BufferDetail.hh

Go to the documentation of this file.

#ifndef avro_BufferDetail_hh__
#define avro_BufferDetail_hh__

#include <boost/shared_ptr.hpp>
#include <boost/shared_array.hpp>
#include <boost/static_assert.hpp>
#include <boost/function.hpp>
#ifdef HAVE_BOOST_ASIO
#include <boost/asio/buffer.hpp>
#endif
#include <exception>
#include <cassert>
#include <deque>

namespace avro {

namespace detail {

typedef char    data_type;
typedef size_t  size_type;
#ifdef HAVE_BOOST_ASIO
typedef boost::asio::const_buffer   ConstAsioBuffer;
typedef boost::asio::mutable_buffer MutableAsioBuffer;
#endif

const size_type kMinBlockSize = 4096;
const size_type kMaxBlockSize = 16384;
const size_type kDefaultBlockSize = kMinBlockSize;

typedef boost::function<void(void)>  free_func;

class CallOnDestroy {
  public:
    CallOnDestroy(const free_func &func) : func_(func)
    { }
    ~CallOnDestroy() {
        if (func_) {
            func_();
        }
    }
  private:
    free_func func_;
};

class Chunk 
{

  public:

    typedef boost::shared_ptr<Chunk> SharedPtr;

    Chunk(size_type size) :
        underlyingBlock_(new data_type[size]), 
        readPos_(underlyingBlock_.get()),
        writePos_(readPos_),
        endPos_(readPos_ + size)
    { }

    Chunk(const data_type *data, size_type size, const free_func &func) :
        callOnDestroy_(new CallOnDestroy(func)),
        readPos_(const_cast<data_type *>(data)),
        writePos_(readPos_ + size),
        endPos_(writePos_)
    { }

  private:
    // reference counted object will call a functor when it's destroyed
    boost::shared_ptr<CallOnDestroy> callOnDestroy_;

  public:

    void truncateFront(size_type howMuch) {
        readPos_ += howMuch;
        assert(readPos_ <= writePos_);
    }

    void truncateBack(size_type howMuch) {
        writePos_ -= howMuch;
        assert(readPos_ <= writePos_);
    }

    data_type *tellWritePos() const {
        return writePos_;
    }

    const data_type *tellReadPos() const {
        return readPos_;
    }

    void incrementCursor(size_type howMuch) {
        writePos_  += howMuch; 
        assert(writePos_ <= endPos_);
    }

    size_type dataSize() const {
        return (writePos_ - readPos_);
    }

    size_type freeSize() const {
        return (endPos_ - writePos_);
    }

    size_type capacity() const {
        return (endPos_ - readPos_);
    }

  private:

    friend bool operator==(const Chunk &lhs, const Chunk &rhs);
    friend bool operator!=(const Chunk &lhs, const Chunk &rhs);

    // more than one buffer can share an underlying block, so use SharedPtr
    boost::shared_array<data_type> underlyingBlock_;

    data_type *readPos_;  
    data_type *writePos_; 
    data_type *endPos_;   
};

inline bool operator==(const Chunk &lhs, const Chunk &rhs) {
    return lhs.underlyingBlock_ == rhs.underlyingBlock_;
}

inline bool operator!=(const Chunk &lhs, const Chunk &rhs) {
    return lhs.underlyingBlock_ != rhs.underlyingBlock_;
}


class BufferImpl : boost::noncopyable
{

    void allocChunkChecked(size_type size = kDefaultBlockSize) 
    {
        writeChunks_.push_back(Chunk(size));
        freeSpace_ += writeChunks_.back().freeSize();
    }

    void allocChunk(size_type size) 
    {
        if(size < kMinBlockSize) {
            size = kMinBlockSize;
        }
        else if (size > kMaxBlockSize) {
            size = kMaxBlockSize;
        }
        allocChunkChecked(size);
    }

    void postWrite(size_type size) 
    {

        // precondition to this function is that the writeChunk_.front()
        // contains the data that was just written, so make sure writeChunks_
        // is not empty:
        
        assert(size <= freeSpace_ && !writeChunks_.empty());

        // This is probably the one tricky part of BufferImpl.  The data that
        // was written now exists in writeChunks_.front().  Now we must make
        // sure that same data exists in readChunks_.back().
        //
        // There are two cases: 
        //
        // 1. readChunks_.last() and writeChunk_.front() refer to the same
        // underlying block, in which case they both just need their cursor
        // updated to reflect the new state.
        //
        // 2. readChunk_.last() is not the same block as writeChunks_.front(),
        // in which case it should be, since the writeChunk.front() contains
        // the next bit of data that will be appended to readChunks_, and
        // therefore needs to be copied there so we can proceed with updating
        // their state.
        //

        // if readChunks_ is not the same as writeChunks_.front(), make a copy
        // of it there
        
        if(readChunks_.empty() || (readChunks_.back() != writeChunks_.front())) {
            const Chunk &curChunk = writeChunks_.front();
            readChunks_.push_back(curChunk);

            // Any data that existed in the write chunk previously doesn't
            // belong to this buffer (otherwise it would have already been
            // added to the readChunk_ list).  Here, adjust the start of the
            // readChunk to begin after any data already existing in curChunk
            
            readChunks_.back().truncateFront( curChunk.dataSize());
        }

        assert(readChunks_.back().freeSize() == writeChunks_.front().freeSize());

        // update the states of both readChunks_ and writeChunks_ to indicate that they are
        // holding the new data
        
        readChunks_.back().incrementCursor(size);
        writeChunks_.front().incrementCursor(size);
        size_ += size;
        freeSpace_ -= size;

        // if there is no more free space in writeChunks_, the next write cannot use
        // it, so dispose of it now
        
        if(writeChunks_.front().freeSize() == 0) {
            writeChunks_.pop_front();
        }
    }

  public:

    typedef std::deque<Chunk> ChunkList;
    typedef boost::shared_ptr<BufferImpl> SharedPtr;
    typedef boost::shared_ptr<const BufferImpl> ConstSharedPtr;

    BufferImpl() :
        freeSpace_(0),
        size_(0)
    { }

    explicit BufferImpl(const BufferImpl &src) :
        readChunks_(src.readChunks_), 
        freeSpace_(0),
        size_(src.size_)
    { }

    size_type size() const {
        return size_;
    }

    size_type freeSpace() const {
        return freeSpace_;
    }

    void reserveFreeSpace(size_type reserveSize) {
        while(freeSpace_ < reserveSize) {
            allocChunk(reserveSize - freeSpace_);
        }
    }

    ChunkList::const_iterator beginRead() const {
        return readChunks_.begin();
    }
    
    ChunkList::const_iterator endRead() const {
        return readChunks_.end();
    }

    ChunkList::const_iterator beginWrite() const {
        return writeChunks_.begin();
    }
    
    ChunkList::const_iterator endWrite() const {
        return writeChunks_.end();
    }

    template<typename T>
    void writeTo(T val, const boost::true_type&)
    {
        if(freeSpace_ && (sizeof(T) <= writeChunks_.front().freeSize())) {
            // fast path, there's enough room in the writeable chunk to just
            // straight out copy it
            *(reinterpret_cast <T*> ( writeChunks_.front().tellWritePos()) ) = val;
            postWrite(sizeof(T));
        }
        else {
            // need to fixup chunks first, so use the regular memcpy 
            // writeTo method
            writeTo(reinterpret_cast<data_type*>(&val), sizeof(T));
        }
    }

    template<typename T>
    void writeTo(T val, const boost::false_type&) 
    {
        BOOST_STATIC_ASSERT(sizeof(T)==0);
    }

    size_type writeTo(const data_type *data, size_type size) 
    {
        size_type bytesLeft = size; 
        while(bytesLeft) {

            if(freeSpace_ == 0) {
                allocChunkChecked();
            }

            Chunk &chunk = writeChunks_.front();
            size_type toCopy = std::min<size_type>(chunk.freeSize(), bytesLeft);
            assert(toCopy);
            memcpy(chunk.tellWritePos(), data, toCopy);
            postWrite(toCopy);
            data      += toCopy; 
            bytesLeft -= toCopy; 
        }
        return size;
    }

    size_type wroteTo(size_type size) 
    {
        assert(size <= freeSpace_);
        size_type bytesLeft = size;
        while (bytesLeft) {

            Chunk &chunk = writeChunks_.front();
            size_type wrote = std::min<size_type>(chunk.freeSize(), bytesLeft);
            assert(wrote);
            postWrite(wrote);
            bytesLeft -= wrote;
        }
        return size;
    }

    void append(const BufferImpl &src) {
        std::copy(src.readChunks_.begin(), src.readChunks_.end(), std::back_inserter(readChunks_));
        size_ += src.size_;
    }

    void discardData() {
        readChunks_.clear();
        size_ = 0;
    }

    void discardData(size_type bytes)
    {
        assert(bytes && bytes <= size_);

        size_type bytesToDiscard = bytes;
        while( bytesToDiscard ) {
          
            size_t currentSize = readChunks_.front().dataSize();

            // see if entire chunk is discarded
            if(currentSize <= bytesToDiscard) {
                readChunks_.pop_front();
                bytesToDiscard -= currentSize;
            }
            else {
                readChunks_.front().truncateFront(bytesToDiscard);
                bytesToDiscard = 0;
            }
        }

        size_ -= bytes;
    }

    void extractData(BufferImpl &dest, size_type bytes)
    {
        assert(bytes && bytes <= size_);

        size_type bytesToExtract = bytes;
        while( bytesToExtract ) {
          
            size_t currentSize = readChunks_.front().dataSize();
            dest.readChunks_.push_back(readChunks_.front());

            // see if entire chunk was extracted 
            if(currentSize <= bytesToExtract) {
                readChunks_.pop_front();
                bytesToExtract -= currentSize;
            }
            else {
                readChunks_.front().truncateFront(bytesToExtract);
                size_t excess = currentSize - bytesToExtract;
                dest.readChunks_.back().truncateBack(excess);
                bytesToExtract = 0;
            }
        }

        size_ -= bytes;
        dest.size_ += bytes;
    }

    void extractData(BufferImpl &dest) 
    {
        assert(dest.readChunks_.empty());
        dest.readChunks_.swap(readChunks_);
        dest.size_ = size_;
        size_ = 0;
    }

    void copyData(BufferImpl &dest, 
                  ChunkList::const_iterator iter,
                  size_type offset, 
                  size_type bytes) const
    {
        // now we are positioned to start the copying, copy as many
        // chunks as we need, the first chunk may have a non-zero offset 
        // if the data to copy is not at the start of the chunk 
        size_type copied = 0;
        while(copied < bytes) {

            dest.readChunks_.push_back(*iter);

            // offset only applies in the first chunk, 
            // all subsequent chunks are copied from the start
            dest.readChunks_.back().truncateFront(offset);
            offset = 0;

            copied += dest.readChunks_.back().dataSize(); 
            ++iter;
        }

        // if the last chunk copied has more bytes than we need, truncate it
        size_type excess = copied - bytes;
        dest.readChunks_.back().truncateBack(excess);

        dest.size_ += bytes;
    }

    int numDataChunks() const {
        return readChunks_.size();
    }

    int numFreeChunks() const {
        return writeChunks_.size();
    }

    void appendForeignData(const data_type *data, size_type size, const free_func &func) {
        readChunks_.push_back(Chunk(data, size, func));
        size_ += size;
    }

  private:

    BufferImpl& operator=(const BufferImpl &src);
    /* {
        readChunks_.assign(src.readChunks_.begin(), src.readChunks_.end());
        size_ = src.size();
        return *this;
    } */

    ChunkList readChunks_;     
    ChunkList writeChunks_;    

    size_type freeSpace_;  
    size_type size_;       

};

} // detail namespace

} // namespace

#endif