RecoveryPointAdapter.hpp

//
// Copyright (c) 2010-2020 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 _RECOVERYPOINTADAPTER_H_
#define _RECOVERYPOINTADAPTER_H_

#include <Field.hpp>
#include <RecoveryPoint.hpp>
#include <deque>
#include <iterator>
#include <memory>
#include <thread>
#include <unordered_map>

namespace AMPS
{

class RecoveryPointAdapter;

class RecoveryPointAdapterImpl : public RefBody
{
public:
    virtual ~RecoveryPointAdapterImpl() { }
    virtual bool next(RecoveryPoint& current_) = 0;
    virtual void update(RecoveryPoint& recoveryPoint_) = 0;
    virtual void purge() = 0;
    virtual void purge(const Field& subId_) = 0;
    virtual void close() = 0;
    virtual void prune() { ; }
};

class RecoveryPointAdapter
{
public:
    class iterator
    {
        RecoveryPointAdapterImpl* _pAdapter;
        RecoveryPoint _current;
        inline void advance()
        {
            if (!_pAdapter || !_pAdapter->next(_current))
            {
                _pAdapter = NULL;
            }
        }

    public:
        iterator() // end
        :_pAdapter(NULL)
        {;}
        iterator(RecoveryPointAdapterImpl* pAdapter_)
        :_pAdapter(pAdapter_)
        {
            advance();
        }

        bool operator==(const iterator& rhs)
        {
            return _pAdapter == rhs._pAdapter;
        }
        bool operator!=(const iterator& rhs)
        {
            return _pAdapter != rhs._pAdapter;
        }
        void operator++(void) { advance(); }
        RecoveryPoint operator*(void) { return _current; }
        RecoveryPoint* operator->(void) { return &_current; }
    };

    RecoveryPointAdapter() : _body() { }
    RecoveryPointAdapter(RecoveryPointAdapterImpl* body_, bool isRef_ = true)
    : _body(body_, isRef_) { }
    RecoveryPointAdapter(const RecoveryPointAdapter& rhs_)
    : _body(rhs_._body)
    { }

    iterator begin() { return iterator(&(_body.get())); }

    iterator end() { return iterator(); }

    void update(RecoveryPoint& recoveryPoint_)
    {
        _body.get().update(recoveryPoint_);
    }

    void purge() { _body.get().purge(); }

    void purge(const Field& subId_) { _body.get().purge(subId_); }

    void close() { _body.get().close(); }

    void prune() { _body.get().prune(); }

    bool isValid() const
    {
        return _body.isValid();
    }
private:
    BorrowRefHandle<RecoveryPointAdapterImpl> _body;
};

class ConflatingRecoveryPointAdapter : public RecoveryPointAdapterImpl
{
public:
    ConflatingRecoveryPointAdapter(
                     const std::shared_ptr<RecoveryPointAdapterImpl>& delegate_,
                     unsigned updateThreshold_ = 10,
                     double timeoutMillis_ = 2000.0,
                     long updateIntervalMillis_ = 2000
                     )
    : _closed(false)
    , _delegate(delegate_)
    , _updateThreshold(updateThreshold_)
    , _timeoutMillis(timeoutMillis_)
    , _updateIntervalMillis(updateIntervalMillis_)
    , _updateAll(false)
    {
        // Start the update thread
        _thread = std::thread(&ConflatingRecoveryPointAdapter::updateThread,
                              this);
    }

    virtual ~ConflatingRecoveryPointAdapter()
    {
        close();
        _thread.join();
        for (UpdateIter purged = _latestUpdates.begin();
             purged != _latestUpdates.end(); ++purged)
        {
            Field clearableSubId = purged->first;
            purged->second.clear();
            clearableSubId.clear();
        }
    }

    virtual bool next(RecoveryPoint& current_)
    {
        return _delegate->next(current_);
    }

    virtual void update(RecoveryPoint& recoveryPoint_)
    {
        Field subId = recoveryPoint_.getSubId();
        Lock<Mutex> lock(_lock);
        UpdateIter lastUpdate = _latestUpdates.find(subId);
        if (lastUpdate == _latestUpdates.end())
        {
            // New sub id, use deep copies and a new Timer.
            subId = subId.deepCopy();
            _latestUpdates[subId] = recoveryPoint_.deepCopy();
            _counts[subId] = 1;
            if (_timeoutMillis != 0.0)
            {
                Timer timer(_timeoutMillis);
                timer.start();
                _timers[subId] = timer;
            }
        }
        else
        {
            // SubId already exists, set to new recovery point.
            lastUpdate->second.deepCopy(recoveryPoint_);
            // Increment and check the count.
            if (++_counts[subId] >= _updateThreshold)
            {
                // Time to update, make sure update thread wakes up.
                _lock.signalAll();
            }
        }
    }

    virtual void purge()
    {
        _delegate->purge();
        Lock<Mutex> lock(_lock);
        _counts.clear();
        _timers.clear();
        for (UpdateIter purged = _latestUpdates.begin();
             purged != _latestUpdates.end(); ++purged)
        {
            Field clearableSubId = purged->first;
            purged->second.clear();
            clearableSubId.clear();
        }
        _latestUpdates.clear();
    }

    virtual void purge(const Field& subId_)
    {
        _delegate->purge(subId_);
        Lock<Mutex> lock(_lock);
        UpdateIter purged = _latestUpdates.find(subId_);
        if (purged != _latestUpdates.end())
        {
            Field clearableSubId = purged->first;
            purged->second.clear();
            _latestUpdates.erase(purged);
            _counts.erase(subId_);
            _timers.erase(subId_);
            clearableSubId.clear();
        }
    }

    virtual void close()
    {
        // Save all cached updates before shutting down update thread.
        if (!_closed)
        {
            Lock<Mutex> lock(_lock);
            _runUpdateAll();
            _closed = true;
            _lock.signalAll();
        }
        _delegate->close();
    }

    virtual void updateAll()
    {
        Lock<Mutex> lock(_lock);
        _runUpdateAll();
    }

    virtual void _runUpdateAll()
    {
        _updateAll = true;
        while (!_counts.empty())
        {
            _lock.signalAll();
            _lock.wait(250);
        }
    }

protected:
    void updateThread()
    {
        // A place to hold updates to save
        std::deque<Update>        _queuedUpdates;
        while (!_closed)
        {
            DeferLock<Mutex> lock(_lock);
            lock.lock();
            // Wait for a signal or update interval
            _lock.wait(_updateIntervalMillis);

            // Check for timeouts
            for (TimerMap::iterator timer = _timers.begin();
                 timer != _timers.end(); )
            {
                if (timer->second.check())
                {
                    UpdateIter update = _latestUpdates.find(timer->first);
                    if (update != _latestUpdates.end())
                    {
                        // Remove subId from all, clear of subId will
                        // occur after save.
                        _queuedUpdates.push_back(*update);
                        _counts.erase(update->first);
                        timer = _timers.erase(timer);
                        _latestUpdates.erase(update);
                    }
                    else
                    {
                        timer++;
                    }
                }
                else
                {
                    timer++;
                }
            }

            // Need a local version so it doesn't change after we unlock to
            // deliver updates.
            volatile bool updateAll = _updateAll;
            // Check for update counts
            for (CountMap::iterator count = _counts.begin();
                 count != _counts.end(); )
            {
                if (updateAll || _timeoutMillis == 0.0
                    || count->second >= _updateThreshold)
                {
                    UpdateIter update = _latestUpdates.find(count->first);
                    if (update != _latestUpdates.end())
                    {
                        // Remove subId from all, clear of subId will
                        // occur after save.
                        _queuedUpdates.push_back(*update);
                        count = _counts.erase(count);
                        _timers.erase(update->first);
                        _latestUpdates.erase(update);
                    }
                    else
                    {
                        count++;
                    }
                }
                else
                {
                    count++;
                }
            }
            // Release the lock unless we're doing an update all, then we
            // hold it until updates are completed and signal when done.
            if (!updateAll)
            {
                lock.unlock();
            }
            // Shouldn't need the lock to send the updates
            for (std::deque<Update>::iterator update = _queuedUpdates.begin();
                 update != _queuedUpdates.end(); ++update)
            {
                _delegate->update(update->second);
                Field clearableSubId(update->first);
                clearableSubId.clear();
                update->second.clear();
            }
            _queuedUpdates.clear();
            if (updateAll)
            {
                _updateAll = false;
                _lock.signalAll();
            }
        }
    }

    // The update thread runs until this is true.
    volatile bool                             _closed;

    // The adapter doing the real saves
    std::shared_ptr<RecoveryPointAdapterImpl> _delegate;

    // Lock used to protect _latestUpdates, _timers, and _counts.
    Mutex                                     _lock;

    // Types for our maps
    typedef std::unordered_map<Field, RecoveryPoint, Field::FieldHash> UpdateMap;
    typedef UpdateMap::value_type                                      Update;
    typedef UpdateMap::iterator                                        UpdateIter;
    typedef std::unordered_map<Field, Timer, Field::FieldHash>         TimerMap;
    typedef TimerMap::iterator                                         TimerIter;
    typedef std::unordered_map<Field, unsigned, Field::FieldHash>      CountMap;
    typedef CountMap::iterator                                         CountIter;

    // Saves the most recent update for each sub id.
    UpdateMap                                 _latestUpdates;

    // Saves a timer for each sub id that is reset each time we save.
    TimerMap                                  _timers;

    // Saves a count of how many updates have come in since last save.
    CountMap                                  _counts;

    // The thread doind the saves.
    std::thread                               _thread;

    // How many updates before we force a save.
    unsigned                                  _updateThreshold;

    // How long between getting first cached update and save.
    double                                    _timeoutMillis;

    // How long between automatic checks of the timers.
    long                                      _updateIntervalMillis;

    // Flag to tell update thread to save everything.
    volatile bool                             _updateAll;
};

}

#endif //_RECOVERYPOINTADAPTER_H_