--- a/rts/Net/GameServer.h +++ b/rts/Net/GameServer.h @@ -34,6 +34,7 @@ class UDPListener; } class CDemoReader; +class WebbarResumeReader; class Action; class CDemoRecorder; class AutohostInterface; @@ -288,6 +289,10 @@ std::unique_ptr udpListener; std::unique_ptr demoReader; + // WebBAR match recovery (WebbarResume.h): recorded input replayed into this live game. + std::unique_ptr resumeReader; + bool resumeActive = false; + void SendResumeData(); std::unique_ptr demoRecorder; std::unique_ptr hostif; --- a/rts/Net/GameServer.cpp +++ b/rts/Net/GameServer.cpp @@ -52,6 +52,7 @@ #include "System/Net/UnpackPacket.h" #include "System/LoadSave/DemoRecorder.h" #include "System/LoadSave/DemoReader.h" +#include "Net/WebbarResume.h" #include "System/Log/ILog.h" #include "System/Platform/errorhandler.h" #include "System/Platform/Threading.h" @@ -188,6 +189,20 @@ demoReader.reset(new CDemoReader(myGameSetup->demoName, modGameTime + 0.1f)); } + if (demoReader == nullptr) { + if (const char* path = std::getenv("WEBBAR_RESUME_STREAM")) { + resumeReader.reset(new WebbarResumeReader()); + if (resumeReader->Load(path)) { + webbarResumeTarget = resumeReader->frames; + Message(spring::format("[WebbarResume] replaying %d recorded frames (%u packets)", resumeReader->frames, unsigned(resumeReader->packets.size())), false); + } else { + Message("[WebbarResume] the recorded stream could not be read; starting live", false); + resumeReader.reset(); + webbarResumeTarget = -2; // failure, reported to the browser + } + } + } + // initialize players, teams & ais { netPingTimings.fill(spring_notime); @@ -550,6 +565,37 @@ } return ret; +} + +void CGameServer::SendResumeData() +{ + // Pace by the local client's progress (not wall time), so catch-up runs as fast as it simulates. + unsigned int sent = 0; + while (resumeReader->HasData() && sent < 2 * GAME_SPEED) { + if (HasLocalClient() && (serverFrameNum - players[localClientNumber].lastFrameResponse) >= 2 * GAME_SPEED) + break; + const WebbarResumeReader::Packet p = resumeReader->Peek(); + resumeReader->Pop(); + const uint8_t* bytes = resumeReader->Bytes(p); + if (!WebbarResumeReader::Forwarded(bytes[0])) + continue; + if (bytes[0] == NETMSG_NEWFRAME || bytes[0] == NETMSG_KEYFRAME) { + ++serverFrameNum; + ++sent; + lastNewFrameTick = spring_gettime(); + #ifdef SYNCCHECK + outstandingSyncFrames.insert(serverFrameNum); + #endif + } + Broadcast(std::make_shared(bytes, p.length)); + } + if (!resumeReader->HasData()) { + resumeActive = false; + UserSpeedChange(1.0f, SERVER_PLAYER); + frameTimeLeft = 0.0f; + lastNewFrameTick = spring_gettime(); + Message(spring::format("[WebbarResume] recording replayed through frame %d; live play continues", serverFrameNum), false); + } } void CGameServer::Broadcast(std::shared_ptr packet) @@ -1049,6 +1095,9 @@ // Ignore packets from clients in process of disconnecting. if (players[a].myState == GameParticipant::DISCONNECTING) + return; + // A resumed match takes all synced input from the recording until it catches up. + if (resumeActive && WebbarResumeReader::LiveInputDropped(msgCode)) return; switch (msgCode) { @@ -2179,6 +2228,9 @@ for (int i = 0; i<16; ++i) gameID.charArray[i] = p[i]; } + + if (resumeReader != nullptr && resumeReader->hasGameID) + std::memcpy(gameID.charArray, resumeReader->gameID, sizeof(gameID.charArray)); Broadcast(CBaseNetProtocol::Get().SendGameID(gameID.charArray)); @@ -2325,6 +2377,11 @@ } Broadcast(CBaseNetProtocol::Get().SendStartPlaying(0)); + resumeActive = (resumeReader != nullptr); + // Catch up at the fastest allowed speed (the client consumes frames at its speed factor); + // the speed factor is not synced state, so this does not change the simulation. + if (resumeActive) + UserSpeedChange(maxUserSpeed, SERVER_PLAYER); if (hostif != nullptr) { if (demoRecorder != nullptr) { @@ -2654,6 +2711,11 @@ if (demoReader != nullptr) { CheckSync(); SendDemoData(-1); + return; + } + if (resumeActive) { + CheckSync(); + SendResumeData(); return; } --- /dev/null +++ b/rts/Net/WebbarResume.h @@ -0,0 +1,100 @@ +/* This file is part of the Spring engine (GPL v2 or later), see LICENSE.html */ +#pragma once + +// WebBAR match recovery (F138). A browser-exported client packet stream (the demo stream after +// the file header and setup text: {DemoStreamChunkHeader, packet} records) is replayed into a +// fresh live local game with the same start script. The server takes frame boundaries and every +// synced input from the recording, drops the live duplicates, then hands over to live play at the +// end of the recording. Inert unless WEBBAR_RESUME_STREAM names a readable file. + +#include +#include +#include +#include +#include +#include +#include + +#include "Net/Protocol/NetMessageTypes.h" +#include "System/LoadSave/demofile.h" + +// Shared with the client-side bridge (WebbarBrowser.h): the last recorded frame, or -1. +inline std::atomic webbarResumeTarget{-1}; + +class WebbarResumeReader +{ +public: + struct Packet { size_t offset; uint32_t length; }; + + bool Load(const char* path) { + std::ifstream in(path, std::ios::binary); + if (!in) + return false; + data.assign(std::istreambuf_iterator(in), std::istreambuf_iterator()); + bool started = false; + size_t o = 0; + while (o + sizeof(DemoStreamChunkHeader) <= data.size()) { + DemoStreamChunkHeader h; + std::memcpy(&h, &data[o], sizeof(h)); + h.swab(); + const size_t p = o + sizeof(h); + if (h.length == 0 || p + h.length > data.size()) + break; // a truncated tail (interrupted export) ends the recording + const uint8_t msg = data[p]; + if (msg == NETMSG_GAMEID && h.length >= 17) { + std::memcpy(gameID, &data[p + 1], 16); + hasGameID = true; + } + if (started) { + packets.push_back({p, h.length}); + frames += (msg == NETMSG_NEWFRAME || msg == NETMSG_KEYFRAME); + } else if (msg == NETMSG_STARTPLAYING) { + started = true; // the live pre-game handshake replaces everything before this + } + o = p + h.length; + } + return started && frames > 0; + } + + bool HasData() const { return next < packets.size(); } + const Packet& Peek() const { return packets[next]; } + void Pop() { ++next; } + const uint8_t* Bytes(const Packet& p) const { return &data[p.offset]; } + + // Recorded server messages that carry synced input or frame boundaries. + static bool Forwarded(uint8_t msg) { + switch (msg) { + case NETMSG_NEWFRAME: case NETMSG_KEYFRAME: + case NETMSG_COMMAND: case NETMSG_SELECT: + case NETMSG_AICOMMAND: case NETMSG_AICOMMANDS: case NETMSG_AICOMMAND_TRACKED: case NETMSG_AISHARE: + case NETMSG_LUAMSG: case NETMSG_SHARE: case NETMSG_SETSHARE: + case NETMSG_DIRECT_CONTROL: case NETMSG_DC_UPDATE: + case NETMSG_ALLIANCE: case NETMSG_TEAM: case NETMSG_CCOMMAND: + case NETMSG_CHAT: case NETMSG_SYSTEMMSG: + return true; + default: + return false; + } + } + // Live client input dropped while the recording drives the game (the recorded copy applies). + static bool LiveInputDropped(uint8_t msg) { + switch (msg) { + case NETMSG_COMMAND: case NETMSG_SELECT: + case NETMSG_AICOMMAND: case NETMSG_AICOMMANDS: case NETMSG_AICOMMAND_TRACKED: case NETMSG_AISHARE: + case NETMSG_LUAMSG: case NETMSG_SHARE: case NETMSG_SETSHARE: + case NETMSG_DIRECT_CONTROL: case NETMSG_DC_UPDATE: + case NETMSG_ALLIANCE: case NETMSG_TEAM: + case NETMSG_PAUSE: case NETMSG_USER_SPEED: + return true; + default: + return false; + } + } + + std::vector data; + std::vector packets; + size_t next = 0; + int frames = 0; + unsigned char gameID[16] = {}; + bool hasGameID = false; +}; --- a/rts/System/LoadSave/DemoRecorder.h +++ b/rts/System/LoadSave/DemoRecorder.h @@ -45,6 +45,8 @@ bool IsValid() const { return (file != nullptr); } + // WebBAR recovery: the recorded packet stream (after the file header and setup text) so far. + std::pair GetPacketStream() const; void WriteSetupText(const std::string& text); void SaveToDemo(const unsigned char* buf, const unsigned length, const float modGameTime); --- a/rts/System/LoadSave/DemoRecorder.cpp +++ b/rts/System/LoadSave/DemoRecorder.cpp @@ -126,6 +126,15 @@ demoStreams[isServerDemo].append(text.c_str(), length); } +std::pair CDemoRecorder::GetPacketStream() const +{ + const std::string& data = demoStreams[isServerDemo]; + const size_t start = sizeof(DemoFileHeader) + fileHeader.scriptSize; + if (file == nullptr || data.size() <= start) + return {nullptr, 0}; + return {data.data() + start, data.size() - start}; +} + void CDemoRecorder::SaveToDemo(const unsigned char* buf, const unsigned length, const float modGameTime) { DemoStreamChunkHeader chunkHeader; --- a/rts/Game/WebbarBrowser.h +++ b/rts/Game/WebbarBrowser.h @@ -2,6 +2,8 @@ #pragma once #include +#include "Net/WebbarResume.h" +#include "System/LoadSave/DemoRecorder.h" #include #include #include @@ -88,7 +90,7 @@ if (!catalogSent) { catalogSent = true; if (std::getenv("WEBBAR_NETWORK") != nullptr) { static WebbarNetworkResult result; } else { static WebbarStatsResult stats; } - MAIN_THREAD_EM_ASM({ if (Module['webbarProtocol']) Module['webbarProtocol'](9, 65535 /* restore | shields | cloak | resurrection | native work effects | area resurrection | manual launch | capture | smart trajectory | energy conversion | geothermal | selectability | build restrictions | queue edit | game speed | engine memory */); }); + MAIN_THREAD_EM_ASM({ if (Module['webbarProtocol']) Module['webbarProtocol'](9, 131071 /* bit 16: match recovery stream export + resume | restore | shields | cloak | resurrection | native work effects | area resurrection | manual launch | capture | smart trajectory | energy conversion | geothermal | selectability | build restrictions | queue edit | game speed | engine memory */); }); PublishWebbarEngineMemory(); MAIN_THREAD_EM_ASM({ if (Module['webbarIdentity']) Module['webbarIdentity']($0, $1, $2, $3); @@ -205,7 +207,10 @@ uint32_t units[64]; }; static_assert(sizeof(BrowserOrder) == 288); - for (int batch = 0; batch < 8; ++batch) { + // A resumed match takes its input from the recording until it catches up. + const int resumeTarget = webbarResumeTarget.load(); + const bool catchingUp = resumeTarget >= 0 && gs->frameNum < resumeTarget; + for (int batch = 0; !catchingUp && batch < 8; ++batch) { BrowserOrder order = {}; const int available = MAIN_THREAD_EM_ASM_INT({ return Module['webbarPollCommand'] ? Module['webbarPollCommand']($0) : 0; @@ -490,11 +495,72 @@ MAIN_THREAD_EM_ASM({ if (Module['webbarPoses']) Module['webbarPoses'](HEAPU8.slice($0, $0 + $1).buffer); }, packet.data(), packet.size()); +} + +// Match recovery (F138): export the client packet stream incrementally so a later page can +// replay it; the browser journals it in IndexedDB. Game thread only (the same thread that +// appends to the stream), so the buffer cannot change during the synchronous host call. +static void PublishWebbarDemoStream() +{ + static size_t exported = 0; + CDemoRecorder* recorder = clientNet != nullptr ? clientNet->GetDemoRecorder() : nullptr; + if (recorder == nullptr || !recorder->IsValid()) + return; + const auto [data, size] = recorder->GetPacketStream(); + if (data == nullptr || size <= exported) + return; + MAIN_THREAD_EM_ASM({ if (Module['webbarDemo']) Module['webbarDemo']($0, $1, $2, $3); }, data + exported, size - exported, exported, gs->frameNum); + exported = size; +} + +// Returns true while a resumed match is still replaying its recording: presentation is then +// published only every 300 frames (the browser verifies its sync checkpoints there). +static bool WebbarResumeCatchingUp() +{ + static bool done = false, failed = false; + const int target = webbarResumeTarget.load(); + if (target == -2 && !failed) { + failed = true; + MAIN_THREAD_EM_ASM({ if (Module['webbarResume']) Module['webbarResume'](3, $0, 0); }, gs->frameNum); + } + if (target < 0 || done) + return false; + if (gs->frameNum >= target) { + done = true; + MAIN_THREAD_EM_ASM({ if (Module['webbarResume']) Module['webbarResume'](2, $0, $1); }, gs->frameNum, target); + return false; + } + if (gs->frameNum % 300 == 0) + MAIN_THREAD_EM_ASM({ if (Module['webbarResume']) Module['webbarResume'](1, $0, $1); }, gs->frameNum, target); + return true; } +// While a resumed match replays its recording, presentation is published only every 300 frames. +// Events queued on replayed frames are stale by then, so drop them (the browser would reject the +// first packet after a skipped stretch). One-time definitions and full-state channels are kept. +static void DiscardWebbarPresentationQueues() +{ + WebbarNano::pending.clear(); WebbarNano::dropped = 0; + webbarWorkAudio::pending.clear(); webbarWorkAudio::dropped = 0; + webbarFireAudio::pending.clear(); webbarFireAudio::dropped = 0; + webbarCeg::pending.clear(); webbarCeg::dropped = 0; + webbarCloakEvents::pending.clear(); webbarCloakEvents::dropped = 0; + WebbarShieldHits::pending.clear(); WebbarShieldHits::dropped = 0; + WebbarEffects& effects = GetWebbarEffects(); + effects.records.clear(); effects.webbarDamage.clear(); effects.musicDamage = 0.0f; effects.dropped = 0; effects.cegDropped = 0; +} + // Read-only observer, after LEAVE_SYNCED_CODE. This never drives game state. static void PublishWebbarFrame() { + static const bool recoveryStream = std::getenv("WEBBAR_PLAYABLE") != nullptr && std::getenv("WEBBAR_NETWORK") == nullptr && std::getenv("WEBBAR_BROWSER") != nullptr; + if (recoveryStream && gs->frameNum % 150 == 0) + PublishWebbarDemoStream(); + if (recoveryStream && WebbarResumeCatchingUp()) { + DiscardWebbarPresentationQueues(); + if (gs->frameNum % 300 != 0) + return; + } static const char* output = std::getenv("WEBBAR_SNAPSHOTS"); static const bool browser = std::getenv("WEBBAR_BROWSER") != nullptr; if (browser && std::getenv("WEBBAR_PLAYABLE") != nullptr && gs->frameNum >= 0) GetWebbarEffects().Publish(); --- a/rts/Game/WebbarNano.h +++ b/rts/Game/WebbarNano.h @@ -2,6 +2,7 @@ #pragma once #include +#include #include #include #include @@ -36,7 +37,7 @@ if (pending.size() >= 1024) { ++dropped; return; } const float3 color = source->unitDef->nanoColor; pending.push_back({++nextId, uint32_t(gs->frameNum), uint32_t(source->id), uint32_t(unit ? unit->id : feature ? feature->id : 0), kind, uint32_t(inverse) | (feature ? 2u : 0u), - {start.x, start.y, start.z}, {end.x, end.y, end.z}, radius, {color.x, color.y, color.z}}); + {start.x, start.y, start.z}, {end.x, end.y, end.z}, std::fabs(radius), {color.x, color.y, color.z}}); } inline void Publish() --- a/rts/Sim/Units/Scripts/UnitScript.cpp +++ b/rts/Sim/Units/Scripts/UnitScript.cpp @@ -215,6 +215,9 @@ // clear doneAnims here to preserve them for DumpState doneAnims.clear(); +#ifdef WEBBAR_SYNC_TRACE + webbarTraceAnims.clear(); +#endif for (auto& ai : anims) { LocalModelPiece& lmp = *pieces[ai.piece]; @@ -226,6 +229,9 @@ // checksum all anims (live + done) checksum = spring::LiteHash(ai, checksum); +#ifdef WEBBAR_SYNC_TRACE + { std::array raw; memcpy(raw.data(), &ai, sizeof(AnimInfo)); webbarTraceAnims.push_back(raw); } +#endif } spring::VectorEraseIfAll(anims, [](const auto& ai) { return ai.done; }); --- a/rts/Sim/Units/Scripts/UnitScript.h +++ b/rts/Sim/Units/Scripts/UnitScript.h @@ -117,6 +117,9 @@ const CUnit* GetUnit() const { return unit; } auto GetAnimArrayChecksum() const { return checksum; } +#ifdef WEBBAR_SYNC_TRACE + std::vector> webbarTraceAnims; +#endif void TickAllAnims(int tickRate); bool TickAnimFinished(); // note: must copy-and-set here (LMP dirty flag, etc) --- a/rts/Sim/Units/Scripts/UnitScriptEngine.cpp +++ b/rts/Sim/Units/Scripts/UnitScriptEngine.cpp @@ -151,6 +151,9 @@ currentScript = animating[i]; // deal with synced checksum here, before animating is possibly popped below cs = spring::hash_combine(currentScript->GetAnimArrayChecksum(), cs); +#ifdef WEBBAR_SYNC_TRACE + { const uint32_t scs = currentScript->GetAnimArrayChecksum(); CSyncChecker::TraceBytes("S", currentScript->GetUnit() ? currentScript->GetUnit()->id : -1, &scs, sizeof(scs)); for (const auto& raw : currentScript->webbarTraceAnims) CSyncChecker::TraceBytes("A", currentScript->GetUnit() ? currentScript->GetUnit()->id : -1, raw.data(), raw.size()); } +#endif if (!currentScript->TickAnimFinished()) { animating[i] = animating.back(); --- a/rts/System/Sync/SyncChecker.cpp +++ b/rts/System/Sync/SyncChecker.cpp @@ -14,6 +14,31 @@ unsigned CSyncChecker::g_prevChecksum; int CSyncChecker::inSyncedCode; +#ifdef WEBBAR_SYNC_TRACE +#include +#include +#include "Sim/Misc/GlobalSynced.h" +// WebBAR sync trace (diagnostic only): WEBBAR_SYNC_TRACE=from:to logs every checksum input of those frames. +const char* CSyncChecker::traceMsg = ""; +static FILE* webbarTraceFile = nullptr; static int webbarTraceFrom = -1, webbarTraceTo = -1, webbarTraceFrame = -2; static bool webbarTraceInit = false; static unsigned webbarTraceIdx = 0; +void* CSyncChecker::TraceFile() { + if (!webbarTraceInit) { webbarTraceInit = true; const char* r = getenv("WEBBAR_SYNC_TRACE"); if (r && sscanf(r, "%d:%d", &webbarTraceFrom, &webbarTraceTo) == 2) { const char* f = getenv("WEBBAR_SYNC_TRACE_FILE"); webbarTraceFile = fopen(f ? f : "sync-trace.txt", "w"); } } + if (webbarTraceFile == nullptr || gs == nullptr) return nullptr; + const int fr = gs->frameNum; if (fr < webbarTraceFrom || fr > webbarTraceTo) { if (fr > webbarTraceTo) fflush(webbarTraceFile); return nullptr; } + if (fr != webbarTraceFrame) { webbarTraceFrame = fr; webbarTraceIdx = 0; fprintf(webbarTraceFile, "F %d\n", fr); fflush(webbarTraceFile); } + return webbarTraceFile; +} +void CSyncChecker::TraceBytes(const char* tag, int id, const void* p, unsigned size) { + FILE* f = static_cast(TraceFile()); if (f == nullptr) return; + fprintf(f, "%s %d %u ", tag, id, size); for (unsigned i = 0; i < size; i++) fprintf(f, "%02x", static_cast(p)[i]); fprintf(f, "\n"); +} +static void WebbarTraceSync(const void* p, unsigned size, unsigned cs) { + FILE* f = static_cast(CSyncChecker::TraceFile()); if (f == nullptr) return; + fprintf(f, "%u %s %u ", webbarTraceIdx++, CSyncChecker::traceMsg ? CSyncChecker::traceMsg : "?", size); for (unsigned i = 0; i < size; i++) fprintf(f, "%02x", static_cast(p)[i]); fprintf(f, " %08x\n", cs); + CSyncChecker::traceMsg = ""; +} +#endif + void CSyncChecker::NewFrame() { g_checksum = 0xfade1eaf; @@ -34,6 +59,9 @@ debugSyncCheckThreading(); #endif g_checksum = spring::hash_combine(val, g_checksum); +#ifdef WEBBAR_SYNC_TRACE + WebbarTraceSync(&val, sizeof(val), g_checksum); +#endif //LOG("[Sync::Checker] chksum=%u\n", g_checksum); #ifdef SYNC_HISTORY @@ -50,6 +78,9 @@ // most common cases first, make it easy for compiler to optimize for it // simple xor is not enough to detect multiple zeroes, e.g. g_checksum = spring::LiteHash(p, size, g_checksum); +#ifdef WEBBAR_SYNC_TRACE + WebbarTraceSync(p, size, g_checksum); +#endif //LOG("[Sync::Checker] chksum=%u\n", g_checksum); #ifdef SYNC_HISTORY --- a/rts/System/Sync/SyncChecker.h +++ b/rts/System/Sync/SyncChecker.h @@ -37,6 +37,11 @@ static void debugSyncCheckThreading(); static void Sync(uint32_t val); static void Sync(const void* p, unsigned size); + #ifdef WEBBAR_SYNC_TRACE + static const char* traceMsg; + static void* TraceFile(); + static void TraceBytes(const char* tag, int id, const void* p, unsigned size); + #endif #ifdef SYNC_HISTORY static std::tuple GetFrameHistory(unsigned rewindFrames); static std::pair GetHistory() { return std::make_pair(nextHistoryIndex, logs.data()); }; --- a/rts/System/Sync/SyncedPrimitiveBase.h +++ b/rts/System/Sync/SyncedPrimitiveBase.h @@ -45,6 +45,9 @@ AssertDebugger(p, size, msg); #ifdef SYNCCHECK assert(CSyncChecker::InSyncedCode()); + #ifdef WEBBAR_SYNC_TRACE + CSyncChecker::traceMsg = msg; + #endif CSyncChecker::Sync(p, size); #ifdef TRACE_SYNC unsigned int crc = CSyncChecker::GetChecksum(); @@ -57,6 +60,9 @@ AssertDebugger(val, msg); #ifdef SYNCCHECK assert(CSyncChecker::InSyncedCode()); + #ifdef WEBBAR_SYNC_TRACE + CSyncChecker::traceMsg = msg; + #endif CSyncChecker::Sync(val); #ifdef TRACE_SYNC unsigned int crc = CSyncChecker::GetChecksum(); --- a/rts/builds/headless/CMakeLists.txt +++ b/rts/builds/headless/CMakeLists.txt @@ -116,7 +116,7 @@ -sUSE_LIBPNG=1 -sUSE_LIBJPEG=1 -sUSE_ZLIB=1 -sUSE_FREETYPE=1 -sPROXY_TO_PTHREAD=1 -sPTHREAD_POOL_SIZE=8 -sALLOW_MEMORY_GROWTH=1 -sINITIAL_MEMORY=268435456 -sMAXIMUM_MEMORY=4294967296 - -sSTACK_SIZE=8388608 -sDEFAULT_PTHREAD_STACK_SIZE=8388608 + -sSTACK_SIZE=8388608 -sDEFAULT_PTHREAD_STACK_SIZE=2097152 -sEXIT_RUNTIME=1 -sENVIRONMENT=web,worker -sFORCE_FILESYSTEM=1 -sEXPORTED_RUNTIME_METHODS=FS,ENV,callMain) set_target_properties(engine-browser PROPERTIES OUTPUT_NAME "spring-browser" SUFFIX ".js" ENABLE_EXPORTS TRUE)