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.hpp290
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