asio/detail/reactive_socket_service.hpp

0001 //
0002 // detail/reactive_socket_service.hpp
0003 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
0004 //
0005 // Copyright (c) 2003-2023 Christopher M. Kohlhoff (chris at kohlhoff dot com)
0006 //
0007 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0008 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0009 //
0010
0011 #ifndef BOOST_ASIO_DETAIL_REACTIVE_SOCKET_SERVICE_HPP
0012 #define BOOST_ASIO_DETAIL_REACTIVE_SOCKET_SERVICE_HPP
0013
0014 #if defined(_MSC_VER) && (_MSC_VER >= 1200)
0015 # pragma once
0016 #endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
0017
0018 #include <boost/asio/detail/config.hpp>
0019
0020 #if !defined(BOOST_ASIO_HAS_IOCP) \
0021   && !defined(BOOST_ASIO_HAS_IO_URING_AS_DEFAULT)
0022
0023 #include <boost/asio/buffer.hpp>
0024 #include <boost/asio/error.hpp>
0025 #include <boost/asio/execution_context.hpp>
0026 #include <boost/asio/socket_base.hpp>
0027 #include <boost/asio/detail/buffer_sequence_adapter.hpp>
0028 #include <boost/asio/detail/memory.hpp>
0029 #include <boost/asio/detail/noncopyable.hpp>
0030 #include <boost/asio/detail/reactive_null_buffers_op.hpp>
0031 #include <boost/asio/detail/reactive_socket_accept_op.hpp>
0032 #include <boost/asio/detail/reactive_socket_connect_op.hpp>
0033 #include <boost/asio/detail/reactive_socket_recvfrom_op.hpp>
0034 #include <boost/asio/detail/reactive_socket_sendto_op.hpp>
0035 #include <boost/asio/detail/reactive_socket_service_base.hpp>
0036 #include <boost/asio/detail/reactor.hpp>
0037 #include <boost/asio/detail/reactor_op.hpp>
0038 #include <boost/asio/detail/socket_holder.hpp>
0039 #include <boost/asio/detail/socket_ops.hpp>
0040 #include <boost/asio/detail/socket_types.hpp>
0041
0042 #include <boost/asio/detail/push_options.hpp>
0043
0044 namespace boost {
0045 namespace asio {
0046 namespace detail {
0047
0048 template <typename Protocol>
0049 class reactive_socket_service :
0050   public execution_context_service_base<reactive_socket_service<Protocol>>,
0051   public reactive_socket_service_base
0052 {
0053 public:
0054   // The protocol type.
0055   typedef Protocol protocol_type;
0056
0057   // The endpoint type.
0058   typedef typename Protocol::endpoint endpoint_type;
0059
0060   // The native type of a socket.
0061   typedef socket_type native_handle_type;
0062
0063   // The implementation type of the socket.
0064   struct implementation_type :
0065     reactive_socket_service_base::base_implementation_type
0066   {
0067     // Default constructor.
0068     implementation_type()
0069       : protocol_(endpoint_type().protocol())
0070     {
0071     }
0072
0073     // The protocol associated with the socket.
0074     protocol_type protocol_;
0075   };
0076
0077   // Constructor.
0078   reactive_socket_service(execution_context& context)
0079     : execution_context_service_base<
0080         reactive_socket_service<Protocol>>(context),
0081       reactive_socket_service_base(context)
0082   {
0083   }
0084
0085   // Destroy all user-defined handler objects owned by the service.
0086   void shutdown()
0087   {
0088     this->base_shutdown();
0089   }
0090
0091   // Move-construct a new socket implementation.
0092   void move_construct(implementation_type& impl,
0093       implementation_type& other_impl) noexcept
0094   {
0095     this->base_move_construct(impl, other_impl);
0096
0097     impl.protocol_ = other_impl.protocol_;
0098     other_impl.protocol_ = endpoint_type().protocol();
0099   }
0100
0101   // Move-assign from another socket implementation.
0102   void move_assign(implementation_type& impl,
0103       reactive_socket_service_base& other_service,
0104       implementation_type& other_impl)
0105   {
0106     this->base_move_assign(impl, other_service, other_impl);
0107
0108     impl.protocol_ = other_impl.protocol_;
0109     other_impl.protocol_ = endpoint_type().protocol();
0110   }
0111
0112   // Move-construct a new socket implementation from another protocol type.
0113   template <typename Protocol1>
0114   void converting_move_construct(implementation_type& impl,
0115       reactive_socket_service<Protocol1>&,
0116       typename reactive_socket_service<
0117         Protocol1>::implementation_type& other_impl)
0118   {
0119     this->base_move_construct(impl, other_impl);
0120
0121     impl.protocol_ = protocol_type(other_impl.protocol_);
0122     other_impl.protocol_ = typename Protocol1::endpoint().protocol();
0123   }
0124
0125   // Open a new socket implementation.
0126   boost::system::error_code open(implementation_type& impl,
0127       const protocol_type& protocol, boost::system::error_code& ec)
0128   {
0129     if (!do_open(impl, protocol.family(),
0130           protocol.type(), protocol.protocol(), ec))
0131       impl.protocol_ = protocol;
0132
0133     BOOST_ASIO_ERROR_LOCATION(ec);
0134     return ec;
0135   }
0136
0137   // Assign a native socket to a socket implementation.
0138   boost::system::error_code assign(implementation_type& impl,
0139       const protocol_type& protocol, const native_handle_type& native_socket,
0140       boost::system::error_code& ec)
0141   {
0142     if (!do_assign(impl, protocol.type(), native_socket, ec))
0143       impl.protocol_ = protocol;
0144
0145     BOOST_ASIO_ERROR_LOCATION(ec);
0146     return ec;
0147   }
0148
0149   // Get the native socket representation.
0150   native_handle_type native_handle(implementation_type& impl)
0151   {
0152     return impl.socket_;
0153   }
0154
0155   // Bind the socket to the specified local endpoint.
0156   boost::system::error_code bind(implementation_type& impl,
0157       const endpoint_type& endpoint, boost::system::error_code& ec)
0158   {
0159     socket_ops::bind(impl.socket_, endpoint.data(), endpoint.size(), ec);
0160
0161     BOOST_ASIO_ERROR_LOCATION(ec);
0162     return ec;
0163   }
0164
0165   // Set a socket option.
0166   template <typename Option>
0167   boost::system::error_code set_option(implementation_type& impl,
0168       const Option& option, boost::system::error_code& ec)
0169   {
0170     socket_ops::setsockopt(impl.socket_, impl.state_,
0171         option.level(impl.protocol_), option.name(impl.protocol_),
0172         option.data(impl.protocol_), option.size(impl.protocol_), ec);
0173
0174     BOOST_ASIO_ERROR_LOCATION(ec);
0175     return ec;
0176   }
0177
0178   // Set a socket option.
0179   template <typename Option>
0180   boost::system::error_code get_option(const implementation_type& impl,
0181       Option& option, boost::system::error_code& ec) const
0182   {
0183     std::size_t size = option.size(impl.protocol_);
0184     socket_ops::getsockopt(impl.socket_, impl.state_,
0185         option.level(impl.protocol_), option.name(impl.protocol_),
0186         option.data(impl.protocol_), &size, ec);
0187     if (!ec)
0188       option.resize(impl.protocol_, size);
0189
0190     BOOST_ASIO_ERROR_LOCATION(ec);
0191     return ec;
0192   }
0193
0194   // Get the local endpoint.
0195   endpoint_type local_endpoint(const implementation_type& impl,
0196       boost::system::error_code& ec) const
0197   {
0198     endpoint_type endpoint;
0199     std::size_t addr_len = endpoint.capacity();
0200     if (socket_ops::getsockname(impl.socket_, endpoint.data(), &addr_len, ec))
0201     {
0202       BOOST_ASIO_ERROR_LOCATION(ec);
0203       return endpoint_type();
0204     }
0205     endpoint.resize(addr_len);
0206     return endpoint;
0207   }
0208
0209   // Get the remote endpoint.
0210   endpoint_type remote_endpoint(const implementation_type& impl,
0211       boost::system::error_code& ec) const
0212   {
0213     endpoint_type endpoint;
0214     std::size_t addr_len = endpoint.capacity();
0215     if (socket_ops::getpeername(impl.socket_,
0216           endpoint.data(), &addr_len, false, ec))
0217     {
0218       BOOST_ASIO_ERROR_LOCATION(ec);
0219       return endpoint_type();
0220     }
0221     endpoint.resize(addr_len);
0222     return endpoint;
0223   }
0224
0225   // Disable sends or receives on the socket.
0226   boost::system::error_code shutdown(base_implementation_type& impl,
0227       socket_base::shutdown_type what, boost::system::error_code& ec)
0228   {
0229     socket_ops::shutdown(impl.socket_, what, ec);
0230
0231     BOOST_ASIO_ERROR_LOCATION(ec);
0232     return ec;
0233   }
0234
0235   // Send a datagram to the specified endpoint. Returns the number of bytes
0236   // sent.
0237   template <typename ConstBufferSequence>
0238   size_t send_to(implementation_type& impl, const ConstBufferSequence& buffers,
0239       const endpoint_type& destination, socket_base::message_flags flags,
0240       boost::system::error_code& ec)
0241   {
0242     typedef buffer_sequence_adapter<boost::asio::const_buffer,
0243         ConstBufferSequence> bufs_type;
0244
0245     size_t n;
0246     if (bufs_type::is_single_buffer)
0247     {
0248       n = socket_ops::sync_sendto1(impl.socket_, impl.state_,
0249           bufs_type::first(buffers).data(),
0250           bufs_type::first(buffers).size(), flags,
0251           destination.data(), destination.size(), ec);
0252     }
0253     else
0254     {
0255       bufs_type bufs(buffers);
0256       n = socket_ops::sync_sendto(impl.socket_, impl.state_,
0257           bufs.buffers(), bufs.count(), flags,
0258           destination.data(), destination.size(), ec);
0259     }
0260
0261     BOOST_ASIO_ERROR_LOCATION(ec);
0262     return n;
0263   }
0264
0265   // Wait until data can be sent without blocking.
0266   size_t send_to(implementation_type& impl, const null_buffers&,
0267       const endpoint_type&, socket_base::message_flags,
0268       boost::system::error_code& ec)
0269   {
0270     // Wait for socket to become ready.
0271     socket_ops::poll_write(impl.socket_, impl.state_, -1, ec);
0272
0273     BOOST_ASIO_ERROR_LOCATION(ec);
0274     return 0;
0275   }
0276
0277   // Start an asynchronous send. The data being sent must be valid for the
0278   // lifetime of the asynchronous operation.
0279   template <typename ConstBufferSequence, typename Handler, typename IoExecutor>
0280   void async_send_to(implementation_type& impl,
0281       const ConstBufferSequence& buffers,
0282       const endpoint_type& destination, socket_base::message_flags flags,
0283       Handler& handler, const IoExecutor& io_ex)
0284   {
0285     bool is_continuation =
0286       boost_asio_handler_cont_helpers::is_continuation(handler);
0287
0288     associated_cancellation_slot_t<Handler> slot
0289       = boost::asio::get_associated_cancellation_slot(handler);
0290
0291     // Allocate and construct an operation to wrap the handler.
0292     typedef reactive_socket_sendto_op<ConstBufferSequence,
0293         endpoint_type, Handler, IoExecutor> op;
0294     typename op::ptr p = { boost::asio::detail::addressof(handler),
0295       op::ptr::allocate(handler), 0 };
0296     p.p = new (p.v) op(success_ec_, impl.socket_,
0297         buffers, destination, flags, handler, io_ex);
0298
0299     // Optionally register for per-operation cancellation.
0300     if (slot.is_connected())
0301     {
0302       p.p->cancellation_key_ =
0303         &slot.template emplace<reactor_op_cancellation>(
0304             &reactor_, &impl.reactor_data_, impl.socket_, reactor::write_op);
0305     }
0306
0307     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0308           &impl, impl.socket_, "async_send_to"));
0309
0310     start_op(impl, reactor::write_op, p.p,
0311         is_continuation, true, false, &io_ex, 0);
0312     p.v = p.p = 0;
0313   }
0314
0315   // Start an asynchronous wait until data can be sent without blocking.
0316   template <typename Handler, typename IoExecutor>
0317   void async_send_to(implementation_type& impl, const null_buffers&,
0318       const endpoint_type&, socket_base::message_flags,
0319       Handler& handler, const IoExecutor& io_ex)
0320   {
0321     bool is_continuation =
0322       boost_asio_handler_cont_helpers::is_continuation(handler);
0323
0324     associated_cancellation_slot_t<Handler> slot
0325       = boost::asio::get_associated_cancellation_slot(handler);
0326
0327     // Allocate and construct an operation to wrap the handler.
0328     typedef reactive_null_buffers_op<Handler, IoExecutor> op;
0329     typename op::ptr p = { boost::asio::detail::addressof(handler),
0330       op::ptr::allocate(handler), 0 };
0331     p.p = new (p.v) op(success_ec_, handler, io_ex);
0332
0333     // Optionally register for per-operation cancellation.
0334     if (slot.is_connected())
0335     {
0336       p.p->cancellation_key_ =
0337         &slot.template emplace<reactor_op_cancellation>(
0338             &reactor_, &impl.reactor_data_, impl.socket_, reactor::write_op);
0339     }
0340
0341     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0342           &impl, impl.socket_, "async_send_to(null_buffers)"));
0343
0344     start_op(impl, reactor::write_op, p.p,
0345         is_continuation, false, false, &io_ex, 0);
0346     p.v = p.p = 0;
0347   }
0348
0349   // Receive a datagram with the endpoint of the sender. Returns the number of
0350   // bytes received.
0351   template <typename MutableBufferSequence>
0352   size_t receive_from(implementation_type& impl,
0353       const MutableBufferSequence& buffers,
0354       endpoint_type& sender_endpoint, socket_base::message_flags flags,
0355       boost::system::error_code& ec)
0356   {
0357     typedef buffer_sequence_adapter<boost::asio::mutable_buffer,
0358         MutableBufferSequence> bufs_type;
0359
0360     std::size_t addr_len = sender_endpoint.capacity();
0361     std::size_t n;
0362     if (bufs_type::is_single_buffer)
0363     {
0364       n = socket_ops::sync_recvfrom1(impl.socket_, impl.state_,
0365           bufs_type::first(buffers).data(), bufs_type::first(buffers).size(),
0366           flags, sender_endpoint.data(), &addr_len, ec);
0367     }
0368     else
0369     {
0370       bufs_type bufs(buffers);
0371       n = socket_ops::sync_recvfrom(impl.socket_, impl.state_, bufs.buffers(),
0372           bufs.count(), flags, sender_endpoint.data(), &addr_len, ec);
0373     }
0374
0375     if (!ec)
0376       sender_endpoint.resize(addr_len);
0377
0378     BOOST_ASIO_ERROR_LOCATION(ec);
0379     return n;
0380   }
0381
0382   // Wait until data can be received without blocking.
0383   size_t receive_from(implementation_type& impl, const null_buffers&,
0384       endpoint_type& sender_endpoint, socket_base::message_flags,
0385       boost::system::error_code& ec)
0386   {
0387     // Wait for socket to become ready.
0388     socket_ops::poll_read(impl.socket_, impl.state_, -1, ec);
0389
0390     // Reset endpoint since it can be given no sensible value at this time.
0391     sender_endpoint = endpoint_type();
0392
0393     BOOST_ASIO_ERROR_LOCATION(ec);
0394     return 0;
0395   }
0396
0397   // Start an asynchronous receive. The buffer for the data being received and
0398   // the sender_endpoint object must both be valid for the lifetime of the
0399   // asynchronous operation.
0400   template <typename MutableBufferSequence,
0401       typename Handler, typename IoExecutor>
0402   void async_receive_from(implementation_type& impl,
0403       const MutableBufferSequence& buffers, endpoint_type& sender_endpoint,
0404       socket_base::message_flags flags, Handler& handler,
0405       const IoExecutor& io_ex)
0406   {
0407     bool is_continuation =
0408       boost_asio_handler_cont_helpers::is_continuation(handler);
0409
0410     associated_cancellation_slot_t<Handler> slot
0411       = boost::asio::get_associated_cancellation_slot(handler);
0412
0413     // Allocate and construct an operation to wrap the handler.
0414     typedef reactive_socket_recvfrom_op<MutableBufferSequence,
0415         endpoint_type, Handler, IoExecutor> op;
0416     typename op::ptr p = { boost::asio::detail::addressof(handler),
0417       op::ptr::allocate(handler), 0 };
0418     int protocol = impl.protocol_.type();
0419     p.p = new (p.v) op(success_ec_, impl.socket_, protocol,
0420         buffers, sender_endpoint, flags, handler, io_ex);
0421
0422     // Optionally register for per-operation cancellation.
0423     if (slot.is_connected())
0424     {
0425       p.p->cancellation_key_ =
0426         &slot.template emplace<reactor_op_cancellation>(
0427             &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op);
0428     }
0429
0430     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0431           &impl, impl.socket_, "async_receive_from"));
0432
0433     start_op(impl,
0434         (flags & socket_base::message_out_of_band)
0435           ? reactor::except_op : reactor::read_op,
0436         p.p, is_continuation, true, false, &io_ex, 0);
0437     p.v = p.p = 0;
0438   }
0439
0440   // Wait until data can be received without blocking.
0441   template <typename Handler, typename IoExecutor>
0442   void async_receive_from(implementation_type& impl, const null_buffers&,
0443       endpoint_type& sender_endpoint, socket_base::message_flags flags,
0444       Handler& handler, const IoExecutor& io_ex)
0445   {
0446     bool is_continuation =
0447       boost_asio_handler_cont_helpers::is_continuation(handler);
0448
0449     associated_cancellation_slot_t<Handler> slot
0450       = boost::asio::get_associated_cancellation_slot(handler);
0451
0452     // Allocate and construct an operation to wrap the handler.
0453     typedef reactive_null_buffers_op<Handler, IoExecutor> op;
0454     typename op::ptr p = { boost::asio::detail::addressof(handler),
0455       op::ptr::allocate(handler), 0 };
0456     p.p = new (p.v) op(success_ec_, handler, io_ex);
0457
0458     // Optionally register for per-operation cancellation.
0459     if (slot.is_connected())
0460     {
0461       p.p->cancellation_key_ =
0462         &slot.template emplace<reactor_op_cancellation>(
0463             &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op);
0464     }
0465
0466     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0467           &impl, impl.socket_, "async_receive_from(null_buffers)"));
0468
0469     // Reset endpoint since it can be given no sensible value at this time.
0470     sender_endpoint = endpoint_type();
0471
0472     start_op(impl,
0473         (flags & socket_base::message_out_of_band)
0474           ? reactor::except_op : reactor::read_op,
0475         p.p, is_continuation, false, false, &io_ex, 0);
0476     p.v = p.p = 0;
0477   }
0478
0479   // Accept a new connection.
0480   template <typename Socket>
0481   boost::system::error_code accept(implementation_type& impl,
0482       Socket& peer, endpoint_type* peer_endpoint, boost::system::error_code& ec)
0483   {
0484     // We cannot accept a socket that is already open.
0485     if (peer.is_open())
0486     {
0487       ec = boost::asio::error::already_open;
0488       BOOST_ASIO_ERROR_LOCATION(ec);
0489       return ec;
0490     }
0491
0492     std::size_t addr_len = peer_endpoint ? peer_endpoint->capacity() : 0;
0493     socket_holder new_socket(socket_ops::sync_accept(impl.socket_,
0494           impl.state_, peer_endpoint ? peer_endpoint->data() : 0,
0495           peer_endpoint ? &addr_len : 0, ec));
0496
0497     // On success, assign new connection to peer socket object.
0498     if (new_socket.get() != invalid_socket)
0499     {
0500       if (peer_endpoint)
0501         peer_endpoint->resize(addr_len);
0502       peer.assign(impl.protocol_, new_socket.get(), ec);
0503       if (!ec)
0504         new_socket.release();
0505     }
0506
0507     BOOST_ASIO_ERROR_LOCATION(ec);
0508     return ec;
0509   }
0510
0511   // Start an asynchronous accept. The peer and peer_endpoint objects must be
0512   // valid until the accept's handler is invoked.
0513   template <typename Socket, typename Handler, typename IoExecutor>
0514   void async_accept(implementation_type& impl, Socket& peer,
0515       endpoint_type* peer_endpoint, Handler& handler, const IoExecutor& io_ex)
0516   {
0517     bool is_continuation =
0518       boost_asio_handler_cont_helpers::is_continuation(handler);
0519
0520     associated_cancellation_slot_t<Handler> slot
0521       = boost::asio::get_associated_cancellation_slot(handler);
0522
0523     // Allocate and construct an operation to wrap the handler.
0524     typedef reactive_socket_accept_op<Socket, Protocol, Handler, IoExecutor> op;
0525     typename op::ptr p = { boost::asio::detail::addressof(handler),
0526       op::ptr::allocate(handler), 0 };
0527     p.p = new (p.v) op(success_ec_, impl.socket_, impl.state_,
0528         peer, impl.protocol_, peer_endpoint, handler, io_ex);
0529
0530     // Optionally register for per-operation cancellation.
0531     if (slot.is_connected() && !peer.is_open())
0532     {
0533       p.p->cancellation_key_ =
0534         &slot.template emplace<reactor_op_cancellation>(
0535             &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op);
0536     }
0537
0538     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0539           &impl, impl.socket_, "async_accept"));
0540
0541     start_accept_op(impl, p.p, is_continuation, peer.is_open(), &io_ex, 0);
0542     p.v = p.p = 0;
0543   }
0544
0545   // Start an asynchronous accept. The peer_endpoint object must be valid until
0546   // the accept's handler is invoked.
0547   template <typename PeerIoExecutor, typename Handler, typename IoExecutor>
0548   void async_move_accept(implementation_type& impl,
0549       const PeerIoExecutor& peer_io_ex, endpoint_type* peer_endpoint,
0550       Handler& handler, const IoExecutor& io_ex)
0551   {
0552     bool is_continuation =
0553       boost_asio_handler_cont_helpers::is_continuation(handler);
0554
0555     associated_cancellation_slot_t<Handler> slot
0556       = boost::asio::get_associated_cancellation_slot(handler);
0557
0558     // Allocate and construct an operation to wrap the handler.
0559     typedef reactive_socket_move_accept_op<Protocol,
0560         PeerIoExecutor, Handler, IoExecutor> op;
0561     typename op::ptr p = { boost::asio::detail::addressof(handler),
0562       op::ptr::allocate(handler), 0 };
0563     p.p = new (p.v) op(success_ec_, peer_io_ex, impl.socket_,
0564         impl.state_, impl.protocol_, peer_endpoint, handler, io_ex);
0565
0566     // Optionally register for per-operation cancellation.
0567     if (slot.is_connected())
0568     {
0569       p.p->cancellation_key_ =
0570         &slot.template emplace<reactor_op_cancellation>(
0571             &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op);
0572     }
0573
0574     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0575           &impl, impl.socket_, "async_accept"));
0576
0577     start_accept_op(impl, p.p, is_continuation, false, &io_ex, 0);
0578     p.v = p.p = 0;
0579   }
0580
0581   // Connect the socket to the specified endpoint.
0582   boost::system::error_code connect(implementation_type& impl,
0583       const endpoint_type& peer_endpoint, boost::system::error_code& ec)
0584   {
0585     socket_ops::sync_connect(impl.socket_,
0586         peer_endpoint.data(), peer_endpoint.size(), ec);
0587     BOOST_ASIO_ERROR_LOCATION(ec);
0588     return ec;
0589   }
0590
0591   // Start an asynchronous connect.
0592   template <typename Handler, typename IoExecutor>
0593   void async_connect(implementation_type& impl,
0594       const endpoint_type& peer_endpoint,
0595       Handler& handler, const IoExecutor& io_ex)
0596   {
0597     bool is_continuation =
0598       boost_asio_handler_cont_helpers::is_continuation(handler);
0599
0600     associated_cancellation_slot_t<Handler> slot
0601       = boost::asio::get_associated_cancellation_slot(handler);
0602
0603     // Allocate and construct an operation to wrap the handler.
0604     typedef reactive_socket_connect_op<Handler, IoExecutor> op;
0605     typename op::ptr p = { boost::asio::detail::addressof(handler),
0606       op::ptr::allocate(handler), 0 };
0607     p.p = new (p.v) op(success_ec_, impl.socket_, handler, io_ex);
0608
0609     // Optionally register for per-operation cancellation.
0610     if (slot.is_connected())
0611     {
0612       p.p->cancellation_key_ =
0613         &slot.template emplace<reactor_op_cancellation>(
0614             &reactor_, &impl.reactor_data_, impl.socket_, reactor::connect_op);
0615     }
0616
0617     BOOST_ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket",
0618           &impl, impl.socket_, "async_connect"));
0619
0620     start_connect_op(impl, p.p, is_continuation,
0621         peer_endpoint.data(), peer_endpoint.size(), &io_ex, 0);
0622     p.v = p.p = 0;
0623   }
0624 };
0625
0626 } // namespace detail
0627 } // namespace asio
0628 } // namespace boost
0629
0630 #include <boost/asio/detail/pop_options.hpp>
0631
0632 #endif // !defined(BOOST_ASIO_HAS_IOCP)
0633        //   && !defined(BOOST_ASIO_HAS_IO_URING_AS_DEFAULT)
0634
0635 #endif // BOOST_ASIO_DETAIL_REACTIVE_SOCKET_SERVICE_HPP