diff --git a/daemon/launcher/trace_command_common.cpp b/daemon/launcher/trace_command_common.cpp index 815c960..6ee5e46 100644 --- a/daemon/launcher/trace_command_common.cpp +++ b/daemon/launcher/trace_command_common.cpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include @@ -25,6 +26,7 @@ #include "gpufl/core/json/json.hpp" #include "gpufl/core/logger/file_compressor.hpp" #include "gpufl/core/logger/log_salvage.hpp" +#include "gpufl/core/logger/window_metadata.hpp" #include "gpufl/inject/inject_entry.hpp" #include "gpufl/upload/upload_logs.hpp" @@ -154,10 +156,14 @@ void readLogLines(const fs::path& path, Fn&& fn) { struct SessionLifecycleInfo { bool saw_job_start = false; + bool saw_segment_end = false; + bool saw_run_end = false; bool saw_shutdown = false; int pid = 0; std::string app = "unknown"; int64_t job_start_ts_ns = 0; + std::string run_id; + int segment_index = -1; }; bool isSessionDirectory(const fs::directory_entry& entry) { @@ -175,6 +181,14 @@ struct SyntheticShutdownContext { }; void inspectLifecycleLine(const std::string& line, SessionLifecycleInfo& info) { + if (line.find("\"type\":\"segment_end\"") != std::string::npos) { + info.saw_segment_end = true; + return; + } + if (line.find("\"type\":\"run_end\"") != std::string::npos) { + info.saw_run_end = true; + return; + } if (line.find("\"type\":\"shutdown\"") != std::string::npos) { info.saw_shutdown = true; return; @@ -194,6 +208,8 @@ void inspectLifecycleLine(const std::string& line, SessionLifecycleInfo& info) { info.pid = doc.value("pid", 0); info.app = doc.value("app", "unknown"); info.job_start_ts_ns = doc.value("ts_ns", 0); + info.run_id = doc.value("run_id", ""); + info.segment_index = doc.value("segment_index", -1); } bool decompressGzipToFile(const fs::path& gz_path, const fs::path& out_path) { @@ -236,6 +252,56 @@ std::string syntheticShutdownLine(const std::string& session_id, return os.str(); } +std::vector syntheticCompletionLines( + const std::string& session_id, const SessionLifecycleInfo& info, + const SyntheticShutdownContext& context, + const bool may_synthesize_run_end) { + std::vector lines; + const int64_t ended_ns = nowNs(); + const bool normal_segmented_exit = + context.exit_code == 0 && !context.signaled && + !context.window_stopped && !info.run_id.empty() && + info.segment_index >= 0; + + // A Windows injection atexit may be unable to drain outstanding CUPTI + // writers safely. SegmentRuntime then deliberately omits false finality, + // but the launcher knows whether the process itself exited normally. + // Repair only that normal-exit case. Crashes and forced window stops must + // remain without run_end so backend liveness marks them incomplete. + if (normal_segmented_exit && !info.saw_segment_end) { + std::ostringstream os; + os << "{\"version\":1,\"type\":\"segment_end\"" + << ",\"session_id\":\"" << json::escape(session_id) << "\"" + << ",\"run_id\":\"" << json::escape(info.run_id) << "\"" + << ",\"segment_index\":" << info.segment_index + << ",\"ts_ns\":" << ended_ns + << ",\"actual_end_ns\":" << ended_ns + << ",\"requested_boundary_ns\":null" + << ",\"boundary_delay_ns\":0" + << ",\"end_reason\":\"process_shutdown\"" + << ",\"deferred_by\":null" + << ",\"records_outside_segment_window\":0" + << ",\"synthetic\":true}"; + lines.push_back(os.str()); + } + if (normal_segmented_exit && may_synthesize_run_end && + !info.saw_run_end) { + std::ostringstream os; + os << "{\"version\":1,\"type\":\"run_end\"" + << ",\"session_id\":\"" << json::escape(session_id) << "\"" + << ",\"run_id\":\"" << json::escape(info.run_id) << "\"" + << ",\"final_segment_index\":" << info.segment_index + << ",\"ts_ns\":" << ended_ns + << ",\"ended_ns\":" << ended_ns + << ",\"synthetic\":true}"; + lines.push_back(os.str()); + } + if (!info.saw_shutdown) { + lines.push_back(syntheticShutdownLine(session_id, info, context)); + } + return lines; +} + bool appendLine(const fs::path& path, const std::string& line) { std::ofstream out(path, std::ios::out | std::ios::app); if (!out.is_open()) return false; @@ -288,11 +354,15 @@ bool appendLineToGzipLog(const fs::path& gz_path, const std::string& line) { return true; } -bool appendSyntheticShutdown(const fs::path& session_dir, - const std::string& session_id, - const SessionLifecycleInfo& info, - const SyntheticShutdownContext& context) { - const std::string line = syntheticShutdownLine(session_id, info, context); +bool appendSyntheticCompletion(const fs::path& session_dir, + const std::string& session_id, + const SessionLifecycleInfo& info, + const SyntheticShutdownContext& context, + const bool may_synthesize_run_end) { + const std::vector lines = + syntheticCompletionLines( + session_id, info, context, may_synthesize_run_end); + if (lines.empty()) return false; // Allocate through the shared allocator, which also counts windows still // sitting in `.tmp`. highestSystemWindowIndex() only scans the session // ROOT, so a deferred window in `.tmp` did not reserve its index and the @@ -306,9 +376,16 @@ bool appendSyntheticShutdown(const fs::path& session_dir, const fs::path window_gz = session_dir / (base + ".gz"); std::error_code ec; if (fs::exists(window_gz, ec)) { - return appendLineToGzipLog(window_gz, line); + for (const auto& line : lines) { + if (!appendLineToGzipLog(window_gz, line)) return false; + } + return true; + } + const fs::path raw = session_dir / base; + for (const auto& line : lines) { + if (!appendLine(raw, line)) return false; } - return appendLine(session_dir / base, line); + return true; } bool isSystemLog(const fs::path& path) { @@ -337,7 +414,6 @@ void inspectLifecycleFiles(const std::vector& paths, readLogLines(path, [&](const std::string& line) { inspectLifecycleLine(line, info); }); - if (info.saw_shutdown) break; } } @@ -347,7 +423,13 @@ int ensureTraceCompletionMarkers(const fs::path& output_dir, std::error_code ec; if (output_dir.empty() || !fs::exists(output_dir, ec)) return 0; - int synthesized = 0; + struct SessionCandidate { + fs::path session_dir; + std::string session_id; + SessionLifecycleInfo info; + }; + std::vector candidates; + for (const auto& session_entry : fs::directory_iterator(output_dir, ec)) { if (ec) break; if (!isSessionDirectory(session_entry)) continue; @@ -377,9 +459,45 @@ int ensureTraceCompletionMarkers(const fs::path& output_dir, inspectLifecycleFiles(fallback_logs, info); } - if (info.saw_job_start && !info.saw_shutdown && - info.job_start_ts_ns >= min_job_start_ts_ns && - appendSyntheticShutdown(session_dir, session_id, info, context)) { + if (info.saw_job_start && + info.job_start_ts_ns >= min_job_start_ts_ns) { + candidates.push_back( + SessionCandidate{session_dir, session_id, std::move(info)}); + } + } + + // run_end is one logical-run terminal marker, not one marker per segment. + // Repair it only in the highest observed segment from this launcher + // invocation. Earlier segments still receive a missing segment_end or + // shutdown, but can never truncate final_segment_index when the actual tail + // window is delayed or lost. + std::unordered_map highest_segment_by_run; + for (const auto& candidate : candidates) { + const auto& info = candidate.info; + if (info.run_id.empty() || info.segment_index < 0) continue; + auto [it, inserted] = highest_segment_by_run.emplace( + info.run_id, info.segment_index); + if (!inserted) { + it->second = std::max(it->second, info.segment_index); + } + } + + int synthesized = 0; + for (const auto& candidate : candidates) { + const auto& info = candidate.info; + const auto highest = highest_segment_by_run.find(info.run_id); + const bool may_synthesize_run_end = + highest != highest_segment_by_run.end() && + info.segment_index == highest->second; + const bool needs_normal_finality = + context.exit_code == 0 && !context.signaled && + !context.window_stopped && !info.run_id.empty() && + (!info.saw_segment_end || + (may_synthesize_run_end && !info.saw_run_end)); + if ((!info.saw_shutdown || needs_normal_finality) && + appendSyntheticCompletion( + candidate.session_dir, candidate.session_id, info, context, + may_synthesize_run_end)) { ++synthesized; } } @@ -458,6 +576,7 @@ int repairUncompressedLogs(const fs::path& root) { int repaired = salvaged.salvaged; GzipFileCompressor compressor; + bool staged_indexed_window = false; std::error_code iter_ec; for (fs::recursive_directory_iterator it(root, fs::directory_options::skip_permission_denied, iter_ec), end; !iter_ec && it != end; @@ -465,8 +584,10 @@ int repairUncompressedLogs(const fs::path& root) { std::error_code entry_ec; if (!it->is_regular_file(entry_ec)) continue; const fs::path path = it->path(); - // .tmp contents were handled by the salvage pass above; a dir - // that survived it is still held - leave it alone. + // The first salvage pass handled artifacts already under `.tmp`. + // Indexed raw files found below are moved into `.tmp` deliberately and + // consumed by the second salvage pass after this traversal. Skipping + // `.tmp` here prevents those staged files from being compressed twice. if (path.parent_path().filename() == ".tmp") continue; if (path.extension() != ".log") continue; @@ -480,6 +601,10 @@ int repairUncompressedLogs(const fs::path& root) { } const fs::path gz_path(path.string() + ".gz"); + std::string channel; + std::size_t sequence = 0; + const bool indexed_window = parsePublishedWindowName( + gz_path.filename().string(), channel, sequence); std::error_code exists_ec; // EXISTS is not VALID. A compress that died midway - or a rename that // hit the disk before its data - leaves a truncated or empty `.gz`, @@ -495,6 +620,20 @@ int repairUncompressedLogs(const fs::path& root) { fs::remove(gz_path, rm_ec); } if (fs::exists(gz_path, exists_ec)) { + // A completed indexed gzip is not authoritative until its immutable + // sidecar exists. Write/validate the identity before removing the + // duplicate raw source; otherwise a current agent sees an + // identity-less payload and permanently escalates it. + if (indexed_window && + !ensureWindowMetadata(path.parent_path(), + path.parent_path().filename().string(), + channel, sequence, gz_path)) { + std::fprintf(stderr, + "[gpufl] warning: could not write metadata for " + "%s; preserving its raw source\n", + gz_path.string().c_str()); + continue; + } std::error_code remove_ec; if (!removeWithRetry(path, remove_ec)) { // Held without delete sharing - truncate so the data @@ -512,12 +651,54 @@ int repairUncompressedLogs(const fs::path& root) { continue; } + if (indexed_window) { + // Do not compress directly into the published name. The agent is + // polling this directory concurrently, and payload-before-metadata + // would briefly downgrade a current-client window to the legacy + // path. Move the raw source back under `.tmp` and let the shared + // salvage transaction do: + // + // compress -> immutable metadata -> payload publish. + // + // This also makes launcher-created synthetic shutdown windows use + // exactly the same crash/collision handling as normal windows. + const fs::path tmp_dir = path.parent_path() / ".tmp"; + std::error_code mkdir_ec; + fs::create_directories(tmp_dir, mkdir_ec); + if (mkdir_ec) { + std::fprintf(stderr, + "[gpufl] warning: could not create %s to stage " + "repaired window %s: %s\n", + tmp_dir.string().c_str(), path.string().c_str(), + mkdir_ec.message().c_str()); + continue; + } + const fs::path staged = tmp_dir / path.filename(); + std::error_code move_ec; + const auto moved = moveFileNoReplace(path, staged, move_ec); + if (moved != MoveFileNoReplaceResult::Moved) { + std::fprintf(stderr, + "[gpufl] warning: could not stage repaired " + "window %s under .tmp: %s\n", + path.string().c_str(), + move_ec.message().c_str()); + continue; + } + staged_indexed_window = true; + continue; + } + + // Legacy, unindexed logs predate transport-window identity. Preserve + // their historical repair behavior; current segmented windows always + // take the indexed transaction above. if (compressor.compress(path.string())) { - // compress() removes (or truncates) the original itself and - // logs when it can't - nothing left to do here. ++repaired; } } + if (staged_indexed_window) { + const auto staged = salvageSessionTempDirs(root); + repaired += staged.salvaged; + } return repaired; } @@ -836,6 +1017,15 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) { } } + // Completion repair must happen while the upload agent is still alive. + // In particular, ensureTraceCompletionMarkers() may append an indexed raw + // system window after a crash or a bounded stop. Repairing it only after + // the agent drained left the final segment permanently local. + const int repaired_logs = repairUncompressedLogs(output_dir); + if (repaired_logs > 0) { + GFL_LOG_DEBUG("compressed ", repaired_logs, " log file(s)"); + } + if (!args.quiet) { std::fprintf(stderr, "[gpufl] inspect: %s\n", output_dir.string().c_str()); } @@ -877,11 +1067,6 @@ int runTraceCommon(const TraceArgs& args, const TracePlatform& platform) { } } - const int repaired_logs = repairUncompressedLogs(output_dir); - if (repaired_logs > 0) { - GFL_LOG_DEBUG("compressed ", repaired_logs, " log file(s)"); - } - return overall_rc; } diff --git a/include/gpufl/core/gpufl.cpp b/include/gpufl/core/gpufl.cpp index e351dec..f081cd5 100644 --- a/include/gpufl/core/gpufl.cpp +++ b/include/gpufl/core/gpufl.cpp @@ -706,7 +706,12 @@ bool init(const InitOptions& opts) { if (g_opts.system_sample_rate_ms > 0 && rt_ptr->collector) { rt_ptr->sampler.configure( rt_ptr->app_name, - [rt_ptr] { return rt_ptr->acquireSegmentContext(); }, + [rt_ptr] { + return rt_ptr->acquireSegmentContext("sampler"); + }, + [rt_ptr] { + return rt_ptr->peekSegmentContext(); + }, rt_ptr->collector, g_opts.system_sample_rate_ms, rt_ptr->host_collector.get(), [rt_ptr](const uint32_t index, const uint64_t rows, @@ -847,7 +852,7 @@ void shutdown() { Monitor::Shutdown(); } GFL_LOG_DEBUG("Shutdown: monitor drained -> finalize logs"); - auto final_segment = rt->acquireSegmentContext(); + auto final_segment = rt->acquireSegmentContext("shutdown"); if (!final_segment || !final_segment->logger) { GFL_LOG_ERROR("Shutdown: active segment context disappeared"); set_runtime(nullptr); diff --git a/include/gpufl/core/logger/log_salvage.cpp b/include/gpufl/core/logger/log_salvage.cpp index e5fc211..735b784 100644 --- a/include/gpufl/core/logger/log_salvage.cpp +++ b/include/gpufl/core/logger/log_salvage.cpp @@ -103,6 +103,14 @@ bool endsWith(const std::string& value, const std::string& suffix) { } // namespace +bool parsePublishedWindowName(const std::string& filename, + std::string& channel, + std::size_t& sequence) { + bool compressed = false; + return parseWindowName(filename, channel, sequence, compressed) && + sequence > 0; +} + MoveFileNoReplaceResult moveFileNoReplace(const fs::path& from, const fs::path& to, std::error_code& ec) { @@ -621,9 +629,13 @@ LogSalvageResult salvageOwnedSessionTempDir(const fs::path& session_dir) { ++result.deferred; continue; } + // Fingerprint the staged file that exists now, but record the + // name it gets when published below - the agent only ever sees + // `target`, and a sidecar naming the staging file is rejected as + // a contract violation. if (!ensureWindowMetadata( session_dir, session_dir.filename().string(), channel, - idx, path)) { + idx, path, target.filename().string())) { ++result.deferred; continue; } @@ -712,9 +724,11 @@ LogSalvageResult salvageOwnedSessionTempDir(const fs::path& session_dir) { ++result.deferred; continue; } + // Same split as above: `staging` holds the bytes right now, `target` + // is the name the published window will carry. if (!ensureWindowMetadata( session_dir, session_dir.filename().string(), channel, idx, - staging)) { + staging, target.filename().string())) { ++result.deferred; continue; } diff --git a/include/gpufl/core/logger/log_salvage.hpp b/include/gpufl/core/logger/log_salvage.hpp index 140aa5d..32d6c42 100644 --- a/include/gpufl/core/logger/log_salvage.hpp +++ b/include/gpufl/core/logger/log_salvage.hpp @@ -56,6 +56,20 @@ MoveFileNoReplaceResult moveFileNoReplace( const std::filesystem::path& to, std::error_code& ec); +/** + * Split a published window filename (`..log` or `..log.gz`) + * into its channel and sequence. Returns false for anything that is not a + * published window, including the un-indexed active file `.log` + * (which yields sequence 0). + * + * Exposed so every publisher - rotator, salvage, launcher repair - derives the + * identity from ONE parser. A second copy of this convention is how the + * launcher previously allocated an index that ignored `.tmp`. + */ +bool parsePublishedWindowName(const std::string& filename, + std::string& channel, + std::size_t& sequence); + /** * True when `path` is a non-empty file that decodes cleanly as gzip all the * way to EOF. Existence is NOT proof: a zero-length file decodes as a clean diff --git a/include/gpufl/core/logger/window_metadata.cpp b/include/gpufl/core/logger/window_metadata.cpp index 17785c9..2c42896 100644 --- a/include/gpufl/core/logger/window_metadata.cpp +++ b/include/gpufl/core/logger/window_metadata.cpp @@ -77,7 +77,7 @@ bool existingMetadataMatches( const std::string& session_id, const std::string& channel, const std::size_t sequence, - const fs::path& payload, + const std::string& published_name, const std::uint64_t payload_bytes, const std::uint32_t payload_crc32) { const auto metadata = json::loadFile(metadata_path.string()); @@ -90,8 +90,7 @@ bool existingMetadataMatches( metadata.value("channel", "") != channel || metadata.value("window_sequence", kMissing) != sequence || - metadata.value("payload_file", "") != - payload.filename().string() || + metadata.value("payload_file", "") != published_name || metadata.value("payload_bytes", kMissing) != payload_bytes || metadata.value("payload_crc32", kMissing) != @@ -121,21 +120,37 @@ bool ensureWindowMetadata(const fs::path& session_dir, const std::size_t sequence, const fs::path& payload, const WindowTiming& timing) { + return ensureWindowMetadata(session_dir, session_id, channel, sequence, + payload, payload.filename().string(), timing); +} + +bool ensureWindowMetadata(const fs::path& session_dir, + const std::string& session_id, + const std::string& channel, + const std::size_t sequence, + const fs::path& fingerprint_source, + const std::string& published_name, + const WindowTiming& timing) { const fs::path target = windowMetadataPath(session_dir, channel, sequence); std::uint64_t payload_bytes = 0; std::uint32_t payload_crc32 = 0; - if (!payloadFingerprint(payload, payload_bytes, payload_crc32)) { + // Read the file that exists NOW; record the name it will have when a + // consumer sees it. A publish is a rename of these exact bytes, so the + // fingerprint stays valid across it. + if (!payloadFingerprint(fingerprint_source, payload_bytes, + payload_crc32)) { GFL_LOG_ERROR("[Logger] cannot fingerprint window payload '", - payload.string(), "'; metadata not published."); + fingerprint_source.string(), + "'; metadata not published."); return false; } std::error_code state_ec; if (fs::is_regular_file(target, state_ec)) { return existingMetadataMatches( - target, session_id, channel, sequence, payload, payload_bytes, - payload_crc32); + target, session_id, channel, sequence, published_name, + payload_bytes, payload_crc32); } WindowMetadata metadata; @@ -148,7 +163,7 @@ bool ensureWindowMetadata(const fs::path& session_dir, std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch()) .count(); - metadata.payload_file = payload.filename().string(); + metadata.payload_file = published_name; metadata.payload_bytes = payload_bytes; metadata.payload_crc32 = payload_crc32; @@ -188,7 +203,7 @@ bool ensureWindowMetadata(const fs::path& session_dir, std::error_code target_ec; if (fs::is_regular_file(target, target_ec)) { return existingMetadataMatches( - target, session_id, channel, sequence, payload, + target, session_id, channel, sequence, published_name, payload_bytes, payload_crc32); } GFL_LOG_ERROR("[Logger] immutable window metadata path '", diff --git a/include/gpufl/core/logger/window_metadata.hpp b/include/gpufl/core/logger/window_metadata.hpp index 1fc6fe4..2e37cd3 100644 --- a/include/gpufl/core/logger/window_metadata.hpp +++ b/include/gpufl/core/logger/window_metadata.hpp @@ -31,6 +31,26 @@ struct WindowMetadata { * The payload may later be deleted, but this small file remains until the * session is retired. That keeps the sequence consumed and gives the agent a * stable idempotency key independent of filename timestamps. + * + * `fingerprint_source` is the file READ NOW to compute size and CRC. + * `published_name` is the basename the window will carry once it is visible + * to a consumer. They differ on every path that fingerprints a staged file + * before renaming it into place - the sidecar has to describe the published + * window, because that is the only name the agent can ever see. Recording the + * staging name made the agent reject the pair as a contract violation and + * refuse to upload it. + */ +bool ensureWindowMetadata(const std::filesystem::path& session_dir, + const std::string& session_id, + const std::string& channel, + std::size_t sequence, + const std::filesystem::path& fingerprint_source, + const std::string& published_name, + const WindowTiming& timing = {}); + +/** + * Overload for callers whose payload is already at its published name. + * Equivalent to passing `payload.filename()` as `published_name`. */ bool ensureWindowMetadata(const std::filesystem::path& session_dir, const std::string& session_id, diff --git a/include/gpufl/core/monitor_batch_manager.cpp b/include/gpufl/core/monitor_batch_manager.cpp index 5fdb8e4..d0c9260 100644 --- a/include/gpufl/core/monitor_batch_manager.cpp +++ b/include/gpufl/core/monitor_batch_manager.cpp @@ -81,7 +81,8 @@ void MonitorBatchManager::flushAll(FlushMode mode) { return; } - const auto context = flushSink_.runtime->acquireSegmentContext(); + const auto context = + flushSink_.runtime->acquireSegmentContext("monitor_batch"); if (!context || !context->logger) { GFL_LOG_ERROR( "MonitorBatchManager::flushAll: active segment context is missing"); @@ -110,7 +111,11 @@ void MonitorBatchManager::flushAll(FlushMode mode) { flushDictionary(); logger.write(model::KernelEventBatchModel(kernelBatch_, session_id, ++kernelBatchId_)); kernelBatch_.clear(); - for (const auto& d : pendingDetails_) { + for (auto& d : pendingDetails_) { + // Details may have been queued before a segment publication. The + // flush lease is the linearization point and therefore the sole + // authority for the session identity written with this batch. + d.session_id = session_id; logger.write(model::KernelDetailModel(d)); } pendingDetails_.clear(); @@ -201,7 +206,8 @@ void MonitorBatchManager::enqueueDisassembly(uint64_t crc, const uint8_t* data, void MonitorBatchManager::flushDisassembly() { if (!flushSink_.available()) return; - const auto context = flushSink_.runtime->acquireSegmentContext(); + const auto context = + flushSink_.runtime->acquireSegmentContext("monitor_batch"); if (!context || !context->logger) return; dictManager_.flushDisassembly(*context->logger, context->session_id); } diff --git a/include/gpufl/core/monitor_record_builders.cpp b/include/gpufl/core/monitor_record_builders.cpp index 22339db..7c7d261 100644 --- a/include/gpufl/core/monitor_record_builders.cpp +++ b/include/gpufl/core/monitor_record_builders.cpp @@ -29,7 +29,10 @@ KernelDetailRow MakeKernelDetailRow(const ActivityRecord& rec, const Runtime& rt) { KernelDetailRow detail; detail.corr_id = rec.corr_id; - detail.session_id = rt.session_id; + // Segment identity is assigned when the pending detail is flushed. The + // Runtime's session_id is the initial process session and intentionally + // does not change at a segment cutover; copying it here would stamp later + // segments with the previous session's identity. detail.pid = GetPid(); detail.app = rt.app_name; detail.grid_x = rec.grid_x; detail.grid_y = rec.grid_y; detail.grid_z = rec.grid_z; diff --git a/include/gpufl/core/runtime.cpp b/include/gpufl/core/runtime.cpp index 685fd65..e11aa3d 100644 --- a/include/gpufl/core/runtime.cpp +++ b/include/gpufl/core/runtime.cpp @@ -6,15 +6,35 @@ namespace gpufl { // handlers can run after normal function-local/static teardown has begun. static auto* g_rt = new std::unique_ptr; -bool SegmentContext::tryAcquireWriter() const noexcept { +bool SegmentContext::tryAcquireWriter(const char* const owner) const noexcept { if (!accepting_writers_.load(std::memory_order_seq_cst)) return false; active_writers_.fetch_add(1, std::memory_order_seq_cst); + if (!run_id.empty()) { + // Diagnostics must never make a previously valid acquisition fail. + // Allocation can throw on the first sighting of an owner label. + try { + std::lock_guard lock(writer_owner_mu_); + ++writer_owners_[owner ? owner : "general"]; + } catch (...) { + } + } if (accepting_writers_.load(std::memory_order_seq_cst)) return true; - releaseWriter(); + releaseWriter(owner); return false; } -void SegmentContext::releaseWriter() const noexcept { +void SegmentContext::releaseWriter(const char* const owner) const noexcept { + if (!run_id.empty()) { + std::lock_guard lock(writer_owner_mu_); + const auto it = writer_owners_.find(owner ? owner : "general"); + if (it != writer_owners_.end()) { + if (it->second <= 1) { + writer_owners_.erase(it); + } else { + --it->second; + } + } + } const uint64_t before = active_writers_.fetch_sub(1, std::memory_order_seq_cst); // The normal hot path is one atomic decrement. Once sealed, the last @@ -27,6 +47,18 @@ void SegmentContext::releaseWriter() const noexcept { } } +std::string SegmentContext::activeWriterSummary() const { + std::lock_guard lock(writer_owner_mu_); + std::string summary; + for (const auto& [owner, count] : writer_owners_) { + if (!summary.empty()) summary += ", "; + summary += owner; + summary += "="; + summary += std::to_string(count); + } + return summary.empty() ? "none" : summary; +} + void SegmentContext::sealForRetirement() const noexcept { accepting_writers_.store(false, std::memory_order_seq_cst); } @@ -46,13 +78,14 @@ bool SegmentContext::waitForWriters(const std::chrono::milliseconds timeout, SegmentWriteLease::~SegmentWriteLease() { reset(); } SegmentWriteLease::SegmentWriteLease(SegmentWriteLease&& other) noexcept - : context_(std::move(other.context_)) {} + : context_(std::move(other.context_)), owner_(other.owner_) {} SegmentWriteLease& SegmentWriteLease::operator=( SegmentWriteLease&& other) noexcept { if (this != &other) { reset(); context_ = std::move(other.context_); + owner_ = other.owner_; } return *this; } @@ -60,17 +93,17 @@ SegmentWriteLease& SegmentWriteLease::operator=( void SegmentWriteLease::reset() noexcept { if (!context_) return; const auto context = std::move(context_); - context->releaseWriter(); + context->releaseWriter(owner_); } SegmentWriteLease -Runtime::acquireSegmentContext() const noexcept { +Runtime::acquireSegmentContext(const char* const owner) const noexcept { for (;;) { auto context = std::atomic_load_explicit( &active_segment_context, std::memory_order_acquire); if (!context) return {}; - if (context->tryAcquireWriter()) { - return SegmentWriteLease(std::move(context)); + if (context->tryAcquireWriter(owner)) { + return SegmentWriteLease(std::move(context), owner); } // Publication sealed this context after our atomic load. Retry against // the newly-published context instead of writing into retirement. diff --git a/include/gpufl/core/runtime.hpp b/include/gpufl/core/runtime.hpp index 915a88f..f8798e9 100644 --- a/include/gpufl/core/runtime.hpp +++ b/include/gpufl/core/runtime.hpp @@ -32,7 +32,8 @@ struct Runtime { std::shared_ptr active_segment_context; std::shared_ptr segment_runtime; - SegmentWriteLease acquireSegmentContext() const noexcept; + SegmentWriteLease acquireSegmentContext( + const char* owner = "general") const noexcept; /** Liveness check only; does not participate in writer drainage. */ bool hasSegmentContext() const noexcept; /** Coordinator-only read. This does not protect a write operation. */ diff --git a/include/gpufl/core/sampler.cpp b/include/gpufl/core/sampler.cpp index 8332b11..fd9bf7a 100644 --- a/include/gpufl/core/sampler.cpp +++ b/include/gpufl/core/sampler.cpp @@ -27,17 +27,22 @@ void Sampler::configure(std::string appName, std::string sessionId, [fixed_runtime]() { return fixed_runtime->acquireSegmentContext(); }, + [fixed_runtime]() { + return fixed_runtime->peekSegmentContext(); + }, std::move(collector), sampleIntervalMs, hostCollector); } void Sampler::configure( std::string appName, SegmentProvider segmentProvider, + SegmentPeekProvider segmentPeekProvider, std::shared_ptr> collector, const int sampleIntervalMs, HostCollector* hostCollector, RowObserver rowObserver) { std::lock_guard lk(mu_); appName_ = std::move(appName); segment_provider_ = std::move(segmentProvider); + segment_peek_provider_ = std::move(segmentPeekProvider); collector_ = std::move(collector); host_collector_ = hostCollector; row_observer_ = std::move(rowObserver); @@ -88,9 +93,26 @@ void Sampler::deactivate() { void Sampler::shutdown() { std::lock_guard lk(mu_); activations_.store(0, std::memory_order_release); + const SegmentContext* const retained_context = batch_context_.get(); + if (retained_context) { + GFL_LOG_DEBUG( + "[Sampler] shutdown before worker join/reset: owners={", + retained_context->activeWriterSummary(), "}"); + } if (running_.load()) { stopWorkerLocked_(); } + // stopWorkerLocked_() joins the producer, so no code can touch the + // retained batch lease after this point. Release it unconditionally: + // an empty/invalid final batch can return from flushBatches_ before its + // normal reset, and retaining that lease would prevent the final segment + // from draining and becoming uploadable. + batch_context_.reset(); + if (retained_context) { + GFL_LOG_DEBUG( + "[Sampler] shutdown after worker join/reset: owners={", + retained_context->activeWriterSummary(), "}"); + } } void Sampler::startWorkerLocked_() { @@ -127,13 +149,14 @@ void Sampler::runLoop_() { while (running_.load()) { next_wake_time += interval; const int64_t ts = detail::GetTimestampNs(); - auto context = - segment_provider_ ? segment_provider_() : nullptr; - if (!context || !context->logger) { + const auto active_context = + segment_peek_provider_ ? segment_peek_provider_() : nullptr; + if (!active_context || !active_context->logger) { std::this_thread::sleep_until(next_wake_time); continue; } - if (batch_context_ && batch_context_.get() != context.get()) { + if (batch_context_ && + batch_context_.get() != active_context.get()) { if (!batch_.empty() || !host_batch_.empty()) { flushBatches_(batch_context_); samples_since_flush = 0; @@ -141,7 +164,14 @@ void Sampler::runLoop_() { batch_context_.reset(); } } - if (!batch_context_) batch_context_ = std::move(context); + if (!batch_context_) { + batch_context_ = + segment_provider_ ? segment_provider_() : nullptr; + } + if (!batch_context_ || !batch_context_->logger) { + std::this_thread::sleep_until(next_wake_time); + continue; + } for (const DeviceSample& d : collector_->sampleAll()) { // A rule reads gauges from here, not by polling: the timestamp has diff --git a/include/gpufl/core/sampler.hpp b/include/gpufl/core/sampler.hpp index dc46292..67cb52f 100644 --- a/include/gpufl/core/sampler.hpp +++ b/include/gpufl/core/sampler.hpp @@ -44,6 +44,8 @@ class ISystemCollector { class Sampler { public: using SegmentProvider = std::function; + using SegmentPeekProvider = + std::function()>; using RowObserver = std::function; @@ -64,12 +66,14 @@ class Sampler { HostCollector* hostCollector = nullptr); /** - * Segmentation-aware configuration. Each sampling iteration acquires one - * immutable context. If publication changes while rows are buffered, the - * old batch is committed through its original context before new rows are - * accepted. + * Segmentation-aware configuration. The sampler retains one writer lease + * for the batch and uses a read-only snapshot to detect publication. This + * is important during Windows injection teardown: ExitProcess can end the + * worker without running stack destructors, so a per-iteration writer + * lease could otherwise leak and block final-segment retirement. */ void configure(std::string appName, SegmentProvider segmentProvider, + SegmentPeekProvider segmentPeekProvider, std::shared_ptr> collector, int sampleIntervalMs, HostCollector* hostCollector = nullptr, @@ -123,6 +127,7 @@ class Sampler { std::string appName_; SegmentProvider segment_provider_; + SegmentPeekProvider segment_peek_provider_; RowObserver row_observer_; SegmentWriteLease batch_context_; std::shared_ptr> collector_; diff --git a/include/gpufl/core/segment_context.hpp b/include/gpufl/core/segment_context.hpp index dc0d201..d11e251 100644 --- a/include/gpufl/core/segment_context.hpp +++ b/include/gpufl/core/segment_context.hpp @@ -6,6 +6,7 @@ #include #include #include +#include #include namespace gpufl { @@ -46,18 +47,25 @@ struct SegmentContext { private: friend class SegmentWriteLease; friend class SegmentRuntime; + friend class Sampler; friend struct Runtime; - bool tryAcquireWriter() const noexcept; - void releaseWriter() const noexcept; + bool tryAcquireWriter(const char* owner) const noexcept; + void releaseWriter(const char* owner) const noexcept; void sealForRetirement() const noexcept; bool waitForWriters(std::chrono::milliseconds timeout, uint64_t* remaining) const noexcept; + std::string activeWriterSummary() const; mutable std::atomic accepting_writers_{true}; mutable std::atomic active_writers_{0}; mutable std::mutex writer_drain_mu_; mutable std::condition_variable writer_drain_cv_; + // Segmentation is opt-in, so retaining owner counts here adds no cost to + // ordinary sessions. When a drain times out this turns "one writer leaked" + // into an actionable producer name instead of an untraceable hang. + mutable std::mutex writer_owner_mu_; + mutable std::unordered_map writer_owners_; }; /** @@ -94,10 +102,12 @@ class SegmentWriteLease { private: friend struct Runtime; explicit SegmentWriteLease( - std::shared_ptr context) noexcept - : context_(std::move(context)) {} + std::shared_ptr context, + const char* owner) noexcept + : context_(std::move(context)), owner_(owner) {} std::shared_ptr context_; + const char* owner_ = "general"; }; } // namespace gpufl diff --git a/include/gpufl/core/segment_runtime.cpp b/include/gpufl/core/segment_runtime.cpp index 8bfa6ff..23b4f62 100644 --- a/include/gpufl/core/segment_runtime.cpp +++ b/include/gpufl/core/segment_runtime.cpp @@ -239,6 +239,7 @@ bool SegmentRuntime::awaitWriterDrain_( "; run=", context->run_id, " session=", context->session_id, " segment=", context->segment_index, " active_writers=", remaining, + " owners={", context->activeWriterSummary(), "}", ". The segment is intentionally left incomplete; its logger and " "ownership lock remain live until process exit."); quarantineUndrainedContext(context); diff --git a/tests/core/test_file_log_sink_rotation.cpp b/tests/core/test_file_log_sink_rotation.cpp index bbc7b8a..28601c2 100644 --- a/tests/core/test_file_log_sink_rotation.cpp +++ b/tests/core/test_file_log_sink_rotation.cpp @@ -105,6 +105,12 @@ class FileLogSinkRotationTest : public ::testing::Test { out.close(); } + static std::string readText(const fs::path& path) { + std::ifstream in(path, std::ios::binary); + return std::string((std::istreambuf_iterator(in)), + std::istreambuf_iterator()); + } + static void writeEmptyFile(const fs::path& path) { fs::create_directories(path.parent_path()); std::ofstream out(path, std::ios::binary | std::ios::trunc); @@ -489,6 +495,81 @@ TEST_F(FileLogSinkRotationTest, EXPECT_FALSE(fs::exists(tmpDir())); } +// The sidecar must name the window the agent will actually see. Salvage +// fingerprints a STAGED file and then renames it, so recording +// `payload.filename()` wrote the staging name ("device.log.gz") while the +// published window was "device.1.log.gz". The agent's isValidFor() compared +// the two, rejected the pair as a contract violation, wrote a durable loss +// marker, and refused to upload - observed on every final segment of a real +// segmented run. +TEST_F(FileLogSinkRotationTest, + SalvagedWindowMetadataNamesThePublishedFileNotTheStagingFile) { + fs::create_directories(tmpDir()); + const std::string payload = R"({"window":"active"})"; + writeText(tmpDir() / "device.log", payload); // un-indexed active file + + const auto result = gpufl::salvageSessionTempDir(sessionDir()); + + ASSERT_EQ(result.deferred, 0u); + ASSERT_EQ(publishedWindows("device"), 1u); + ASSERT_TRUE(fs::is_regular_file(sessionDir() / "device.1.log.gz")); + + const fs::path sidecar = sessionDir() / ".gpufl-window.device.1.json"; + ASSERT_TRUE(fs::is_regular_file(sidecar)); + const std::string json = readText(sidecar); + EXPECT_NE(json.find("\"payload_file\":\"device.1.log.gz\""), + std::string::npos) + << "sidecar must name the published window; got: " << json; + EXPECT_EQ(json.find("\"payload_file\":\"device.log.gz\""), + std::string::npos) + << "sidecar names the staging file, which the agent will reject"; +} + +// The other salvage publish path: an UN-INDEXED gzip in `.tmp` +// (`device.log.gz`), which salvage publishes under a freshly assigned index. +// The name split only appears when the staged basename differs from the +// target's - an already-indexed staged file has the same basename in both +// places, so it cannot expose this. The final segment of a real segmented run +// leaves exactly this shape behind. +TEST_F(FileLogSinkRotationTest, + UnindexedStagedGzipSalvageMetadataNamesThePublishedFile) { + fs::create_directories(tmpDir()); + const fs::path raw = tmpDir() / "device.log"; + const fs::path staged = tmpDir() / "device.log.gz"; + writeText(raw, R"({"window":"active","payload":"staged before publish"})"); + gpufl::GzipFileCompressor compressor; + ASSERT_TRUE(compressor.compressTo(raw.string(), staged.string())); + fs::remove(raw); // the rotator had already consumed its raw source + + const auto result = gpufl::salvageSessionTempDir(sessionDir()); + + ASSERT_EQ(result.deferred, 0u); + ASSERT_TRUE(fs::is_regular_file(sessionDir() / "device.1.log.gz")); + + const fs::path sidecar = sessionDir() / ".gpufl-window.device.1.json"; + ASSERT_TRUE(fs::is_regular_file(sidecar)); + const std::string json = readText(sidecar); + EXPECT_NE(json.find("\"payload_file\":\"device.1.log.gz\""), + std::string::npos) + << "sidecar must name the published window; got: " << json; + EXPECT_EQ(json.find("\"payload_file\":\"device.log.gz\""), + std::string::npos) + << "sidecar names the staging file, which the agent will reject"; +} + +TEST_F(FileLogSinkRotationTest, + PublishedWindowParserRejectsTheUnindexedActiveFilename) { + std::string channel; + std::size_t sequence = 99; + + EXPECT_FALSE(gpufl::parsePublishedWindowName( + "device.log.gz", channel, sequence)); + EXPECT_TRUE(gpufl::parsePublishedWindowName( + "device.7.log.gz", channel, sequence)); + EXPECT_EQ(channel, "device"); + EXPECT_EQ(sequence, 7u); +} + TEST_F(FileLogSinkRotationTest, SalvagePrefersCompletedGzipOverDuplicateRaw) { fs::create_directories(tmpDir()); diff --git a/tests/core/test_sampler.cpp b/tests/core/test_sampler.cpp index baac16d..9ba0903 100644 --- a/tests/core/test_sampler.cpp +++ b/tests/core/test_sampler.cpp @@ -20,13 +20,19 @@ #include +#include #include #include +#include #include #include +#include #include "gpufl/core/logger/logger.hpp" +#include "gpufl/core/logger/log_sink.hpp" +#include "gpufl/core/runtime.hpp" #include "gpufl/core/sampler.hpp" +#include "gpufl/core/segment_runtime.hpp" namespace { @@ -53,6 +59,55 @@ std::shared_ptr makeUnopenedLogger() { return std::make_shared(); } +class RecordingSink final : public gpufl::ILogSink { + public: + explicit RecordingSink( + std::shared_ptr> lines) + : lines_(std::move(lines)) {} + + void write(gpufl::Channel, std::string_view json) override { + lines_->emplace_back(json); + } + void close() override {} + + private: + std::shared_ptr> lines_; +}; + +class BlockingSecondCallCollector final + : public gpufl::ISystemCollector { + public: + std::vector sampleAll() override { + std::unique_lock lock(mu_); + ++calls_; + cv_.notify_all(); + if (calls_ == 2) { + cv_.wait(lock, [this] { return release_second_; }); + } + return {}; + } + + bool waitForSecondCall() { + std::unique_lock lock(mu_); + return cv_.wait_for(lock, std::chrono::seconds(1), + [this] { return calls_ >= 2; }); + } + + void releaseSecondCall() { + { + std::lock_guard lock(mu_); + release_second_ = true; + } + cv_.notify_all(); + } + + private: + std::mutex mu_; + std::condition_variable cv_; + int calls_ = 0; + bool release_second_ = false; +}; + } // namespace TEST(SamplerRefCount, ConfiguredButNotActivated) { @@ -193,3 +248,95 @@ TEST(SamplerRefCount, ReactivationAfterFullReleaseSpawnsFreshWorker) { EXPECT_GT(coll->calls(), calls_after_first_run) << "Second activation should produce additional collector calls."; } + +TEST(SamplerRefCount, ShutdownReleasesEverySegmentWriterLease) { + gpufl::Runtime runtime; + runtime.run_id = "12345678-1234-4123-8123-123456789abc"; + runtime.session_id = "sampler-final-segment"; + auto lines = std::make_shared>(); + auto logger = std::make_shared(); + logger->addSink(std::make_unique(lines)); + runtime.logger = logger; + ASSERT_TRUE(runtime.publishSegmentContext( + std::make_shared( + runtime.run_id, runtime.session_id, 0, 1, logger))); + + gpufl::InitEvent init; + init.pid = 1; + init.app = "sampler-lease-test"; + init.session_id = runtime.session_id; + init.run_id = runtime.run_id; + init.segment_index = 0; + + gpufl::SegmentRuntime::Options segment_options; + segment_options.runtime = &runtime; + segment_options.init_template = init; + segment_options.segment_max_rows = 1; + segment_options.retirement_drain_timeout_ms = 30; + auto segmented = + std::make_shared(std::move(segment_options)); + runtime.segment_runtime = segmented; + ASSERT_TRUE(segmented->start()); + + auto collector = std::make_shared(); + gpufl::Sampler sampler; + sampler.configure( + "sampler-lease-test", + [&runtime] { return runtime.acquireSegmentContext(); }, + [&runtime] { return runtime.peekSegmentContext(); }, + collector, /*sampleIntervalMs=*/100); + sampler.activate(); + const auto sampled_deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(1); + while (collector->calls() == 0 && + std::chrono::steady_clock::now() < sampled_deadline) { + std::this_thread::yield(); + } + ASSERT_GT(collector->calls(), 0); + + // The worker is now normally sleeping between samples. shutdown() must + // join it and release its retained-batch lease; + // otherwise SegmentRuntime refuses to publish finality. + sampler.shutdown(); + segmented->finish(2); + + EXPECT_TRUE(std::any_of( + lines->begin(), lines->end(), [](const std::string& line) { + return line.find("\"type\":\"run_end\"") != + std::string::npos; + })); +} + +TEST(SamplerRefCount, ReusesBatchLeaseWhileSamplingSameSegment) { + gpufl::Runtime runtime; + runtime.run_id = "12345678-1234-4123-8123-123456789abc"; + runtime.session_id = "sampler-single-lease"; + auto logger = makeUnopenedLogger(); + ASSERT_TRUE(runtime.publishSegmentContext( + std::make_shared( + runtime.run_id, runtime.session_id, 0, 1, logger))); + + auto collector = std::make_shared(); + std::atomic writer_acquisitions{0}; + gpufl::Sampler sampler; + sampler.configure( + "sampler-single-lease", + [&runtime, &writer_acquisitions] { + writer_acquisitions.fetch_add(1, std::memory_order_relaxed); + return runtime.acquireSegmentContext(); + }, + [&runtime] { return runtime.peekSegmentContext(); }, + collector, /*sampleIntervalMs=*/5); + + sampler.activate(); + const bool reached_second_call = collector->waitForSecondCall(); + EXPECT_TRUE(reached_second_call); + // The retained batch lease already protects the second sample. Taking a + // second writer lease here is unsafe in Windows injection teardown: + // ExitProcess may terminate the worker without unwinding its stack. + if (reached_second_call) { + EXPECT_EQ(writer_acquisitions.load(std::memory_order_relaxed), 1); + } + collector->releaseSecondCall(); + sampler.shutdown(); +} diff --git a/tests/core/test_segment_context.cpp b/tests/core/test_segment_context.cpp index 4c47676..94610e1 100644 --- a/tests/core/test_segment_context.cpp +++ b/tests/core/test_segment_context.cpp @@ -18,6 +18,7 @@ #include "gpufl/core/logger/log_sink.hpp" #include "gpufl/core/logger/logger.hpp" #include "gpufl/core/logger/session_ownership.hpp" +#include "gpufl/core/monitor_batch_manager.hpp" #include "gpufl/core/model/lifecycle_model.hpp" #include "gpufl/core/runtime.hpp" #include "gpufl/core/segment_runtime.hpp" @@ -144,6 +145,43 @@ TEST(SegmentContextTest, DictionaryEmissionIsIndependentPerSegment) { EXPECT_NE((*lines)[3].find("\"2\":\"kernel_b\""), std::string::npos); } +TEST(SegmentContextTest, PendingKernelDetailUsesTheFlushSegmentIdentity) { + gpufl::Runtime runtime; + runtime.session_id = "initial-process-session"; + auto lines = std::make_shared>(); + auto logger = std::make_shared(); + logger->addSink(std::make_unique(lines)); + ASSERT_TRUE(runtime.publishSegmentContext( + std::make_shared( + "12345678-1234-4123-8123-123456789abc", + "current-segment-session", 1, 1000, logger))); + + gpufl::detail::MonitorBatchManager batches; + batches.bindFlushRuntime(&runtime); + gpufl::KernelBatchRow kernel; + kernel.kernel_id = 1; + gpufl::KernelDetailRow detail; + detail.session_id = "stale-previous-segment"; + detail.corr_id = 7; + ASSERT_FALSE(batches.pushKernel(kernel, &detail)); + batches.flushAll( + gpufl::detail::MonitorBatchManager::FlushMode::Full); + + const auto detail_line = std::find_if( + lines->begin(), lines->end(), [](const std::string& line) { + return line.find("\"type\":\"kernel_detail\"") != + std::string::npos; + }); + ASSERT_NE(detail_line, lines->end()); + EXPECT_NE(detail_line->find( + "\"session_id\":\"current-segment-session\""), + std::string::npos); + EXPECT_EQ(detail_line->find("stale-previous-segment"), + std::string::npos); + EXPECT_EQ(detail_line->find("initial-process-session"), + std::string::npos); +} + TEST(SegmentContextTest, ProductionRuntimePublishesAndRetiresTwoSegments) { const fs::path root = fs::temp_directory_path() /