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/rotatingfilesink.cpp | 249 +++++++++++++++++++++++++++++++ 1 file changed, 249 insertions(+) create mode 100644 src/zenutil/logging/rotatingfilesink.cpp (limited to 'src/zenutil/logging/rotatingfilesink.cpp') diff --git a/src/zenutil/logging/rotatingfilesink.cpp b/src/zenutil/logging/rotatingfilesink.cpp new file mode 100644 index 000000000..23cf60d16 --- /dev/null +++ b/src/zenutil/logging/rotatingfilesink.cpp @@ -0,0 +1,249 @@ +// Copyright Epic Games, Inc. All Rights Reserved. + +#include + +#include +#include +#include +#include +#include +#include + +#include + +namespace zen::logging { + +struct RotatingFileSink::Impl +{ + Impl(const std::filesystem::path& BaseFilename, std::size_t MaxSize, std::size_t MaxFiles, bool RotateOnOpen) + : m_BaseFilename(BaseFilename) + , m_MaxSize(MaxSize) + , m_MaxFiles(MaxFiles) + , m_Formatter(std::make_unique()) + { + ZEN_MEMSCOPE(ELLMTag::Logging); + + std::error_code Ec; + if (RotateOnOpen) + { + RwLock::ExclusiveLockScope RotateLock(m_Lock); + Rotate(RotateLock, Ec); + } + else + { + m_CurrentFile.Open(m_BaseFilename, BasicFile::Mode::kWrite, Ec); + if (!Ec) + { + m_CurrentSize = m_CurrentFile.FileSize(Ec); + } + if (!Ec) + { + if (m_CurrentSize > m_MaxSize) + { + RwLock::ExclusiveLockScope RotateLock(m_Lock); + Rotate(RotateLock, Ec); + } + } + } + + if (Ec) + { + throw std::system_error(Ec, fmt::format("Failed to open log file '{}'", m_BaseFilename.string())); + } + } + + ~Impl() + { + try + { + RwLock::ExclusiveLockScope RotateLock(m_Lock); + m_CurrentFile.Close(); + } + catch (const std::exception&) + { + } + } + + void Rotate(RwLock::ExclusiveLockScope&, std::error_code& OutEc) + { + ZEN_MEMSCOPE(ELLMTag::Logging); + + m_CurrentFile.Close(); + + OutEc = RotateFiles(m_BaseFilename, m_MaxFiles); + if (OutEc) + { + return; + } + + m_CurrentFile.Open(m_BaseFilename, BasicFile::Mode::kWrite, OutEc); + if (OutEc) + { + return; + } + + m_CurrentSize = m_CurrentFile.FileSize(OutEc); + if (OutEc) + { + // FileSize failed but we have an open file — reset to 0 + // so we can at least attempt writes from the start + m_CurrentSize = 0; + OutEc.clear(); + } + } + + bool TrySinkIt(const LogMessage& Msg, MemoryBuffer& OutFormatted) + { + ZEN_MEMSCOPE(ELLMTag::Logging); + + RwLock::SharedLockScope Lock(m_Lock); + if (!m_CurrentFile.IsOpen()) + { + return false; + } + m_Formatter->Format(Msg, OutFormatted); + helpers::StripAnsiSgrSequences(OutFormatted); + size_t AddSize = OutFormatted.size(); + size_t WritePos = m_CurrentSize.fetch_add(AddSize); + if (WritePos + AddSize > m_MaxSize) + { + return false; + } + std::error_code Ec; + m_CurrentFile.Write(OutFormatted.data(), OutFormatted.size(), WritePos, Ec); + if (Ec) + { + return false; + } + m_NeedFlush = true; + return true; + } + + bool TrySinkIt(const MemoryBuffer& Formatted) + { + ZEN_MEMSCOPE(ELLMTag::Logging); + + RwLock::SharedLockScope Lock(m_Lock); + if (!m_CurrentFile.IsOpen()) + { + return false; + } + size_t AddSize = Formatted.size(); + size_t WritePos = m_CurrentSize.fetch_add(AddSize); + if (WritePos + AddSize > m_MaxSize) + { + return false; + } + + std::error_code Ec; + m_CurrentFile.Write(Formatted.data(), Formatted.size(), WritePos, Ec); + if (Ec) + { + return false; + } + m_NeedFlush = true; + return true; + } + + RwLock m_Lock; + const std::filesystem::path m_BaseFilename; + const std::size_t m_MaxSize; + const std::size_t m_MaxFiles; + std::unique_ptr m_Formatter; + std::atomic_size_t m_CurrentSize; + BasicFile m_CurrentFile; + std::atomic m_NeedFlush = false; +}; + +RotatingFileSink::RotatingFileSink(const std::filesystem::path& BaseFilename, std::size_t MaxSize, std::size_t MaxFiles, bool RotateOnOpen) +: m_Impl(std::make_unique(BaseFilename, MaxSize, MaxFiles, RotateOnOpen)) +{ +} + +RotatingFileSink::~RotatingFileSink() = default; + +void +RotatingFileSink::Log(const LogMessage& Msg) +{ + ZEN_MEMSCOPE(ELLMTag::Logging); + + try + { + MemoryBuffer Formatted; + if (m_Impl->TrySinkIt(Msg, Formatted)) + { + return; + } + + // This intentionally has no limit on the number of retries, see + // comment above. + for (;;) + { + { + RwLock::ExclusiveLockScope RotateLock(m_Impl->m_Lock); + // Only rotate if no-one else has rotated before us + if (m_Impl->m_CurrentSize > m_Impl->m_MaxSize || !m_Impl->m_CurrentFile.IsOpen()) + { + std::error_code Ec; + m_Impl->Rotate(RotateLock, Ec); + if (Ec) + { + return; + } + } + } + if (m_Impl->TrySinkIt(Formatted)) + { + return; + } + } + } + catch (const std::exception&) + { + // Silently eat errors + } +} + +void +RotatingFileSink::Flush() +{ + if (!m_Impl->m_NeedFlush) + { + return; + } + + ZEN_MEMSCOPE(ELLMTag::Logging); + + try + { + RwLock::SharedLockScope Lock(m_Impl->m_Lock); + if (m_Impl->m_CurrentFile.IsOpen()) + { + m_Impl->m_CurrentFile.Flush(); + } + } + catch (const std::exception&) + { + // Silently eat errors + } + + m_Impl->m_NeedFlush = false; +} + +void +RotatingFileSink::SetFormatter(std::unique_ptr InFormatter) +{ + ZEN_MEMSCOPE(ELLMTag::Logging); + + try + { + RwLock::ExclusiveLockScope _(m_Impl->m_Lock); + m_Impl->m_Formatter = std::move(InFormatter); + } + catch (const std::exception&) + { + // Silently eat errors + } +} + +} // namespace zen::logging -- cgit v1.2.3