MobileGL/MobileGL/MG_Remote/Transport/Ring.h

// MobileGL - MobileGL/MG_Remote/Transport/Ring.h
// Copyright (c) 2025-2026 MobileGL-Dev
// Licensed under the GNU Lesser General Public License v3.0:
//   https://www.gnu.org/licenses/gpl-3.0.txt
//   https://www.gnu.org/licenses/lgpl-3.0.txt
// SPDX-License-Identifier: LGPL-3.0-only
// End of Source File Header

// SEG_CMD / SEG_STAGE ring control and the SPSC producer/consumer over it.
//
// RingControl is the shared page at the head of SEG_CMD, laid out exactly as
// the inherited transport design (plan section 8.1, referring the earlier
// plan's section 6.2) specifies:
//
//   - TWO independent cursor triples, one for SEG_CMD and one for SEG_STAGE.
//     The stage ring needs its own because "SEG_STAGE has less than a quarter
//     left" is a publish trigger and that occupancy cannot be derived from the
//     command ring's cursors, and because a stage slot retires on a different
//     event than a command record does.
//   - THREE separate sequence watermarks. Conflating them is the classic bug:
//     appliedSeq releases *AppliedTail, submittedSeq releases staging,
//     retiredSeq / completedFrameSerial release *RetiredTail and adopted
//     stores.
//   - TWO tails per ring, not one. Once the server borrows a ring slot into
//     the GPU timeline instead of copying it out again, that slot can only be
//     recycled after completedFrameSerial; a single tail would silently
//     degrade to conservative reclaim the day borrowing lands.
//   - Both park flags, because the doorbell is bidirectional: without the
//     server->client direction every client wait degenerates into a
//     cross-process spin on one shared cache line (a whole 16.6ms frame of a
//     big core, on a phone, competing with the GPU and the game's JVM).
//
// Cursors are monotonically increasing byte counts; the ring is indexed with a
// power-of-two mask. They are never reset, so a torn read can never look like
// a valid earlier position. ringGeneration is bumped after a hard drain to
// invalidate every cached offset.
//
// Record framing inside the ring is the 8-byte header below, which is the
// layout the plan's RecHeader already fixes ({u16 kind, u16 flags, u32 size},
// size including the header and a multiple of 8). The record CATALOGUE
// (Records.def / PipeCalls.def) is a separate deliverable; the ring itself
// only needs kind/flags/size, so it can carry the real records the day they
// land without changing shape.
//
// ---------------------------------------------------------------------------
// THE FIVE WATERMARKS (P5 R-9). One sentence each, and they are a contract:
// every one of the five was declared here at P0 and written by nobody but
// InitRingControl, so until P5 there was nothing to disagree with.
//
//   submittedSeq        Advanced by the PRODUCER after it publishes. NOBODY
//                       WAITS ON IT - it is diagnostic, the answer to "how far
//                       ahead of the server is the client right now".
//   appliedSeq          Advanced by the CONSUMER for EVERY SINGLE RECORD it
//                       applies. The client's verb barrier and every reply wait
//                       read it, so it is the one watermark P5 FORBIDS BATCHING:
//                       the sixty-four-record batching this ring was designed
//                       for makes a waiter block on work that already ran, or -
//                       far worse - resume on work that has not.
//   retiredSeq          Advanced by the CONSUMER once it has finished with the
//                       SEG_STAGE bytes a record referenced. The staging
//                       allocator reclaims behind it, and nothing else may.
//   completedFrameSerial Advanced by the SERVER when a present completes. What
//                       recycling and ageing wait on; it trails appliedSeq by
//                       the GPU's own depth and must never be conflated with it.
//   presentAckSerial    Advanced by the SERVER when it returns a present credit.
//                       The client's present throttle waits on it; it is the
//                       only back-pressure that bounds latency rather than bytes.
//
// Every wait on all five is `>=`, never `==`: a waiter that tests equality
// misses the wakeup the moment a producer or consumer moves by more than one.
//
// BATCHING MAY ONLY MAKE A WATERMARK LATE. All five except appliedSeq may be
// published lazily, because a waiter that sees an old value waits longer than
// it had to and is still correct. NONE of them may ever be published EARLY: a
// watermark that reports more than was actually done turns every waiter into a
// silent use of work that has not happened, and there is no checksum anywhere
// on this ring that would catch it.
//
// kRecPad DOES NOT ADVANCE SEQ. A wrap filler is framing, not a record: it has
// no opcode, no payload meaning and no reply slot. Both sides must skip it
// BEFORE counting. If one side counts it and the other does not, the two seq
// spaces drift by one at every wrap - and because seq IS the reply-slot id
// (P5 R-3), a drifted seq silently reads another call's answer rather than
// failing. Nothing on this ring would detect that, which is why the rule is
// stated here rather than left to each side's loop.
// ---------------------------------------------------------------------------

#pragma once

#include "../Protocol/mg_protocol_base.h"

#include <atomic>
#include <cstddef>
#include <cstdint>

namespace MobileGL::MG_Remote::Transport {

    // The shared control page. One 4 KiB page so it can be mapped alone, with
    // each contended group on its own cache line.
    struct alignas(4096) RingControl {
        // ---- SEG_CMD cursors ------------------------------------------------
        alignas(64) std::atomic<std::uint64_t> cmdHead;        // producer: bytes written
        alignas(64) std::atomic<std::uint64_t> cmdAppliedTail; // consumer: bytes decoded/copied out
        std::atomic<std::uint64_t> cmdRetiredTail;             // consumer: borrowed slots released

        // ---- SEG_STAGE cursors ----------------------------------------------
        alignas(64) std::atomic<std::uint64_t> stageHead;
        alignas(64) std::atomic<std::uint64_t> stageAppliedTail;
        std::atomic<std::uint64_t> stageRetiredTail;

        // ---- sequence / frame watermarks -------------------------------------
        alignas(64) std::atomic<std::uint64_t> appliedSeq; // records applied
        std::atomic<std::uint64_t> submittedSeq;           // handed to the driver
        std::atomic<std::uint64_t> retiredSeq;             // GPU finished
        std::atomic<std::uint64_t> completedFrameSerial;
        std::atomic<std::uint64_t> presentAckSerial;

        // ---- doorbell / generation -------------------------------------------
        alignas(64) std::atomic<std::uint32_t> serverEpoch;    // ++ on context loss / server restart
        std::atomic<std::uint32_t> ringGeneration;             // ++ after a hard drain
        std::atomic<std::uint32_t> consumerParked;             // server asleep, producer must ring
        std::atomic<std::uint32_t> producerParked;             // client asleep, server must ring
        std::atomic<std::uint32_t> eventRingFull;              // SEG_EVENT full, server stopped applying
        std::atomic<std::uint32_t> eventDropped;               // dropped lossy events
    };

    static_assert(sizeof(RingControl) == 4096, "RingControl must be exactly one page");
    static_assert(alignof(RingControl) == 4096, "RingControl must be page aligned");
    static_assert(std::atomic<std::uint64_t>::is_always_lock_free,
                  "the ring cursors are shared across processes: they must be lock-free");
    static_assert(std::atomic<std::uint32_t>::is_always_lock_free,
                  "the doorbell flags are shared across processes: they must be lock-free");

    // Per-record header. Prefix-identical to the plan's RecHeader so the
    // generated record catalogue drops straight in.
    struct RingRecordHeader {
        std::uint16_t kind;
        std::uint16_t flags;
        std::uint32_t size; // header + payload + alignment padding, multiple of 8
    };
    static_assert(sizeof(RingRecordHeader) == 8, "RecHeader is 8 bytes on the wire");

    enum RingRecordFlags : std::uint16_t {
        kRecNone = 0,
        kRecNeedsAck = 1u << 0,
        kRecHasBlob = 1u << 1,
        kRecPad = 1u << 2,       // filler to the wrap boundary, no payload meaning
        kRecBorrowSlot = 1u << 3, // slot is borrowed into the GPU timeline; retires late
        kRecVarTail = 1u << 4,
    };

    // Reserved kind for the wrap filler. The catalogue starts at 1.
    inline constexpr std::uint16_t kRingPadRecordKind = 0;

    inline constexpr std::uint64_t kRingRecordAlignment = 8;

    // Largest ring the 8-byte header can describe. Both a record's size and a
    // wrap filler's size are bounded only by the capacity and are stored in
    // RingRecordHeader::size, which is 32 bits by wire contract: a ring of
    // 4 GiB or more would silently truncate them, and the consumer would then
    // bounds-check the truncated value against the real one. SEG_CMD is 8 MiB
    // and SEG_STAGE 32 MiB today, so this is unreachable - it is the same
    // class of construction-time guard as the power-of-two check beside it.
    inline constexpr std::uint64_t kMaxRingCapacity = 0xFFFFFFFFull;

    // Smallest ring: two record headers. A record may be at most HALF the ring
    // (see RingProducer::Reserve), so a ring of one header could carry nothing
    // at all - not even the smallest record, a bare header.
    inline constexpr std::uint64_t kMinRingCapacity = 2 * sizeof(RingRecordHeader);

    // Which cursor triple a producer/consumer pair drives.
    enum class RingCursorSet : std::uint32_t {
        Cmd = 0,
        Stage = 1,
    };

    // Zeroes every cursor and starts serverEpoch / ringGeneration at 1, so that
    // a zero read is always "uninitialized", never a legal generation.
    void InitRingControl(RingControl& control);

    // head >= appliedTail >= retiredTail, and the ring never holds more than
    // its capacity. False means the shared page is corrupt (or a peer is
    // misbehaving), which is a Fatal{ProtocolCorruption}, never a retry.
    bool RingCursorsValid(const RingControl& control, RingCursorSet cursors,
                          std::uint64_t capacityBytes);

    // Bumps ringGeneration, invalidating every offset either side has cached.
    // Both sides must be quiesced and the ring fully drained
    // (head == appliedTail == retiredTail); otherwise this returns
    // MOBILEGL_ERR_INVALID_ARGUMENT and changes nothing.
    MobileGLResult HardDrainRing(RingControl& control, RingCursorSet cursors);

    // A record as seen by the consumer.
    struct RingRecordView {
        std::uint16_t kind = 0;
        std::uint16_t flags = 0;
        const void* payload = nullptr;
        std::uint64_t payloadSize = 0;
        std::uint64_t cursor = 0; // producer cursor at the START of this record
    };

    // Single producer. Not thread-safe: one writer thread, by construction.
    class RingProducer {
    public:
        RingProducer() = default;
        // `base` is the ring's byte area (NOT the control page) and
        // `capacityBytes` must be a power of two between kMinRingCapacity and
        // kMaxRingCapacity. Anything else leaves Valid() false.
        RingProducer(RingControl* control, void* base, std::uint64_t capacityBytes,
                     RingCursorSet cursors);

        bool Valid() const { return m_control != nullptr; }

        // Bytes still writable before the consumer has to catch up.
        std::uint64_t FreeBytes() const;

        // Reserves room for one record and returns a pointer to its payload,
        // or nullptr when the ring is full. The payload is uninitialized;
        // alignment padding at its tail is NOT zeroed. Emits a pad record
        // automatically when the record would straddle the wrap boundary, so
        // every record is contiguous.
        //
        // A record whose total (header + payload, rounded up to 8) exceeds
        // MaxRecordBytes() == Capacity()/2 is refused outright, with an error
        // log and however empty the ring is: chunking it is the emitter's job
        // (plan section 8.2, the G3 chunking rule). Half is exact, not
        // conservative - it is the largest record EVERY head offset can place,
        // because a wrap pad costs at most total-8 bytes on top of the record
        // and 2*total-8 <= capacity-8 holds exactly up to capacity/2. Above it
        // a record is placeable at some offsets and not at others, and a
        // producer waiting for FreeBytes() >= total stalls forever on an empty
        // ring. So: nullptr with FreeBytes() >= total never means "wait"; it
        // can only mean "too big, chunk".
        void* Reserve(std::uint16_t kind, std::uint16_t flags, std::uint64_t payloadBytes);

        // The largest header+payload total Reserve accepts: Capacity()/2. This
        // is the number the emitter chunks against.
        std::uint64_t MaxRecordBytes() const { return m_capacity / 2; }

        // Makes every reserved record visible to the consumer (release store on
        // the head cursor). Cheap: publishing per record is fine, batching 8-16
        // only amortizes the doorbell store.
        void Publish();

        // Producer-local cursor including records not yet published.
        std::uint64_t LocalHead() const { return m_localHead; }
        std::uint64_t Capacity() const { return m_capacity; }

    private:
        std::uint64_t TailForReclaim() const;
        std::uint8_t* SlotAt(std::uint64_t cursor) const {
            return m_base + static_cast<std::size_t>(cursor & m_mask);
        }

        RingControl* m_control = nullptr;
        std::uint8_t* m_base = nullptr;
        std::uint64_t m_capacity = 0;
        std::uint64_t m_mask = 0;
        std::uint64_t m_localHead = 0;
        RingCursorSet m_cursors = RingCursorSet::Cmd;
    };

    // Single consumer. Not thread-safe: one reader thread, by construction.
    class RingConsumer {
    public:
        RingConsumer() = default;
        RingConsumer(RingControl* control, void* base, std::uint64_t capacityBytes,
                     RingCursorSet cursors);

        bool Valid() const { return m_control != nullptr; }

        // Pops the next record, skipping wrap fillers. Returns false when the
        // ring is empty at this moment. A record whose header is impossible
        // (size not 8-aligned, smaller than a header, or larger than what the
        // producer has published) is refused: *outCorrupt is set, which the
        // caller must escalate to Fatal{ProtocolCorruption} rather than retry.
        bool Pop(RingRecordView& out, bool* outCorrupt = nullptr);

        // Publishes the applied cursor, releasing those bytes to the producer.
        void PublishApplied();
        // Publishes the retired cursor. Records without kRecBorrowSlot retire
        // as soon as they are applied; borrowed slots retire on
        // completedFrameSerial, which is why this is a separate call.
        void PublishRetired();
        void PublishRetiredUpTo(std::uint64_t cursor);

        std::uint64_t LocalTail() const { return m_localTail; }
        std::uint64_t Capacity() const { return m_capacity; }

    private:
        RingControl* m_control = nullptr;
        const std::uint8_t* m_base = nullptr;
        std::uint64_t m_capacity = 0;
        std::uint64_t m_mask = 0;
        std::uint64_t m_localTail = 0;
        RingCursorSet m_cursors = RingCursorSet::Cmd;
    };

} // namespace MobileGL::MG_Remote::Transport