From 7d485c96721839ffbc09894ee1c20471ab417c1a Mon Sep 17 00:00:00 2001 From: Cyd Date: Thu, 13 Aug 2026 12:59:25 -0500 Subject: [PATCH] The wire keeps what it could not send Cyd asked for an analysis of the networking stack and what would make the simulation feel better over the internet. The analysis found something more urgent than latency: the transport has been losing data silently since the arcade, and nothing in the game could see it happen. Every send result was discarded - L4NET, the console, all of it. On the 1ms arcade LAN the socket buffer never filled, so it never mattered. Over the internet it matters twice. A peer stalled in its own 10-30 second mission load stops reading, its window closes, and our nonblocking send starts answering would-block, which threw the message away; or worse, answering a PARTIAL count, and since framing on that stream is recovered purely from each message's length prefix, the bytes that never followed sheared it for good. Both are reachable in an ordinary race, because every race has a load in it. So sends go through a bounded per-connection queue now. What the wire will not take is kept, byte-exact, and retried at three flush points - before the render (the present blocks on vsync, and this frame's state should be travelling while it does), at the top of the receive pump, and before a connect sequence. Nothing is ever dropped from the middle: these are reliable ordered messages carrying entity creation, damage and race control, so a queue that overflows its 256K declares the connection dead and lets the disconnect path run rather than quietly desyncing the stream. RP412NETSENDQ=0 restores the old behaviour and still logs what it would have lost, which is the honest way to A/B it. On Steam the same queue finally surfaces k_EResultLimitExceeded, which the old code collapsed into -1 and discarded - that was backpressure, unlogged. The receive side gained the check the release build never had. The length prefix is untrusted input; Verify() compiles away in release, so a corrupt one went to memmove as a negative, or copied 4096 bytes of assembled packet into a 1600-byte stack buffer, or named a size the pad could never complete and wedged the connection forever. It is now validated against the same bounds the sender works to, and a stream that fails them is dropped like any other lost peer. And a fry that never ends: drop zones are map entities dealt round-robin at load, ownership transfer is not implemented, so a leaver's pads stay in the DropZones group. The respawn request dispatched to one goes to a host that is gone - dropped at the send, the 'no host N in the table' path - and the two-second retry re-dispatches to the same dead owner forever. The pad scan now skips zones whose owner has left, and re-validates one assigned earlier before reusing it. The rest is measurement, because the symptoms this work exists to chase are all reported in prose and none of them are in any log. Sixteen logs from the six-player night contain zero player-facing latency lines. A race now ends with a NetLog summary: per remote pod, how many updates arrived and how evenly (median and p95 out of a log2 histogram), the widest gap, how many gaps were long enough to mean a quiet sender versus short enough to mean OUR loop stalled, how often its motion snapped instead of blending, and how far arriving updates moved it. Per peer, whether the clock alignment ever had to step mid-race - which is the input for deciding if it needs slewing, rather than guessing. The mission t0 tick goes in the log too, alongside the console's per-pod RunMission send ticks, because nothing has ever measured how far apart the machines actually start; the clockwork doors inherit that skew directly. RP412NETSTATS adds the transport's own view - per connection: messages, bytes, wire writes, partials, refusals, how much sat queued - and on Steam the first read this codebase has ever taken of GetConnectionRealTime Status. Ping, quality, pending and unacked bytes, and one route description per connection at teardown. The API was vendored and never called; there was no RTT number anywhere in the game. Finally, rpl4opt -spoolstats reads any recording offline. The data was already in every spool ever made and nothing read it that way: the recorder restamps each packet with local arrival time while the update records inside keep the sender's sim-grid stamp, so the difference is clock offset plus one-way delay, and the same running-minimum estimator the game runs live separates them. It prints delay above the per-host minimum, and decomposes each entity's gaps into sender pacing versus delivery jitter - which no live counter can do. It lives in the game exe rather than RPL4TOOL because the tool is deliberately not /Zp1 and would misread every struct in the file. Verified on the two-pod loopback harness: mesh up, egg fed, 60s raced, stopped on command, scores collected, and both summaries reading exactly what a pair of PARKED pods should read - heartbeat cadence, one snap per heartbeat, sub-quarter-metre corrections, no clock steps. The t0 ticks and the netclock offsets agree with each other to the two seconds the pods launched apart. The latency tier is deliberately NOT here. TCP_NODELAY, the Steam NoNagle flag, per-frame coalescing and the pre-sim receive drain are all scoped and all wait on this build's numbers, because the point of shipping measurement first is to find out whether the thing we would fix is the thing that hurts. Nagle is still on. Interest management is still inert. The wire format is untouched, so this build and the last one still race each other. Co-Authored-By: Claude Fable 5 --- MUNGA/APP.cpp | 142 +++++++++ MUNGA/ENTITY.cpp | 22 ++ MUNGA/ENTITY.h | 71 +++++ MUNGA/MOVER.cpp | 29 ++ MUNGA/NETWORK.h | 10 + MUNGA/PLAYER.cpp | 153 +++++++--- MUNGA/SIMULATE.cpp | 46 +++ MUNGA/SIMULATE.h | 1 + MUNGA_L4/L4NET.CPP | 78 ++++- MUNGA_L4/L4NET.H | 3 + MUNGA_L4/L4NETTRANSPORT.cpp | 432 +++++++++++++++++++++++++++- MUNGA_L4/L4NETTRANSPORT.h | 101 +++++++ MUNGA_L4/L4STEAMTRANSPORT.cpp | 274 +++++++++++++++++- README.md | 1 + RP_L4/RPL4.CPP | 28 ++ RP_L4/RPL4CONSOLE.cpp | 14 + RP_L4/RPL4ENVIRON.cpp | 35 +++ RP_L4/RPL4SPOOLSTATS.cpp | 526 ++++++++++++++++++++++++++++++++++ RP_L4/RP_L4.vcxproj | 1 + docs/NET-TEST.md | 85 ++++++ 20 files changed, 1994 insertions(+), 58 deletions(-) create mode 100644 RP_L4/RPL4SPOOLSTATS.cpp create mode 100644 docs/NET-TEST.md diff --git a/MUNGA/APP.cpp b/MUNGA/APP.cpp index 1365dd6..33192c4 100644 --- a/MUNGA/APP.cpp +++ b/MUNGA/APP.cpp @@ -53,6 +53,27 @@ Logical return enabled ? True : False; } +// +// RP412NETLOG - the race-end network summary, on unless =0. The latch +// keeps the summary to exactly one print per mission: RunMission arms +// it, the first stop or abort fires it (see NetLogReportRace, further +// down beside the stop handlers). +// +static Logical gNetRaceReported = True; + +static Logical + NetLogEnabled() +{ + static int enabled = -1; + + if (enabled < 0) + { + const char *setting = getenv("RP412NETLOG"); + enabled = (setting != NULL && atoi(setting) == 0) ? 0 : 1; + } + return enabled ? True : False; +} + //############################################################################# //########################### Application ############################### //############################################################################# @@ -815,6 +836,14 @@ Time endInterest = Now(); // renderer executes this frame. //-------------------------------------------------------------------------- // + // + // Flush queued sends before the render: the present can block on + // vsync, and this frame's state updates should be on the wire while + // that happens, not behind it. + // + Check(networkManager); + networkManager->FlushSends(); + SET_RENDERER_MANAGER(); Time startRender = Now(); Check(rendererManager); @@ -1720,6 +1749,20 @@ void Tell("Application::RunMissionMessageHandler - running mission\n"); applicationState.SetState(RunningMission); gameStarted = Now(); + + // + // Mission t0, in this machine's ticks (RP412NETLOG). Every + // machine stamps its own t0 when this message is PROCESSED, so + // the mission clocks - and everything phase-derived from them, + // the clockwork doors included - skew by the one-way spread of + // the console's RunMission sends plus frame quantization. This + // line plus the console's per-pod send ticks plus the netclock + // offsets are what let the skew be computed offline for the + // first time (plan item M4). + // + DEBUG_STREAM << "NetLog: mission t0 at tick " << gameStarted.ticks + << "\n" << std::flush; + gNetRaceReported = False; // arm the race-end summary break; case WaitingForLaunch: @@ -1855,6 +1898,96 @@ void } } +// +//############################################################################# +// The race-end network summary (RP412NETLOG, on unless =0). +// +// Playtest logs used to carry ZERO player-symptom lines - every reported +// tick, warp and stall lived in commit prose while sixteen logs said +// nothing. These few lines are the fix: per-replicant race totals +// (accumulated ungated in Entity::netRaceStats) and the per-peer clock +// lines, printed once when the first stop or abort lands, before +// teardown. The latch (gNetRaceReported, defined up beside RPCameraLog) +// is armed by RunMission, so menu traffic never prints and a second +// stop cannot print twice. +//############################################################################# +// +// A percentile out of the log2-millisecond histogram, reported as the +// bucket's lower bound: "~32 ms" means the value fell in [32, 64). +// +static unsigned long + NetLogPercentileMs( + const unsigned long *buckets, + unsigned long total, + unsigned long percent) +{ + if (total == 0) + { + return 0; + } + unsigned long target = (total * percent + 99) / 100; + unsigned long seen = 0; + for (int b = 0; b < 16; ++b) + { + seen += buckets[b]; + if (seen >= target) + { + return (b == 0) ? 0 : (1UL << (b - 1)); + } + } + return 1UL << 14; +} + +static void + NetLogReportRace(Application *the_application) +{ + if (gNetRaceReported || !NetLogEnabled()) + { + return; + } + gNetRaceReported = True; + + DEBUG_STREAM << "NetLog: race summary at tick " << Now().ticks + << ", elapsed " << the_application->GetMissionElapsed() + << " s\n" << std::flush; + + HostManager::DynamicReplicantEntityIterator + replicants(the_application->GetHostManager()); + Entity *entity; + + while ((entity = replicants.ReadAndNext()) != NULL) + { + Check(entity); + Entity::NetRaceStats *stats = &entity->netRaceStats; + if (stats->updateCount == 0) + { + continue; + } + DEBUG_STREAM << "NetLog: pod " << entity->GetEntityID() + << " host " << entity->GetOwnerID() + << ": " << stats->updateCount << " updates" + << ", interval med/p95 ~" + << NetLogPercentileMs(stats->gapHistogram, stats->updateCount, 50) + << "/~" + << NetLogPercentileMs(stats->gapHistogram, stats->updateCount, 95) + << " ms, widest " << stats->widestGapSeconds << " s, " + << stats->longGapCount << " long / " + << stats->queuedGapCount << " queued, " + << stats->snapCount << " snaps, corrections " + << stats->correctionCount; + if (stats->correctionCount > 0) + { + DEBUG_STREAM << " mean " + << (stats->correctionTotal / (Scalar) stats->correctionCount) + << " m worst " << stats->correctionWorst << " m"; + } + DEBUG_STREAM << ", stale " << stats->staleCount + << "\n" << std::flush; + } + + NetClock_ReportRaceStats(); +} + // //############################################################################# // StopMissionMessageHandler @@ -1873,6 +2006,12 @@ void Check(message); Verify(message->messageID == StopMissionMessageID); + // + // The network race summary goes out on the FIRST stop, while the + // replicants still exist (teardown is downstream of here). + // + NetLogReportRace(this); + // //-------------------------------------------------------------------------- // If the application is already stopping then ignore the message @@ -1930,6 +2069,9 @@ void Check(message); Verify(message->messageID == AbortMissionMessageID); + // same first-stop summary as StopMission - an aborted race counts + NetLogReportRace(this); + // //-------------------------------------------------------------------------- // If the application is already stopping then ignore the message diff --git a/MUNGA/ENTITY.cpp b/MUNGA/ENTITY.cpp index 89e1ff3..f4538b7 100644 --- a/MUNGA/ENTITY.cpp +++ b/MUNGA/ENTITY.cpp @@ -453,6 +453,28 @@ void // Reported per entity, on a five second clock, so a busy race // does not bury the log. // + // + // Race totals for the RP412NETLOG summary - per entity, + // ungated (one subtract and a length per arriving update), + // where the CamLog block below is windowed, gated, and + // shared across all replicants. + // + if (GetInstance() == ReplicantInstance) + { + Vector3D race_drift; + race_drift.Subtract( + update->localOrigin.linearPosition, + localOrigin.linearPosition + ); + Scalar race_distance = race_drift.Length(); + netRaceStats.correctionCount++; + netRaceStats.correctionTotal += race_distance; + if (race_distance > netRaceStats.correctionWorst) + { + netRaceStats.correctionWorst = race_distance; + } + } + if (RPCameraLog() && GetInstance() == ReplicantInstance) { static Scalar next_say = 0.0f; diff --git a/MUNGA/ENTITY.h b/MUNGA/ENTITY.h index 115f94b..9fccc73 100644 --- a/MUNGA/ENTITY.h +++ b/MUNGA/ENTITY.h @@ -186,6 +186,77 @@ public: // Logical renderStepTaken; + //###################################################################### + //################ Race-total network statistics ################### + //###################################################################### + // + // Whole-race arrival statistics for a REPLICANT entity, accumulated + // ungated (the arithmetic is a subtract and a length per arriving + // update) and printed once at mission end under RP412NETLOG - so a + // playtest log finally carries the symptoms instead of nothing. The + // windowed CamLog counters near these update sites reset every trace + // print and cover one latched entity; these cover every replicant for + // the whole race. + // + // gapHistogram is log2 milliseconds: bucket 0 is a sub-millisecond + // gap, bucket b covers [2^(b-1), 2^b) ms, bucket 15 collects + // everything from ~16 s up. Median and p95 fall out of a cumulative + // walk at print time. + // + class NetRaceStats + { + public: + unsigned long updateCount; // updates applied to this entity + Scalar widestGapSeconds; // worst inter-arrival gap (<10 s stream) + unsigned long longGapCount; // gaps > 0.200 s (sender quiet) + unsigned long queuedGapCount; // gaps < 0.005 s (our loop stalled) + unsigned long snapCount; // lerp->snap transitions while flowing + unsigned long staleCount; // rejected out-of-order records (0 + // until the stale guard ships) + unsigned long correctionCount; // arriving updates that moved us + Scalar correctionTotal; // metres, for the mean + Scalar correctionWorst; // metres + unsigned long gapHistogram[16]; + Logical wasSnapped; // edge detector for snapCount + + NetRaceStats(): + updateCount(0), + widestGapSeconds(0.0f), + longGapCount(0), + queuedGapCount(0), + snapCount(0), + staleCount(0), + correctionCount(0), + correctionTotal(0.0f), + correctionWorst(0.0f), + wasSnapped(False) + { + for (int i = 0; i < 16; ++i) + { + gapHistogram[i] = 0; + } + } + + void + CountGap(Scalar seconds) + { + if (seconds > widestGapSeconds) + { + widestGapSeconds = seconds; + } + unsigned long ms = (unsigned long) (seconds * 1000.0f); + int bucket = 0; + while (ms != 0 && bucket < 15) + { + ms >>= 1; + ++bucket; + } + gapHistogram[bucket]++; + } + }; + + NetRaceStats netRaceStats; + // // The transform to DRAW with. Falls back to localToWorld verbatim when // interpolation is off, when there is no fixed step to interpolate diff --git a/MUNGA/MOVER.cpp b/MUNGA/MOVER.cpp index e54cbba..47ad378 100644 --- a/MUNGA/MOVER.cpp +++ b/MUNGA/MOVER.cpp @@ -627,6 +627,7 @@ void // for the RP412CAMLOG trace at the end of this function gLastLerpUsed = True; gLastPercent = percent; + netRaceStats.wasSnapped = False; // //------------------------------------------ @@ -678,6 +679,19 @@ void else { gLastLerpUsed = False; // snapped, not blended + + // + // Race totals: count the TRANSITION into snap mode, and only + // while updates are flowing - a departed sender's replicant + // snaps every step forever and would bury the number. + // + if (!netRaceStats.wasSnapped && + (lastPerformance - lastUpdate) < 2.0f) + { + netRaceStats.snapCount++; + } + netRaceStats.wasSnapped = True; + localOrigin = projectedOrigin; worldLinearVelocity = projectedVelocity.linearMotion; localVelocity.angularMotion = projectedVelocity.angularMotion; @@ -1149,6 +1163,21 @@ void nextUpdate = Now(); Scalar diff = nextUpdate - lastUpdate; Scalar anchorInterval = (Scalar) 0; + + // + // Race totals for the RP412NETLOG summary: every update + // counts; gaps only within a live stream (the same <10 s + // bound the predictor uses), so a menu pause or join does + // not pollute the histogram. + // + netRaceStats.updateCount++; + if (diff < 10.0f) + { + netRaceStats.CountGap(diff); + if (diff > kLongGapThreshold) { netRaceStats.longGapCount++; } + if (diff < kQueuedGapThreshold) { netRaceStats.queuedGapCount++; } + } + if (diff < 10.0f) { if (UseMedianPrediction()) diff --git a/MUNGA/NETWORK.h b/MUNGA/NETWORK.h index 5e445af..9489796 100644 --- a/MUNGA/NETWORK.h +++ b/MUNGA/NETWORK.h @@ -137,6 +137,16 @@ public: virtual Logical ExecuteBackground(); + // + // Retry sends a connection could not take earlier. The application + // calls this at its flush points; the base engine has no queues, so + // the default is a no-op (L4 forwards it to the wire transport). + // + virtual void + FlushSends() + { + } + NetworkClient* GetNetworkClientPointer(ClientID client_id); diff --git a/MUNGA/PLAYER.cpp b/MUNGA/PLAYER.cpp index a460ab6..501d025 100644 --- a/MUNGA/PLAYER.cpp +++ b/MUNGA/PLAYER.cpp @@ -218,6 +218,81 @@ void Check_Fpu(); } +//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// +// A drop zone is only usable if the machine that owns it is still with +// us. Zones are map entities dealt round-robin across the hosts at load, +// and ownership transfer is not implemented - so a leaver's pads stay in +// the DropZones group forever, a request dispatched to one is silently +// dropped at the send ('Send: no host N in the table'), and the fry +// retry re-dispatches to the same dead owner every two seconds: the +// player never respawns. Locally-owned zones are always usable. +// +static Logical + DropZoneOwnerReachable(DropZone *dropzone) +{ + Check(dropzone); + if (dropzone->GetInstance() != Entity::ReplicantInstance) + { + return True; + } + Check(application); + HostManager *host_manager = application->GetHostManager(); + Check(host_manager); + Host *owner = host_manager->GetRemoteHost(dropzone->GetOwnerID()); + return + (owner != NULL && + owner->GetConnectStatus() == Host::OnLineConnectionStatus) ? + True : False; +} + +// +// The closest non-podium drop zone, optionally restricted to pads whose +// owner is still connected. One function because the dead-owner case +// needs a second pass: when every pad's owner is gone the old behaviour +// (closest pad regardless) is the graceful floor to land on. +// +static DropZone * + Player_FindClosestDropZone( + Player *player, + Logical require_reachable_owner) +{ + EntityManager *entity_manager = application->GetEntityManager(); + Check(entity_manager); + EntityGroup *dropzones = entity_manager->FindGroup("DropZones"); + Check(dropzones); + ChainIteratorOf iterator(dropzones->groupMembers); + + DropZone *dropzone; + DropZone *closest_dropzone = NULL; + Vector3D range; + Scalar closest_range = 0.0f; + + while ((dropzone = (DropZone*)iterator.ReadAndNext()) != NULL) + { + Check(dropzone); + if (!strnicmp(dropzone->GetDropZoneName(), "win", 3)) + { + continue; + } + if (require_reachable_owner && !DropZoneOwnerReachable(dropzone)) + { + continue; + } + range.Subtract( + player->localOrigin.linearPosition, + dropzone->localOrigin.linearPosition + ); + Scalar len = range.LengthSquared(); + if (closest_dropzone == NULL || len < closest_range) + { + closest_dropzone = dropzone; + closest_range = len; + } + } + return closest_dropzone; +} + //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ // void @@ -242,11 +317,32 @@ void { DropZone *dropzone; - + // + //--------------------------------------------------------------------- + // A pad assigned on an earlier try may have died with its owner - the + // fry retry reuses message->dropZoneID forever, so left unchecked it + // re-dispatches to the same dead host every two seconds. Reset it and + // let the closest-pad scan below pick a live one. + //--------------------------------------------------------------------- + // + if (message->dropZoneID != EntityID::Null) + { + HostManager *host_manager = application->GetHostManager(); + Check(host_manager); + Entity *assigned = host_manager->GetEntityPointer(message->dropZoneID); + if (assigned == NULL || !DropZoneOwnerReachable((DropZone*)assigned)) + { + DEBUG_STREAM << "Respawn: assigned drop zone " + << ((assigned == NULL) ? "is gone" : "belongs to a departed host") + << " - picking a new pad\n" << std::flush; + message->dropZoneID = EntityID::Null; + } + } // //--------------------------------------------------------------------- // If no dropzone has been assigned to this death, find the closest one + // whose owner is still connected //--------------------------------------------------------------------- // if (message->dropZoneID == EntityID::Null) @@ -254,52 +350,19 @@ void //DEBUG_STREAM << " Find Closest" << endl << std::flush; - EntityManager *entity_manager = application->GetEntityManager(); - Check(entity_manager); - EntityGroup *dropzones = entity_manager->FindGroup("DropZones"); - Check(dropzones); - ChainIteratorOf iterator(dropzones->groupMembers); - - DropZone *closest_dropzone; - Vector3D range; - Scalar closest_range; - - while ((dropzone = (DropZone*)iterator.ReadAndNext()) != NULL) + DropZone *closest_dropzone = Player_FindClosestDropZone(this, True); + if (closest_dropzone == NULL) { - Check(dropzone); - - if (!strnicmp(dropzone->GetDropZoneName(), "win", 3)) - { - continue; - } - - closest_dropzone = dropzone; - range.Subtract( - localOrigin.linearPosition, - dropzone->localOrigin.linearPosition - ); - closest_range = range.LengthSquared(); - break; - } - - while ((dropzone = (DropZone*)iterator.ReadAndNext()) != NULL) - { - Check(dropzone); - if (!strnicmp(dropzone->GetDropZoneName(), "win", 3)) - { - continue; - } - range.Subtract( - localOrigin.linearPosition, - dropzone->localOrigin.linearPosition - ); - Scalar len = range.LengthSquared(); - if (len < closest_range) - { - closest_dropzone = dropzone; - closest_range = len; - } + // + // Every pad's owner is gone. Fall back to the old behaviour + // (closest pad regardless) so the retry loop keeps its old + // shape rather than crashing here. + // + DEBUG_STREAM << "Respawn: no drop zone with a live owner - " + << "falling back to the closest regardless\n" << std::flush; + closest_dropzone = Player_FindClosestDropZone(this, False); } + Check(closest_dropzone); message->dropZoneID = closest_dropzone->GetEntityID(); dropzone = closest_dropzone; } diff --git a/MUNGA/SIMULATE.cpp b/MUNGA/SIMULATE.cpp index d07bc3c..ea57d16 100644 --- a/MUNGA/SIMULATE.cpp +++ b/MUNGA/SIMULATE.cpp @@ -298,6 +298,16 @@ namespace long offsetTicks; // our clock - their clock long windowMinTicks; int windowCount; + + // + // Race totals for the RP412NETLOG summary: how often a window + // close actually moved the estimate (>5 ms), and the worst such + // step. This is the decision input for whether a bounded slew is + // ever needed - a step mid-race means SDR rerouted or a clock + // drifted, and today it lands as one hard jump. + // + unsigned long raceStepCount; + long raceStepWorstMs; }; PeerClock gPeerClocks[netClockMaxPeers]; @@ -345,6 +355,8 @@ namespace free_slot->offsetTicks = 0; free_slot->windowMinTicks = 0; free_slot->windowCount = 0; + free_slot->raceStepCount = 0; + free_slot->raceStepWorstMs = 0; } return free_slot; } @@ -367,6 +379,30 @@ void NetClock_Reset() gUpdateSenderValid = False; } +// +// Race-end summary, one line per peer (RP412NETLOG). The offset's +// absolute value only says how far apart the two processes launched +// (clocks are ms-since-process-start); the movement counters are the +// health signal - a mid-race step means SDR rerouted or a clock +// drifted, and today each one lands as a hard jump in every replicant +// from that peer. This is the decision input for the deferred bounded +// slew (plan item D2). +// +void NetClock_ReportRaceStats() +{ + for (int i = 0; i < netClockMaxPeers; ++i) + { + PeerClock *peer = &gPeerClocks[i]; + if (peer->inUse && peer->settled) + { + DEBUG_STREAM << "NetLog: peer " << peer->host + << ": netclock offset " << peer->offsetTicks + << " ms, window steps >5ms " << peer->raceStepCount + << ", worst " << peer->raceStepWorstMs << " ms\n" << std::flush; + } + } +} + //~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ // void @@ -424,6 +460,16 @@ void // slower rather than staying pinned to one old packet. // long moved = peer->windowMinTicks - peer->offsetTicks; + long moved_abs = (moved < 0) ? -moved : moved; + if (moved_abs > 5) + { + // race totals for the NetLog summary (see PeerClock) + peer->raceStepCount++; + if (moved_abs > peer->raceStepWorstMs) + { + peer->raceStepWorstMs = moved_abs; + } + } if (moved > 50 || moved < -50) { DEBUG_STREAM << "NetClock: host " << peer->host diff --git a/MUNGA/SIMULATE.h b/MUNGA/SIMULATE.h index 69fe855..78d93d3 100644 --- a/MUNGA/SIMULATE.h +++ b/MUNGA/SIMULATE.h @@ -39,6 +39,7 @@ void NetClock_BeginUpdate(HostID sender); // around one message's records void NetClock_EndUpdate(); void NetClock_Reset(); // forget every peer (new mission) +void NetClock_ReportRaceStats(); // RP412NETLOG race-end peer lines class Simulation__SharedData; class Simulation__IndexData; diff --git a/MUNGA_L4/L4NET.CPP b/MUNGA_L4/L4NET.CPP index d0ca059..1afc36c 100644 --- a/MUNGA_L4/L4NET.CPP +++ b/MUNGA_L4/L4NET.CPP @@ -64,12 +64,22 @@ #define CLEAR_LOST_DATA() #endif -#define BATCHED_TRANSMIT True // true sends broadcast messages with a single call to netnub -#define BATCHED_RECEIVE True // true receives from all streams with a single call to netnub -#define REPORT_LOST_DATA False // Print a message if data is lost (may be toxic) +// +// These three were written as `True`, which is an ENUM (style.h), not a +// macro - so the preprocessor evaluated the bare identifiers as 0 and +// the batched/buffered netnub paths never compiled. That accident is +// what actually shipped in every arcade and every 4.12 build; the dead +// branches reference netnub symbols that no longer exist, so "enabling" +// them cannot compile. Spelled as literal 0 now so the guards say what +// the build does. Send-side loss handling lives in the transport's +// send queue these days (RP412NETSENDQ, l4nettransport.h). +// +#define BATCHED_TRANSMIT 0 // dead netnub path - see note above +#define BATCHED_RECEIVE 0 // dead netnub path - see note above +#define REPORT_LOST_DATA 0 // dead netnub path - see note above #if !defined(MESSAGE_BUFFERING) - #define MESSAGE_BUFFERING True // true uses munga level buffering for dropped packets + #define MESSAGE_BUFFERING 0 // dead netnub path - see note above #endif #define CONSOLE_NET_PORT 1501 // Port number the console will connect on @@ -416,6 +426,13 @@ void // the game port is the console listening port + 1 unsigned short localGamePort = (unsigned short)((L4Application *)application)->GetNetworkCommonFlatAddress() + 1; + // + // Flush before the connect sequence: the egg-ACK ordering the mesh + // relies on must not have console bytes parked in a queue while a + // blocking connect holds the frame. + // + NetTransport_Get()->FlushSends(); + // // This should be entered with no network connections to other game hosts up, // so we initialize this count to zero. @@ -2120,8 +2137,25 @@ Logical // line, waiting for connections and so on. // //WinSock support :ADB 01/06/07 +//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// L4NetworkManager::FlushSends Retry anything the wire refused when it was +// first sent. Forwards to the transport's queue drain; called from the +// application's flush points (before the render, at the top of the receive +// pump, and before a connect sequence starts). +// +void + L4NetworkManager::FlushSends() +{ + NetTransport_Get()->FlushSends(); +} + Logical L4NetworkManager::CheckBuffers(NetworkPacket *network_packet) { + // + // Every receive-pump round is also a send flush point, so bytes a + // stalled peer could not take retry at least as often as we poll. + // + NetTransport_Get()->FlushSends(); //char *current_receive_ptr; int // i, @@ -2513,7 +2547,8 @@ Logical { short i, - host_count, + host_count; + int move_size, receive_packet_size; Host @@ -2566,8 +2601,35 @@ Logical // figure how many bytes the complete packet should be // incoming_packet = (NetworkPacket*)remote_host->pad_buffer; - receive_packet_size = (short)(incoming_packet->messageData.messageLength + sizeof(NetworkPacketHeader)); - Verify(receive_packet_size > 0); + // + // The length prefix is untrusted input off the wire. Framing on + // this stream is recovered purely from it, so a corrupt or + // sheared length is unrecoverable: too small and the stream + // never advances, too large and it either overruns the caller's + // NETWORKMANAGER_BUFFER_SIZE buffer or exceeds the pad so the + // packet can never complete. Any of those means the stream can + // no longer be re-framed - treat it exactly like a hard + // disconnect (the Verify that used to stand here compiles away + // in release). The lower bound mirrors Send's own floor of 8. + // + if((incoming_packet->messageData.messageLength < 8) || + (incoming_packet->messageData.messageLength > + NETWORKMANAGER_BUFFER_SIZE - sizeof(NetworkPacketHeader))) + { + DEBUG_STREAM << "Receive: host " << remote_host->GetHostID() + << " framing broken (message length " + << (unsigned long)incoming_packet->messageData.messageLength + << ") - dropping the connection\n" << std::flush; + remote_host->pad_tail = 0; + HostDisconnectedMessage myHostDisconnected( + remote_host->GetHostID(), + remote_host->GetNetworkSocket()); + NetworkClient *dead_stream_client = GetNetworkClientPointer(NetworkManagerClientID); + Check(dead_stream_client); + dead_stream_client->ReceiveNetworkPacket(NULL, &myHostDisconnected); + continue; + } + receive_packet_size = (int)(incoming_packet->messageData.messageLength + sizeof(NetworkPacketHeader)); // Are there enough bytes in the buffer to make up this packet? if(remote_host->pad_tail >= receive_packet_size) { @@ -2578,7 +2640,7 @@ Logical remote_host->pad_buffer, receive_packet_size, NETWORKMANAGER_BUFFER_SIZE); - move_size = (short)(remote_host->pad_tail - receive_packet_size); + move_size = (int)(remote_host->pad_tail - receive_packet_size); // don't do the next step if the buffer is empty // we need to use memmove because the source and destination addresses overlap // and memcopy doesn't know how to deal with that. diff --git a/MUNGA_L4/L4NET.H b/MUNGA_L4/L4NET.H index eecd95f..8a8dbc3 100644 --- a/MUNGA_L4/L4NET.H +++ b/MUNGA_L4/L4NET.H @@ -209,6 +209,9 @@ public: void StartConnecting(Mission *mission); + // forwards to the wire transport's queue drain (l4nettransport.h) + void FlushSends(); + Logical Shutdown(); Logical CheckBuffers(NetworkPacket *packet); diff --git a/MUNGA_L4/L4NETTRANSPORT.cpp b/MUNGA_L4/L4NETTRANSPORT.cpp index 7fa4db9..8f8db3e 100644 --- a/MUNGA_L4/L4NETTRANSPORT.cpp +++ b/MUNGA_L4/L4NETTRANSPORT.cpp @@ -134,6 +134,213 @@ void gWaitCancelled = False; } +//######################################################################## +// Outbound send queue - shared machinery for both transports. +// See the block comment in l4nettransport.h for why this exists. +//######################################################################## + +Logical + NetTransport_SendQueueEnabled() +{ + static int cached = -1; + + if (cached < 0) + { + const char *setting = getenv("RP412NETSENDQ"); + + cached = (setting != NULL && atoi(setting) == 0) ? 0 : 1; + } + return cached ? True : False; +} + +int + NetTransport_SendQueueCapBytes() +{ + static int cached = -1; + + if (cached < 0) + { + const char *setting = getenv("RP412NETQUEUEKB"); + int kilobytes = (setting != NULL) ? atoi(setting) : 256; + + if (kilobytes < 16) { kilobytes = 16; } + if (kilobytes > 4096) { kilobytes = 4096; } + cached = kilobytes * 1024; + } + return cached; +} + +Logical + NetTransport_StatsEnabled() +{ + static int cached = -1; + + if (cached < 0) + { + const char *setting = getenv("RP412NETSTATS"); + cached = (setting != NULL && atoi(setting) != 0) ? 1 : 0; + } + return cached ? True : False; +} + +NetSendQueueTable::NetSendQueueTable() +{ + for (int i = 0; i < MaxQueues; ++i) + { + queues[i].connection = NetTransport::InvalidConnection; + queues[i].buffer = NULL; + queues[i].head = 0; + queues[i].tail = 0; + queues[i].dead = False; + queues[i].sentMessages = 0; + queues[i].sentBytes = 0; + queues[i].wireWrites = 0; + queues[i].partialWrites = 0; + queues[i].wouldBlocks = 0; + queues[i].queuedHighWater = 0; + } +} + +NetSendQueueTable::~NetSendQueueTable() +{ + ReleaseAll(); +} + +NetSendQueue * + NetSendQueueTable::Find(NetTransport::Connection connection) +{ + for (int i = 0; i < MaxQueues; ++i) + { + if (queues[i].connection == connection) + { + return &queues[i]; + } + } + return NULL; +} + +NetSendQueue * + NetSendQueueTable::FindOrAdopt(NetTransport::Connection connection) +{ + NetSendQueue *queue = Find(connection); + + if (queue != NULL) + { + return queue; + } + for (int i = 0; i < MaxQueues; ++i) + { + if (queues[i].connection == NetTransport::InvalidConnection) + { + queue = &queues[i]; + queue->connection = connection; + queue->head = 0; + queue->tail = 0; + queue->dead = False; + queue->sentMessages = 0; + queue->sentBytes = 0; + queue->wireWrites = 0; + queue->partialWrites = 0; + queue->wouldBlocks = 0; + queue->queuedHighWater = 0; + return queue; + } + } + DEBUG_STREAM << "NetSendQueue: table full - connection " + << (unsigned long) connection << " gets no queue\n" << std::flush; + return NULL; +} + +void + NetSendQueueTable::Release(NetTransport::Connection connection) +{ + NetSendQueue *queue = Find(connection); + + if (queue == NULL) + { + return; + } + if (queue->buffer != NULL) + { + free(queue->buffer); + queue->buffer = NULL; + } + queue->connection = NetTransport::InvalidConnection; + queue->head = 0; + queue->tail = 0; + queue->dead = False; +} + +void + NetSendQueueTable::ReleaseAll() +{ + for (int i = 0; i < MaxQueues; ++i) + { + if (queues[i].connection != NetTransport::InvalidConnection) + { + Release(queues[i].connection); + } + } +} + +Logical + NetSendQueueTable::Append( + NetSendQueue *queue, + const void *data, + int size) +{ + int capacity = NetTransport_SendQueueCapBytes(); + int pending = queue->PendingBytes(); + + if (pending + size > capacity) + { + // + // Overflow. These are reliable ordered messages - dropping any of + // them desyncs the stream - so the only honest answer is to give + // up on the connection and let the disconnect path run. + // + DEBUG_STREAM << "NetSendQueue: connection " + << (unsigned long) queue->connection << " overflowed (" + << pending << " pending + " << size << " > " << capacity + << ") - declaring it dead\n" << std::flush; + if (queue->buffer != NULL) + { + free(queue->buffer); + queue->buffer = NULL; + } + queue->head = 0; + queue->tail = 0; + queue->dead = True; + return False; + } + if (queue->buffer == NULL) + { + queue->buffer = (unsigned char *) malloc(capacity); + if (queue->buffer == NULL) + { + queue->dead = True; + return False; + } + } + // + // Rejustify before appending when the tail would run off the end - + // the bytes ahead of head are already on the wire. + // + if (queue->tail + size > capacity && queue->head > 0) + { + memmove(queue->buffer, &queue->buffer[queue->head], pending); + queue->head = 0; + queue->tail = pending; + } + memcpy(&queue->buffer[queue->tail], data, size); + queue->tail += size; + if ((unsigned long) queue->PendingBytes() > queue->queuedHighWater) + { + queue->queuedHighWater = (unsigned long) queue->PendingBytes(); + } + return True; +} + //######################################################################## // WinsockNetTransport - the TCP wire the arcade always used, moved // verbatim out of L4NET.CPP behind the NetTransport seam. Behavior @@ -152,7 +359,8 @@ namespace { public: WinsockNetTransport(): - started(False) + started(False), + statsNextTick(0) { } @@ -179,6 +387,8 @@ namespace { if (started) { + FlushSends(); + sendQueues.ReleaseAll(); WSACleanup(); started = False; } @@ -430,6 +640,19 @@ namespace void Close(Connection connection) { + // + // Best-effort last drain, so an orderly close does not strand + // bytes a slow connection still owes the wire. + // + NetSendQueue *queue = sendQueues.Find(connection); + if (queue != NULL) + { + if (!queue->dead && queue->PendingBytes() > 0) + { + DrainQueue(queue); + } + sendQueues.Release(connection); + } shutdown((SOCKET) connection, SD_BOTH); closesocket((SOCKET) connection); } @@ -441,7 +664,151 @@ namespace int size ) { - return send((SOCKET) connection, (const char *) data, size, 0); + NetSendQueue *queue = sendQueues.Find(connection); + + if (queue != NULL && queue->dead) + { + return -1; + } + + if (!NetTransport_SendQueueEnabled()) + { + // + // Legacy direct send (RP412NETSENDQ=0) - but count and + // name the losses the queue would have prevented. + // + int sent = send((SOCKET) connection, (const char *) data, size, 0); + if (sent != size) + { + static int said = 0; + if (said < 8) + { + ++said; + DEBUG_STREAM << "WinsockNetTransport::Send: legacy mode " + << ((sent < 0) ? "lost a message of " : "sheared the stream at ") + << size << " bytes (send returned " << sent + << ", error " << WSAGetLastError() << ")\n" << std::flush; + } + } + return sent; + } + + // + // Every live connection gets a table slot (the buffer itself + // stays unallocated until something actually queues), so the + // RP412NETSTATS counters exist for healthy connections too. + // + if (queue == NULL) + { + queue = sendQueues.FindOrAdopt(connection); + if (queue == NULL) + { + // table full: behave exactly like the legacy send + return send((SOCKET) connection, (const char *) data, size, 0); + } + } + queue->sentMessages++; + queue->sentBytes += size; + + // + // FIFO discipline: bytes already queued must go first, so a + // queued connection only ever appends. + // + if (queue->PendingBytes() > 0) + { + if (!sendQueues.Append(queue, data, size)) + { + return -1; + } + DrainQueue(queue); + return size; + } + + int sent = send((SOCKET) connection, (const char *) data, size, 0); + if (sent == size) + { + queue->wireWrites++; + return size; + } + + // + // The wire would not take all of it. Keep the byte-exact + // remainder - a partial send left unhealed is a sheared + // stream, the failure this queue exists to close. + // + if (sent < 0) + { + DWORD error = WSAGetLastError(); + if (error != WSAEWOULDBLOCK) + { + DEBUG_STREAM << "WinsockNetTransport::Send: hard error " + << error << " - declaring the connection dead\n" << std::flush; + queue->dead = True; + return -1; + } + queue->wouldBlocks++; + sent = 0; + } + else + { + queue->wireWrites++; + queue->partialWrites++; + } + if (!sendQueues.Append(queue, (const char *) data + sent, size - sent)) + { + return -1; + } + return size; + } + + void + FlushSends() + { + for (int i = 0; i < NetSendQueueTable::MaxQueues; ++i) + { + NetSendQueue *queue = &sendQueues.queues[i]; + + if (queue->connection != InvalidConnection && + !queue->dead && + queue->PendingBytes() > 0) + { + DrainQueue(queue); + } + } + + // + // RP412NETSTATS: 5 s per-connection rollup, from here because + // FlushSends is the one transport entry the engine calls on a + // steady cadence. + // + if (NetTransport_StatsEnabled()) + { + DWORD now = GetTickCount(); + if ((LONG) (now - statsNextTick) >= 0) + { + statsNextTick = now + 5000; + for (int j = 0; j < NetSendQueueTable::MaxQueues; ++j) + { + NetSendQueue *queue = &sendQueues.queues[j]; + if (queue->connection == InvalidConnection || + queue->sentMessages == 0) + { + continue; + } + DEBUG_STREAM << "NetStats: tcp conn " + << (unsigned long) queue->connection << ": " + << queue->sentMessages << " msgs " + << queue->sentBytes << " B, " + << queue->wireWrites << " writes (" + << queue->partialWrites << " partial, " + << queue->wouldBlocks << " would-block), queued high " + << queue->queuedHighWater << " B, pending " + << queue->PendingBytes() + << (queue->dead ? " B, DEAD" : " B") + << "\n" << std::flush; + } + } + } } int @@ -451,6 +818,16 @@ namespace int size ) { + // + // A connection the send side declared dead reports as a + // disconnect here, so the one disconnect path upstairs runs. + // + NetSendQueue *queue = sendQueues.Find(connection); + if (queue != NULL && queue->dead) + { + return ReceiveDisconnected; + } + int received = recv((SOCKET) connection, (char *) buffer, size, 0); if (received > 0) { @@ -488,8 +865,55 @@ namespace } private: - Logical started; - WSADATA winsockData; + // + // Push a queue's pending bytes at the wire until it is empty or + // the wire refuses. Partial progress advances head byte-exactly. + // + void + DrainQueue(NetSendQueue *queue) + { + while (queue->PendingBytes() > 0 && !queue->dead) + { + int sent = send( + (SOCKET) queue->connection, + (const char *) &queue->buffer[queue->head], + queue->PendingBytes(), 0); + + if (sent > 0) + { + queue->wireWrites++; + queue->head += sent; + if (queue->PendingBytes() == 0) + { + queue->head = 0; + queue->tail = 0; + } + else + { + queue->partialWrites++; + break; + } + continue; + } + + DWORD error = WSAGetLastError(); + if (sent < 0 && error == WSAEWOULDBLOCK) + { + queue->wouldBlocks++; + break; + } + DEBUG_STREAM << "WinsockNetTransport::DrainQueue: hard error " + << error << " with " << queue->PendingBytes() + << " bytes pending - declaring the connection dead\n" << std::flush; + queue->dead = True; + break; + } + } + + Logical started; + WSADATA winsockData; + NetSendQueueTable sendQueues; + DWORD statsNextTick; }; WinsockNetTransport gWinsockTransport; diff --git a/MUNGA_L4/L4NETTRANSPORT.h b/MUNGA_L4/L4NETTRANSPORT.h index 705aa73..5336fe0 100644 --- a/MUNGA_L4/L4NETTRANSPORT.h +++ b/MUNGA_L4/L4NETTRANSPORT.h @@ -110,6 +110,15 @@ public: int size ) = 0; + // Retry anything a connection could not take when Send was called. + // The engine calls this at its flush points (before the render, at + // the top of the receive pump, before a connect sequence); default + // is a no-op so a transport without queues needs nothing. + virtual void + FlushSends() + { + } + // remote endpoint of a live connection (mesh identity checks) virtual Logical GetRemoteAddress( @@ -125,6 +134,98 @@ NetTransport * void NetTransport_Set(NetTransport *transport); +//######################################################################## +// Outbound send queue (RP412NETSENDQ). +// +// Historically every send result was discarded. On the 1 ms arcade LAN +// the socket buffer never filled, so nothing was ever lost - but over +// the internet a peer stalled in its 10-30 s mission load stops reading, +// its window closes, and our nonblocking send starts returning +// would-block (message silently vanished) or a PARTIAL count. A partial +// send is the worse of the two: framing on the stream is recovered +// purely from each message's length prefix, so the bytes that never +// followed shear the stream for good. +// +// The queue closes both holes: a connection that cannot take the bytes +// now keeps them, byte-exact, and FlushSends() retries at the engine's +// flush points. Nothing is ever dropped from the middle of the stream - +// these are reliable, ordered, non-idempotent messages, so the only +// legal overflow response is to declare the connection dead and let the +// normal disconnect path run (Receive() reports it). +// +// RP412NETSENDQ=0 restores the old direct sends but keeps counting and +// logging the losses it would have prevented. +//######################################################################## + +Logical + NetTransport_SendQueueEnabled(); // RP412NETSENDQ, default on +int + NetTransport_SendQueueCapBytes(); // RP412NETQUEUEKB * 1024, default 256K +Logical + NetTransport_StatsEnabled(); // RP412NETSTATS, default off: 5 s + // per-connection rollups (and, on + // Steam, a 1 Hz ping/quality poll) + // printed from FlushSends + +class NetSendQueue +{ +public: + NetTransport::Connection + connection; // InvalidConnection marks a free slot + unsigned char + *buffer; // allocated on first append + int + head, // first unsent byte + tail; // one past the last queued byte + Logical + dead; // overflow or hard error; Receive() turns this + // into ReceiveDisconnected + + // telemetry, printed under RP412NETSTATS + unsigned long + sentMessages, // engine Send() calls accepted + sentBytes, + wireWrites, // actual writes on the wire + partialWrites, // wire writes that took only part of the bytes + wouldBlocks, // wire writes refused outright + queuedHighWater; // worst PendingBytes() seen + + int + PendingBytes() const + { + return tail - head; + } +}; + +class NetSendQueueTable +{ +public: + enum { MaxQueues = 24 }; // 8 pods' mesh + console + listeners + slack + + NetSendQueueTable(); + ~NetSendQueueTable(); + + NetSendQueue * + Find(NetTransport::Connection connection); + NetSendQueue * + FindOrAdopt(NetTransport::Connection connection); + void + Release(NetTransport::Connection connection); + void + ReleaseAll(); + + // Append bytes to a queue. False = overflow: the pending bytes are + // freed and the queue is marked dead (see the block comment above). + Logical + Append( + NetSendQueue *queue, + const void *data, + int size); + + NetSendQueue + queues[MaxQueues]; +}; + //######################################################################## // Waiting for a peer without hanging the window. // diff --git a/MUNGA_L4/L4STEAMTRANSPORT.cpp b/MUNGA_L4/L4STEAMTRANSPORT.cpp index bf0b5e3..55e347d 100644 --- a/MUNGA_L4/L4STEAMTRANSPORT.cpp +++ b/MUNGA_L4/L4STEAMTRANSPORT.cpp @@ -301,6 +301,11 @@ namespace public NetTransport { public: + SteamNetTransport(): + statsNextTick(0) + { + } + Logical Startup() { @@ -314,6 +319,29 @@ namespace { // mission teardown (single-binary race loop): drop the // wire but keep the Steam API up - the lobby lives on + FlushSends(); + sendQueues.ReleaseAll(); + + // + // RP412NETSTATS: one detailed status per connection on the + // way out - the route description (direct vs which relay) + // lives nowhere else. + // + if (NetTransport_StatsEnabled()) + { + for (int d = 0; d < gConnectionCount; ++d) + { + char detail[2048]; + if (SteamNetworkingSockets()->GetDetailedConnectionStatus( + gConnections[d].handle, detail, sizeof(detail)) >= 0) + { + DEBUG_STREAM << "NetStats: steam conn " + << (unsigned long) gConnections[d].handle + << " detail:\n" << detail << "\n" << std::flush; + } + } + } + for (int i = 0; i < gConnectionCount; ++i) { SteamNetworkingSockets()->CloseConnection( @@ -689,6 +717,20 @@ namespace listener->pendingCount = 0; return; } + // + // Best-effort last drain so an orderly close does not strand + // queued bytes, then let the connection linger (true) so + // Steam delivers what it already accepted. + // + NetSendQueue *queue = sendQueues.Find(connection); + if (queue != NULL) + { + if (!queue->dead && queue->PendingBytes() > 0) + { + DrainQueue(queue); + } + sendQueues.Release(connection); + } SteamNetworkingSockets()->CloseConnection( (HSteamNetConnection) connection, 0, "closed", true); RemoveConnection((HSteamNetConnection) connection); @@ -701,10 +743,163 @@ namespace int size ) { + NetSendQueue *queue = sendQueues.Find(connection); + + if (queue != NULL && queue->dead) + { + return -1; + } + + if (!NetTransport_SendQueueEnabled()) + { + // + // Legacy direct send (RP412NETSENDQ=0) - but name the + // losses the queue would have prevented. Backpressure + // (LimitExceeded) used to vanish into this return code. + // + EResult result = SteamNetworkingSockets()->SendMessageToConnection( + (HSteamNetConnection) connection, data, (uint32) size, + k_nSteamNetworkingSend_Reliable, NULL); + if (result != k_EResultOK) + { + static int said = 0; + if (said < 8) + { + ++said; + DEBUG_STREAM << "SteamNetTransport::Send: legacy mode lost a " + << size << " byte message (result " << (int) result + << ")\n" << std::flush; + } + return -1; + } + return size; + } + + if (queue == NULL) + { + queue = sendQueues.FindOrAdopt(connection); + if (queue == NULL) + { + // table full: behave exactly like the legacy send + EResult result = SteamNetworkingSockets()->SendMessageToConnection( + (HSteamNetConnection) connection, data, (uint32) size, + k_nSteamNetworkingSend_Reliable, NULL); + return (result == k_EResultOK) ? size : -1; + } + } + queue->sentMessages++; + queue->sentBytes += size; + + // + // FIFO discipline: anything already queued goes first. + // + if (queue->PendingBytes() > 0) + { + if (!sendQueues.Append(queue, data, size)) + { + MarkSendDead(connection, queue); + return -1; + } + DrainQueue(queue); + return size; + } + EResult result = SteamNetworkingSockets()->SendMessageToConnection( (HSteamNetConnection) connection, data, (uint32) size, k_nSteamNetworkingSend_Reliable, NULL); - return (result == k_EResultOK) ? size : -1; + if (result == k_EResultOK) + { + queue->wireWrites++; + return size; + } + if (result == k_EResultLimitExceeded) + { + // + // Steam's send buffer is full (a peer stalled in its + // mission load, most likely). Keep the message; the + // flush points retry it in order. + // + queue->wouldBlocks++; + if (!sendQueues.Append(queue, data, size)) + { + MarkSendDead(connection, queue); + return -1; + } + return size; + } + DEBUG_STREAM << "SteamNetTransport::Send: result " << (int) result + << " - declaring the connection dead\n" << std::flush; + MarkSendDead(connection, queue); + return -1; + } + + void + FlushSends() + { + for (int i = 0; i < NetSendQueueTable::MaxQueues; ++i) + { + NetSendQueue *queue = &sendQueues.queues[i]; + + if (queue->connection != InvalidConnection && + !queue->dead && + queue->PendingBytes() > 0) + { + DrainQueue(queue); + } + } + + // + // RP412NETSTATS: the 5 s rollup, plus the first read this + // codebase has ever taken of Steam's own connection health - + // ping, quality, and how much is sitting unsent or unacked + // (the backpressure the old send path silently discarded). + // + if (NetTransport_StatsEnabled()) + { + DWORD now = GetTickCount(); + if ((LONG) (now - statsNextTick) >= 0) + { + statsNextTick = now + 5000; + + for (int j = 0; j < NetSendQueueTable::MaxQueues; ++j) + { + NetSendQueue *queue = &sendQueues.queues[j]; + if (queue->connection == InvalidConnection || + queue->sentMessages == 0) + { + continue; + } + DEBUG_STREAM << "NetStats: steam conn " + << (unsigned long) queue->connection << ": " + << queue->sentMessages << " msgs " + << queue->sentBytes << " B, " + << queue->wireWrites << " writes (" + << queue->wouldBlocks << " limit-exceeded), queued high " + << queue->queuedHighWater << " B, pending " + << queue->PendingBytes() + << (queue->dead ? " B, DEAD" : " B") + << "\n" << std::flush; + } + + for (int k = 0; k < gConnectionCount; ++k) + { + SteamNetConnectionRealTimeStatus_t status; + if (SteamNetworkingSockets()->GetConnectionRealTimeStatus( + gConnections[k].handle, &status, 0, NULL) == k_EResultOK) + { + DEBUG_STREAM << "NetStats: steam conn " + << (unsigned long) gConnections[k].handle + << ": ping " << status.m_nPing + << " ms, quality " << status.m_flConnectionQualityLocal + << "/" << status.m_flConnectionQualityRemote + << ", pending reliable " << status.m_cbPendingReliable + << " B, unacked " << status.m_cbSentUnackedReliable + << " B, queue " << (long) status.m_usecQueueTime + << " us\n" << std::flush; + } + } + } + } } int @@ -794,6 +989,83 @@ namespace address->sin_port = record->remoteEnginePort; return True; } + + private: + // + // Wire messages are capped so the receiver's leftover staging + // buffer can always hold the tail of one: the pump reads with at + // least MAX_RECEIVE_DATA_SIZE (1600) of room, so a 2048-byte + // message leaves at most 448 bytes over, against 4096 of staging. + // + enum { maxWireMessageBytes = 2048 }; + + // + // Send-side death has to surface where the engine looks for it: + // Receive() reports ConnectionRecord::dead as a disconnect once + // the leftovers drain. Mark both sides so drains stop too. + // + void + MarkSendDead( + Connection connection, + NetSendQueue *queue) + { + if (queue != NULL) + { + queue->dead = True; + } + ConnectionRecord *record = FindConnection((HSteamNetConnection) connection); + if (record != NULL) + { + record->dead = True; + } + } + + // + // Push pending bytes at the connection in wire-message chunks. + // A MUNGA message split across two wire messages is fine - the + // receiver treats the message sequence as a byte stream and + // re-frames from the length prefixes. + // + void + DrainQueue(NetSendQueue *queue) + { + while (queue->PendingBytes() > 0 && !queue->dead) + { + int chunk = queue->PendingBytes(); + if (chunk > maxWireMessageBytes) + { + chunk = maxWireMessageBytes; + } + EResult result = SteamNetworkingSockets()->SendMessageToConnection( + (HSteamNetConnection) queue->connection, + &queue->buffer[queue->head], (uint32) chunk, + k_nSteamNetworkingSend_Reliable, NULL); + if (result == k_EResultOK) + { + queue->wireWrites++; + queue->head += chunk; + if (queue->PendingBytes() == 0) + { + queue->head = 0; + queue->tail = 0; + } + continue; + } + if (result == k_EResultLimitExceeded) + { + queue->wouldBlocks++; + break; + } + DEBUG_STREAM << "SteamNetTransport::DrainQueue: result " << (int) result + << " with " << queue->PendingBytes() + << " bytes pending - declaring the connection dead\n" << std::flush; + MarkSendDead(queue->connection, queue); + break; + } + } + + NetSendQueueTable sendQueues; + DWORD statsNextTick; }; SteamNetTransport gSteamTransport; diff --git a/README.md b/README.md index 2186056..991cd5d 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,7 @@ mission. Full map with pad and keyboard diagrams: | [docs/RP412-FRONTEND-DESIGN.md](docs/RP412-FRONTEND-DESIGN.md) | TeslaConsole analysis, the egg format, the console protocol, and the Steam mapping — with status notes as each layer landed | | [docs/STEAM-3-MACHINE-TEST.md](docs/STEAM-3-MACHINE-TEST.md) | Multiplayer test procedure, Steam Input notes, the abort key | | [docs/RP412-LINUX.md](docs/RP412-LINUX.md) | Linux compatibility assessment — Proton path vs native port; parked until late playtesting | +| [docs/NET-TEST.md](docs/NET-TEST.md) | Network test protocol — WAN emulation profiles, the NetLog/NetStats telemetry, the stall-recovery test, `-spoolstats` | | [BUILD.md](BUILD.md) | Toolchain and build steps | Dev tooling: `tools/two-pod-test.ps1` races two pods on loopback, diff --git a/RP_L4/RPL4.CPP b/RP_L4/RPL4.CPP index 464110a..dfe8788 100644 --- a/RP_L4/RPL4.CPP +++ b/RP_L4/RPL4.CPP @@ -337,6 +337,34 @@ int WINAPI WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPSTR lpCmdLine // int argc; LPWSTR* argv = CommandLineToArgvW(GetCommandLineW(), &argc); + + // + //---------------------------------------------------------------------- + // -spoolstats : headless spool analysis (RPL4SPOOLSTATS.cpp). + // Runs before any window, renderer, or engine object exists and exits + // with the analyzer's code. Output lands on the parent console when + // launched from one, or a fresh console otherwise. + //---------------------------------------------------------------------- + // + for (int spool_arg = 1; spool_arg < argc - 1; ++spool_arg) + { + if (_wcsicmp(argv[spool_arg], L"-spoolstats") == 0) + { + char spool_path[512]; + size_t converted = 0; + wcstombs_s(&converted, spool_path, sizeof(spool_path), + argv[spool_arg + 1], _TRUNCATE); + if (!AttachConsole(ATTACH_PARENT_PROCESS)) + { + AllocConsole(); + } + FILE *spool_out = NULL; + freopen_s(&spool_out, "CONOUT$", "w", stdout); + extern int RPL4SpoolStats_Run(const char *path); + return RPL4SpoolStats_Run(spool_path); + } + } + Logical run_application = RPL4Application::ParseCommandLine(argc, argv); if (!run_application) { diff --git a/RP_L4/RPL4CONSOLE.cpp b/RP_L4/RPL4CONSOLE.cpp index 8e57baf..d4fa952 100644 --- a/RP_L4/RPL4CONSOLE.cpp +++ b/RP_L4/RPL4CONSOLE.cpp @@ -847,10 +847,24 @@ namespace for (int i = 0; i < gRemotePodCount; ++i) { Application::RunMissionMessage run; + // + // Per-pod send tick (RP412NETLOG / plan M4): + // with each pod's own "mission t0 at tick" + // line and the netclock offsets, these let + // the mission-clock skew between machines be + // computed offline. Expected to show the + // serialization cost here is microseconds + // and the real skew is path spread. + // + DEBUG_STREAM << "NetLog: RunMission to " + << gRemotePods[i].address << " at tick " + << Now().ticks << "\n" << std::flush; SendWire(&gRemotePods[i], NetworkClient::ApplicationClientID, &run, (int) run.messageLength); } Application::RunMissionMessage local_run; + DEBUG_STREAM << "NetLog: RunMission local at tick " + << Now().ticks << "\n" << std::flush; DeliverLocal( NetworkClient::ApplicationClientID, &local_run, diff --git a/RP_L4/RPL4ENVIRON.cpp b/RP_L4/RPL4ENVIRON.cpp index 9f39c2f..0a1edf1 100644 --- a/RP_L4/RPL4ENVIRON.cpp +++ b/RP_L4/RPL4ENVIRON.cpp @@ -321,6 +321,32 @@ namespace "# the old arrival-time behaviour if you want to compare.\n" "#RP412NETCLOCK=0\n" "\n" +"# 0 = no race-end network summary. On by default: when a race stops,\n" +"# rpl4.log gets a few NetLog lines - per remote pod, how many updates\n" +"# arrived and how evenly, how often its motion snapped instead of\n" +"# blending, and how far arriving updates moved it; per peer, whether\n" +"# the clock alignment ever had to step mid-race. Playtest logs used to\n" +"# say nothing at all about how the network felt; these lines are how a\n" +"# report like 'pod four was ticking' comes with numbers attached.\n" +"#RP412NETLOG=0\n" +"\n" +"# 0 = send network data the way the arcade did: fire and forget. On by\n" +"# default: bytes a connection cannot take right now are kept, in order,\n" +"# and retried - nothing is silently lost, and a partially-sent message\n" +"# can no longer shear the stream for good. The arcade LAN never filled\n" +"# a buffer so the old code never noticed; over the internet a peer\n" +"# stalled in its 10-30 second mission load stops reading and every\n" +"# machine still racing used to drop or shear whatever it sent them.\n" +"# The 0 setting still logs what it would have lost, for comparing.\n" +"#RP412NETSENDQ=0\n" +"\n" +"# How much a single connection may hold waiting to retry, in kilobytes.\n" +"# 16 to 4096, default 256 - seventeen-plus seconds of a full grid's\n" +"# traffic, far past the longest mission-load stall. A connection that\n" +"# overflows even this is declared dead rather than have the stream\n" +"# drop something from its middle.\n" +"#RP412NETQUEUEKB=256\n" +"\n" "# How long each connection attempt to another machine may take, in\n" "# seconds. 2 to 300, default 20, and up to three attempts are made.\n" "#\n" @@ -358,6 +384,15 @@ namespace "# unless set and cost nothing when off; none belongs in a real race.\n" "# They exist so a claim about the game can be tested instead of argued.\n" "\n" +"# 1 = transport telemetry to rpl4.log: a five-second rollup per network\n" +"# connection (messages, bytes, wire writes, partial writes and refusals,\n" +"# how much sat queued for retry), and on Steam the connection's own\n" +"# health - ping, quality, unsent and unacknowledged bytes - plus one\n" +"# detailed route description per connection at teardown (direct or\n" +"# which relay). The first read anything in this codebase has ever\n" +"# taken of what the wire is actually doing.\n" +"#RP412NETSTATS=1\n" +"\n" "# Dump the gauge profile to rpl4.log every N seconds: every cockpit\n" "# display with its rate mask and tier, how often it ran and what it\n" "# cost. This is the engine's own ProfileReport, which was only ever\n" diff --git a/RP_L4/RPL4SPOOLSTATS.cpp b/RP_L4/RPL4SPOOLSTATS.cpp new file mode 100644 index 0000000..0c05718 --- /dev/null +++ b/RP_L4/RPL4SPOOLSTATS.cpp @@ -0,0 +1,526 @@ +#include "..\munga_l4\mungal4.h" +#pragma hdrstop + +#include +#include "..\munga\app.h" +#include "..\munga\network.h" +#include "..\munga\interest.h" +#include "..\munga\entity.h" +#include "..\munga\simulate.h" + +//######################################################################## +// rpl4opt -spoolstats - offline latency analysis of a spool +// +// Every Live Cam / pod recording already contains a complete one-way- +// delay dataset and nothing ever read it that way: the recorder restamps +// each packet's header with LOCAL ARRIVAL time (SPOOLER.cpp), while the +// UpdateRecords inside still carry the SENDER's sim-grid timestamp +// untouched. Per sender, +// +// sample = arrival - senderStamp = clockOffset + oneWayDelay +// +// and the running minimum of sample converges on the offset from above - +// the same estimator RP412NETCLOCK runs live (SIMULATE.cpp). Subtracting +// it back out gives one-way delay ABOVE THE MINIMUM per packet: jitter +// and queueing, which is what the dead reckoner actually fights. The +// minimum itself (the flat latency floor) is unobservable without a +// synchronized clock and is absorbed into the offset - every figure +// printed here is delay above that floor, not absolute delay. +// +// Also decomposed per entity: gaps on the SENDER timeline (send pacing) +// versus gaps on the ARRIVAL timeline (network + receive-loop jitter). +// The send side of the tick story lives in the first, the delivery side +// in the second, and no live counter can separate them. +// +// This must live in the game exe, not RPL4TOOL: the engine compiles +// /Zp1 and RPL4TOOL deliberately does not, so the tool would misread +// every struct in the file. +// +// Caveats printed with the output: +// - arrival stamps are quantized by the receive drain (post-render, +// so roughly frame cadence) +// - sender stamps sit on the sender's fixed-step grid (20 ms at 50 Hz) +// - a racer's spool never contains its own pod; a Live Cam hears all +//######################################################################## + +namespace +{ + enum + { + statsMaxHosts = 32, + statsMaxEntities = 128, + statsWindow = 128, // mirrors netClockWindow (SIMULATE.cpp) + statsBuckets = 16 + }; + + // + // log2-millisecond histogram, same shape as Entity::NetRaceStats - + // bucket 0 is sub-millisecond, bucket b covers [2^(b-1), 2^b) ms. + // + struct Histogram + { + unsigned long buckets[statsBuckets]; + unsigned long total; + long worstMs; + + void + Count(long ms) + { + if (ms > worstMs) + { + worstMs = ms; + } + unsigned long value = (ms > 0) ? (unsigned long) ms : 0; + int bucket = 0; + while (value != 0 && bucket < statsBuckets - 1) + { + value >>= 1; + ++bucket; + } + buckets[bucket]++; + total++; + } + + unsigned long + PercentileMs(unsigned long percent) const + { + if (total == 0) + { + return 0; + } + unsigned long target = (total * percent + 99) / 100; + unsigned long seen = 0; + for (int b = 0; b < statsBuckets; ++b) + { + seen += buckets[b]; + if (seen >= target) + { + return (b == 0) ? 0 : (1UL << (b - 1)); + } + } + return 1UL << (statsBuckets - 2); + } + }; + + // + // Per sender host: packet accounting and the netclock replica. + // + struct HostStats + { + HostID host; + Logical inUse; + unsigned long packetCount; + unsigned long updateMessageCount; + unsigned long byteCount; + + Logical settled; + long offsetTicks; + long windowMinTicks; + int windowCount; + + Histogram owdAboveMin; + }; + + // + // Per entity: one entry per replicated entity seen in the spool, + // with the sender-vs-arrival gap decomposition. + // + struct EntityStats + { + Logical inUse; + int entityHost; // EntityID host part + int entityLocal; // EntityID local part + HostID fromHost; + unsigned long updateCount; + long lastSenderTicks; + long lastArrivalTicks; + Logical haveLast; + Histogram senderGaps; + Histogram arrivalGaps; + }; + + HostStats gHosts[statsMaxHosts]; + EntityStats gEntities[statsMaxEntities]; + + HostStats * + FindHost(HostID host) + { + HostStats *free_slot = NULL; + for (int i = 0; i < statsMaxHosts; ++i) + { + if (gHosts[i].inUse) + { + if (gHosts[i].host == host) + { + return &gHosts[i]; + } + } + else if (free_slot == NULL) + { + free_slot = &gHosts[i]; + } + } + if (free_slot != NULL) + { + memset(free_slot, 0, sizeof(*free_slot)); + free_slot->inUse = True; + free_slot->host = host; + } + return free_slot; + } + + EntityStats * + FindEntity(int entity_host, int entity_local, HostID from_host) + { + EntityStats *free_slot = NULL; + for (int i = 0; i < statsMaxEntities; ++i) + { + if (gEntities[i].inUse) + { + if (gEntities[i].entityHost == entity_host && + gEntities[i].entityLocal == entity_local) + { + return &gEntities[i]; + } + } + else if (free_slot == NULL) + { + free_slot = &gEntities[i]; + } + } + if (free_slot != NULL) + { + memset(free_slot, 0, sizeof(*free_slot)); + free_slot->inUse = True; + free_slot->entityHost = entity_host; + free_slot->entityLocal = entity_local; + free_slot->fromHost = from_host; + } + return free_slot; + } + + // + // The netclock replica: returns delay-above-minimum for one sample, + // updating the host's running-minimum offset exactly the way the + // live estimator does (immediate adopt of a shorter path, window + // close follows drift). + // + long + OwdAboveMin(HostStats *host, long sample) + { + if (!host->settled) + { + host->settled = True; + host->offsetTicks = sample; + host->windowMinTicks = sample; + host->windowCount = 0; + return 0; + } + if (sample < host->windowMinTicks) + { + host->windowMinTicks = sample; + } + if (sample < host->offsetTicks) + { + host->offsetTicks = sample; + } + if (++host->windowCount >= statsWindow) + { + host->offsetTicks = host->windowMinTicks; + host->windowMinTicks = sample; + host->windowCount = 0; + } + long above = sample - host->offsetTicks; + return (above > 0) ? above : 0; + } + + // + // Is there a plausible packet chain at this offset? The same check + // playback runs against a wrong egg, run over a few packets in a + // row - used to find where the host-pair table ends without needing + // the egg at all. + // + Logical + PacketChainPlausible( + const unsigned char *buffer, + size_t size, + size_t offset) + { + int checked = 0; + while (checked < 5) + { + if (offset == size) + { + return (checked > 0) ? True : False; + } + if (offset + sizeof(NetworkPacket) > size) + { + // a truncated tail is normal (the buffer filled mid-write) + return (checked > 0) ? True : False; + } + const NetworkPacket *packet = (const NetworkPacket *) (buffer + offset); + size_t length = packet->messageData.messageLength; + if (length < sizeof(Receiver::Message) || length > 65536) + { + return False; + } + if (offset + length + sizeof(NetworkPacketHeader) > size) + { + return (checked > 0) ? True : False; + } + offset += length + sizeof(NetworkPacketHeader); + ++checked; + } + return True; + } +} + +//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// +// Returns a process exit code: 0 analysis written, 1 something wrong. +// +int + RPL4SpoolStats_Run(const char *path) +{ + printf("SpoolStats: %s\n", path); + + FILE *file = fopen(path, "rb"); + if (file == NULL) + { + printf("SpoolStats: cannot open the file\n"); + return 1; + } + fseek(file, 0, SEEK_END); + long file_size = ftell(file); + fseek(file, 0, SEEK_SET); + if (file_size <= 0) + { + printf("SpoolStats: empty file\n"); + fclose(file); + return 1; + } + unsigned char *buffer = (unsigned char *) malloc((size_t) file_size); + if (buffer == NULL || + fread(buffer, 1, (size_t) file_size, file) != (size_t) file_size) + { + printf("SpoolStats: could not read %ld bytes\n", file_size); + fclose(file); + free(buffer); + return 1; + } + fclose(file); + size_t size = (size_t) file_size; + + // + // Skip the header: ApplicationID + resource major revision, then one + // (Logical remote, HostID) pair per host named in the egg. The pair + // count is not recorded, so find it by trying each count and asking + // whether a plausible packet chain starts where the table would end. + // + size_t table_start = sizeof(ApplicationID) + sizeof(int); + size_t pair_size = sizeof(Logical) + sizeof(HostID); + size_t packets_start = 0; + int host_pairs = -1; + + for (int try_count = 0; try_count <= 16; ++try_count) + { + size_t offset = table_start + (size_t) try_count * pair_size; + if (offset > size) + { + break; + } + // + // Each pair must read as a Logical 0/1 to be believable. + // + Logical pairs_believable = True; + for (int p = 0; p < try_count; ++p) + { + long remote = *(const long *) (buffer + table_start + (size_t) p * pair_size); + if (remote != 0 && remote != 1) + { + pairs_believable = False; + break; + } + } + if (pairs_believable && PacketChainPlausible(buffer, size, offset)) + { + packets_start = offset; + host_pairs = try_count; + break; + } + } + if (host_pairs < 0) + { + printf("SpoolStats: no plausible packet chain found - not a spool?\n"); + free(buffer); + return 1; + } + printf("SpoolStats: header holds %d host pair(s), %lu bytes of packets\n", + host_pairs, (unsigned long) (size - packets_start)); + + memset(gHosts, 0, sizeof(gHosts)); + memset(gEntities, 0, sizeof(gEntities)); + + char csv_path[512]; + _snprintf(csv_path, sizeof(csv_path) - 1, "%s.netstats.csv", path); + csv_path[sizeof(csv_path) - 1] = 0; + FILE *csv = fopen(csv_path, "w"); + if (csv != NULL) + { + fprintf(csv, "fromHost,entityHost,entityLocal,senderMs,arrivalMs,owdAboveMinMs\n"); + } + + // + // The packet walk. + // + size_t offset = packets_start; + unsigned long packet_total = 0; + long first_arrival = 0; + long last_arrival = 0; + Logical have_arrival = False; + + while (offset + sizeof(NetworkPacket) <= size) + { + NetworkPacket *packet = (NetworkPacket *) (buffer + offset); + size_t length = packet->messageData.messageLength; + if (length < sizeof(Receiver::Message) || length > 65536 || + offset + length + sizeof(NetworkPacketHeader) > size) + { + break; // truncated tail (buffer filled mid-write) + } + + long arrival = packet->timeStamp.ticks; + if (!have_arrival) + { + first_arrival = arrival; + have_arrival = True; + } + last_arrival = arrival; + + HostStats *host = FindHost(packet->fromHost); + if (host != NULL) + { + host->packetCount++; + host->byteCount += (unsigned long) (length + sizeof(NetworkPacketHeader)); + } + ++packet_total; + + // + // Entity update messages carry the sender sim-grid stamps. + // + if (host != NULL && + packet->clientID == NetworkClient::InterestManagerClientID && + packet->messageData.messageID == + InterestManager::EntityUpdateReplicantsMessageID && + length >= sizeof(Entity::Message) + sizeof(Simulation::UpdateRecord)) + { + host->updateMessageCount++; + + Entity::Message *entity_message = + (Entity::Message *) &packet->messageData; + + // + // EntityID is (hostID, localID), both 32-bit under /Zp1; + // its members are private and the analyzer only needs the + // two numbers, so read them as the pair they are on disk. + // + int entity_id[2]; + memcpy(entity_id, &entity_message->entityID, sizeof(entity_id)); + + // + // First record's stamp speaks for the message - every record + // in it was written at the same lastPerformance. + // + Simulation::UpdateRecord *record = + (Simulation::UpdateRecord *) + ((unsigned char *) entity_message + sizeof(Entity::Message)); + long sender_ticks = record->timeStamp.ticks; + long sample = arrival - sender_ticks; + long above = OwdAboveMin(host, sample); + host->owdAboveMin.Count(above); + + EntityStats *entity = + FindEntity(entity_id[0], entity_id[1], packet->fromHost); + if (entity != NULL) + { + entity->updateCount++; + if (entity->haveLast) + { + entity->senderGaps.Count(sender_ticks - entity->lastSenderTicks); + entity->arrivalGaps.Count(arrival - entity->lastArrivalTicks); + } + entity->lastSenderTicks = sender_ticks; + entity->lastArrivalTicks = arrival; + entity->haveLast = True; + } + + if (csv != NULL) + { + fprintf(csv, "%d,%d,%d,%ld,%ld,%ld\n", + (int) packet->fromHost, entity_id[0], entity_id[1], + sender_ticks, arrival, above); + } + } + + offset += length + sizeof(NetworkPacketHeader); + } + + // + // The report. + // + printf("SpoolStats: %lu packets spanning %.1f s of arrivals\n", + packet_total, + have_arrival ? (last_arrival - first_arrival) / 1000.0f : 0.0f); + printf("SpoolStats: figures are delay ABOVE the per-host minimum - the\n"); + printf("SpoolStats: flat latency floor is absorbed into the clock offset\n"); + printf("SpoolStats: and is unobservable without a synchronized clock.\n"); + printf("SpoolStats: arrivals are quantized by the receive drain (~frame\n"); + printf("SpoolStats: cadence); sender stamps sit on the 20 ms sim grid.\n"); + + int h; + for (h = 0; h < statsMaxHosts; ++h) + { + HostStats *host = &gHosts[h]; + if (!host->inUse) + { + continue; + } + printf("SpoolStats: host %d: %lu packets %lu B, %lu update msgs, " + "owd-above-min med/p95 ~%lu/~%lu ms, worst %ld ms, offset %ld ms\n", + (int) host->host, + host->packetCount, + host->byteCount, + host->updateMessageCount, + host->owdAboveMin.PercentileMs(50), + host->owdAboveMin.PercentileMs(95), + host->owdAboveMin.worstMs, + host->offsetTicks); + } + for (int e = 0; e < statsMaxEntities; ++e) + { + EntityStats *entity = &gEntities[e]; + if (!entity->inUse || entity->updateCount < 2) + { + continue; + } + printf("SpoolStats: entity %d:%d (host %d): %lu updates, " + "sender gaps med/p95 ~%lu/~%lu ms, arrival gaps med/p95 ~%lu/~%lu ms, " + "worst %ld/%ld ms\n", + entity->entityHost, entity->entityLocal, (int) entity->fromHost, + entity->updateCount, + entity->senderGaps.PercentileMs(50), + entity->senderGaps.PercentileMs(95), + entity->arrivalGaps.PercentileMs(50), + entity->arrivalGaps.PercentileMs(95), + entity->senderGaps.worstMs, + entity->arrivalGaps.worstMs); + } + + if (csv != NULL) + { + fclose(csv); + printf("SpoolStats: per-update rows written to %s\n", csv_path); + } + free(buffer); + return 0; +} diff --git a/RP_L4/RP_L4.vcxproj b/RP_L4/RP_L4.vcxproj index 94bd219..e3e2142 100644 --- a/RP_L4/RP_L4.vcxproj +++ b/RP_L4/RP_L4.vcxproj @@ -136,6 +136,7 @@ + diff --git a/docs/NET-TEST.md b/docs/NET-TEST.md new file mode 100644 index 0000000..86159e8 --- /dev/null +++ b/docs/NET-TEST.md @@ -0,0 +1,85 @@ +# Network testing — WAN emulation, telemetry, and the stall test + +How to make the loopback two-pod harness behave like the internet, and +how to read what the new instrumentation says about it. Added with the +first networking measurement+correctness build (2026-08-13); the plan it +serves is `~/.claude/plans` material, summarized here so the procedure +outlives the session that wrote it. + +Scope honestly stated: everything below exercises the **plain TCP +transport** on loopback. There is no SDR emulation — claims about the +Steam path rest on live telemetry (`RP412NETSTATS=1`) gathered during +real Steam races ([STEAM-3-MACHINE-TEST.md](STEAM-3-MACHINE-TEST.md)). + +## The instrumentation (what a race now writes) + +| Switch | Default | What appears in rpl4.log | +|--------|---------|--------------------------| +| `RP412NETLOG` | on | `NetLog:` lines — mission t0 tick at the green light; per-pod race totals at the stop (updates, interval med/p95, widest gap, long/queued counts, snaps, correction mean/worst); per-peer netclock offset and window-step counts; the console's per-pod RunMission send ticks | +| `RP412NETSTATS` | off | `NetStats:` lines — 5 s per-connection transport rollups (messages, bytes, wire writes, partials, would-blocks, retry-queue high water); on Steam also ping/quality/pending/unacked per connection and a route description at teardown | +| `RP412NETSENDQ` | on | set `=0` to restore fire-and-forget sends (still logs what it would have lost) — the A/B for the send queue | + +Offline: `rpl4opt -spoolstats SPOOLS\.spl` reads any Live Cam or +pod recording and prints per-host one-way-delay-above-minimum and +per-entity sender-vs-arrival gap decompositions, writing the raw rows to +`.spl.netstats.csv` beside it. Every spool ever recorded is +retroactively a latency dataset; the RECORDING toggle on the setup +screen is how new ones get made. + +## WAN emulation with clumsy + +[clumsy](https://jagt.github.io/clumsy/) (WinDivert underneath) is the +one Windows emulator that captures loopback traffic, which is what the +two-pod harness runs on. Run it as administrator, set the filter, pick a +profile, **then** start `tools\two-pod-test.ps1`. + +Filter — the game mesh only (console channels stay clean): + + tcp and (tcp.DstPort == 1502 or tcp.SrcPort == 1502 or tcp.DstPort == 1602 or tcp.SrcPort == 1602) + +Add `1501`/`1601` terms to also stress the console marshaling (egg feed, +state polls, the stop). + +Profiles (stock clumsy has fixed lag, not jitter — say which was used +when reporting numbers): + +| Profile | clumsy settings | What it stands in for | +|---------|-----------------|----------------------| +| **Steady internet** | Lag 80 ms, inbound + outbound | a good SDR route (~160 ms RTT) | +| **Asymmetric** | Lag 150 ms, inbound only | one slow direction, the shape NETCLOCK's min-filter has to survive | +| **Disorderly** | Lag 30 ms both + Out-of-order 2% | reorder torture — TCP re-orders below the engine, so the engine-visible symptom is added jitter, not reordering; the stale counter should stay 0 | + +What to look at afterward: both pods' `rpl4.log` (the harness leaves +them in its scratch `podA\` / `podB\` folders) — the `NetLog:` race +summary should show interval medians tracking the imposed lag pattern, +snaps staying rare, and corrections staying under ~0.5 m mean. Compare +against a clean run of the same script; that pair of summaries *is* the +result. + +## The stall-recovery test (the send queue's reason to exist) + +A peer stalled in its 10–30 s mission load stops reading; before the +send queue, every machine still racing silently dropped — or worse, +half-sent and permanently sheared — whatever it sent them. To reproduce +the stall deliberately: + + # freeze pod B for 20 s mid-race, then thaw it + $p = Get-Process rpl4opt | Sort-Object StartTime | Select-Object -Last 1 + # (suspend/resume via SysInternals: pssuspend $p.Id ; sleep 20 ; pssuspend -r $p.Id) + +Expected with the queue (default): pod A's `NetStats` rollup shows +would-blocks and a rising retry queue during the freeze, draining after +the thaw; the race continues; no `framing broken` line ever appears. +Expected with `RP412NETSENDQ=0`: pod A logs `legacy mode lost a +message` / `sheared the stream` — the old behavior, now at least named. +A `Receive: host N framing broken ... dropping the connection` line +means the receive-side validation caught a sheared stream and dropped +the peer rather than crashing — correct behavior, but on the queue path +it should never be needed. + +## Version discipline + +The wire format is unchanged by all of this. A mixed run — one pod on +the previous build, one on this — must interoperate, and doing that once +per build is part of the checklist: it is the proof the format did not +drift.