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 | 240 |
1 files changed, 160 insertions, 80 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..2124eaf9a9d 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-2021 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) @@ -21,7 +21,7 @@ #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 +45,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 +73,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 +88,7 @@ public: // Move-construct a new socket implementation. void move_construct(implementation_type& impl, - implementation_type& other_impl) + implementation_type& other_impl) ASIO_NOEXCEPT { this->base_move_construct(impl, other_impl); @@ -109,6 +110,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) { @@ -217,12 +219,23 @@ 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); + if (bufs_type::is_single_buffer) + { + return 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); + return socket_ops::sync_sendto(impl.socket_, impl.state_, + bufs.buffers(), bufs.count(), flags, + destination.data(), destination.size(), ec); + } } // Wait until data can be sent without blocking. @@ -238,21 +251,33 @@ public: // 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); + typename associated_cancellation_slot<Handler>::type 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")); @@ -262,18 +287,30 @@ public: } // 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); + typename associated_cancellation_slot<Handler>::type 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)")); @@ -290,13 +327,25 @@ 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 bytes_recvd; + if (bufs_type::is_single_buffer) + { + bytes_recvd = 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); + bytes_recvd = 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); @@ -321,22 +370,35 @@ public: // 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); + typename associated_cancellation_slot<Handler>::type 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")); @@ -349,19 +411,30 @@ public: } // 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); + typename associated_cancellation_slot<Handler>::type 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)")); @@ -406,49 +479,32 @@ public: 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); + typename associated_cancellation_slot<Handler>::type 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")); @@ -460,20 +516,32 @@ public: #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); + typename associated_cancellation_slot<Handler>::type 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")); @@ -493,18 +561,30 @@ public: } // 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); + typename associated_cancellation_slot<Handler>::type 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")); |