asio 调度器实现 - timer 实现详解

template <typename Time_Traits>
class timer_queue
  : public timer_queue_base
{
public:
  // The time type.
  using time_type = typename Time_Traits::time_type;

  // The duration type.
  using duration_type = typename Time_Traits::duration_type;

  // Per-timer data.
  class per_timer_data
  {
  public:
    per_timer_data() :
      heap_index_((std::numeric_limits<std::size_t>::max)()),
      next_(0), prev_(0)
    {
    }
  private:
    friend class timer_queue;

    // The operations waiting on the timer.
    op_queue<wait_op> op_queue_;

    // The index of the timer in the heap.
    std::size_t heap_index_;

    // Pointers to adjacent timers in a linked list.
    per_timer_data* next_;
    per_timer_data* prev_;
  };


private:
  // The head of a linked list of all active timers.
  per_timer_data* timers_;

  struct heap_entry
  {
    // The time when the timer should fire.
    time_type time_;

    // The associated timer with enqueued operations.
    per_timer_data* timer_;
  };

  // The heap of timers, with the earliest timer at the front.
  std::vector<heap_entry> heap_;
};

// Move the item at the given index up the heap to its correct position.
  void up_heap(std::size_t index)
  {
    while (index > 0)
    {
      std::size_t parent = (index - 1) / 2;
      if (!Time_Traits::less_than(heap_[index].time_, heap_[parent].time_))
        break;
      swap_heap(index, parent);
      index = parent;
    }
  }

  // Move the item at the given index down the heap to its correct position.
  void down_heap(std::size_t index)
  {
    std::size_t child = index * 2 + 1;
    while (child < heap_.size())
    {
      std::size_t min_child = (child + 1 == heap_.size()
          || Time_Traits::less_than(
            heap_[child].time_, heap_[child + 1].time_))
        ? child : child + 1;
      if (Time_Traits::less_than(heap_[index].time_, heap_[min_child].time_))
        break;
      swap_heap(index, min_child);
      index = min_child;
      child = index * 2 + 1;
    }
  }

  // Swap two entries in the heap.
  void swap_heap(std::size_t index1, std::size_t index2)
  {
    heap_entry tmp = heap_[index1];
    heap_[index1] = heap_[index2];
    heap_[index2] = tmp;
    heap_[index1].timer_->heap_index_ = index1;
    heap_[index2].timer_->heap_index_ = index2;
  }

// Add a new timer to the queue. Returns true if this is the timer that is
 // earliest in the queue, in which case the reactor's event demultiplexing
 // function call may need to be interrupted and restarted.
 bool enqueue_timer(const time_type& time, per_timer_data& timer, wait_op* op)
 {
 // Enqueue the timer object.
 if (timer.prev_ == 0 && &timer != timers_)
 {
  if (this->is_positive_infinity(time))
  {
    // No heap entry is required for timers that never expire.
    timer.heap_index_ = (std::numeric_limits<std::size_t>::max)();
  }
  else
  {
    // Put the new timer at the correct position in the heap. This is done
    // first since push_back() can throw due to allocation failure.
    timer.heap_index_ = heap_.size();
    heap_entry entry = { time, &timer };
    heap_.push_back(entry);
    up_heap(heap_.size() - 1);
  }

  // Insert the new timer into the linked list of active timers.
  timer.next_ = timers_;
  timer.prev_ = 0;
  if (timers_)
    timers_->prev_ = &timer;
  timers_ = &timer;
 }

 // Enqueue the individual timer operation.
 timer.op_queue_.push(op);

 // Interrupt reactor only if newly added timer is first to expire.
 return timer.heap_index_ == 0 && timer.op_queue_.front() == op;
 }

#if defined(ASIO_WINDOWS_RUNTIME)
# include "asio/detail/winrt_timer_schedupp"
#elif defined(ASIO_HAS_EPOLL)
# include "asio/detail/epoll_reactor.hpp"
#elif defined(ASIO_HAS_KQUEUE)
# include "asio/detail/kqueue_reactor.hpp"
#elif defined(ASIO_HAS_DEV_POLL)
# include "asio/detail/dev_poll_reactor.hpp"
#else
# include "asio/detail/select_reactor.hpp"
#endif

void winrt_timer_scheduler::run_thread()
{
  asio::detail::mutex::scoped_lock lock(mutex_);
  while (!stop_thread_)
  {
    const long max_wait_duration = 5 * 60 * 1000000;
    long wait_duration = timer_queues_.wait_duration_usec(max_wait_duration);
    event_.wait_for_usec(lock, wait_duration);
    event_.clear(lock);
    op_queue<operation> ops;
    timer_queues_.get_ready_timers(ops);
    if (!ops.empty())
    {
      lock.unlock();
      scheduler_.post_deferred_completions(ops);
      lock.lock();
    }
  }
}

template <typename Time_Traits>
void winrt_timer_scheduler::schedule_timer(timer_queue<Time_Traits>& queue, const typename Time_Traits::time_type& time, typename timer_queue<Time_Traits>::per_timer_data& timer, wait_op* op) {
  asio::detail::mutex::scoped_lock lock(mutex_);

  if (shutdown_) {
    scheduler_.post_immediate_completion(op, false);
    return;
  }

  bool earliest = queue.enqueue_timer(time, timer, op);
  scheduler_.work_started();
  if (earliest) event_.signal(lock);
}

// Add the timer descriptor to epoll.
 if (timer_fd_ != -1)
 {
 ev.events = EPOLLIN | EPOLLERR;
 ev.data.ptr = &timer_fd_;
 epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &ev);
 }

void epoll_reactor::run(long usec, op_queue<operation>& ops)
{
 int timeout;
 if (usec == 0)
   timeout = 0;
 else
 {
   timeout = (usec < 0) ? -1 : ((usec - 1) / 1000 + 1);
 }

 bool check_timers = false
 // Block on the epoll descriptor.
 epoll_event events[128];
 int num_events = epoll_wait(epoll_fd_, events, 128, timeout);

 // Dispatch the waiting events.
 for (int i = 0; i < num_events; ++i)
 {
   void* ptr = events[i].data.ptr;
   if (ptr == &timer_fd_)
   {
     check_timers = true;
   }
 }

 if (check_timers)
 {
   mutex::scoped_lock common_lock(mutex_);
   timer_queues_.get_ready_timers(ops);

   itimerspec new_timeout;
   itimerspec old_timeout;
   int flags = get_timeout(new_timeout);
   timerfd_settime(timer_fd_, flags, &new_timeout, &old_timeout);
 }
}

using system_timer = basic_waitable_timer<std::chrono::system_clock>;
using steady_timer = basic_waitable_timer<std::chrono::steady_clock>;
using high_resolution_timer = basic_waitable_timer<
    std::chrono::high_resolution_clock>;

asio::steady_timer t(ctx);
  t.expires_after(1s);
  t.async_wait([](asio::error_code ec) { 
    std::cout << "timeout!" << std::endl; 
  });