template <typename Protocol, typename Executor>
class basic_socket_acceptor : public socket_base {
  void open(const protocol_type &protocol = protocol_type()) {
    boost::system::error_code ec;
    impl_.get_service().open(impl_.get_implementation(), protocol, ec);
    boost::asio::detail::throw_error(ec, "open");
  }
}

template <BOOST_ASIO_COMPLETION_TOKEN_FOR(
              void(boost::system::error_code,
                   typename Protocol::socket::template rebind_executor<
                       executor_type>::other))
              MoveAcceptToken = default_completion_token_t<executor_type>>
auto async_accept(
    MoveAcceptToken &&token = default_completion_token_t<executor_type>())
    -> decltype(async_initiate<
                MoveAcceptToken,
                void(boost::system::error_code,
                     typename Protocol::socket::template rebind_executor<
                         executor_type>::other)>(
        declval<initiate_async_move_accept>(), token,
        declval<const executor_type &>(), static_cast<endpoint_type *>(0),
        static_cast<typename Protocol::socket::template rebind_executor<
            executor_type>::other *>(0))) {
  return async_initiate<
      MoveAcceptToken, void(boost::system::error_code,
                            typename Protocol::socket::template rebind_executor<
                                executor_type>::other)>(
      initiate_async_move_accept(this), token, impl_.get_executor(),
      static_cast<endpoint_type *>(0),
      static_cast<typename Protocol::socket::template rebind_executor<
          executor_type>::other *>(0));
}

class initiate_async_move_accept {
public:
  typedef Executor executor_type;

  explicit initiate_async_move_accept(basic_socket_acceptor *self)
      : self_(self) {}

  const executor_type &get_executor() const noexcept {
    return self_->get_executor();
  }

  template <typename MoveAcceptHandler, typename Executor1, typename Socket>
  void operator()(MoveAcceptHandler &&handler, const Executor1 &peer_ex,
                  endpoint_type *peer_endpoint, Socket *) const {
    detail::non_const_lvalue<MoveAcceptHandler> handler2(handler);
    self_->impl_.get_service().async_move_accept(
        self_->impl_.get_implementation(), peer_ex, peer_endpoint,
        handler2.value, self_->impl_.get_executor());
  }

private:
  basic_socket_acceptor *self_;
};

template <typename MutableBufferSequence,
          BOOST_ASIO_COMPLETION_TOKEN_FOR(void(boost::system::error_code,
                                               std::size_t))
              ReadToken = default_completion_token_t<executor_type>>
auto async_read_some(
    const MutableBufferSequence &buffers,
    ReadToken &&token = default_completion_token_t<executor_type>())
    -> decltype(async_initiate<ReadToken, void(boost::system::error_code,
                                               std::size_t)>(
        declval<initiate_async_receive>(), token, buffers,
        socket_base::message_flags(0))) {
  return async_initiate<ReadToken,
                        void(boost::system::error_code, std::size_t)>(
      initiate_async_receive(this), token, buffers,
      socket_base::message_flags(0));
}

class initiate_async_receive {
public:
  typedef Executor executor_type;

  explicit initiate_async_receive(basic_stream_socket * self) : self_(self) {}

  const executor_type &get_executor() const noexcept {
    return self_->get_executor();
  }

  template <typename ReadHandler, typename MutableBufferSequence>
  void operator()(ReadHandler &&handler, const MutableBufferSequence &buffers,
                  socket_base::message_flags flags) const {
    detail::non_const_lvalue<ReadHandler> handler2(handler);
    self_->impl_.get_service().async_receive(self_->impl_.get_implementation(),
                                             buffers, flags, handler2.value,
                                             self_->impl_.get_executor());
  }

private:
  basic_stream_socket *self_;
};

template <typename IoObjectService,
    typename Executor = io_context::executor_type>
class io_object_impl {
  io_object_impl(int, const executor_type &ex)
      : service_(&boost::asio::use_service<IoObjectService>(
            io_object_impl::get_context(ex))),
        executor_(ex) {
    service_->construct(implementation_);
  }
};

class reactive_socket_service_base {
  struct base_implementation_type {
    // 原始 socket
    socket_type socket_;
    // 状态
    socket_ops::state_type state_;
    // 挂载到 reactor 上的异步 op 指针
    descriptor_state *reactor_data_;
  };

  // 构造 epoll_reactor
  reactive_socket_service_base(execution_context &context)
      : reactor_(use_service<reactor>(context)) {}

  boost::system::error_code
  do_open(reactive_socket_service_base::base_implementation_type &impl, int af,
          int type, int protocol, boost::system::error_code &ec) {
    if (is_open(impl)) {
      ec = boost::asio::error::already_open;
      return ec;
    }

    // 创建 server socket
    socket_holder sock(socket_ops::socket(af, type, protocol, ec));
    if (sock.get() == invalid_socket)
      return ec;

    // 往 epoll_reactor 上注册当前 server socket 相关的 I/O 事件
    // 初始化 reactor_data_
    // 设置 ptr 为 acceptor 的 impl.reactor_data_ 指针
    if (int err =
            reactor_.register_descriptor(sock.get(), impl.reactor_data_)) {
      ec = boost::system::error_code(err,
                                     boost::asio::error::get_system_category());
      return ec;
    }

    impl.socket_ = sock.release();
    switch (type) {
    case SOCK_STREAM:
      impl.state_ = socket_ops::stream_oriented | extra_state_;
      break;
    case SOCK_DGRAM:
      impl.state_ = socket_ops::datagram_oriented | extra_state_;
      break;
    default:
      impl.state_ = 0;
      break;
    }
    ec = boost::system::error_code();
    return ec;
  }

  boost::system::error_code reactive_socket_service_base::do_assign(
      reactive_socket_service_base::base_implementation_type &impl, int type,
      const reactive_socket_service_base::native_handle_type &native_socket,
      boost::system::error_code &ec) {
    if (is_open(impl)) {
      ec = boost::asio::error::already_open;
      return ec;
    }

    // 往 epoll_reactor 上注册当前 socket 相关的 I/O 事件
    if (int err =
            reactor_.register_descriptor(native_socket, impl.reactor_data_)) {
      ec = boost::system::error_code(err,
                                     boost::asio::error::get_system_category());
      return ec;
    }

    // 转移当前 socket
    impl.socket_ = native_socket;
    switch (type) {
    case SOCK_STREAM:
      impl.state_ = socket_ops::stream_oriented | extra_state_;
      break;
    case SOCK_DGRAM:
      impl.state_ = socket_ops::datagram_oriented | extra_state_;
      break;
    default:
      impl.state_ = 0;
      break;
    }
    impl.state_ |= socket_ops::possible_dup;
    ec = boost::system::error_code();
    return ec;
  }

  void construct(base_implementation_type& impl);

  reactor& reactor_;
};

template <typename PeerIoExecutor, typename Handler, typename IoExecutor>
void async_move_accept(implementation_type &impl,
                       const PeerIoExecutor &peer_io_ex,
                       endpoint_type *peer_endpoint, Handler &handler,
                       const IoExecutor &io_ex) {
  bool is_continuation =
      BOOST_ASIO_VERSIONED_NAME(handler_cont_helpers)::is_continuation(handler);

  associated_cancellation_slot_t<Handler> slot =
      boost::asio::get_associated_cancellation_slot(handler);

  typedef reactive_socket_move_accept_op<Protocol, PeerIoExecutor, Handler,
                                         IoExecutor>
      op;
  typename op::ptr p = {boost::asio::detail::addressof(handler),
                        op::ptr::allocate(handler), 0};
  p.p = new (p.v) op(success_ec_, peer_io_ex, impl.socket_, impl.state_,
                     impl.protocol_, peer_endpoint, handler, io_ex);

  if (slot.is_connected()) {
    p.p->cancellation_key_ = &slot.template emplace<reactor_op_cancellation>(
        &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op);
  }

  start_accept_op(impl, p.p, is_continuation, false, &io_ex, 0);
  p.v = p.p = 0;
}

                 typename IoExecutor>
class reactive_socket_move_accept_op
    : private Protocol::socket::template rebind_executor<PeerIoExecutor>::other,
      public reactive_socket_accept_op_base<
          typename Protocol::socket::template rebind_executor<
              PeerIoExecutor>::other,
          Protocol> {
public:
  typedef Handler handler_type;
  typedef IoExecutor io_executor_type;

  // RAII 资源管理指针
  BOOST_ASIO_DEFINE_HANDLER_PTR(reactive_socket_move_accept_op);

  reactive_socket_move_accept_op(const boost::system::error_code &success_ec,
                                 const PeerIoExecutor &peer_io_ex,
                                 socket_type socket,
                                 socket_ops::state_type state,
                                 const Protocol &protocol,
                                 typename Protocol::endpoint *peer_endpoint,
                                 Handler &handler, const IoExecutor &io_ex)
      : peer_socket_type(peer_io_ex),
        reactive_socket_accept_op_base<peer_socket_type, Protocol>(
            success_ec, socket, state, *this, protocol, peer_endpoint,
            &reactive_socket_move_accept_op::do_complete),
        handler_(static_cast<Handler &&>(handler)), work_(handler_, io_ex) {}

  // 就绪时调用的通知函数
  static void do_complete(void *owner, operation *base,
                          const boost::system::error_code & /*ec*/,
                          std::size_t /*bytes_transferred*/) {
    reactive_socket_move_accept_op* o(
        static_cast<reactive_socket_move_accept_op*>(base));
    ptr p = { boost::asio::detail::addressof(o->handler_), o, o };

    // 将 socket 从 new_socket_ 转移到自身
    if (owner)
      o->do_assign();

    handler_work<Handler, IoExecutor> w(
        static_cast<handler_work<Handler, IoExecutor>&&>(
          o->work_));

    // 绑定异步事件的 I/O 结果到 handler，即 error_code 和 socket
    detail::move_binder2<Handler,
      boost::system::error_code, peer_socket_type>
        handler(0, static_cast<Handler&&>(o->handler_), o->ec_,
          static_cast<peer_socket_type&&>(*o));
    p.h = boost::asio::detail::addressof(handler.handler_);
    p.reset();

    if (owner) {
      // 内存屏障
      fenced_block b(fenced_block::half);
      // 调用实际的通知函数，将 handler 提交到 handler.handler_ 绑定的 executor 去执行
      w.complete(handler, handler.handler_);
    }
  }

private:
  typedef typename Protocol::socket::template
    rebind_executor<PeerIoExecutor>::other peer_socket_type;
  Handler handler_;
  handler_work<Handler, IoExecutor> work_;
};

template <typename Socket, typename Protocol>
class reactive_socket_accept_op_base : public reactor_op {
public:
  reactive_socket_accept_op_base(const boost::system::error_code &success_ec,
                                 socket_type socket,
                                 socket_ops::state_type state, Socket &peer,
                                 const Protocol &protocol,
                                 typename Protocol::endpoint *peer_endpoint,
                                 func_type complete_func)
      : reactor_op(success_ec, &reactive_socket_accept_op_base::do_perform,
                   complete_func),
        socket_(socket), state_(state), peer_(peer), protocol_(protocol),
        peer_endpoint_(peer_endpoint),
        addrlen_(peer_endpoint ? peer_endpoint->capacity() : 0) {}

  // 事件到达时执行
  // 从 socket_ops::accept 系统调用中取出 socket 并存入 new_socket_ 中
  static status do_perform(reactor_op *base) {
    BOOST_ASIO_ASSUME(base != 0);
    reactive_socket_accept_op_base* o(
        static_cast<reactive_socket_accept_op_base*>(base));

    socket_type new_socket = invalid_socket;
    // accept 系统调用
    status result = socket_ops::non_blocking_accept(o->socket_,
        o->state_, o->peer_endpoint_ ? o->peer_endpoint_->data() : 0,
        o->peer_endpoint_ ? &o->addrlen_ : 0, o->ec_, new_socket)
    ? done : not_done;
    o->new_socket_.reset(new_socket);
    return result;
  }

  // 从 new_socket_ 中转移给自身
  void do_assign() {
    if (new_socket_.get() != invalid_socket) {
      if (peer_endpoint_)
        peer_endpoint_->resize(addrlen_);
      peer_.assign(protocol_, new_socket_.get(), ec_);
      if (!ec_)
        new_socket_.release();
    }
  }

private:
  socket_type socket_;
  socket_ops::state_type state_;
  socket_holder new_socket_;
  Socket& peer_;
  Protocol protocol_;
  typename Protocol::endpoint* peer_endpoint_;
  std::size_t addrlen_;
};

template <typename Handler, typename IoExecutor, typename = void>
class handler_work
    : handler_work_base<IoExecutor>,
      handler_work_base<associated_executor_t<Handler, IoExecutor>,
                        IoExecutor> {
public:
  typedef handler_work_base<IoExecutor> base1_type;
  typedef handler_work_base<associated_executor_t<Handler, IoExecutor>,
      IoExecutor> base2_type;

  handler_work(Handler &handler, const IoExecutor &io_ex) noexcept
      : base1_type(0, 0, io_ex),
        base2_type(base1_type::owns_work(),
                   boost::asio::get_associated_executor(handler, io_ex),
                   io_ex) {}

  template <typename Function>
  void complete(Function &function, Handler &handler) {
    if (!base1_type::owns_work() && !base2_type::owns_work()) {
      static_cast<Function&&>(function)();
    } else {
      base2_type::dispatch(function, handler);
    }
  }
};

template <typename MutableBufferSequence,
typename Handler, typename IoExecutor>
void async_receive(base_implementation_type& impl,
      const MutableBufferSequence& buffers, socket_base::message_flags flags,
      Handler& handler, const IoExecutor& io_ex) {
  bool is_continuation =
      BOOST_ASIO_VERSIONED_NAME(handler_cont_helpers)::is_continuation(handler);

  associated_cancellation_slot_t<Handler> slot =
      boost::asio::get_associated_cancellation_slot(handler);

  typedef reactive_socket_recv_op<MutableBufferSequence, Handler, IoExecutor>
      op;
  typename op::ptr p = {boost::asio::detail::addressof(handler),
                        op::ptr::allocate(handler), 0};
  p.p = new (p.v) op(success_ec_, impl.socket_, impl.state_, buffers, flags,
                     handler, io_ex);

  if (slot.is_connected()) {
    p.p->cancellation_key_ = &slot.template emplace<reactor_op_cancellation>(
        &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op);
  }

  start_op(impl,
           (flags & socket_base::message_out_of_band) ? reactor::except_op
                                                      : reactor::read_op,
           p.p, is_continuation,
           (flags & socket_base::message_out_of_band) == 0,
           ((impl.state_ & socket_ops::stream_oriented) &&
            buffer_sequence_adapter<boost::asio::mutable_buffer,
                                    MutableBufferSequence>::all_empty(buffers)),
           true, &io_ex, 0);
  p.v = p.p = 0;
}

// 一个 descriptor_state 对应一个 fd
// 其上挂载了所有等待它的异步操作
struct descriptor_state : operation {
  descriptor_state *next_;
  descriptor_state *prev_;

  mutex mutex_;
  epoll_reactor *reactor_;
  int descriptor_;
  uint32_t registered_events_;
  op_queue<reactor_op> op_queue_[max_ops];
  bool try_speculative_[max_ops];
  bool shutdown_;

  descriptor_state(bool locking, int spin_count);
  void set_ready_events(uint32_t events) { task_result_ = events; }
  void add_ready_events(uint32_t events) { task_result_ |= events; }

  // 检测 epoll_wait 返回的 event 类型(read_op, connect_or_write_op, except_op)
  // 收集对应激活的 op_queue_[activate_op_flag...] 到 ops_
  // 返回第一个 op 并交由 do_complete 中执行
  // 其余由追加到 scheduler 的 op_queue_ 中由事件循环推动
  operation *perform_io(uint32_t events) {
    mutex_.lock();
    perform_io_cleanup_on_block_exit io_cleanup(reactor_);
    mutex::scoped_lock descriptor_lock(mutex_, mutex::scoped_lock::adopt_lock);
    static const int flag[max_ops] = {EPOLLIN, EPOLLOUT, EPOLLPRI};
    for (int j = max_ops - 1; j >= 0; --j) {
      if (events & (flag[j] | EPOLLERR | EPOLLHUP)) {
        try_speculative_[j] = true;
        while (reactor_op *op = op_queue_[j].front()) {
          if (reactor_op::status status = op->perform()) {
            op_queue_[j].pop();
            io_cleanup.ops_.push(op);
            if (status == reactor_op::done_and_exhausted) {
              try_speculative_[j] = false;
              break;
            }
          } else
            break;
        }
      }
    }
    io_cleanup.first_op_ = io_cleanup.ops_.front();
    io_cleanup.ops_.pop();
    return io_cleanup.first_op_;
  }

  // 就绪时需要触发的钩子
  // 在事件循环中的 o->complete(this, ec, task_result) 中被调用
  static void do_complete(void *owner, operation *base,
                          const boost::system::error_code &ec,
                          std::size_t bytes_transferred) {
    if (owner) {
      // 检查记录的 owner 是否存活，使用 void* 避免多态开销
      descriptor_state *descriptor_data = static_cast<descriptor_state *>(base);
      uint32_t events = static_cast<uint32_t>(bytes_transferred);
      if (operation *op = descriptor_data->perform_io(events)) {
        op->complete(owner, ec, 0);
      }
    }
  }

  // 继承自 operation，保存 &do_complete 指针到父类 func_ 然后通过 complete 调用
  descriptor_state(bool locking, int spin_count)
      : operation(&epoll_reactor::descriptor_state::do_complete),
        mutex_(locking, spin_count) {}
};

void epoll_reactor::start_op(int op_type, socket_type descriptor,
                        epoll_reactor::per_descriptor_data &descriptor_data,
                        reactor_op *op, bool is_continuation,
                        bool allow_speculative,
                        void (*on_immediate)(operation *, bool, const void *),
                        const void *immediate_arg) {
  if (!descriptor_data) {
    op->ec_ = boost::asio::error::bad_descriptor;
    on_immediate(op, is_continuation, immediate_arg);
    return;
  }

  mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);

  if (descriptor_data->shutdown_) {
    on_immediate(op, is_continuation, immediate_arg);
    return;
  }

  if (descriptor_data->op_queue_[op_type].empty()) {
    if (allow_speculative &&
        (op_type != read_op || descriptor_data->op_queue_[except_op].empty())) {
      if (descriptor_data->try_speculative_[op_type]) {
        if (reactor_op::status status = op->perform()) {
          if (status == reactor_op::done_and_exhausted)
            if (descriptor_data->registered_events_ != 0)
              descriptor_data->try_speculative_[op_type] = false;
          descriptor_lock.unlock();
          on_immediate(op, is_continuation, immediate_arg);
          return;
        }
      }

      if (descriptor_data->registered_events_ == 0) {
        op->ec_ = boost::asio::error::operation_not_supported;
        on_immediate(op, is_continuation, immediate_arg);
        return;
      }

      if (op_type == write_op) {
        if ((descriptor_data->registered_events_ & EPOLLOUT) == 0) {
          epoll_event ev = { 0, { 0 } };
          ev.events = descriptor_data->registered_events_ | EPOLLOUT;
          ev.data.ptr = descriptor_data;
          if (epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, descriptor, &ev) == 0) {
            descriptor_data->registered_events_ |= ev.events;
          } else {
            op->ec_ = boost::system::error_code(
                errno, boost::asio::error::get_system_category());
            on_immediate(op, is_continuation, immediate_arg);
            return;
          }
        }
      }
    } else if (descriptor_data->registered_events_ == 0) {
      op->ec_ = boost::asio::error::operation_not_supported;
      on_immediate(op, is_continuation, immediate_arg);
      return;
    } else {
      if (op_type == write_op) {
        descriptor_data->registered_events_ |= EPOLLOUT;
      }

      epoll_event ev = {0, {0}};
      ev.events = descriptor_data->registered_events_;
      ev.data.ptr = descriptor_data;
      epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, descriptor, &ev);
    }
  }

  descriptor_data->op_queue_[op_type].push(op);
  scheduler_.work_started();
}

void epoll_reactor::run(long usec, op_queue<operation> &ops) {
  // Calculate timeout. Check the timer queues only if timerfd is not in use.
  int timeout;
  if (usec == 0)
    timeout = 0;
  else {
    timeout = (usec < 0) ? -1 : ((usec - 1) / 1000 + 1);
    if (timer_fd_ == -1) {
      mutex::scoped_lock lock(mutex_);
      timeout = get_timeout(timeout);
    }
  }

  // 获取 epoll_wait 就绪事件
  epoll_event events[128];
  int num_events = epoll_wait(epoll_fd_, events, 128, timeout);

  for (int i = 0; i < num_events; ++i) {
    // 获取 fd 的上下文
    void* ptr = events[i].data.ptr;
    descriptor_state *descriptor_data = static_cast<descriptor_state *>(ptr);
    if (!ops.is_enqueued(descriptor_data)) {
      descriptor_data->set_ready_events(events[i].events);
      ops.push(descriptor_data);
    } else {
      descriptor_data->add_ready_events(events[i].events);
    }
  }

  // 裁剪了 timer_fd 相关的逻辑
}

io_context::count_type io_context::run() {
  count_type s = impl_.run(ec);
  return s;
}

struct scheduler_thread_info : public thread_info_base
{
  op_queue<scheduler_operation> private_op_queue;
  long private_outstanding_work;
};

std::size_t scheduler::run(boost::system::error_code &ec) {
  thread_info this_thread; // 创建当前线程存储块
  thread_call_stack::context ctx(this, this_thread);

  mutex::scoped_lock lock(mutex_); // 获取 op_queue_ 锁

  std::size_t n = 0;
  for (; do_run_one(lock, this_thread, ec); lock.lock())
    if (n != (std::numeric_limits<std::size_t>::max)())
      ++n;
  return n;
}

std::size_t scheduler::do_run_one(mutex::scoped_lock &lock,
                                  scheduler::thread_info &this_thread,
                                  const boost::system::error_code &ec) {
  while (!stopped_) {
    if (!op_queue_.empty()) {
      // 获取待执行 op
      operation *o = op_queue_.front();
      op_queue_.pop();

      if (o == &task_operation_) { // 代表执行 epoll_wait 监听的标记
        // lock.unlock
        // 唤醒其他事件循环线程
        // ...

        // 保持只有一个线程处于 epoll_wait
        // on_exit 析构时重新添加 task_operation_ 进 op_queue_
        task_cleanup on_exit = {this, &lock, &this_thread};

        // 监听 epoll 并把就绪 op 收集到自身 private_op_queue 队列
        // on_exit 析构时添加进 op_queue_
        task_->run(more_handlers ? 0 : task_usec_,
            this_thread.private_op_queue);

        // 减少锁竞争优化
        // on_exit 析构时会去获取 lock 并更新 op_queue_ 队列
        // 所以这里并不会如同执行用户回调 op 一样直接返回，而是再次获取待执行 op
      } else { // 执行就绪的 op
        // lock.unlock
        // 唤醒其他事件循环线程
        // ...

        work_cleanup on_exit = { this, &lock, &this_thread };

        // 可能有立刻就绪的 op 产生，收集到 private_op_queue 队列
        // on_exit 析构时添加进 op_queue_
        o->complete(this, ec, o->task_result_);

        return 1; // 统计执行用户提交的异步操作次数
      }
    } else {
      // 空轮询优化
      // ...
    }
  }

  return 0;
}

class scheduler_task {
public:
  virtual void run(long usec, op_queue<scheduler_operation>& ops) = 0;
  virtual void interrupt() = 0;
};

class scheduler_operation {
public:
  typedef scheduler_operation operation_type;

  void complete(void *owner, const boost::system::error_code &ec,
                std::size_t bytes_transferred) {
    func_(owner, this, ec, bytes_transferred);
  }

  void destroy() { func_(0, this, boost::system::error_code(), 0); }

protected:
  typedef void (*func_type)(void *, scheduler_operation *,
                            const boost::system::error_code &, std::size_t);

  scheduler_operation(func_type func)
      : next_(0), func_(func), task_result_(0) {}

  ~scheduler_operation() {}

private:
  friend class op_queue_access;
  scheduler_operation *next_;
  func_type func_;
protected:
  friend class scheduler;
  unsigned int task_result_; // 记录接受到的字节数
};

// executor 通过存储指针 target_ 关联对应的 io_context
// 在 cpp 内存对齐下，指针末两位始终为0
// asio 将其作为标志位表示是否允许阻塞、是否延续执行

struct io_context_bits {
  static constexpr uintptr_t blocking_never = 1;
  static constexpr uintptr_t relationship_continuation = 2;
  static constexpr uintptr_t outstanding_work_tracked = 4;
  static constexpr uintptr_t runtime_bits = 3;
};

template <typename Allocator, uintptr_t Bits>
class io_context::basic_executor_type : detail::io_context_bits, Allocator {
public:
  /// Copy constructor.
  basic_executor_type(const basic_executor_type &other) noexcept
      : Allocator(static_cast<const Allocator &>(other)),
        target_(other.target_) {}

  // 设置 runtime_bits 仅由 execute 使用
  constexpr basic_executor_type
  require(execution::blocking_t::possibly_t) const {
    return basic_executor_type(context_ptr(),
        *this, bits() & ~blocking_never);
  }

  // 查询 runtime_bits
  constexpr execution::blocking_t query(execution::blocking_t) const noexcept {
    return (bits() & blocking_never)
      ? execution::blocking_t(execution::blocking.never)
      : execution::blocking_t(execution::blocking.possibly);
  }

  // 标准接口，适配 std
  template <typename Function> void execute(Function &&f) const;

  template <typename Function, typename OtherAllocator>
  void dispatch(Function &&f, const OtherAllocator &a) const;

  template <typename Function, typename OtherAllocator>
  void post(Function &&f, const OtherAllocator &a) const;

  template <typename Function, typename OtherAllocator>
  void defer(Function &&f, const OtherAllocator &a) const;

private:
  io_context *context_ptr() const noexcept {
    return reinterpret_cast<io_context *>(target_ & ~runtime_bits);
  }

  uintptr_t bits() const noexcept { return target_ & runtime_bits; }

  uintptr_t target_;
};

template <typename Allocator, uintptr_t Bits>
template <typename Function>
void io_context::basic_executor_type<Allocator, Bits>::execute(
    Function &&f) const {
  typedef decay_t<Function> function_type;

  // 非阻塞时且当前处于 io_context 循环时，可以直接执行
  if ((bits() & blocking_never) == 0 && context_ptr()->impl_.can_dispatch()) {
    function_type tmp(static_cast<Function&&>(f));

    detail::fenced_block b(detail::fenced_block::full);
    static_cast<function_type &&>(tmp)();
    return;
  }

  // 否则转为 operation 入队等待调度执行
  typedef detail::executor_op<function_type, Allocator, detail::operation> op;
  typename op::ptr p = {
      detail::addressof(static_cast<const Allocator &>(*this)),
      op::ptr::allocate(static_cast<const Allocator &>(*this)), 0};
  p.p = new (p.v) op(static_cast<Function&&>(f),
      static_cast<const Allocator&>(*this));

  BOOST_ASIO_HANDLER_CREATION((*context_ptr(), *p.p,
        "io_context", context_ptr(), 0, "execute"));

  context_ptr()->impl_.post_immediate_completion(p.p,
      (bits() & relationship_continuation) != 0);
  p.v = p.p = 0;
}

bool scheduler::can_dispatch() {
  return thread_call_stack::contains(this) != 0;
}

class thread_context {
public:
  static thread_info_base* top_of_thread_call_stack();

protected:
  typedef call_stack<thread_context, thread_info_base> thread_call_stack;
};

template <typename Key, typename Value = unsigned char> class call_stack {
public:
  class context : private noncopyable {
  public:
    explicit context(Key *k) : key_(k), next_(call_stack<Key, Value>::top_) {
      value_ = reinterpret_cast<unsigned char*>(this);
      call_stack<Key, Value>::top_ = this;
    }

    context(Key *k, Value &v)
        : key_(k), value_(&v), next_(call_stack<Key, Value>::top_) {
      call_stack<Key, Value>::top_ = this;
    }

    ~context() { call_stack<Key, Value>::top_ = next_; }

    Value *next_by_key() const {
      context* elem = next_;
      while (elem) {
        if (elem->key_ == key_)
          return elem->value_;
        elem = elem->next_;
      }
      return 0;
    }

  private:
    friend class call_stack<Key, Value>;
    Key* key_;
    Value* value_;
    context* next_;
  };

  friend class context;

  static Value *contains(Key *k) {
    context* elem = top_;
    while (elem) {
      if (elem->key_ == k)
        return elem->value_;
      elem = elem->next_;
    }
    return 0;
  }

  static Value *top() {
    context* elem = top_;
    return elem ? elem->value_ : 0;
  }

private:
  static tss_ptr<context> top_;
};

co_await socket.async_read_some(buffer, asio::use_awaitable);

template <typename Executor, typename R, typename... Args>
class async_result<use_awaitable_t<Executor>, R(Args...)> {
public:
  typedef typename detail::awaitable_handler<
      Executor, decay_t<Args>...> handler_type;
  typedef typename handler_type::awaitable_type return_type;

  template <typename Initiation, typename... InitArgs>
  static handler_type *do_init(
      detail::awaitable_frame_base<Executor> * frame, Initiation & initiation,
      use_awaitable_t<Executor> u, InitArgs & ...args) {
    // 因为调用异步接口，理应挂起当前协程等待完成回调
    // 运行当前协程的线程所使用的数据结构 awaitable_thread 被断开
    // 并重新被塞进 handler 里等待异步完成时调用并借助该数据结构恢复挂起的协程
    handler_type handler(frame->detach_thread());
    std::move(initiation)(std::move(handler), std::move(args)...);
    return nullptr;
  }

  template <typename Initiation, typename... InitArgs>
  static return_type initiate(Initiation initiation,
                              use_awaitable_t<Executor> u, InitArgs... args) {
    // 仿函数特化 await_transform，传入当前协程帧
    co_await
        [&](auto *frame) { return do_init(frame, initiation, u, args...); };

    for (;;) {} // Never reached.
  }
};

template <typename T, typename Executor, typename... Args>
struct coroutine_traits<boost::asio::awaitable<T, Executor>, Args...> {
  // awaitable_frame 作为 awaitable 类型的 promise_type
  typedef boost::asio::detail::awaitable_frame<T, Executor> promise_type;
}; 

template <typename T, typename Executor = any_io_executor> class awaitable {
public:
  typedef T value_type;

  typedef Executor executor_type;

  constexpr awaitable() noexcept : frame_(nullptr) {}

  awaitable(awaitable &&other) noexcept
      : frame_(std::exchange(other.frame_, nullptr)) {}

  ~awaitable() {
    if (frame_)
      frame_->destroy();
  }

  awaitable &operator=(awaitable &&other) noexcept {
    if (this != &other) {
      if (frame_)
        frame_->destroy();
      frame_ = std::exchange(other.frame_, nullptr);
    }
    return *this;
  }

  bool valid() const noexcept { return !!frame_; }

  bool await_ready() const noexcept { return false; }

  // 当 co_await 一个同为 awaitable<U, Executor> 类型的新协程体时
  // 实际将新协程体链接到当前协程帧的尾部
  template <class U>
  void await_suspend(
      detail::coroutine_handle<detail::awaitable_frame<U, Executor>> h) {
    frame_->push_frame(&h.promise());
  }

  // 当前协程体恢复，取出类型 T 的结果并返回给上一级的 co_await 表达式
  T await_resume() {
    return awaitable(static_cast<awaitable &&>(*this)).frame_->get();
  }

private:
  template <typename> friend class detail::awaitable_thread;
  template <typename, typename> friend class detail::awaitable_frame;

  awaitable(const awaitable &) = delete;
  awaitable &operator=(const awaitable &) = delete;

  explicit awaitable(detail::awaitable_frame<T, Executor> *a) : frame_(a) {}

  // 协程帧数据结构，维护嵌套式的协程
  detail::awaitable_frame<T, Executor> *frame_;
};

class awaitable_launch_context {
public:
  void launch(void (*pump_fn)(void*), void* arg) {
    call_stack<awaitable_launch_context>::context ctx(this);
    pump_fn(arg);
  }
  bool is_launching() {
    return !!call_stack<awaitable_launch_context>::contains(this);
  }
};

template <typename Executor>
class awaitable_frame_base : public awaitable_launch_context {
public:
  auto initial_suspend() noexcept { return suspend_always(); }

  auto final_suspend() noexcept {
    struct result {
      awaitable_frame_base *this_;

      bool await_ready() const noexcept { return false; }

      void await_suspend(coroutine_handle<void>) noexcept {
        this->this_->pop_frame();
      }

      void await_resume() const noexcept {}
    };

    return result{this};
  }

  // 设置异常标记
  template <typename Disposition>
  void set_disposition(Disposition &&d) noexcept {
    pending_exception_ = (to_exception_ptr)(static_cast<Disposition &&>(d));
  }

  void unhandled_exception() { set_disposition(std::current_exception()); }

  // 抛出异常
  void rethrow_exception() {
    if (pending_exception_) {
      std::exception_ptr ex = std::exchange(pending_exception_, nullptr);
      std::rethrow_exception(ex);
    }
  }

  // async_op 特化
  template <typename Op>
  auto await_transform(Op && op,
                       constraint_t<is_async_operation<Op>::value> = 0) {
    return awaitable_async_op<completion_signature_of_t<Op>, decay_t<Op>,
                              Executor>{std::forward<Op>(op), this};
  }

  // 仿函数特化 绑定 initiate
  template <typename Function>
  auto await_transform(
      Function f,
      enable_if_t<is_convertible<result_of_t<Function(awaitable_frame_base *)>,
                                 awaitable_thread<Executor> *>::value> * =
          nullptr) {
    struct result {
      Function function_;
      awaitable_frame_base *this_;

      // 直接挂起
      bool await_ready() const noexcept { return false; }

      // 延迟 initiate 函数到挂起后执行
      // 避免即刻完成的回调在协程未挂起时就尝试恢复协程
      void await_suspend(coroutine_handle<void>) noexcept {
        this_->after_suspend(
            [](void *arg) {
              result *r = static_cast<result *>(arg);
              r->function_(r->this_);
            },
            this);
      }

      // 挂起后通过传入的 initiate 函数完成了挂载到 awaitable_thread 的 frame 链上
      // 后续通过 awaitable_thread 的 pump 机制完成协程的恢复
      void await_resume() const noexcept {}
    };

    // 传入当前协程帧 this
    return result{std::move(f), this};
  }

  void attach_thread(awaitable_thread<Executor> *handler) noexcept {
    attached_thread_ = handler;
  }

  awaitable_thread<Executor> *detach_thread() noexcept {
    return std::exchange(attached_thread_, nullptr);
  }

  void push_frame(awaitable_frame_base<Executor> *caller) noexcept {
    caller_ = caller;
    attached_thread_ = caller_->attached_thread_;
    attached_thread_->entry_point()->top_of_stack_ = this;
    caller_->attached_thread_ = nullptr;
  }

  void pop_frame() noexcept {
    if (caller_)
      caller_->attached_thread_ = attached_thread_;
    attached_thread_->entry_point()->top_of_stack_ = caller_;
    attached_thread_ = nullptr;
    caller_ = nullptr;
  }

  struct resume_context {
    void (*after_suspend_fn_)(void *) = nullptr;
    void *after_suspend_arg_ = nullptr;
  };

  void resume() {
    resume_context context;
    resume_context_ = &context;
    coro_.resume();
    if (context.after_suspend_fn_)
      context.after_suspend_fn_(context.after_suspend_arg_);
  }

  void after_suspend(void (*fn)(void *), void *arg) {
    resume_context_->after_suspend_fn_ = fn;
    resume_context_->after_suspend_arg_ = arg;
  }

  void destroy() { coro_.destroy(); }

protected:
  coroutine_handle<void> coro_ = nullptr;
  awaitable_thread<Executor> *attached_thread_ = nullptr;
  awaitable_frame_base<Executor> *caller_ = nullptr;
  std::exception_ptr pending_exception_ = nullptr;
  resume_context *resume_context_ = nullptr;
};

template <typename T, typename Executor>
class awaitable_frame : public awaitable_frame_base<Executor> {
public:
  awaitable_frame() noexcept {}

  awaitable_frame(awaitable_frame &&other) noexcept
      : awaitable_frame_base<Executor>(std::move(other)) {}

  ~awaitable_frame() {
    if (has_result_)
      std::launder(static_cast<T *>(static_cast<void *>(result_)))->~T();
  }

  awaitable<T, Executor> get_return_object() noexcept {
    this->coro_ = coroutine_handle<awaitable_frame>::from_promise(*this);
    return awaitable<T, Executor>(this);
  }

  template <typename U> void return_value(U &&u) {
    new (&result_) T(std::forward<U>(u));
    has_result_ = true;
  }

  template <typename... Us> void return_values(Us &&...us) {
    this->return_value(std::forward_as_tuple(std::forward<Us>(us)...));
  }

  T get() {
    this->caller_ = nullptr;
    this->rethrow_exception();
    return std::move(
        *std::launder(static_cast<T *>(static_cast<void *>(result_))));
  }

private:
  alignas(T) unsigned char result_[sizeof(T)];
  bool has_result_ = false;
};