Index/include / boost / corosio / native / detail / reactor / reactor_backend.hpp

include/boost/corosio/native/detail/reactor/reactor_backend.hpp

81.9% Lines (59/0/72) 100.0% List of functions (1/1/2)
[ ] prev/next file n p prev/next uncovered line
reactor_backend.hpp
f(x) Functions (2)
Function Calls Lines Blocks
<unknown function 37> :37 – – – boost::corosio::detail::reactor_acceptor_impl<boost::corosio::detail::epoll_tcp_acceptor, boost::corosio::detail::epoll_traits, boost::corosio::detail::epoll_tcp_acceptor_service, boost::corosio::detail::epoll_tcp_socket, boost::corosio::tcp_acceptor::implementation, boost::corosio::endpoint>::accept(std::__n4861::coroutine_handle<void>, boost::capy::executor_ref, std::stop_token, std::error_code*, boost::corosio::io_object::implementation**) :37 1x 31.9% 26.0%
Line TLA Hits Source Code
1 //
2 // Copyright (c) 2026 Michael Vandeberg
3 //
4 // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6 //
7 // Official repository: https://github.com/cppalliance/corosio
8 //
9
10 #ifndef BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BACKEND_HPP
11 #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BACKEND_HPP
12
13 /* Reactor backend: acceptor accept() implementation.
14
15 Contains the accept() method body for reactor_acceptor_impl,
16 which needs all socket/service types to be complete. Included
17 by per-backend type files (epoll_types.hpp, etc.) after all
18 named types are defined.
19 */
20
21 #include <boost/corosio/native/detail/reactor/reactor_service_finals.hpp>
22 #include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
23 #include <boost/corosio/native/detail/endpoint_convert.hpp>
24 #include <boost/corosio/detail/dispatch_coro.hpp>
25
26 #include <mutex>
27
28 namespace boost::corosio::detail {
29
30 // ============================================================
31 // Acceptor accept() implementation
32 // ============================================================
33
34 template<class Derived, class Traits, class Service,
35 class SocketFinal, class AccImplBase, class Endpoint>
36 std::coroutine_handle<>
37 4441x reactor_acceptor_impl<Derived, Traits, Service, SocketFinal, AccImplBase, Endpoint>::accept(
38 std::coroutine_handle<> h,
39 capy::executor_ref ex,
40 std::stop_token token,
41 std::error_code* ec,
42 io_object::implementation** impl_out)
43 {
44 4441x auto& op = this->acc_;
45 4441x op.reset();
46 4441x op.h = h;
47 4441x op.ex = ex;
48 4441x op.ec_out = ec;
49 4441x op.impl_out = impl_out;
50 4441x op.fd = this->fd_;
51 4441x op.start(token, static_cast<Derived*>(this));
52
53 4441x sockaddr_storage peer_storage{};
54 4441x socklen_t peer_addrlen = 0;
55
56 4441x int accepted = Traits::accept_policy::do_accept(
57 this->fd_, peer_storage, peer_addrlen);
58
59 4441x if (accepted >= 0)
60 {
61 {
62 32x std::lock_guard lock(this->desc_state_.mutex);
63 32x this->desc_state_.read_ready = false;
64 32x }
65
66 32x if (this->svc_.scheduler().try_consume_inline_budget())
67 {
68 6x auto* socket_svc = this->svc_.stream_service();
69 6x if (socket_svc)
70 {
71 auto& impl =
72 6x static_cast<SocketFinal&>(*socket_svc->construct());
73 6x impl.set_socket(accepted);
74
75 6x impl.desc_state_.fd = accepted;
76 {
77 6x std::lock_guard lock(impl.desc_state_.mutex);
78 6x impl.desc_state_.read_op = nullptr;
79 6x impl.desc_state_.write_op = nullptr;
80 6x impl.desc_state_.connect_op = nullptr;
81 6x }
82 6x auto reg_ec = socket_svc->scheduler().register_descriptor(
83 accepted, &impl.desc_state_);
84 6x if (reg_ec)
85 {
86 // destroy() closes the fd the impl already owns.
87 ✗ socket_svc->destroy(&impl);
88 ✗ *ec = reg_ec;
89 ✗ if (impl_out)
90 ✗ *impl_out = nullptr;
91 }
92 else
93 {
94 6x impl.set_endpoints(
95 this->local_endpoint_,
96 6x from_sockaddr_as(
97 peer_storage, peer_addrlen, Endpoint{}));
98
99 6x *ec = {};
100 6x if (impl_out)
101 6x *impl_out = &impl;
102 }
103 }
104 else
105 {
106 ✗ ::close(accepted);
107 ✗ *ec = make_err(ENOENT);
108 ✗ if (impl_out)
109 ✗ *impl_out = nullptr;
110 }
111 6x op.cont.h = h;
112 6x return dispatch_coro(ex, op.cont);
113 }
114
115 26x op.accepted_fd = accepted;
116 26x op.peer_storage = peer_storage;
117 26x op.peer_addrlen = peer_addrlen;
118 26x op.complete(0, 0);
119 26x op.impl_ptr = this->shared_from_this();
120 26x this->svc_.post(&op);
121 26x return std::noop_coroutine();
122 }
123
124 4409x if (errno == EAGAIN || errno == EWOULDBLOCK)
125 {
126 4405x op.impl_ptr = this->shared_from_this();
127 4405x this->svc_.work_started();
128
129 4405x std::lock_guard lock(this->desc_state_.mutex);
130 4405x bool io_done = false;
131 4405x if (this->desc_state_.read_ready)
132 {
133 ✗ this->desc_state_.read_ready = false;
134 ✗ op.perform_io();
135 ✗ io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
136 ✗ if (!io_done)
137 ✗ op.errn = 0;
138 }
139
140 4405x if (io_done || op.cancelled.load(std::memory_order_acquire))
141 {
142 2x this->svc_.post(&op);
143 2x this->svc_.work_finished();
144 }
145 else
146 {
147 4403x this->desc_state_.read_op = &op;
148 }
149 4405x return std::noop_coroutine();
150 4405x }
151
152 4x op.complete(errno, 0);
153 4x op.impl_ptr = this->shared_from_this();
154 4x this->svc_.post(&op);
155 4x return std::noop_coroutine();
156 }
157
158 } // namespace boost::corosio::detail
159
160 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BACKEND_HPP
161