HybridPublishStore.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 _HYBRIDPUBLISHSTORE_H_
#define _HYBRIDPUBLISHSTORE_H_

#include <ampsplusplus.hpp>
#include <MemoryPublishStore.hpp>
#include <PublishStore.hpp>


namespace AMPS
{
class HybridPublishStore : public StoreImpl
{
 class HandlerData
 {
 public:
 HybridPublishStore* _store;
 PublishStoreResizeHandler _handler;
 void* _data;

 HandlerData()
 : _store(NULL), _handler(NULL), _data(NULL)
 { ; }

 void init(PublishStoreResizeHandler handler_, void* data_)
 {
 _handler = handler_;
 _data = data_;
 }
 };

 class SwappingOutReplayer : public StoreReplayer
 {
 PublishStore* _pStore;
 volatile size_t _entries, _errorCount;
 volatile amps_uint64_t _lastIndex;
 public:
 SwappingOutReplayer(PublishStore* pStore_, size_t entries_)
 : _pStore(pStore_), _entries(entries_)
 , _errorCount(0), _lastIndex(0)
 { }

 size_t getErrors() { return _errorCount; }

 amps_uint64_t lastIndex() { return _lastIndex; }

 void execute(Message& message_)
 {
 if(_entries > 0 && _errorCount == 0)
 {
 try
 {
 {
 _pStore->store(message_, false);
 }
 _lastIndex = amps_message_get_field_uint64(message_.getMessage(),
 AMPS_Sequence);
 }
 catch(...)
 {
 ++_errorCount;
 }
 --_entries;
 }
 }
 };

public:
 HybridPublishStore(const char* fileName_, size_t maxMemoryCapacity_)
 : StoreImpl(), _memStore(maxMemoryCapacity_), _fileStore(fileName_),
 _cap(maxMemoryCapacity_),
 _lowWatermark((size_t)((double)maxMemoryCapacity_ * 0.5)),
 _lowestIndexInMemory(0), _holdSwapping(false)
 {
 _handlerData._store = this;
 _memStore.addRef();
 _fileStore.addRef();
 }

 HybridPublishStore(const std::string& fileName_, size_t maxMemoryCapacity_)
 : StoreImpl(), _memStore(maxMemoryCapacity_), _fileStore(fileName_),
 _cap(maxMemoryCapacity_),
 _lowWatermark((size_t)((double)maxMemoryCapacity_ * 0.5)),
 _lowestIndexInMemory(0), _holdSwapping(false)
 {
 _handlerData._store = this;
 _memStore.addRef();
 _fileStore.addRef();
 }

 void setLowWatermark(size_t lowWatermark_)
 {
 Lock<Mutex> guard(_lock);
 _lowWatermark = lowWatermark_;
 }

 size_t getLowWatermark()
 {
 Lock<Mutex> guard(_lock);
 return _lowWatermark;
 }

 void discardUpTo(amps_uint64_t index_)
 {
 Lock<Mutex> guard(_lock);
 while (_holdSwapping)
 {
 Unlock<Mutex> u(_lock);
 AMPS_YIELD();
 }
 // Set _holdSwapping true to end of function
 FlagFlip flip(&_holdSwapping);
 Unlock<Mutex> u(_lock);
 if (!index_)
 {
 _memStore.discardUpTo(_fileStore.getLastPersisted());
 return;
 }
 _fileStore.discardUpTo(index_);
 if (_lowestIndexInMemory <= index_)
 {
 _memStore.discardUpTo(index_);
 _lowestIndexInMemory = index_ + 1;
 }
 _lock.signalAll();
 }

 void replay(StoreReplayer& replayer_)
 {
 Lock<Mutex> guard(_lock);
 while (_holdSwapping)
 {
 Unlock<Mutex> u(_lock);
 AMPS_YIELD();
 }
 // Set _holdSwapping true to end of function
 FlagFlip flip(&_holdSwapping);
 Unlock<Mutex> u(_lock);
 _fileStore.replay(replayer_);
 _memStore.replay(replayer_);
 _lock.signalAll();
 }

 size_t unpersistedCount() const
 {
 return _fileStore.unpersistedCount() + _memStore.unpersistedCount();
 }

 virtual void flush(long timeout_)
 {
 Lock<Mutex> guard(_lock);
 amps_uint64_t waitFor = _getHybridHighestUnpersisted();
 amps_uint64_t unset = getUnsetSequence();
 // Check that we aren't already empty
 if (waitFor == unset) return;
 if (timeout_ > 0)
 {
 bool timedOut = false;
 AMPS_START_TIMER(timeout_)
 // While timeout hasn't expired and we haven't had everything acked
 while (!timedOut && waitFor >= _getHybridLowestUnpersisted() &&
 _getHybridLowestUnpersisted() != unset)
 {
 if (!_lock.wait(timeout_))
 {
 // May have woken up early, check real time
 AMPS_RESET_TIMER(timedOut, timeout_);
 }
 }
 // If we timed out and still haven't caught up with the acks
 if (timedOut && waitFor >= _getHybridLowestUnpersisted() &&
 _getHybridLowestUnpersisted() != unset)
 {
 throw TimedOutException("Timed out waiting to flush publish store.");
 }
 }
 else
 {
 while (waitFor >= _getHybridLowestUnpersisted() &&
 _getHybridLowestUnpersisted() != unset)
 {
 // Use timeout version so python can interrupt
 _lock.wait(1000);
 Unlock<Mutex> unlck(_lock);
 amps_invoke_waiting_function();
 }
 }
 }

 bool replaySingle(StoreReplayer& replayer_, amps_uint64_t index_)
 {
 amps_uint64_t lowestIndexInMemory;
 {
 Lock<Mutex> guard(_lock);
 lowestIndexInMemory = _lowestIndexInMemory;
 }
 if(index_ < lowestIndexInMemory)
 {
 return _fileStore.replaySingle(replayer_, index_);
 }
 else
 {
 return _memStore.replaySingle(replayer_, index_);
 }
 }

 amps_uint64_t store(const Message& message_)
 {
 Lock<Mutex> guard(_lock);
 //while (_holdSwapping && _memStore.unpersistedCount() >= _cap)
 while (_holdSwapping)
 {
 Unlock<Mutex> u(_lock);
 AMPS_YIELD();
 }
 if(_memStore.unpersistedCount() >= _cap && !_holdSwapping)
 {
 // Set _holdSwapping true to end of function
 FlagFlip flip(&_holdSwapping);
 SwappingOutReplayer swapper(&_fileStore,
 _memStore.unpersistedCount() - _lowWatermark);
 {
 Unlock<Mutex> u(_lock);
 _memStore.replay(swapper);
 }
 if(swapper.getErrors() == 0)
 {
 _lowestIndexInMemory = swapper.lastIndex();
 _memStore.discardUpTo(_lowestIndexInMemory++);
 }
 }
 return _memStore.store(message_);
 }

 void setResizeHandler(PublishStoreResizeHandler handler_, void* data_)
 {
 _handlerData.init(handler_, data_);
 _fileStore.setResizeHandler(HybridPublishStore::resizeHandler,
 (void*)&_handlerData);
 }

 inline virtual PublishStoreResizeHandler getResizeHandler() const
 {
 return _handlerData._handler;
 }

 amps_uint64_t getLowestUnpersisted() const
 {
 Lock<Mutex> guard(_lock);
 return _getHybridLowestUnpersisted();
 }

 amps_uint64_t getHighestUnpersisted() const
 {
 Lock<Mutex> guard(_lock);
 return _getHybridHighestUnpersisted();
 }

 amps_uint64_t getLastPersisted(void)
 {
 Lock<Mutex> guard(_lock);
 return _getHybridLastPersisted();
 }

private:

 static bool resizeHandler(Store store_, size_t size_, void* data_)
 {
 HandlerData* handlerData = (HandlerData*)data_;
 //Unlock<Mutex> hybridUnlock(handlerData->_store->_lock);
 return handlerData->_handler(store_, size_, handlerData->_data);
 }

 // Lock should be held
 amps_uint64_t _getHybridLowestUnpersisted() const
 {
 amps_uint64_t filemin = _fileStore.getLowestUnpersisted();
 amps_uint64_t memmin = _memStore.getLowestUnpersisted();
 if (filemin == AMPS_UNSET_SEQUENCE)
 return memmin;
 if (memmin == AMPS_UNSET_SEQUENCE || memmin > filemin)
 return filemin;
 // Only left with memmin <= filemin
 return memmin;
 }

 // Lock should be held
 amps_uint64_t _getHybridHighestUnpersisted() const
 {
 amps_uint64_t filemax = _fileStore.getHighestUnpersisted();
 amps_uint64_t memmax = _memStore.getHighestUnpersisted();
 if (filemax == AMPS_UNSET_SEQUENCE)
 return memmax;
 if (memmax == AMPS_UNSET_SEQUENCE || memmax < filemax)
 return filemax;
 // Only left with memmax >= filemax
 return memmax;
 }

 amps_uint64_t _getHybridLastPersisted()
 {
 // If we've never swapped and nothing is in file
 if (!_lowestIndexInMemory &&
 _fileStore.unpersistedCount() == 0)
 {
 _fileStore.discardUpTo(_memStore.getLastPersisted());
 return _fileStore.getLastPersisted();
 }
 amps_uint64_t memLast = _memStore.getLastPersisted();
 amps_uint64_t fileLast = _fileStore.getLastPersisted();
 return (memLast < fileLast) ? memLast : fileLast;
 }

 MemoryPublishStore _memStore;
 PublishStore _fileStore;
 size_t _cap;
 size_t _lowWatermark;
 amps_uint64_t _lowestIndexInMemory;
 mutable Mutex _lock;
 HandlerData _handlerData;
 volatile bool _holdSwapping;

};//end HybridPublishStore

}//end namespace AMPS

#endif //_HYBRIDPUBLISHSTORE_H_