BlockStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2024 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 <amps/ampsplusplus.hpp>
#include <amps/Buffer.hpp>
#include <sstream>
#include <string>
#include <map>
#include <amps/ampscrc.hpp>
#if __cplusplus >= 201103L || _MSC_VER >= 1900
  #include <atomic>
#endif

#ifdef _WIN32
  #include <intrin.h>
  #include <sys/timeb.h>
#else
  #include <sys/time.h>
#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 associated 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
    AMPS_ATOMIC_BASE_TYPE put(amps_uint64_t sequence_)
    {
      assert(_usedList);
      assert(_endOfUsedList);
      Block* used = _usedList;
      AMPS_ATOMIC_BASE_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
#if __cplusplus >= 201103L || _MSC_VER >= 1900
    std::atomic<bool>   _resizing;
#else
    volatile bool       _resizing;
#endif

    // Lock for _buffer
    mutable Mutex   _lock;

  };

}

#endif