#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]; } void RPL4LocalConsole_ClearResults() { gResultCount = 0; gPilotNameCount = 0; } void RPL4LocalConsole_InjectResult(int host_ID, int score, const char *name) { if (gResultCount >= maxResults) { return; } gResults[gResultCount].hostID = host_ID; gResults[gResultCount].score = score; ++gResultCount; int index = host_ID - firstPilotHostID; if (name != NULL && index >= 0 && index < maxRemotePods + 1) { strncpy(gPilotNames[index], name, sizeof(gPilotNames[index]) - 1); gPilotNames[index][sizeof(gPilotNames[index]) - 1] = '\0'; if (index >= gPilotNameCount) { gPilotNameCount = index + 1; } } }