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