SyncInterrupt_Internal.cpp

 
 
#if defined(RSI_TEST)
#include <condition_variable>
#include <array>
#include <deque>
#include <mutex>
#include <vector>
 
#include "SyncInterrupt.h"
 
 
 
struct WelfordStats
{
  double mean, M2, variance, count;
  void Reset()
  {
    mean = 0;
    M2 = 0;
    variance = 0;
    count = 0;
  }
  // Calculate running mean and variance using...
  // Welford's Algorithm - Courtesy of StackOverflow + Wikipedia
  // https://stackoverflow.com/a/17053010
  WelfordStats UpdateCalculateMeanVariance(const std::vector<double>& vec)
  {
    double& mean = this->mean, & M2 = this->M2, & variance = this->variance;
    double& count = this->count;
    size_t n = vec.size();
    for (size_t i = 0; i < n; ++i, ++count)
    {
      double delta = vec[i] - mean;
      mean += delta / (count + 1);
      M2 += delta * (vec[i] - mean);
      variance = M2 / (count + 1);
      if (i >= 2)
      {
        // <-- You can use the running mean and variance here 
      }
    }
    return *this;
  }
 
  // TODO once we get Welford's working we will probably want something like...
  // A Streaming Parallel Decision Tree Algorithm - Y.Ben-Haim, E. Tom-Tov
  // https://stackoverflow.com/a/73355085
  // points to https://github.com/apache/spark/blob/4c7888dd9159dc203628b0d84f0ee2f90ab4bf13/sql/catalyst/src/main/java/org/apache/spark/sql/util/NumericHistogram.java
};
 
template<typename DataType, size_t _Size>
struct locked_buffer_template
{
  std::mutex mutex;
  std::vector<DataType> data;
  bool full;
  size_t index;
 
  locked_buffer_template() :
    full(false),
    index(0)
  {
    this->data.resize(_Size, 0);
  }
 
  void Reset()
  {
    this->full = false;
    this->index = 0;
  }
 
  // add data
  void AddData(const DataType& datum)
  {
    if (!this->IsFull())
    {
      this->data[this->index++] = datum;
    }
  }
  bool IsFull() const
  {
    return !(this->index < _Size);
  }
};
template<typename DataType, size_t rotating_buffer_count, size_t rotating_buffer_size>
class StatisticsBufferTemplate_RSI : public StatisticsBufferTemplate<DataType> {
public:
  
 
  using locked_buffer = locked_buffer_template<DataType, rotating_buffer_size>;
 
  std::array<locked_buffer, rotating_buffer_count> buffers;
  int buffer_idx;
 
  locked_buffer* buffer_in_use;
 
  WelfordStats running_stat; // should this get locked???
 
  // These get locked together!
  std::mutex mtx_nextData;
  std::condition_variable cv;
  std::deque<locked_buffer*> nextData;
  
  StatisticsBufferTemplate_RSI() :
    buffer_idx(0),
    buffer_in_use(&buffers[0])
  {
  }
 
  // You must call this from your running thread.
  // If you do as I did and let the static initializer call it, the main thread
  // gets the mutex, and calling unlock fails badly
  void Init()
  {
    running_stat.Reset();
    for (auto& buf : this->buffers)
    {
      std::scoped_lock<std::mutex> lock(buf.mutex);
      buf.Reset();
    }
    this->buffer_in_use->mutex.lock(); // start with first buffer locked!
  }
  void Reset()
  {
    buffer_in_use->mutex.unlock();
    this->cv.notify_all();
  }
  void AddData(const DataType& datum)
  {
    this->StatisticsBufferTemplate<DataType>::AddData(datum);
    if (buffer_in_use->IsFull())
    {
      this->RotateBuffer();
    }
 
    this->buffer_in_use->AddData(datum);
  }
 
private:
  // The thread that rotates the buffer "owns" the current buffer to be filled
  // When the current buffer is full, its unlocked, marked as full, and put into the nextData queue
  // The next buffer is then locked and marked as not full
  void RotateBuffer()
  {
    // lock our deque so we can add to it
    std::scoped_lock<std::mutex> lock(this->mtx_nextData);
    this->nextData.emplace_back(buffer_in_use);
 
    this->buffer_idx = (this->buffer_idx + 1) % this->buffers.size();
    auto* next_buffer = &this->buffers[this->buffer_idx];
    // Acquire the lock for the next buffer
    if (!next_buffer->mutex.try_lock())
    {
      // something is wrong. BUT WHAT?!
 
      // resolve or log...
 
      // ... or just wait.
      next_buffer->mutex.lock();
    }
    next_buffer->Reset();
    // next buffer is locked (and loaded) and ready for new data
 
    this->buffer_in_use->full = true;
    this->buffer_in_use->mutex.unlock();
    // current buffer is done and unlocked and unloaded
 
    this->buffer_in_use = next_buffer;
    this->cv.notify_all();
  }
};
using StatisticsBufferImpl = StatisticsBufferTemplate_RSI<double, 10, 10>;
 
 
 
 
 
StatisticsBufferImpl buffered_stats_impl;
StatisticsBuffer& buffered_stats = buffered_stats_impl;
 
 
void StatisticsThread()
{
  StatisticsBufferImpl::locked_buffer* buffer;
  while (!readyToCleanup) // the CV should sleep the thread!
  {
    {
      std::unique_lock<std::mutex> lock(buffered_stats_impl.mtx_nextData);
      buffered_stats_impl.cv.wait(lock, [] { return readyToCleanup || !buffered_stats_impl.nextData.empty(); });
      if (readyToCleanup)
      {
        break;
      }
      buffer = buffered_stats_impl.nextData.front();
      buffered_stats_impl.nextData.pop_front();
 
 
      if (!buffer->mutex.try_lock())
      {
        // if we cant unlock writer thread is still writing...
        // do something?
 
        buffer->mutex.lock();// wait for lock...
      }
    } // unlock buffered_data.mtx_nextData
 
    
    buffered_stats_impl.running_stat.UpdateCalculateMeanVariance(buffer->data);
    buffer->mutex.unlock();
    buffer->Reset();
  }
}
 
void PrintTimingInfo()
{
  // && iterations to wait until we start looping.
  // short circuit on bPrint
  if (print && syncInterruptIterations)
  {
    printf("\t\t%ld\t|\t%8.1lfus\t|\t%8.1lfus\t|\t%8.1lfus\t|\t%8.1lfus\t|\t%8.1lf\r",
      syncInterruptIterations, deltaMicroseconds, buffered_stats_impl.min, buffered_stats_impl.max,
      buffered_stats_impl.running_stat.mean, buffered_stats_impl.running_stat.variance);
  }
}
void printTimingHeaderString()
{
  printf("Number Processed\t|\tDeltaT (uS)\t|\tMin (uS)\t|\tMax (uS)\t|\tMean (uS)\t|\tVariance\n");
}
 
#endif