feat(vf): add Assembly and InFlightDatagram to Transport
This commit is contained in:
1 parent
7f1575defe
commit
fac200dcf3
1 file changed
+120
@@ -157,3 +157,123 @@ public:
|
|||||||
uint32_t nextExpectedMessageId() const { return nextDeliverMessageId; }
|
uint32_t nextExpectedMessageId() const { return nextDeliverMessageId; }
|
||||||
std::size_t reorderBufferSize() const { return completed.size(); }
|
std::size_t reorderBufferSize() const { return completed.size(); }
|
||||||
|
|
||||||
|
static uint64_t fragmentKey(uint32_t messageId, uint32_t fragmentIndex);
|
||||||
|
|
||||||
|
private:
|
||||||
|
struct Assembly {
|
||||||
|
uint32_t fragmentCount = 0;
|
||||||
|
uint32_t received = 0;
|
||||||
|
std::vector<ByteBuffer> parts;
|
||||||
|
std::vector<uint8_t> have;
|
||||||
|
};
|
||||||
|
|
||||||
|
void drainCompleted();
|
||||||
|
|
||||||
|
std::map<uint64_t, OutgoingFragment> outgoing;
|
||||||
|
uint32_t nextOutgoingMessageId = 0;
|
||||||
|
|
||||||
|
std::map<uint32_t, Assembly> incoming;
|
||||||
|
std::map<uint32_t, ByteBuffer> completed;
|
||||||
|
uint32_t nextDeliverMessageId = 0;
|
||||||
|
std::deque<ByteBuffer> delivered;
|
||||||
|
};
|
||||||
|
|
||||||
|
// Unordered, send-once stream for state that ages badly. A message older than
|
||||||
|
// the newest one already delivered is thrown away instead of queued, because a
|
||||||
|
// stale entity position is worse than no update at all.
|
||||||
|
class UnreliableChannel {
|
||||||
|
public:
|
||||||
|
void queueMessage(const uint8_t* payload, std::size_t size, std::size_t fragmentPayload);
|
||||||
|
|
||||||
|
const Datagram::Fragment* peekOutgoing() const;
|
||||||
|
void dropOutgoing();
|
||||||
|
std::size_t outgoingFragmentCount() const { return outgoing.size(); }
|
||||||
|
|
||||||
|
bool accept(const Datagram::Fragment& fragment);
|
||||||
|
bool popDelivered(ByteBuffer& out);
|
||||||
|
std::size_t deliveredWaiting() const { return delivered.size(); }
|
||||||
|
uint32_t highestDeliveredMessageId() const { return highestDelivered; }
|
||||||
|
uint64_t staleDroppedCount() const { return stale; }
|
||||||
|
|
||||||
|
private:
|
||||||
|
struct Assembly {
|
||||||
|
uint32_t fragmentCount = 0;
|
||||||
|
uint32_t received = 0;
|
||||||
|
std::vector<ByteBuffer> parts;
|
||||||
|
std::vector<uint8_t> have;
|
||||||
|
};
|
||||||
|
|
||||||
|
std::deque<Datagram::Fragment> outgoing;
|
||||||
|
uint32_t nextOutgoingMessageId = 0;
|
||||||
|
|
||||||
|
std::map<uint32_t, Assembly> incoming;
|
||||||
|
uint32_t highestDelivered = 0;
|
||||||
|
bool anyDelivered = false;
|
||||||
|
std::deque<ByteBuffer> delivered;
|
||||||
|
uint64_t stale = 0;
|
||||||
|
};
|
||||||
|
|
||||||
|
// Reliability layered over the unreliable datagram link: sequencing, acks with
|
||||||
|
// a 32-entry bitfield, timeout-driven retransmission, a bounded send window and
|
||||||
|
// the two message channels. Time is the tick handed in from outside; the class
|
||||||
|
// never reads a clock.
|
||||||
|
class Transport {
|
||||||
|
public:
|
||||||
|
Transport(LoopbackLink& link, int endpoint, TransportConfig config = {});
|
||||||
|
|
||||||
|
void sendReliable(const uint8_t* payload, std::size_t size);
|
||||||
|
void sendReliable(const ByteBuffer& payload);
|
||||||
|
void sendUnreliable(const uint8_t* payload, std::size_t size);
|
||||||
|
void sendUnreliable(const ByteBuffer& payload);
|
||||||
|
|
||||||
|
// Reads arrivals and applies their acks.
|
||||||
|
void pump(uint64_t tick);
|
||||||
|
// Expires unacked datagrams, then packs and sends whatever fits.
|
||||||
|
void flush(uint64_t tick);
|
||||||
|
void update(uint64_t tick);
|
||||||
|
|
||||||
|
bool popReliable(ByteBuffer& out);
|
||||||
|
bool popUnreliable(ByteBuffer& out);
|
||||||
|
|
||||||
|
const TransportStats& stats() const { return statistics; }
|
||||||
|
const TransportConfig& settings() const { return config; }
|
||||||
|
std::size_t maxFragmentPayload() const;
|
||||||
|
std::size_t inFlightDatagramCount() const { return inFlight.size(); }
|
||||||
|
std::size_t unackedFragmentCount() const { return reliable.unackedFragmentCount(); }
|
||||||
|
std::size_t pendingOutgoingCount() const;
|
||||||
|
ReliableChannel& reliableChannel() { return reliable; }
|
||||||
|
UnreliableChannel& unreliableChannel() { return unreliable; }
|
||||||
|
|
||||||
|
private:
|
||||||
|
struct InFlightDatagram {
|
||||||
|
SequenceNumber sequence = 0;
|
||||||
|
uint64_t sentTick = 0;
|
||||||
|
std::vector<uint64_t> reliableKeys;
|
||||||
|
};
|
||||||
|
|
||||||
|
bool registerIncoming(SequenceNumber sequence);
|
||||||
|
void applyAcks(SequenceNumber ackLatest, AckBitfield ackHistory, uint64_t tick);
|
||||||
|
void acknowledge(SequenceNumber sequence, uint64_t tick);
|
||||||
|
bool buildDatagram(Datagram& out);
|
||||||
|
void sampleRtt(uint64_t sample);
|
||||||
|
|
||||||
|
LoopbackLink* link;
|
||||||
|
int endpoint;
|
||||||
|
TransportConfig config;
|
||||||
|
|
||||||
|
ReliableChannel reliable;
|
||||||
|
UnreliableChannel unreliable;
|
||||||
|
|
||||||
|
SequenceNumber nextSequence = 0;
|
||||||
|
std::vector<InFlightDatagram> inFlight;
|
||||||
|
|
||||||
|
SequenceNumber remoteLatest = 0;
|
||||||
|
AckBitfield remoteHistory = 0;
|
||||||
|
bool anyReceived = false;
|
||||||
|
bool ackPending = false;
|
||||||
|
|
||||||
|
std::vector<ByteBuffer> arrivals;
|
||||||
|
TransportStats statistics;
|
||||||
|
};
|
||||||
|
|
||||||
|
} // namespace vf
|
||||||
Reference in new issue
Block a user