HANDLE   CreateIoCompletionPort   (   
      HANDLE   FileHandle,                                 //   handle   to   file   
      HANDLE   ExistingCompletionPort,           //   handle   to   I/O   completion   port   
      ULONG_PTR   CompletionKey,                 //   completion   key   
      DWORD   NumberOfConcurrentThreads   //   number   of   threads   to   execute   concurrently   
  );   

CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0)

CreateIoCompletionPort(ioHandle, iocp, (ULONG_PTR)fn, 0)

BOOL WINAPI GetQueuedCompletionStatus(  
  __in   HANDLE          CompletionPort,    // 这个就是我们建立的那个唯一的完成端口   
  __out  LPDWORD         lpNumberOfBytes,   //这个是操作完成后返回的字节数   
  __out  PULONG_PTR      lpCompletionKey,   // 这个是我们建立完成端口的时候绑定的那个自定义结构体参数   
  __out  LPOVERLAPPED    *lpOverlapped,     // 这个是我们在连入Socket的时候一起建立的那个重叠结构   
  __in   DWORD           dwMilliseconds     // 等待完成端口的超时时间，如果线程不需要做其他的事情，那就INFINITE就行了   
   ); 

BOOL WINAPI PostQueuedCompletionStatus(  
   __in      HANDLE CompletionPort,  
   __in      DWORD dwNumberOfBytesTransferred,  
   __in      ULONG_PTR dwCompletionKey,  
   __in_opt  LPOVERLAPPED lpOverlapped  
);  

for (int i = 0; i < m_nThreads; i++)  
{  
      PostQueuedCompletionStatus(m_hIOCompletionPort, 0, (DWORD) NULL, NULL);  
}  

private:
#if defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
  detail::winsock_init<> init_;
#elif defined(__sun) || defined(__QNX__) || defined(__hpux) || defined(_AIX) \
  || defined(__osf__)
  detail::signal_init<> init_;
#endif

  // The service registry.
  boost::asio::detail::service_registry* service_registry_;

  // The implementation.
  impl_type& impl_;

typedef detail::io_service_impl impl_type;

namespace detail {
#if defined(BOOST_ASIO_HAS_IOCP)
  typedef class win_iocp_io_service io_service_impl;
  class win_iocp_overlapped_ptr;
#else
  typedef class task_io_service io_service_impl;
#endif
  class service_registry;
} //

io_service::io_service()
  : service_registry_(new boost::asio::detail::service_registry(
        *this, static_cast<impl_type*>(0),
        (std::numeric_limits<std::size_t>::max)())),
    impl_(service_registry_->first_service<impl_type>())
{
}

explicit basic_io_object(boost::asio::io_service& io_service)
: service(boost::asio::use_service<IoObjectService>(io_service))
{
    service.construct(implementation);
}

typedef basic_resolver<tcp> resolver;

template <typename ResolveHandler>
void async_resolve(const query& q, BOOST_ASIO_MOVE_ARG(ResolveHandler) handler)
{
// If you get an error on the following line it means that your handler does
// not meet the documented type requirements for a ResolveHandler.
BOOST_ASIO_RESOLVE_HANDLER_CHECK(
    ResolveHandler, handler, iterator) type_check;

return this->service.async_resolve(this->implementation, q,
    BOOST_ASIO_MOVE_CAST(ResolveHandler)(handler));
}

/// Asynchronously resolve a query to a list of entries.
template <typename ResolveHandler>
BOOST_ASIO_INITFN_RESULT_TYPE(ResolveHandler,
  void (boost::system::error_code, iterator_type))
async_resolve(implementation_type& impl, const query_type& query,
  BOOST_ASIO_MOVE_ARG(ResolveHandler) handler)
{
boost::asio::detail::async_result_init<
  ResolveHandler, void (boost::system::error_code, iterator_type)> init(
    BOOST_ASIO_MOVE_CAST(ResolveHandler)(handler));

service_impl_.async_resolve(impl, query, init.handler);

return init.result.get();
}

#if defined(BOOST_ASIO_WINDOWS_RUNTIME)
  typedef boost::asio::detail::winrt_resolver_service<InternetProtocol>
    service_impl_type;
#else
  typedef boost::asio::detail::resolver_service<InternetProtocol>
    service_impl_type;
#endif

// Asynchronously resolve a query to a list of entries.
template <typename Handler>
void async_resolve(implementation_type& impl,
  const query_type& query, Handler& handler)
{
// Allocate and construct an operation to wrap the handler.
typedef resolve_op<Protocol, Handler> op;
typename op::ptr p = { boost::asio::detail::addressof(handler),
  boost_asio_handler_alloc_helpers::allocate(
    sizeof(op), handler), 0 };
p.p = new (p.v) op(impl, query, io_service_impl_, handler);

BOOST_ASIO_HANDLER_CREATION((p.p, "resolver", &impl, "async_resolve"));

start_resolve_op(p.p);
p.v = p.p = 0;
}

void resolver_service_base::start_resolve_op(operation* op)
{
  start_work_thread();
  io_service_impl_.work_started();
  work_io_service_impl_.post_immediate_completion(op, false);
}

namespace boost {
namespace asio {
namespace detail {

#if defined(BOOST_ASIO_HAS_IOCP)
typedef win_iocp_operation operation;
#else
typedef task_io_service_operation operation;
#endif

} // namespace detail
} // namespace asio
} // namespace boost

void complete(win_iocp_io_service& 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);
}
typedef void (*func_type)(
win_iocp_io_service*, win_iocp_operation*,
const boost::system::error_code&, std::size_t);

win_iocp_operation(func_type func)
: next_(0),
func_(func)
{
    reset();
}
win_iocp_operation* next_;
func_type func_;

resolve_op(socket_ops::weak_cancel_token_type cancel_token,
const query_type& query, io_service_impl& ios, Handler& handler)
: operation(&resolve_op::do_complete),
cancel_token_(cancel_token),
query_(query),
io_service_impl_(ios),
handler_(BOOST_ASIO_MOVE_CAST(Handler)(handler)),
addrinfo_(0)
{
}

class win_iocp_operation
  : public OVERLAPPED

static void do_complete(io_service_impl* owner, operation* base,
const boost::system::error_code& /*ec*/,
std::size_t /*bytes_transferred*/)

void win_iocp_socket_service_base::start_receive_op(
    win_iocp_socket_service_base::base_implementation_type& impl,
    WSABUF* buffers, std::size_t buffer_count,
    socket_base::message_flags flags, bool noop, operation* op)
{
  update_cancellation_thread_id(impl);
  iocp_service_.work_started();
 
  if (noop)
    iocp_service_.on_completion(op);
  else if (!is_open(impl))
    iocp_service_.on_completion(op, boost::asio::error::bad_descriptor);
  else
  {
    DWORD bytes_transferred = 0;
    DWORD recv_flags = flags;
    int result = ::WSARecv(impl.socket_, buffers, static_cast<DWORD>(buffer_count),
        &bytes_transferred, &recv_flags, op, 0);
    DWORD last_error = ::WSAGetLastError();
    if (last_error == ERROR_NETNAME_DELETED)
      last_error = WSAECONNRESET;
    else if (last_error == ERROR_PORT_UNREACHABLE)
      last_error = WSAECONNREFUSED;
    if (result != 0 && last_error != WSA_IO_PENDING)
      iocp_service_.on_completion(op, last_error, bytes_transferred);
    else
      iocp_service_.on_pending(op);
  }
}

void win_iocp_io_service::on_pending(win_iocp_operation* op)
{
  if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
  {
    // Enqueue the operation on the I/O completion port.
    if (!::PostQueuedCompletionStatus(iocp_.handle,
          0, overlapped_contains_result, op))
    {
      // Out of resources. Put on completed queue instead.
      mutex::scoped_lock lock(dispatch_mutex_);
      completed_ops_.push(op);
      ::InterlockedExchange(&dispatch_required_, 1);
    }
  }
}

class task_io_service_operation BOOST_ASIO_INHERIT_TRACKED_HANDLER
{
public:
  void complete(task_io_service& 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)(task_io_service*,
      task_io_service_operation*,
      const boost::system::error_code&, std::size_t);

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

  // Prevents deletion through this type.
  ~task_io_service_operation()
  {
  }

private:
  friend class op_queue_access;
  task_io_service_operation* next_;
  func_type func_;
protected:
  friend class task_io_service;
  unsigned int task_result_; // Passed into bytes transferred.
};

class descriptor_state : operation
  {
    friend class epoll_reactor;
    friend class object_pool_access;

    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 shutdown_;

    BOOST_ASIO_DECL descriptor_state();
    void set_ready_events(uint32_t events) { task_result_ = events; }
    BOOST_ASIO_DECL operation* perform_io(uint32_t events);
    BOOST_ASIO_DECL static void do_complete(
        io_service_impl* owner, operation* base,
        const boost::system::error_code& ec, std::size_t bytes_transferred);
  };

epoll_reactor::descriptor_state::descriptor_state()
  : operation(&epoll_reactor::descriptor_state::do_complete)
{
}

void epoll_reactor::descriptor_state::do_complete(
    io_service_impl* owner, operation* base,
    const boost::system::error_code& ec, std::size_t bytes_transferred)
{
  if (owner)
  {
    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* epoll_reactor::descriptor_state::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);

  // Exception operations must be processed first to ensure that any
  // out-of-band data is read before normal data.
  static const int flag[max_ops] = { EPOLLIN, EPOLLOUT, EPOLLPRI };
  for (int j = max_ops - 1; j >= 0; --j)
  {
    if (events & (flag[j] | EPOLLERR | EPOLLHUP))
    {
      while (reactor_op* op = op_queue_[j].front())
      {
        if (op->perform())
        {
          op_queue_[j].pop();
          io_cleanup.ops_.push(op);
        }
        else
          break;
      }
    }
  }

  // The first operation will be returned for completion now. The others will
  // be posted for later by the io_cleanup object's destructor.
  io_cleanup.first_op_ = io_cleanup.ops_.front();
  io_cleanup.ops_.pop();
  return io_cleanup.first_op_;
}

enum op_types { read_op = 0, write_op = 1,
    connect_op = 1, except_op = 2, max_ops = 3 };

class reactor_op
  : public operation
{
public:
  // The error code to be passed to the completion handler.
  boost::system::error_code ec_;

  // The number of bytes transferred, to be passed to the completion handler.
  std::size_t bytes_transferred_;

  // Perform the operation. Returns true if it is finished.
  bool perform()
  {
    return perform_func_(this);
  }

protected:
  typedef bool (*perform_func_type)(reactor_op*);

  reactor_op(perform_func_type perform_func, func_type complete_func)
    : operation(complete_func),
      bytes_transferred_(0),
      perform_func_(perform_func)
  {
  }

private:
  perform_func_type perform_func_;
};

boost::asio::io_service::strand strand_(&io); //注意io_service对象地址作为他的参数。
timer1_.async_wait(strand_.wrap(boost::bind(&printer::print1, this)));
timer2_.async_wait(strand_.wrap(boost::bind(&printer::print2, this)));

template <typename Handler>
#if defined(GENERATING_DOCUMENTATION)
unspecified
#else
detail::wrapped_handler<strand, Handler, detail::is_continuation_if_running>
#endif
wrap(Handler handler)
{
return detail::wrapped_handler<io_service::strand, Handler,
    detail::is_continuation_if_running>(*this, handler);
}

wrapped_handler(Dispatcher dispatcher, Handler& handler)
: dispatcher_(dispatcher),
  handler_(BOOST_ASIO_MOVE_CAST(Handler)(handler))
{
}
void operator()()
{
    dispatcher_.dispatch(BOOST_ASIO_MOVE_CAST(Handler)(handler_));
}

template <typename CompletionHandler>
BOOST_ASIO_INITFN_RESULT_TYPE(CompletionHandler, void ())
dispatch(BOOST_ASIO_MOVE_ARG(CompletionHandler) handler)
{
    // If you get an error on the following line it means that your handler does
    // not meet the documented type requirements for a CompletionHandler.
    BOOST_ASIO_COMPLETION_HANDLER_CHECK(CompletionHandler, handler) type_check;

    detail::async_result_init<
      CompletionHandler, void ()> init(
        BOOST_ASIO_MOVE_CAST(CompletionHandler)(handler));

    service_.dispatch(impl_, init.handler);

    return init.result.get();
}

class strand_impl
: public operation
{
    public:
    strand_impl();

    private:
    // Only this service will have access to the internal values.
    friend class strand_service;
    friend struct on_do_complete_exit;
    friend struct on_dispatch_exit;

    // Mutex to protect access to internal data.
    boost::asio::detail::mutex mutex_;

    // Indicates whether the strand is currently "locked" by a handler. This
    // means that there is a handler upcall in progress, or that the strand
    // itself has been scheduled in order to invoke some pending handlers.
    bool locked_;

    // The handlers that are waiting on the strand but should not be run until
    // after the next time the strand is scheduled. This queue must only be
    // modified while the mutex is locked.
    op_queue<operation> waiting_queue_;

    // The handlers that are ready to be run. Logically speaking, these are the
    // handlers that hold the strand's lock. The ready queue is only modified
    // from within the strand and so may be accessed without locking the mutex.
    op_queue<operation> ready_queue_;
};

template <typename Handler>
void strand_service::dispatch(strand_service::implementation_type& impl,
    Handler& handler)
{
  // If we are already in the strand then the handler can run immediately.
  if (call_stack<strand_impl>::contains(impl))
  {
    fenced_block b(fenced_block::full);
    boost_asio_handler_invoke_helpers::invoke(handler, handler);
    return;
  }

  // Allocate and construct an operation to wrap the handler.
  typedef completion_handler<Handler> op;
  typename op::ptr p = { boost::asio::detail::addressof(handler),
    boost_asio_handler_alloc_helpers::allocate(
      sizeof(op), handler), 0 };
  p.p = new (p.v) op(handler);

  BOOST_ASIO_HANDLER_CREATION((p.p, "strand", impl, "dispatch"));

  bool dispatch_immediately = do_dispatch(impl, p.p);
  operation* o = p.p;
  p.v = p.p = 0;

  if (dispatch_immediately)
  {
    // Indicate that this strand is executing on the current thread.
    call_stack<strand_impl>::context ctx(impl);

    // Ensure the next handler, if any, is scheduled on block exit.
    on_dispatch_exit on_exit = { &io_service_, impl };
    (void)on_exit;

    completion_handler<Handler>::do_complete(
        &io_service_, o, boost::system::error_code(), 0);
  }
}

bool strand_service::do_dispatch(implementation_type& impl, operation* op)
{
  // If we are running inside the io_service, and no other handler already
  // holds the strand lock, then the handler can run immediately.
  bool can_dispatch = io_service_.can_dispatch();
  impl->mutex_.lock();
  if (can_dispatch && !impl->locked_)
  {
    // Immediate invocation is allowed.
    impl->locked_ = true;
    impl->mutex_.unlock();
    return true;
  }

  if (impl->locked_)
  {
    // Some other handler already holds the strand lock. Enqueue for later.
    impl->waiting_queue_.push(op);
    impl->mutex_.unlock();
  }
  else
  {
    // The handler is acquiring the strand lock and so is responsible for
    // scheduling the strand.
    impl->locked_ = true;
    impl->mutex_.unlock();
    impl->ready_queue_.push(op);
    io_service_.post_immediate_completion(impl, false);
  }

  return false;
}

struct strand_service::on_dispatch_exit
{
  io_service_impl* io_service_;
  strand_impl* impl_;

  ~on_dispatch_exit()
  {
    impl_->mutex_.lock();
    impl_->ready_queue_.push(impl_->waiting_queue_);
    bool more_handlers = impl_->locked_ = !impl_->ready_queue_.empty();
    impl_->mutex_.unlock();

    if (more_handlers)
      io_service_->post_immediate_completion(impl_, false);
  }
};

win_iocp_io_service::win_iocp_io_service(
    boost::asio::io_service& io_service, size_t concurrency_hint)
  : boost::asio::detail::service_base<win_iocp_io_service>(io_service),
    iocp_(),
    outstanding_work_(0),
    stopped_(0),
    stop_event_posted_(0),
    shutdown_(0),
    gqcs_timeout_(get_gqcs_timeout()),
    dispatch_required_(0)
{
  BOOST_ASIO_HANDLER_TRACKING_INIT;

  iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0,
      static_cast<DWORD>(concurrency_hint < DWORD(~0)
        ? concurrency_hint : DWORD(~0)));
  if (!iocp_.handle)
  {
    DWORD last_error = ::GetLastError();
    boost::system::error_code ec(last_error,
        boost::asio::error::get_system_category());
    boost::asio::detail::throw_error(ec, "iocp");
  }
}

struct auto_handle
{
HANDLE handle;
auto_handle() : handle(0) {}
~auto_handle() { if (handle) ::CloseHandle(handle); }
};

// The IO completion port used for queueing operations.
auto_handle iocp_;

boost::system::error_code win_iocp_io_service::register_handle(
    HANDLE handle, boost::system::error_code& ec)
{
  if (::CreateIoCompletionPort(handle, iocp_.handle, 0, 0) == 0)
  {
    DWORD last_error = ::GetLastError();
    ec = boost::system::error_code(last_error,
        boost::asio::error::get_system_category());
  }
  else
  {
    ec = boost::system::error_code();
  }
  return ec;
}

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

  socket_holder sock(socket_ops::socket(family, type, protocol, ec));
  if (sock.get() == invalid_socket)
    return ec;

  HANDLE sock_as_handle = reinterpret_cast<HANDLE>(sock.get());
  if (iocp_service_.register_handle(sock_as_handle, ec))
    return ec;

  impl.socket_ = sock.release();
  switch (type)
  {
  case SOCK_STREAM: impl.state_ = socket_ops::stream_oriented; break;
  case SOCK_DGRAM: impl.state_ = socket_ops::datagram_oriented; break;
  default: impl.state_ = 0; break;
  }
  impl.cancel_token_.reset(static_cast<void*>(0), socket_ops::noop_deleter());
  ec = boost::system::error_code();
  return ec;
}

thread1=std::thread(&io_service.run);
thread1.join();

size_t win_iocp_io_service::run(boost::system::error_code& ec)
{
  if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
  {
    stop();
    ec = boost::system::error_code();
    return 0;
  }
 
  win_iocp_thread_info this_thread;
  thread_call_stack::context ctx(this, this_thread);
 
  size_t n = 0;
   while (do_one(true, ec))
    if (n != (std::numeric_limits<size_t>::max)())
      ++n;
  return n;
}

  for (;;)

// Try to acquire responsibility for dispatching timers and completed ops.
if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1)
{
  mutex::scoped_lock lock(dispatch_mutex_);

  // Dispatch pending timers and operations.
  op_queue<win_iocp_operation> ops;
  ops.push(completed_ops_);
  timer_queues_.get_ready_timers(ops);
  post_deferred_completions(ops);
  update_timeout();
}

// Get the next operation from the queue.
DWORD bytes_transferred = 0;
dword_ptr_t completion_key = 0;
LPOVERLAPPED overlapped = 0;
::SetLastError(0);
BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
    &completion_key, &overlapped, block ? gqcs_timeout_ : 0);
DWORD last_error = ::GetLastError();

win_iocp_operation* op = static_cast<win_iocp_operation*>(overlapped);
boost::system::error_code result_ec(last_error,
  boost::asio::error::get_system_category());

// We may have been passed the last_error and bytes_transferred in the
// OVERLAPPED structure itself.
if (completion_key == overlapped_contains_result)
{
result_ec = boost::system::error_code(static_cast<int>(op->Offset),
    *reinterpret_cast<boost::system::error_category*>(op->Internal));
bytes_transferred = op->OffsetHigh;
}

// Otherwise ensure any result has been saved into the OVERLAPPED
// structure.
else
{
op->Internal = reinterpret_cast<ulong_ptr_t>(&result_ec.category());
op->Offset = result_ec.value();
op->OffsetHigh = bytes_transferred;
}

// Dispatch the operation only if ready. The operation may not be ready
// if the initiating function (e.g. a call to WSARecv) has not yet
// returned. This is because the initiating function still wants access
// to the operation's OVERLAPPED structure.
if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
{
    // Ensure the count of outstanding work is decremented on block exit.
    work_finished_on_block_exit on_exit = { this };
    (void)on_exit;

    op->complete(*this, result_ec, bytes_transferred);
    ec = boost::system::error_code();
    return 1;
}

void win_iocp_io_service::post_deferred_completion(win_iocp_operation* op)
{
  // Flag the operation as ready.
  op->ready_ = 1;

  // Enqueue the operation on the I/O completion port.
  if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, op))
  {
    // Out of resources. Put on completed queue instead.
    mutex::scoped_lock lock(dispatch_mutex_);
    completed_ops_.push(op);
    ::InterlockedExchange(&dispatch_required_, 1);
  }
}

void win_iocp_io_service::post_deferred_completions(
    op_queue<win_iocp_operation>& ops)
{
  while (win_iocp_operation* op = ops.front())
  {
    ops.pop();

    // Flag the operation as ready.
    op->ready_ = 1;

    // Enqueue the operation on the I/O completion port.
    if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, op))
    {
      // Out of resources. Put on completed queue instead.
      mutex::scoped_lock lock(dispatch_mutex_);
      completed_ops_.push(op);
      completed_ops_.push(ops);
      ::InterlockedExchange(&dispatch_required_, 1);
    }
  }
}

std::size_t task_io_service::run(boost::system::error_code& ec)
{
  ec = boost::system::error_code();
  if (outstanding_work_ == 0)
  {
    stop();
    return 0;
  }

  thread_info this_thread;
  this_thread.private_outstanding_work = 0;
  thread_call_stack::context ctx(this, this_thread);

  mutex::scoped_lock lock(mutex_);

  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;
}

// Run at most one operation. May block.
BOOST_ASIO_DECL std::size_t do_run_one(mutex::scoped_lock& lock,
  thread_info& this_thread, const boost::system::error_code& ec);

// Poll for at most one operation.
BOOST_ASIO_DECL std::size_t do_poll_one(mutex::scoped_lock& lock,
  thread_info& this_thread, const boost::system::error_code& ec);

// Stop the task and all idle threads.
BOOST_ASIO_DECL void stop_all_threads(mutex::scoped_lock& lock);

// Wake a single idle thread, or the task, and always unlock the mutex.
BOOST_ASIO_DECL void wake_one_thread_and_unlock(
  mutex::scoped_lock& lock);

std::size_t task_io_service::do_run_one(mutex::scoped_lock& lock,
    task_io_service::thread_info& this_thread,
    const boost::system::error_code& ec)
{
  while (!stopped_)
  {
    if (!op_queue_.empty())
    {
      // Prepare to execute first handler from queue.
      operation* o = op_queue_.front();
      op_queue_.pop();
      bool more_handlers = (!op_queue_.empty());

      //执行具体任务
    }
    else
    {
      wakeup_event_.clear(lock);
      wakeup_event_.wait(lock);
    }
  }
  return 0;
}

if (o == &task_operation_)
  {
    task_interrupted_ = more_handlers;

    if (more_handlers && !one_thread_)
      wakeup_event_.unlock_and_signal_one(lock);
    else
      lock.unlock();

    task_cleanup on_exit = { this, &lock, &this_thread };
    (void)on_exit;

    // Run the task. May throw an exception. Only block if the operation
    // queue is empty and we're not polling, otherwise we want to return
    // as soon as possible.
    task_->run(!more_handlers, this_thread.private_op_queue);
  }

// Operation object to represent the position of the task in the queue.
struct task_operation : operation
{
    task_operation() : operation(0) {}
} task_operation_;

void task_io_service::wake_one_thread_and_unlock(
    mutex::scoped_lock& lock)
{
  if (!wakeup_event_.maybe_unlock_and_signal_one(lock))
  {
    if (!task_interrupted_ && task_)
    {
      task_interrupted_ = true;
      task_->interrupt();
    }
    lock.unlock();
  }
}

#if !defined(BOOST_ASIO_HAS_THREADS)
typedef null_event event;
#elif defined(BOOST_ASIO_WINDOWS)
typedef win_event event;
#elif defined(BOOST_ASIO_HAS_PTHREADS)
typedef posix_event event;
#elif defined(BOOST_ASIO_HAS_STD_MUTEX_AND_CONDVAR)
typedef std_event event;
#endif

// The task to be run by this service.
reactor* task_;

struct task_io_service::task_cleanup
{
  ~task_cleanup()
  {
    if (this_thread_->private_outstanding_work > 0)
    {
      boost::asio::detail::increment(
          task_io_service_->outstanding_work_,
          this_thread_->private_outstanding_work);
    }
    this_thread_->private_outstanding_work = 0;

    // Enqueue the completed operations and reinsert the task at the end of
    // the operation queue.
    lock_->lock();
    task_io_service_->task_interrupted_ = true;
    task_io_service_->op_queue_.push(this_thread_->private_op_queue);
    task_io_service_->op_queue_.push(&task_io_service_->task_operation_);
  }

  task_io_service* task_io_service_;
  mutex::scoped_lock* lock_;
  thread_info* this_thread_;
};

#if defined(BOOST_ASIO_WINDOWS_RUNTIME)
typedef class null_reactor reactor;
#elif defined(BOOST_ASIO_HAS_IOCP)
typedef class select_reactor reactor;
#elif defined(BOOST_ASIO_HAS_EPOLL)
typedef class epoll_reactor reactor;
#elif defined(BOOST_ASIO_HAS_KQUEUE)
typedef class kqueue_reactor reactor;
#elif defined(BOOST_ASIO_HAS_DEV_POLL)
typedef class dev_poll_reactor reactor;
#else
typedef class select_reactor reactor;
#endif

void epoll_reactor::run(bool block, op_queue<operation>& ops)
  // This code relies on the fact that the task_io_service queues the reactor
  // task behind all descriptor operations generated by this function. This
  // means, that by the time we reach this point, any previously returned
  // descriptor operations have already been dequeued. Therefore it is now safe
  // for us to reuse and return them for the task_io_service to queue again.

// Calculate a timeout only if timerfd is not used.
int timeout;
if (timer_fd_ != -1)
timeout = block ? -1 : 0;
else
{
mutex::scoped_lock lock(mutex_);
timeout = block ? get_timeout() : 0;
}

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

#if defined(BOOST_ASIO_HAS_TIMERFD)
  bool check_timers = (timer_fd_ == -1);
#else // defined(BOOST_ASIO_HAS_TIMERFD)
  bool check_timers = true;
#endif // defined(BOOST_ASIO_HAS_TIMERFD)

  // Dispatch the waiting events.
  for (int i = 0; i < num_events; ++i)
  {
    void* ptr = events[i].data.ptr;
    if (ptr == &interrupter_)
    {
      // No need to reset the interrupter since we're leaving the descriptor
      // in a ready-to-read state and relying on edge-triggered notifications
      // to make it so that we only get woken up when the descriptor's epoll
      // registration is updated.

#if defined(BOOST_ASIO_HAS_TIMERFD)
      if (timer_fd_ == -1)
        check_timers = true;
#else // defined(BOOST_ASIO_HAS_TIMERFD)
      check_timers = true;
#endif // defined(BOOST_ASIO_HAS_TIMERFD)
    }
#if defined(BOOST_ASIO_HAS_TIMERFD)
    else if (ptr == &timer_fd_)
    {
      check_timers = true;
    }
#endif // defined(BOOST_ASIO_HAS_TIMERFD)
    else
    {
      // The descriptor operation doesn't count as work in and of itself, so we
      // don't call work_started() here. This still allows the io_service to
      // stop if the only remaining operations are descriptor operations.
      descriptor_state* descriptor_data = static_cast<descriptor_state*>(ptr);
      descriptor_data->set_ready_events(events[i].events);
      ops.push(descriptor_data);
    }
  }

void epoll_reactor::interrupt()
{
  epoll_event ev = { 0, { 0 } };
  ev.events = EPOLLIN | EPOLLERR | EPOLLET;
  ev.data.ptr = &interrupter_;
  epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, interrupter_.read_descriptor(), &ev);
}

  else
  {
    std::size_t task_result = o->task_result_;

    if (more_handlers && !one_thread_)
      wake_one_thread_and_unlock(lock);
    else
      lock.unlock();

    // Ensure the count of outstanding work is decremented on block exit.
    work_cleanup on_exit = { this, &lock, &this_thread };
    (void)on_exit;

    // Complete the operation. May throw an exception. Deletes the object.
    o->complete(*this, ec, task_result);

    return 1;
  }

void task_io_service::stop_all_threads(
    mutex::scoped_lock& lock)
{
  stopped_ = true;
  wakeup_event_.signal_all(lock);

  if (!task_interrupted_ && task_)
  {
    task_interrupted_ = true;
    task_->interrupt();
  }
}