Time series dataset with temporally enforced object lifetime

Question

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.

Answer 1

Design

The "Single Responsibility Principle" says (paraphrasing) that a class should be responsible for only one thing. TemporalBuffer contains the circular buffer implementation, but also logic for dealing with time-stamps etc.

It would be cleaner to move the circular buffer implementation to a separate class (which would also make it easier to switch out one implementation for another if necessary).

I'd suggest defining your own iterator to go with the circular buffer class, which would allow operations like find, remove etc. using the std:: algorithms.

Code

The Sample comparison operators don't seem to be used anywhere and could be confusing (it seems arbitrary to have the operator compare the timestamp instead of the data). It might be better to just compare timestamps directly where necessary.

AgeLimit, Head and Tail are not useful typedefs. They hide the actual type of the variables, and imply a type difference that doesn't exist (head and tail are the exact same type). The Clock and DataSet declarations are fine. TimePoint and Duration are ok, though they don't save much typing and hide the relationship with the Clock type.

Clean should probably use IncrementIterator.

The member initializer list can be used for all variables in the TemporalBuffer constructor, not just mAgeLimit.

startingBufferSize should be a std::size_t, not an int and if it's set to 0 as an initial value, the program will crash. This needs to at least be checked in code with an assertion.

If I'm not mistaken, AddSample can use IncrementIterator to find the next head position, and compare it with mTail in place of those rather complicated if statements. The "growing" of the container should probably be in a separate function.

IncrementIterator increments the iterator once using std::next, and then again using it++. This could be simplified. It would also be more flexible to model it after std::next and return a value, instead of altering a reference.

GetSamples() and GetRange() should be const. This requires a const version of IncrementIterator that takes a DataSet::const_iterator. Certain variables (auto variables assigned from mHead and mTail in the const functions) may need explicit types, or a call to the DataSet::const_iterator constructor that takes a non-const iterator.

In the for loops in GetSamples and GetRange, the IncrementIterator call can be in the iteration expression in the loop header, rather than in the body of the loop.

GetNearest() should also be const.

Prefer the c++11 facilities in the <random> header to calling rand().