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

   1 //
   2 // detail/impl/win_iocp_socket_service_base.ipp
   3 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
   4 //
   5 // Copyright (c) 2003-2022 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_SOCKET_SERVICE_BASE_IPP
  12 #define BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_SOCKET_SERVICE_BASE_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/detail/win_iocp_socket_service_base.hpp>
  23
  24 #include <boost/asio/detail/push_options.hpp>
  25
  26 namespace boost {
  27 namespace asio {
  28 namespace detail {
  29
  30 win_iocp_socket_service_base::win_iocp_socket_service_base(
  31     execution_context& context)
  32   : context_(context),
  33     iocp_service_(use_service<win_iocp_io_context>(context)),
  34     reactor_(0),
  35     connect_ex_(0),
  36     nt_set_info_(0),
  37     mutex_(),
  38     impl_list_(0)
  39 {
  40 }
  41
  42 void win_iocp_socket_service_base::base_shutdown()
  43 {
  44   // Close all implementations, causing all operations to complete.
  45   boost::asio::detail::mutex::scoped_lock lock(mutex_);
  46   base_implementation_type* impl = impl_list_;
  47   while (impl)
  48   {
  49     close_for_destruction(*impl);
  50     impl = impl->next_;
  51   }
  52 }
  53
  54 void win_iocp_socket_service_base::construct(
  55     win_iocp_socket_service_base::base_implementation_type& impl)
  56 {
  57   impl.socket_ = invalid_socket;
  58   impl.state_ = 0;
  59   impl.cancel_token_.reset();
  60 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
  61   impl.safe_cancellation_thread_id_ = 0;
  62 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
  63
  64   // Insert implementation into linked list of all implementations.
  65   boost::asio::detail::mutex::scoped_lock lock(mutex_);
  66   impl.next_ = impl_list_;
  67   impl.prev_ = 0;
  68   if (impl_list_)
  69     impl_list_->prev_ = &impl;
  70   impl_list_ = &impl;
  71 }
  72
  73 void win_iocp_socket_service_base::base_move_construct(
  74     win_iocp_socket_service_base::base_implementation_type& impl,
  75     win_iocp_socket_service_base::base_implementation_type& other_impl)
  76   BOOST_ASIO_NOEXCEPT
  77 {
  78   impl.socket_ = other_impl.socket_;
  79   other_impl.socket_ = invalid_socket;
  80
  81   impl.state_ = other_impl.state_;
  82   other_impl.state_ = 0;
  83
  84   impl.cancel_token_ = other_impl.cancel_token_;
  85   other_impl.cancel_token_.reset();
  86
  87 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
  88   impl.safe_cancellation_thread_id_ = other_impl.safe_cancellation_thread_id_;
  89   other_impl.safe_cancellation_thread_id_ = 0;
  90 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
  91
  92   // Insert implementation into linked list of all implementations.
  93   boost::asio::detail::mutex::scoped_lock lock(mutex_);
  94   impl.next_ = impl_list_;
  95   impl.prev_ = 0;
  96   if (impl_list_)
  97     impl_list_->prev_ = &impl;
  98   impl_list_ = &impl;
  99 }
 100
 101 void win_iocp_socket_service_base::base_move_assign(
 102     win_iocp_socket_service_base::base_implementation_type& impl,
 103     win_iocp_socket_service_base& other_service,
 104     win_iocp_socket_service_base::base_implementation_type& other_impl)
 105 {
 106   close_for_destruction(impl);
 107
 108   if (this != &other_service)
 109   {
 110     // Remove implementation from linked list of all implementations.
 111     boost::asio::detail::mutex::scoped_lock lock(mutex_);
 112     if (impl_list_ == &impl)
 113       impl_list_ = impl.next_;
 114     if (impl.prev_)
 115       impl.prev_->next_ = impl.next_;
 116     if (impl.next_)
 117       impl.next_->prev_= impl.prev_;
 118     impl.next_ = 0;
 119     impl.prev_ = 0;
 120   }
 121
 122   impl.socket_ = other_impl.socket_;
 123   other_impl.socket_ = invalid_socket;
 124
 125   impl.state_ = other_impl.state_;
 126   other_impl.state_ = 0;
 127
 128   impl.cancel_token_ = other_impl.cancel_token_;
 129   other_impl.cancel_token_.reset();
 130
 131 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
 132   impl.safe_cancellation_thread_id_ = other_impl.safe_cancellation_thread_id_;
 133   other_impl.safe_cancellation_thread_id_ = 0;
 134 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
 135
 136   if (this != &other_service)
 137   {
 138     // Insert implementation into linked list of all implementations.
 139     boost::asio::detail::mutex::scoped_lock lock(other_service.mutex_);
 140     impl.next_ = other_service.impl_list_;
 141     impl.prev_ = 0;
 142     if (other_service.impl_list_)
 143       other_service.impl_list_->prev_ = &impl;
 144     other_service.impl_list_ = &impl;
 145   }
 146 }
 147
 148 void win_iocp_socket_service_base::destroy(
 149     win_iocp_socket_service_base::base_implementation_type& impl)
 150 {
 151   close_for_destruction(impl);
 152
 153   // Remove implementation from linked list of all implementations.
 154   boost::asio::detail::mutex::scoped_lock lock(mutex_);
 155   if (impl_list_ == &impl)
 156     impl_list_ = impl.next_;
 157   if (impl.prev_)
 158     impl.prev_->next_ = impl.next_;
 159   if (impl.next_)
 160     impl.next_->prev_= impl.prev_;
 161   impl.next_ = 0;
 162   impl.prev_ = 0;
 163 }
 164
 165 boost::system::error_code win_iocp_socket_service_base::close(
 166     win_iocp_socket_service_base::base_implementation_type& impl,
 167     boost::system::error_code& ec)
 168 {
 169   if (is_open(impl))
 170   {
 171     BOOST_ASIO_HANDLER_OPERATION((iocp_service_.context(),
 172           "socket", &impl, impl.socket_, "close"));
 173
 174     // Check if the reactor was created, in which case we need to close the
 175     // socket on the reactor as well to cancel any operations that might be
 176     // running there.
 177     select_reactor* r = static_cast<select_reactor*>(
 178           interlocked_compare_exchange_pointer(
 179             reinterpret_cast<void**>(&reactor_), 0, 0));
 180     if (r)
 181       r->deregister_descriptor(impl.socket_, impl.reactor_data_, true);
 182
 183     socket_ops::close(impl.socket_, impl.state_, false, ec);
 184
 185     if (r)
 186       r->cleanup_descriptor_data(impl.reactor_data_);
 187   }
 188   else
 189   {
 190     ec = boost::system::error_code();
 191   }
 192
 193   impl.socket_ = invalid_socket;
 194   impl.state_ = 0;
 195   impl.cancel_token_.reset();
 196 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
 197   impl.safe_cancellation_thread_id_ = 0;
 198 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
 199
 200   return ec;
 201 }
 202
 203 socket_type win_iocp_socket_service_base::release(
 204     win_iocp_socket_service_base::base_implementation_type& impl,
 205     boost::system::error_code& ec)
 206 {
 207   if (!is_open(impl))
 208     return invalid_socket;
 209
 210   cancel(impl, ec);
 211   if (ec)
 212     return invalid_socket;
 213
 214   nt_set_info_fn fn = get_nt_set_info();
 215   if (fn == 0)
 216   {
 217     ec = boost::asio::error::operation_not_supported;
 218     return invalid_socket;
 219   }
 220
 221   HANDLE sock_as_handle = reinterpret_cast<HANDLE>(impl.socket_);
 222   ULONG_PTR iosb[2] = { 0, 0 };
 223   void* info[2] = { 0, 0 };
 224   if (fn(sock_as_handle, iosb, &info, sizeof(info),
 225         61 /* FileReplaceCompletionInformation */))
 226   {
 227     ec = boost::asio::error::operation_not_supported;
 228     return invalid_socket;
 229   }
 230
 231   socket_type tmp = impl.socket_;
 232   impl.socket_ = invalid_socket;
 233   return tmp;
 234 }
 235
 236 boost::system::error_code win_iocp_socket_service_base::cancel(
 237     win_iocp_socket_service_base::base_implementation_type& impl,
 238     boost::system::error_code& ec)
 239 {
 240   if (!is_open(impl))
 241   {
 242     ec = boost::asio::error::bad_descriptor;
 243     return ec;
 244   }
 245
 246   BOOST_ASIO_HANDLER_OPERATION((iocp_service_.context(),
 247         "socket", &impl, impl.socket_, "cancel"));
 248
 249   if (FARPROC cancel_io_ex_ptr = ::GetProcAddress(
 250         ::GetModuleHandleA("KERNEL32"), "CancelIoEx"))
 251   {
 252     // The version of Windows supports cancellation from any thread.
 253     typedef BOOL (WINAPI* cancel_io_ex_t)(HANDLE, LPOVERLAPPED);
 254     cancel_io_ex_t cancel_io_ex = reinterpret_cast<cancel_io_ex_t>(
 255         reinterpret_cast<void*>(cancel_io_ex_ptr));
 256     socket_type sock = impl.socket_;
 257     HANDLE sock_as_handle = reinterpret_cast<HANDLE>(sock);
 258     if (!cancel_io_ex(sock_as_handle, 0))
 259     {
 260       DWORD last_error = ::GetLastError();
 261       if (last_error == ERROR_NOT_FOUND)
 262       {
 263         // ERROR_NOT_FOUND means that there were no operations to be
 264         // cancelled. We swallow this error to match the behaviour on other
 265         // platforms.
 266         ec = boost::system::error_code();
 267       }
 268       else
 269       {
 270         ec = boost::system::error_code(last_error,
 271             boost::asio::error::get_system_category());
 272       }
 273     }
 274     else
 275     {
 276       ec = boost::system::error_code();
 277     }
 278   }
 279 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
 280   else if (impl.safe_cancellation_thread_id_ == 0)
 281   {
 282     // No operations have been started, so there's nothing to cancel.
 283     ec = boost::system::error_code();
 284   }
 285   else if (impl.safe_cancellation_thread_id_ == ::GetCurrentThreadId())
 286   {
 287     // Asynchronous operations have been started from the current thread only,
 288     // so it is safe to try to cancel them using CancelIo.
 289     socket_type sock = impl.socket_;
 290     HANDLE sock_as_handle = reinterpret_cast<HANDLE>(sock);
 291     if (!::CancelIo(sock_as_handle))
 292     {
 293       DWORD last_error = ::GetLastError();
 294       ec = boost::system::error_code(last_error,
 295           boost::asio::error::get_system_category());
 296     }
 297     else
 298     {
 299       ec = boost::system::error_code();
 300     }
 301   }
 302   else
 303   {
 304     // Asynchronous operations have been started from more than one thread,
 305     // so cancellation is not safe.
 306     ec = boost::asio::error::operation_not_supported;
 307   }
 308 #else // defined(BOOST_ASIO_ENABLE_CANCELIO)
 309   else
 310   {
 311     // Cancellation is not supported as CancelIo may not be used.
 312     ec = boost::asio::error::operation_not_supported;
 313   }
 314 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
 315
 316   // Cancel any operations started via the reactor.
 317   if (!ec)
 318   {
 319     select_reactor* r = static_cast<select_reactor*>(
 320           interlocked_compare_exchange_pointer(
 321             reinterpret_cast<void**>(&reactor_), 0, 0));
 322     if (r)
 323       r->cancel_ops(impl.socket_, impl.reactor_data_);
 324   }
 325
 326   return ec;
 327 }
 328
 329 boost::system::error_code win_iocp_socket_service_base::do_open(
 330     win_iocp_socket_service_base::base_implementation_type& impl,
 331     int family, int type, int protocol, boost::system::error_code& ec)
 332 {
 333   if (is_open(impl))
 334   {
 335     ec = boost::asio::error::already_open;
 336     return ec;
 337   }
 338
 339   socket_holder sock(socket_ops::socket(family, type, protocol, ec));
 340   if (sock.get() == invalid_socket)
 341     return ec;
 342
 343   HANDLE sock_as_handle = reinterpret_cast<HANDLE>(sock.get());
 344   if (iocp_service_.register_handle(sock_as_handle, ec))
 345     return ec;
 346
 347   impl.socket_ = sock.release();
 348   switch (type)
 349   {
 350   case SOCK_STREAM: impl.state_ = socket_ops::stream_oriented; break;
 351   case SOCK_DGRAM: impl.state_ = socket_ops::datagram_oriented; break;
 352   default: impl.state_ = 0; break;
 353   }
 354   impl.cancel_token_.reset(static_cast<void*>(0), socket_ops::noop_deleter());
 355   ec = boost::system::error_code();
 356   return ec;
 357 }
 358
 359 boost::system::error_code win_iocp_socket_service_base::do_assign(
 360     win_iocp_socket_service_base::base_implementation_type& impl,
 361     int type, socket_type native_socket, boost::system::error_code& ec)
 362 {
 363   if (is_open(impl))
 364   {
 365     ec = boost::asio::error::already_open;
 366     return ec;
 367   }
 368
 369   HANDLE sock_as_handle = reinterpret_cast<HANDLE>(native_socket);
 370   if (iocp_service_.register_handle(sock_as_handle, ec))
 371     return ec;
 372
 373   impl.socket_ = native_socket;
 374   switch (type)
 375   {
 376   case SOCK_STREAM: impl.state_ = socket_ops::stream_oriented; break;
 377   case SOCK_DGRAM: impl.state_ = socket_ops::datagram_oriented; break;
 378   default: impl.state_ = 0; break;
 379   }
 380   impl.cancel_token_.reset(static_cast<void*>(0), socket_ops::noop_deleter());
 381   ec = boost::system::error_code();
 382   return ec;
 383 }
 384
 385 void win_iocp_socket_service_base::start_send_op(
 386     win_iocp_socket_service_base::base_implementation_type& impl,
 387     WSABUF* buffers, std::size_t buffer_count,
 388     socket_base::message_flags flags, bool noop, operation* op)
 389 {
 390   update_cancellation_thread_id(impl);
 391   iocp_service_.work_started();
 392
 393   if (noop)
 394     iocp_service_.on_completion(op);
 395   else if (!is_open(impl))
 396     iocp_service_.on_completion(op, boost::asio::error::bad_descriptor);
 397   else
 398   {
 399     DWORD bytes_transferred = 0;
 400     int result = ::WSASend(impl.socket_, buffers,
 401         static_cast<DWORD>(buffer_count), &bytes_transferred, flags, op, 0);
 402     DWORD last_error = ::WSAGetLastError();
 403     if (last_error == ERROR_PORT_UNREACHABLE)
 404       last_error = WSAECONNREFUSED;
 405     if (result != 0 && last_error != WSA_IO_PENDING)
 406       iocp_service_.on_completion(op, last_error, bytes_transferred);
 407     else
 408       iocp_service_.on_pending(op);
 409   }
 410 }
 411
 412 void win_iocp_socket_service_base::start_send_to_op(
 413     win_iocp_socket_service_base::base_implementation_type& impl,
 414     WSABUF* buffers, std::size_t buffer_count,
 415     const socket_addr_type* addr, int addrlen,
 416     socket_base::message_flags flags, operation* op)
 417 {
 418   update_cancellation_thread_id(impl);
 419   iocp_service_.work_started();
 420
 421   if (!is_open(impl))
 422     iocp_service_.on_completion(op, boost::asio::error::bad_descriptor);
 423   else
 424   {
 425     DWORD bytes_transferred = 0;
 426     int result = ::WSASendTo(impl.socket_, buffers,
 427         static_cast<DWORD>(buffer_count),
 428         &bytes_transferred, flags, addr, addrlen, op, 0);
 429     DWORD last_error = ::WSAGetLastError();
 430     if (last_error == ERROR_PORT_UNREACHABLE)
 431       last_error = WSAECONNREFUSED;
 432     if (result != 0 && last_error != WSA_IO_PENDING)
 433       iocp_service_.on_completion(op, last_error, bytes_transferred);
 434     else
 435       iocp_service_.on_pending(op);
 436   }
 437 }
 438
 439 void win_iocp_socket_service_base::start_receive_op(
 440     win_iocp_socket_service_base::base_implementation_type& impl,
 441     WSABUF* buffers, std::size_t buffer_count,
 442     socket_base::message_flags flags, bool noop, operation* op)
 443 {
 444   update_cancellation_thread_id(impl);
 445   iocp_service_.work_started();
 446
 447   if (noop)
 448     iocp_service_.on_completion(op);
 449   else if (!is_open(impl))
 450     iocp_service_.on_completion(op, boost::asio::error::bad_descriptor);
 451   else
 452   {
 453     DWORD bytes_transferred = 0;
 454     DWORD recv_flags = flags;
 455     int result = ::WSARecv(impl.socket_, buffers,
 456         static_cast<DWORD>(buffer_count),
 457         &bytes_transferred, &recv_flags, op, 0);
 458     DWORD last_error = ::WSAGetLastError();
 459     if (last_error == ERROR_NETNAME_DELETED)
 460       last_error = WSAECONNRESET;
 461     else if (last_error == ERROR_PORT_UNREACHABLE)
 462       last_error = WSAECONNREFUSED;
 463     if (result != 0 && last_error != WSA_IO_PENDING)
 464       iocp_service_.on_completion(op, last_error, bytes_transferred);
 465     else
 466       iocp_service_.on_pending(op);
 467   }
 468 }
 469
 470 int win_iocp_socket_service_base::start_null_buffers_receive_op(
 471     win_iocp_socket_service_base::base_implementation_type& impl,
 472     socket_base::message_flags flags, reactor_op* op, operation* iocp_op)
 473 {
 474   if ((impl.state_ & socket_ops::stream_oriented) != 0)
 475   {
 476     // For stream sockets on Windows, we may issue a 0-byte overlapped
 477     // WSARecv to wait until there is data available on the socket.
 478     ::WSABUF buf = { 0, 0 };
 479     start_receive_op(impl, &buf, 1, flags, false, iocp_op);
 480     return -1;
 481   }
 482   else
 483   {
 484     int op_type = (flags & socket_base::message_out_of_band)
 485       ? select_reactor::except_op : select_reactor::read_op;
 486     start_reactor_op(impl, op_type, op);
 487     return op_type;
 488   }
 489 }
 490
 491 void win_iocp_socket_service_base::start_receive_from_op(
 492     win_iocp_socket_service_base::base_implementation_type& impl,
 493     WSABUF* buffers, std::size_t buffer_count, socket_addr_type* addr,
 494     socket_base::message_flags flags, int* addrlen, operation* op)
 495 {
 496   update_cancellation_thread_id(impl);
 497   iocp_service_.work_started();
 498
 499   if (!is_open(impl))
 500     iocp_service_.on_completion(op, boost::asio::error::bad_descriptor);
 501   else
 502   {
 503     DWORD bytes_transferred = 0;
 504     DWORD recv_flags = flags;
 505     int result = ::WSARecvFrom(impl.socket_, buffers,
 506         static_cast<DWORD>(buffer_count),
 507         &bytes_transferred, &recv_flags, addr, addrlen, op, 0);
 508     DWORD last_error = ::WSAGetLastError();
 509     if (last_error == ERROR_PORT_UNREACHABLE)
 510       last_error = WSAECONNREFUSED;
 511     if (result != 0 && last_error != WSA_IO_PENDING)
 512       iocp_service_.on_completion(op, last_error, bytes_transferred);
 513     else
 514       iocp_service_.on_pending(op);
 515   }
 516 }
 517
 518 void win_iocp_socket_service_base::start_accept_op(
 519     win_iocp_socket_service_base::base_implementation_type& impl,
 520     bool peer_is_open, socket_holder& new_socket, int family, int type,
 521     int protocol, void* output_buffer, DWORD address_length, operation* op)
 522 {
 523   update_cancellation_thread_id(impl);
 524   iocp_service_.work_started();
 525
 526   if (!is_open(impl))
 527     iocp_service_.on_completion(op, boost::asio::error::bad_descriptor);
 528   else if (peer_is_open)
 529     iocp_service_.on_completion(op, boost::asio::error::already_open);
 530   else
 531   {
 532     boost::system::error_code ec;
 533     new_socket.reset(socket_ops::socket(family, type, protocol, ec));
 534     if (new_socket.get() == invalid_socket)
 535       iocp_service_.on_completion(op, ec);
 536     else
 537     {
 538       DWORD bytes_read = 0;
 539       BOOL result = ::AcceptEx(impl.socket_, new_socket.get(), output_buffer,
 540           0, address_length, address_length, &bytes_read, op);
 541       DWORD last_error = ::WSAGetLastError();
 542       if (!result && last_error != WSA_IO_PENDING)
 543         iocp_service_.on_completion(op, last_error);
 544       else
 545         iocp_service_.on_pending(op);
 546     }
 547   }
 548 }
 549
 550 void win_iocp_socket_service_base::restart_accept_op(
 551     socket_type s, socket_holder& new_socket, int family, int type,
 552     int protocol, void* output_buffer, DWORD address_length,
 553     long* cancel_requested, operation* op)
 554 {
 555   new_socket.reset();
 556   iocp_service_.work_started();
 557
 558   // Check if we were cancelled after the first AcceptEx completed.
 559   if (cancel_requested)
 560     if (::InterlockedExchangeAdd(cancel_requested, 0) == 1)
 561       iocp_service_.on_completion(op, boost::asio::error::operation_aborted);
 562
 563   boost::system::error_code ec;
 564   new_socket.reset(socket_ops::socket(family, type, protocol, ec));
 565   if (new_socket.get() == invalid_socket)
 566     iocp_service_.on_completion(op, ec);
 567   else
 568   {
 569     DWORD bytes_read = 0;
 570     BOOL result = ::AcceptEx(s, new_socket.get(), output_buffer,
 571         0, address_length, address_length, &bytes_read, op);
 572     DWORD last_error = ::WSAGetLastError();
 573     if (!result && last_error != WSA_IO_PENDING)
 574       iocp_service_.on_completion(op, last_error);
 575     else
 576     {
 577 #if defined(_WIN32_WINNT) && (_WIN32_WINNT >= 0x0600)
 578       if (cancel_requested)
 579       {
 580         if (::InterlockedExchangeAdd(cancel_requested, 0) == 1)
 581         {
 582           HANDLE sock_as_handle = reinterpret_cast<HANDLE>(s);
 583           ::CancelIoEx(sock_as_handle, op);
 584         }
 585       }
 586 #endif // defined(_WIN32_WINNT) && (_WIN32_WINNT >= 0x0600)
 587       iocp_service_.on_pending(op);
 588     }
 589   }
 590 }
 591
 592 void win_iocp_socket_service_base::start_reactor_op(
 593     win_iocp_socket_service_base::base_implementation_type& impl,
 594     int op_type, reactor_op* op)
 595 {
 596   select_reactor& r = get_reactor();
 597   update_cancellation_thread_id(impl);
 598
 599   if (is_open(impl))
 600   {
 601     r.start_op(op_type, impl.socket_, impl.reactor_data_, op, false, false);
 602     return;
 603   }
 604   else
 605     op->ec_ = boost::asio::error::bad_descriptor;
 606
 607   iocp_service_.post_immediate_completion(op, false);
 608 }
 609
 610 int win_iocp_socket_service_base::start_connect_op(
 611     win_iocp_socket_service_base::base_implementation_type& impl,
 612     int family, int type, const socket_addr_type* addr, std::size_t addrlen,
 613     win_iocp_socket_connect_op_base* op, operation* iocp_op)
 614 {
 615   // If ConnectEx is available, use that.
 616   if (family == BOOST_ASIO_OS_DEF(AF_INET)
 617       || family == BOOST_ASIO_OS_DEF(AF_INET6))
 618   {
 619     if (connect_ex_fn connect_ex = get_connect_ex(impl, type))
 620     {
 621       union address_union
 622       {
 623         socket_addr_type base;
 624         sockaddr_in4_type v4;
 625         sockaddr_in6_type v6;
 626       } a;
 627
 628       using namespace std; // For memset.
 629       memset(&a, 0, sizeof(a));
 630       a.base.sa_family = family;
 631
 632       socket_ops::bind(impl.socket_, &a.base,
 633           family == BOOST_ASIO_OS_DEF(AF_INET)
 634           ? sizeof(a.v4) : sizeof(a.v6), op->ec_);
 635       if (op->ec_ && op->ec_ != boost::asio::error::invalid_argument)
 636       {
 637         iocp_service_.post_immediate_completion(op, false);
 638         return -1;
 639       }
 640
 641       op->connect_ex_ = true;
 642       update_cancellation_thread_id(impl);
 643       iocp_service_.work_started();
 644
 645       BOOL result = connect_ex(impl.socket_,
 646           addr, static_cast<int>(addrlen), 0, 0, 0, iocp_op);
 647       DWORD last_error = ::WSAGetLastError();
 648       if (!result && last_error != WSA_IO_PENDING)
 649         iocp_service_.on_completion(iocp_op, last_error);
 650       else
 651         iocp_service_.on_pending(iocp_op);
 652       return -1;
 653     }
 654   }
 655
 656   // Otherwise, fall back to a reactor-based implementation.
 657   select_reactor& r = get_reactor();
 658   update_cancellation_thread_id(impl);
 659
 660   if ((impl.state_ & socket_ops::non_blocking) != 0
 661       || socket_ops::set_internal_non_blocking(
 662         impl.socket_, impl.state_, true, op->ec_))
 663   {
 664     if (socket_ops::connect(impl.socket_, addr, addrlen, op->ec_) != 0)
 665     {
 666       if (op->ec_ == boost::asio::error::in_progress
 667           || op->ec_ == boost::asio::error::would_block)
 668       {
 669         op->ec_ = boost::system::error_code();
 670         r.start_op(select_reactor::connect_op, impl.socket_,
 671             impl.reactor_data_, op, false, false);
 672         return select_reactor::connect_op;
 673       }
 674     }
 675   }
 676
 677   r.post_immediate_completion(op, false);
 678   return -1;
 679 }
 680
 681 void win_iocp_socket_service_base::close_for_destruction(
 682     win_iocp_socket_service_base::base_implementation_type& impl)
 683 {
 684   if (is_open(impl))
 685   {
 686     BOOST_ASIO_HANDLER_OPERATION((iocp_service_.context(),
 687           "socket", &impl, impl.socket_, "close"));
 688
 689     // Check if the reactor was created, in which case we need to close the
 690     // socket on the reactor as well to cancel any operations that might be
 691     // running there.
 692     select_reactor* r = static_cast<select_reactor*>(
 693           interlocked_compare_exchange_pointer(
 694             reinterpret_cast<void**>(&reactor_), 0, 0));
 695     if (r)
 696       r->deregister_descriptor(impl.socket_, impl.reactor_data_, true);
 697
 698     boost::system::error_code ignored_ec;
 699     socket_ops::close(impl.socket_, impl.state_, true, ignored_ec);
 700
 701     if (r)
 702       r->cleanup_descriptor_data(impl.reactor_data_);
 703   }
 704
 705   impl.socket_ = invalid_socket;
 706   impl.state_ = 0;
 707   impl.cancel_token_.reset();
 708 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
 709   impl.safe_cancellation_thread_id_ = 0;
 710 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
 711 }
 712
 713 void win_iocp_socket_service_base::update_cancellation_thread_id(
 714     win_iocp_socket_service_base::base_implementation_type& impl)
 715 {
 716 #if defined(BOOST_ASIO_ENABLE_CANCELIO)
 717   if (impl.safe_cancellation_thread_id_ == 0)
 718     impl.safe_cancellation_thread_id_ = ::GetCurrentThreadId();
 719   else if (impl.safe_cancellation_thread_id_ != ::GetCurrentThreadId())
 720     impl.safe_cancellation_thread_id_ = ~DWORD(0);
 721 #else // defined(BOOST_ASIO_ENABLE_CANCELIO)
 722   (void)impl;
 723 #endif // defined(BOOST_ASIO_ENABLE_CANCELIO)
 724 }
 725
 726 select_reactor& win_iocp_socket_service_base::get_reactor()
 727 {
 728   select_reactor* r = static_cast<select_reactor*>(
 729         interlocked_compare_exchange_pointer(
 730           reinterpret_cast<void**>(&reactor_), 0, 0));
 731   if (!r)
 732   {
 733     r = &(use_service<select_reactor>(context_));
 734     interlocked_exchange_pointer(reinterpret_cast<void**>(&reactor_), r);
 735   }
 736   return *r;
 737 }
 738
 739 win_iocp_socket_service_base::connect_ex_fn
 740 win_iocp_socket_service_base::get_connect_ex(
 741     win_iocp_socket_service_base::base_implementation_type& impl, int type)
 742 {
 743 #if defined(BOOST_ASIO_DISABLE_CONNECTEX)
 744   (void)impl;
 745   (void)type;
 746   return 0;
 747 #else // defined(BOOST_ASIO_DISABLE_CONNECTEX)
 748   if (type != BOOST_ASIO_OS_DEF(SOCK_STREAM)
 749       && type != BOOST_ASIO_OS_DEF(SOCK_SEQPACKET))
 750     return 0;
 751
 752   void* ptr = interlocked_compare_exchange_pointer(&connect_ex_, 0, 0);
 753   if (!ptr)
 754   {
 755     GUID guid = { 0x25a207b9, 0xddf3, 0x4660,
 756       { 0x8e, 0xe9, 0x76, 0xe5, 0x8c, 0x74, 0x06, 0x3e } };
 757
 758     DWORD bytes = 0;
 759     if (::WSAIoctl(impl.socket_, SIO_GET_EXTENSION_FUNCTION_POINTER,
 760           &guid, sizeof(guid), &ptr, sizeof(ptr), &bytes, 0, 0) != 0)
 761     {
 762       // Set connect_ex_ to a special value to indicate that ConnectEx is
 763       // unavailable. That way we won't bother trying to look it up again.
 764       ptr = this;
 765     }
 766
 767     interlocked_exchange_pointer(&connect_ex_, ptr);
 768   }
 769
 770   return reinterpret_cast<connect_ex_fn>(ptr == this ? 0 : ptr);
 771 #endif // defined(BOOST_ASIO_DISABLE_CONNECTEX)
 772 }
 773
 774 win_iocp_socket_service_base::nt_set_info_fn
 775 win_iocp_socket_service_base::get_nt_set_info()
 776 {
 777   void* ptr = interlocked_compare_exchange_pointer(&nt_set_info_, 0, 0);
 778   if (!ptr)
 779   {
 780     if (HMODULE h = ::GetModuleHandleA("NTDLL.DLL"))
 781       ptr = reinterpret_cast<void*>(GetProcAddress(h, "NtSetInformationFile"));
 782
 783     // On failure, set nt_set_info_ to a special value to indicate that the
 784     // NtSetInformationFile function is unavailable. That way we won't bother
 785     // trying to look it up again.
 786     interlocked_exchange_pointer(&nt_set_info_, ptr ? ptr : this);
 787   }
 788
 789   return reinterpret_cast<nt_set_info_fn>(ptr == this ? 0 : ptr);
 790 }
 791
 792 void* win_iocp_socket_service_base::interlocked_compare_exchange_pointer(
 793     void** dest, void* exch, void* cmp)
 794 {
 795 #if defined(_M_IX86)
 796   return reinterpret_cast<void*>(InterlockedCompareExchange(
 797         reinterpret_cast<PLONG>(dest), reinterpret_cast<LONG>(exch),
 798         reinterpret_cast<LONG>(cmp)));
 799 #else
 800   return InterlockedCompareExchangePointer(dest, exch, cmp);
 801 #endif
 802 }
 803
 804 void* win_iocp_socket_service_base::interlocked_exchange_pointer(
 805     void** dest, void* val)
 806 {
 807 #if defined(_M_IX86)
 808   return reinterpret_cast<void*>(InterlockedExchange(
 809         reinterpret_cast<PLONG>(dest), reinterpret_cast<LONG>(val)));
 810 #else
 811   return InterlockedExchangePointer(dest, val);
 812 #endif
 813 }
 814
 815 } // namespace detail
 816 } // namespace asio
 817 } // namespace boost
 818
 819 #include <boost/asio/detail/pop_options.hpp>
 820
 821 #endif // defined(BOOST_ASIO_HAS_IOCP)
 822
 823 #endif // BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_SOCKET_SERVICE_BASE_IPP