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

   1 //
   2 // detail/impl/win_iocp_io_context.ipp
   3 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
   4 //
   5 // Copyright (c) 2003-2019 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_WIN_IOCP_IO_CONTEXT_IPP
  12 #define BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_CONTEXT_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
  22 #include <boost/asio/error.hpp>
  23 #include <boost/asio/detail/cstdint.hpp>
  24 #include <boost/asio/detail/handler_alloc_helpers.hpp>
  25 #include <boost/asio/detail/handler_invoke_helpers.hpp>
  26 #include <boost/asio/detail/limits.hpp>
  27 #include <boost/asio/detail/thread.hpp>
  28 #include <boost/asio/detail/throw_error.hpp>
  29 #include <boost/asio/detail/win_iocp_io_context.hpp>
  30
  31 #include <boost/asio/detail/push_options.hpp>
  32
  33 namespace boost {
  34 namespace asio {
  35 namespace detail {
  36
  37 struct win_iocp_io_context::thread_function
  38 {
  39   explicit thread_function(win_iocp_io_context* s)
  40     : this_(s)
  41   {
  42   }
  43
  44   void operator()()
  45   {
  46     boost::system::error_code ec;
  47     this_->run(ec);
  48   }
  49
  50   win_iocp_io_context* this_;
  51 };
  52
  53 struct win_iocp_io_context::work_finished_on_block_exit
  54 {
  55   ~work_finished_on_block_exit()
  56   {
  57     io_context_->work_finished();
  58   }
  59
  60   win_iocp_io_context* io_context_;
  61 };
  62
  63 struct win_iocp_io_context::timer_thread_function
  64 {
  65   void operator()()
  66   {
  67     while (::InterlockedExchangeAdd(&io_context_->shutdown_, 0) == 0)
  68     {
  69       if (::WaitForSingleObject(io_context_->waitable_timer_.handle,
  70             INFINITE) == WAIT_OBJECT_0)
  71       {
  72         ::InterlockedExchange(&io_context_->dispatch_required_, 1);
  73         ::PostQueuedCompletionStatus(io_context_->iocp_.handle,
  74             0, wake_for_dispatch, 0);
  75       }
  76     }
  77   }
  78
  79   win_iocp_io_context* io_context_;
  80 };
  81
  82 win_iocp_io_context::win_iocp_io_context(
  83     boost::asio::execution_context& ctx, int concurrency_hint, bool own_thread)
  84   : execution_context_service_base<win_iocp_io_context>(ctx),
  85     iocp_(),
  86     outstanding_work_(0),
  87     stopped_(0),
  88     stop_event_posted_(0),
  89     shutdown_(0),
  90     gqcs_timeout_(get_gqcs_timeout()),
  91     dispatch_required_(0),
  92     concurrency_hint_(concurrency_hint)
  93 {
  94   BOOST_ASIO_HANDLER_TRACKING_INIT;
  95
  96   iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0,
  97       static_cast<DWORD>(concurrency_hint >= 0 ? concurrency_hint : DWORD(~0)));
  98   if (!iocp_.handle)
  99   {
 100     DWORD last_error = ::GetLastError();
 101     boost::system::error_code ec(last_error,
 102         boost::asio::error::get_system_category());
 103     boost::asio::detail::throw_error(ec, "iocp");
 104   }
 105
 106   if (own_thread)
 107   {
 108     ::InterlockedIncrement(&outstanding_work_);
 109     thread_.reset(new boost::asio::detail::thread(thread_function(this)));
 110   }
 111 }
 112
 113 win_iocp_io_context::~win_iocp_io_context()
 114 {
 115   if (thread_.get())
 116   {
 117     thread_->join();
 118     thread_.reset();
 119   }
 120 }
 121
 122 void win_iocp_io_context::shutdown()
 123 {
 124   ::InterlockedExchange(&shutdown_, 1);
 125
 126   if (timer_thread_.get())
 127   {
 128     LARGE_INTEGER timeout;
 129     timeout.QuadPart = 1;
 130     ::SetWaitableTimer(waitable_timer_.handle, &timeout, 1, 0, 0, FALSE);
 131   }
 132
 133   if (thread_.get())
 134   {
 135     thread_->join();
 136     thread_.reset();
 137     ::InterlockedDecrement(&outstanding_work_);
 138   }
 139
 140   while (::InterlockedExchangeAdd(&outstanding_work_, 0) > 0)
 141   {
 142     op_queue<win_iocp_operation> ops;
 143     timer_queues_.get_all_timers(ops);
 144     ops.push(completed_ops_);
 145     if (!ops.empty())
 146     {
 147       while (win_iocp_operation* op = ops.front())
 148       {
 149         ops.pop();
 150         ::InterlockedDecrement(&outstanding_work_);
 151         op->destroy();
 152       }
 153     }
 154     else
 155     {
 156       DWORD bytes_transferred = 0;
 157       dword_ptr_t completion_key = 0;
 158       LPOVERLAPPED overlapped = 0;
 159       ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
 160           &completion_key, &overlapped, gqcs_timeout_);
 161       if (overlapped)
 162       {
 163         ::InterlockedDecrement(&outstanding_work_);
 164         static_cast<win_iocp_operation*>(overlapped)->destroy();
 165       }
 166     }
 167   }
 168
 169   if (timer_thread_.get())
 170     timer_thread_->join();
 171 }
 172
 173 boost::system::error_code win_iocp_io_context::register_handle(
 174     HANDLE handle, boost::system::error_code& ec)
 175 {
 176   if (::CreateIoCompletionPort(handle, iocp_.handle, 0, 0) == 0)
 177   {
 178     DWORD last_error = ::GetLastError();
 179     ec = boost::system::error_code(last_error,
 180         boost::asio::error::get_system_category());
 181   }
 182   else
 183   {
 184     ec = boost::system::error_code();
 185   }
 186   return ec;
 187 }
 188
 189 size_t win_iocp_io_context::run(boost::system::error_code& ec)
 190 {
 191   if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
 192   {
 193     stop();
 194     ec = boost::system::error_code();
 195     return 0;
 196   }
 197
 198   win_iocp_thread_info this_thread;
 199   thread_call_stack::context ctx(this, this_thread);
 200
 201   size_t n = 0;
 202   while (do_one(INFINITE, ec))
 203     if (n != (std::numeric_limits<size_t>::max)())
 204       ++n;
 205   return n;
 206 }
 207
 208 size_t win_iocp_io_context::run_one(boost::system::error_code& ec)
 209 {
 210   if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
 211   {
 212     stop();
 213     ec = boost::system::error_code();
 214     return 0;
 215   }
 216
 217   win_iocp_thread_info this_thread;
 218   thread_call_stack::context ctx(this, this_thread);
 219
 220   return do_one(INFINITE, ec);
 221 }
 222
 223 size_t win_iocp_io_context::wait_one(long usec, boost::system::error_code& ec)
 224 {
 225   if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
 226   {
 227     stop();
 228     ec = boost::system::error_code();
 229     return 0;
 230   }
 231
 232   win_iocp_thread_info this_thread;
 233   thread_call_stack::context ctx(this, this_thread);
 234
 235   return do_one(usec < 0 ? INFINITE : ((usec - 1) / 1000 + 1), ec);
 236 }
 237
 238 size_t win_iocp_io_context::poll(boost::system::error_code& ec)
 239 {
 240   if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
 241   {
 242     stop();
 243     ec = boost::system::error_code();
 244     return 0;
 245   }
 246
 247   win_iocp_thread_info this_thread;
 248   thread_call_stack::context ctx(this, this_thread);
 249
 250   size_t n = 0;
 251   while (do_one(0, ec))
 252     if (n != (std::numeric_limits<size_t>::max)())
 253       ++n;
 254   return n;
 255 }
 256
 257 size_t win_iocp_io_context::poll_one(boost::system::error_code& ec)
 258 {
 259   if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
 260   {
 261     stop();
 262     ec = boost::system::error_code();
 263     return 0;
 264   }
 265
 266   win_iocp_thread_info this_thread;
 267   thread_call_stack::context ctx(this, this_thread);
 268
 269   return do_one(0, ec);
 270 }
 271
 272 void win_iocp_io_context::stop()
 273 {
 274   if (::InterlockedExchange(&stopped_, 1) == 0)
 275   {
 276     if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
 277     {
 278       if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
 279       {
 280         DWORD last_error = ::GetLastError();
 281         boost::system::error_code ec(last_error,
 282             boost::asio::error::get_system_category());
 283         boost::asio::detail::throw_error(ec, "pqcs");
 284       }
 285     }
 286   }
 287 }
 288
 289 void win_iocp_io_context::post_deferred_completion(win_iocp_operation* op)
 290 {
 291   // Flag the operation as ready.
 292   op->ready_ = 1;
 293
 294   // Enqueue the operation on the I/O completion port.
 295   if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, op))
 296   {
 297     // Out of resources. Put on completed queue instead.
 298     mutex::scoped_lock lock(dispatch_mutex_);
 299     completed_ops_.push(op);
 300     ::InterlockedExchange(&dispatch_required_, 1);
 301   }
 302 }
 303
 304 void win_iocp_io_context::post_deferred_completions(
 305     op_queue<win_iocp_operation>& ops)
 306 {
 307   while (win_iocp_operation* op = ops.front())
 308   {
 309     ops.pop();
 310
 311     // Flag the operation as ready.
 312     op->ready_ = 1;
 313
 314     // Enqueue the operation on the I/O completion port.
 315     if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, op))
 316     {
 317       // Out of resources. Put on completed queue instead.
 318       mutex::scoped_lock lock(dispatch_mutex_);
 319       completed_ops_.push(op);
 320       completed_ops_.push(ops);
 321       ::InterlockedExchange(&dispatch_required_, 1);
 322     }
 323   }
 324 }
 325
 326 void win_iocp_io_context::abandon_operations(
 327     op_queue<win_iocp_operation>& ops)
 328 {
 329   while (win_iocp_operation* op = ops.front())
 330   {
 331     ops.pop();
 332     ::InterlockedDecrement(&outstanding_work_);
 333     op->destroy();
 334   }
 335 }
 336
 337 void win_iocp_io_context::on_pending(win_iocp_operation* op)
 338 {
 339   if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
 340   {
 341     // Enqueue the operation on the I/O completion port.
 342     if (!::PostQueuedCompletionStatus(iocp_.handle,
 343           0, overlapped_contains_result, op))
 344     {
 345       // Out of resources. Put on completed queue instead.
 346       mutex::scoped_lock lock(dispatch_mutex_);
 347       completed_ops_.push(op);
 348       ::InterlockedExchange(&dispatch_required_, 1);
 349     }
 350   }
 351 }
 352
 353 void win_iocp_io_context::on_completion(win_iocp_operation* op,
 354     DWORD last_error, DWORD bytes_transferred)
 355 {
 356   // Flag that the operation is ready for invocation.
 357   op->ready_ = 1;
 358
 359   // Store results in the OVERLAPPED structure.
 360   op->Internal = reinterpret_cast<ulong_ptr_t>(
 361       &boost::asio::error::get_system_category());
 362   op->Offset = last_error;
 363   op->OffsetHigh = bytes_transferred;
 364
 365   // Enqueue the operation on the I/O completion port.
 366   if (!::PostQueuedCompletionStatus(iocp_.handle,
 367         0, overlapped_contains_result, op))
 368   {
 369     // Out of resources. Put on completed queue instead.
 370     mutex::scoped_lock lock(dispatch_mutex_);
 371     completed_ops_.push(op);
 372     ::InterlockedExchange(&dispatch_required_, 1);
 373   }
 374 }
 375
 376 void win_iocp_io_context::on_completion(win_iocp_operation* op,
 377     const boost::system::error_code& ec, DWORD bytes_transferred)
 378 {
 379   // Flag that the operation is ready for invocation.
 380   op->ready_ = 1;
 381
 382   // Store results in the OVERLAPPED structure.
 383   op->Internal = reinterpret_cast<ulong_ptr_t>(&ec.category());
 384   op->Offset = ec.value();
 385   op->OffsetHigh = bytes_transferred;
 386
 387   // Enqueue the operation on the I/O completion port.
 388   if (!::PostQueuedCompletionStatus(iocp_.handle,
 389         0, overlapped_contains_result, op))
 390   {
 391     // Out of resources. Put on completed queue instead.
 392     mutex::scoped_lock lock(dispatch_mutex_);
 393     completed_ops_.push(op);
 394     ::InterlockedExchange(&dispatch_required_, 1);
 395   }
 396 }
 397
 398 size_t win_iocp_io_context::do_one(DWORD msec, boost::system::error_code& ec)
 399 {
 400   for (;;)
 401   {
 402     // Try to acquire responsibility for dispatching timers and completed ops.
 403     if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1)
 404     {
 405       mutex::scoped_lock lock(dispatch_mutex_);
 406
 407       // Dispatch pending timers and operations.
 408       op_queue<win_iocp_operation> ops;
 409       ops.push(completed_ops_);
 410       timer_queues_.get_ready_timers(ops);
 411       post_deferred_completions(ops);
 412       update_timeout();
 413     }
 414
 415     // Get the next operation from the queue.
 416     DWORD bytes_transferred = 0;
 417     dword_ptr_t completion_key = 0;
 418     LPOVERLAPPED overlapped = 0;
 419     ::SetLastError(0);
 420     BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle,
 421         &bytes_transferred, &completion_key, &overlapped,
 422         msec < gqcs_timeout_ ? msec : gqcs_timeout_);
 423     DWORD last_error = ::GetLastError();
 424
 425     if (overlapped)
 426     {
 427       win_iocp_operation* op = static_cast<win_iocp_operation*>(overlapped);
 428       boost::system::error_code result_ec(last_error,
 429           boost::asio::error::get_system_category());
 430
 431       // We may have been passed the last_error and bytes_transferred in the
 432       // OVERLAPPED structure itself.
 433       if (completion_key == overlapped_contains_result)
 434       {
 435         result_ec = boost::system::error_code(static_cast<int>(op->Offset),
 436             *reinterpret_cast<boost::system::error_category*>(op->Internal));
 437         bytes_transferred = op->OffsetHigh;
 438       }
 439
 440       // Otherwise ensure any result has been saved into the OVERLAPPED
 441       // structure.
 442       else
 443       {
 444         op->Internal = reinterpret_cast<ulong_ptr_t>(&result_ec.category());
 445         op->Offset = result_ec.value();
 446         op->OffsetHigh = bytes_transferred;
 447       }
 448
 449       // Dispatch the operation only if ready. The operation may not be ready
 450       // if the initiating function (e.g. a call to WSARecv) has not yet
 451       // returned. This is because the initiating function still wants access
 452       // to the operation's OVERLAPPED structure.
 453       if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
 454       {
 455         // Ensure the count of outstanding work is decremented on block exit.
 456         work_finished_on_block_exit on_exit = { this };
 457         (void)on_exit;
 458
 459         op->complete(this, result_ec, bytes_transferred);
 460         ec = boost::system::error_code();
 461         return 1;
 462       }
 463     }
 464     else if (!ok)
 465     {
 466       if (last_error != WAIT_TIMEOUT)
 467       {
 468         ec = boost::system::error_code(last_error,
 469             boost::asio::error::get_system_category());
 470         return 0;
 471       }
 472
 473       // If we're waiting indefinitely we need to keep going until we get a
 474       // real handler.
 475       if (msec == INFINITE)
 476         continue;
 477
 478       ec = boost::system::error_code();
 479       return 0;
 480     }
 481     else if (completion_key == wake_for_dispatch)
 482     {
 483       // We have been woken up to try to acquire responsibility for dispatching
 484       // timers and completed operations.
 485     }
 486     else
 487     {
 488       // Indicate that there is no longer an in-flight stop event.
 489       ::InterlockedExchange(&stop_event_posted_, 0);
 490
 491       // The stopped_ flag is always checked to ensure that any leftover
 492       // stop events from a previous run invocation are ignored.
 493       if (::InterlockedExchangeAdd(&stopped_, 0) != 0)
 494       {
 495         // Wake up next thread that is blocked on GetQueuedCompletionStatus.
 496         if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
 497         {
 498           if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
 499           {
 500             last_error = ::GetLastError();
 501             ec = boost::system::error_code(last_error,
 502                 boost::asio::error::get_system_category());
 503             return 0;
 504           }
 505         }
 506
 507         ec = boost::system::error_code();
 508         return 0;
 509       }
 510     }
 511   }
 512 }
 513
 514 DWORD win_iocp_io_context::get_gqcs_timeout()
 515 {
 516   OSVERSIONINFOEX osvi;
 517   ZeroMemory(&osvi, sizeof(osvi));
 518   osvi.dwOSVersionInfoSize = sizeof(osvi);
 519   osvi.dwMajorVersion = 6ul;
 520
 521   const uint64_t condition_mask = ::VerSetConditionMask(
 522       0, VER_MAJORVERSION, VER_GREATER_EQUAL);
 523
 524   if (!!::VerifyVersionInfo(&osvi, VER_MAJORVERSION, condition_mask))
 525     return INFINITE;
 526
 527   return default_gqcs_timeout;
 528 }
 529
 530 void win_iocp_io_context::do_add_timer_queue(timer_queue_base& queue)
 531 {
 532   mutex::scoped_lock lock(dispatch_mutex_);
 533
 534   timer_queues_.insert(&queue);
 535
 536   if (!waitable_timer_.handle)
 537   {
 538     waitable_timer_.handle = ::CreateWaitableTimer(0, FALSE, 0);
 539     if (waitable_timer_.handle == 0)
 540     {
 541       DWORD last_error = ::GetLastError();
 542       boost::system::error_code ec(last_error,
 543           boost::asio::error::get_system_category());
 544       boost::asio::detail::throw_error(ec, "timer");
 545     }
 546
 547     LARGE_INTEGER timeout;
 548     timeout.QuadPart = -max_timeout_usec;
 549     timeout.QuadPart *= 10;
 550     ::SetWaitableTimer(waitable_timer_.handle,
 551         &timeout, max_timeout_msec, 0, 0, FALSE);
 552   }
 553
 554   if (!timer_thread_.get())
 555   {
 556     timer_thread_function thread_function = { this };
 557     timer_thread_.reset(new thread(thread_function, 65536));
 558   }
 559 }
 560
 561 void win_iocp_io_context::do_remove_timer_queue(timer_queue_base& queue)
 562 {
 563   mutex::scoped_lock lock(dispatch_mutex_);
 564
 565   timer_queues_.erase(&queue);
 566 }
 567
 568 void win_iocp_io_context::update_timeout()
 569 {
 570   if (timer_thread_.get())
 571   {
 572     // There's no point updating the waitable timer if the new timeout period
 573     // exceeds the maximum timeout. In that case, we might as well wait for the
 574     // existing period of the timer to expire.
 575     long timeout_usec = timer_queues_.wait_duration_usec(max_timeout_usec);
 576     if (timeout_usec < max_timeout_usec)
 577     {
 578       LARGE_INTEGER timeout;
 579       timeout.QuadPart = -timeout_usec;
 580       timeout.QuadPart *= 10;
 581       ::SetWaitableTimer(waitable_timer_.handle,
 582           &timeout, max_timeout_msec, 0, 0, FALSE);
 583     }
 584   }
 585 }
 586
 587 } // namespace detail
 588 } // namespace asio
 589 } // namespace boost
 590
 591 #include <boost/asio/detail/pop_options.hpp>
 592
 593 #endif // defined(BOOST_ASIO_HAS_IOCP)
 594
 595 #endif // BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_CONTEXT_IPP