summaryrefslogtreecommitdiffstatshomepage
path: root/3rdparty/asio/include/asio/detail/reactive_socket_service.hpp
diff options
context:
space:
mode:
Diffstat (limited to '3rdparty/asio/include/asio/detail/reactive_socket_service.hpp')
-rw-r--r--3rdparty/asio/include/asio/detail/reactive_socket_service.hpp240
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"));