diff --git a/core/src/Transport.cpp b/core/src/Transport.cpp index e31ae31..669a3dd 100644 --- a/core/src/Transport.cpp +++ b/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(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(ackLatest - 1 - bit), tick); + } +} + +void Transport::sampleRtt(uint64_t sample) { + double value = static_cast(sample); + if (statistics.smoothedRttTicks <= 0.0) { + statistics.smoothedRttTicks = value; + return; + } + double weight = static_cast(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 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(); + } +