ceph/src/boost/boost/asio/detail/impl/select_reactor.ipp

   1 //
   2 // detail/impl/select_reactor.ipp
   3 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
   4 //
   5 // Copyright (c) 2003-2020 Christopher M. Kohlhoff (chris at kohlhoff dot com)
   6 //
   7 // Distributed under the Boost Software License, Version 1.0. (See accompanying
   8 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   9 //
  10
  11 #ifndef BOOST_ASIO_DETAIL_IMPL_SELECT_REACTOR_IPP
  12 #define BOOST_ASIO_DETAIL_IMPL_SELECT_REACTOR_IPP
  13
  14 #if defined(_MSC_VER) && (_MSC_VER >= 1200)
  15 # pragma once
  16 #endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
  17
  18 #include <boost/asio/detail/config.hpp>
  19
  20 #if defined(BOOST_ASIO_HAS_IOCP) \
  21   || (!defined(BOOST_ASIO_HAS_DEV_POLL) \
  22       && !defined(BOOST_ASIO_HAS_EPOLL) \
  23       && !defined(BOOST_ASIO_HAS_KQUEUE) \
  24       && !defined(BOOST_ASIO_WINDOWS_RUNTIME))
  25
  26 #include <boost/asio/detail/fd_set_adapter.hpp>
  27 #include <boost/asio/detail/select_reactor.hpp>
  28 #include <boost/asio/detail/signal_blocker.hpp>
  29 #include <boost/asio/detail/socket_ops.hpp>
  30
  31 #include <boost/asio/detail/push_options.hpp>
  32
  33 namespace boost {
  34 namespace asio {
  35 namespace detail {
  36
  37 #if defined(BOOST_ASIO_HAS_IOCP)
  38 class select_reactor::thread_function
  39 {
  40 public:
  41   explicit thread_function(select_reactor* r)
  42     : this_(r)
  43   {
  44   }
  45
  46   void operator()()
  47   {
  48     this_->run_thread();
  49   }
  50
  51 private:
  52   select_reactor* this_;
  53 };
  54 #endif // defined(BOOST_ASIO_HAS_IOCP)
  55
  56 select_reactor::select_reactor(boost::asio::execution_context& ctx)
  57   : execution_context_service_base<select_reactor>(ctx),
  58     scheduler_(use_service<scheduler_type>(ctx)),
  59     mutex_(),
  60     interrupter_(),
  61 #if defined(BOOST_ASIO_HAS_IOCP)
  62     stop_thread_(false),
  63     thread_(0),
  64 #endif // defined(BOOST_ASIO_HAS_IOCP)
  65     shutdown_(false)
  66 {
  67 #if defined(BOOST_ASIO_HAS_IOCP)
  68   boost::asio::detail::signal_blocker sb;
  69   thread_ = new boost::asio::detail::thread(thread_function(this));
  70 #endif // defined(BOOST_ASIO_HAS_IOCP)
  71 }
  72
  73 select_reactor::~select_reactor()
  74 {
  75   shutdown();
  76 }
  77
  78 void select_reactor::shutdown()
  79 {
  80   boost::asio::detail::mutex::scoped_lock lock(mutex_);
  81   shutdown_ = true;
  82 #if defined(BOOST_ASIO_HAS_IOCP)
  83   stop_thread_ = true;
  84   if (thread_)
  85     interrupter_.interrupt();
  86 #endif // defined(BOOST_ASIO_HAS_IOCP)
  87   lock.unlock();
  88
  89 #if defined(BOOST_ASIO_HAS_IOCP)
  90   if (thread_)
  91   {
  92     thread_->join();
  93     delete thread_;
  94     thread_ = 0;
  95   }
  96 #endif // defined(BOOST_ASIO_HAS_IOCP)
  97
  98   op_queue<operation> ops;
  99
 100   for (int i = 0; i < max_ops; ++i)
 101     op_queue_[i].get_all_operations(ops);
 102
 103   timer_queues_.get_all_timers(ops);
 104
 105   scheduler_.abandon_operations(ops);
 106 }
 107
 108 void select_reactor::notify_fork(
 109     boost::asio::execution_context::fork_event fork_ev)
 110 {
 111   if (fork_ev == boost::asio::execution_context::fork_child)
 112     interrupter_.recreate();
 113 }
 114
 115 void select_reactor::init_task()
 116 {
 117   scheduler_.init_task();
 118 }
 119
 120 int select_reactor::register_descriptor(socket_type,
 121     select_reactor::per_descriptor_data&)
 122 {
 123   return 0;
 124 }
 125
 126 int select_reactor::register_internal_descriptor(
 127     int op_type, socket_type descriptor,
 128     select_reactor::per_descriptor_data&, reactor_op* op)
 129 {
 130   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 131
 132   op_queue_[op_type].enqueue_operation(descriptor, op);
 133   interrupter_.interrupt();
 134
 135   return 0;
 136 }
 137
 138 void select_reactor::move_descriptor(socket_type,
 139     select_reactor::per_descriptor_data&,
 140     select_reactor::per_descriptor_data&)
 141 {
 142 }
 143
 144 void select_reactor::start_op(int op_type, socket_type descriptor,
 145     select_reactor::per_descriptor_data&, reactor_op* op,
 146     bool is_continuation, bool)
 147 {
 148   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 149
 150   if (shutdown_)
 151   {
 152     post_immediate_completion(op, is_continuation);
 153     return;
 154   }
 155
 156   bool first = op_queue_[op_type].enqueue_operation(descriptor, op);
 157   scheduler_.work_started();
 158   if (first)
 159     interrupter_.interrupt();
 160 }
 161
 162 void select_reactor::cancel_ops(socket_type descriptor,
 163     select_reactor::per_descriptor_data&)
 164 {
 165   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 166   cancel_ops_unlocked(descriptor, boost::asio::error::operation_aborted);
 167 }
 168
 169 void select_reactor::deregister_descriptor(socket_type descriptor,
 170     select_reactor::per_descriptor_data&, bool)
 171 {
 172   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 173   cancel_ops_unlocked(descriptor, boost::asio::error::operation_aborted);
 174 }
 175
 176 void select_reactor::deregister_internal_descriptor(
 177     socket_type descriptor, select_reactor::per_descriptor_data&)
 178 {
 179   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 180   op_queue<operation> ops;
 181   for (int i = 0; i < max_ops; ++i)
 182     op_queue_[i].cancel_operations(descriptor, ops);
 183 }
 184
 185 void select_reactor::cleanup_descriptor_data(
 186     select_reactor::per_descriptor_data&)
 187 {
 188 }
 189
 190 void select_reactor::run(long usec, op_queue<operation>& ops)
 191 {
 192   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 193
 194 #if defined(BOOST_ASIO_HAS_IOCP)
 195   // Check if the thread is supposed to stop.
 196   if (stop_thread_)
 197     return;
 198 #endif // defined(BOOST_ASIO_HAS_IOCP)
 199
 200   // Set up the descriptor sets.
 201   for (int i = 0; i < max_select_ops; ++i)
 202     fd_sets_[i].reset();
 203   fd_sets_[read_op].set(interrupter_.read_descriptor());
 204   socket_type max_fd = 0;
 205   bool have_work_to_do = !timer_queues_.all_empty();
 206   for (int i = 0; i < max_select_ops; ++i)
 207   {
 208     have_work_to_do = have_work_to_do || !op_queue_[i].empty();
 209     fd_sets_[i].set(op_queue_[i], ops);
 210     if (fd_sets_[i].max_descriptor() > max_fd)
 211       max_fd = fd_sets_[i].max_descriptor();
 212   }
 213
 214 #if defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
 215   // Connection operations on Windows use both except and write fd_sets.
 216   have_work_to_do = have_work_to_do || !op_queue_[connect_op].empty();
 217   fd_sets_[write_op].set(op_queue_[connect_op], ops);
 218   if (fd_sets_[write_op].max_descriptor() > max_fd)
 219     max_fd = fd_sets_[write_op].max_descriptor();
 220   fd_sets_[except_op].set(op_queue_[connect_op], ops);
 221   if (fd_sets_[except_op].max_descriptor() > max_fd)
 222     max_fd = fd_sets_[except_op].max_descriptor();
 223 #endif // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
 224
 225   // We can return immediately if there's no work to do and the reactor is
 226   // not supposed to block.
 227   if (!usec && !have_work_to_do)
 228     return;
 229
 230   // Determine how long to block while waiting for events.
 231   timeval tv_buf = { 0, 0 };
 232   timeval* tv = usec ? get_timeout(usec, tv_buf) : &tv_buf;
 233
 234   lock.unlock();
 235
 236   // Block on the select call until descriptors become ready.
 237   boost::system::error_code ec;
 238   int retval = socket_ops::select(static_cast<int>(max_fd + 1),
 239       fd_sets_[read_op], fd_sets_[write_op], fd_sets_[except_op], tv, ec);
 240
 241   // Reset the interrupter.
 242   if (retval > 0 && fd_sets_[read_op].is_set(interrupter_.read_descriptor()))
 243   {
 244     if (!interrupter_.reset())
 245     {
 246       lock.lock();
 247       interrupter_.recreate();
 248     }
 249     --retval;
 250   }
 251
 252   lock.lock();
 253
 254   // Dispatch all ready operations.
 255   if (retval > 0)
 256   {
 257 #if defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
 258     // Connection operations on Windows use both except and write fd_sets.
 259     fd_sets_[except_op].perform(op_queue_[connect_op], ops);
 260     fd_sets_[write_op].perform(op_queue_[connect_op], ops);
 261 #endif // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
 262
 263     // Exception operations must be processed first to ensure that any
 264     // out-of-band data is read before normal data.
 265     for (int i = max_select_ops - 1; i >= 0; --i)
 266       fd_sets_[i].perform(op_queue_[i], ops);
 267   }
 268   timer_queues_.get_ready_timers(ops);
 269 }
 270
 271 void select_reactor::interrupt()
 272 {
 273   interrupter_.interrupt();
 274 }
 275
 276 #if defined(BOOST_ASIO_HAS_IOCP)
 277 void select_reactor::run_thread()
 278 {
 279   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 280   while (!stop_thread_)
 281   {
 282     lock.unlock();
 283     op_queue<operation> ops;
 284     run(true, ops);
 285     scheduler_.post_deferred_completions(ops);
 286     lock.lock();
 287   }
 288 }
 289 #endif // defined(BOOST_ASIO_HAS_IOCP)
 290
 291 void select_reactor::do_add_timer_queue(timer_queue_base& queue)
 292 {
 293   mutex::scoped_lock lock(mutex_);
 294   timer_queues_.insert(&queue);
 295 }
 296
 297 void select_reactor::do_remove_timer_queue(timer_queue_base& queue)
 298 {
 299   mutex::scoped_lock lock(mutex_);
 300   timer_queues_.erase(&queue);
 301 }
 302
 303 timeval* select_reactor::get_timeout(long usec, timeval& tv)
 304 {
 305   // By default we will wait no longer than 5 minutes. This will ensure that
 306   // any changes to the system clock are detected after no longer than this.
 307   const long max_usec = 5 * 60 * 1000 * 1000;
 308   usec = timer_queues_.wait_duration_usec(
 309       (usec < 0 || max_usec < usec) ? max_usec : usec);
 310   tv.tv_sec = usec / 1000000;
 311   tv.tv_usec = usec % 1000000;
 312   return &tv;
 313 }
 314
 315 void select_reactor::cancel_ops_unlocked(socket_type descriptor,
 316     const boost::system::error_code& ec)
 317 {
 318   bool need_interrupt = false;
 319   op_queue<operation> ops;
 320   for (int i = 0; i < max_ops; ++i)
 321     need_interrupt = op_queue_[i].cancel_operations(
 322         descriptor, ops, ec) || need_interrupt;
 323   scheduler_.post_deferred_completions(ops);
 324   if (need_interrupt)
 325     interrupter_.interrupt();
 326 }
 327
 328 } // namespace detail
 329 } // namespace asio
 330 } // namespace boost
 331
 332 #include <boost/asio/detail/pop_options.hpp>
 333
 334 #endif // defined(BOOST_ASIO_HAS_IOCP)
 335        //   || (!defined(BOOST_ASIO_HAS_DEV_POLL)
 336        //       && !defined(BOOST_ASIO_HAS_EPOLL)
 337        //       && !defined(BOOST_ASIO_HAS_KQUEUE))
 338        //       && !defined(BOOST_ASIO_WINDOWS_RUNTIME))
 339
 340 #endif // BOOST_ASIO_DETAIL_IMPL_SELECT_REACTOR_IPP