BlockStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2021 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;
 BufferUnlock guard(_lock);
 AMPS_YIELD();
 }
 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
 {
 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