The purpose of this class is to have that can maintain data set where the samples contained within the set have an expiry time. Samples are inserted in chronological order. In an attempt to memory consumption, allocation and deallocations a circular buffer has been implemented.
template<typename T>
class TemporalBuffer
{
public:
using Clock = std::chrono::steady_clock;
using TimePoint = Clock::time_point;
using Duration = Clock::duration;
struct Sample
{
TimePoint timestamp{};
T data{};
friend bool operator<(const Sample& lhs, const Sample& rhs)
{
return lhs.timestamp < rhs.timestamp;
}
friend bool operator<=(const Sample& lhs, const Sample& rhs)
{
return !(rhs < lhs);
}
friend bool operator>(const Sample& lhs, const Sample& rhs)
{
return rhs < lhs;
}
friend bool operator>=(const Sample& lhs, const Sample& rhs)
{
return !(lhs < rhs);
}
};
using DataSet = std::vector<Sample>;
using Head = typename DataSet::iterator;
using Tail = typename DataSet::iterator;
using AgeLimit = Duration;
private:
AgeLimit mAgeLimit;
DataSet mContainer;
Head mHead;
Tail mTail;
long long GetTimeDelta(TimePoint point1, TimePoint point2) const
{
return point1 > point2 ?
std::chrono::duration_cast<std::chrono::microseconds>(point1 - point2).count() :
std::chrono::duration_cast<std::chrono::microseconds>(point2 - point1).count();
}
void Clean()
{
auto expiryTimestamp = Clock::now() - mAgeLimit;
while (mTail->timestamp < expiryTimestamp) {
if (std::next(mTail) == std::end(mContainer)) {
mTail = std::begin(mContainer);
}
else {
++mTail;
}
}
}
public:
explicit TemporalBuffer(AgeLimit ageLimit, int startingBufferSize = 32)
: mAgeLimit(ageLimit)
{
mContainer = DataSet(startingBufferSize);
mHead = std::begin(mContainer);
mTail = std::begin(mContainer);
};
~TemporalBuffer() = default;
void AddSample(T&& sampleData)
{
if (std::next(mHead) == std::end(mContainer) && mTail != std::begin(mContainer)) {
mHead = std::begin(mContainer);
}
else if (std::next(mHead) == mTail || (std::next(mHead) == std::end(mContainer) && mTail == std::begin(mContainer))) { //Our vector is not big enough
auto headLocation = mHead - std::begin(mContainer);
auto it = std::rotate(mContainer.rbegin(), mContainer.rbegin() + (mContainer.end() - mTail), mContainer.rend());
auto headIncremented = mContainer.size() - (std::rend(mContainer) - it);
mContainer.resize(mContainer.size() * 2);
mTail = std::begin(mContainer);
mHead = std::begin(mContainer) + headLocation + headIncremented;
}
(*mHead).timestamp = Clock::now();
(*mHead).data = std::move(sampleData);
++mHead;
Clean();
}
void IncrementIterator(typename DataSet::iterator& it)
{
if (std::next(it) == std::end(mContainer)) {
it = std::begin(mContainer);
}
else {
++it;
}
}
std::vector<T> GetSamples()
{
std::vector<T> samples;
samples.reserve(Size());
for (auto it = mTail; it != mHead;) {
samples.emplace_back(it->data);
IncrementIterator(it);
}
return samples;
}
DataSet GetRange(TimePoint start, TimePoint end)
{
DataSet sampleRange{};
sampleRange.reserve(Size());
for (auto it = mTail; it != mHead;) {
if (it->timestamp >= start && it->timestamp <= end) {
sampleRange.emplace_back(*it);
}
IncrementIterator(it);
}
return sampleRange;
}
const DataSet& GetDataSet() const
{
return mContainer;
}
typename DataSet::size_type Size() const
{
return mTail <= mHead ? mHead - mTail : mContainer.size() - (mTail - mHead);
}
boost::optional<Sample> GetNearest(TimePoint timestamp)
{
if (mContainer.empty() || mTail == mHead) {
return boost::none;
}
if (timestamp <= mTail->timestamp) {
return *mTail;
}
auto olderIt = std::end(mContainer);
auto newerIt = mTail;
while (newerIt != mHead) {
if (newerIt->timestamp > timestamp) {
olderIt = newerIt == mContainer.begin() ? std::prev(mContainer.end()) : std::prev(newerIt);
break;
}
IncrementIterator(newerIt);
}
if (olderIt == std::end(mContainer) || newerIt == std::end(mContainer)) {
return boost::none;
}
auto timeToOlder = GetTimeDelta(olderIt->timestamp, timestamp);
auto timeToNewer = GetTimeDelta(newerIt->timestamp, timestamp);
return timeToOlder <= timeToNewer ? *olderIt : *newerIt;
}
};
The following is a simple example of this code.
#include <iostream>
#include <thread>
#include "TemporalBuffer.h"
int main()
{
TemporalBuffer<double> timeSeries(std::chrono::seconds{ 60 });
auto lastStatistics = std::chrono::steady_clock::now();
while (true) {
timeSeries.AddSample(static_cast<double>(std::rand() % 100));
auto now = std::chrono::steady_clock::now();
if (now - lastStatistics > std::chrono::seconds{ 1 }) {
lastStatistics = now;
auto dataSet = timeSeries.GetSamples();
auto end = std::chrono::steady_clock::now();
auto delta = std::chrono::milliseconds{ rand() % 30000 };
auto lookupTime = now - delta;
auto optional = timeSeries.GetNearest(lookupTime);
auto startTime = now - std::chrono::milliseconds{ 2000 + rand() % 5000 };
auto endTime = now - std::chrono::milliseconds{ 2000 - rand() % 2000 };
auto sampleRange = timeSeries.GetRange(startTime, endTime);
std::cout << "Size: " << dataSet.size() << ". Dataset population time: " << std::chrono::duration_cast<std::chrono::microseconds>(end - now).count() << "us. ";
std::cout << "Sample size is " << sampleRange.size() << ", for a " << std::chrono::duration_cast<std::chrono::milliseconds>(endTime-startTime).count() << "ms window. ";
if (optional) {
std::cout << delta.count() << "ms ago, the value was: " << optional.value().data << ".";
}
std::cout << '\n';
}
std::this_thread::sleep_for(std::chrono::milliseconds{ 100 });
}
}
My initial implementation used a std::map container. I was pleasantly surprised at the performance gain in moving to vector (~50-60x faster). If you have a view on this, please provide feedback/critism and if you know of any implementations similar to this, please provide a link as I would like to see how others have done this - however I didn't find an solutions.