From 7f1575defe51562d1c821af26427fd176852f773 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 6 Sep 2026 11:42:10 +0200 Subject: [PATCH] feat(vf): add LinkQuality and PendingDatagram --- core/include/vf/Transport.h | 159 ++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 core/include/vf/Transport.h diff --git a/core/include/vf/Transport.h b/core/include/vf/Transport.h new file mode 100644 index 0000000..03bff85 --- /dev/null +++ b/core/include/vf/Transport.h @@ -0,0 +1,159 @@ +#pragma once + +#include +#include +#include +#include +#include +#include "Codec.h" + +namespace vf { + +// Datagram numbering is 16-bit and wraps, so every comparison between two +// sequence numbers has to be modular rather than arithmetic. +using SequenceNumber = uint16_t; + +// Bit i means "the datagram numbered ackLatest - 1 - i also arrived", so one +// 32-bit word confirms 33 datagrams and a single surviving ack can recover the +// whole window that came before it. +using AckBitfield = uint32_t; + +// Everything a hostile link does to a datagram, as integer percents and tick +// counts. The seed makes a run reproducible: the same sequence of sends always +// produces the same sequence of arrivals. +struct LinkQuality { + uint32_t lossPercent = 0; + uint32_t duplicatePercent = 0; + uint32_t reorderPercent = 0; + uint32_t latencyTicks = 1; + uint32_t reorderExtraTicks = 3; + uint64_t seed = 0x9e3779b97f4a7c15ull; +}; + +// A two-endpoint in-memory pipe: no sockets, no threads, no wall clock. +// Delivery is a tick-stamped queue and every impairment is drawn from a seeded +// stream, which is what makes the transport tests reproducible. +class LoopbackLink { +public: + explicit LoopbackLink(LinkQuality quality = {}); + + void send(int fromEndpoint, const uint8_t* datagram, std::size_t size, uint64_t tick); + void send(int fromEndpoint, const ByteBuffer& datagram, uint64_t tick); + + // Every datagram addressed to `endpoint` whose delivery tick has arrived, + // oldest first. Same-tick arrivals keep their send order. + void receive(int endpoint, uint64_t tick, std::vector& out); + + // Re-seeds the impairment stream, so a test can change the link mid-run. + void setQuality(const LinkQuality& newQuality); + const LinkQuality& settings() const { return quality; } + void reset(); + + uint64_t sentCount() const { return sent; } + uint64_t droppedCount() const { return dropped; } + uint64_t duplicatedCount() const { return duplicated; } + uint64_t delayedCount() const { return delayed; } + std::size_t inFlightCount() const { return pending.size(); } + +private: + struct PendingDatagram { + uint64_t deliveryTick = 0; + uint64_t order = 0; + int toEndpoint = 0; + ByteBuffer bytes; + }; + + uint64_t nextRandom(); + bool roll(uint32_t percent); + void enqueue(int toEndpoint, const ByteBuffer& bytes, uint64_t deliveryTick); + + LinkQuality quality; + uint64_t rngState; + uint64_t orderCounter = 0; + std::vector pending; + uint64_t sent = 0; + uint64_t dropped = 0; + uint64_t duplicated = 0; + uint64_t delayed = 0; +}; + +// One unit on the wire: an ack header plus however many message fragments fit +// under the MTU. A fragment carries its own message identity so a datagram can +// be lost without losing track of what was in it. +struct Datagram { + struct Fragment { + bool reliable = false; + uint32_t messageId = 0; + uint32_t fragmentIndex = 0; + uint32_t fragmentCount = 1; + ByteBuffer bytes; + }; + + SequenceNumber sequence = 0; + SequenceNumber ackLatest = 0; + AckBitfield ackHistory = 0; + // Until the sender has itself received something, ackLatest is meaningless + // and must not be mistaken for an acknowledgement of sequence 0. + bool ackValid = false; + std::vector fragments; + + void encode(ByteWriter& out) const; + bool decode(const uint8_t* data, std::size_t size); + + static std::size_t headerSize() { return 11; } + static std::size_t maxFragmentOverhead() { return 21; } +}; + +struct TransportConfig { + std::size_t mtu = 512; + uint32_t retransmitTimeoutTicks = 8; + uint32_t sendWindow = 32; + uint32_t rttSmoothingPercent = 20; +}; + +struct TransportStats { + uint64_t datagramsSent = 0; + uint64_t datagramsReceived = 0; + uint64_t datagramsLost = 0; + uint64_t retransmitted = 0; + uint64_t duplicatesDropped = 0; + uint64_t staleDropped = 0; + uint64_t messagesQueued = 0; + uint64_t messagesDelivered = 0; + uint64_t bytesSent = 0; + uint64_t bytesReceived = 0; + double smoothedRttTicks = 0.0; +}; + +// Ordered, exactly-once message stream. Outgoing fragments live until acked; +// incoming fragments are reassembled per message and then held back until +// every earlier message id has been handed over. +class ReliableChannel { +public: + struct OutgoingFragment { + uint32_t messageId = 0; + uint32_t fragmentIndex = 0; + uint32_t fragmentCount = 1; + ByteBuffer bytes; + bool inFlight = false; + }; + + void queueMessage(const uint8_t* payload, std::size_t size, std::size_t fragmentPayload); + + void collectReady(std::vector& keys) const; + const OutgoingFragment* fragment(uint64_t key) const; + void markInFlight(uint64_t key); + void markAcked(uint64_t key); + void markLost(uint64_t key); + + std::size_t unackedFragmentCount() const { return outgoing.size(); } + std::size_t readyFragmentCount() const; + + // False when the fragment belongs to a message already handed over, which + // is what a retransmission of an already-delivered message looks like. + bool accept(const Datagram::Fragment& fragment); + bool popDelivered(ByteBuffer& out); + std::size_t deliveredWaiting() const { return delivered.size(); } + uint32_t nextExpectedMessageId() const { return nextDeliverMessageId; } + std::size_t reorderBufferSize() const { return completed.size(); } +