26 #ifndef _BLOCKSTORE_H_ 27 #define _BLOCKSTORE_H_ 33 #include <ampscrc.hpp> 37 #include <sys/timeb.h> 65 DEFAULT_BLOCK_HEADER_SIZE = 32,
66 DEFAULT_BLOCKS_PER_REALLOC = 1000,
67 DEFAULT_BLOCK_SIZE = 2048
78 amps_uint64_t _sequence;
85 Block(
size_t offset_) : _offset(offset_), _sequence(0)
86 , _nextInChain(0), _nextInList(0)
91 Block() : _offset(0), _sequence(0)
92 , _nextInChain(0), _nextInList((
Block*)(
this+1))
96 Block* init(
size_t index_, amps_uint32_t blockSize_)
98 _offset = index_*blockSize_;
103 Block* setOffset(
size_t offset_)
113 typedef Lock<Mutex> BufferLock;
114 typedef Unlock<Mutex> BufferUnlock;
115 typedef bool (*ResizeHandler)(size_t,
void*);
116 typedef std::vector<Block*> BlockList;
135 amps_uint32_t blocksPerRealloc_ = DEFAULT_BLOCKS_PER_REALLOC,
136 amps_uint32_t blockHeaderSize_ = DEFAULT_BLOCK_HEADER_SIZE,
137 amps_uint32_t blockSize_ = DEFAULT_BLOCK_SIZE)
138 : _buffer(buffer_), _freeList(0), _usedList(0)
139 , _endOfUsedList(0), _blocksPerRealloc(blocksPerRealloc_)
140 , _blockSize(blockSize_), _blockHeaderSize(blockHeaderSize_)
141 , _blocksAvailable(0), _resizing(false)
149 for (BlockList::iterator i = _blockList.begin();
150 i != _blockList.end(); ++i)
168 return _blockHeaderSize;
207 return _lock.wait(timeout_);
221 _resizeHandler = resizeHandler_;
222 _resizeUserData = userData_;
241 return _endOfUsedList;
251 _blocksAvailable = freeCount_;
267 _endOfUsedList = block_;
276 _blockList.push_back(blockArray_);
284 Block*
get(amps_uint32_t numBlocksInChain_)
289 while (_blocksAvailable < numBlocksInChain_)
292 unsigned int blocksNeeded = numBlocksInChain_ - _blocksAvailable;
293 amps_uint32_t addedBlocks = (blocksNeeded/_blocksPerRealloc + 1)
295 size_t size = _buffer->getSize() + (addedBlocks * _blockSize);
302 for (
unsigned int i=0; i<numBlocksInChain_; ++i)
306 _freeList = _freeList->_nextInList;
307 next->_nextInList = 0;
317 last->_nextInChain = next;
329 _endOfUsedList->_nextInList = first;
331 _endOfUsedList = first;
332 _blocksAvailable -= numBlocksInChain_;
343 assert(_endOfUsedList);
345 if (_usedList == block_)
348 _usedList = _usedList->_nextInList;
357 Block* used=_usedList;
360 if (used->_nextInList == block_)
362 used->_nextInList = block_->_nextInList;
365 used = used->_nextInList;
373 _flattenToFreeList(block_);
380 ATOMIC_TYPE
put(amps_uint64_t sequence_)
383 assert(_endOfUsedList);
384 Block* used=_usedList;
385 ATOMIC_TYPE removalCount = 0;
386 while (used && used->_sequence <= sequence_)
388 Block* next = used->_nextInList;
390 _flattenToFreeList(used);
409 Block* newEndOfUsedList = 0;
410 for (
Block* used = _usedList; used; used = used->_nextInList)
414 if (newEndOfUsedList)
416 newEndOfUsedList->_nextInList = 0;
422 _endOfUsedList = newEndOfUsedList;
424 newEndOfUsedList = used;
428 for (
Block* block = block_; block; block = next)
430 next = block->_nextInList;
431 _flattenToFreeList(block);
440 size_t startSize = _buffer->getSize();
444 startSize = _buffer->getSize();
447 amps_uint32_t numBlocks = (amps_uint32_t)(startSize) /
getBlockSize();
448 _freeList =
new Block[numBlocks];
449 _blockList.push_back(_freeList);
450 for (
size_t i = 0; i < numBlocks; ++i)
454 _freeList[numBlocks-1]._nextInList = 0;
455 _blocksAvailable += numBlocks;
457 assert(_blocksAvailable);
464 return _blocksPerRealloc * _blockSize;
471 return _blocksPerRealloc;
487 if (_buffer->getSize() >= size_)
return freeList;
488 BufferUnlock guard(_lock);
491 FlagFlip flip(&_resizing);
492 bool okToResize =
false;
495 BufferUnlock u(_lock);
497 okToResize = _canResize(size_);
505 size_t oldSize = _buffer->getSize();
506 amps_uint32_t oldBlocks = (amps_uint32_t)(oldSize /
getBlockSize());
507 if (oldSize >= size_)
512 _buffer->setSize(size_);
513 _buffer->zero(oldSize, size_-oldSize);
515 *pNewBlocks_ = (amps_uint32_t)((size_-oldSize) /
getBlockSize());
516 freeList =
new Block[*pNewBlocks_];
517 for (
size_t i = 0; i < *pNewBlocks_; ++i)
521 freeList[*pNewBlocks_-1]._nextInList = 0;
524 catch(
const std::bad_alloc&)
526 catch(
const std::bad_alloc& e)
529 std::ostringstream os;
530 os <<
"BlockStore failed to allocate " << size_
531 <<
" bytes for resize of store from " << _buffer->getSize()
533 throw StoreException(os.str());
544 amps_uint32_t newBlocks = 0;
546 if (!addedBlockList || !newBlocks)
550 if (_buffer->getSize() >= size_)
return;
552 std::ostringstream os;
553 os <<
"BlockStore failed to allocate " << size_
554 <<
" bytes for resize of buffer from " << _buffer->getSize()
556 throw StoreException(os.str());
558 _blockList.push_back(addedBlockList);
559 addedBlockList[newBlocks-1]._nextInList = _freeList;
560 _freeList = addedBlockList;
561 _blocksAvailable += newBlocks;
563 assert(_blocksAvailable);
571 if (_usedList || _freeList)
575 amps_uint32_t oldSize = _blockSize;
576 _blockSize = blockSize_;
586 if (_usedList || _freeList)
590 amps_uint32_t oldSize = _blockHeaderSize;
591 _blockHeaderSize = blockHeaderSize_;
605 bool _canResize(
size_t requestedSize_)
609 return _resizeHandler(requestedSize_, _resizeUserData);
618 void _flattenToFreeList(
Block* block_)
621 Block* current = block_;
624 Block* chain = current->_nextInChain;
626 _buffer->zero(current->_offset, _blockHeaderSize);
628 current->_nextInList = _freeList;
631 current->_sequence = (amps_uint64_t)0;
632 current->_nextInChain = 0;
636 assert(_blocksAvailable);
646 Block* _endOfUsedList;
648 amps_uint32_t _blocksPerRealloc;
650 amps_uint32_t _blockSize;
651 amps_uint32_t _blockHeaderSize;
653 amps_uint32_t _blocksAvailable;
655 ResizeHandler _resizeHandler;
657 void* _resizeUserData;
659 BlockList _blockList;
661 volatile bool _resizing;
void wait()
Wait for a signal.
Definition: BlockStore.hpp:196
void setResizeHandler(ResizeHandler resizeHandler_, void *userData_)
Set a resize hanlder that is called with the new total size of the Buffer.
Definition: BlockStore.hpp:219
amps_uint32_t setBlockSize(amps_uint32_t blockSize_)
Set the size to use for all Blocks.
Definition: BlockStore.hpp:569
void acquireRead() const
Acquire the lock for this object.
Definition: BlockStore.hpp:175
Constants
Default constant values for BlockStore.
Definition: BlockStore.hpp:63
Buffer * getBuffer()
Return the buffer underlying the store for direct write/read.
Definition: BlockStore.hpp:598
void put(Block *block_)
Return the given chain of Blocks to the free list for reuse.
Definition: BlockStore.hpp:340
~BlockStore()
Destructor that cleans up the buffer and other associated memory.
Definition: BlockStore.hpp:147
void releaseRead() const
Release the lock for this object. Used by RAII templates.
Definition: BlockStore.hpp:182
Block * front() const
Get the first used block in the store.
Definition: BlockStore.hpp:229
amps_uint32_t getDefaultResizeBlocks() const
Return the default number of blocks for each resize.
Definition: BlockStore.hpp:469
void addBlocks(Block *blockArray_)
Allow users to create Block arrays during recovery that are tracked for cleanup here with all other B...
Definition: BlockStore.hpp:274
bool wait(long timeout_)
Wait timeout_ ms for a signal.
Definition: BlockStore.hpp:205
Block * back() const
Get the last used block in the store.
Definition: BlockStore.hpp:239
void resize(size_t size_)
Resize the buffer to the requested size, adding all new space as unused Blocks for the free list...
Definition: BlockStore.hpp:542
ATOMIC_TYPE put(amps_uint64_t sequence_)
Return all Blocks with sequence <= sequence_ for reuse.
Definition: BlockStore.hpp:380
Used as a base class for other stores in the AMPS C++ client, this is an implementation that breaks a...
Definition: BlockStore.hpp:58
void init()
Initialize, assuming that _buffer has no existing information.
Definition: BlockStore.hpp:438
void setEndOfUsedList(Block *block_)
Allow containing classes to initialize the used list in recovery.
Definition: BlockStore.hpp:265
size_t getDefaultResizeSize() const
Return the default number of bytes for each resize.
Definition: BlockStore.hpp:462
void setUsedList(Block *block_)
Allow containing classes to initialize the used list in recovery.
Definition: BlockStore.hpp:257
Core type, function, and class declarations for the AMPS C++ client.
Provides AMPS::Buffer, an abstract base class used by the store implementations in the AMPS client...
amps_uint32_t setBlockHeaderSize(amps_uint32_t blockHeaderSize_)
Set the size to use for the header for all Blocks.
Definition: BlockStore.hpp:584
void setFreeList(Block *block_, amps_uint32_t freeCount_)
Allow containing classes to initialize the free list in recovery.
Definition: BlockStore.hpp:248
amps_uint32_t getBlockHeaderSize() const
Get the size of a header within each Block, as set in the constructor.
Definition: BlockStore.hpp:166
Abstract base class for implementing a buffer to be used by a StoreImpl for storage of publish messag...
Definition: Buffer.hpp:40
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)
Create a BlockStore using buffer_ and default block size, that grows by blocksPerRealloc_ blocks when...
Definition: BlockStore.hpp:134
void putAll(Block *block_)
Return all Blocks starting with the given Block to the free list.
Definition: BlockStore.hpp:406
Used as metadata for each block in a Buffer.
Definition: BlockStore.hpp:72
Block * resizeBuffer(size_t size_, amps_uint32_t *pNewBlocks_)
Resize the buffer to the requested size, returning all new space.
Definition: BlockStore.hpp:482
void signalAll()
Signal lock waiters.
Definition: BlockStore.hpp:189
amps_uint32_t getBlockSize() const
Get the size of each Block, as set in the constructor.
Definition: BlockStore.hpp:159
Definition: ampsplusplus.hpp:136