Buffer.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_BUFFER_HXX_
#define _UTILS_BUFFER_HXX_
 
// Enable this to collect the pointer of all buffers live.
//#define DEBUG_BUFFER_MEMORY
 
#include <memory>
#include <atomic>
#include <new>
#include <cstdint>
#include <cstdlib>
#include <cstdarg>
 
#include "executor/Executable.hxx"
#include "executor/Notifiable.hxx"
#include "os/OS.hxx"
#include "utils/Atomic.hxx"
#include "utils/MultiMap.hxx"
#include "utils/QMember.hxx"
#include "utils/Queue.hxx"
#include "utils/macros.h"
 
class DynamicPool;
class FixedPool;
class LimitedPool;
class Pool;
template <class T> class Buffer;
class BufferBase;
class ForwardAllocator;
 
namespace openlcb
{
class AsyncIfTest;
}
 
extern const unsigned LARGEST_BUFFERPOOL_BUCKET;
 
extern DynamicPool *mainBufferPool;
 
Pool* init_main_buffer_pool();
 
extern void* g_current_alloc;
 
class BufferBase : public QMember
{
public:
    uint16_t references()
    {
        return count_.load();
    }
 
    void set_done(BarrierNotifiable *done)
    {
        if (done_)
        {
            done_->notify();
        }
        done_ = done;
    }
 
    BarrierNotifiable *new_child()
    {
        if (done_)
        {
            return done_->new_child();
        }
        else
        {
            return nullptr;
        }
    }
 
    size_t size()
    {
        return size_;
    }
 
protected:
    Pool *pool()
    {
        return pool_;
    }
 
    Pool *pool_;
 
    BarrierNotifiable *done_;
 
    uint16_t size_;
 
    std::atomic_uint_least16_t count_;
 
    BufferBase(size_t size, Pool *pool)
        : QMember()
        , pool_(pool)
        , done_(NULL)
        , size_(size)
        , count_(1)
    {
    }
 
    ~BufferBase()
    {
        if (done_)
        {
            done_->notify();
            done_ = nullptr;
        }
    }
 
    friend class openlcb::AsyncIfTest;
 
    friend class Pool;
 
    friend class DynamicPool;
 
    friend class FixedPool;
 
    friend class LimitedPool;
 
    friend class DataBufferPool;
 
    DISALLOW_COPY_AND_ASSIGN(BufferBase);
};
 
template <class T> class Buffer : public BufferBase
{
public:
    typedef T value_type;
 
    Buffer<T> *ref()
    {
        ++count_;
        return this;
    }
 
    inline void unref();
 
    T *data()
    {
        return &data_;
    }
 
private:
    Buffer(Pool *pool)
        : BufferBase(sizeof(Buffer<T>), pool)
    {
    }
 
    ~Buffer()
    {
    }
 
    DISALLOW_COPY_AND_ASSIGN(Buffer);
 
    friend class Pool;
    friend class DataBuffer;
 
    T data_;
};
 
template<typename T> struct BufferDelete {
    void operator()(Buffer<T>* b) {
        if (b) b->unref();
    }
};
 
template<typename T> using AutoReleaseBuffer = std::unique_ptr<Buffer<T>, BufferDelete<T>>;
template<typename T> using BufferPtr = AutoReleaseBuffer<T>;
 
template<typename T> BufferPtr<T> get_buffer_deleter(Buffer<T>* b) {
    return BufferPtr<T>(b);
}
 
class Pool
{
public:
    size_t total_size()
    {
        return totalSize;
    }
 
    template <class BufferType>
    void alloc(Buffer<BufferType> **result, Executable *flow = NULL)
    {
#ifdef DEBUG_BUFFER_MEMORY
        g_current_alloc = &&alloc;
        alloc:
#endif
        *result = static_cast<Buffer<BufferType> *>(
            alloc_untyped(sizeof(Buffer<BufferType>), flow));
        if (*result && !flow)
        {
            new (*result) Buffer<BufferType>(this);
        }
    }
 
    template <class BufferType> void alloc(BufferPtr<BufferType> *result)
    {
        Buffer<BufferType> *p;
        alloc(&p);
        result->reset(p);
    }
 
    template <class BufferType> void alloc_async(Executable *flow)
    {
        Buffer<BufferType> *buffer;
        alloc(&buffer, flow);
    }
 
    template <class BufferType>
    static void alloc_async_init(BufferBase *base, Buffer<BufferType> **result)
    {
        HASSERT(base);
        HASSERT(sizeof(Buffer<BufferType>) == base->size());
        *result = static_cast<Buffer<BufferType> *>(base);
        new (*result) Buffer<BufferType>(base->pool());
    }
 
    virtual size_t free_items() = 0;
 
    virtual size_t free_items(size_t size) = 0;
 
protected:
    Pool()
        : totalSize(0)
    {
    }
 
    virtual ~Pool()
    {
    }
 
    virtual BufferBase *alloc_untyped(size_t size, Executable *flow) = 0;
 
    virtual void free(BufferBase *item) = 0;
 
    size_t totalSize;
 
private:
    friend class BufferBase;
    friend class LimitedPool;
    friend class DataBufferPool;
 
    template <class T> friend class Buffer;
 
    DISALLOW_COPY_AND_ASSIGN(Pool);
};
 
class Bucket : public Q
{
public:
    static Bucket *init(int s, ...)
    {
        va_list ap, aq;
        va_start(ap, s);
        va_copy(aq, ap);
        int count = 1;
        int current = s;
 
        while (current != 0)
        {
            ++count;
            int next = va_arg(ap, int);
            HASSERT(next > current || next == 0);
            current = next;
        }
 
        Bucket *bucket = (Bucket *)malloc(sizeof(Bucket) * count);
        Bucket *now = bucket;
 
        new (now) Bucket(s);
        now++;
 
        for (int i = 1; i < count; ++i)
        {
            new (now) Bucket(va_arg(aq, int));
            now++;
        }
 
        va_end(aq);
        va_end(ap);
        return bucket;
    }
 
    static void destroy(Bucket *bucket)
    {
        free(bucket);
    }
 
    size_t size()
    {
        return size_;
    }
 
    QMember *executables()
    {
        return pending_.next().item;
    }
 
private:
    Bucket(size_t size)
        : Q()
        , size_(size)
        , pending_()
    {
    }
 
    ~Bucket()
    {
    }
 
    size_t size_; 
public:
    size_t allocCount_{0}; 
private:
    Q pending_;
};
 
class DynamicPool : public Pool, private Atomic
{
public:
    DynamicPool(Bucket sizes[])
        : Pool()
        , buckets(sizes)
    {
    }
 
    ~DynamicPool()
    {
#ifdef GTEST
        for (unsigned i = 0; buckets[i].size() != 0; ++i)
        {
            // Frees all memory left in the bucket.
            do
            {
                auto *p = static_cast<BufferBase *>(buckets[i].next().item);
                if (!p)
                {
                    break;
                }
                ::free(p);
            } while (true);
        }
#endif
        Bucket::destroy(buckets);
    }
 
    size_t free_items() override;
 
    size_t free_items(size_t size) override;
 
protected:
    Bucket *buckets;
 
private:
    BufferBase *alloc_untyped(size_t size, Executable *flow) override;
 
    void* alloc_large(size_t size);
    void free_large(void* block);
 
    void free(BufferBase *item) override;
 
    DynamicPool();
 
    friend class ForwardAllocator;
 
    DISALLOW_COPY_AND_ASSIGN(DynamicPool);
};
 
class FixedPool : public Pool, public Atomic
{
public:
    bool valid(QMember *item)
    {
        if ((char *)item >= mempool &&
            (char *)item < (mempool + (items * itemSize)))
        {
            return true;
        }
        return false;
    }
 
    FixedPool(size_t item_size, size_t items)
        : Pool()
        , mempool(new char[(item_size * items)])
        , itemSize(item_size)
        , items(items)
        , empty(false)
    {
        // HASSERT(item_size != 0 && items != 0);
        QMember *current = (QMember *)mempool;
        for (size_t i = 0; i < items; ++i)
        {
            current->init();
            queue.insert(current);
            current = (QMember *)((char *)current + item_size);
        }
    }
 
    ~FixedPool()
    {
        delete[] mempool;
    }
 
    size_t free_items() override
    {
        return items - (totalSize / itemSize);
    }
 
    size_t free_items(size_t size) override
    {
        return size == itemSize ? free_items() : 0;
    }
 
protected:
    Q queue;
 
    char *mempool;
 
    size_t itemSize;
 
    size_t items;
 
    bool empty;
 
private:
    BufferBase *alloc_untyped(size_t size, Executable *flow) override;
 
    void free(BufferBase *item) override;
 
    FixedPool();
 
    DISALLOW_COPY_AND_ASSIGN(FixedPool);
};
 
template <class T> void Buffer<T>::unref()
{
    if (count_.fetch_sub(1) == 1u)
    {
        this->~Buffer();
        pool_->free(this);
    }
}
 
#endif /* _UTILS_BUFFER_HXX_ */