circularBufferConcurrent.ipp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2020-2025 European Southern Observatory (ESO)
//
// SPDX-License-Identifier: LGPL-3.0-only
/*
 * circularBufferConcurrent.hpp
 *
 *  Created on: Jul 7, 2021
 *      Author: hsommer
 */
 
#include <mutex>
#include <condition_variable>
#include <functional>
 
#include <boost/circular_buffer.hpp>
 
 
// HSO TODO: Check async calls in ProcessRequest in ~/git-rtc/rtctk/componentFramework/services/oldb/src/oldbAdapter.cpp
 
 
namespace hlcc::oldbmux {
 
template<typename T>
CircularBufferConcurrent<T>::CircularBufferConcurrent(
        const std::string& name, std::size_t capacity, const log4cplus::Logger& logger)
    : m_logger {logger},
      m_closed {false},
      cb {capacity},
      mutex {},
      buffer_not_empty {},
      discard_listener {},
      name {name}
{
    if (capacity < 1) {
        // TODO what to throw?    Java has throw new IllegalArgumentException("bufferCapacity must be >= 1.");
    }
}
 
 
template<typename T>
CircularBufferConcurrent<T>::~CircularBufferConcurrent() {
    RAD_TRACE(m_logger);
}
 
 
template<typename T>
void CircularBufferConcurrent<T>::Close() {
    RAD_TRACE(m_logger);
 
    // we better close right away
    bool expected = false; // TODO: how to pass bool& as a 'false' literal, without declaring the variable?
    // if m_closed == false then set it to true and execute the code block that closes the buffer.
    if (m_closed.compare_exchange_strong(expected, true)) {  
 
        // release waiting thread if Poll was called with a timeout
        buffer_not_empty.notify_all();
 
        /*
         * To speed up destruction, we discard from the buffer all but the last data.
         * For discarded data, we call the discard_listener.
         * 
         * Clients such as CiiOldbDataPointAsync may decrement a counter for ongoing write operations
         * in their discard callback functions, to block their dtors until all data is written or discarded.
         * There is a race condition with the background worker thread of CiiOldbDataPointAsync, in which 
         * the thread may be scheduled before Close() gets called, but runs only afterwards. We leave the last data item
         * for such a thread to process and decrement the counter, to make sure that the thread does not wake up 
         * using the invalid memory location of an already destructed CiiOldbDataPointAsync instance. 
         * If not discarding one data item becomes a performance problem during shutdown, then we should discard 
         * all queued data, but put back a special sentinel data that the background thread can clear in no time, 
         * but which nonetheless keeps the writer count up until the thread is done. 
         */
        std::lock_guard lck {mutex};
        while (cb.size() > 1) {
            std::optional<T> data_opt = Poll();
            if (data_opt && discard_listener) {
                discard_listener(*data_opt);
            }
        }
    }
}
 
template<typename T>
void CircularBufferConcurrent<T>::Push(T&& data) {
    if (!m_closed) { 
        { // lock scope
            std::lock_guard lck {mutex};
 
            std::optional<T> data_old {};
 
            if (cb.full() && discard_listener) {
                data_old = std::move(cb.front()); // saves a copy
            }
 
            cb.push_back(std::move(data)); // discards the front data if buffer is full (i.e., assigns the new data to the old element).
            
            if (data_old && discard_listener) {
                discard_listener(*data_old);
            }
 
            // release lck before notifying the condition variable. 
            // If the thread waiting in Poll wakes immediately, it doesn't then 
            // have to block again, waiting for us to unlock the mutex.
        } 
        buffer_not_empty.notify_one();
    } else {
        // We call discard_listener directly for data given to Push after a Close call, without using cb.
        if (discard_listener) {
            discard_listener(data);
        }
    }
}
 
 
template<typename T>
std::optional<T> CircularBufferConcurrent<T>::Poll(std::chrono::milliseconds timeout) {
 
    // Note that it is important for this method to work even when m_closed==true.
    // As explained in Poll, the UC is for CiiOldbDataPointAsync, whose worker thread must process 
    // the last buffered item.
 
    // condition variable needs to be able to lock and unlock the mutex.
    // Our normal lock_guard doesn't allow this, thus we use unique_lock.
    std::unique_lock lck {mutex};
    // wait_for with predicate keeps looping after spurious wakeups.
    // It yields the lock and re-acquires it after wakeup, see https://en.cppreference.com/w/cpp/thread/condition_variable/wait_for
    if (buffer_not_empty.wait_for(lck, timeout, [this](){return (!cb.empty() || m_closed);})) {
        // no timeout but condition met: Either got data, or got notified by Close.
        // If it was Close, 
        //   - then perhaps Close has already emptied the buffer and notified discard_listener. 
        //   - and if there is still data, then we return that last data item. The others will be removed by Close right after this.
        if (!cb.empty()) {
            T data = std::move(cb.front());
            cb.pop_front();
            return std::make_optional<T>(std::move(data));
        } 
    } 
    // got no data because: 
    // - wait_for returned false for !cb.empty() (i.e., empty buffer) or
    // - wait_for returned true for m_closed and there was no data in the buffer or
    // - wait_for timed out (if timeout > 0) and thus predicate value was still false.
    return {};
    // RAII unlocks mutex
}
 
 
template<typename T>
void CircularBufferConcurrent<T>::Clear() {
    std::lock_guard lck {mutex};
    cb.clear();
}
 
 
template<typename T>
int CircularBufferConcurrent<T>::Size() const {
    std::lock_guard lck {mutex};
    return cb.size();
}
 
 
template<typename T>
const std::unique_ptr<std::scoped_lock<std::recursive_mutex>> CircularBufferConcurrent<T>::Lock() const {
    return std::make_unique<std::scoped_lock<std::recursive_mutex>>(mutex);
}
 
 
template<typename T>
void CircularBufferConcurrent<T>::SetDiscardListener(std::function<void(T&)> discard_listener) {
    std::lock_guard lck {mutex};
    this->discard_listener = discard_listener;
}
 
template<typename T>
boost::circular_buffer<T>& CircularBufferConcurrent<T>::GetCb() {
    return cb;
}
 
 
} // end namespace hlcc::oldbmux