work_stealing_pool_backend.hpp

Go to the documentation of this file.

#pragma once
 
#include "../callable/bind.hpp"
#include "../callable/move_only_function.hpp"
#include "callbacks.hpp"
#include "deadline.hpp"
#include "shutdown_policy_backend.hpp"
#include "work_stealing_deque.hpp"
#include "worker_context_guard.hpp"
#include "worker_count.hpp"
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <random>
#include <shared_mutex>
#include <string>
#include <system_error>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace threadschedule::detail
{
 
class work_stealing_pool_backend
{
public:
  using task_type = std::function<void()>;
  using queued_task = detail::move_only_function<void()>;
 
  struct statistics
  {
    size_t total_threads;
    size_t active_threads;
    size_t pending_tasks;
    size_t completed_tasks;
    size_t stolen_tasks;
    double tasks_per_second;
    std::chrono::microseconds avg_task_time;
  };
 
  explicit work_stealing_pool_backend(size_t num_threads = default_worker_count(),
                                      size_t deque_capacity = work_stealing_deque<queued_task>::default_capacity,
                                      bool register_workers = false)
      : num_threads_(checked_worker_count(num_threads)), register_workers_(register_workers), stop_(false),
        next_victim_(0), start_time_(std::chrono::steady_clock::now())
  {
    worker_queues_.resize(num_threads_);
    for (size_t i = 0; i < num_threads_; ++i)
      {
        worker_queues_[i] = std::make_unique<work_stealing_deque<queued_task>>(deque_capacity);
      }
 
    workers_.reserve(num_threads_);
 
    try
      {
        for (size_t i = 0; i < num_threads_; ++i)
          workers_.emplace_back(&work_stealing_pool_backend::worker_function, this, i);
        startup_.wait(num_threads_);
      }
    catch (...)
      {
        stop_.store(true, std::memory_order_release);
        submissions_quiesced_.store(true, std::memory_order_release);
        wakeup_condition_.notify_all();
        for (auto& worker : workers_)
          if (worker.joinable())
            worker.join();
        throw;
      }
  }
 
  template <typename Bool, std::enable_if_t<std::is_same_v<std::decay_t<Bool>, bool>, int> = 0>
  work_stealing_pool_backend(size_t num_threads, Bool register_workers)
      : work_stealing_pool_backend(num_threads, work_stealing_deque<queued_task>::default_capacity, register_workers)
  {
  }
 
  work_stealing_pool_backend(work_stealing_pool_backend const&) = delete;
  auto operator=(work_stealing_pool_backend const&) -> work_stealing_pool_backend& = delete;
 
  ~work_stealing_pool_backend()
  {
    shutdown(shutdown_policy_backend::drain);
  }
 
  void
  shutdown(shutdown_policy_backend policy = shutdown_policy_backend::drain)
  {
    if (is_current_worker())
      detail::throw_worker_deadlock();
    std::lock_guard<std::recursive_timed_mutex> shutdown_lock(shutdown_mutex_);
    if (workers_.empty())
      return;
    stop_.store(true, std::memory_order_release);
    bool const completed_all = finish_shutdown(policy);
    shutdown_completed_all_ = shutdown_completed_all_ && completed_all;
    shutdown_completed_at_ = std::chrono::steady_clock::now();
  }
 
  auto
  shutdown_for(std::chrono::milliseconds timeout) -> bool
  {
    if (is_current_worker())
      detail::throw_worker_deadlock();
    auto const deadline = shutdown_deadline_after(timeout);
    std::unique_lock<std::recursive_timed_mutex> shutdown_lock(shutdown_mutex_, std::defer_lock);
    if (deadline == std::chrono::steady_clock::time_point::max())
      shutdown_lock.lock();
    else if (!shutdown_lock.try_lock_until(deadline))
      return false;
 
    if (workers_.empty())
      return shutdown_completed_all_ && shutdown_completed_at_ <= deadline;
 
    stop_.store(true, std::memory_order_release);
    if (!submissions_quiesced_.load(std::memory_order_acquire))
      {
        std::unique_lock<std::shared_timed_mutex> submission_lock(submission_mutex_, std::defer_lock);
        if (deadline == std::chrono::steady_clock::time_point::max())
          submission_lock.lock();
        else if (!submission_lock.try_lock_until(deadline))
          return false;
        submissions_quiesced_.store(true, std::memory_order_release);
      }
    wakeup_condition_.notify_all();
 
    std::unique_lock<std::mutex> lock(completion_mutex_);
    bool const drained = completion_condition_.wait_until(
        lock, deadline, [this] { return outstanding_tasks_.load(std::memory_order_acquire) == 0; });
    lock.unlock();
 
    if (!drained)
      {
        size_t const dropped_tasks = discard_pending_tasks();
        shutdown_completed_all_ = shutdown_completed_all_ && dropped_tasks == 0;
      }
 
    wakeup_condition_.notify_all();
    completion_condition_.notify_all();
 
    if (deadline == std::chrono::steady_clock::time_point::max())
      {
        for (auto& worker : workers_)
          if (worker.joinable())
            worker.join();
        workers_.clear();
      }
    if (drained && shutdown_completed_at_ == std::chrono::steady_clock::time_point{})
      shutdown_completed_at_ = std::chrono::steady_clock::now();
    return drained && shutdown_completed_all_;
  }
 
  template <typename F, typename... Args>
  auto
  try_submit(F&& f, Args&&... args) -> expected<std::future<bind_result_t<F, Args...>>, std::error_code>
  {
    using return_type = bind_result_t<F, Args...>;
 
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        detail::bind_args(std::forward<F>(f), std::forward<Args>(args)...));
 
    std::future<return_type> result = task->get_future();
    queued_task queued([task]() { (*task)(); });
 
    std::shared_lock<std::shared_timed_mutex> submission_lock(submission_mutex_);
 
    if (stop_.load(std::memory_order_acquire))
      return unexpected(std::make_error_code(std::errc::operation_canceled));
 
    size_t const preferred_queue = next_victim_.fetch_add(1, std::memory_order_relaxed) % num_threads_;
 
    outstanding_tasks_.fetch_add(1, std::memory_order_release);
    if (worker_queues_[preferred_queue]->push(std::move(queued)))
      {
        wakeup_condition_.notify_one();
        return result;
      }
    outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
 
    for (size_t attempts = 0; attempts < (std::min)(num_threads_, size_t(3)); ++attempts)
      {
        size_t const idx = (preferred_queue + attempts + 1) % num_threads_;
        outstanding_tasks_.fetch_add(1, std::memory_order_release);
        // A failed push does not consume the queued task.
        // NOLINTNEXTLINE(bugprone-use-after-move)
        if (worker_queues_[idx]->push(std::move(queued)))
          {
            wakeup_condition_.notify_one();
            return result;
          }
        outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
      }
 
    {
      std::lock_guard<std::mutex> lock(overflow_mutex_);
      if (stop_.load(std::memory_order_relaxed))
        return unexpected(std::make_error_code(std::errc::operation_canceled));
      // Failed deque pushes preserve the queued task.
      // NOLINTNEXTLINE(bugprone-use-after-move)
      overflow_tasks_.emplace(std::move(queued));
      outstanding_tasks_.fetch_add(1, std::memory_order_release);
    }
 
    wakeup_condition_.notify_all();
    return result;
  }
 
  template <typename F, typename... Args>
  auto
  submit(F&& f, Args&&... args) -> std::future<bind_result_t<F, Args...>>
  {
    auto result = try_submit(std::forward<F>(f), std::forward<Args>(args)...);
    if (!result.has_value())
      throw std::runtime_error("work_stealing_pool_backend is shutting down");
    return std::move(result.value());
  }
 
  template <typename F, typename... Args>
  void
  post(F&& f, Args&&... args)
  {
    auto r = try_post(std::forward<F>(f), std::forward<Args>(args)...);
    if (!r.has_value())
      throw std::runtime_error("work_stealing_pool_backend is shutting down");
  }
 
  template <typename F, typename... Args>
  auto
  try_post(F&& f, Args&&... args) -> expected<void, std::error_code>
  {
    queued_task bound(
        detail::make_move_only_function<void()>(detail::bind_args(std::forward<F>(f), std::forward<Args>(args)...)));
 
    std::shared_lock<std::shared_timed_mutex> submission_lock(submission_mutex_);
 
    if (stop_.load(std::memory_order_acquire))
      return unexpected(std::make_error_code(std::errc::operation_canceled));
 
    size_t const preferred_queue = next_victim_.fetch_add(1, std::memory_order_relaxed) % num_threads_;
 
    outstanding_tasks_.fetch_add(1, std::memory_order_release);
    if (worker_queues_[preferred_queue]->push(std::move(bound)))
      {
        wakeup_condition_.notify_one();
        return {};
      }
    outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
 
    for (size_t attempts = 0; attempts < (std::min)(num_threads_, size_t(3)); ++attempts)
      {
        size_t const idx = (preferred_queue + attempts + 1) % num_threads_;
        outstanding_tasks_.fetch_add(1, std::memory_order_release);
        // work_stealing_deque::push only moves from the task after it has
        // confirmed capacity; a failed push deliberately preserves it.
        // NOLINTNEXTLINE(bugprone-use-after-move)
        if (worker_queues_[idx]->push(std::move(bound)))
          {
            wakeup_condition_.notify_one();
            return {};
          }
        outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
      }
 
    {
      std::lock_guard<std::mutex> lock(overflow_mutex_);
      if (stop_.load(std::memory_order_relaxed))
        return unexpected(std::make_error_code(std::errc::operation_canceled));
      // Failed deque pushes preserve the queued task.
      // NOLINTNEXTLINE(bugprone-use-after-move)
      overflow_tasks_.emplace(std::move(bound));
      outstanding_tasks_.fetch_add(1, std::memory_order_release);
    }
 
    wakeup_condition_.notify_all();
    return {};
  }
 
  template <typename Iterator>
  auto
  try_submit_batch(Iterator begin, Iterator end) -> expected<std::vector<std::future<void>>, std::error_code>
  {
    std::vector<std::future<void>> futures;
    size_t const batch_size_hint = detail::multipass_range_size(begin, end);
    futures.reserve(batch_size_hint);
    std::vector<queued_task> prepared;
    prepared.reserve(batch_size_hint);
 
    for (auto it = begin; it != end; ++it)
      {
        auto task = std::make_shared<std::packaged_task<void()>>(*it);
        futures.push_back(task->get_future());
        prepared.emplace_back([task]() { (*task)(); });
      }
 
    size_t const batch_size = prepared.size();
 
    std::shared_lock<std::shared_timed_mutex> submission_lock(submission_mutex_);
 
    if (stop_.load(std::memory_order_acquire))
      return unexpected(std::make_error_code(std::errc::operation_canceled));
 
    size_t queue_idx = next_victim_.fetch_add(batch_size, std::memory_order_relaxed) % num_threads_;
 
    try
      {
        for (auto& queued : prepared)
          {
            bool enqueued = false;
            for (size_t attempts = 0; attempts < num_threads_; ++attempts)
              {
                outstanding_tasks_.fetch_add(1, std::memory_order_release);
                // A failed push does not consume the queued task.
                // NOLINTNEXTLINE(bugprone-use-after-move)
                if (worker_queues_[queue_idx]->push(std::move(queued)))
                  {
                    enqueued = true;
                    queue_idx = (queue_idx + 1) % num_threads_;
                    break;
                  }
                outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
                queue_idx = (queue_idx + 1) % num_threads_;
              }
 
            if (!enqueued)
              {
                std::lock_guard<std::mutex> lock(overflow_mutex_);
                // Failed deque pushes preserve the queued task.
                // NOLINTNEXTLINE(bugprone-use-after-move)
                overflow_tasks_.emplace(std::move(queued));
                outstanding_tasks_.fetch_add(1, std::memory_order_release);
              }
          }
      }
    catch (...)
      {
        wakeup_condition_.notify_all();
        throw;
      }
 
    wakeup_condition_.notify_all();
    return futures;
  }
 
  template <typename Iterator>
  auto
  submit_batch(Iterator begin, Iterator end) -> std::vector<std::future<void>>
  {
    auto result = try_submit_batch(begin, end);
    if (!result.has_value())
      throw std::runtime_error("work_stealing_pool_backend is shutting down");
    return std::move(result.value());
  }
 
  template <typename Iterator, typename F>
  void
  parallel_for_each(Iterator begin, Iterator end, F&& func)
  {
    if (is_current_worker())
      detail::throw_worker_deadlock();
    detail::parallel_for_each_chunked(*this, begin, end, std::forward<F>(func), num_threads_);
  }
 
 
  [[nodiscard]] auto
  size() const noexcept -> size_t
  {
    return num_threads_;
  }
 
  [[nodiscard]] auto
  pending_tasks() const -> size_t
  {
    size_t total = 0;
    for (auto const& queue : worker_queues_)
      {
        total += queue->size();
      }
 
    std::lock_guard<std::mutex> lock(overflow_mutex_);
    total += overflow_tasks_.size();
    return total;
  }
 
  auto
  get_statistics() const -> statistics
  {
    auto const now = std::chrono::steady_clock::now();
    auto const elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - start_time_);
 
    statistics stats;
    stats.total_threads = num_threads_;
    stats.active_threads = active_tasks_.load(std::memory_order_acquire);
    stats.pending_tasks = pending_tasks();
    stats.completed_tasks = completed_tasks_.load(std::memory_order_acquire);
    stats.stolen_tasks = stolen_tasks_.load(std::memory_order_acquire);
 
    if (elapsed.count() > 0)
      {
        stats.tasks_per_second = static_cast<double>(stats.completed_tasks) / elapsed.count();
      }
    else
      {
        stats.tasks_per_second = 0.0;
      }
 
    auto const total_task_time = total_task_time_.load(std::memory_order_acquire);
    if (stats.completed_tasks > 0)
      {
        stats.avg_task_time = std::chrono::microseconds(total_task_time / stats.completed_tasks);
      }
    else
      {
        stats.avg_task_time = std::chrono::microseconds(0);
      }
 
    return stats;
  }
 
 
 
  auto
  configure_threads(std::string const& name_prefix, native_scheduling_policy policy = native_scheduling_policy::other,
                    native_thread_priority priority = native_thread_priority::normal())
      -> expected<void, std::error_code>
  {
    return detail::configure_worker_threads(workers_, name_prefix, policy, priority,
                                            register_workers_ ? &runtime_registry() : nullptr);
  }
 
  auto
  configure_threads(native_thread_config const& config) -> expected<void, std::error_code>
  {
    return detail::configure_worker_threads(workers_, config, register_workers_ ? &runtime_registry() : nullptr);
  }
 
  auto
  set_affinity(native_thread_affinity const& affinity) -> expected<void, std::error_code>
  {
    return detail::set_worker_affinity(workers_, affinity);
  }
 
  auto
  distribute_across_cpus() -> expected<void, std::error_code>
  {
    return detail::distribute_workers_across_cpus(workers_);
  }
 
 
 
  void
  wait_for_tasks()
  {
    if (is_current_worker())
      detail::throw_worker_deadlock();
    std::unique_lock<std::mutex> lock(completion_mutex_);
    completion_condition_.wait(lock, [this] { return outstanding_tasks_.load(std::memory_order_acquire) == 0; });
  }
 
  [[nodiscard]] auto
  is_current_worker() const noexcept -> bool
  {
    return current_pool == this;
  }
 
 
 
  void
  set_on_task_start(task_start_callback cb)
  {
    std::lock_guard<std::mutex> lock(trace_mutex_);
    on_task_start_ = task_start_callback_storage(std::move(cb));
  }
 
  template <typename Callback,
            std::enable_if_t<!std::is_same_v<detail::remove_cvref_t<Callback>, task_start_callback>, int> = 0>
  void
  set_on_task_start(Callback&& cb)
  {
    static_assert(std::is_invocable_r_v<void, Callback&, std::chrono::steady_clock::time_point, std::thread::id>,
                  "Task start callback must accept (time_point, std::thread::id)");
    std::lock_guard<std::mutex> lock(trace_mutex_);
    on_task_start_ = detail::make_copyable_function<void(std::chrono::steady_clock::time_point, std::thread::id)>(
        std::forward<Callback>(cb));
  }
 
  void
  set_on_task_end(task_end_callback cb)
  {
    std::lock_guard<std::mutex> lock(trace_mutex_);
    on_task_end_ = task_end_callback_storage(std::move(cb));
  }
 
  template <typename Callback,
            std::enable_if_t<!std::is_same_v<detail::remove_cvref_t<Callback>, task_end_callback>, int> = 0>
  void
  set_on_task_end(Callback&& cb)
  {
    static_assert(std::is_invocable_r_v<void, Callback&, std::chrono::steady_clock::time_point, std::thread::id,
                                        std::chrono::microseconds>,
                  "Task end callback must accept (time_point, std::thread::id, "
                  "std::chrono::microseconds)");
    std::lock_guard<std::mutex> lock(trace_mutex_);
    on_task_end_ = detail::make_copyable_function<void(std::chrono::steady_clock::time_point, std::thread::id,
                                                       std::chrono::microseconds)>(std::forward<Callback>(cb));
  }
 
 
private:
  size_t num_threads_;
  bool register_workers_;
  detail::worker_startup_latch startup_;
  std::vector<detail::thread_backend> workers_;
  std::vector<std::unique_ptr<work_stealing_deque<queued_task>>> worker_queues_;
 
  std::queue<queued_task> overflow_tasks_;
  mutable std::mutex overflow_mutex_;
  mutable std::shared_timed_mutex submission_mutex_;
  std::recursive_timed_mutex shutdown_mutex_;
 
  std::atomic<bool> stop_;
  std::atomic<bool> submissions_quiesced_{ false };
  bool shutdown_completed_all_{ true };
  std::chrono::steady_clock::time_point shutdown_completed_at_{};
  std::condition_variable wakeup_condition_;
  std::mutex wakeup_mutex_;
 
  std::condition_variable completion_condition_;
  std::mutex completion_mutex_;
 
  std::atomic<size_t> next_victim_;
  std::atomic<size_t> active_tasks_{ 0 };
  std::atomic<size_t> outstanding_tasks_{ 0 };
  std::atomic<size_t> completed_tasks_{ 0 };
  std::atomic<size_t> stolen_tasks_{ 0 };
  std::atomic<uint64_t> total_task_time_{ 0 };
 
  std::mutex trace_mutex_;
  task_start_callback_storage on_task_start_;
  task_end_callback_storage on_task_end_;
 
  std::chrono::steady_clock::time_point start_time_;
  inline static thread_local work_stealing_pool_backend* current_pool = nullptr;
 
  auto
  discard_pending_tasks() -> size_t
  {
    size_t dropped_tasks = 0;
    std::queue<queued_task> discarded_overflow;
    {
      std::lock_guard<std::mutex> lock(overflow_mutex_);
      dropped_tasks += overflow_tasks_.size();
      overflow_tasks_.swap(discarded_overflow);
    }
 
    if (dropped_tasks != 0)
      {
        std::lock_guard<std::mutex> lock(completion_mutex_);
        outstanding_tasks_.fetch_sub(dropped_tasks, std::memory_order_acq_rel);
      }
 
    if (dropped_tasks != 0)
      completion_condition_.notify_all();
 
    if (dropped_tasks != 0)
      {
        std::queue<queued_task> empty;
        discarded_overflow.swap(empty);
      }
 
    for (auto& queue : worker_queues_)
      {
        queued_task discarded;
        while (queue->steal(discarded))
          {
            ++dropped_tasks;
            {
              std::lock_guard<std::mutex> lock(completion_mutex_);
              outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
            }
            completion_condition_.notify_all();
            discarded = queued_task{};
          }
      }
 
    return dropped_tasks;
  }
 
  auto
  finish_shutdown(shutdown_policy_backend policy) -> bool
  {
    {
      std::unique_lock<std::shared_timed_mutex> submission_lock(submission_mutex_);
      submissions_quiesced_.store(true, std::memory_order_release);
    }
 
    size_t const dropped_tasks = policy == shutdown_policy_backend::drop_pending ? discard_pending_tasks() : size_t(0);
 
    wakeup_condition_.notify_all();
 
    for (auto& worker : workers_)
      if (worker.joinable())
        worker.join();
 
    workers_.clear();
    return dropped_tasks == 0;
  }
 
  // NOLINTNEXTLINE(readability-function-cognitive-complexity)
  void
  worker_function(size_t worker_id)
  {
    detail::worker_context_guard<work_stealing_pool_backend> worker_context(current_pool, this);
    std::optional<registration_guard_backend> reg_guard;
    try
      {
        if (register_workers_)
          reg_guard.emplace("hp_worker_" + std::to_string(worker_id), "threadschedule.pool");
        startup_.arrive();
      }
    catch (...)
      {
        startup_.arrive(std::current_exception());
        return;
      }
 
    thread_local std::mt19937 gen = [worker_id]()
      {
        try
          {
            std::random_device device;
            return std::mt19937(device());
          }
        catch (...)
          {
            auto const seed = static_cast<std::mt19937::result_type>(
                std::mt19937::default_seed ^ static_cast<std::mt19937::result_type>(worker_id));
            return std::mt19937(seed);
          }
      }();
 
    queued_task task;
    std::uniform_int_distribution<size_t> dist(0, num_threads_ - 1);
 
    while (true)
      {
        bool found_task = false;
 
        if (worker_queues_[worker_id]->pop(task))
          {
            found_task = true;
          }
        else
          {
            size_t const max_steal_attempts = (std::min)(num_threads_, size_t(4));
            for (size_t attempts = 0; attempts < max_steal_attempts; ++attempts)
              {
                size_t const victim_id = dist(gen);
                if (victim_id != worker_id && worker_queues_[victim_id]->steal(task))
                  {
                    found_task = true;
                    stolen_tasks_.fetch_add(1, std::memory_order_relaxed);
                    break;
                  }
              }
          }
 
        if (!found_task)
          {
            std::lock_guard<std::mutex> lock(overflow_mutex_);
            if (!overflow_tasks_.empty())
              {
                task = std::move(overflow_tasks_.front());
                overflow_tasks_.pop();
                found_task = true;
              }
          }
 
        if (found_task)
          {
            active_tasks_.fetch_add(1, std::memory_order_relaxed);
 
            auto const start_time = std::chrono::steady_clock::now();
            auto const tid = std::this_thread::get_id();
 
            try
              {
                task_start_callback_storage on_task_start;
                {
                  std::lock_guard<std::mutex> tl(trace_mutex_);
                  on_task_start = on_task_start_;
                }
                if (on_task_start)
                  on_task_start(start_time, tid);
              }
            catch (...)
              {
              }
 
            // For submit() tasks the callable is a packaged_task which
            // catches exceptions internally and stores them in the
            // std::future shared state - those never reach this catch.
            // For post() tasks (fire-and-forget) the catch prevents an
            // unhandled exception from terminating the worker thread.
            try
              {
                task();
              }
            catch (...)
              {
              }
            task = queued_task{};
            auto const end_time = std::chrono::steady_clock::now();
 
            auto const task_duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
            total_task_time_.fetch_add(task_duration.count(), std::memory_order_relaxed);
 
            try
              {
                task_end_callback_storage on_task_end;
                {
                  std::lock_guard<std::mutex> tl(trace_mutex_);
                  on_task_end = on_task_end_;
                }
                if (on_task_end)
                  on_task_end(end_time, tid, task_duration);
              }
            catch (...)
              {
              }
 
            active_tasks_.fetch_sub(1, std::memory_order_relaxed);
            {
              std::lock_guard<std::mutex> lock(completion_mutex_);
              outstanding_tasks_.fetch_sub(1, std::memory_order_acq_rel);
            }
            completed_tasks_.fetch_add(1, std::memory_order_relaxed);
 
            completion_condition_.notify_all();
            wakeup_condition_.notify_all();
          }
        else
          {
            if (stop_.load(std::memory_order_acquire) && submissions_quiesced_.load(std::memory_order_acquire)
                && outstanding_tasks_.load(std::memory_order_acquire) == 0)
              {
                break;
              }
 
            std::unique_lock<std::mutex> lock(wakeup_mutex_);
            wakeup_condition_.wait_for(lock, std::chrono::microseconds(100));
          }
      }
  }
};
 
} // namespace threadschedule::detail