C++: algorithm that uses fixed-size buffer of data that are produced in stream, faster than the algorithm speed; modified version

Question

This is a, I hope, an improved version of this code, limiting atomic usage (following @GSliepen, @G.Sliepen advice) (NB I'm limited to, at most C++17):

#include <algorithm>
#include <array>
#include <atomic>
#include <cassert>
#include <cmath>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <iostream>
#include <mutex>
#include <thread>

// #define LOG
// #define USEACTIVESLEEP

#ifdef USEACTIVESLEEP
#define MAYBEUNSUSED(var) static_cast<void>(var)
// functions to sleep for short period of times
// active wait but thread sleep has a too large overhead to allow for short
// delays
namespace {
// Iterations per nanosec
double gk_ItPerns;

void EstimateItPerns() noexcept {
    auto start = std::chrono::steady_clock::now();
    constexpr std::size_t NbIt{1000000};
    for (std::size_t i = 0; i < NbIt; ++i) {
        volatile std::size_t DoNotOptimize = 0;
        MAYBEUNSUSED(DoNotOptimize);
    }
    auto end = std::chrono::steady_clock::now();
    auto delay =
        std::chrono::duration_cast<std::chrono::nanoseconds>(end - start)
            .count();
    gk_ItPerns = static_cast<double>(NbIt) / static_cast<double>(delay);
}

void ActiveSleep(double i_ns) noexcept {
    std::size_t NbIt = static_cast<std::size_t>(i_ns * gk_ItPerns);
    for (std::size_t i = 0; i < NbIt; ++i) {
        volatile std::size_t DoNotOptimize = 0;
        MAYBEUNSUSED(DoNotOptimize);
    }
}
}  // namespace
#endif

class CAsyncAlgo {
   public:
    using Data_t = std::size_t;

   private:
    static constexpr std::size_t mNbData = 1024;

    std::size_t mWorkingIndex = 1;
    std::size_t mBufferIndex = 0;

    // type of data buffer
    using DataBuffer_t = std::array<Data_t, mNbData>;

    std::size_t mIndex = 0;
    bool mHasData = false;

    std::array<DataBuffer_t, 2> mSamples;

    // Mutex for condition_variable and atomics
    std::mutex mMutex;
    // Condition variable used to wake up the working thread
    std::condition_variable mWakeUp;
    // To stop the worker
    std::atomic<bool> mStop{false};
    // Is an Algo instance running?
    std::atomic<bool> mBusy{false};
    // 1- Can an Algo instance be launched (for testing spurious wake-up)?
    // not atomic because always accessed inside critical section
    bool mReady{false};

    // working thread
    std::thread Worker;

    // WorkLoad internals
    // previous seen max value in buffer
    Data_t mMaxVal = 0;
    // number of processed data
    Data_t mProcessed = 0;

   private:
    bool Stop() const noexcept {
        // 2- no synch needed?
        return (mStop.load(std::memory_order_relaxed));
    }
    bool Ready() const noexcept { return (mReady); }
    void WaitForJob() {
        std::unique_lock<std::mutex> lock(mMutex);
#ifdef LOG
        std::cout << "entering waiting state " << std::boolalpha << mStop
                  << std::endl;
#endif
        // 3- std::memory_order_relaxed possible as locking mMutex will prevent
        // it to be reordered before the previous Workload call
        mBusy.store(false, std::memory_order_relaxed);
        assert(lock.owns_lock());
        mWakeUp.wait(lock, [this]() -> bool { return (Stop() || Ready()); });
        assert(lock.owns_lock());
        assert(mBusy || Stop());
        mReady = false;
#ifdef LOG
        std::cout << "waked up " << std::this_thread::get_id() << std::endl;
#endif
    }
    // Check if the working buffer is holding increasing successive integers
    // from some point max value must be strictly greater than the one of the
    // previous call {5,6,7,3,4} is valid if previous greatest value is strictly
    // smaller than 7 {5,6,7,2,4} is invalid smallest value must also be
    // strictly greater than mMaxVal as buffers do not overlap
    void WorkLoad() {
        Data_t Max = mSamples[mWorkingIndex][mNbData - 1];
        Data_t Min = mSamples[mWorkingIndex][0];
        for (std::size_t i = 1; i < mNbData; ++i) {
            if (mSamples[mWorkingIndex][i] !=
                mSamples[mWorkingIndex][i - 1] + 1) {
                assert(mSamples[mWorkingIndex][i - 1] ==
                       (mSamples[mWorkingIndex][i] + mNbData - 1));
                Max = mSamples[mWorkingIndex][i - 1];
                Min = mSamples[mWorkingIndex][i];
            }
        }
        assert(Max > mMaxVal);
        assert(Min > mMaxVal);
        mMaxVal = Max;
        mProcessed += mNbData;
    }
    void MainLoop() {
        while (!Stop()) {
            WaitForJob();
            if (Stop()) {
                return;
            }
            WorkLoad();
        }
    }

   public:
    CAsyncAlgo() : Worker([this]() mutable -> void { MainLoop(); }) {}
    void Push(Data_t const Sample, std::size_t) {
        // writing one sample in current circular buffer
        mSamples[mBufferIndex][mIndex] = Sample;
        mIndex = (mIndex + 1) % mNbData;
        if (mIndex == 0) {
            // buffer is full
            mHasData = true;
        }
    }
    bool IsReady() {
        if (mHasData && (mBusy.load(std::memory_order_acquire) == false)) {
            return true;
        }
        return false;
    }

    void SubmitJob() {
#ifdef LOG
        std::cout << "SubmitJob" << std::endl;
#endif
        {
            std::lock_guard<std::mutex> lock(mMutex);
            mReady = true;
            // 4- std::memory_order_relaxed because no synch needed, read only
            // by this thread
            mBusy.store(true, std::memory_order_relaxed);
            std::swap(mWorkingIndex, mBufferIndex);
            mIndex = 0;
            mHasData = false;
        }
        mWakeUp.notify_one();
    }
    void Run(double const, double &) const {
        // NOP
    }

    // destructor
    ///\details finishing computation and releasing resources
    ///\todo explicitely "close" computation before the end of life of the
    /// object
    ~CAsyncAlgo() {
        {
#ifdef LOG
            std::cout << "closing" << std::endl;
#endif
            std::lock_guard<std::mutex> lock(mMutex);
            // 5- std::memory_order_relaxed: on unlocking may synchronise with
            // the lock in wait in this case, the worker will see true
            mStop.store(true, std::memory_order_relaxed);
        }
        mWakeUp.notify_one();
        if (Worker.joinable()) {
#ifdef LOG
            std::cout << "waiting for last run" << std::endl;
#endif
            Worker.join();
#ifdef LOG
            std::cout << "finished" << std::endl;
#endif
            std::cout << "Processed " << GetNbProcessed() << " data"
                      << std::endl;
        }
    }

    std::size_t GetNbProcessed() { return mProcessed; }
};

static constexpr std::size_t NbSamples = 1000000;

int main() {
    CAsyncAlgo Algo;

    std::cout << std::this_thread::get_id() << std::endl;

#ifdef USEACTIVESLEEP
    EstimateItPerns();
    std::size_t acc{0};
#endif
    for (std::size_t i = 0; i < NbSamples; ++i) {
#ifdef USEACTIVESLEEP
        double period = 10000.;  // ns
        // manage data production frequency
        auto start = std::chrono::steady_clock::now();
#endif
        CAsyncAlgo::Data_t data =
            static_cast<CAsyncAlgo::Data_t>(i + 1);  // 0 is reserved
        Algo.Push(data, i);
#ifdef USEACTIVESLEEP
        auto end = std::chrono::steady_clock::now();
        // no more synchro needed as only this thread is designed to launch a
        // new computation
        if (static_cast<double>(
                std::chrono::duration_cast<std::chrono::nanoseconds>(end -
                                                                     start)
                    .count()) < period) {
            ActiveSleep(
                period -
                static_cast<double>(
                    std::chrono::duration_cast<std::chrono::nanoseconds>(end -
                                                                         start)
                        .count()));
        }
        end = std::chrono::steady_clock::now();
        acc = acc + static_cast<std::size_t>(
                        std::chrono::duration_cast<std::chrono::microseconds>(
                            end - start)
                            .count());
#endif
        if (Algo.IsReady()) {
#ifdef LOG
#ifdef USEACTIVESLEEP
            std::cout << "Ready " << i << " "
                      << static_cast<double>(acc) /
                             (static_cast<double>(i) + 1.)
                      << " us/Sample" << std::endl;
#endif
#endif
            Algo.SubmitJob();
        }
        double res;
        Algo.Run(3.14, res);
    }

#ifdef USEACTIVESLEEP
    std::cout << static_cast<double>(acc) / NbSamples << "us/Sample"
              << std::endl;
#endif

    return 0;
}

Live One atomic has been replaced by a simple bool as it was only accessed in critical sections. The other where kept due to access out of critical sections but with less constraint memory order as, once again, when inside such sections, they provides enough synchronization and ordering constraints.

Separating buffers management from workflow would still have to be done.

Yet I'd just like to have feedback on remaining comments (1 to 5)

Answer 1

A large part of my answer to your previous version still applies. I'll revisit some of it below, but first lets start with the changes and the remaining questions.

Can we remove all atomics?

I already mentioned that you should avoid mixing mutexes and atomic variables. That still applies. You have two atomic variables left. Inside WaitForJob() and SubmitJob(), those variables are accessed with the mutex held. There are a few places where these atomic variables are read while not holding a mutex. We can avoid needing those reads by having WaitForJob() return the value of mStop, and by creating a TrySubmitJob() that itself check mReady:

class CAsyncAlgo {
    …
    std::mutex mMutex;
    std::condition_variable mWakeUp;
    bool mStop{false};
    bool mBusy{false};
    …
    bool WaitForJob() {
        std::unique_lock<std::mutex> lock(mMutex);
        mBusy = false;
        mWakeUp.wait(lock, [&]{ return mStop || mReady; });
        mReady = false;
        return !mStop;
    }

    void TrySubmitJob() {
        {
            std::lock_guard<std::mutex> lock(mMutex);
            if (!mHasData || mBusy)
                return;
            mReady = true;
            mBusy = true
            std::swap(mWorkingIndex, mBufferIndex);
            mIndex = 0;
            mHasData = false;
        }
        mWakeUp.notify_one();
    }

    void MainLoop() {
        while (WaitForJob())
            WorkLoad();
    }
    …
};

Note how MainLoop() has simplified a lot. WaitForJob() does less atomic operations now, so it's actually more efficient.

The only issue is that TrySubmitJob() will always lock the mutex, while if (Algo.IsReady()) Algo.SubmitJob() avoided locking the mutex if the consumer was still busy. So if you are going to call this every time you Push() one number to the buffer, then it probably is worth it to keep mBusy as an atomic variable. However, its value quickly diminishes if you do a lot more work between calls to TrySubmitJob().

Questions in the comments

// 1- Can an Algo instance be launched (for testing spurious wake-up)?

I don't know what you mean by this.

// 2- no synch needed?
return (mStop.load(std::memory_order_relaxed));

It's probably fine here because you check it in WaitForJob() with the mutex held as well, and it's only ever set with a held mutex, so that will ensure things will be synchronized. For the two calls of Stop() that are done without the mutex held: the first one is just an unnecessary check (it looks like an optimization, but because it is mostly false for most of the time your program runs, it's actually a pessimization), the second will be guaranteed to see true if WaitForJob() saw that it was true. As mentioned above, it's better if you make this non-atomic.

// 3- std::memory_order_relaxed possible as locking mMutex will prevent
// it to be reordered before the previous Workload call
mBusy.store(false, std::memory_order_relaxed);

This is not true. Locking the mutex is a std::memory_order_acquire action. From cppreference.com:

memory_order_acquire: A load operation with this memory order performs the acquire operation on the affected memory location: no reads or writes in the current thread can be reordered before this load. All writes in other threads that release the same atomic variable are visible in the current thread […]

Note the emphasis. The write to mBusy cannot be reordered before the acquire action on the mutex. But whatever WorkLoad() is doing has nothing to do with the mutex nor mBusy. Reads and writes from WorkLoad() could be reordered to after the locking of the mutex (at least as seen from other threads).

It will still be OK in your code: at worst SubmitJob() is called too early, but since it takes the mutex as well, it will at least not run before WaitForJob() has released its mutex.

// 4- std::memory_order_relaxed because no synch needed, read only
// by this thread
mBusy.store(true, std::memory_order_relaxed);

One of the assert() calls in WaitForJob() reads mBusy, so it's not "read only by this thread" unless you compile with assertions disabled (this might thus be a Heisenbug). Apart from that it's fine.

// 5- std::memory_order_relaxed: on unlocking may synchronise with
// the lock in wait in this case, the worker will see true
mStop.store(true, std::memory_order_relaxed);

See comment 2.

Using atomics correctly is tricky, especially if you want to have the most relaxed memory ordering possible. Your intution might fail you. What works fine on one architecture might not work on another. I think your reasoning is incorrect in some cases, and that might lead to bugs. I personally would stick to using release/acquire semantics, or just purely relying on mutexes instead of atomics, as that is easier to understand.

Separate the data structure from the workers

Again, the class CSyncAlgo is too complicated, making it harder to understand, and making it less flexible. I strongly suggest you split things up. Create a class that just manages the buffers, then you can have two threads that access an object of that class. Consider being able to write:

int main() {
    DoubleBuffer<std::size_t, 1024> dblbuf;

    std::thread producer([&]{
        for (…) {
            …
            dblbuf.Push(data, i);
            dblbuf.TrySubmit();
            …
        }
        dblbuf.Stop();
    });

    std::thread consumer([&]{
        while (dblbuf.WaitForJob()) {
            for (…) {
                …
                auto sample = dblBuf.Pop(i);
                …
            }
        }
    });

    consumer.join();
    producer.join();
}

Consider using at least three buffers for better performance

The way your code works now, it checks for every sample written if it can call SubmitJob(). Even if you do this with an atomic variable, it's still one atomic operation per sample. It would be better if you would not have to do this per-sample check. If you have three buffers, then the producer can write a full buffer, and then swap to another buffer. This way, you only need a check for one full buffer.