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},
      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>::Push(T&& data) {
    {
        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 PollBlocking wakes immediately, it doesn't then 
        // have to block again, waiting for us to unlock the mutex.
    } 
    buffer_not_empty.notify_one();
}
 
 
template<typename T>
std::optional<T> CircularBufferConcurrent<T>::Poll(std::chrono::milliseconds timeout) {
    // 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.
    if (buffer_not_empty.wait_for(lck, timeout, [=](){return !cb.empty();})) {
        // no timeout
        T data = std::move(cb.front());
        cb.pop_front();
        return std::make_optional<T>(std::move(data));
    } else {
        // timeout
        return {};
    }
}
 
 
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