Something went wrong. Try again.
Toolkit for tile-based quality reconstruction of astronomical image stacks
dwarf stacking two-seestar
Something went wrong. Try again.
8.8 kB · 207 lines
C++
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208#include "job_store.hpp"#include "time_utils.hpp"#include <algorithm>#include <sstream>
namespace {
/// Returns a JSON string or null if the string is empty.inline nlohmann::json json_str_or_null(const std::string& s) { return s.empty() ? nlohmann::json(nullptr) : nlohmann::json(s);}
} // namespace
/// @brief Implements job to json./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.nlohmann::json job_to_json(const Job& j) { // Prefer run_id from data payload if present, fall back to j.run_id. nlohmann::json run_id_val = (j.data.is_object() && j.data.contains("run_id")) ? j.data["run_id"] : json_str_or_null(j.run_id); return { {"job_id", j.job_id}, {"type", j.type}, {"run_id", run_id_val}, {"state", job_state_str(j.state)}, {"pid", j.pid.has_value() ? nlohmann::json(*j.pid) : nlohmann::json(nullptr)}, {"exit_code", j.data.is_object() && j.data.contains("exit_code") ? j.data["exit_code"] : nlohmann::json(nullptr)}, {"created_at", json_str_or_null(j.created_at)}, {"updated_at", json_str_or_null(j.updated_at)}, {"started_at", json_str_or_null(j.started_at)}, {"ended_at", json_str_or_null(j.ended_at)}, {"data", j.data}, {"error", j.error_message}, {"progress", j.progress}, };}
/// @brief Creates create./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.std::string InMemoryJobStore::create(const std::string& type, const std::string& run_id) { std::lock_guard<std::mutex> lk(_mutex); const auto epoch_mod = std::chrono::system_clock::now().time_since_epoch().count() % 100000; std::ostringstream oss; oss << "job_" << (++_counter) << "_" << epoch_mod; std::string id = oss.str(); Job j; j.job_id = id; j.type = type; j.run_id = run_id; j.state = JobState::pending; j.created_at = utc_now_iso(); j.updated_at = j.created_at; _jobs[id] = std::move(j); _order.push_back(id); prune_locked(); return id;}
/// @brief Implements configure retention./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.void InMemoryJobStore::configure_retention(size_t max_retained_jobs) { std::lock_guard<std::mutex> lk(_mutex); _max_retained_jobs = std::max<size_t>(1, max_retained_jobs); prune_locked();}
/// @brief Implements get./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.std::optional<Job> InMemoryJobStore::get(const std::string& job_id) const { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return std::nullopt; return it->second;}
bool InMemoryJobStore::update_state(const std::string& job_id, JobState state, const nlohmann::json& data, const std::string& error) { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return false; if (state == JobState::running && it->second.started_at.empty()) { it->second.started_at = utc_now_iso(); } it->second.state = state; if (!data.is_null()) it->second.data = data; if (!error.empty()) it->second.error_message = error; it->second.updated_at = utc_now_iso(); if ((state == JobState::ok || state == JobState::error || state == JobState::cancelled) && it->second.ended_at.empty()) { it->second.ended_at = it->second.updated_at; prune_locked(); } return true;}
/// @brief Implements merge data./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.bool InMemoryJobStore::merge_data(const std::string& job_id, const nlohmann::json& patch) { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return false; if (!patch.is_object()) { it->second.data = patch; } else { if (!it->second.data.is_object()) it->second.data = nlohmann::json::object(); for (auto patch_it = patch.begin(); patch_it != patch.end(); ++patch_it) { it->second.data[patch_it.key()] = patch_it.value(); } } it->second.updated_at = utc_now_iso(); return true;}
/// @brief Updates progress./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.bool InMemoryJobStore::update_progress(const std::string& job_id, double progress) { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return false; it->second.progress = progress; it->second.updated_at = utc_now_iso(); return true;}
/// @brief Implements set pid./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.bool InMemoryJobStore::set_pid(const std::string& job_id, std::optional<int> pid) { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return false; it->second.pid = pid; it->second.updated_at = utc_now_iso(); return true;}
/// @brief Checks whether cancelled./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.bool InMemoryJobStore::is_cancelled(const std::string& job_id) const { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return false; return it->second.state == JobState::cancelled;}
/// @brief Cancels cancel./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.bool InMemoryJobStore::cancel(const std::string& job_id) { std::lock_guard<std::mutex> lk(_mutex); auto it = _jobs.find(job_id); if (it == _jobs.end()) return false; if (it->second.state == JobState::running || it->second.state == JobState::pending) it->second.state = JobState::cancelled; it->second.updated_at = utc_now_iso(); if (it->second.ended_at.empty()) it->second.ended_at = it->second.updated_at; return true;}
/// @brief Lists list./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.std::vector<Job> InMemoryJobStore::list(int limit) const { std::lock_guard<std::mutex> lk(_mutex); std::vector<Job> result; result.reserve(std::min(static_cast<size_t>(limit), _order.size())); int start = (int)_order.size() - limit; if (start < 0) start = 0; for (int i = (int)_order.size() - 1; i >= start; --i) { auto it = _jobs.find(_order[i]); if (it != _jobs.end()) result.push_back(it->second); } return result;}
/// @brief Implements prune locked./// @details This implementation serializes and mutates transient backend job records; it keeps JSON shapes, filesystem/// access, process handling, and error reporting localized to this backend component.void InMemoryJobStore::prune_locked() { while (_order.size() > _max_retained_jobs) { bool removed = false; for (auto it = _order.begin(); it != _order.end(); ++it) { const auto job_it = _jobs.find(*it); if (job_it == _jobs.end()) { _order.erase(it); removed = true; break; }
const JobState state = job_it->second.state; if (state == JobState::pending || state == JobState::running) continue;
_jobs.erase(job_it); _order.erase(it); removed = true; break; } if (!removed) break; }}