diff options
Diffstat (limited to '3rdparty/asio/include/asio/detail/reactive_socket_service.hpp')
-rw-r--r-- | 3rdparty/asio/include/asio/detail/reactive_socket_service.hpp | 290 |
1 files changed, 199 insertions, 91 deletions
diff --git a/3rdparty/asio/include/asio/detail/reactive_socket_service.hpp b/3rdparty/asio/include/asio/detail/reactive_socket_service.hpp index 4ba255667c1..c828e7f1f2a 100644 --- a/3rdparty/asio/include/asio/detail/reactive_socket_service.hpp +++ b/3rdparty/asio/include/asio/detail/reactive_socket_service.hpp @@ -2,7 +2,7 @@ // detail/reactive_socket_service.hpp // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ // -// Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff dot com) +// Copyright (c) 2003-2024 Christopher M. Kohlhoff (chris at kohlhoff dot com) // // Distributed under the Boost Software License, Version 1.0. (See accompanying // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) @@ -17,11 +17,12 @@ #include "asio/detail/config.hpp" -#if !defined(ASIO_HAS_IOCP) +#if !defined(ASIO_HAS_IOCP) \ + && !defined(ASIO_HAS_IO_URING_AS_DEFAULT) #include "asio/buffer.hpp" #include "asio/error.hpp" -#include "asio/io_context.hpp" +#include "asio/execution_context.hpp" #include "asio/socket_base.hpp" #include "asio/detail/buffer_sequence_adapter.hpp" #include "asio/detail/memory.hpp" @@ -45,7 +46,7 @@ namespace detail { template <typename Protocol> class reactive_socket_service : - public service_base<reactive_socket_service<Protocol> >, + public execution_context_service_base<reactive_socket_service<Protocol>>, public reactive_socket_service_base { public: @@ -73,9 +74,10 @@ public: }; // Constructor. - reactive_socket_service(asio::io_context& io_context) - : service_base<reactive_socket_service<Protocol> >(io_context), - reactive_socket_service_base(io_context) + reactive_socket_service(execution_context& context) + : execution_context_service_base< + reactive_socket_service<Protocol>>(context), + reactive_socket_service_base(context) { } @@ -87,7 +89,7 @@ public: // Move-construct a new socket implementation. void move_construct(implementation_type& impl, - implementation_type& other_impl) + implementation_type& other_impl) noexcept { this->base_move_construct(impl, other_impl); @@ -109,6 +111,7 @@ public: // Move-construct a new socket implementation from another protocol type. template <typename Protocol1> void converting_move_construct(implementation_type& impl, + reactive_socket_service<Protocol1>&, typename reactive_socket_service< Protocol1>::implementation_type& other_impl) { @@ -125,6 +128,8 @@ public: if (!do_open(impl, protocol.family(), protocol.type(), protocol.protocol(), ec)) impl.protocol_ = protocol; + + ASIO_ERROR_LOCATION(ec); return ec; } @@ -135,6 +140,8 @@ public: { if (!do_assign(impl, protocol.type(), native_socket, ec)) impl.protocol_ = protocol; + + ASIO_ERROR_LOCATION(ec); return ec; } @@ -149,6 +156,8 @@ public: const endpoint_type& endpoint, asio::error_code& ec) { socket_ops::bind(impl.socket_, endpoint.data(), endpoint.size(), ec); + + ASIO_ERROR_LOCATION(ec); return ec; } @@ -160,6 +169,8 @@ public: socket_ops::setsockopt(impl.socket_, impl.state_, option.level(impl.protocol_), option.name(impl.protocol_), option.data(impl.protocol_), option.size(impl.protocol_), ec); + + ASIO_ERROR_LOCATION(ec); return ec; } @@ -174,6 +185,8 @@ public: option.data(impl.protocol_), &size, ec); if (!ec) option.resize(impl.protocol_, size); + + ASIO_ERROR_LOCATION(ec); return ec; } @@ -184,7 +197,10 @@ public: endpoint_type endpoint; std::size_t addr_len = endpoint.capacity(); if (socket_ops::getsockname(impl.socket_, endpoint.data(), &addr_len, ec)) + { + ASIO_ERROR_LOCATION(ec); return endpoint_type(); + } endpoint.resize(addr_len); return endpoint; } @@ -197,7 +213,10 @@ public: std::size_t addr_len = endpoint.capacity(); if (socket_ops::getpeername(impl.socket_, endpoint.data(), &addr_len, false, ec)) + { + ASIO_ERROR_LOCATION(ec); return endpoint_type(); + } endpoint.resize(addr_len); return endpoint; } @@ -207,6 +226,8 @@ public: socket_base::shutdown_type what, asio::error_code& ec) { socket_ops::shutdown(impl.socket_, what, ec); + + ASIO_ERROR_LOCATION(ec); return ec; } @@ -217,12 +238,27 @@ public: const endpoint_type& destination, socket_base::message_flags flags, asio::error_code& ec) { - buffer_sequence_adapter<asio::const_buffer, - ConstBufferSequence> bufs(buffers); + typedef buffer_sequence_adapter<asio::const_buffer, + ConstBufferSequence> bufs_type; - return socket_ops::sync_sendto(impl.socket_, impl.state_, - bufs.buffers(), bufs.count(), flags, - destination.data(), destination.size(), ec); + size_t n; + if (bufs_type::is_single_buffer) + { + n = socket_ops::sync_sendto1(impl.socket_, impl.state_, + bufs_type::first(buffers).data(), + bufs_type::first(buffers).size(), flags, + destination.data(), destination.size(), ec); + } + else + { + bufs_type bufs(buffers); + n = socket_ops::sync_sendto(impl.socket_, impl.state_, + bufs.buffers(), bufs.count(), flags, + destination.data(), destination.size(), ec); + } + + ASIO_ERROR_LOCATION(ec); + return n; } // Wait until data can be sent without blocking. @@ -233,52 +269,79 @@ public: // Wait for socket to become ready. socket_ops::poll_write(impl.socket_, impl.state_, -1, ec); + ASIO_ERROR_LOCATION(ec); return 0; } // Start an asynchronous send. The data being sent must be valid for the // lifetime of the asynchronous operation. - template <typename ConstBufferSequence, typename Handler> + template <typename ConstBufferSequence, typename Handler, typename IoExecutor> void async_send_to(implementation_type& impl, const ConstBufferSequence& buffers, const endpoint_type& destination, socket_base::message_flags flags, - Handler& handler) + Handler& handler, const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. typedef reactive_socket_sendto_op<ConstBufferSequence, - endpoint_type, Handler> op; + endpoint_type, Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; - p.p = new (p.v) op(impl.socket_, buffers, destination, flags, handler); + p.p = new (p.v) op(success_ec_, impl.socket_, + buffers, destination, flags, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::write_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_send_to")); - start_op(impl, reactor::write_op, p.p, is_continuation, true, false); + start_op(impl, reactor::write_op, p.p, + is_continuation, true, false, &io_ex, 0); p.v = p.p = 0; } // Start an asynchronous wait until data can be sent without blocking. - template <typename Handler> + template <typename Handler, typename IoExecutor> void async_send_to(implementation_type& impl, const null_buffers&, - const endpoint_type&, socket_base::message_flags, Handler& handler) + const endpoint_type&, socket_base::message_flags, + Handler& handler, const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. - typedef reactive_null_buffers_op<Handler> op; + typedef reactive_null_buffers_op<Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; - p.p = new (p.v) op(handler); + p.p = new (p.v) op(success_ec_, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::write_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_send_to(null_buffers)")); - start_op(impl, reactor::write_op, p.p, is_continuation, false, false); + start_op(impl, reactor::write_op, p.p, + is_continuation, false, false, &io_ex, 0); p.v = p.p = 0; } @@ -290,18 +353,29 @@ public: endpoint_type& sender_endpoint, socket_base::message_flags flags, asio::error_code& ec) { - buffer_sequence_adapter<asio::mutable_buffer, - MutableBufferSequence> bufs(buffers); + typedef buffer_sequence_adapter<asio::mutable_buffer, + MutableBufferSequence> bufs_type; std::size_t addr_len = sender_endpoint.capacity(); - std::size_t bytes_recvd = socket_ops::sync_recvfrom( - impl.socket_, impl.state_, bufs.buffers(), bufs.count(), - flags, sender_endpoint.data(), &addr_len, ec); + std::size_t n; + if (bufs_type::is_single_buffer) + { + n = socket_ops::sync_recvfrom1(impl.socket_, impl.state_, + bufs_type::first(buffers).data(), bufs_type::first(buffers).size(), + flags, sender_endpoint.data(), &addr_len, ec); + } + else + { + bufs_type bufs(buffers); + n = socket_ops::sync_recvfrom(impl.socket_, impl.state_, bufs.buffers(), + bufs.count(), flags, sender_endpoint.data(), &addr_len, ec); + } if (!ec) sender_endpoint.resize(addr_len); - return bytes_recvd; + ASIO_ERROR_LOCATION(ec); + return n; } // Wait until data can be received without blocking. @@ -315,28 +389,42 @@ public: // Reset endpoint since it can be given no sensible value at this time. sender_endpoint = endpoint_type(); + ASIO_ERROR_LOCATION(ec); return 0; } // Start an asynchronous receive. The buffer for the data being received and // the sender_endpoint object must both be valid for the lifetime of the // asynchronous operation. - template <typename MutableBufferSequence, typename Handler> + template <typename MutableBufferSequence, + typename Handler, typename IoExecutor> void async_receive_from(implementation_type& impl, const MutableBufferSequence& buffers, endpoint_type& sender_endpoint, - socket_base::message_flags flags, Handler& handler) + socket_base::message_flags flags, Handler& handler, + const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. typedef reactive_socket_recvfrom_op<MutableBufferSequence, - endpoint_type, Handler> op; + endpoint_type, Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; int protocol = impl.protocol_.type(); - p.p = new (p.v) op(impl.socket_, protocol, - buffers, sender_endpoint, flags, handler); + p.p = new (p.v) op(success_ec_, impl.socket_, protocol, + buffers, sender_endpoint, flags, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_receive_from")); @@ -344,24 +432,35 @@ public: start_op(impl, (flags & socket_base::message_out_of_band) ? reactor::except_op : reactor::read_op, - p.p, is_continuation, true, false); + p.p, is_continuation, true, false, &io_ex, 0); p.v = p.p = 0; } // Wait until data can be received without blocking. - template <typename Handler> - void async_receive_from(implementation_type& impl, - const null_buffers&, endpoint_type& sender_endpoint, - socket_base::message_flags flags, Handler& handler) + template <typename Handler, typename IoExecutor> + void async_receive_from(implementation_type& impl, const null_buffers&, + endpoint_type& sender_endpoint, socket_base::message_flags flags, + Handler& handler, const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. - typedef reactive_null_buffers_op<Handler> op; + typedef reactive_null_buffers_op<Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; - p.p = new (p.v) op(handler); + p.p = new (p.v) op(success_ec_, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_receive_from(null_buffers)")); @@ -372,7 +471,7 @@ public: start_op(impl, (flags & socket_base::message_out_of_band) ? reactor::except_op : reactor::read_op, - p.p, is_continuation, false, false); + p.p, is_continuation, false, false, &io_ex, 0); p.v = p.p = 0; } @@ -385,6 +484,7 @@ public: if (peer.is_open()) { ec = asio::error::already_open; + ASIO_ERROR_LOCATION(ec); return ec; } @@ -403,85 +503,79 @@ public: new_socket.release(); } + ASIO_ERROR_LOCATION(ec); return ec; } -#if defined(ASIO_HAS_MOVE) - // Accept a new connection. - typename Protocol::socket accept(implementation_type& impl, - io_context* peer_io_context, endpoint_type* peer_endpoint, - asio::error_code& ec) - { - typename Protocol::socket peer( - peer_io_context ? *peer_io_context : io_context_); - - std::size_t addr_len = peer_endpoint ? peer_endpoint->capacity() : 0; - socket_holder new_socket(socket_ops::sync_accept(impl.socket_, - impl.state_, peer_endpoint ? peer_endpoint->data() : 0, - peer_endpoint ? &addr_len : 0, ec)); - - // On success, assign new connection to peer socket object. - if (new_socket.get() != invalid_socket) - { - if (peer_endpoint) - peer_endpoint->resize(addr_len); - peer.assign(impl.protocol_, new_socket.get(), ec); - if (!ec) - new_socket.release(); - } - - return peer; - } -#endif // defined(ASIO_HAS_MOVE) - // Start an asynchronous accept. The peer and peer_endpoint objects must be // valid until the accept's handler is invoked. - template <typename Socket, typename Handler> + template <typename Socket, typename Handler, typename IoExecutor> void async_accept(implementation_type& impl, Socket& peer, - endpoint_type* peer_endpoint, Handler& handler) + endpoint_type* peer_endpoint, Handler& handler, const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. - typedef reactive_socket_accept_op<Socket, Protocol, Handler> op; + typedef reactive_socket_accept_op<Socket, Protocol, Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; - p.p = new (p.v) op(impl.socket_, impl.state_, peer, - impl.protocol_, peer_endpoint, handler); + p.p = new (p.v) op(success_ec_, impl.socket_, impl.state_, + peer, impl.protocol_, peer_endpoint, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected() && !peer.is_open()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_accept")); - start_accept_op(impl, p.p, is_continuation, peer.is_open()); + start_accept_op(impl, p.p, is_continuation, peer.is_open(), &io_ex, 0); p.v = p.p = 0; } -#if defined(ASIO_HAS_MOVE) // Start an asynchronous accept. The peer_endpoint object must be valid until // the accept's handler is invoked. - template <typename Handler> - void async_accept(implementation_type& impl, - asio::io_context* peer_io_context, - endpoint_type* peer_endpoint, Handler& handler) + template <typename PeerIoExecutor, typename Handler, typename IoExecutor> + void async_move_accept(implementation_type& impl, + const PeerIoExecutor& peer_io_ex, endpoint_type* peer_endpoint, + Handler& handler, const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. - typedef reactive_socket_move_accept_op<Protocol, Handler> op; + typedef reactive_socket_move_accept_op<Protocol, + PeerIoExecutor, Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; - p.p = new (p.v) op(peer_io_context ? *peer_io_context : io_context_, - impl.socket_, impl.state_, impl.protocol_, peer_endpoint, handler); + p.p = new (p.v) op(success_ec_, peer_io_ex, impl.socket_, + impl.state_, impl.protocol_, peer_endpoint, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::read_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_accept")); - start_accept_op(impl, p.p, is_continuation, false); + start_accept_op(impl, p.p, is_continuation, false, &io_ex, 0); p.v = p.p = 0; } -#endif // defined(ASIO_HAS_MOVE) // Connect the socket to the specified endpoint. asio::error_code connect(implementation_type& impl, @@ -489,28 +583,41 @@ public: { socket_ops::sync_connect(impl.socket_, peer_endpoint.data(), peer_endpoint.size(), ec); + ASIO_ERROR_LOCATION(ec); return ec; } // Start an asynchronous connect. - template <typename Handler> + template <typename Handler, typename IoExecutor> void async_connect(implementation_type& impl, - const endpoint_type& peer_endpoint, Handler& handler) + const endpoint_type& peer_endpoint, + Handler& handler, const IoExecutor& io_ex) { bool is_continuation = asio_handler_cont_helpers::is_continuation(handler); + associated_cancellation_slot_t<Handler> slot + = asio::get_associated_cancellation_slot(handler); + // Allocate and construct an operation to wrap the handler. - typedef reactive_socket_connect_op<Handler> op; + typedef reactive_socket_connect_op<Handler, IoExecutor> op; typename op::ptr p = { asio::detail::addressof(handler), op::ptr::allocate(handler), 0 }; - p.p = new (p.v) op(impl.socket_, handler); + p.p = new (p.v) op(success_ec_, impl.socket_, handler, io_ex); + + // Optionally register for per-operation cancellation. + if (slot.is_connected()) + { + p.p->cancellation_key_ = + &slot.template emplace<reactor_op_cancellation>( + &reactor_, &impl.reactor_data_, impl.socket_, reactor::connect_op); + } ASIO_HANDLER_CREATION((reactor_.context(), *p.p, "socket", &impl, impl.socket_, "async_connect")); start_connect_op(impl, p.p, is_continuation, - peer_endpoint.data(), peer_endpoint.size()); + peer_endpoint.data(), peer_endpoint.size(), &io_ex, 0); p.v = p.p = 0; } }; @@ -521,5 +628,6 @@ public: #include "asio/detail/pop_options.hpp" #endif // !defined(ASIO_HAS_IOCP) + // && !defined(ASIO_HAS_IO_URING_AS_DEFAULT) #endif // ASIO_DETAIL_REACTIVE_SOCKET_SERVICE_HPP |