RecoveryPointAdapter.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2025 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 <amps/Field.hpp>
#include <amps/RecoveryPoint.hpp>
#include <vector>
#include <iterator>
#include <memory>
#include <thread>
#include <unordered_map>
#if __cplusplus >= 201100L || _MSC_VER >= 1900
  #include <atomic>
#endif


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 // -V690
  {
  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) const
      {
        return _pAdapter == rhs._pAdapter;
      }
      bool operator!=(const iterator& rhs) const
      {
        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
    )
      : _delegate(delegate_)
      , _updateThreshold(updateThreshold_)
      , _timeoutMillis(timeoutMillis_)
      , _updateIntervalMillis(updateIntervalMillis_)
      , _closed(false)
      , _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_)
    {
      if (_closed)
      {
        return;
      }
      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) // -V550
        {
          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()
    {
      if (_closed)
      {
        return;
      }
      _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_)
    {
      if (_closed)
      {
        return;
      }
      _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()
    {
      _close();
    }

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

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

  protected:
    void _close()
    {
      // Save all cached updates before shutting down update thread.
      if (!_closed)
      {
        Lock<Mutex> lock(_lock);
        _runUpdateAll();
        _closed = true;
        _lock.signalAll();
      }
      _delegate->close();
    }
    void updateThread()
    {
      // A place to hold updates to save
      std::vector<SavedUpdate> _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.
        bool updateAll = (bool)_updateAll;
        // Check for update counts
        for (CountMap::iterator count = _counts.begin();
             count != _counts.end(); )
        {
          if (updateAll || _timeoutMillis == 0.0 // -V550
              || 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::vector<SavedUpdate>::iterator update = _queuedUpdates.begin(), end = _queuedUpdates.end(); update != end; ++update)
        {
          _delegate->update(update->second);
          Field clearableSubId(update->first);
          clearableSubId.clear();
          update->second.clear();
        }
        _queuedUpdates.clear();
        if (updateAll)
        {
          _updateAll = false;
          _lock.signalAll();
        }
      } // -V1020
    }

    // 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 std::pair<Field, RecoveryPoint>                       SavedUpdate;
    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 doing 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;

#if __cplusplus >= 201100L || _MSC_VER >= 1900
    // The update thread runs until this is true.
    std::atomic<bool>                         _closed;

    // Flag to tell update thread to save everything.
    std::atomic<bool>                         _updateAll;
#else
    // The update thread runs until this is true.
    volatile bool                             _closed;

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

}

#endif //_RECOVERYPOINTADAPTER_H_