Buffer.cxx

Go to the documentation of this file.

 
#include "utils/Buffer.hxx"
#include "utils/ByteBuffer.hxx"
 
DynamicPool *mainBufferPool = nullptr;
Pool *rawBufferPool = nullptr;
 
Pool* init_main_buffer_pool()
{
    if (!rawBufferPool)
    {
        rawBufferPool = new DynamicPool(Bucket::init(sizeof(RawBuffer), 0));
    }
    if (!mainBufferPool)
    {
        mainBufferPool =
            new DynamicPool(Bucket::init(32, 48, LARGEST_BUFFERPOOL_BUCKET, 0));
    }
    return mainBufferPool;
}
 
size_t DynamicPool::free_items()
{
    size_t total = 0;
    for (Bucket *current = buckets; current->size() != 0; ++current)
    {
        total += current->pending();
    }
    return total;
}
 
size_t DynamicPool::free_items(size_t size)
{
    for (Bucket *current = buckets; current->size() != 0; ++current)
    {
        if (current->size() >= size)
        {
            return current->pending();
        }
    }
    return 0;
}
 
#ifdef DEBUG_BUFFER_MEMORY
/* key: buffer pointer. Value: instruction pointer for allocation caller. */
std::map<BufferBase*, void*> g_alloc_source;
Atomic g_alloc_atomic;
void* g_current_alloc;
#endif
 
extern "C" {
extern void *buffer_malloc(size_t length);
}
 
BufferBase *DynamicPool::alloc_untyped(size_t size, Executable *flow)
{
    BufferBase *result = NULL;
 
    for (Bucket *current = buckets; current->size() != 0; ++current)
    {
        if (size <= current->size())
        {
            result = static_cast<BufferBase*>(current->next().item);
            if (result == NULL)
            {
                result = (BufferBase*)buffer_malloc(current->size());
                {
                    AtomicHolder h(this);
                    if (0 && totalSize < 5000 && totalSize + current->size() >= 5000) {
                        HASSERT(0);
                    }
                    current->allocCount_++;
                    totalSize += current->size();
                }
            }
            new (result) BufferBase(size, this);
            break;
        }
    }
 
    if (!result)
    {
        /* big items are just malloc'd freely */
        result = (BufferBase*)alloc_large(size);
        new (result) BufferBase(size, this);
        {
            AtomicHolder h(this);
            totalSize += size;
        }
    }
#ifdef DEBUG_BUFFER_MEMORY
    {
        AtomicHolder h(&g_alloc_atomic);
        g_alloc_source[result] = g_current_alloc;
    }
#endif
    if (flow)
    {
        flow->alloc_result(result);
    }
    return result;
}
 
void* DynamicPool::alloc_large(size_t size) {
    return malloc(size);
}
 
void DynamicPool::free_large(void* buffer) {
    ::free(buffer);
}
 
void DynamicPool::free(BufferBase *item)
{
#ifdef DEBUG_BUFFER_MEMORY
    {
        AtomicHolder h(&g_alloc_atomic);
        g_alloc_source.erase(item);
    }
#endif
    for (Bucket *current = buckets; current->size() != 0; ++current)
    {
        if (item->size() <= current->size())
        {
            current->insert(item);
            return;
        }
    }
    /* big items are just freed */
    {
        AtomicHolder h(this);
        totalSize -= item->size();
    }
    free_large(item);
}
 
BufferBase *FixedPool::alloc_untyped(size_t size, Executable *flow)
{
    BufferBase *result = NULL;
    
    {
        AtomicHolder h(this);
        if (empty == false)
        {
            result = static_cast<BufferBase *>(queue.next(0));
            if (result)
            {
                new (result) BufferBase(size, this);
                totalSize += itemSize;
            }
            else
            {
                empty = true;
            }
        }
        if (flow && empty)
        {
            queue.insert(flow);
        }
    }
    if (result && flow)
    {
        flow->alloc_result(result);
    }
    return result;
}
 
void FixedPool::free(BufferBase *item)
{
    HASSERT(valid(item));
    HASSERT(item->size() <= itemSize);
    
    Executable *waiting = NULL;
    {
        AtomicHolder h(this);
        if (empty == true)
        {
            waiting = static_cast<Executable *>(queue.next().item);
            if (queue.empty())
            {
                empty = false;
            }
            if (!waiting)
            {
                queue.insert(item);
                totalSize -= itemSize;
            }
        }
        else
        {
            queue.insert(item);
            totalSize -= itemSize;
        }
    }
    if (waiting)
    {
        waiting->alloc_result(item);
    }
 
}