feat: expand Transport

core/src/Transport.cpp

@@ -323,3 +323,162 @@ bool UnreliableChannel::accept(const Datagram::Fragment& piece) {
     anyDelivered = true;
     incoming.erase(it);
 
+    // Anything still half-assembled below the newest complete message can only
+    // ever produce a stale update, so it is abandoned right away.
+    for (auto stalePart = incoming.begin(); stalePart != incoming.end();) {
+        if (stalePart->first < highestDelivered) {
+            ++stale;
+            stalePart = incoming.erase(stalePart);
+        } else {
+            ++stalePart;
+        }
+    }
+    return true;
+}
+
+bool UnreliableChannel::popDelivered(ByteBuffer& out) {
+    if (delivered.empty()) return false;
+    out = std::move(delivered.front());
+    delivered.pop_front();
+    return true;
+}
+
+Transport::Transport(LoopbackLink& link, int endpoint, TransportConfig config)
+    : link(&link), endpoint(endpoint), config(config) {
+    std::size_t floorMtu = Datagram::headerSize() + Datagram::maxFragmentOverhead() + 16;
+    if (this->config.mtu < floorMtu) this->config.mtu = floorMtu;
+    if (this->config.sendWindow == 0) this->config.sendWindow = 1;
+    if (this->config.retransmitTimeoutTicks == 0) this->config.retransmitTimeoutTicks = 1;
+}
+
+std::size_t Transport::maxFragmentPayload() const {
+    return config.mtu - Datagram::headerSize() - Datagram::maxFragmentOverhead();
+}
+
+void Transport::sendReliable(const uint8_t* payload, std::size_t size) {
+    reliable.queueMessage(payload, size, maxFragmentPayload());
+    ++statistics.messagesQueued;
+}
+
+void Transport::sendReliable(const ByteBuffer& payload) {
+    sendReliable(payload.data(), payload.size());
+}
+
+void Transport::sendUnreliable(const uint8_t* payload, std::size_t size) {
+    unreliable.queueMessage(payload, size, maxFragmentPayload());
+    ++statistics.messagesQueued;
+}
+
+void Transport::sendUnreliable(const ByteBuffer& payload) {
+    sendUnreliable(payload.data(), payload.size());
+}
+
+bool Transport::registerIncoming(SequenceNumber sequence) {
+    if (!anyReceived) {
+        anyReceived = true;
+        remoteLatest = sequence;
+        remoteHistory = 0;
+        return true;
+    }
+    if (sequence == remoteLatest) return false;
+
+    if (sequenceGreater(sequence, remoteLatest)) {
+        uint16_t shift = sequenceDistance(sequence, remoteLatest);
+        if (shift >= 32) {
+            remoteHistory = 0;
+        } else {
+            remoteHistory = (remoteHistory << shift) | (1u << (shift - 1));
+        }
+        remoteLatest = sequence;
+        return true;
+    }
+
+    uint16_t behind = sequenceDistance(remoteLatest, sequence);
+    if (behind > 32) return false;
+    uint32_t bit = 1u << (behind - 1);
+    if ((remoteHistory & bit) != 0) return false;
+    remoteHistory |= bit;
+    return true;
+}
+
+void Transport::acknowledge(SequenceNumber sequence, uint64_t tick) {
+    for (std::size_t i = 0; i < inFlight.size(); ++i) {
+        if (inFlight[i].sequence != sequence) continue;
+        sampleRtt(tick - inFlight[i].sentTick);
+        for (uint64_t key : inFlight[i].reliableKeys) reliable.markAcked(key);
+        inFlight.erase(inFlight.begin() + static_cast<std::ptrdiff_t>(i));
+        return;
+    }
+}
+
+void Transport::applyAcks(SequenceNumber ackLatest, AckBitfield ackHistory, uint64_t tick) {
+    acknowledge(ackLatest, tick);
+    for (int bit = 0; bit < 32; ++bit) {
+        if ((ackHistory & (1u << bit)) == 0) continue;
+        acknowledge(static_cast<SequenceNumber>(ackLatest - 1 - bit), tick);
+    }
+}
+
+void Transport::sampleRtt(uint64_t sample) {
+    double value = static_cast<double>(sample);
+    if (statistics.smoothedRttTicks <= 0.0) {
+        statistics.smoothedRttTicks = value;
+        return;
+    }
+    double weight = static_cast<double>(config.rttSmoothingPercent) / 100.0;
+    statistics.smoothedRttTicks += (value - statistics.smoothedRttTicks) * weight;
+}
+
+void Transport::pump(uint64_t tick) {
+    link->receive(endpoint, tick, arrivals);
+    for (const ByteBuffer& raw : arrivals) {
+        ++statistics.datagramsReceived;
+        statistics.bytesReceived += raw.size();
+
+        Datagram datagram;
+        if (!datagram.decode(raw.data(), raw.size())) continue;
+        if (!registerIncoming(datagram.sequence)) {
+            ++statistics.duplicatesDropped;
+            continue;
+        }
+        if (datagram.ackValid) applyAcks(datagram.ackLatest, datagram.ackHistory, tick);
+        if (!datagram.fragments.empty()) ackPending = true;
+
+        for (const Datagram::Fragment& piece : datagram.fragments) {
+            bool accepted = piece.reliable ? reliable.accept(piece) : unreliable.accept(piece);
+            if (!accepted) ++statistics.staleDropped;
+        }
+    }
+}
+
+bool Transport::buildDatagram(Datagram& out) {
+    out.fragments.clear();
+    std::size_t used = Datagram::headerSize();
+
+    std::vector<uint64_t> ready;
+    reliable.collectReady(ready);
+    for (uint64_t key : ready) {
+        const ReliableChannel::OutgoingFragment* piece = reliable.fragment(key);
+        std::size_t cost = fragmentEncodedSize(piece->messageId, piece->fragmentIndex,
+                                               piece->fragmentCount, piece->bytes.size());
+        if (used + cost > config.mtu) break;
+        Datagram::Fragment wire;
+        wire.reliable = true;
+        wire.messageId = piece->messageId;
+        wire.fragmentIndex = piece->fragmentIndex;
+        wire.fragmentCount = piece->fragmentCount;
+        wire.bytes = piece->bytes;
+        out.fragments.push_back(std::move(wire));
+        used += cost;
+        reliable.markInFlight(key);
+    }
+
+    while (const Datagram::Fragment* piece = unreliable.peekOutgoing()) {
+        std::size_t cost = fragmentEncodedSize(piece->messageId, piece->fragmentIndex,
+                                               piece->fragmentCount, piece->bytes.size());
+        if (used + cost > config.mtu) break;
+        out.fragments.push_back(*piece);
+        used += cost;
+        unreliable.dropOutgoing();
+    }
+