BlockStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2023 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//

#ifndef _BLOCKSTORE_H_
#define _BLOCKSTORE_H_
#include <ampsplusplus.hpp>
#include <Buffer.hpp>
#include <sstream>
#include <string>
#include <map>
#include <ampscrc.hpp>

#ifdef _WIN32
#include <intrin.h>
#include <sys/timeb.h>
#else
#include <sys/time.h>
#if AMPS_SSE_42
#include <cpuid.h>
#endif
#endif
#include <iostream>


namespace AMPS
{
class BlockStore
{
public:
    enum Constants : amps_uint32_t
    {
        DEFAULT_BLOCK_HEADER_SIZE  =   32,
        DEFAULT_BLOCKS_PER_REALLOC = 1000,
        DEFAULT_BLOCK_SIZE         = 2048
    };

    class Block
    {
        public:
            // The offset of the Block's data in the buffer.
            size_t          _offset;
            // The sequence number assoicated with the Block.
            amps_uint64_t   _sequence;
            // The next Block in the chain when data is in multiple Blocks.
            Block*          _nextInChain;
            // The next Block in list of available or free Blocks.
            Block*          _nextInList;

            // Create Block with given offset
            Block(size_t offset_) : _offset(offset_), _sequence(0)
                                  , _nextInChain(0), _nextInList(0)
            { ; }

            // Create Block with _nextInList at an address one Block farther
            // than self. Convenient for creating arrays of Blocks.
            Block() : _offset(0), _sequence(0)
                    , _nextInChain(0), _nextInList((Block*)(this+1))
            { ; }

            // Init Block to an offset at index_ * blockSize_
            Block* init(size_t index_, amps_uint32_t blockSize_)
            {
                _offset = index_*blockSize_;
                return this;
            }

            // Set Block to given offset and return pointer to self
            Block* setOffset(size_t offset_)
            {
                _offset = offset_;
                return this;
            }

    };

private:
    // Typedefs
    typedef Lock<Mutex> BufferLock;
    typedef Unlock<Mutex> BufferUnlock;
    typedef bool (*ResizeHandler)(size_t,void*);
    typedef std::vector<Block*> BlockList;

public:
    BlockStore(Buffer* buffer_,
               amps_uint32_t blocksPerRealloc_ = DEFAULT_BLOCKS_PER_REALLOC,
               amps_uint32_t blockHeaderSize_ = DEFAULT_BLOCK_HEADER_SIZE,
               amps_uint32_t blockSize_ = DEFAULT_BLOCK_SIZE)
        : _buffer(buffer_), _freeList(0), _usedList(0)
        , _endOfUsedList(0), _blocksPerRealloc(blocksPerRealloc_)
        , _blockSize(blockSize_), _blockHeaderSize(blockHeaderSize_)
        , _blocksAvailable(0), _resizeHandler(0), _resizeUserData(0)
        , _resizing(false)
    {
    }

    ~BlockStore()
    {
        for (BlockList::iterator i = _blockList.begin();
             i != _blockList.end(); ++i)
        {
            delete[] *i;
        }
        delete _buffer;
    }

    amps_uint32_t getBlockSize() const
    {
        return _blockSize;
    }

    amps_uint32_t getBlockHeaderSize() const
    {
        return _blockHeaderSize;
    }

    void acquireRead() const
    {
        _lock.acquireRead();
    }

    void releaseRead() const
    {
        _lock.releaseRead();
    }

    void signalAll()
    {
        _lock.signalAll();
    }

    void wait()
    {
        _lock.wait();
    }

    bool wait(long timeout_)
    {
        return _lock.wait(timeout_);
    }

    void setResizeHandler(ResizeHandler resizeHandler_, void* userData_)
    {
        _resizeHandler = resizeHandler_;
        _resizeUserData = userData_;
    }

    // Lock should already be acquired
    Block* front() const
    {
        return _usedList;
    }

    // Lock should already be acquired
    Block* back() const
    {
        return _endOfUsedList;
    }

    // Lock should already be acquired
    void setFreeList(Block* block_, amps_uint32_t freeCount_)
    {
        _freeList = block_;
        _blocksAvailable = freeCount_;
    }

    // Lock should already be acquired
    void setUsedList(Block* block_)
    {
        _usedList = block_;
    }

    // Lock should already be acquired
    void setEndOfUsedList(Block* block_)
    {
        _endOfUsedList = block_;
    }

    // Lock should already be acquired
    void addBlocks(Block* blockArray_)
    {
        _blockList.push_back(blockArray_);
    }

    // Lock should already be acquired
    Block* get(amps_uint32_t numBlocksInChain_)
    {
        // Check that we have enough blocks
        // Do this in a loop since resize can possibly return without resizing
        // and may still leave us needing more space.
        while (_blocksAvailable < numBlocksInChain_)
        {
            // Resize by required multiple of blockPerRealloc
            unsigned int blocksNeeded = numBlocksInChain_ - _blocksAvailable;
            amps_uint32_t addedBlocks = (blocksNeeded/_blocksPerRealloc + 1)
                                        * _blocksPerRealloc;
            size_t size = _buffer->getSize() + (addedBlocks * _blockSize);
            resize(size);
        }
        // Return first free block with others as _nextInChain
        Block* first = 0;
        Block* last = 0;
        Block* next = 0;
        for (unsigned int i=0; i<numBlocksInChain_; ++i)
        {
            // Take from free list and advance
            next = _freeList;
            _freeList = _freeList->_nextInList;
            next->_nextInList = 0;
            if (!first)
            {
                // First, set it up
                first = next;
                last = next;
            }
            else
            {
                // Not first, add it to chain
                last->_nextInChain = next;
                last = next;
            }
        }
        assert(first);
        // Set _usedList or add it to the end of the used list
        if (!_usedList)
        {
            _usedList = first;
        }
        else
        {
            _endOfUsedList->_nextInList = first;
        }
        _endOfUsedList = first;
        _blocksAvailable -= numBlocksInChain_;
        return first;
    }

    // Lock should already be acquired
    void put(Block* block_)
    {
        assert(_usedList);
        assert(_endOfUsedList);
        // Remove from used list
        if (_usedList == block_)
        {
            // Easy
            _usedList = _usedList->_nextInList;
            if (!_usedList)
            {
                _endOfUsedList = 0;
            }
        }
        else
        {
            // Search and remove the block
            Block* used=_usedList;
            while (used)
            {
                if (used->_nextInList == block_)
                {
                    used->_nextInList = block_->_nextInList;
                    break;
                }
                used = used->_nextInList;
                if (!_usedList) // -V1051
                {
                    _endOfUsedList = 0;
                }
            }
        }
        // Add to free list
        _flattenToFreeList(block_);
    }

    // Lock should already be acquired
    ATOMIC_TYPE put(amps_uint64_t sequence_)
    {
        assert(_usedList);
        assert(_endOfUsedList);
        Block* used=_usedList;
        ATOMIC_TYPE removalCount = 0;
        while (used && used->_sequence <= sequence_)
        {
            Block* next = used->_nextInList;
            // Add to free list
            _flattenToFreeList(used);
            used = next;
            ++removalCount;
        }
        _usedList = used;
        if (!used)
        {
            _endOfUsedList = 0;
        }
        return removalCount;
    }

    // Lock should already be acquired
    void putAll(Block* block_)
    {
        // Remove from used list
        Block* newEndOfUsedList = 0;
        for (Block* used = _usedList; used; used = used->_nextInList)
        {
            if (used == block_)
            {
                if (newEndOfUsedList)
                {
                    newEndOfUsedList->_nextInList = 0;
                }
                else
                {
                    _usedList = 0;
                }
                _endOfUsedList = newEndOfUsedList;
            }
            newEndOfUsedList = used;
        }
        // Add all remaining to free list
        Block* next = 0;
        for (Block* block = block_; block; block = next)
        {
            next = block->_nextInList;
            _flattenToFreeList(block);
        }
    }

    // Lock should already be held
    void init()
    {
        size_t startSize = _buffer->getSize();
        if (!startSize)
        {
            resize(getDefaultResizeSize());
            startSize = _buffer->getSize();
        }
        // How many blocks are we resizing
        amps_uint32_t numBlocks = (amps_uint32_t)(startSize) / getBlockSize();
        _freeList = new Block[numBlocks];
        _blockList.push_back(_freeList);
        for (size_t i = 0; i < numBlocks; ++i)
        {
            _freeList[i].init(i, getBlockSize());
        }
        _freeList[numBlocks-1]._nextInList = 0;
        _blocksAvailable += numBlocks;
        assert(_freeList);
        assert(_blocksAvailable);
    }

    size_t getDefaultResizeSize() const
    {
        return _blocksPerRealloc * _blockSize;
    }

    amps_uint32_t getDefaultResizeBlocks() const
    {
        return _blocksPerRealloc;
    }

    // Lock should already be held
    Block* resizeBuffer(size_t size_, amps_uint32_t* pNewBlocks_)
    {
        Block* freeList = 0;
        while (_resizing)
        {
            if (_buffer->getSize() >= size_) return freeList;
            if (!_lock.wait(1000))
            {
                amps_invoke_waiting_function();
            }
        }
        FlagFlip flip(&_resizing);
        bool okToResize = false;
        if (true)
        {
            BufferUnlock u(_lock);
            // Don't do anything if resizeHandler says no
            okToResize = _canResize(size_);
        }
        if (!okToResize)
        {
            return freeList;
        }
        try
        {
            _lock.signalAll();
            size_t oldSize = _buffer->getSize();
            amps_uint32_t oldBlocks = (amps_uint32_t)(oldSize / getBlockSize());
            if (oldSize >= size_)
            {
                *pNewBlocks_ = 0;
                return freeList;
            }
            _buffer->setSize(size_);
            _buffer->zero(oldSize, size_-oldSize);
            // How many blocks are we resizing
            *pNewBlocks_ = (amps_uint32_t)((size_-oldSize) / getBlockSize());
            freeList = new Block[*pNewBlocks_];
            for (size_t i = 0; i < *pNewBlocks_; ++i)
            {
                freeList[i].init(oldBlocks+i, getBlockSize());
            }
            freeList[*pNewBlocks_-1]._nextInList = 0;
        }
#ifdef _WIN32
        catch(const std::bad_alloc&)
#else
        catch(const std::bad_alloc& e)
#endif
        {
            std::ostringstream os;
            os << "BlockStore failed to allocate " << size_
               << " bytes for resize of store from " << _buffer->getSize()
               << " bytes.";
            throw StoreException(os.str());
        }
        return freeList;
    }

    // Lock should already be held
    void resize(size_t size_)
    {
        amps_uint32_t newBlocks = 0;
        Block* addedBlockList = resizeBuffer(size_, &newBlocks);
        if (!addedBlockList || !newBlocks)
        {
            // Maybe we didn't have to allocate in this thread
            return;
        }
        _blockList.push_back(addedBlockList);
        addedBlockList[newBlocks-1]._nextInList = _freeList;
        _freeList = addedBlockList;
        _blocksAvailable += newBlocks;
        assert(_blocksAvailable);
    }

    // Lock should be held, no blocks should be used or allocated
    amps_uint32_t setBlockSize(amps_uint32_t blockSize_)
    {
        if (_usedList || _freeList)
        {
            return 0;
        }
        amps_uint32_t oldSize = _blockSize;
        _blockSize = blockSize_;
        return oldSize;
    }

    // Lock should be held, no blocks should be used or allocated
    amps_uint32_t setBlockHeaderSize(amps_uint32_t blockHeaderSize_)
    {
        if (_usedList || _freeList)
        {
            return 0;
        }
        amps_uint32_t oldSize = _blockHeaderSize;
        _blockHeaderSize = blockHeaderSize_;
        return oldSize;
    }

    // Lock should already be held
    Buffer* getBuffer()
    {
        return _buffer;
    }

private:
    bool _canResize(size_t requestedSize_)
    {
        if (_resizeHandler)
        {
            return _resizeHandler(requestedSize_, _resizeUserData);
        }
        else
        {
            return true;
        }
    }

    // Lock should already be acquired
    void _flattenToFreeList(Block* block_)
    {
        // Flatten chain to front of free list
        Block* current = block_;
        while (current)
        {
            Block* chain = current->_nextInChain;
            // Clear the header
            _buffer->zero(current->_offset, _blockHeaderSize);
            // Prepend to the free list and clear other values
            current->_nextInList = _freeList;
            _freeList = current;
            ++_blocksAvailable;
            current->_sequence = (amps_uint64_t)0;
            current->_nextInChain = 0;
            current = chain;
        }
        assert(_freeList);
        assert(_blocksAvailable);
    }

    // Member variables
    // Buffer to use for storage
    Buffer*         _buffer;

    // The Block accounting
    Block*              _freeList;
    Block*              _usedList;
    Block*              _endOfUsedList;
    // How much to resize buffer when needed
    amps_uint32_t       _blocksPerRealloc;
    // How big is each Block, and what part is header
    amps_uint32_t       _blockSize;
    amps_uint32_t       _blockHeaderSize;
    // How many blocks are free
    amps_uint32_t       _blocksAvailable;
    // ResizeHandler to call before resizing
    ResizeHandler       _resizeHandler;
    // ResizeHandler data
    void*               _resizeUserData;
    // List of every allocated slab of Blocks
    BlockList           _blockList;
    // Flag to control resizing
    volatile bool       _resizing;

    // Lock for _buffer
    mutable Mutex   _lock;

};

}

#endif