Jungfraujoch/image_puller/ReceiveRingBuffer.cpp

// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only

#include "ReceiveRingBuffer.h"

#include <algorithm>
#include <optional>

struct ReceiveRingBuffer::State {
    // Uninitialised on purpose: a ring that is never filled should not cost its capacity in
    // resident memory.
    std::unique_ptr<uint8_t[]> memory;
    const size_t capacity;

    std::mutex m;
    std::condition_variable cv;

    // Live regions in arrival order. Only the oldest can give back space, so a region released out of
    // turn is marked and waits.
    struct Slot {
        size_t offset;
        size_t size;
        bool released = false;
    };
    std::deque<Slot> slots;
    // Next byte to hand out. The ring is empty when slots is empty; the oldest live byte is then
    // slots.front().offset.
    size_t tail = 0;

    size_t utilization = 0;
    size_t max_utilization = 0;
    bool stopped = false;

    explicit State(size_t in_capacity)
            : memory(in_capacity > 0 ? std::make_unique_for_overwrite<uint8_t[]>(in_capacity) : nullptr),
              capacity(in_capacity) {}

    // Where a region of `size` bytes would start, or nothing if it does not fit right now. Called
    // under the mutex.
    std::optional<size_t> FindSpace(size_t size) const {
        if (slots.empty())
            return 0;
        const size_t head = slots.front().offset;
        // tail has come round to head: treated as full. (After a wrap the piece between the old tail
        // and the end is free, but it is behind the live data and not worth tracking.)
        if (tail == head)
            return std::nullopt;
        if (tail > head) {
            // Live data is [head, tail): free space is behind it to the end, or before it at the start.
            if (capacity - tail >= size)
                return tail;
            if (head >= size)
                return 0;
            return std::nullopt;
        }
        // Wrapped: live data is [head, capacity) + [0, tail), free space is [tail, head).
        if (head - tail >= size)
            return tail;
        return std::nullopt;
    }

    void Release(size_t offset) {
        std::unique_lock ul(m);
        for (auto &s: slots) {
            if (s.offset == offset) {
                s.released = true;
                break;
            }
        }
        while (!slots.empty() && slots.front().released) {
            utilization -= slots.front().size;
            slots.pop_front();
        }
        if (slots.empty())
            tail = 0;
        cv.notify_all();
    }

    void ReleaseHeap(size_t size) {
        std::unique_lock ul(m);
        utilization -= size;
    }
};

ReceiveRingBuffer::ReceiveRingBuffer(size_t capacity) : state(std::make_shared<State>(capacity)) {}

std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
    // Nothing to hold, and a zero-length slot would share its offset with the next real one.
    if (size == 0)
        return std::make_shared<ReceiveBufferRegion>();

    std::unique_lock ul(state->m);

    if (size > state->capacity) {
        // Heap fallback: counted, not bounded.
        state->utilization += size;
        state->max_utilization = std::max(state->max_utilization, state->utilization);
        auto *region = new ReceiveBufferRegion{new uint8_t[size], size};
        return std::shared_ptr<ReceiveBufferRegion>(region, [s = state](ReceiveBufferRegion *r) {
            delete[] r->data;
            s->ReleaseHeap(r->size);
            delete r;
        });
    }

    std::optional<size_t> offset;
    state->cv.wait(ul, [&] {
        if (state->stopped)
            return true;
        offset = state->FindSpace(size);
        return offset.has_value();
    });
    if (!offset)
        return nullptr;

    state->slots.push_back({*offset, size});
    state->tail = *offset + size;
    state->utilization += size;
    state->max_utilization = std::max(state->max_utilization, state->utilization);

    auto *region = new ReceiveBufferRegion{state->memory.get() + *offset, size};
    return std::shared_ptr<ReceiveBufferRegion>(region, [s = state, off = *offset](ReceiveBufferRegion *r) {
        s->Release(off);
        delete r;
    });
}

void ReceiveRingBuffer::Stop() {
    std::unique_lock ul(state->m);
    state->stopped = true;
    state->cv.notify_all();
}

size_t ReceiveRingBuffer::GetCapacity() const {
    return state->capacity;
}

size_t ReceiveRingBuffer::GetUtilization() const {
    std::unique_lock ul(state->m);
    return state->utilization;
}

size_t ReceiveRingBuffer::GetMaxUtilization() const {
    std::unique_lock ul(state->m);
    return state->max_utilization;
}