#include "mungal4.h" #pragma hdrstop #include "l4steamtransport.h" #ifdef BT412_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: 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 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 DWORD deadline = GetTickCount() + 120 * 1000; int attempt = 0; for (;;) { ++attempt; HSteamNetConnection handle = SteamNetworkingSockets()->ConnectByIPAddress(target, 0, NULL); if (handle == k_HSteamNetConnection_Invalid) { DEBUG_STREAM << "SteamNetTransport: ConnectByIPAddress refused the call\n" << std::flush; 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; } Sleep(25); } 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; return (Connection) handle; } // attempt failed - drop it and retry until the deadline DEBUG_STREAM << "SteamNetTransport: attempt " << attempt << " ended in state " << (int) state << "\n" << std::flush; SteamNetworkingSockets()->CloseConnection(handle, 0, "retry", false); RemoveConnection(handle); if ((LONG)(GetTickCount() - deadline) >= 0) { DEBUG_STREAM << "SteamNetTransport: connect timed out\n" << std::flush; return InvalidConnection; } Sleep(1000); } } 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; } SteamNetworkingSockets()->CloseConnection( (HSteamNetConnection) connection, 0, "closed", true); RemoveConnection((HSteamNetConnection) connection); } int Send( Connection connection, const void *data, int size ) { EResult result = SteamNetworkingSockets()->SendMessageToConnection( (HSteamNetConnection) connection, data, (uint32) size, k_nSteamNetworkingSend_Reliable, NULL); return (result == k_EResultOK) ? size : -1; } 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; } }; SteamNetTransport gSteamTransport; } //######################################################################## // Install: bring Steam up, get our FakeIP identity, take over the wire //######################################################################## Logical SteamNetTransport_Install() { if (gSteamReady) { return True; } 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: BT412STEAMSELFTEST=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("BT412STEAMSELFTEST"); 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 // BT412_STEAM