buffered_range.d

module buffered_range;

import std.algorithm, std.conv, std.exception, std.range, std.traits, std.typecons;
import core.memory, core.stdc.stdlib;

/*
 * Adds the save function to an input range thus promoting
 * it to a forward range. This is done by buffering the input range
 * in blocks of size blockSize.
 */
struct BufferedRange(R) if(isInputRange!R)
{
    alias ElementType!R T;
    immutable defaultBlockSize = max(4096 / T.sizeof, 1);

    this(R input, size_t blockSize = defaultBlockSize)
    {
        _storage = RefCounted!Storage(input, blockSize);
        fetchBlock();
    }

    @property bool empty() const
    {
        return _curbuf.empty && input.empty;
    }

    @property T front()
    {
        return _curbuf.front;
    }

    @property void popFront()
    {
        _curbuf.popFront;
        if (_curbuf.empty)
            fetchBlock();
    }

    ~this()
    {
        decref();
    }

    this(this)
    {
        incref();
    }

    void opAssign(typeof(this) rhs)
    {
        rhs.incref();
        decref();

        _storage = rhs._storage;
        _curbuf = rhs._curbuf;
        _block = rhs._block;
    }

    @property BufferedRange save()
    {
        return this;
    }

private:
    void fetchBlock()
    {
        // no block at all
        if (_block is null)
        {
            _block = _storage.allocBlock();
            ++_block._refcount;
            _storage.fillBlock(_block);
            _curbuf = _block._data;
        }
        // already has successor, possibly free current block
        else if (_block._next !is null)
        {
            auto next = _block._next;
            if (--_block._refcount == 0)
                _storage.freeBlock(_block);
            _block = next;
            _curbuf = _block._data;
        }
        // sole owner reuse block
        else if (_block._refcount == 1)
        {
            _storage.fillBlock(_block);
            _curbuf = _block._data;
        }
        // still referenced => need to fetch a new one
        else
        {
            _block._next = _storage.allocBlock();
            _block._next._refcount = _block._refcount;
            --_block._refcount;
            _block = _block._next;
            _storage.fillBlock(_block);
            _curbuf = _block._data;
        }
    }

    void incref()
    {
        auto p = _block;
        while (p !is null)
        {
            assert(p._refcount);
            ++p._refcount;
            p = p._next;
        }
    }

    void decref()
    {
        if (_block is null)
            return;

        auto p = _block;

        if (_block._refcount == 1)
            _block = null;

        while (p !is null)
        {
            auto next = p._next;
            assert(p._refcount);
            if (--p._refcount == 0)
                _storage.freeBlock(p);
            p = next;
        }
    }

    @property ref R input()
    {
        return _storage._input;
    }

    @property ref const(R) input() const
    {
        return _storage._input;
    }

    static struct Block
    {
        T[] _data;
        size_t _refcount;
        Block* _next;
    }

    static struct Storage
    {
        ~this()
        {
            foreach(block; _freeList)
            {
                static if (hasIndirections!T)
                    GC.removeRange(block._data.ptr);
                free(block);
            }
        }

        Block* allocBlock()
        {
            typeof(return) res;
            if (_freeList.empty)
            {
                static assert(Block.alignof <= 8 && T.alignof <= 8);
                static assert(Block.alignof % T.alignof == 0);

                auto p = malloc(Block.sizeof + T.sizeof * _blockSize);
                res = emplace(cast(Block*)p);

                auto pt = p + Block.sizeof;
                static if (hasElaborateAssign!T)
                    memset(pt, 0, T.sizeof * _blockSize);
                static if (hasIndirections!T)
                    GC.addRange(pt, T.sizeof * _blockSize);

                res._data = (cast(T*)pt)[0 .. _blockSize];
            }
            else
            {
                res = _freeList.back;
                _freeList.popBack;
            }
            assert(res._refcount == 0);
            return res;
        }

        void freeBlock(Block* p)
        {
            assert(p._refcount == 0);
            _freeList ~= p;
        }

        void fillBlock(Block* p)
        {
            assert(p._refcount);
            auto rdbuf = p._data.ptr[0 .. _blockSize];
            for (; !_input.empty && !rdbuf.empty; _input.popFront, rdbuf.popFront)
            {
                rdbuf.front = _input.front;
            }
            p._data = p._data.ptr[0 .. _blockSize - rdbuf.length];
        }

        R _input;
        /* immutable */ size_t _blockSize;
        Block*[] _freeList;
    }

    RefCounted!Storage _storage;
    T[] _curbuf;
    Block* _block;
    alias input this;
}

template BufferedRange(R) if(isForwardRange!R)
{
     alias R BufferedRange;
}

unittest
{
    static struct InputRange
    {
        this(size_t *pn)
        {
            _pn = pn;
        }

        bool empty() const { return *_pn == 1_000_000; }
        size_t front() { return *_pn; }
        void popFront() { ++*_pn; }
        void reset() { *_pn = 0; }
        size_t* _pn;
    }

    auto inp = InputRange(new size_t);
    auto inp2 = inp;
    assert(array(take(inp, 100)) != array(take(inp2, 100)));
    inp.reset;

    alias BufferedRange!InputRange FwdRange;

    {
        auto fwd1 = FwdRange(inp);
        auto fwd2 = fwd1.save;
        assert(array(take(fwd1, 100)) == array(take(fwd2, 100)));

        // test alias this
        assert(fwd1._pn !is null);

        inp.reset;
    }

    {
        // this is inefficient due to the implicit save-through-copy in foreach
        auto fwd = FwdRange(inp);
        foreach(e; fwd)
        {}

        inp.reset;

        // this is efficient because no second copy is made
        foreach(e; BufferedRange!InputRange(inp))
        {}

        inp.reset;
    }
}