circular_buffer.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "types.hpp"
 
#if defined(DEBUG_CIRC_BUF)
#include DEBUG_CIRC_BUF
#else
#include "ndebug.hpp"
#endif
 
#include <stddef.h>
#include <stdint.h>
 
#include <array>
#include <bit>
#include <optional>
#include <type_traits>
#include <utility>
 
template<typename Value, size_t Size, size_t MaxPacketSize>
class CircularBuffer
{
public:
    static_assert(std::popcount(Size) == 1, "Size needs to be a power of 2");
 
    using value_type = Value;
 
    static constexpr size_t sizeBytes = sizeof(SmallestTypeT<MaxPacketSize>);
 
    constexpr size_t capacity() const { return Size; }
    size_t size() const { return write - read; }
    size_t readable() const { return notified - read; }
    size_t unnotified() const { return std::max(sizeBytes, write - notified) - sizeBytes; }
    bool empty() const { return size() == 0; }
    bool full() const { return size() == capacity(); }
    bool dropping() const { return dropped; }

    template<typename Value_>
        // Here to avoid an extra copy of the code with int for e.g. uint8_t
        requires std::same_as<std::remove_cvref_t<Value>, std::remove_cvref_t<Value_>>
    void push_back(Value_ && v)
    {
        if (unnotified() >= MaxPacketSize || size() >= capacity())
        {
            dropped = true;
        }
        if ((!dropped) && (notified == write))
        {
            write += sizeBytes;
        }
        if (unnotified() >= MaxPacketSize || size() >= capacity())
        {
            dropped = true;
        }
        if (!dropped)
        {
            buf[write++ % Size] = std::forward<Value_>(v);
        }
    }

    void pop_front()
    {
        if (read < notified)
        {
            ++read;
        }
    }

    Value & front()
    {
        return buf[read % Size];
    }

    const Value & front() const
    {
        return buf[read % Size];
    }

    void reset_dropped()
    {
        if (dropped)
        {
            dropped = false;
            write = notified;
        }
    }

    void notify()
    {
        // Since we are about to notify of a partial packet, we want to
        // call `reset_dropped` to make sure it is treated as a full
        // packet.
        reset_dropped();
        size_t size = unnotified();
        for (size_t i = sizeBytes; i > 0; --i)
        {
            buf[notified++ % Size] = static_cast<uint8_t>(size);
            size >>= 8;
        }
        notified = write;
    }
 
    class Frame;
    class Iterator
    {
    public:
        using value_type = Value;
        using difference_type = ptrdiff_t;
 
        Iterator(const CircularBuffer * parent_, size_t index_)
          : parent(parent_), index(index_) {}
        Iterator & operator++() { ++index; return *this; }
        Iterator operator++(int) { const auto tmp = *this; ++*this; return tmp; }
        const Value & operator*() const { return parent->buf[index % Size]; }
        bool operator!=(const Iterator & other) const
        {
            return parent != other.parent || index != other.index;
        }
        bool operator==(const Iterator & other) const { return !(*this != other); }
 
    private:
        friend class Frame;
        const CircularBuffer * parent;
        size_t index;
    };
 
    class Frame
    {
    public:
        Frame(CircularBuffer * parent_, size_t start, uint8_t len)
          : parent(parent_),
            begin_(parent, start),
            end_(parent, start + len) {}
        Frame(const Frame &) = delete;
        Frame & operator=(const Frame &) = delete;
        Frame(Frame && o) : begin_(o.begin_), end_(o.end_)
        {
            std::swap(parent, o.parent);
        }
        Frame & operator=(Frame && o)
        {
            parent = o.parent;
            o.parent = nullptr;
            begin_ = o.begin_;
            end_ = o.end_;
            return *this;
        }
 
        ~Frame()
        {
            if (parent)
            {
                parent->read = end_.index;
            }
        }
 
        Iterator & begin() { return begin_; }
        Iterator & end() { return end_; }
 
        CircularBuffer * parent = nullptr;
        Iterator begin_;
        Iterator end_;
    };

    bool get_frame(std::optional<Frame> & frame)
    {
        // Ensure we drop the old one first, before reading from the queue
        frame.reset();
        if (dropped)
        {
            debugf(DEBUG "No packet as truncating" END LOGLEVEL_ARGS);
            return false;
        }
        auto start = read;
        const auto end = notified;
        if (start == end)
        {
            // No next packet, compare with the next two `start != end`
            return false;
        }
        size_t size = 0;
        for (size_t i = sizeBytes; i > 0; --i)
        {
            if (start != end)
            {
                size <<= 8;
                size |= buf[start++ % Size];
            }
            else
            {
                // A bad state, and we checked for truncating above
                debugf(ERROR "Truncated next packet size" END);
            }
        }
        if (start != end)
        {
            frame.emplace(this, start, size);
        }
        else
        {
            // Almost certainly an error, we checked for truncating above
            debugf(ERROR "Zero length item" END LOGLEVEL_ARGS);
        }
        return frame.has_value();
    }

    void reset()
    {
        read = 0;
        notified = 0;
        write = 0;
        dropped = false;
    }
 
private:
    std::array<Value, Size> buf;
    size_t read = 0;
    size_t notified = 0;
    size_t write = 0;
    bool dropped = false;
};
 
// This is a header, undefine the debugf macro
#include "debug_end.hpp"