thread.hpp

Go to the documentation of this file.

#pragma once
 
 
#include <threadman/config.hpp>
#include <threadman/display.hpp>
#include <threadman/log.hpp>
#include <threadman/metrics.hpp>
#include <threadman/stats.hpp>
 
#include <commons/display_info.hpp>
 
#include <atomic>
#include <chrono>
#include <concepts>
#include <cstdint>
#include <exception>
#include <memory>
#include <mutex>
#include <optional>
#include <stop_token>
#include <string>
#include <thread>
#include <type_traits>
#include <utility>
 
#if THREADMAN_PLATFORM_APPLE || THREADMAN_PLATFORM_LINUX
#include <pthread.h>
#endif
 
namespace threadman {
 
class ThreadManager;  // forward
 
namespace detail {
 
inline std::uint64_t next_thread_id() noexcept {
    static std::atomic<std::uint64_t> counter{1};
    return counter.fetch_add(1, std::memory_order_relaxed);
}
 
inline std::uint64_t current_native_id() noexcept {
#if THREADMAN_PLATFORM_APPLE
    std::uint64_t tid = 0;
    pthread_threadid_np(nullptr, &tid);
    return tid;
#elif THREADMAN_PLATFORM_LINUX
    return static_cast<std::uint64_t>(pthread_self());
#else
    return static_cast<std::uint64_t>(std::hash<std::thread::id>{}(std::this_thread::get_id()));
#endif
}
 
inline void apply_native_name(const std::string& name) noexcept {
#if THREADMAN_PLATFORM_APPLE
    // macOS allows setting only the current thread's name; truncate to 63 chars.
    pthread_setname_np(name.substr(0, 63).c_str());
#elif THREADMAN_PLATFORM_LINUX
    // Linux limits the name to 16 bytes including NUL.
    pthread_setname_np(pthread_self(), name.substr(0, 15).c_str());
#else
    (void)name;
#endif
}
 
}  // namespace detail
 
class ManagedThread {
public:
    struct ControlBlock {
        std::uint64_t id = 0;
        std::string name;
        std::uint64_t native_id = 0;
        std::optional<std::uint64_t> pool_id;
        bool is_core = true;
        std::atomic<ThreadState> state{ThreadState::Starting};
        std::chrono::steady_clock::time_point started_at;
        std::chrono::steady_clock::time_point ended_at;
        std::exception_ptr exception;
        mutable std::mutex mtx;
    };
 
    struct Options {
        std::string name;
        std::optional<std::uint64_t> pool_id;
        bool is_core = true;
        ThreadManager* manager = nullptr;  // null → singleton
    };
 
    template <class Fn, class... Args>
        requires(std::invocable<Fn, std::stop_token, Args...> || std::invocable<Fn, Args...>)
    explicit ManagedThread(Options opts, Fn&& fn, Args&&... args)
        : cb_(std::make_shared<ControlBlock>()), opts_manager_(opts.manager) {
        cb_->id = detail::next_thread_id();
        cb_->name = std::move(opts.name);
        cb_->pool_id = opts.pool_id;
        cb_->is_core = opts.is_core;
 
        register_with_manager();
 
        if constexpr (std::invocable<Fn, std::stop_token, Args...>) {
            jth_ = std::jthread(
                [cb = cb_, fn = std::forward<Fn>(fn), ... captured_args = std::forward<Args>(args)](
                    // forwarded via std::move below; const& would break bodies
                    // taking std::stop_token&&.
                    // NOLINTNEXTLINE(performance-unnecessary-value-param)
                    std::stop_token tok) mutable {
                    enter(cb);
                    try {
                        fn(std::move(tok), std::move(captured_args)...);
                        exit_completed(cb);
                    } catch (...) {
                        exit_failed(cb, std::current_exception());
                    }
                });
        } else {
            jth_ = std::jthread(
                [cb = cb_, fn = std::forward<Fn>(fn), ... captured_args = std::forward<Args>(args)](
                    const std::stop_token&) mutable {
                    enter(cb);
                    try {
                        fn(std::move(captured_args)...);
                        exit_completed(cb);
                    } catch (...) {
                        exit_failed(cb, std::current_exception());
                    }
                });
        }
    }
 
    ManagedThread(const ManagedThread&) = delete;
    ManagedThread& operator=(const ManagedThread&) = delete;
    void unregister_from_manager() noexcept;
    ManagedThread(ManagedThread&&) noexcept = default;
    ManagedThread& operator=(ManagedThread&&) noexcept = default;
 
    ~ManagedThread() {
        unregister_from_manager();
    }
 
    [[nodiscard]] std::uint64_t id() const noexcept {
        return cb_ ? cb_->id : 0;
    }
    [[nodiscard]] const std::string& name() const noexcept {
        static constexpr std::string empty;
        return cb_ ? cb_->name : empty;
    }
    [[nodiscard]] std::uint64_t native_id() const noexcept {
        return cb_ ? cb_->native_id : 0;
    }
    [[nodiscard]] std::optional<std::uint64_t> pool_id() const noexcept {
        return cb_ ? cb_->pool_id : std::nullopt;
    }
    [[nodiscard]] bool is_core() const noexcept {
        return cb_ ? cb_->is_core : true;
    }
    [[nodiscard]] ThreadState state() const noexcept {
        return cb_ ? cb_->state.load(std::memory_order_acquire) : ThreadState::Completed;
    }
    [[nodiscard]] bool failed() const noexcept {
        return state() == ThreadState::Failed;
    }
 
    [[nodiscard]] std::chrono::nanoseconds run_duration() const noexcept {
        if (!cb_) {
            return std::chrono::nanoseconds{0};
        }
        std::scoped_lock guard(cb_->mtx);
        if (cb_->started_at.time_since_epoch().count() == 0) {
            return std::chrono::nanoseconds{0};
        }
        const auto end = cb_->ended_at.time_since_epoch().count() == 0
                             ? std::chrono::steady_clock::now()
                             : cb_->ended_at;
        return end - cb_->started_at;
    }
 
    [[nodiscard]] std::stop_source get_stop_source() noexcept {
        return jth_.get_stop_source();
    }
    [[nodiscard]] std::stop_token get_stop_token() const noexcept {
        return jth_.get_stop_token();
    }
 
    bool request_stop() noexcept {
        return jth_.request_stop();
    }
    [[nodiscard]] bool joinable() const noexcept {
        return jth_.joinable();
    }
    void join() {
        jth_.join();
    }
    void detach() {
        jth_.detach();
    }
 
    [[nodiscard]] std::thread::id native_thread_id() const noexcept {
        return jth_.get_id();
    }
 
    [[nodiscard]] std::shared_ptr<ControlBlock> control_block() const noexcept {
        return cb_;
    }
 
    [[nodiscard]] ThreadSnapshot snapshot() const {
        ThreadSnapshot snap;
        if (!cb_) {
            return snap;
        }
        snap.state = cb_->state.load(std::memory_order_acquire);
        snap.failed = (snap.state == ThreadState::Failed);
        snap.run_duration = run_duration();
        std::scoped_lock guard(cb_->mtx);
        snap.id = cb_->id;
        snap.name = cb_->name;
        snap.native_id = cb_->native_id;
        snap.pool_id = cb_->pool_id;
        snap.is_core = cb_->is_core;
        return snap;
    }
 
    [[nodiscard]] static ControlBlock* current() noexcept {
        return current_cb();
    }
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "ManagedThread",
            .description = "RAII wrapper around std::jthread with lifecycle "
                           "introspection and ThreadManager integration.",
            .icon = comms::Icon::from("mdi:cog-outline"),
        };
        return info;
    }
 
private:
    static ControlBlock*& current_cb() noexcept {
        thread_local ControlBlock* tls = nullptr;
        return tls;
    }
 
    static void enter(const std::shared_ptr<ControlBlock>& cb) noexcept {
        current_cb() = cb.get();
        const auto now = std::chrono::steady_clock::now();
        const auto native = detail::current_native_id();
        if (!cb->name.empty()) {
            detail::apply_native_name(cb->name);
        }
        {
            std::scoped_lock guard(cb->mtx);
            cb->started_at = now;
            cb->native_id = native;
        }
        cb->state.store(ThreadState::Running, std::memory_order_release);
        metrics::threads_created().inc();
        metrics::threads_live().inc();
        if (const auto& lg = ::threadman::log::thread();
            lg && lg->should_log(spdlog::level::debug)) {
            lg->debug("started '{}' (id={}, native={})", cb->name, cb->id, native);
        }
    }
 
    static void exit_completed(const std::shared_ptr<ControlBlock>& cb) noexcept {
        {
            std::scoped_lock guard(cb->mtx);
            cb->ended_at = std::chrono::steady_clock::now();
        }
        cb->state.store(ThreadState::Completed, std::memory_order_release);
        metrics::threads_completed().inc();
        metrics::threads_live().dec();
        current_cb() = nullptr;
    }
 
    static void exit_failed(const std::shared_ptr<ControlBlock>& cb,
                            const std::exception_ptr& ex) noexcept {
        std::string what_msg = "(unknown exception)";
        try {
            if (ex) {
                std::rethrow_exception(ex);
            }
        } catch (const std::exception& e) {
            what_msg = e.what();
        } catch (...) {
            // keep default
        }
        {
            std::scoped_lock guard(cb->mtx);
            cb->ended_at = std::chrono::steady_clock::now();
            cb->exception = ex;
        }
        cb->state.store(ThreadState::Failed, std::memory_order_release);
        metrics::threads_failed().inc();
        metrics::threads_live().dec();
        if (const auto& lg = ::threadman::log::thread()) {
            lg->warn("'{}' failed: {}", cb->name, what_msg);
        }
        current_cb() = nullptr;
    }
 
    void register_with_manager();
    void unregister_from_manager() const noexcept;
 
    std::shared_ptr<ControlBlock> cb_;
    ThreadManager* opts_manager_ = nullptr;
    std::jthread jth_;
};
 
}  // namespace threadman