From eba410c4168e23d7908827eb34b7cf0c58a5dc48 Mon Sep 17 00:00:00 2001 From: Stefan Boberg Date: Wed, 18 Mar 2026 11:19:10 +0100 Subject: Compute batching (#849) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Compute Batch Submission - Consolidate duplicated action submission logic in `httpcomputeservice` into a single `HandleSubmitAction` supporting both single-action and batch (actions array) payloads - Group actions by queue in `RemoteHttpRunner` and submit as batches with configurable chunk size, falling back to individual submission on failure - Extract shared helpers: `MakeErrorResult`, `ValidateQueueForEnqueue`, `ActivateActionInQueue`, `RemoveActionFromActiveMaps` ### Retracted Action State - Add `Retracted` state to `RunnerAction` for retry-free rescheduling — an explicit request to pull an action back and reschedule it on a different runner without incrementing `RetryCount` - Implement idempotent `RetractAction()` on `RunnerAction` and `ComputeServiceSession` - Add `POST jobs/{lsn}/retract` and `queues/{queueref}/jobs/{lsn}/retract` HTTP endpoints - Add state machine documentation and per-state comments to `RunnerAction` ### Compute Race Fixes - Fix race in `HandleActionUpdates` where actions enqueued between session abandon and scheduler tick were never abandoned, causing `GetActionResult` to return 202 indefinitely - Fix queue `ActiveCount` race where `NotifyQueueActionComplete` was called after releasing `m_ResultsLock`, allowing callers to observe stale counters immediately after `GetActionResult` returned OK ### Logging Optimization and ANSI improvements - Improve `AnsiColorStdoutSink` write efficiency — single write call, dirty-flag flush, `RwLock` instead of `std::mutex` - Move ANSI color emission from sink into formatters via `Formatter::SetColorEnabled()`; remove `ColorRangeStart`/`End` from `LogMessage` - Extract color helpers (`AnsiColorForLevel`, `StripAnsiSgrSequences`) into `helpers.h` - Strip upstream ANSI SGR escapes in non-color output mode. This enables colour in log messages without polluting log files with ANSI control sequences - Move `RotatingFileSink`, `JsonFormatter`, and `FullFormatter` from header-only to pimpl with `.cpp` files ### CLI / Exec Refactoring - Extract `ExecSessionRunner` class from ~920-line `ExecUsingSession` into focused methods and a `ExecSessionConfig` struct - Replace monolithic `ExecCommand` with subcommand-based architecture (`http`, `inproc`, `beacon`, `dump`, `buildlog`) - Allow parent options to appear after subcommand name by parsing subcommand args permissively and forwarding unmatched tokens to the parent parser ### Testing Improvements - Fix `--test-suite` filter being ignored due to accumulation with default wildcard filter - Add test suite banners to test listener output - Made `function.session.abandon_pending` test more robust ### Startup / Reliability Fixes - Fix silent exit when a second zenserver instance detects a port conflict — use `ZEN_CONSOLE_*` for log calls that precede `InitializeLogging()` - Fix two potential SIGSEGV paths during early startup: guard `sentry_options_new()` returning nullptr, and throw on `ZenServerState::Register()` returning nullptr instead of dereferencing - Fail on unrecognized zenserver `--mode` instead of silently defaulting to store ### Other - Show host details (hostname, platform, CPU count, memory) when discovering new compute workers - Move frontend `html.zip` from source tree into build directory - Add format specifications for Compact Binary and Compressed Buffer wire formats - Add `WriteCompactBinaryObject` to zencore - Extended `ConsoleTui` with additional functionality - Add `--vscode` option to `xmake sln` for clangd / `compile_commands.json` support - Disable compute/horde/nomad in release builds (not yet production-ready) - Disable unintended `ASIO_HAS_IO_URING` enablement - Fix crashpad patch missing leading whitespace - Clean up code triggering gcc false positives --- src/zenutil/logging/jsonformatter.cpp | 198 ++++++++++++++++++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 src/zenutil/logging/jsonformatter.cpp (limited to 'src/zenutil/logging/jsonformatter.cpp') diff --git a/src/zenutil/logging/jsonformatter.cpp b/src/zenutil/logging/jsonformatter.cpp new file mode 100644 index 000000000..673a03c94 --- /dev/null +++ b/src/zenutil/logging/jsonformatter.cpp @@ -0,0 +1,198 @@ +// Copyright Epic Games, Inc. All Rights Reserved. + +#include + +#include +#include +#include +#include + +#include +#include +#include + +namespace zen::logging { + +using namespace std::literals; + +static void +WriteEscapedString(MemoryBuffer& Dest, std::string_view Text) +{ + // Strip ANSI SGR sequences before escaping so they don't appear in JSON output + static const auto IsEscapeStart = [](char C) { return C == '\033'; }; + + const char* RangeStart = Text.data(); + const char* End = Text.data() + Text.size(); + + static const std::unordered_map SpecialCharacterMap{ + {'\b', "\\b"sv}, + {'\f', "\\f"sv}, + {'\n', "\\n"sv}, + {'\r', "\\r"sv}, + {'\t', "\\t"sv}, + {'"', "\\\""sv}, + {'\\', "\\\\"sv}, + }; + + for (const char* It = RangeStart; It != End; ++It) + { + // Skip ANSI SGR escape sequences (\033[...m) + if (*It == '\033' && (It + 1) < End && *(It + 1) == '[') + { + if (RangeStart != It) + { + Dest.append(RangeStart, It); + } + const char* Seq = It + 2; + while (Seq < End && *Seq != 'm') + { + ++Seq; + } + if (Seq < End) + { + ++Seq; // skip 'm' + } + It = Seq - 1; // -1 because the for loop will ++It + RangeStart = Seq; + continue; + } + + if (auto SpecialIt = SpecialCharacterMap.find(*It); SpecialIt != SpecialCharacterMap.end()) + { + if (RangeStart != It) + { + Dest.append(RangeStart, It); + } + helpers::AppendStringView(SpecialIt->second, Dest); + RangeStart = It + 1; + } + } + if (RangeStart != End) + { + Dest.append(RangeStart, End); + } +} + +struct JsonFormatter::Impl +{ + explicit Impl(std::string_view LogId) : m_LogId(LogId) {} + + std::tm m_CachedTm{0, 0, 0, 0, 0, 0, 0, 0, 0}; + std::chrono::seconds m_LastLogSecs{0}; + MemoryBuffer m_CachedDatetime; + std::string m_LogId; + RwLock m_TimestampLock; +}; + +JsonFormatter::JsonFormatter(std::string_view LogId) : m_Impl(std::make_unique(LogId)) +{ +} + +JsonFormatter::~JsonFormatter() = default; + +std::unique_ptr +JsonFormatter::Clone() const +{ + return std::make_unique(m_Impl->m_LogId); +} + +void +JsonFormatter::Format(const LogMessage& Msg, MemoryBuffer& Dest) +{ + ZEN_MEMSCOPE(ELLMTag::Logging); + + using std::chrono::duration_cast; + using std::chrono::milliseconds; + using std::chrono::seconds; + + auto Secs = duration_cast(Msg.GetTime().time_since_epoch()); + if (Secs != m_Impl->m_LastLogSecs) + { + RwLock::ExclusiveLockScope _(m_Impl->m_TimestampLock); + m_Impl->m_CachedTm = helpers::SafeLocaltime(LogClock::to_time_t(Msg.GetTime())); + m_Impl->m_LastLogSecs = Secs; + + // cache the date/time part for the next second. + m_Impl->m_CachedDatetime.clear(); + + helpers::AppendInt(m_Impl->m_CachedTm.tm_year + 1900, m_Impl->m_CachedDatetime); + m_Impl->m_CachedDatetime.push_back('-'); + + helpers::Pad2(m_Impl->m_CachedTm.tm_mon + 1, m_Impl->m_CachedDatetime); + m_Impl->m_CachedDatetime.push_back('-'); + + helpers::Pad2(m_Impl->m_CachedTm.tm_mday, m_Impl->m_CachedDatetime); + m_Impl->m_CachedDatetime.push_back(' '); + + helpers::Pad2(m_Impl->m_CachedTm.tm_hour, m_Impl->m_CachedDatetime); + m_Impl->m_CachedDatetime.push_back(':'); + + helpers::Pad2(m_Impl->m_CachedTm.tm_min, m_Impl->m_CachedDatetime); + m_Impl->m_CachedDatetime.push_back(':'); + + helpers::Pad2(m_Impl->m_CachedTm.tm_sec, m_Impl->m_CachedDatetime); + + m_Impl->m_CachedDatetime.push_back('.'); + } + helpers::AppendStringView("{"sv, Dest); + helpers::AppendStringView("\"time\": \""sv, Dest); + { + RwLock::SharedLockScope _(m_Impl->m_TimestampLock); + Dest.append(m_Impl->m_CachedDatetime.begin(), m_Impl->m_CachedDatetime.end()); + } + auto Millis = helpers::TimeFraction(Msg.GetTime()); + helpers::Pad3(static_cast(Millis.count()), Dest); + helpers::AppendStringView("\", "sv, Dest); + + helpers::AppendStringView("\"status\": \""sv, Dest); + helpers::AppendStringView(helpers::LevelToShortString(Msg.GetLevel()), Dest); + helpers::AppendStringView("\", "sv, Dest); + + helpers::AppendStringView("\"source\": \""sv, Dest); + helpers::AppendStringView("zenserver"sv, Dest); + helpers::AppendStringView("\", "sv, Dest); + + helpers::AppendStringView("\"service\": \""sv, Dest); + helpers::AppendStringView("zencache"sv, Dest); + helpers::AppendStringView("\", "sv, Dest); + + if (!m_Impl->m_LogId.empty()) + { + helpers::AppendStringView("\"id\": \""sv, Dest); + helpers::AppendStringView(m_Impl->m_LogId, Dest); + helpers::AppendStringView("\", "sv, Dest); + } + + if (Msg.GetLoggerName().size() > 0) + { + helpers::AppendStringView("\"logger.name\": \""sv, Dest); + helpers::AppendStringView(Msg.GetLoggerName(), Dest); + helpers::AppendStringView("\", "sv, Dest); + } + + if (Msg.GetThreadId() != 0) + { + helpers::AppendStringView("\"logger.thread_name\": \""sv, Dest); + helpers::PadUint(Msg.GetThreadId(), 0, Dest); + helpers::AppendStringView("\", "sv, Dest); + } + + if (Msg.GetSource()) + { + helpers::AppendStringView("\"file\": \""sv, Dest); + WriteEscapedString(Dest, helpers::ShortFilename(Msg.GetSource().Filename)); + helpers::AppendStringView("\","sv, Dest); + + helpers::AppendStringView("\"line\": \""sv, Dest); + helpers::AppendInt(Msg.GetSource().Line, Dest); + helpers::AppendStringView("\","sv, Dest); + } + + helpers::AppendStringView("\"message\": \""sv, Dest); + WriteEscapedString(Dest, Msg.GetPayload()); + helpers::AppendStringView("\""sv, Dest); + + helpers::AppendStringView("}\n"sv, Dest); +} + +} // namespace zen::logging -- cgit v1.2.3