Something went wrong. Try again.
Toolkit for tile-based quality reconstruction of astronomical image stacks
dwarf stacking two-seestar
Something went wrong. Try again.
28 kB · 663 lines
C++
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664#include "subprocess_manager.hpp"#include <algorithm>#include <filesystem>#include <fstream>#include <iostream>#include <sstream>#include <array>#include <stdexcept>#include <thread>#include <chrono>
#ifdef _WIN32# include <windows.h>#else# include <unistd.h># include <sys/wait.h># include <fcntl.h># include <signal.h>#endif
namespace {
using json = nlohmann::json;constexpr size_t MAX_JSON_STRING_BYTES = 8 * 1024;constexpr size_t MAX_JSON_ARRAY_ITEMS = 256;constexpr size_t MAX_JSON_OBJECT_ITEMS = 256;constexpr int MAX_JSON_DEPTH = 8;namespace fs = std::filesystem;
struct CapturedText { std::string text; size_t total_bytes{0}; bool truncated{false};};
std::string fnv1a_hex(const std::string& value) { uint64_t hash = 1469598103934665603ull; for (unsigned char ch : value) { hash ^= ch; hash *= 1099511628211ull; } std::ostringstream out; out << std::hex << hash; return out.str();}
void write_scan_metrics_cache(const std::string& cwd, const std::string& job_id, const json& data, const json& result) { if (!data.contains("cache_key") || !data["cache_key"].is_string()) return; if (!result.is_object() || !result.value("ok", false)) return; const std::string cache_key = data["cache_key"].get<std::string>(); if (cache_key.empty()) return; try { fs::path cache_dir = fs::path(cwd) / "runs" / ".pi_memory" / "scan_metrics_cache"; fs::create_directories(cache_dir); json payload = { {"schema_version", "tile_compile.scan_metrics_cache.v1"}, {"cache_key", cache_key}, {"job_id", job_id}, {"created_at", data.value("ended_at", data.value("updated_at", std::string()))}, {"input_path", data.value("input_path", std::string())}, {"object_name", data.value("object_name", std::string())}, {"frame_count", data.value("frame_count", 0)}, {"result", result} }; std::ofstream out(cache_dir / (fnv1a_hex(cache_key) + ".json")); out << payload.dump(2); } catch (...) { // Cache write failures must not fail the scan-metrics job. }}
/// @brief Implements truncate text./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.std::string truncate_text(const std::string& raw, size_t max_bytes) { if (raw.size() <= max_bytes) return raw; if (max_bytes == 0) return {}; static const std::string suffix = "\n...[truncated]"; if (max_bytes <= suffix.size()) return raw.substr(0, max_bytes); return raw.substr(0, max_bytes - suffix.size()) + suffix;}
json compact_json_for_job_storage(const json& value, const BackendGuardLimits& limits, int depth = 0);
json compact_json_array(const json& value, size_t max_items, const BackendGuardLimits& limits, int depth = 0) { json out = json::array(); if (!value.is_array()) return out; const size_t limit = std::min(value.size(), max_items); for (size_t i = 0; i < limit; ++i) out.push_back(compact_json_for_job_storage(value[i], limits, depth + 1)); return out;}
json compact_json_for_job_storage(const json& value, const BackendGuardLimits& limits, int depth) { if (depth >= MAX_JSON_DEPTH) return "[truncated depth]"; if (value.is_string()) return truncate_text(value.get<std::string>(), MAX_JSON_STRING_BYTES); if (value.is_array()) return compact_json_array(value, MAX_JSON_ARRAY_ITEMS, limits, depth); if (value.is_object()) { json out = json::object(); size_t count = 0; for (auto it = value.begin(); it != value.end() && count < MAX_JSON_OBJECT_ITEMS; ++it, ++count) { out[it.key()] = compact_json_for_job_storage(it.value(), limits, depth + 1); } if (value.size() > MAX_JSON_OBJECT_ITEMS) { out["_truncated_fields"] = static_cast<long long>(value.size() - MAX_JSON_OBJECT_ITEMS); } return out; } return value;}
/// @brief Compacts common scan fields (errors, warnings, frames, color_mode_candidates)./// @details Shared logic between compact_scan_per_dir_result and compact_scan_job_result.static void compact_scan_common_fields(json& out, const json& raw, const BackendGuardLimits& limits, size_t frames_preview) { if (raw.contains("errors") && raw["errors"].is_array()) { out["errors"] = compact_json_array(raw["errors"], limits.scan_messages_preview, limits); out["errors_total"] = raw["errors"].size(); out["errors_truncated"] = raw["errors"].size() > limits.scan_messages_preview; } if (raw.contains("warnings") && raw["warnings"].is_array()) { out["warnings"] = compact_json_array(raw["warnings"], limits.scan_messages_preview, limits); out["warnings_total"] = raw["warnings"].size(); out["warnings_truncated"] = raw["warnings"].size() > limits.scan_messages_preview; } if (raw.contains("frames") && raw["frames"].is_array()) { out["frames"] = compact_json_array(raw["frames"], frames_preview, limits); out["frames_total"] = raw["frames"].size(); out["frames_truncated"] = raw["frames"].size() > frames_preview; } else if (!out.contains("frames")) { out["frames"] = json::array(); out["frames_total"] = 0; out["frames_truncated"] = false; } if (raw.contains("color_mode_candidates") && raw["color_mode_candidates"].is_array()) { out["color_mode_candidates"] = compact_json_array(raw["color_mode_candidates"], limits.scan_color_candidates_preview, limits); out["color_mode_candidates_total"] = raw["color_mode_candidates"].size(); out["color_mode_candidates_truncated"] = raw["color_mode_candidates"].size() > limits.scan_color_candidates_preview; }}
/// @brief Compacts scan per dir result./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.json compact_scan_per_dir_result(const json& raw, const BackendGuardLimits& limits) { json out = compact_json_for_job_storage(raw, limits); if (!raw.is_object()) return out; compact_scan_common_fields(out, raw, limits, limits.scan_per_dir_frames_preview); return out;}
/// @brief Compacts scan job result./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.json compact_scan_job_result(const json& raw, const BackendGuardLimits& limits) { json out = compact_json_for_job_storage(raw, limits); if (!raw.is_object()) return out; compact_scan_common_fields(out, raw, limits, limits.scan_frames_preview); if (raw.contains("per_dir_results") && raw["per_dir_results"].is_array()) { json per_dir = json::array(); const size_t limit = std::min(raw["per_dir_results"].size(), limits.scan_per_dir_results_preview); for (size_t i = 0; i < limit; ++i) per_dir.push_back(compact_scan_per_dir_result(raw["per_dir_results"][i], limits)); out["per_dir_results"] = std::move(per_dir); out["per_dir_results_total"] = raw["per_dir_results"].size(); out["per_dir_results_truncated"] = raw["per_dir_results"].size() > limits.scan_per_dir_results_preview; } return out;}
json compact_scan_metrics_job_result(const json& raw, const BackendGuardLimits& limits) { json out = compact_json_for_job_storage(raw, limits); if (!raw.is_object()) return out; if (raw.contains("frames") && raw["frames"].is_array()) { out["frames"] = raw["frames"]; out["frames_metrics_total"] = raw["frames"].size(); out["frames_metrics_truncated"] = false; } if (raw.contains("aggregate")) out["aggregate"] = compact_json_for_job_storage(raw["aggregate"], limits); return out;}
void store_process_output(json& data, const char* key, const std::string& text, size_t total_bytes, bool truncated, const BackendGuardLimits& limits) { data[key] = truncate_text(text, limits.job_stdio_store_bytes); data[std::string(key) + "_bytes"] = total_bytes; data[std::string(key) + "_truncated"] = truncated || text.size() > limits.job_stdio_store_bytes;}
#ifndef _WIN32struct SpawnedProcess { pid_t pid{-1}; int stdout_fd{-1}; int stderr_fd{-1};};
/// @brief Implements drain fd./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.CapturedText drain_fd(int fd, size_t capture_limit_bytes) { CapturedText out; if (fd < 0) return out; char buf[4096]; ssize_t n = 0; while ((n = read(fd, buf, sizeof(buf))) > 0) { out.total_bytes += static_cast<size_t>(n); const size_t remaining = out.text.size() < capture_limit_bytes ? (capture_limit_bytes - out.text.size()) : 0; const size_t to_copy = std::min(static_cast<size_t>(n), remaining); if (to_copy > 0) out.text.append(buf, to_copy); if (to_copy < static_cast<size_t>(n) || out.total_bytes > capture_limit_bytes) out.truncated = true; } close(fd); return out;}
bool spawn_subprocess(const std::vector<std::string>& args, const std::string& cwd, const std::string& stdin_text, SpawnedProcess& proc_out) { int pfd_out[2], pfd_err[2], pfd_in[2]; if (pipe(pfd_out) || pipe(pfd_err) || pipe(pfd_in)) return false;
pid_t pid = fork(); if (pid < 0) { // Close all pipe ends to avoid FD leak. close(pfd_out[0]); close(pfd_out[1]); close(pfd_err[0]); close(pfd_err[1]); close(pfd_in[0]); close(pfd_in[1]); return false; }
if (pid == 0) { close(pfd_out[0]); close(pfd_err[0]); close(pfd_in[1]); dup2(pfd_out[1], STDOUT_FILENO); dup2(pfd_err[1], STDERR_FILENO); dup2(pfd_in[0], STDIN_FILENO); close(pfd_out[1]); close(pfd_err[1]); close(pfd_in[0]); setpgid(0, 0); if (!cwd.empty() && chdir(cwd.c_str()) != 0) _exit(126);
std::vector<const char*> argv; for (auto& a : args) argv.push_back(a.c_str()); argv.push_back(nullptr); execvp(argv[0], const_cast<char* const*>(argv.data())); _exit(127); }
setpgid(pid, pid); close(pfd_out[1]); close(pfd_err[1]); close(pfd_in[0]); if (!stdin_text.empty()) { ssize_t total = 0; while (total < static_cast<ssize_t>(stdin_text.size())) { ssize_t n = write(pfd_in[1], stdin_text.data() + total, stdin_text.size() - static_cast<size_t>(total)); if (n <= 0) break; total += n; } } close(pfd_in[1]);
proc_out.pid = pid; proc_out.stdout_fd = pfd_out[0]; proc_out.stderr_fd = pfd_err[0]; return true;}
void terminate_spawned_process(int pid, int signal) { if (pid <= 0) return; // Send to the entire process group (pgid == pid, set via setpgid above) // so all child processes spawned by the runner are also terminated. if (kill(-static_cast<pid_t>(pid), signal) != 0) { kill(static_cast<pid_t>(pid), signal); }}
/// @brief Implements wait for process./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.int wait_for_process(BackgroundProcess& proc) { int status = 0; bool term_sent = false; int term_wait_cycles = 0; // Allow up to ~3 s for graceful shutdown after SIGTERM before sending SIGKILL. constexpr int SIGKILL_AFTER_CYCLES = 20; // 20 * 150 ms = 3 s while (true) { pid_t rc = waitpid(static_cast<pid_t>(proc.pid.load()), &status, WNOHANG); if (rc == static_cast<pid_t>(proc.pid.load())) return status; if (rc < 0) return -1;
if (proc.cancelled.load()) { if (!term_sent) { terminate_spawned_process(proc.pid.load(), SIGTERM); term_sent = true; term_wait_cycles = 0; } else if (++term_wait_cycles >= SIGKILL_AFTER_CYCLES) { terminate_spawned_process(proc.pid.load(), SIGKILL); } } std::this_thread::sleep_for(std::chrono::milliseconds(term_sent ? 150 : 100)); }}#endif
} // namespace
/// @brief Executes a foreground subprocess and captures bounded stdout/stderr./// @details Forks or spawns the requested command, writes optional stdin, drains output pipes,/// applies backend capture limits, and returns exit status plus truncation metadata.SubprocessResult run_subprocess(const std::vector<std::string>& args, const std::string& cwd, const std::string& stdin_text, const BackendGuardLimits* limits_override) { SubprocessResult res; if (args.empty()) { res.exit_code = -1; return res; } const BackendGuardLimits limits = limits_override ? *limits_override : backend_guard_limits_from_env();
#ifdef _WIN32 // Build a Windows command line with proper quoting. // Each argument must be double-quoted, and any backslashes preceding a // closing quote must be doubled (e.g. C:\path\ -> "C:\path\\"). // See: https://learn.microsoft.com/en-us/archive/blogs/twistylittlepassagesallalike/everyone-quotes-command-line-arguments-the-wrong-way std::string cmd; for (auto& a : args) { cmd += '"'; for (size_t i = 0; i < a.size(); ++i) { if (a[i] == '"') { cmd += "\\\""; } else if (a[i] == '\\' && i + 1 == a.size()) { // Trailing backslash before closing quote: double it cmd += "\\\\"; } else { cmd += a[i]; } } cmd += "\" "; }
std::cerr << "[subprocess] Windows command line: " << cmd << std::endl;
SECURITY_ATTRIBUTES sa{}; sa.nLength = sizeof(sa); sa.bInheritHandle = TRUE;
HANDLE hStdoutR, hStdoutW, hStderrR, hStderrW, hStdinR, hStdinW; CreatePipe(&hStdoutR, &hStdoutW, &sa, 0); CreatePipe(&hStderrR, &hStderrW, &sa, 0); CreatePipe(&hStdinR, &hStdinW, &sa, 0); SetHandleInformation(hStdoutR, HANDLE_FLAG_INHERIT, 0); SetHandleInformation(hStderrR, HANDLE_FLAG_INHERIT, 0); SetHandleInformation(hStdinW, HANDLE_FLAG_INHERIT, 0);
STARTUPINFOA si{}; si.cb = sizeof(si); si.hStdOutput = hStdoutW; si.hStdError = hStderrW; si.hStdInput = hStdinR; si.dwFlags |= STARTF_USESTDHANDLES;
// CreateProcessA does not support UNC paths (\\server\share) as lpCurrentDirectory. // If cwd is a UNC path, pass nullptr so the child inherits the parent's CWD. const char* cwd_ptr = nullptr; if (!cwd.empty()) { if (cwd.size() >= 2 && cwd[0] == '\\' && cwd[1] == '\\') { // UNC path — cannot be used as lpCurrentDirectory cwd_ptr = nullptr; } else { cwd_ptr = cwd.c_str(); } }
PROCESS_INFORMATION pi{}; // CreateProcessA may modify the command-line buffer, so provide a mutable copy. std::vector<char> cmd_buf(cmd.begin(), cmd.end()); cmd_buf.push_back('\0'); bool ok = CreateProcessA(nullptr, cmd_buf.data(), nullptr, nullptr, TRUE, 0, nullptr, cwd_ptr, &si, &pi); CloseHandle(hStdoutW); CloseHandle(hStderrW); CloseHandle(hStdinR);
if (!ok) { DWORD err = GetLastError(); CloseHandle(hStdinW); res.exit_code = -1; res.stderr_str = "CreateProcessA failed (error " + std::to_string(err) + "): " + cmd; return res; }
if (!stdin_text.empty()) { DWORD written = 0; WriteFile(hStdinW, stdin_text.data(), static_cast<DWORD>(stdin_text.size()), &written, nullptr); // Ignore partial write; child will receive what was written before pipe closes. } CloseHandle(hStdinW);
auto read_pipe = [capture_limit = limits.subprocess_capture_bytes](HANDLE h) { CapturedText out; char buf[4096]; DWORD n; while (ReadFile(h, buf, sizeof(buf), &n, nullptr) && n > 0) { out.total_bytes += static_cast<size_t>(n); const size_t remaining = out.text.size() < capture_limit ? (capture_limit - out.text.size()) : 0; const size_t to_copy = std::min(static_cast<size_t>(n), remaining); if (to_copy > 0) out.text.append(buf, to_copy); if (to_copy < static_cast<size_t>(n) || out.total_bytes > capture_limit) out.truncated = true; } return out; }; CapturedText stdout_capture = read_pipe(hStdoutR); CapturedText stderr_capture = read_pipe(hStderrR); res.stdout_str = std::move(stdout_capture.text); res.stderr_str = std::move(stderr_capture.text); res.stdout_bytes = stdout_capture.total_bytes; res.stderr_bytes = stderr_capture.total_bytes; res.stdout_truncated = stdout_capture.truncated; res.stderr_truncated = stderr_capture.truncated; CloseHandle(hStdoutR); CloseHandle(hStderrR);
WaitForSingleObject(pi.hProcess, INFINITE); DWORD ec; GetExitCodeProcess(pi.hProcess, &ec); res.exit_code = (int)ec; CloseHandle(pi.hProcess); CloseHandle(pi.hThread);
#else int pfd_out[2], pfd_err[2], pfd_in[2]; if (pipe(pfd_out) || pipe(pfd_err) || pipe(pfd_in)) { res.exit_code = -1; return res; }
pid_t pid = fork(); if (pid < 0) { close(pfd_out[0]); close(pfd_out[1]); close(pfd_err[0]); close(pfd_err[1]); close(pfd_in[0]); close(pfd_in[1]); res.exit_code = -1; return res; }
if (pid == 0) { close(pfd_out[0]); close(pfd_err[0]); close(pfd_in[1]); dup2(pfd_out[1], STDOUT_FILENO); dup2(pfd_err[1], STDERR_FILENO); dup2(pfd_in[0], STDIN_FILENO); close(pfd_out[1]); close(pfd_err[1]); close(pfd_in[0]);
if (!cwd.empty() && chdir(cwd.c_str()) != 0) _exit(126);
std::vector<const char*> argv; for (auto& a : args) argv.push_back(a.c_str()); argv.push_back(nullptr); execvp(argv[0], const_cast<char* const*>(argv.data())); _exit(127); }
close(pfd_out[1]); close(pfd_err[1]); close(pfd_in[0]);
if (!stdin_text.empty()) { ssize_t total = 0; while (total < static_cast<ssize_t>(stdin_text.size())) { ssize_t n = write(pfd_in[1], stdin_text.data() + total, stdin_text.size() - static_cast<size_t>(total)); if (n <= 0) break; total += n; } } close(pfd_in[1]);
// Read stdout and stderr concurrently to avoid deadlock when either pipe // buffer fills up while the parent is blocked reading the other pipe. CapturedText stdout_capture, stderr_capture; std::thread stdout_thread([&stdout_capture, fd = pfd_out[0], cap = limits.subprocess_capture_bytes]() { stdout_capture = drain_fd(fd, cap); }); std::thread stderr_thread([&stderr_capture, fd = pfd_err[0], cap = limits.subprocess_capture_bytes]() { stderr_capture = drain_fd(fd, cap); }); stdout_thread.join(); stderr_thread.join();
res.stdout_str = std::move(stdout_capture.text); res.stderr_str = std::move(stderr_capture.text); res.stdout_bytes = stdout_capture.total_bytes; res.stderr_bytes = stderr_capture.total_bytes; res.stdout_truncated = stdout_capture.truncated; res.stderr_truncated = stderr_capture.truncated;
int status = 0; waitpid(pid, &status, 0); res.exit_code = WIFEXITED(status) ? WEXITSTATUS(status) : -1;#endif
return res;}
/// @brief Starts a background subprocess as a tracked job./// @details Creates the job record, launches a worker thread, streams bounded diagnostics back/// into job data, and records completion, cancellation, or error state for API clients.std::string SubprocessManager::launch(const std::string& type, const std::vector<std::string>& args, const std::string& cwd, const std::string& run_id, const nlohmann::json& initial_data, const std::string& stdin_text, std::function<void(const std::string&, JobState)> on_complete) { std::string job_id = _store.create(type, run_id); _store.update_state(job_id, JobState::running, initial_data);
auto proc = std::make_shared<BackgroundProcess>(); proc->job_id = job_id;
{ std::lock_guard<std::mutex> lk(_procs_mutex); _procs[job_id] = proc; }
proc->thread = std::thread([this, job_id, type, args, cwd, stdin_text, proc, on_complete]() { SubprocessResult res;#ifdef _WIN32 res = run_subprocess(args, cwd, stdin_text, &_limits);#else SpawnedProcess spawned; if (!spawn_subprocess(args, cwd, stdin_text, spawned)) { res.exit_code = -1; res.stderr_str = "failed to spawn subprocess"; } else { proc->pid.store(static_cast<int>(spawned.pid)); _store.set_pid(job_id, static_cast<int>(spawned.pid));
CapturedText stdout_capture; CapturedText stderr_capture; std::thread stdout_thread([&stdout_capture, fd = spawned.stdout_fd, capture_limit = _limits.subprocess_capture_bytes]() { stdout_capture = drain_fd(fd, capture_limit); }); std::thread stderr_thread([&stderr_capture, fd = spawned.stderr_fd, capture_limit = _limits.subprocess_capture_bytes]() { stderr_capture = drain_fd(fd, capture_limit); });
int status = wait_for_process(*proc); stdout_thread.join(); stderr_thread.join(); res.stdout_str = std::move(stdout_capture.text); res.stderr_str = std::move(stderr_capture.text); res.stdout_bytes = stdout_capture.total_bytes; res.stderr_bytes = stderr_capture.total_bytes; res.stdout_truncated = stdout_capture.truncated; res.stderr_truncated = stderr_capture.truncated; if (status >= 0 && WIFEXITED(status)) res.exit_code = WEXITSTATUS(status); else if (status >= 0 && WIFSIGNALED(status)) res.exit_code = 128 + WTERMSIG(status); else res.exit_code = -1; }#endif nlohmann::json data = nlohmann::json::object(); if (auto snapshot = _store.get(job_id); snapshot.has_value() && snapshot->data.is_object()) { data = snapshot->data; } store_process_output(data, "stdout", res.stdout_str, res.stdout_bytes, res.stdout_truncated, _limits); store_process_output(data, "stderr", res.stderr_str, res.stderr_bytes, res.stderr_truncated, _limits); data["exit_code"] = res.exit_code; auto parsed = nlohmann::json::parse(res.stdout_str, nullptr, false); if (!parsed.is_discarded()) { const json compact = (type == "scan") ? compact_scan_job_result(parsed, _limits) : ((type == "scan-metrics") ? compact_scan_metrics_job_result(parsed, _limits) : compact_json_for_job_storage(parsed, _limits)); data["result"] = compact; if (compact.is_object()) { for (auto it = compact.begin(); it != compact.end(); ++it) { if (!data.contains(it.key()) || data[it.key()].is_null()) { data[it.key()] = it.value(); } } } if (type == "scan-metrics") { write_scan_metrics_cache(cwd, job_id, data, compact); } } JobState final_state; if (proc->cancelled.load()) { final_state = JobState::cancelled; _store.update_state(job_id, final_state, data); } else if (res.exit_code == 0) { final_state = JobState::ok; _store.update_state(job_id, final_state, data); } else { final_state = JobState::error; _store.update_state(job_id, final_state, data, res.stderr_str.empty() ? "exit " + std::to_string(res.exit_code) : res.stderr_str.substr(0, 256)); } _store.set_pid(job_id, std::nullopt); if (on_complete) { try { on_complete(job_id, final_state); } catch (...) { // A follow-up hook must never take the worker thread down with it. } } std::lock_guard<std::mutex> lk(_procs_mutex); _procs.erase(job_id); }); proc->thread.detach(); return job_id;}
/// @brief Cancels cancel./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component./// @brief Cancels one tracked background process by job id./// @details Sets the cooperative cancellation flag, terminates the live process group when/// possible, and lets the worker thread finalize the job state.bool SubprocessManager::cancel(const std::string& job_id) { std::lock_guard<std::mutex> lk(_procs_mutex); auto it = _procs.find(job_id); if (it == _procs.end()) return _store.cancel(job_id); auto proc = it->second; _store.cancel(job_id);#ifndef _WIN32 std::thread([proc]() { std::this_thread::sleep_for(std::chrono::milliseconds(250)); proc->cancelled.store(true); if (proc->pid.load() > 0) { terminate_spawned_process(proc->pid.load(), SIGTERM); } }).detach();#else proc->cancelled.store(true);#endif return true;}
/// @brief Cancels by run./// @details This implementation captures subprocess output and coordinates asynchronous cancellable jobs; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component./// @brief Cancels all tracked background processes for a run id./// @details Scans the active process map for matching job metadata and forwards each match to/// the single-job cancellation path.void SubprocessManager::cancel_by_run(const std::string& run_id) { auto jobs = _store.list(500); for (auto& j : jobs) { if (j.run_id == run_id && (j.state == JobState::running || j.state == JobState::pending)) { cancel(j.job_id); } }}