OpenRTX/openrtx/include/core/ringbuf.hpp

/***************************************************************************
 *   Copyright (C) 2022 by Federico Amedeo Izzo IU2NUO,                    *
 *                         Niccolò Izzo IU2KIN                             *
 *                         Frederik Saraci IU2NRO                          *
 *                         Silvano Seva IU2KWO                             *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 3 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, see <http://www.gnu.org/licenses/>   *
 ***************************************************************************/

#ifndef RINGBUF_H
#define RINGBUF_H

#ifndef __cplusplus
#error This header is C++ only!
#endif

#include <pthread.h>
#include <cstdint>

/**
 * Class implementing a statically allocated circular buffer with blocking and
 * non-blocking push and pop functions.
 */
template < typename T, size_t N >
class RingBuffer
{
public:

    /**
     * Constructor.
     */
    RingBuffer() : readPos(0), writePos(0), numElements(0)
    {
        pthread_mutex_init(&mutex, NULL);
        pthread_cond_init(&not_empty, NULL);
        pthread_cond_init(&not_full, NULL);
    }

    /**
     * Destructor.
     */
    ~RingBuffer()
    {
        pthread_mutex_destroy(&mutex);
        pthread_cond_destroy(&not_empty);
        pthread_cond_destroy(&not_full);
    }

    /**
     * Push an element to the buffer.
     *
     * @param elem: element to be pushed.
     * @param blocking: if set to true, when the buffer is full this function
     * blocks the execution flow until at least one empty slot is available.
     * @return true if the element has been successfully pushed to the queue,
     * false if the queue is empty.
     */
    bool push(const T& elem, bool blocking)
    {
        pthread_mutex_lock(&mutex);

        if((numElements >= N) && (blocking == false))
        {
            // No space available and non-blocking call: unlock mutex and return
            pthread_mutex_unlock(&mutex);
            return false;
        }

        // The call is blocking: wait until there is some free space
        while(numElements >= N)
        {
            pthread_cond_wait(&not_full, &mutex);
        }

        // There is free space, push data into the queue
        data[writePos] = elem;
        writePos = (writePos + 1) % N;

        // Signal that the queue is not empty
        if(numElements == 0) pthread_cond_signal(&not_empty);
        numElements += 1;

        pthread_mutex_unlock(&mutex);
        return true;
    }

    /**
     * Pop an element from the buffer.
     *
     * @param elem: place where to store the popped element.
     * @param blocking: if set to true, when the buffer is empty this function
     * blocks the execution flow until at least one element is available.
     * @return true if the element has been successfully popped from the queue,
     * false if the queue is empty.
     */
    bool pop(T& elem, bool blocking)
    {
        pthread_mutex_lock(&mutex);

        if((numElements == 0) && (blocking == false))
        {
            // No elements present and non-blocking call: unlock mutex and return
            pthread_mutex_unlock(&mutex);
            return false;
        }

        // The call is blocking: wait until there is something into the queue
        while(numElements == 0)
        {
            pthread_cond_wait(&not_empty, &mutex);
        }

        // At least one element present pop one.
        elem    = data[readPos];
        readPos = (readPos + 1) % N;

        // Signal that the queue is no more full
        if(numElements >= N) pthread_cond_signal(&not_full);
        numElements -= 1;

        pthread_mutex_unlock(&mutex);

        return true;
    }

    /**
     * Check if the buffer is empty.
     *
     * @return true if the buffer is empty.
     */
    bool empty()
    {
        return numElements == 0;
    }

    /**
     * Check if the buffer is full.
     *
     * @return true if the buffer is full.
     */
    bool full()
    {
        return numElements >= N;
    }

    /**
     * Discard one element from the buffer's tail, creating a new empty slot.
     * In case the buffer is full calling this function unlocks the eventual
     * threads waiting to push data.
     */
    void eraseElement()
    {
        // Nothing to erase
        if(numElements == 0) return;

        pthread_mutex_lock(&mutex);

        // Chomp away one element just by advancing the read pointer.
        readPos = (readPos + 1) % N;

        if(numElements >= N) pthread_cond_signal(&not_full);
        numElements -= 1;

        pthread_mutex_unlock(&mutex);
    }

private:

    size_t readPos;      ///< Read pointer.
    size_t writePos;     ///< Write pointer.
    size_t numElements;  ///< Number of elements currently present.
    T      data[N];      ///< Data storage.

    pthread_mutex_t mutex;      ///< Mutex for concurrent access.
    pthread_cond_t  not_empty;  ///< Queue not empty condition.
    pthread_cond_t  not_full;   ///< Queue not full condition.
};

#endif  // RINGBUF_H