vwemirror/RP412
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d9149a6a8c6152a9e49aef32893153c023192531
RP412/MUNGA_L4/L4STEAMTRANSPORT.cpp
T
CydandClaude Fable 5 7d485c9672 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 <noreply@anthropic.com>
2026-08-13 12:59:25 -05:00

1313 lines
37 KiB
C++
Raw Blame History

#include "mungal4.h"
#pragma hdrstop
#include "l4steamtransport.h"
//########################################################################
// Is steam_api.dll actually here?
//
// Defined outside the RP412_STEAM guard on purpose: callers ask without
// caring how the build was configured, and a build without the SDK
// truthfully has no client library. See the header for why every Steam
// path has to come through here first.
//
// The answer is cached because it is asked on menu paint, and because a
// DLL that appeared halfway through a session is not a case worth
// supporting - the delay-load helper would have bound the first miss
// anyway.
//########################################################################
Logical
SteamNetTransport_ClientLibraryPresent()
{
#ifdef RP412_STEAM
static int present = -1;
if (present < 0)
{
present = (LoadLibraryA("steam_api.dll") != NULL) ? 1 : 0;
if (!present)
{
DEBUG_STREAM << "Steam: steam_api.dll not found beside the exe - "
<< "Steam features off, staying on TCP\n" << std::flush;
}
}
return present ? True : False;
#else
return False;
#endif
}
#ifdef RP412_STEAM
#include "l4nettransport.h"
// The engine builds with /Zp1; Valve's ABI expects default packing.
// Callback structs carry their own pack pragmas, but everything else
// must see the platform default.
#pragma pack(push, 8)
#include "steam/steam_api.h"
#include "steam/isteamnetworkingsockets.h"
#include "steam/isteamnetworkingutils.h"
#include "steam/steamnetworkingfakeip.h"
#pragma pack(pop)
//########################################################################
// SteamNetTransport - see l4steamtransport.h for the design. The
// engine is single threaded and so is this: every entry point pumps
// SteamAPI_RunCallbacks, which is where the status-changed callback
// fires (accepting incoming connections and marking drops).
//########################################################################
namespace
{
enum
{
maxListeners = 4,
maxConnections = 32,
maxPeers = 16,
maxPending = 8,
leftoverSize = 4096
};
struct ListenerRecord
{
HSteamListenSocket handle;
int fakePortIndex;
unsigned short enginePort; // host order
HSteamNetConnection pending[maxPending];
int pendingCount;
// The engine closes listeners with TCP semantics (accepted
// connections survive). Steam's CloseListenSocket kills the
// accepted connections ungracefully - so an engine close only
// marks the listener here; the real close waits for Cleanup.
Logical closed;
};
struct ConnectionRecord
{
HSteamNetConnection handle;
unsigned long remoteIP; // network order
unsigned short remoteEnginePort; // network order
Logical connected;
Logical dead;
char leftover[leftoverSize];
int leftoverCount;
};
struct PeerRecord
{
unsigned long fakeIP; // host order
unsigned short fakeConsolePort; // host order
unsigned short fakeGamePort; // host order
unsigned __int64 steamID; // identity -> global FakeIP on accept
};
// engine-side port convention (lobby default; game port is +1)
unsigned short gEngineConsolePort = 1501;
Logical gSteamReady = False;
unsigned long gLocalFakeIP = 0; // host order
unsigned short gLocalFakePorts[2] = { 0, 0 };
char gLocalFakeAddressString[32] = "";
// engine ports in the order Listen sees them: [0] console, [1] game
unsigned short gEnginePorts[2] = { 0, 0 };
int gEnginePortCount = 0;
ListenerRecord gListeners[maxListeners];
int gListenerCount = 0;
ConnectionRecord gConnections[maxConnections];
int gConnectionCount = 0;
PeerRecord gPeers[maxPeers];
int gPeerCount = 0;
ListenerRecord *
FindListener(HSteamListenSocket handle)
{
for (int i = 0; i < gListenerCount; ++i)
{
if (gListeners[i].handle == handle)
{
return &gListeners[i];
}
}
return NULL;
}
ConnectionRecord *
FindConnection(HSteamNetConnection handle)
{
for (int i = 0; i < gConnectionCount; ++i)
{
if (gConnections[i].handle == handle)
{
return &gConnections[i];
}
}
return NULL;
}
ConnectionRecord *
AddConnection(
HSteamNetConnection handle,
unsigned long remote_ip_net,
unsigned short remote_engine_port_net
)
{
if (gConnectionCount >= maxConnections)
{
return NULL;
}
ConnectionRecord *record = &gConnections[gConnectionCount++];
memset(record, 0, sizeof(*record));
record->handle = handle;
record->remoteIP = remote_ip_net;
record->remoteEnginePort = remote_engine_port_net;
return record;
}
void
RemoveConnection(HSteamNetConnection handle)
{
for (int i = 0; i < gConnectionCount; ++i)
{
if (gConnections[i].handle == handle)
{
gConnections[i] = gConnections[gConnectionCount - 1];
--gConnectionCount;
return;
}
}
}
// engine port in a SOCKADDR_IN -> the peer's matching fake port
unsigned short
LookupPeerFakePort(unsigned long fake_ip_host, unsigned short engine_port)
{
for (int i = 0; i < gPeerCount; ++i)
{
if (gPeers[i].fakeIP == fake_ip_host)
{
return (engine_port == gEngineConsolePort)
? gPeers[i].fakeConsolePort
: gPeers[i].fakeGamePort;
}
}
return 0;
}
//---------------------------------------------------------------
// Status-changed handler (fires inside SteamAPI_RunCallbacks on
// the game thread): accept incoming, queue connected, mark drops.
// Registered through the CCallback dispatcher below - the
// config-value function pointer is NOT dispatched by the steam_api
// flavor of the library (Valve's own example uses STEAM_CALLBACK).
//---------------------------------------------------------------
void
OnConnectionStatusChanged(SteamNetConnectionStatusChangedCallback_t *status)
{
switch (status->m_info.m_eState)
{
case k_ESteamNetworkingConnectionState_Connecting:
if (status->m_info.m_hListenSocket != k_HSteamListenSocket_Invalid)
{
// no more callers once the engine closed the listener
ListenerRecord *listener = FindListener(status->m_info.m_hListenSocket);
if (listener == NULL || listener->closed)
{
DEBUG_STREAM << "SteamNetTransport: incoming ["
<< status->m_info.m_szConnectionDescription
<< "] on a closed listener - rejecting\n" << std::flush;
SteamNetworkingSockets()->CloseConnection(
status->m_hConn, 0, "listener closed", false);
break;
}
// incoming: accept immediately, queue it when Connected
DEBUG_STREAM << "SteamNetTransport: incoming ["
<< status->m_info.m_szConnectionDescription
<< "] - accepting\n" << std::flush;
EResult accepted = SteamNetworkingSockets()->AcceptConnection(status->m_hConn);
if (accepted != k_EResultOK)
{
DEBUG_STREAM << "SteamNetTransport: AcceptConnection failed (EResult "
<< (int) accepted << ")\n" << std::flush;
}
}
break;
case k_ESteamNetworkingConnectionState_Connected:
DEBUG_STREAM << "SteamNetTransport: connected ["
<< status->m_info.m_szConnectionDescription << "]\n" << std::flush;
if (status->m_info.m_hListenSocket != k_HSteamListenSocket_Invalid)
{
ListenerRecord *listener = FindListener(status->m_info.m_hListenSocket);
if (listener != NULL && listener->pendingCount < maxPending)
{
listener->pending[listener->pendingCount++] = status->m_hConn;
}
}
else
{
ConnectionRecord *record = FindConnection(status->m_hConn);
if (record != NULL)
{
record->connected = True;
}
}
break;
case k_ESteamNetworkingConnectionState_ProblemDetectedLocally:
case k_ESteamNetworkingConnectionState_ClosedByPeer:
{
DEBUG_STREAM << "SteamNetTransport: dropped ["
<< status->m_info.m_szConnectionDescription
<< "] end reason " << status->m_info.m_eEndReason
<< ": " << status->m_info.m_szEndDebug << "\n" << std::flush;
ConnectionRecord *record = FindConnection(status->m_hConn);
if (record != NULL)
{
record->dead = True;
}
}
break;
}
}
//---------------------------------------------------------------
// The callback listener: constructed after SteamAPI_Init so the
// CCallback registration lands in a live callback manager
//---------------------------------------------------------------
class SteamTransportCallbacks
{
public:
STEAM_CALLBACK(SteamTransportCallbacks, OnStatusChanged,
SteamNetConnectionStatusChangedCallback_t);
};
void SteamTransportCallbacks::OnStatusChanged(
SteamNetConnectionStatusChangedCallback_t *status)
{
OnConnectionStatusChanged(status);
}
SteamTransportCallbacks *gCallbacks = NULL;
//---------------------------------------------------------------
// The transport
//---------------------------------------------------------------
class SteamNetTransport:
public NetTransport
{
public:
SteamNetTransport():
statsNextTick(0)
{
}
Logical
Startup()
{
// Install() already brought Steam up; the network manager
// just confirms the transport is live
return gSteamReady;
}
void
Cleanup()
{
// 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(
gConnections[i].handle, 0, "mission teardown", true);
}
gConnectionCount = 0;
for (int j = 0; j < gListenerCount; ++j)
{
SteamNetworkingSockets()->CloseListenSocket(gListeners[j].handle);
}
gListenerCount = 0;
gEnginePortCount = 0;
}
int
GetLocalAddresses(
unsigned long *addresses,
int max_count
)
{
if (!gSteamReady || max_count < 1)
{
return 0;
}
addresses[0] = htonl(gLocalFakeIP);
return 1;
}
Logical
Resolve(
const char *host_name,
SOCKADDR_IN *address
)
{
memset(address, 0, sizeof(SOCKADDR_IN));
address->sin_family = AF_INET;
SteamNetworkingIPAddr parsed;
if (!parsed.ParseString(host_name))
{
return False;
}
address->sin_addr.S_un.S_addr = htonl(parsed.GetIPv4());
address->sin_port = htons(parsed.m_port); // 0 when absent
return True;
}
Connection
Connect(
const SOCKADDR_IN *remote,
int local_port
)
{
unsigned long remote_fake_ip = ntohl(remote->sin_addr.S_un.S_addr);
unsigned short fake_port =
LookupPeerFakePort(remote_fake_ip, ntohs(remote->sin_port));
if (fake_port == 0)
{
DEBUG_STREAM << "SteamNetTransport: no registered peer for "
<< inet_ntoa(remote->sin_addr) << " - lobby did not feed it\n" << std::flush;
return InvalidConnection;
}
DEBUG_STREAM << "SteamNetTransport: connecting to "
<< inet_ntoa(remote->sin_addr) << ":" << ntohs(remote->sin_port)
<< " (fake port " << fake_port << ")...\n" << std::flush;
SteamNetworkingIPAddr target;
target.Clear();
target.SetIPv4(remote_fake_ip, fake_port);
//
// Mirror the TCP retry-while-refused loop, bounded: the
// egg-ACK ordering means the peer may not be listening yet.
//
// Every wait below goes through NetTransport_PumpAndSleep, so
// the window keeps painting and answering instead of being
// declared "Not responding" for the whole attempt. There is no
// render loop yet at this point - this runs before the engine
// block - so nothing else is keeping it alive.
//
//
// The deadline is PER ATTEMPT, and attempts are capped, rather
// than one budget across all attempts.
//
// The 2026-08-11 playtest showed why, on every machine that met
// the one badly-NATed peer: the first attempt goes for a direct
// path, burns around ten seconds, and dies with 'timed out
// attempting to connect' (5003) or 'negotiate rendezvous'
// (5008); the SECOND attempt comes up through Valve's relay and
// succeeds - whenever it is given time. Under one shared budget
// the relay attempt inherited whatever the direct attempt left,
// and the logs show it being cut off mid-connect ('attempt 2
// ended in state 1' - still connecting) at the 20s mark. Races
// then could not assemble, because the mesh needs every pair.
//
// Three attempts at RP412CONNECTWAIT each: the observed failure
// connects on attempt 2 at about half a minute, the truly
// unreachable peer costs about a minute instead of twenty
// seconds, and ESC still works throughout.
//
DWORD wait_seconds = (DWORD) NetTransport_ConnectWaitSeconds();
DWORD started = GetTickCount();
int attempt = 0;
const int kMaxAttempts = 3;
char progress[256];
sprintf(
progress,
"Red Planet - connecting to %s ...",
inet_ntoa(remote->sin_addr)
);
NetTransport_SetWaitProgress(progress);
for (;;)
{
++attempt;
DWORD deadline = GetTickCount() + wait_seconds * 1000;
HSteamNetConnection handle =
SteamNetworkingSockets()->ConnectByIPAddress(target, 0, NULL);
if (handle == k_HSteamNetConnection_Invalid)
{
DEBUG_STREAM << "SteamNetTransport: ConnectByIPAddress refused the call\n" << std::flush;
NetTransport_SetWaitProgress(NULL);
return InvalidConnection;
}
AddConnection(handle, remote->sin_addr.S_un.S_addr, remote->sin_port);
//
// Poll the connection state directly (the callback also
// runs, for the log and the accept queues)
//
ESteamNetworkingConnectionState state =
k_ESteamNetworkingConnectionState_Connecting;
while ((LONG)(GetTickCount() - deadline) < 0)
{
SteamAPI_RunCallbacks();
SteamNetConnectionInfo_t info;
if (!SteamNetworkingSockets()->GetConnectionInfo(handle, &info))
{
state = k_ESteamNetworkingConnectionState_Dead;
break;
}
state = info.m_eState;
if (state == k_ESteamNetworkingConnectionState_Connected ||
state == k_ESteamNetworkingConnectionState_ProblemDetectedLocally ||
state == k_ESteamNetworkingConnectionState_ClosedByPeer ||
state == k_ESteamNetworkingConnectionState_None)
{
break;
}
//
// Count down out loud. Two minutes of a silent frozen
// window gave a host nothing to act on - not which peer
// was missing, not how long was left, not a way out.
//
DWORD left = (DWORD)((LONG)(deadline - GetTickCount()) / 1000);
sprintf(
progress,
"Red Planet - connecting to %s (try %d of %d) ... %us left, ESC to cancel",
inet_ntoa(remote->sin_addr),
attempt, kMaxAttempts,
(unsigned) left
);
NetTransport_SetWaitProgress(progress);
NetWaitResult wait = NetTransport_PumpAndSleep(25);
if (wait != NetWaitContinue)
{
DEBUG_STREAM << "SteamNetTransport: connect "
<< ((wait == NetWaitCancelled) ? "cancelled" : "abandoned, quitting")
<< " after " << ((GetTickCount() - started) / 1000)
<< "s\n" << std::flush;
SteamNetworkingSockets()->CloseConnection(handle, 0, "cancelled", false);
RemoveConnection(handle);
NetTransport_SetWaitProgress(NULL);
return InvalidConnection;
}
}
if (state == k_ESteamNetworkingConnectionState_Connected)
{
ConnectionRecord *record = FindConnection(handle);
if (record != NULL)
{
record->connected = True;
}
DEBUG_STREAM << "SteamNetTransport: connect succeeded (attempt "
<< attempt << ")\n" << std::flush;
NetTransport_SetWaitProgress(NULL);
return (Connection) handle;
}
// attempt failed - drop it and retry with a fresh window
DEBUG_STREAM << "SteamNetTransport: attempt " << attempt
<< " ended in state " << (int) state << "\n" << std::flush;
SteamNetworkingSockets()->CloseConnection(handle, 0, "retry", false);
RemoveConnection(handle);
if (attempt >= kMaxAttempts)
{
DEBUG_STREAM << "SteamNetTransport: gave up after "
<< attempt << " attempts of " << wait_seconds
<< "s each (RP412CONNECTWAIT), "
<< ((GetTickCount() - started) / 1000)
<< "s in all\n" << std::flush;
NetTransport_SetWaitProgress(NULL);
return InvalidConnection;
}
//
// A second between redials, still answering the window.
//
for (int slept = 0; slept < 1000; slept += 50)
{
NetWaitResult wait = NetTransport_PumpAndSleep(50);
if (wait != NetWaitContinue)
{
DEBUG_STREAM << "SteamNetTransport: connect "
<< ((wait == NetWaitCancelled) ? "cancelled" : "abandoned, quitting")
<< " between attempts\n" << std::flush;
NetTransport_SetWaitProgress(NULL);
return InvalidConnection;
}
}
}
}
Connection
Listen(
int local_port,
int backlog
)
{
if (gListenerCount >= maxListeners)
{
return InvalidConnection;
}
// a logically-closed listener for this engine port reopens
// (the Steam socket outlives engine closes; see Close)
for (int existing = 0; existing < gListenerCount; ++existing)
{
if (gListeners[existing].enginePort == (unsigned short) local_port)
{
gListeners[existing].closed = False;
gListeners[existing].pendingCount = 0;
DEBUG_STREAM << "SteamNetTransport: reopened listener on engine port "
<< local_port << "\n" << std::flush;
return (Connection) gListeners[existing].handle;
}
}
// engine port -> fake port index, in order of appearance:
// the console channel always listens first, the mesh second
int index = -1;
for (int i = 0; i < gEnginePortCount; ++i)
{
if (gEnginePorts[i] == (unsigned short) local_port)
{
index = i;
}
}
if (index < 0)
{
if (gEnginePortCount >= 2)
{
DEBUG_STREAM << "SteamNetTransport: only two fake ports allocated!\n" << std::flush;
return InvalidConnection;
}
index = gEnginePortCount;
gEnginePorts[gEnginePortCount++] = (unsigned short) local_port;
}
DEBUG_STREAM << "SteamNetTransport: listening on engine port "
<< local_port << " (fake port " << gLocalFakePorts[index]
<< ")...\n" << std::flush;
HSteamListenSocket handle =
SteamNetworkingSockets()->CreateListenSocketP2PFakeIP(index, 0, NULL);
if (handle == k_HSteamListenSocket_Invalid)
{
return InvalidConnection;
}
ListenerRecord *listener = &gListeners[gListenerCount++];
memset(listener, 0, sizeof(*listener));
listener->handle = handle;
listener->fakePortIndex = index;
listener->enginePort = (unsigned short) local_port;
return (Connection) handle;
}
Connection
Accept(Connection listener_handle)
{
SteamAPI_RunCallbacks();
ListenerRecord *listener = FindListener((HSteamListenSocket) listener_handle);
if (listener == NULL || listener->closed || listener->pendingCount == 0)
{
return InvalidConnection;
}
HSteamNetConnection handle = listener->pending[0];
for (int i = 1; i < listener->pendingCount; ++i)
{
listener->pending[i - 1] = listener->pending[i];
}
--listener->pendingCount;
//
// The remote engine port is the same engine port we accept
// on (all pods share the -net convention under Steam). The
// caller's address: Steam reports incoming connections
// under a locally-allocated ALIAS FakeIP, so resolve the
// caller's identity through the peer table to the GLOBAL
// FakeIP the egg's [pilots] list promised - the engine's
// mesh identity checks compare against that.
//
unsigned long remote_ip_net = 0;
SteamNetConnectionInfo_t info;
if (SteamNetworkingSockets()->GetConnectionInfo(handle, &info))
{
unsigned __int64 caller_id = info.m_identityRemote.GetSteamID64();
for (int p = 0; p < gPeerCount; ++p)
{
if (gPeers[p].steamID != 0 && gPeers[p].steamID == caller_id)
{
remote_ip_net = htonl(gPeers[p].fakeIP);
break;
}
}
}
if (remote_ip_net == 0)
{
// unregistered caller: fall back to the alias address
SteamNetworkingIPAddr remote_fake;
if (SteamNetworkingSockets()->GetRemoteFakeIPForConnection(
handle, &remote_fake) == k_EResultOK)
{
remote_ip_net = htonl(remote_fake.GetIPv4());
}
}
ConnectionRecord *record = AddConnection(
handle, remote_ip_net, htons(listener->enginePort));
if (record != NULL)
{
record->connected = True;
}
return (Connection) handle;
}
void
Close(Connection connection)
{
ListenerRecord *listener = FindListener((HSteamListenSocket) connection);
if (listener != NULL)
{
//
// TCP semantics: accepted connections must survive a
// listener close. Steam's CloseListenSocket kills them
// ungracefully, so only mark it closed (rejecting any
// new callers) and destroy it for real in Cleanup.
//
listener->closed = True;
for (int i = 0; i < listener->pendingCount; ++i)
{
SteamNetworkingSockets()->CloseConnection(
listener->pending[i], 0, "listener closed", false);
}
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);
}
int
Send(
Connection connection,
const void *data,
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);
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
Receive(
Connection connection,
void *buffer,
int size
)
{
SteamAPI_RunCallbacks();
ConnectionRecord *record = FindConnection((HSteamNetConnection) connection);
if (record == NULL)
{
return ReceiveNoData;
}
char *out = (char *) buffer;
int copied = 0;
// leftover bytes from a message that outsized the last call
if (record->leftoverCount > 0)
{
int take = (record->leftoverCount < size) ? record->leftoverCount : size;
memcpy(out, record->leftover, take);
memmove(record->leftover, record->leftover + take,
record->leftoverCount - take);
record->leftoverCount -= take;
out += take;
copied += take;
}
// drain messages while they fit; stash any partial tail
while (copied < size && record->leftoverCount == 0)
{
SteamNetworkingMessage_t *message = NULL;
int count = SteamNetworkingSockets()->ReceiveMessagesOnConnection(
record->handle, &message, 1);
if (count <= 0)
{
break;
}
int room = size - copied;
int take = ((int) message->m_cbSize < room) ? (int) message->m_cbSize : room;
memcpy(out, message->m_pData, take);
out += take;
copied += take;
int rest = (int) message->m_cbSize - take;
if (rest > 0)
{
if (rest > leftoverSize)
{
rest = leftoverSize; // cannot happen: engine packets < 2K
}
memcpy(record->leftover, (const char *) message->m_pData + take, rest);
record->leftoverCount = rest;
}
message->Release();
}
if (copied > 0)
{
return copied;
}
if (record->dead)
{
return ReceiveDisconnected;
}
return ReceiveNoData;
}
Logical
GetRemoteAddress(
Connection connection,
SOCKADDR_IN *address
)
{
memset(address, 0, sizeof(SOCKADDR_IN));
ConnectionRecord *record = FindConnection((HSteamNetConnection) connection);
if (record == NULL || !record->connected)
{
return False;
}
address->sin_family = AF_INET;
address->sin_addr.S_un.S_addr = record->remoteIP;
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;
}
//########################################################################
// Install: bring Steam up, get our FakeIP identity, take over the wire
//########################################################################
Logical
SteamNetTransport_Install()
{
if (gSteamReady)
{
return True;
}
//
// Before ANY Steam symbol: steam_api.dll is delay-loaded, so calling
// into it when it is missing raises the helper's fatal exception
// rather than failing. This is the gate that makes the DLL optional.
//
if (!SteamNetTransport_ClientLibraryPresent())
{
return False;
}
if (!SteamAPI_Init())
{
DEBUG_STREAM << "SteamNetTransport: SteamAPI_Init failed "
<< "(Steam not running, or no steam_appid.txt) - staying on TCP\n" << std::flush;
return False;
}
// connection status arrives through the CCallback dispatcher
if (gCallbacks == NULL)
{
gCallbacks = new SteamTransportCallbacks;
}
SteamNetworkingUtils()->InitRelayNetworkAccess();
//
// Mission load stalls the game thread for 10-30s with nothing
// pumping - Steam's default 10s connected-timeout would shear
// every connection mid-load (seen live: end reason 4001 with rx
// ages right at load duration). TCP never timed out idle arcade
// links; 90s keeps that spirit.
//
SteamNetworkingUtils()->SetGlobalConfigValueInt32(
k_ESteamNetworkingConfig_TimeoutConnected, 90 * 1000);
if (!SteamNetworkingSockets()->BeginAsyncRequestFakeIP(2))
{
DEBUG_STREAM << "SteamNetTransport: BeginAsyncRequestFakeIP failed - staying on TCP\n" << std::flush;
return False;
}
// FakeIP allocation is async: pump until it lands (or give up)
SteamNetworkingFakeIPResult_t fake;
memset(&fake, 0, sizeof(fake));
DWORD deadline = GetTickCount() + 20 * 1000;
for (;;)
{
SteamAPI_RunCallbacks();
SteamNetworkingSockets()->GetFakeIP(0, &fake);
if (fake.m_eResult == k_EResultOK)
{
break;
}
if (fake.m_eResult != k_EResultBusy && fake.m_eResult != k_EResultNoMatch)
{
DEBUG_STREAM << "SteamNetTransport: FakeIP allocation failed (EResult "
<< (int) fake.m_eResult << ") - staying on TCP\n" << std::flush;
return False;
}
if ((LONG)(GetTickCount() - deadline) >= 0)
{
DEBUG_STREAM << "SteamNetTransport: FakeIP allocation timed out - staying on TCP\n" << std::flush;
return False;
}
Sleep(50);
}
gLocalFakeIP = fake.m_unIP;
gLocalFakePorts[0] = fake.m_unPorts[0];
gLocalFakePorts[1] = fake.m_unPorts[1];
SteamNetworkingIPAddr mine;
mine.Clear();
mine.SetIPv4(gLocalFakeIP, 0);
mine.ToString(gLocalFakeAddressString, sizeof(gLocalFakeAddressString), false);
DEBUG_STREAM << "SteamNetTransport: up as " << gLocalFakeAddressString
<< " (fake ports " << gLocalFakePorts[0] << " console, "
<< gLocalFakePorts[1] << " game)\n" << std::flush;
gSteamReady = True;
NetTransport_Set(&gSteamTransport);
//
// dev: RP412STEAMSELFTEST=1 loops a connection back to ourselves,
// proving the listen/accept/connect/send/receive machinery (and
// the status-callback dispatch) without a second machine
//
const char *self_test = getenv("RP412STEAMSELFTEST");
if (self_test != NULL && atoi(self_test) != 0)
{
DEBUG_STREAM << "SteamNetTransport: SELF TEST starting\n" << std::flush;
SteamNetTransport_RegisterPeer(
gLocalFakeAddressString, gLocalFakePorts[0], gLocalFakePorts[1],
SteamUser()->GetSteamID().ConvertToUint64());
NetTransport *transport = &gSteamTransport;
NetTransport::Connection listener = transport->Listen(1501, 1);
SOCKADDR_IN self_address;
transport->Resolve(gLocalFakeAddressString, &self_address);
self_address.sin_port = htons(1501);
NetTransport::Connection outgoing = transport->Connect(&self_address, 0);
NetTransport::Connection incoming = NetTransport::InvalidConnection;
DWORD accept_deadline = GetTickCount() + 10 * 1000;
while (incoming == NetTransport::InvalidConnection &&
(LONG)(GetTickCount() - accept_deadline) < 0)
{
incoming = transport->Accept(listener);
Sleep(50);
}
Logical ok = False;
Logical survives_close = False;
if (outgoing != NetTransport::InvalidConnection &&
incoming != NetTransport::InvalidConnection)
{
transport->Send(outgoing, "PING", 4);
char buffer[16];
DWORD recv_deadline = GetTickCount() + 5 * 1000;
while ((LONG)(GetTickCount() - recv_deadline) < 0)
{
if (transport->Receive(incoming, buffer, sizeof(buffer)) == 4)
{
ok = (memcmp(buffer, "PING", 4) == 0);
break;
}
Sleep(25);
}
//
// The arcade engine closes a listener right after
// accepting - the accepted connection MUST survive
// (Steam kills children on CloseListenSocket; the
// transport defers the real close to Cleanup)
//
transport->Close(listener);
transport->Send(incoming, "PONG", 4);
recv_deadline = GetTickCount() + 5 * 1000;
while ((LONG)(GetTickCount() - recv_deadline) < 0)
{
if (transport->Receive(outgoing, buffer, sizeof(buffer)) == 4)
{
survives_close = (memcmp(buffer, "PONG", 4) == 0);
break;
}
Sleep(25);
}
}
DEBUG_STREAM << "SteamNetTransport: SELF TEST "
<< ((ok && survives_close) ? "PASSED" : "FAILED")
<< " (out " << (outgoing != NetTransport::InvalidConnection)
<< ", in " << (incoming != NetTransport::InvalidConnection)
<< ", ping " << ok
<< ", survives listener close " << survives_close
<< ")\n" << std::flush;
// leave the session pristine for the real race
gSteamTransport.Cleanup();
gPeerCount = 0;
}
return True;
}
const char *
SteamNetTransport_GetFakeAddressString()
{
return gLocalFakeAddressString;
}
int
SteamNetTransport_GetFakeConsolePort()
{
return gLocalFakePorts[0];
}
int
SteamNetTransport_GetFakeGamePort()
{
return gLocalFakePorts[1];
}
void
SteamNetTransport_SetEnginePorts(int console_port)
{
gEngineConsolePort = (unsigned short) console_port;
}
void
SteamNetTransport_RegisterPeer(
const char *fake_ip,
int fake_console_port,
int fake_game_port,
unsigned __int64 steam_id
)
{
SteamNetworkingIPAddr parsed;
if (!parsed.ParseString(fake_ip))
{
return;
}
for (int i = 0; i < gPeerCount; ++i)
{
if (gPeers[i].fakeIP == parsed.GetIPv4())
{
gPeers[i].fakeConsolePort = (unsigned short) fake_console_port;
gPeers[i].fakeGamePort = (unsigned short) fake_game_port;
gPeers[i].steamID = steam_id;
return;
}
}
if (gPeerCount >= maxPeers)
{
return;
}
gPeers[gPeerCount].fakeIP = parsed.GetIPv4();
gPeers[gPeerCount].fakeConsolePort = (unsigned short) fake_console_port;
gPeers[gPeerCount].fakeGamePort = (unsigned short) fake_game_port;
gPeers[gPeerCount].steamID = steam_id;
++gPeerCount;
DEBUG_STREAM << "SteamNetTransport: peer registered: " << fake_ip
<< " (console " << fake_console_port << ", game "
<< fake_game_port << ", id " << steam_id << ")\n" << std::flush;
}
#endif // RP412_STEAM
Reference in New Issue View Git Blame Copy Permalink