json.hpp

Go to the documentation of this file.

#pragma once
 
 
#include <threadman/config.hpp>
 
#if !THREADMAN_WITH_NLOHMANN_JSON
#error "threadman/json.hpp requires THREADMAN_WITH_NLOHMANN_JSON to be enabled"
#endif
 
#include <threadman/stats.hpp>
 
#include <nlohmann/json.hpp>
 
#include <chrono>
 
namespace threadman {
 
NLOHMANN_JSON_SERIALIZE_ENUM(ThreadState,
                             {
                                 {ThreadState::Starting, "starting"},
                                 {ThreadState::Running, "running"},
                                 {ThreadState::Idle, "idle"},
                                 {ThreadState::Retiring, "retiring"},
                                 {ThreadState::Completed, "completed"},
                                 {ThreadState::Failed, "failed"},
                             })
 
NLOHMANN_JSON_SERIALIZE_ENUM(TaskState,
                             {
                                 {TaskState::Queued, "queued"},
                                 {TaskState::Running, "running"},
                                 {TaskState::Completed, "completed"},
                                 {TaskState::Failed, "failed"},
                                 {TaskState::Cancelled, "cancelled"},
                             })
 
NLOHMANN_JSON_SERIALIZE_ENUM(PoolState,
                             {
                                 {PoolState::Running, "running"},
                                 {PoolState::ShuttingDown, "shutting_down"},
                                 {PoolState::ShutdownNow, "shutdown_now"},
                                 {PoolState::Terminated, "terminated"},
                             })
 
inline void to_json(nlohmann::json& j, const ThreadSnapshot& s) {
    j = nlohmann::json{
        {"id", s.id},
        {"name", s.name},
        {"native_id", s.native_id},
        {"state", s.state},
        {"run_duration_ns", s.run_duration.count()},
        {"failed", s.failed},
        {"is_core", s.is_core},
    };
    if (s.pool_id.has_value()) {
        j["pool_id"] = *s.pool_id;
    }
}
inline void from_json(const nlohmann::json& j, ThreadSnapshot& s) {
    j.at("id").get_to(s.id);
    j.at("name").get_to(s.name);
    j.at("native_id").get_to(s.native_id);
    j.at("state").get_to(s.state);
    long long ns = 0;
    j.at("run_duration_ns").get_to(ns);
    s.run_duration = std::chrono::nanoseconds{ns};
    j.at("failed").get_to(s.failed);
    j.at("is_core").get_to(s.is_core);
    if (j.contains("pool_id")) {
        s.pool_id = j.at("pool_id").get<std::uint64_t>();
    } else {
        s.pool_id.reset();
    }
}
 
inline void to_json(nlohmann::json& j, const ThreadPoolStats& s) {
    j = nlohmann::json{
        {"pool_id", s.pool_id},
        {"name", s.name},
        {"state", s.state},
        {"workers", s.workers},
        {"core_workers", s.core_workers},
        {"max_workers", s.max_workers},
        {"active", s.active},
        {"idle", s.idle},
        {"queued", s.queued},
        {"completed", s.completed},
        {"failed", s.failed},
        {"avg_execution_duration_ns", s.avg_execution_duration.count()},
        {"scale_ups", s.scale_ups},
        {"scale_downs", s.scale_downs},
    };
}
inline void from_json(const nlohmann::json& j, ThreadPoolStats& s) {
    j.at("pool_id").get_to(s.pool_id);
    j.at("name").get_to(s.name);
    j.at("state").get_to(s.state);
    j.at("workers").get_to(s.workers);
    j.at("core_workers").get_to(s.core_workers);
    j.at("max_workers").get_to(s.max_workers);
    j.at("active").get_to(s.active);
    j.at("idle").get_to(s.idle);
    j.at("queued").get_to(s.queued);
    j.at("completed").get_to(s.completed);
    j.at("failed").get_to(s.failed);
    long long ns = 0;
    j.at("avg_execution_duration_ns").get_to(ns);
    s.avg_execution_duration = std::chrono::nanoseconds{ns};
    j.at("scale_ups").get_to(s.scale_ups);
    j.at("scale_downs").get_to(s.scale_downs);
}
 
namespace detail {
 
template <class Clock>
inline long long tp_to_ns(const std::chrono::time_point<Clock>& tp) {
    return std::chrono::duration_cast<std::chrono::nanoseconds>(tp.time_since_epoch()).count();
}
 
template <class Clock>
inline std::chrono::time_point<Clock> ns_to_tp(const long long ns) {
    using TpDuration = typename Clock::duration;
    return std::chrono::time_point<Clock>{
        std::chrono::duration_cast<TpDuration>(std::chrono::nanoseconds{ns})};
}
 
}  // namespace detail
 
inline void to_json(nlohmann::json& j, const TaskSnapshot& s) {
    j = nlohmann::json{
        {"id", s.id},
        {"pool_id", s.pool_id},
        {"state", s.state},
        {"created_at_ns", detail::tp_to_ns(s.created_at)},
        {"failed", s.failed},
    };
    if (s.name.has_value()) {
        j["name"] = *s.name;
    }
    if (s.started_at.has_value()) {
        j["started_at_ns"] = detail::tp_to_ns(*s.started_at);
    }
    if (s.finished_at.has_value()) {
        j["finished_at_ns"] = detail::tp_to_ns(*s.finished_at);
    }
}
inline void from_json(const nlohmann::json& j, TaskSnapshot& s) {
    j.at("id").get_to(s.id);
    j.at("pool_id").get_to(s.pool_id);
    j.at("state").get_to(s.state);
    long long ns = 0;
    j.at("created_at_ns").get_to(ns);
    s.created_at = detail::ns_to_tp<std::chrono::steady_clock>(ns);
    j.at("failed").get_to(s.failed);
    if (j.contains("name")) {
        s.name = j.at("name").get<std::string>();
    } else {
        s.name.reset();
    }
    if (j.contains("started_at_ns")) {
        s.started_at =
            detail::ns_to_tp<std::chrono::steady_clock>(j.at("started_at_ns").get<long long>());
    } else {
        s.started_at.reset();
    }
    if (j.contains("finished_at_ns")) {
        s.finished_at =
            detail::ns_to_tp<std::chrono::steady_clock>(j.at("finished_at_ns").get<long long>());
    } else {
        s.finished_at.reset();
    }
}
 
inline void to_json(nlohmann::json& j, const FutureSnapshot& s) {
    j = nlohmann::json{
        {"ready", s.ready},
        {"has_exception", s.has_exception},
        {"continuation_count", s.continuation_count},
        {"shared", s.shared},
    };
}
inline void from_json(const nlohmann::json& j, FutureSnapshot& s) {
    j.at("ready").get_to(s.ready);
    j.at("has_exception").get_to(s.has_exception);
    j.at("continuation_count").get_to(s.continuation_count);
    j.at("shared").get_to(s.shared);
}
 
inline void to_json(nlohmann::json& j, const ManagerSummary& s) {
    j = nlohmann::json{
        {"wall_clock_ns", detail::tp_to_ns(s.wall_clock)},
        {"live_threads", s.live_threads},
        {"total_pools", s.total_pools},
        {"total_live_workers", s.total_live_workers},
        {"total_queued", s.total_queued},
        {"total_completed", s.total_completed},
        {"total_failed", s.total_failed},
        {"per_pool_stats", s.per_pool_stats},
        {"stuck_tasks", s.stuck_tasks},
    };
}
inline void from_json(const nlohmann::json& j, ManagerSummary& s) {
    long long ns = 0;
    j.at("wall_clock_ns").get_to(ns);
    s.wall_clock = detail::ns_to_tp<std::chrono::system_clock>(ns);
    j.at("live_threads").get_to(s.live_threads);
    j.at("total_pools").get_to(s.total_pools);
    j.at("total_live_workers").get_to(s.total_live_workers);
    j.at("total_queued").get_to(s.total_queued);
    j.at("total_completed").get_to(s.total_completed);
    j.at("total_failed").get_to(s.total_failed);
    j.at("per_pool_stats").get_to(s.per_pool_stats);
    j.at("stuck_tasks").get_to(s.stuck_tasks);
}
 
inline void to_json(nlohmann::json& j, const StuckTaskEvent& s) {
    j = nlohmann::json{
        {"task", s.task},
        {"running_for_ns", s.running_for.count()},
        {"pool_id", s.pool_id},
    };
}
inline void from_json(const nlohmann::json& j, StuckTaskEvent& s) {
    j.at("task").get_to(s.task);
    long long ns = 0;
    j.at("running_for_ns").get_to(ns);
    s.running_for = std::chrono::nanoseconds{ns};
    j.at("pool_id").get_to(s.pool_id);
}
 
}  // namespace threadman