RP412/RP_L4/RPL4CONSOLE.cpp

#include "..\munga_l4\mungal4.h"
#pragma hdrstop

#include "rpl4console.h"
#include "rpl4fe.h"
#include "..\munga\appmgr.h"
#include "..\munga\appmsg.h"
#include "..\munga\console.h"
#include "..\rp\rpcnsl.h"
#include "..\munga_l4\l4app.h"
#include "..\munga_l4\l4net.h"
#include "..\munga_l4\l4nettransport.h"

#define CONSOLE_NET_PORT	1501	// arcade default (matches L4NET.CPP)

//########################################################################
// The local console runs on ITS OWN THREAD, like the real console: it
// stays alive across the whole session, owns the mission clock, and
// raises the stop request when the selected length expires. The game
// thread's per-frame tick is the only place engine calls happen - it
// reports state transitions to the console thread and executes the
// requested StopMissionMessage dispatch (the engine is single
// threaded; cross-thread dispatch is not safe).
//
// NETWORK RACES (lobby owner as console): the same tick additionally
// marshals REMOTE pods over the NetTransport wire, speaking the exact
// arcade console protocol - egg chunks + ACK, state queries,
// RunMission when everyone reaches WaitingForLaunch, StopMission at
// expiry, EndMission score intake. The owner's own pod runs in
// network mode (it meshes like any pod) but is fed its egg locally
// and driven by direct engine calls, so the console never needs a
// connection to itself.
//
// Results flow in through gConsoleScoreSink (RP layer) for the local
// pod and EndMission wire messages for remote pods: the same final
// scores every pod sent the arcade console at mission end.
//########################################################################

namespace
{
	enum ConsolePhase
	{
		PhaseWaiting = 0,	// waiting for the mission to start running
		PhaseRunning,		// mission running, console thread watching the clock
		PhaseStopped		// stop dispatched, waiting for teardown
	};

	ConsolePhase	gPhase = PhaseWaiting;
	int				gMissionSeconds = 0;
	Application		*gWatchedApp = NULL;

	HANDLE			gConsoleThread = NULL;

	// shared with the console thread
	volatile LONG	gMissionRunning = 0;
	volatile LONG	gStopRequested = 0;
	volatile LONG	gShuttingDown = 0;
	volatile LONG	gRunStartTick = 0;
	volatile LONG	gLengthMs = 0;

	// collected mission results (this session's last race)
	enum { maxResults = 16 };
	struct FinalScore
	{
		int	hostID;
		int	score;
	};
	FinalScore		gResults[maxResults];
	int				gResultCount = 0;

	//---------------------------------------------------------------
	// Network race state: remote pods marshaled over the wire
	//---------------------------------------------------------------
	enum { maxRemotePods = 8 };
	enum { remoteRxSize = 8192 };

	struct RemotePod
	{
		char		address[64];		// console channel, "ip[:port]"
		NetTransport::Connection
					connection;
		int			state;				// last reported application state (-1 unknown)
		Logical		eggAcknowledged;
		DWORD		lastQueryTick;
		DWORD		eggSentTick;		// 0 = never sent
		Logical		scored;
		char		rx[remoteRxSize];	// wire frame reassembly
		int			rxCount;
	};

	RemotePod		gRemotePods[maxRemotePods];
	int				gRemotePodCount = 0;

	Logical			gNetworkRace = False;
	char			gEggPath[MAX_PATH] = "";
	char			*gEggWire = NULL;		// newline->NUL image for chunking
	int				gEggWireSize = 0;
	Logical			gLocalEggFed = False;
	Logical			gRunSent = False;
	Logical			gRemoteStopsSent = False;
	DWORD			gRemoteStopTick = 0;

	// pilot names in [pilots] order; host IDs start at FirstLegalHostID+1
	// (the console reserves the first), so host 2 = pilot index 0
	enum { firstPilotHostID = 2 };
	char			gPilotNames[maxRemotePods + 1][32];
	int				gPilotNameCount = 0;

	//---------------------------------------------------------------
	// The console thread: the mission clock lives here
	//---------------------------------------------------------------
	DWORD WINAPI ConsoleThreadProc(LPVOID)
	{
		while (!gShuttingDown)
		{
			Sleep(250);

			if (gMissionRunning && !gStopRequested)
			{
				LONG length_ms = gLengthMs;
				if (length_ms > 0 &&
					(LONG)(GetTickCount() - (DWORD) gRunStartTick) >= length_ms)
				{
					InterlockedExchange(&gStopRequested, 1);
				}
			}
		}
		return 0;
	}

	//---------------------------------------------------------------
	// Final-score intake (game thread: the RP-layer sink for the
	// local pod, the wire pump for remote pods)
	//---------------------------------------------------------------
	void CollectFinalScore(int host_ID, int score)
	{
		if (gResultCount < maxResults)
		{
			gResults[gResultCount].hostID = host_ID;
			gResults[gResultCount].score = score;
			++gResultCount;
		}
		DEBUG_STREAM << "LocalConsole: final score, host " << host_ID
			<< " = " << score << "\n" << std::flush;
	}

	//---------------------------------------------------------------
	// The wire: the arcade console protocol over NetTransport
	//---------------------------------------------------------------
	void SendWire(RemotePod *pod, int client_ID, const void *message, int size)
	{
		char packet[sizeof(NetworkPacketHeader) + 1400];
		if (size > (int) sizeof(packet) - (int) sizeof(NetworkPacketHeader))
		{
			return;
		}
		memset(packet, 0, sizeof(NetworkPacketHeader));
		NetworkPacketHeader *header = (NetworkPacketHeader *) packet;
		header->clientID = (NetworkClient::ClientID) client_ID;
		header->gameID = 0;
		header->fromHost = 1;		// the console's reserved host ID
		memcpy(packet + sizeof(NetworkPacketHeader), message, size);
		NetTransport_Get()->Send(
			pod->connection, packet, (int) sizeof(NetworkPacketHeader) + size);
	}

	void SendEggTo(RemotePod *pod)
	{
		int chunk_count = (gEggWireSize + 999) / 1000;
		for (int i = 0; i < chunk_count; ++i)
		{
			int offset = i * 1000;
			int length = gEggWireSize - offset;
			if (length > 1000)
			{
				length = 1000;
			}
			NetworkManager__ReceiveEggFileMessage chunk(
				i, gEggWireSize, gEggWire + offset, length);
			SendWire(pod, NetworkClient::NetworkManagerClientID,
				&chunk, (int) chunk.messageLength);
		}
		pod->eggSentTick = GetTickCount();
		DEBUG_STREAM << "LocalConsole: egg sent to " << pod->address
			<< " (" << chunk_count << " chunks)\n" << std::flush;
	}

	void PumpRemote(RemotePod *pod)
	{
		//
		// Read whatever the wire has pending
		//
		for (;;)
		{
			int space = remoteRxSize - pod->rxCount;
			if (space <= 0)
			{
				break;
			}
			int received = NetTransport_Get()->Receive(
				pod->connection, pod->rx + pod->rxCount, space);
			if (received <= 0)
			{
				break;		// no data / disconnected
			}
			pod->rxCount += received;
			if (received < space)
			{
				break;
			}
		}

		//
		// Parse complete frames: NetworkPacketHeader + engine message
		//
		const int header_size = (int) sizeof(NetworkPacketHeader);
		const int base_size = (int) sizeof(Receiver__Message);
		for (;;)
		{
			if (pod->rxCount < header_size + base_size)
			{
				break;
			}
			NetworkPacketHeader *header = (NetworkPacketHeader *) pod->rx;
			Receiver__Message *base = (Receiver__Message *)(pod->rx + header_size);
			int total = header_size + (int) base->messageLength;
			if (total < header_size + base_size || total > remoteRxSize)
			{
				DEBUG_STREAM << "LocalConsole: garbage frame from "
					<< pod->address << " - dropping buffer\n" << std::flush;
				pod->rxCount = 0;
				break;
			}
			if (pod->rxCount < total)
			{
				break;
			}

			if ((int) header->clientID == (int) NetworkClient::ConsoleClientID)
			{
				if ((int) base->messageID == ConsoleApplicationStateResponseMessageID)
				{
					ConsoleApplicationStateResponseMessage *message =
						(ConsoleApplicationStateResponseMessage *) base;
					if (pod->state != (int) message->GetApplicationState())
					{
						DEBUG_STREAM << "LocalConsole: " << pod->address
							<< " state -> " << (int) message->GetApplicationState()
							<< "\n" << std::flush;
					}
					pod->state = (int) message->GetApplicationState();
				}
				else if ((int) base->messageID == ConsoleApplicationEndMissionMessageID)
				{
					ConsoleApplicationEndMissionMessage *message =
						(ConsoleApplicationEndMissionMessage *) base;
					CollectFinalScore(
						(int) message->GetPlayerHostID(),
						(int) message->GetFinalScore());
					pod->scored = True;
				}
				// VTV telemetry (IDs 2-6) skips through for now
			}
			else if ((int) header->clientID == (int) NetworkClient::NetworkManagerClientID)
			{
				if ((int) base->messageID == (int) L4NetworkManager::AcknowledgeEggFileMessageID)
				{
					if (!pod->eggAcknowledged)
					{
						DEBUG_STREAM << "LocalConsole: " << pod->address
							<< " EGG ACK (mesh complete)\n" << std::flush;
					}
					pod->eggAcknowledged = True;
				}
			}

			memmove(pod->rx, pod->rx + total, pod->rxCount - total);
			pod->rxCount -= total;
		}
	}

	void MarshalRemotes()
	{
		DWORD now = GetTickCount();
		for (int i = 0; i < gRemotePodCount; ++i)
		{
			RemotePod *pod = &gRemotePods[i];

			// state poll, once a second (the arcade console's cadence)
			if ((LONG)(now - pod->lastQueryTick) >= 1000)
			{
				Application::StateQueryMessage query(1);
				SendWire(pod, NetworkClient::ApplicationClientID,
					&query, (int) query.messageLength);
				pod->lastQueryTick = now;
			}

			PumpRemote(pod);

			// egg feed: 5s retry until the pod ACKs (post-mesh)
			if (pod->state == (int) Application::WaitingForEgg &&
				!pod->eggAcknowledged &&
				(pod->eggSentTick == 0 || (LONG)(now - pod->eggSentTick) >= 5000))
			{
				SendEggTo(pod);
			}
		}
	}

	Logical AllRemotesInState(int state)
	{
		for (int i = 0; i < gRemotePodCount; ++i)
		{
			if (gRemotePods[i].state != state)
			{
				return False;
			}
		}
		return True;
	}

	Logical AllRemotesScored()
	{
		for (int i = 0; i < gRemotePodCount; ++i)
		{
			if (!gRemotePods[i].scored)
			{
				return False;
			}
		}
		return True;
	}

	void DisconnectRemotes()
	{
		for (int i = 0; i < gRemotePodCount; ++i)
		{
			if (gRemotePods[i].connection != NetTransport::InvalidConnection)
			{
				NetTransport_Get()->Close(gRemotePods[i].connection);
				gRemotePods[i].connection = NetTransport::InvalidConnection;
			}
		}
	}

	void DispatchLocalStop()
	{
		DEBUG_STREAM << "LocalConsole: stopping local pod\n" << std::flush;
		InterlockedExchange(&gMissionRunning, 0);
		Application::StopMissionMessage message(0);
		application->Dispatch(&message);
		gPhase = PhaseStopped;
	}

	//---------------------------------------------------------------
	// The game-thread tick: state reporting + engine-safe execution
	//---------------------------------------------------------------
	void ConsoleTick()
	{
		if (application == NULL)
		{
			return;
		}
		if (gPhase != PhaseWaiting && application != gWatchedApp)
		{
			return;
		}

		int state = application->GetApplicationState();

		switch (gPhase)
		{
		case PhaseWaiting:
			if (gNetworkRace)
			{
				MarshalRemotes();

				//
				// Feed our own pod its egg locally: it meshes like any
				// pod but the console drives it without a connection
				//
				if (!gLocalEggFed && state == Application::WaitingForEgg)
				{
					L4NetworkManager *network_manager =
						(L4NetworkManager *) application->GetNetworkManager();
					if (network_manager != NULL)
					{
						network_manager->FeedLocalEgg(gEggPath);
						gLocalEggFed = True;
						DEBUG_STREAM << "LocalConsole: local egg fed\n" << std::flush;
					}
				}

				//
				// Everyone staged: launch the race everywhere
				//
				if (!gRunSent &&
					state == Application::WaitingForLaunch &&
					AllRemotesInState(Application::WaitingForLaunch))
				{
					DEBUG_STREAM << "LocalConsole: all pods staged - RUN\n" << std::flush;
					for (int i = 0; i < gRemotePodCount; ++i)
					{
						Application::RunMissionMessage run;
						SendWire(&gRemotePods[i], NetworkClient::ApplicationClientID,
							&run, (int) run.messageLength);
					}
					Application::RunMissionMessage local_run;
					application->Dispatch(&local_run);
					gRunSent = True;
				}
			}

			if (state == Application::RunningMission)
			{
				gPhase = PhaseRunning;
				gWatchedApp = application;
				gResultCount = 0;
				InterlockedExchange(&gRunStartTick, (LONG) GetTickCount());
				InterlockedExchange(&gStopRequested, 0);
				InterlockedExchange(&gMissionRunning, 1);
				DEBUG_STREAM << "LocalConsole: mission running, length "
					<< gMissionSeconds << "s\n" << std::flush;
			}
			break;

		case PhaseRunning:
			if (gNetworkRace)
			{
				// telemetry + final scores keep flowing during the race
				for (int i = 0; i < gRemotePodCount; ++i)
				{
					PumpRemote(&gRemotePods[i]);
				}
			}

			if (state != Application::RunningMission)
			{
				// mission ended some other way (pilot exit etc.)
				InterlockedExchange(&gMissionRunning, 0);
				gPhase = PhaseStopped;
				DisconnectRemotes();
			}
			else if (gStopRequested)
			{
				//-----------------------------------------------------
				// The console clock expired: end the race exactly the
				// way the arcade console did. Remote pods stop first;
				// the local pod holds on briefly so their EndMission
				// scores can land before our own teardown.
				//-----------------------------------------------------
				if (!gNetworkRace)
				{
					DEBUG_STREAM << "LocalConsole: time expired - stopping mission\n" << std::flush;
					DispatchLocalStop();
				}
				else if (!gRemoteStopsSent)
				{
					DEBUG_STREAM << "LocalConsole: time expired - stopping remote pods\n" << std::flush;
					for (int i = 0; i < gRemotePodCount; ++i)
					{
						Application::StopMissionMessage stop(0);
						SendWire(&gRemotePods[i], NetworkClient::ApplicationClientID,
							&stop, (int) stop.messageLength);
					}
					gRemoteStopsSent = True;
					gRemoteStopTick = GetTickCount();
				}
				else if (AllRemotesScored() ||
					(LONG)(GetTickCount() - gRemoteStopTick) >= 5000)
				{
					DispatchLocalStop();
					DisconnectRemotes();
				}
			}
			break;

		case PhaseStopped:
			// The application tears itself down after a stop (arcade
			// pods were relaunched per mission). WinMain's race loop
			// asks MissionCompleted() and cycles back to the setup
			// screen in the same process.
			break;
		}
	}

	//---------------------------------------------------------------
	// Shared install plumbing
	//---------------------------------------------------------------
	void InstallCommon(int mission_seconds)
	{
		// debug: L4CONSOLELEN overrides the mission length (test races)
		const char *override_string = getenv("L4CONSOLELEN");
		if (override_string != NULL && atoi(override_string) > 0)
		{
			mission_seconds = atoi(override_string);
			DEBUG_STREAM << "LocalConsole: L4CONSOLELEN override, "
				<< mission_seconds << "s\n" << std::flush;
		}

		gMissionSeconds = mission_seconds;
		InterlockedExchange(&gLengthMs, (LONG) mission_seconds * 1000);
		gPhase = PhaseWaiting;
		gWatchedApp = NULL;
		gRunSent = False;
		gRemoteStopsSent = False;
		gLocalEggFed = False;

		// game-thread execution point
		gPerFrameHook = &ConsoleTick;

		// results intake from the RP layer
		gConsoleScoreSink = &CollectFinalScore;

		// the console itself lives on its own thread, like the real one
		if (gConsoleThread == NULL)
		{
			gConsoleThread = CreateThread(
				NULL, 0, ConsoleThreadProc, NULL, 0, NULL);
		}

		DEBUG_STREAM << "LocalConsole: installed (length "
			<< mission_seconds << "s, console thread "
			<< (gConsoleThread != NULL ? "up" : "FAILED") << ")\n" << std::flush;
	}
}

void
	RPL4LocalConsole_Install(int mission_seconds)
{
	gNetworkRace = False;
	gRemotePodCount = 0;
	gPilotNameCount = 0;
	// single player launches itself (the engine's no-console self-run)
	gConsoleMarshalsLaunch = False;
	InstallCommon(mission_seconds);
}

Logical
	RPL4LocalConsole_InstallNetworkRace(
		int mission_seconds,
		const char *egg_path,
		const char *remote_pod_list,
		const char *pilot_names
	)
{
	gNetworkRace = True;
	gRemotePodCount = 0;
	gPilotNameCount = 0;

	// the owner pod must stage at WaitingForLaunch with everyone else -
	// this console launches the whole mesh at once
	gConsoleMarshalsLaunch = True;

	strncpy(gEggPath, egg_path, sizeof(gEggPath) - 1);
	gEggPath[sizeof(gEggPath) - 1] = '\0';

	//
	// The wire image of the egg: file newlines become NULs, exactly
	// what the arcade console sent (RPMission.ToEggFileMessages)
	//
	if (gEggWire != NULL)
	{
		delete[] gEggWire;
		gEggWire = NULL;
		gEggWireSize = 0;
	}
	FILE *egg_file = fopen(egg_path, "rb");
	if (egg_file == NULL)
	{
		DEBUG_STREAM << "LocalConsole: cannot read egg " << egg_path << "\n" << std::flush;
		return False;
	}
	fseek(egg_file, 0, SEEK_END);
	long raw_size = ftell(egg_file);
	fseek(egg_file, 0, SEEK_SET);
	char *raw = new char[raw_size];
	fread(raw, 1, raw_size, egg_file);
	fclose(egg_file);

	gEggWire = new char[raw_size];
	gEggWireSize = 0;
	for (long b = 0; b < raw_size; ++b)
	{
		if (raw[b] == '\r')
		{
			continue;			// \r\n collapses to one NUL
		}
		gEggWire[gEggWireSize++] = (raw[b] == '\n') ? '\0' : raw[b];
	}
	delete[] raw;

	//
	// Pilot names in [pilots] order (results screen labels)
	//
	if (pilot_names != NULL)
	{
		const char *cursor = pilot_names;
		while (*cursor != '\0' && gPilotNameCount < maxRemotePods + 1)
		{
			int length = 0;
			while (cursor[length] != '\0' && cursor[length] != ',' &&
				length < (int) sizeof(gPilotNames[0]) - 1)
			{
				gPilotNames[gPilotNameCount][length] = cursor[length];
				++length;
			}
			gPilotNames[gPilotNameCount][length] = '\0';
			++gPilotNameCount;
			cursor += length;
			if (*cursor == ',')
			{
				++cursor;
			}
		}
	}

	//
	// Connect to every remote pod's console channel. Blocking with
	// retry, like the arcade console redialing a pod that is still
	// booting; runs before the engine block so nothing is waiting.
	//
	NetTransport_Get()->Startup();
	const char *cursor = remote_pod_list;
	while (*cursor != '\0' && gRemotePodCount < maxRemotePods)
	{
		RemotePod *pod = &gRemotePods[gRemotePodCount];
		memset(pod, 0, sizeof(*pod));
		pod->state = -1;
		pod->connection = NetTransport::InvalidConnection;

		int length = 0;
		while (cursor[length] != '\0' && cursor[length] != ',' &&
			length < (int) sizeof(pod->address) - 1)
		{
			pod->address[length] = cursor[length];
			++length;
		}
		pod->address[length] = '\0';
		cursor += length;
		if (*cursor == ',')
		{
			++cursor;
		}

		SOCKADDR_IN console_address;
		NetTransport_Get()->Resolve(pod->address, &console_address);
		if (console_address.sin_port == 0)
		{
			console_address.sin_port = htons(CONSOLE_NET_PORT);
		}

		DEBUG_STREAM << "LocalConsole: connecting to pod " << pod->address
			<< "...\n" << std::flush;
		pod->connection = NetTransport_Get()->Connect(&console_address, 0);
		if (pod->connection == NetTransport::InvalidConnection)
		{
			DEBUG_STREAM << "LocalConsole: could not reach pod "
				<< pod->address << "\n" << std::flush;
			return False;
		}
		++gRemotePodCount;
	}

	DEBUG_STREAM << "LocalConsole: network race, " << gRemotePodCount
		<< " remote pod(s) connected\n" << std::flush;

	InstallCommon(mission_seconds);
	return True;
}

Logical
	RPL4LocalConsole_MissionCompleted()
{
	return gPhase == PhaseStopped;
}

int
	RPL4LocalConsole_ResultCount()
{
	return gResultCount;
}

Logical
	RPL4LocalConsole_GetResult(int index, int *host_ID, int *score)
{
	if (index < 0 || index >= gResultCount)
	{
		return False;
	}
	*host_ID = gResults[index].hostID;
	*score = gResults[index].score;
	return True;
}

const char *
	RPL4LocalConsole_GetResultName(int host_ID)
{
	int index = host_ID - firstPilotHostID;
	if (index < 0 || index >= gPilotNameCount)
	{
		return NULL;
	}
	return gPilotNames[index];
}