vwemirror/RP412
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Network races: the local console marshals remote pods over the wire

Browse Source 
The lobby-owner-as-console architecture, in-engine. RPL4CONSOLE gains
RPL4LocalConsole_InstallNetworkRace: the owner pod switches to network
mode and meshes like any pod (egg fed locally via FeedLocalEgg, which
now opens the ConsoleOnly state gate), while the console tick also
marshals REMOTE pods over NetTransport speaking the exact arcade
protocol - egg chunks with 5s-retry-until-ACK, 1Hz state polling,
RunMission once every pod stages at WaitingForLaunch, StopMission at
expiry (remotes first, local pod holds until their EndMission scores
land), score intake labeled with [pilots]-order names on the results
screen.

The front end builds the multi-pilot egg: RP412HOSTPODS lists member
console channels (lobby stand-in; the Steam lobby feeds the same
path), RP412HOSTPORT/RP412HOSTADDR set the owner side.

Winsock Connect now redials with a fresh socket per attempt (a refused
TCP socket is dead; the old loop reused it) bounded at 120s - needed
whenever a peer boots after the caller, which is the normal Steam
lobby launch order.

Verified on loopback: member pod in -net, owner hosting from its menu;
mesh completed both sides, 30s race, remote score collected over the
wire (host 3), local stop after the drain, results screen shows both
pilots by name in one process that returns to the menu.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Cyd
2026-07-12 21:08:18 -05:00
co-authored by Claude Fable 5
parent a8053ad784
commit c66cb00a7e
8 changed files with 810 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
RP_L4/RPL4CONSOLE.cpp
+518 -35
View File
@@ -5,9 +5,13 @@
#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
@@ -18,8 +22,18 @@
// requested StopMissionMessage dispatch (the engine is single
// threaded; cross-thread dispatch is not safe).
//
// Results flow in through gConsoleScoreSink (RP layer): the same final
// scores RPPlayer sent the arcade console at mission end.
// 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
@@ -54,6 +68,44 @@ namespace
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
//---------------------------------------------------------------
@@ -77,8 +129,8 @@ namespace
}
//---------------------------------------------------------------
// Final-score intake (called on the game thread from RPPlayer's
// mission-ending path, via the RP-layer sink)
// 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)
{
@@ -92,6 +144,214 @@ namespace
<< " = " << 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
//---------------------------------------------------------------
@@ -111,6 +371,46 @@ namespace
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;
@@ -125,24 +425,53 @@ namespace
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, otherwise the mission
// clock rolls past 00:00 and counts up forever.
// 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.
//-----------------------------------------------------
DEBUG_STREAM << "LocalConsole: time expired - stopping mission\n" << std::flush;
InterlockedExchange(&gMissionRunning, 0);
Application::StopMissionMessage message(0);
application->Dispatch(&message);
gPhase = PhaseStopped;
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;
@@ -154,41 +483,184 @@ namespace
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)
{
// debug: L4CONSOLELEN overrides the mission length (test races)
const char *override_string = getenv("L4CONSOLELEN");
if (override_string != NULL && atoi(override_string) > 0)
gNetworkRace = False;
gRemotePodCount = 0;
gPilotNameCount = 0;
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;
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)
{
mission_seconds = atoi(override_string);
DEBUG_STREAM << "LocalConsole: L4CONSOLELEN override, "
<< mission_seconds << "s\n" << std::flush;
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;
}
}
}
gMissionSeconds = mission_seconds;
InterlockedExchange(&gLengthMs, (LONG) mission_seconds * 1000);
gPhase = PhaseWaiting;
gWatchedApp = NULL;
// 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)
//
// 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)
{
gConsoleThread = CreateThread(
NULL, 0, ConsoleThreadProc, NULL, 0, NULL);
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: installed (length "
<< mission_seconds << "s, console thread "
<< (gConsoleThread != NULL ? "up" : "FAILED") << ")\n" << std::flush;
DEBUG_STREAM << "LocalConsole: network race, " << gRemotePodCount
<< " remote pod(s) connected\n" << std::flush;
InstallCommon(mission_seconds);
return True;
}
Logical
@@ -214,3 +686,14 @@ Logical
*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];
}