thread_registry.hpp

Go to the documentation of this file.

#pragma once

 
#include "expected.hpp"
#include "scheduler_policy.hpp"
#include "thread_wrapper.hpp" // for ThreadInfo, ThreadAffinity
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <string>
#include <thread>
#include <tuple>
#include <unordered_map>
#include <utility>
#include <vector>
 
#ifdef _WIN32
#include <windows.h>
#else
#include <pthread.h>
#include <sched.h>
#include <sys/types.h>
#endif
 
namespace threadschedule
{
 
// Optional export macro for building a runtime (shared/dll) variant
#if defined(_WIN32) || defined(_WIN64)
#if defined(THREADSCHEDULE_EXPORTS)
#define THREADSCHEDULE_API __declspec(dllexport)
#else
#define THREADSCHEDULE_API __declspec(dllimport)
#endif
#else
#define THREADSCHEDULE_API
#endif
 
#ifdef _WIN32
using Tid = unsigned long; // DWORD thread id
#else
using Tid = pid_t; // Linux TID via gettid()
#endif

struct RegisteredThreadInfo
{
    Tid tid{};
    std::thread::id stdId;
    std::string name;
    std::string componentTag;
    bool alive{true};
    std::shared_ptr<class ThreadControlBlock> control;
};

class ThreadControlBlock
{
  public:
    ThreadControlBlock() = default;
    ThreadControlBlock(ThreadControlBlock const&) = delete;
    auto operator=(ThreadControlBlock const&) -> ThreadControlBlock& = delete;
    ThreadControlBlock(ThreadControlBlock&&) = delete;
    auto operator=(ThreadControlBlock&&) -> ThreadControlBlock& = delete;
 
    ~ThreadControlBlock()
    {
#ifdef _WIN32
        if (handle_)
        {
            CloseHandle(handle_);
            handle_ = nullptr;
        }
#endif
    }
 
    [[nodiscard]] auto tid() const noexcept -> Tid
    {
        return tid_;
    }
    [[nodiscard]] auto std_id() const noexcept -> std::thread::id
    {
        return stdId_;
    }
  private:
    [[nodiscard]] auto native_handle() const
    {
#ifdef _WIN32
        return handle_;
#else
        return pthreadHandle_;
#endif
    }
 
  public:
    [[nodiscard]] auto set_affinity(ThreadAffinity const& affinity) const -> expected<void, std::error_code>
    {
        return detail::apply_affinity(native_handle(), affinity);
    }
 
    [[nodiscard]] auto set_priority(ThreadPriority priority) const -> expected<void, std::error_code>
    {
        return detail::apply_priority(native_handle(), priority);
    }
 
    [[nodiscard]] auto set_scheduling_policy(SchedulingPolicy policy, ThreadPriority priority) const
        -> expected<void, std::error_code>
    {
        return detail::apply_scheduling_policy(native_handle(), policy, priority);
    }
 
    [[nodiscard]] auto set_name(std::string const& name) const -> expected<void, std::error_code>
    {
        return detail::apply_name(native_handle(), name);
    }
 
    static auto create_for_current_thread() -> std::shared_ptr<ThreadControlBlock>
    {
        auto block = std::make_shared<ThreadControlBlock>();
        block->tid_ = ThreadInfo::get_thread_id();
        block->stdId_ = std::this_thread::get_id();
#ifdef _WIN32
        HANDLE realHandle = nullptr;
        DuplicateHandle(GetCurrentProcess(), GetCurrentThread(), GetCurrentProcess(), &realHandle,
                        THREAD_SET_INFORMATION | THREAD_QUERY_INFORMATION, FALSE, 0);
        block->handle_ = realHandle;
#else
        block->pthreadHandle_ = pthread_self();
#endif
        return block;
    }
 
  private:
    Tid tid_{};
    std::thread::id stdId_;
#ifdef _WIN32
    HANDLE handle_ = nullptr;
#else
    pthread_t pthreadHandle_{};
#endif
};
 
namespace detail
{

template <typename Derived>
class QueryFacadeMixin
{
    auto self() const -> Derived const& { return static_cast<Derived const&>(*this); }
 
  public:
    template <typename Predicate>
    [[nodiscard]] auto filter(Predicate&& pred) const
    {
        return self().query().filter(std::forward<Predicate>(pred));
    }
 
    [[nodiscard]] auto count() const -> size_t { return self().query().count(); }
 
    [[nodiscard]] auto empty() const -> bool { return self().query().empty(); }
 
    template <typename Fn>
    void for_each(Fn&& fn) const
    {
        self().query().for_each(std::forward<Fn>(fn));
    }
 
    template <typename Predicate, typename Fn>
    void apply(Predicate&& pred, Fn&& fn) const
    {
        self().query().filter(std::forward<Predicate>(pred)).for_each(std::forward<Fn>(fn));
    }
 
    template <typename Fn>
    [[nodiscard]] auto map(Fn&& fn) const -> std::vector<std::invoke_result_t<Fn, RegisteredThreadInfo const&>>
    {
        return self().query().map(std::forward<Fn>(fn));
    }
 
    template <typename Predicate>
    [[nodiscard]] auto find_if(Predicate&& pred) const -> std::optional<RegisteredThreadInfo>
    {
        return self().query().find_if(std::forward<Predicate>(pred));
    }
 
    template <typename Predicate>
    [[nodiscard]] auto any(Predicate&& pred) const -> bool
    {
        return self().query().any(std::forward<Predicate>(pred));
    }
 
    template <typename Predicate>
    [[nodiscard]] auto all(Predicate&& pred) const -> bool
    {
        return self().query().all(std::forward<Predicate>(pred));
    }
 
    template <typename Predicate>
    [[nodiscard]] auto none(Predicate&& pred) const -> bool
    {
        return self().query().none(std::forward<Predicate>(pred));
    }
 
    [[nodiscard]] auto take(size_t n) const { return self().query().take(n); }
 
    [[nodiscard]] auto skip(size_t n) const { return self().query().skip(n); }
};
 
} // namespace detail

class ThreadRegistry : public detail::QueryFacadeMixin<ThreadRegistry>
{
  public:
    ThreadRegistry() = default;
    ThreadRegistry(ThreadRegistry const&) = delete;
    auto operator=(ThreadRegistry const&) -> ThreadRegistry& = delete;
 
    void register_current_thread(std::string name = std::string(), std::string componentTag = std::string())
    {
        RegisteredThreadInfo info;
        info.tid = ThreadInfo::get_thread_id();
        info.stdId = std::this_thread::get_id();
        info.name = std::move(name);
        info.componentTag = std::move(componentTag);
        info.alive = true;
        try_register(std::move(info));
    }
 
    void register_current_thread(std::shared_ptr<ThreadControlBlock> const& controlBlock,
                                 std::string name = std::string(), std::string componentTag = std::string())
    {
        if (!controlBlock)
            return;
        RegisteredThreadInfo info;
        info.tid = controlBlock->tid();
        info.stdId = controlBlock->std_id();
        info.name = std::move(name);
        info.componentTag = std::move(componentTag);
        info.alive = true;
        info.control = controlBlock;
        try_register(std::move(info));
    }
 
    void unregister_current_thread()
    {
        Tid const tid = ThreadInfo::get_thread_id();
        std::unique_lock<std::shared_mutex> lock(mutex_);
        auto it = threads_.find(tid);
        if (it != threads_.end())
        {
            it->second.alive = false;
            auto info = it->second;
            threads_.erase(it);
            if (onUnregister_)
            {
                auto cb = onUnregister_;
                lock.unlock();
                cb(info);
            }
        }
    }
 
    // Lookup
    [[nodiscard]] auto get(Tid tid) const -> std::optional<RegisteredThreadInfo>
    {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        auto it = threads_.find(tid);
        if (it == threads_.end())
            return std::nullopt;
        return it->second;
    }

    class QueryView
    {
      public:
        explicit QueryView(std::vector<RegisteredThreadInfo> entries) : entries_(std::move(entries))
        {
        }
 
        template <typename Predicate>
        auto filter(Predicate&& pred) const -> QueryView
        {
            std::vector<RegisteredThreadInfo> filtered;
            filtered.reserve(entries_.size());
            for (auto const& entry : entries_)
            {
                if (pred(entry))
                    filtered.push_back(entry);
            }
            return QueryView(std::move(filtered));
        }
 
        template <typename Fn>
        void for_each(Fn&& fn) const
        {
            for (auto const& entry : entries_)
            {
                fn(entry);
            }
        }
 
        [[nodiscard]] auto count() const -> size_t
        {
            return entries_.size();
        }
 
        [[nodiscard]] auto empty() const -> bool
        {
            return entries_.empty();
        }
 
        [[nodiscard]] auto entries() const -> std::vector<RegisteredThreadInfo> const&
        {
            return entries_;
        }
 
        // Transform entries to a vector of another type
        template <typename Fn>
        [[nodiscard]] auto map(Fn&& fn) const -> std::vector<std::invoke_result_t<Fn, RegisteredThreadInfo const&>>
        {
            std::vector<std::invoke_result_t<Fn, RegisteredThreadInfo const&>> result;
            result.reserve(entries_.size());
            for (auto const& entry : entries_)
            {
                result.push_back(fn(entry));
            }
            return result;
        }
 
        // Find first entry matching predicate
        template <typename Predicate>
        [[nodiscard]] auto find_if(Predicate&& pred) const -> std::optional<RegisteredThreadInfo>
        {
            for (auto const& entry : entries_)
            {
                if (pred(entry))
                    return entry;
            }
            return std::nullopt;
        }
 
        template <typename Predicate>
        [[nodiscard]] auto any(Predicate&& pred) const -> bool
        {
            for (auto const& entry : entries_)
            {
                if (pred(entry))
                    return true;
            }
            return false;
        }
 
        template <typename Predicate>
        [[nodiscard]] auto all(Predicate&& pred) const -> bool
        {
            for (auto const& entry : entries_)
            {
                if (!pred(entry))
                    return false;
            }
            return true;
        }
 
        template <typename Predicate>
        [[nodiscard]] auto none(Predicate&& pred) const -> bool
        {
            return !any(std::forward<Predicate>(pred));
        }
 
        [[nodiscard]] auto take(size_t n) const -> QueryView
        {
            auto result = entries_;
            if (result.size() > n)
                result.resize(n);
            return QueryView(std::move(result));
        }
 
        [[nodiscard]] auto skip(size_t n) const -> QueryView
        {
            std::vector<RegisteredThreadInfo> result;
            if (n < entries_.size())
            {
                result.assign(entries_.begin() + n, entries_.end());
            }
            return QueryView(std::move(result));
        }
 
      private:
        std::vector<RegisteredThreadInfo> entries_;
    };
 
    // Create a query view over all registered threads
    [[nodiscard]] auto query() const -> QueryView
    {
        std::vector<RegisteredThreadInfo> snapshot;
        std::shared_lock<std::shared_mutex> lock(mutex_);
        snapshot.reserve(threads_.size());
        for (auto const& kv : threads_)
        {
            snapshot.push_back(kv.second);
        }
        return QueryView(std::move(snapshot));
    }
 
    [[nodiscard]] auto set_affinity(Tid tid, ThreadAffinity const& affinity) const -> expected<void, std::error_code>
    {
        auto blk = lock_block(tid);
        if (!blk)
            return unexpected(std::make_error_code(std::errc::no_such_process));
        return blk->set_affinity(affinity);
    }
 
    [[nodiscard]] auto set_priority(Tid tid, ThreadPriority priority) const -> expected<void, std::error_code>
    {
        auto blk = lock_block(tid);
        if (!blk)
            return unexpected(std::make_error_code(std::errc::no_such_process));
        return blk->set_priority(priority);
    }
 
    [[nodiscard]] auto set_scheduling_policy(Tid tid, SchedulingPolicy policy, ThreadPriority priority) const
        -> expected<void, std::error_code>
    {
        auto blk = lock_block(tid);
        if (!blk)
            return unexpected(std::make_error_code(std::errc::no_such_process));
        return blk->set_scheduling_policy(policy, priority);
    }
 
    [[nodiscard]] auto set_name(Tid tid, std::string const& name) const -> expected<void, std::error_code>
    {
        auto blk = lock_block(tid);
        if (!blk)
            return unexpected(std::make_error_code(std::errc::no_such_process));
        return blk->set_name(name);
    }
 
    // Register/unregister hooks (system integration)
    void set_on_register(std::function<void(RegisteredThreadInfo const&)> cb)
    {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        onRegister_ = std::move(cb);
    }
 
    void set_on_unregister(std::function<void(RegisteredThreadInfo const&)> cb)
    {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        onUnregister_ = std::move(cb);
    }
 
  private:
    void try_register(RegisteredThreadInfo info)
    {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        auto it = threads_.find(info.tid);
        if (it != threads_.end())
            return;
        auto stored = info;
        threads_.emplace(info.tid, std::move(info));
        if (onRegister_)
        {
            auto cb = onRegister_;
            lock.unlock();
            cb(stored);
        }
    }
 
    [[nodiscard]] auto lock_block(Tid tid) const -> std::shared_ptr<ThreadControlBlock>
    {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        auto it = threads_.find(tid);
        if (it == threads_.end())
            return nullptr;
        return it->second.control;
    }
    mutable std::shared_mutex mutex_;
    std::unordered_map<Tid, RegisteredThreadInfo> threads_;
 
    std::function<void(RegisteredThreadInfo const&)> onRegister_;
    std::function<void(RegisteredThreadInfo const&)> onUnregister_;
};

 
#if defined(THREADSCHEDULE_RUNTIME)
THREADSCHEDULE_API auto registry() -> ThreadRegistry&;
THREADSCHEDULE_API void set_external_registry(ThreadRegistry* reg);
#else
inline auto registry_storage() -> ThreadRegistry*&
{
    static ThreadRegistry* external = nullptr;
    return external;
}

inline auto registry() -> ThreadRegistry&
{
    ThreadRegistry*& ext = registry_storage();
    if (ext != nullptr)
        return *ext;
    static ThreadRegistry local;
    return local;
}

inline void set_external_registry(ThreadRegistry* reg)
{
    registry_storage() = reg;
}

#endif

enum class BuildMode : std::uint8_t
{
    HEADER_ONLY, 
    RUNTIME      
};
 
#if defined(THREADSCHEDULE_RUNTIME)
inline constexpr bool is_runtime_build = true; 

THREADSCHEDULE_API auto build_mode() -> BuildMode;
#else
inline constexpr bool is_runtime_build = false; 

inline auto build_mode() -> BuildMode
{
    return BuildMode::HEADER_ONLY;
}
#endif

inline auto build_mode_string() -> char const*
{
    return is_runtime_build ? "runtime" : "header-only";
}

class CompositeThreadRegistry : public detail::QueryFacadeMixin<CompositeThreadRegistry>
{
  public:
    void attach(ThreadRegistry* reg)
    {
        if (reg == nullptr)
            return;
        std::lock_guard<std::mutex> lock(mutex_);
        registries_.push_back(reg);
    }
 
    [[nodiscard]] auto query() const -> ThreadRegistry::QueryView
    {
        std::vector<RegisteredThreadInfo> merged;
        std::vector<ThreadRegistry*> regs;
        {
            std::lock_guard<std::mutex> lock(mutex_);
            regs = registries_;
        }
        for (auto* r : regs)
        {
            auto view = r->query();
            auto const& entries = view.entries();
            merged.insert(merged.end(), entries.begin(), entries.end());
        }
        return ThreadRegistry::QueryView(std::move(merged));
    }
 
  private:
    mutable std::mutex mutex_;
    std::vector<ThreadRegistry*> registries_;
};

class AutoRegisterCurrentThread
{
  public:
    explicit AutoRegisterCurrentThread(std::string const& name = std::string(),
                                       std::string const& componentTag = std::string())
        : active_(true), externalReg_(nullptr)
    {
        auto block = ThreadControlBlock::create_for_current_thread();
        (void)block->set_name(name);
        registry().register_current_thread(block, name, componentTag);
    }
 
    explicit AutoRegisterCurrentThread(ThreadRegistry& reg, std::string const& name = std::string(),
                                       std::string const& componentTag = std::string())
        : active_(true), externalReg_(&reg)
    {
        auto block = ThreadControlBlock::create_for_current_thread();
        (void)block->set_name(name);
        externalReg_->register_current_thread(block, name, componentTag);
    }
    ~AutoRegisterCurrentThread()
    {
        if (active_)
        {
            if (externalReg_ != nullptr)
                externalReg_->unregister_current_thread();
            else
                registry().unregister_current_thread();
        }
    }
    AutoRegisterCurrentThread(AutoRegisterCurrentThread const&) = delete;
    auto operator=(AutoRegisterCurrentThread const&) -> AutoRegisterCurrentThread& = delete;
    AutoRegisterCurrentThread(AutoRegisterCurrentThread&& other) noexcept
        : active_(other.active_), externalReg_(other.externalReg_)
    {
        other.active_ = false;
        other.externalReg_ = nullptr;
    }
    auto operator=(AutoRegisterCurrentThread&& other) noexcept -> AutoRegisterCurrentThread&
    {
        if (this != &other)
        {
            if (active_)
            {
                if (externalReg_ != nullptr)
                    externalReg_->unregister_current_thread();
                else
                    registry().unregister_current_thread();
            }
            active_ = other.active_;
            externalReg_ = other.externalReg_;
            other.active_ = false;
            other.externalReg_ = nullptr;
        }
        return *this;
    }
 
  private:
    bool active_;
    ThreadRegistry* externalReg_;
};
 
} // namespace threadschedule
 
#ifndef _WIN32
namespace threadschedule
{
inline auto cgroup_attach_tid(std::string const& cgroupDir, Tid tid) -> expected<void, std::error_code>
{
    std::vector<std::string> candidates = {"cgroup.threads", "tasks", "cgroup.procs"};
    for (auto const& file : candidates)
    {
        std::string path = cgroupDir + "/" + file;
        std::ofstream out(path);
        if (!out)
            continue;
        out << tid;
        out.flush();
        if (out)
            return {};
    }
    return unexpected(std::make_error_code(std::errc::operation_not_permitted));
}
} // namespace threadschedule
#endif