repubBuf.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_REUSABLECOMPONENT_TELREPUB_TELREPUB_BUF_HPP_
#define RTCTK_REUSABLECOMPONENT_TELREPUB_TELREPUB_BUF_HPP_
 
#include <array>
#include <condition_variable>
#include <mutex>
 
#include "rtctk/componentFramework/ddsCommon.hpp"
#include "rtctk/componentFramework/logger.hpp"
 
namespace rtctk::telRepub {
 
template <typename T = rtctk::componentFramework::AgnosticTopic>
class RepubBuf {
public:
    RepubBuf(const std::string &id, uint16_t size)
        : m_identifier(id), m_buf_size(size), m_free_slots(size), m_min_free_slots(size) {
        m_buf.resize(m_buf_size);
    }
 
    RepubBuf(const char *id, uint16_t size)
        : m_identifier(id), m_buf_size(size), m_free_slots(size), m_min_free_slots(size) {
        m_buf.resize(m_buf_size);
    }
 
    RepubBuf(uint16_t size) : m_buf_size(size), m_free_slots(size), m_min_free_slots(size) {
        m_buf.resize(m_buf_size);
    }
 
    ~RepubBuf() noexcept {
        DumpStatistic();
    }
 
    RepubBuf(RepubBuf &&rhs) = delete;
    RepubBuf &operator=(RepubBuf &&rhs) = delete;
 
    T *GetNextFree() {
        // TBD added flag check if GetNextFree was *not* called before
        std::unique_lock lock(m_mutex);
        if (m_free_slots <= 0) {
            return nullptr;
        }
 
        return &(m_buf[m_next_free]);
    }  // GetNextFree
 
    void MakeAvailable() {
        // TBD added flag check if GetNextFree was called before
        std::unique_lock lock(m_mutex);
        if (m_next_free >= (m_buf_size - 1)) {
            m_next_free = 0;
        } else {
            m_next_free++;
        }
        m_free_slots--;
        if (m_free_slots < m_min_free_slots) {
            m_min_free_slots = m_free_slots;
        }
 
        lock.unlock();
        m_not_empty.notify_one();
    }  // MakeAvailable
 
    void ReleaseFree() {
        // TBD added flag check if GetNextFree was called before and set it
        //        std::unique_lock lock(m_mutex);
    }  // ReleaseFree
 
    T *GetNextAvail() {
        std::unique_lock lock(m_mutex);
        if (m_free_slots >= m_buf_size) {
            return nullptr;
        }
 
        return &(m_buf[m_next_avail]);
    }  // GetNextAvail
 
    T *WaitNextAvail() {
        std::unique_lock lock(m_mutex);
        // if (m_free_slots>=B_SIZE)
        m_not_empty.wait(lock, [&]() {
            return ((m_free_slots < m_buf_size) || m_exit_wait);
        });  // TBD: to be replaced with wait_until and timeout
 
        return !m_exit_wait ? &(m_buf[m_next_avail]) : nullptr;
    }  // GetNextAvail
 
    void MakeFree() {
        // TBD added flag check if GetNextAvail was called before
        std::unique_lock lock(m_mutex);
        if (m_next_avail >= (m_buf_size - 1)) {
            m_next_avail = 0;
        } else {
            m_next_avail++;
        }
        m_free_slots++;
    }  // MakeFree
 
    void ReleaseWait() {
        m_exit_wait = true;
        m_not_empty.notify_one();
 
    }  // ReleaseWait
 
    uint16_t GetFreeCapacity() {
        std::unique_lock lock(m_mutex);
        return m_free_slots;
    }
 
    uint16_t GetAvail() {
        std::unique_lock lock(m_mutex);
        return (m_buf_size - m_free_slots);
    }
 
    void DumpStatistic() {
        LOG4CPLUS_DEBUG_FMT(rtctk::componentFramework::GetLogger(),
                            "[%s] Max TelRepub buffer occupancy %d (%f %%).\n",
                            m_identifier.c_str(),
                            (m_buf_size - m_min_free_slots),
                            float(m_buf_size - m_min_free_slots) / m_buf_size);
    }  // dumpStatistic
 
    uint16_t GetMaxOccupancy() {
        std::unique_lock lock(m_mutex);
        return (m_buf_size - m_min_free_slots);
    }
 
    uint16_t GetSize() {
        return m_buf_size;
    }
private:
    std::string m_identifier;  
 
    std::vector<T> m_buf;  
    uint16_t m_buf_size;   
 
    uint16_t m_free_slots;      
    uint16_t m_min_free_slots;  
 
    uint16_t m_next_free = 0;   
    uint16_t m_next_avail = 0;  
 
    mutable std::mutex m_mutex;
    std::condition_variable m_not_empty;
    std::atomic<bool> m_exit_wait = false;  
 
};  // class RepubBuf
 
}  // namespace rtctk::telRepub
 
#endif  // RTCTK_REUSABLECOMPONENT_TELREPUB_TELREPUB_BUF_HPP_