BT411/engine/MUNGA_L4/L4STEAMNET.cpp

//###########################################################################
// L4STEAMNET -- the Steam internet transport (BT_STEAM only; this TU is
// only in the build when the gate is on -- see CMakeLists.txt).
// Design: L4STEAMNET.h.
//###########################################################################

#include "l4steamnet.h"

#include <windows.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#pragma pack(push, 8)
#include "steam/steam_api.h"
#include "steam/isteamnetworkingsockets.h"
#include "steam/isteamnetworkingutils.h"
#include "steam/steamnetworkingfakeip.h"
#pragma pack(pop)

#include <iostream>
#define STEAMNET_LOG(x) \
	do { std::cout << "[steamnet] " << x << "\n" << std::flush; } while (0)

//###########################################################################
// State
//###########################################################################

enum
{
	MaxConnections   = 16,
	RingCapacity     = 256 * 1024,		// per-connection reassembly ring
	PseudoSocketBase = 0x5EA00000,
	AcceptQueueSize  = 8,
	FakePortCount    = 2				// [0] console, [1] game
};

struct SteamConnection
{
	int						inUse;
	HSteamNetConnection		connection;
	SOCKADDR_IN				peerAddress;	// the peer's FAKE ipv4:port
	int						connected;		// handshake complete
	int						closedByPeer;
	unsigned char			*ring;
	int						ringHead, ringTail;	// [tail, head) valid
};

static int				steamActive = 0;
static SteamConnection	connections[MaxConnections];
static HSteamListenSocket listenSockets[FakePortCount];
static SOCKET			acceptQueue[FakePortCount][AcceptQueueSize];
static int				acceptHead[FakePortCount], acceptTail[FakePortCount];
static unsigned long	lastPumpTick = 0;

//
// The TOKEN TABLE (the FakeIP-stability lesson, live-verified: a fresh
// process gets a DIFFERENT FakeIP, so menu-time fake addresses go stale
// by mission time).  Roster addresses are therefore opaque ipv4-shaped
// TOKENS minted by the lobby host and mapped to Steam IDENTITIES; the
// mission process receives the map via env:
//   BT_FE_MYFAKE   = my own token ip (joins GetMyAddress -- self-match)
//   BT_FE_STEAMMAP = ip=steamid64;ip=steamid64;...
// Console channel = P2P virtual port 0 (token port 1501), game = virtual
// port 1 (token port 1502).
//
enum { TokenConsolePort = 1501, TokenGamePort = 1502 };
struct SteamToken
{
	unsigned long		addressBE;
	unsigned long long	steamID;
};
static SteamToken		tokens[8];
static int				tokenCount = 0;
static unsigned long	myTokenBE = 0;

static SteamConnection *
	ConnectionOf(SOCKET wire_socket)
{
	unsigned long value = (unsigned long)wire_socket;
	if ((value & 0xFFFF0000) != PseudoSocketBase)
	{
		return NULL;
	}
	int index = (int)(value & 0xFFFF);
	if (index < 0 || index >= MaxConnections || !connections[index].inUse)
	{
		return NULL;
	}
	return &connections[index];
}

static SOCKET
	PseudoSocketOf(SteamConnection *connection)
{
	return (SOCKET)(PseudoSocketBase | (int)(connection - connections));
}

static SteamConnection *
	AllocateConnection(HSteamNetConnection handle)
{
	for (int i = 0; i < MaxConnections; ++i)
	{
		if (!connections[i].inUse)
		{
			SteamConnection &c = connections[i];
			memset(&c.peerAddress, 0, sizeof(c.peerAddress));
			c.inUse = 1;
			c.connection = handle;
			c.connected = 0;
			c.closedByPeer = 0;
			if (c.ring == NULL)
			{
				c.ring = (unsigned char *)malloc(RingCapacity);
			}
			c.ringHead = c.ringTail = 0;
			return &c;
		}
	}
	return NULL;
}

//
// The peer's TOKEN ipv4:port as a SOCKADDR_IN: the connection's remote
// Steam identity looked up in the token table; the port encodes the
// channel (console/game) the connection rides.
//
static void
	FillPeerAddressFromIdentity(SteamConnection *connection, int channel)
{
	SteamNetConnectionInfo_t info;
	if (!SteamNetworkingSockets()->GetConnectionInfo(
		connection->connection, &info))
	{
		return;
	}
	unsigned long long remote_id = info.m_identityRemote.GetSteamID64();
	for (int t = 0; t < tokenCount; ++t)
	{
		if (tokens[t].steamID == remote_id)
		{
			connection->peerAddress.sin_family = AF_INET;
			connection->peerAddress.sin_addr.s_addr = tokens[t].addressBE;
			connection->peerAddress.sin_port = htons((unsigned short)
				((channel == 0) ? TokenConsolePort : TokenGamePort));
			return;
		}
	}
}

//###########################################################################
// The status-changed callback (invoked from RunCallbacks on our thread).
//###########################################################################

static void __cdecl
	OnConnectionStatusChanged(SteamNetConnectionStatusChangedCallback_t *status)
{
	ISteamNetworkingSockets *sockets = SteamNetworkingSockets();

	switch (status->m_info.m_eState)
	{
	case k_ESteamNetworkingConnectionState_Connecting:
		//
		// Incoming connection on one of our listen sockets: accept it and
		// queue the pseudo-socket for BTNetAccept.
		//
		if (status->m_eOldState == k_ESteamNetworkingConnectionState_None &&
			status->m_info.m_hListenSocket != k_HSteamListenSocket_Invalid)
		{
			int channel = -1;
			for (int i = 0; i < FakePortCount; ++i)
			{
				if (listenSockets[i] == status->m_info.m_hListenSocket)
				{
					channel = i;
				}
			}
			if (channel < 0)
			{
				sockets->CloseConnection(status->m_hConn, 0, "unknown listener", false);
				return;
			}
			int next = (acceptHead[channel] + 1) % AcceptQueueSize;
			if (next == acceptTail[channel])
			{
				sockets->CloseConnection(status->m_hConn, 0, "accept queue full", false);
				return;
			}
			if (sockets->AcceptConnection(status->m_hConn) != k_EResultOK)
			{
				sockets->CloseConnection(status->m_hConn, 0, "accept failed", false);
				return;
			}
			SteamConnection *connection = AllocateConnection(status->m_hConn);
			if (connection == NULL)
			{
				sockets->CloseConnection(status->m_hConn, 0, "table full", false);
				return;
			}
			FillPeerAddressFromIdentity(connection, channel);
			acceptQueue[channel][acceptHead[channel]] = PseudoSocketOf(connection);
			acceptHead[channel] = next;
			STEAMNET_LOG("incoming connection queued on channel " << channel);
		}
		break;

	case k_ESteamNetworkingConnectionState_Connected:
		for (int i = 0; i < MaxConnections; ++i)
		{
			if (connections[i].inUse &&
				connections[i].connection == status->m_hConn)
			{
				connections[i].connected = 1;
				STEAMNET_LOG("connection " << i << " up");
			}
		}
		break;

	case k_ESteamNetworkingConnectionState_ClosedByPeer:
	case k_ESteamNetworkingConnectionState_ProblemDetectedLocally:
		for (int i = 0; i < MaxConnections; ++i)
		{
			if (connections[i].inUse &&
				connections[i].connection == status->m_hConn)
			{
				connections[i].closedByPeer = 1;
				STEAMNET_LOG("connection " << i << " closed ("
					<< status->m_info.m_szEndDebug << ")");
			}
		}
		break;

	default:
		break;
	}
}

//###########################################################################
// The pump: Steam callbacks + draining every connection's messages into
// its ring.  Called lazily from the seam entries (the game thread), rate-
// limited to once per millisecond tick.
//###########################################################################

void
	BTSteamNet_Pump()
{
	if (!steamActive)
	{
		return;
	}
	unsigned long now = GetTickCount();
	if (now == lastPumpTick)
	{
		return;
	}
	lastPumpTick = now;

	//
	// Both pumps: the general Steam dispatch (FakeIP results, auth) AND
	// the networking-sockets callbacks (connection status).
	//
	SteamAPI_RunCallbacks();
	ISteamNetworkingSockets *sockets = SteamNetworkingSockets();
	sockets->RunCallbacks();

	for (int i = 0; i < MaxConnections; ++i)
	{
		SteamConnection &c = connections[i];
		if (!c.inUse || c.connection == k_HSteamNetConnection_Invalid)
		{
			continue;
		}
		for (;;)
		{
			SteamNetworkingMessage_t *messages[16];
			int received = sockets->ReceiveMessagesOnConnection(
				c.connection, messages, 16);
			if (received <= 0)
			{
				break;
			}
			for (int m = 0; m < received; ++m)
			{
				int length = (int)messages[m]->m_cbSize;
				const unsigned char *data =
					(const unsigned char *)messages[m]->m_pData;
				int space = RingCapacity - 1 -
					((c.ringHead - c.ringTail + RingCapacity) % RingCapacity);
				if (length > space)
				{
					STEAMNET_LOG("ring overflow on connection " << i
						<< " -- dropping " << length << " bytes");
				}
				else
				{
					for (int b = 0; b < length; ++b)
					{
						c.ring[c.ringHead] = data[b];
						c.ringHead = (c.ringHead + 1) % RingCapacity;
					}
				}
				messages[m]->Release();
			}
		}
	}
}

//###########################################################################
// Install
//###########################################################################

int
	BTSteamNet_Install()
{
	if (steamActive)
	{
		return 0;
	}

	//
	// The 480 (Spacewar) test id if no appid file exists -- dev tooling.
	//
	FILE *appid = fopen("steam_appid.txt", "rt");
	if (appid == NULL)
	{
		appid = fopen("steam_appid.txt", "wt");
		if (appid != NULL)
		{
			fputs("480\n", appid);
		}
	}
	if (appid != NULL)
	{
		fclose(appid);
	}

	if (!SteamAPI_Init())
	{
		STEAMNET_LOG("SteamAPI_Init failed (Steam not running / no login) "
			"-- staying on Winsock");
		return -1;
	}

	ISteamNetworkingSockets *sockets = SteamNetworkingSockets();
	SteamNetworkingUtils()->InitRelayNetworkAccess();
	SteamNetworkingUtils()->SetGlobalCallback_SteamNetConnectionStatusChanged(
		OnConnectionStatusChanged);

	//
	// The token handoff (lobby -> mission process): my own token + the
	// token->identity map.  A lobby/menu process has neither -- fine, it
	// only needs matchmaking.
	//
	tokenCount = 0;
	myTokenBE = 0;
	const char *my_token = getenv("BT_FE_MYFAKE");
	if (my_token != NULL && my_token[0])
	{
		myTokenBE = inet_addr(my_token);
	}
	const char *map_text = getenv("BT_FE_STEAMMAP");
	if (map_text != NULL && map_text[0])
	{
		char map_copy[512];
		strncpy(map_copy, map_text, sizeof(map_copy) - 1);
		map_copy[sizeof(map_copy) - 1] = 0;
		char *cursor = strtok(map_copy, ";");
		while (cursor != NULL && tokenCount < 8)
		{
			char *equals = strchr(cursor, '=');
			if (equals != NULL)
			{
				*equals = 0;
				tokens[tokenCount].addressBE = inet_addr(cursor);
				tokens[tokenCount].steamID = _strtoui64(equals + 1, NULL, 10);
				if (tokens[tokenCount].addressBE != INADDR_NONE &&
					tokens[tokenCount].steamID != 0)
				{
					++tokenCount;
				}
			}
			cursor = strtok(NULL, ";");
		}
	}

	//
	// Listen on both identity channels (console = 0, game = 1).
	//
	for (int i = 0; i < FakePortCount; ++i)
	{
		listenSockets[i] = sockets->CreateListenSocketP2P(i, 0, NULL);
		acceptHead[i] = acceptTail[i] = 0;
		if (listenSockets[i] == k_HSteamListenSocket_Invalid)
		{
			STEAMNET_LOG("CreateListenSocketP2P(" << i << ") failed");
		}
	}

	steamActive = 1;
	STEAMNET_LOG("up -- identity "
		<< (unsigned long long)SteamUser()->GetSteamID().ConvertToUint64()
		<< ", " << tokenCount << " roster token(s)"
		<< (myTokenBE ? " incl. self" : ""));
	return 0;
}

int
	BTSteamNet_Active()
{
	return steamActive;
}

//###########################################################################
// The wire-seam surface
//###########################################################################

int
	BTSteamNet_Owns(SOCKET wire_socket)
{
	return steamActive && ConnectionOf(wire_socket) != NULL;
}

int
	BTSteamNet_Send(SOCKET wire_socket, const char *data, int length)
{
	BTSteamNet_Pump();
	SteamConnection *connection = ConnectionOf(wire_socket);
	if (connection == NULL || connection->closedByPeer)
	{
		WSASetLastError(WSAECONNRESET);
		return SOCKET_ERROR;
	}
	EResult result = SteamNetworkingSockets()->SendMessageToConnection(
		connection->connection, data, (uint32)length,
		k_nSteamNetworkingSend_ReliableNoNagle, NULL);
	if (result != k_EResultOK)
	{
		WSASetLastError(WSAEWOULDBLOCK);
		return SOCKET_ERROR;
	}
	return length;
}

int
	BTSteamNet_Recv(SOCKET wire_socket, char *buffer, int capacity)
{
	BTSteamNet_Pump();
	SteamConnection *connection = ConnectionOf(wire_socket);
	if (connection == NULL)
	{
		WSASetLastError(WSAENOTSOCK);
		return SOCKET_ERROR;
	}
	int available =
		(connection->ringHead - connection->ringTail + RingCapacity)
		% RingCapacity;
	if (available == 0)
	{
		if (connection->closedByPeer)
		{
			return 0;						// the engine's disconnect path
		}
		WSASetLastError(WSAEWOULDBLOCK);	// nonblocking no-data semantics
		return SOCKET_ERROR;
	}
	int count = (available < capacity) ? available : capacity;
	for (int b = 0; b < count; ++b)
	{
		buffer[b] = (char)connection->ring[connection->ringTail];
		connection->ringTail = (connection->ringTail + 1) % RingCapacity;
	}
	return count;
}

SOCKET
	BTSteamNet_Accept(SOCKET listener_socket)
{
	(void)listener_socket;
	if (!steamActive)
	{
		return INVALID_SOCKET;
	}
	BTSteamNet_Pump();
	//
	// The engine's console/game listeners map onto our two fake-port
	// channels; either TCP listener polling through BTNetAccept also
	// offers whatever Steam has queued.  Console arrivals are only
	// meaningful pre-mission, game arrivals during the mesh -- the
	// engine's own swap logic sorts them by peer address.
	//
	for (int channel = 0; channel < FakePortCount; ++channel)
	{
		if (acceptTail[channel] != acceptHead[channel])
		{
			SOCKET wire_socket = acceptQueue[channel][acceptTail[channel]];
			acceptTail[channel] = (acceptTail[channel] + 1) % AcceptQueueSize;
			return wire_socket;
		}
	}
	return INVALID_SOCKET;
}

void
	BTSteamNet_Close(SOCKET wire_socket)
{
	SteamConnection *connection = ConnectionOf(wire_socket);
	if (connection == NULL)
	{
		return;
	}
	SteamNetworkingSockets()->CloseConnection(
		connection->connection, 0, "closed", true);
	connection->inUse = 0;
	connection->connection = k_HSteamNetConnection_Invalid;
}

int
	BTSteamNet_PeerAddress(SOCKET wire_socket, SOCKADDR_IN *out)
{
	SteamConnection *connection = ConnectionOf(wire_socket);
	if (connection == NULL)
	{
		return 0;
	}
	*out = connection->peerAddress;
	return connection->peerAddress.sin_family != 0;
}

//###########################################################################
// Connection establishment + the lobby surface
//###########################################################################

int
	BTSteamNet_IsFakeAddress(unsigned long internet_address_be)
{
	if (!steamActive)
	{
		return 0;
	}
	for (int t = 0; t < tokenCount; ++t)
	{
		if (tokens[t].addressBE == internet_address_be)
		{
			return 1;
		}
	}
	return 0;
}

SOCKET
	BTSteamNet_Connect(unsigned long internet_address_be, int remote_port)
{
	if (!steamActive)
	{
		return INVALID_SOCKET;
	}
	unsigned long long steam_id = 0;
	for (int t = 0; t < tokenCount; ++t)
	{
		if (tokens[t].addressBE == internet_address_be)
		{
			steam_id = tokens[t].steamID;
		}
	}
	if (steam_id == 0)
	{
		return INVALID_SOCKET;
	}
	int channel = (remote_port == TokenConsolePort) ? 0 : 1;
	SteamNetworkingIdentity identity;
	identity.Clear();
	identity.SetSteamID64(steam_id);

	HSteamNetConnection handle =
		SteamNetworkingSockets()->ConnectP2P(identity, channel, 0, NULL);
	if (handle == k_HSteamNetConnection_Invalid)
	{
		return INVALID_SOCKET;
	}
	SteamConnection *connection = AllocateConnection(handle);
	if (connection == NULL)
	{
		SteamNetworkingSockets()->CloseConnection(handle, 0, "table full", false);
		return INVALID_SOCKET;
	}
	connection->peerAddress.sin_family = AF_INET;
	connection->peerAddress.sin_addr.s_addr = internet_address_be;
	connection->peerAddress.sin_port = htons((unsigned short)remote_port);
	(void)identity;

	//
	// The engine's OpenConnection is synchronous (blocking connect); wait
	// out the SDR handshake here, bounded.
	//
	for (int wait = 0; wait < 300; ++wait)		// <= 30 s
	{
		BTSteamNet_Pump();
		if (connection->connected)
		{
			STEAMNET_LOG("connected to peer " << steam_id
				<< " channel " << channel);
			return PseudoSocketOf(connection);
		}
		if (connection->closedByPeer)
		{
			break;
		}
		Sleep(100);
	}
	STEAMNET_LOG("connect failed/timed out");
	BTSteamNet_Close(PseudoSocketOf(connection));
	return INVALID_SOCKET;
}

int
	BTSteamNet_GetFakeAddressRaw(unsigned long *internet_address_be)
{
	if (!steamActive || myTokenBE == 0)
	{
		return -1;
	}
	*internet_address_be = myTokenBE;
	return 0;
}

unsigned long long
	BTSteamNet_MySteamID()
{
	if (!steamActive)
	{
		return 0;
	}
	return SteamUser()->GetSteamID().ConvertToUint64();
}