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/logging.cpp | 264 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 264 insertions(+) create mode 100644 src/zenutil/logging/logging.cpp (limited to 'src/zenutil/logging/logging.cpp') diff --git a/src/zenutil/logging/logging.cpp b/src/zenutil/logging/logging.cpp new file mode 100644 index 000000000..ea2448a42 --- /dev/null +++ b/src/zenutil/logging/logging.cpp @@ -0,0 +1,264 @@ +// Copyright Epic Games, Inc. All Rights Reserved. + +#include "zenutil/logging.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace zen { + +static bool g_IsLoggingInitialized; +logging::SinkPtr g_FileSink; + +logging::SinkPtr +GetFileSink() +{ + return g_FileSink; +} + +void +InitializeLogging(const LoggingOptions& LogOptions) +{ + BeginInitializeLogging(LogOptions); + FinishInitializeLogging(LogOptions); +} + +static std::terminate_handler OldTerminateHandler = nullptr; + +void +BeginInitializeLogging(const LoggingOptions& LogOptions) +{ + ZEN_MEMSCOPE(ELLMTag::Logging); + + zen::logging::InitializeLogging(); + zen::logging::EnableVTMode(); + + // Sinks + + logging::SinkPtr FileSink; + + if (!LogOptions.AbsLogFile.empty()) + { + if (LogOptions.AbsLogFile.has_parent_path()) + { + zen::CreateDirectories(LogOptions.AbsLogFile.parent_path()); + } + + FileSink = logging::SinkPtr(new zen::logging::RotatingFileSink(LogOptions.AbsLogFile, + /* max size */ 128 * 1024 * 1024, + /* max files */ 16, + /* rotate on open */ true)); + if (LogOptions.AbsLogFile.extension() == ".json") + { + FileSink->SetFormatter(std::make_unique(LogOptions.LogId)); + } + else + { + FileSink->SetFormatter(std::make_unique(LogOptions.LogId)); // this will have a date prefix + } + } + + OldTerminateHandler = std::set_terminate([]() { + try + { + constexpr int SkipFrameCount = 4; + constexpr int FrameCount = 8; + uint8_t CallstackBuffer[CallstackRawMemorySize(SkipFrameCount, FrameCount)]; + CallstackFrames* Callstack = GetCallstackRaw(&CallstackBuffer[0], SkipFrameCount, FrameCount); + + fmt::basic_memory_buffer Message; + auto Appender = fmt::appender(Message); + fmt::format_to(Appender, "Program exited abnormally via std::terminate()"); + + if (Callstack->FrameCount > 0) + { + fmt::format_to(Appender, "\n"); + + CallstackToStringRaw(Callstack, &Message, [](void* UserData, uint32_t FrameIndex, const char* FrameText) { + ZEN_UNUSED(FrameIndex); + fmt::basic_memory_buffer* Message = (fmt::basic_memory_buffer*)UserData; + auto Appender = fmt::appender(*Message); + fmt::format_to(Appender, " {}\n", FrameText); + }); + } + Message.push_back('\0'); + + // We use direct ZEN_LOG here instead of ZEN_ERROR as we don't care about *this* code location in the log + ZEN_LOG(Log(), zen::logging::Critical, "{}", Message.data()); + zen::logging::FlushLogging(); + } + catch (const std::exception&) + { + // Ignore any exceptions in terminate callback + } + if (OldTerminateHandler) + { + OldTerminateHandler(); + } + }); + + // Default + + LoggerRef DefaultLogger = zen::logging::Default(); + + // Collect sinks into a local vector first so we can optionally wrap them + std::vector Sinks; + + if (LogOptions.NoConsoleOutput) + { + zen::logging::SuppressConsoleLog(); + } + else + { + logging::SinkPtr ConsoleSink(new logging::AnsiColorStdoutSink()); + if (LogOptions.QuietConsole) + { + ConsoleSink->SetLevel(logging::Warn); + } + Sinks.push_back(ConsoleSink); + } + + if (FileSink) + { + Sinks.push_back(FileSink); + } + +#if ZEN_PLATFORM_WINDOWS + if (zen::IsDebuggerPresent() && LogOptions.IsDebug) + { + logging::SinkPtr DebugSink(new logging::MsvcSink()); + DebugSink->SetLevel(logging::Debug); + Sinks.push_back(DebugSink); + } +#endif + + bool IsAsync = LogOptions.AllowAsync && !LogOptions.IsDebug && !LogOptions.IsTest; + + if (IsAsync) + { + std::vector AsyncSinks; + AsyncSinks.emplace_back(new logging::AsyncSink(std::move(Sinks))); + DefaultLogger->SetSinks(std::move(AsyncSinks)); + } + else + { + DefaultLogger->SetSinks(std::move(Sinks)); + } + + static struct : logging::ErrorHandler + { + void HandleError(const std::string_view& ErrorMsg) override + { + if (ErrorMsg == std::bad_alloc().what()) + { + return; + } + static constinit logging::LogPoint ErrorPoint{{}, logging::Err, "{}"}; + if (auto ErrLogger = zen::logging::ErrorLog()) + { + try + { + ErrLogger->Log(ErrorPoint, fmt::make_format_args(ErrorMsg)); + } + catch (const std::exception&) + { + } + } + try + { + Log()->Log(ErrorPoint, fmt::make_format_args(ErrorMsg)); + } + catch (const std::exception&) + { + } + } + } s_ErrorHandler; + logging::Registry::Instance().SetErrorHandler(&s_ErrorHandler); + + g_FileSink = std::move(FileSink); +} + +void +FinishInitializeLogging(const LoggingOptions& LogOptions) +{ + ZEN_MEMSCOPE(ELLMTag::Logging); + + logging::LogLevel LogLevel = logging::Info; + + if (LogOptions.IsDebug) + { + LogLevel = logging::Debug; + } + + if (LogOptions.IsTest || LogOptions.IsVerbose) + { + LogLevel = logging::Trace; + } + + // Configure all registered loggers according to settings + + logging::RefreshLogLevels(LogLevel); + logging::Registry::Instance().FlushOn(logging::Err); + logging::Registry::Instance().FlushEvery(std::chrono::seconds{2}); + logging::Registry::Instance().SetFormatter(std::make_unique( + LogOptions.LogId, + std::chrono::system_clock::now() - std::chrono::milliseconds(GetTimeSinceProcessStart()))); // default to duration prefix + + // If the console logger was initialized before, the above will change the output format + // so we need to reset it + + logging::ResetConsoleLog(); + + if (g_FileSink) + { + if (LogOptions.AbsLogFile.extension() == ".json") + { + g_FileSink->SetFormatter(std::make_unique(LogOptions.LogId)); + } + else + { + g_FileSink->SetFormatter(std::make_unique(LogOptions.LogId)); // this will have a date prefix + } + + const std::string StartLogTime = zen::DateTime::Now().ToIso8601(); + + logging::Registry::Instance().ApplyAll([&](auto Logger) { + static constinit logging::LogPoint LogStartPoint{{}, logging::Info, "log starting at {}"}; + Logger->Log(LogStartPoint, fmt::make_format_args(StartLogTime)); + }); + } + + g_IsLoggingInitialized = true; +} + +void +ShutdownLogging() +{ + if (g_IsLoggingInitialized && g_FileSink) + { + auto DefaultLogger = zen::logging::Default(); + ZEN_LOG_INFO(DefaultLogger, "log ending at {}", zen::DateTime::Now().ToIso8601()); + } + + zen::logging::ShutdownLogging(); + + g_FileSink = nullptr; +} + +} // namespace zen -- cgit v1.2.3