HybridPublishStore.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 _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)
        {
            if (!_lock.wait(1000))
            {
                Unlock<Mutex> u(_lock);
                amps_invoke_waiting_function();
            }
        }
        // Set _holdSwapping true to end of function
        FlagFlip flip(&_holdSwapping);
        {
            Unlock<Mutex> u(_lock);
            if (!index_)
            {
                _memStore.discardUpTo(_fileStore.getLastPersisted());
                Lock<Mutex> l(_lock);
                _lock.signalAll();
                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)
        {
            if (!_lock.wait(1000))
            {
                amps_invoke_waiting_function();
            }
        }
        // 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;
            long waitTime = (timeout_ < 1000) ? timeout_ : 1000;
            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(waitTime))
                {
                    // May have woken up early, check real time
                    AMPS_RESET_TIMER(timedOut, timeout_);
                    waitTime = (timeout_ < 1000) ? timeout_ : 1000;
                    Unlock<Mutex> unlck(_lock);
                    amps_invoke_waiting_function();
                }
            }
            // 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)
        {
            if (!_lock.wait(1000))
            {
                Unlock<Mutex> u(_lock);
                amps_invoke_waiting_function();
            }
        }
        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);
            }
            _lock.signalAll();
            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_