Index/libs / corosio / src / corosio / src / detail / select / op.hpp
Prev Index Next

libs/corosio/src/corosio/src/detail/select/op.hpp

74.2% Lines (95/128) 84.2% Functions (16/19) 65.7% Branches (23/35)
libs/corosio/src/corosio/src/detail/select/op.hpp
Line Branch Hits Source Code
1 //
2 // Copyright (c) 2026 Steve Gerbino
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_DETAIL_SELECT_OP_HPP
11 #define BOOST_COROSIO_DETAIL_SELECT_OP_HPP
12
13 #include <boost/corosio/detail/platform.hpp>
14
15 #if BOOST_COROSIO_HAS_SELECT
16
17 #include <boost/corosio/detail/config.hpp>
18 #include <boost/corosio/io_object.hpp>
19 #include <boost/corosio/endpoint.hpp>
20 #include <boost/capy/ex/executor_ref.hpp>
21 #include <boost/capy/coro.hpp>
22 #include <boost/capy/error.hpp>
23 #include <system_error>
24
25 #include "src/detail/make_err.hpp"
26 #include "src/detail/scheduler_op.hpp"
27 #include "src/detail/endpoint_convert.hpp"
28
29 #include <unistd.h>
30 #include <errno.h>
31 #include <fcntl.h>
32
33 #include <atomic>
34 #include <cstddef>
35 #include <memory>
36 #include <optional>
37 #include <stop_token>
38
39 #include <netinet/in.h>
40 #include <sys/select.h>
41 #include <sys/socket.h>
42 #include <sys/uio.h>
43
44 /*
45 select Operation State
46 ======================
47
48 Each async I/O operation has a corresponding select_op-derived struct that
49 holds the operation's state while it's in flight. The socket impl owns
50 fixed slots for each operation type (conn_, rd_, wr_), so only one
51 operation of each type can be pending per socket at a time.
52
53 This mirrors the epoll_op design for consistency across backends.
54
55 Completion vs Cancellation Race
56 -------------------------------
57 The `registered` atomic uses a tri-state (unregistered, registering,
58 registered) to handle two races: (1) between register_fd() and the
59 reactor seeing an event, and (2) between reactor completion and cancel().
60
61 The registering state closes the window where an event could arrive
62 after register_fd() but before the boolean was set. The reactor and
63 cancel() both treat registering the same as registered when claiming.
64
65 Whoever atomically exchanges to unregistered "claims" the operation
66 and is responsible for completing it. The loser sees unregistered and
67 does nothing. The initiating thread uses compare_exchange to transition
68 from registering to registered; if this fails, the reactor or cancel
69 already claimed the op.
70
71 Impl Lifetime Management
72 ------------------------
73 When cancel() posts an op to the scheduler's ready queue, the socket impl
74 might be destroyed before the scheduler processes the op. The `impl_ptr`
75 member holds a shared_ptr to the impl, keeping it alive until the op
76 completes.
77
78 EOF Detection
79 -------------
80 For reads, 0 bytes with no error means EOF. But an empty user buffer also
81 returns 0 bytes. The `empty_buffer_read` flag distinguishes these cases.
82
83 SIGPIPE Prevention
84 ------------------
85 Writes use sendmsg() with MSG_NOSIGNAL instead of writev() to prevent
86 SIGPIPE when the peer has closed.
87 */
88
89 namespace boost::corosio::detail {
90
91 // Forward declarations for cancellation support
92 class select_socket_impl;
93 class select_acceptor_impl;
94
95 /** Registration state for async operations.
96
97 Tri-state enum to handle the race between register_fd() and
98 run_reactor() seeing an event. Setting REGISTERING before
99 calling register_fd() ensures events delivered during the
100 registration window are not dropped.
101 */
102 enum class select_registration_state : std::uint8_t
103 {
104 unregistered, ///< Not registered with reactor
105 registering, ///< register_fd() called, not yet confirmed
106 registered ///< Fully registered, ready for events
107 };
108
109 struct select_op : scheduler_op
110 {
111 struct canceller
112 {
113 select_op* op;
114 void operator()() const noexcept;
115 };
116
117 capy::coro h;
118 capy::executor_ref ex;
119 std::error_code* ec_out = nullptr;
120 std::size_t* bytes_out = nullptr;
121
122 int fd = -1;
123 int errn = 0;
124 std::size_t bytes_transferred = 0;
125
126 std::atomic<bool> cancelled{false};
127 std::atomic<select_registration_state> registered{select_registration_state::unregistered};
128 std::optional<std::stop_callback<canceller>> stop_cb;
129
130 // Prevents use-after-free when socket is closed with pending ops.
131 std::shared_ptr<void> impl_ptr;
132
133 // For stop_token cancellation - pointer to owning socket/acceptor impl.
134 select_socket_impl* socket_impl_ = nullptr;
135 select_acceptor_impl* acceptor_impl_ = nullptr;
136
137 17652 select_op() = default;
138
139 232091 void reset() noexcept
140 {
141 232091 fd = -1;
142 232091 errn = 0;
143 232091 bytes_transferred = 0;
144 232091 cancelled.store(false, std::memory_order_relaxed);
145 232091 registered.store(select_registration_state::unregistered, std::memory_order_relaxed);
146 232091 impl_ptr.reset();
147 232091 socket_impl_ = nullptr;
148 232091 acceptor_impl_ = nullptr;
149 232091 }
150
151 226229 void operator()() override
152 {
153 226229 stop_cb.reset();
154
155
1/2
✓ Branch 0 taken 226229 times.
✗ Branch 1 not taken.
 226229 if (ec_out)
156 {
157
2/2
✓ Branch 1 taken 202 times.
✓ Branch 2 taken 226027 times.
 226229 if (cancelled.load(std::memory_order_acquire))
158 202 *ec_out = capy::error::canceled;
159
2/2
✓ Branch 0 taken 1 time.
✓ Branch 1 taken 226026 times.
 226027 else if (errn != 0)
160 1 *ec_out = make_err(errn);
161
6/6
✓ Branch 1 taken 112976 times.
✓ Branch 2 taken 113050 times.
✓ Branch 3 taken 5 times.
✓ Branch 4 taken 112971 times.
✓ Branch 5 taken 5 times.
✓ Branch 6 taken 226021 times.
 226026 else if (is_read_operation() && bytes_transferred == 0)
162 5 *ec_out = capy::error::eof;
163 else
164 226021 *ec_out = {};
165 }
166
167
1/2
✓ Branch 0 taken 226229 times.
✗ Branch 1 not taken.
 226229 if (bytes_out)
168 226229 *bytes_out = bytes_transferred;
169
170 // Move to stack before destroying the frame
171 226229 capy::executor_ref saved_ex( std::move( ex ) );
172 226229 capy::coro saved_h( std::move( h ) );
173 226229 impl_ptr.reset();
174
1/1
✓ Branch 1 taken 226229 times.
 226229 saved_ex.dispatch( saved_h );
175 226229 }
176
177 113049 virtual bool is_read_operation() const noexcept { return false; }
178 virtual void cancel() noexcept = 0;
179
180 ✗ void destroy() override
181 {
182 ✗ stop_cb.reset();
183 ✗ impl_ptr.reset();
184 ✗ }
185
186 26899 void request_cancel() noexcept
187 {
188 26899 cancelled.store(true, std::memory_order_release);
189 26899 }
190
191 void start(std::stop_token token)
192 {
193 cancelled.store(false, std::memory_order_release);
194 stop_cb.reset();
195 socket_impl_ = nullptr;
196 acceptor_impl_ = nullptr;
197
198 if (token.stop_possible())
199 stop_cb.emplace(token, canceller{this});
200 }
201
202 229159 void start(std::stop_token token, select_socket_impl* impl)
203 {
204 229159 cancelled.store(false, std::memory_order_release);
205 229159 stop_cb.reset();
206 229159 socket_impl_ = impl;
207 229159 acceptor_impl_ = nullptr;
208
209
2/2
✓ Branch 1 taken 100 times.
✓ Branch 2 taken 229059 times.
 229159 if (token.stop_possible())
210 100 stop_cb.emplace(token, canceller{this});
211 229159 }
212
213 2932 void start(std::stop_token token, select_acceptor_impl* impl)
214 {
215 2932 cancelled.store(false, std::memory_order_release);
216 2932 stop_cb.reset();
217 2932 socket_impl_ = nullptr;
218 2932 acceptor_impl_ = impl;
219
220
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2932 times.
 2932 if (token.stop_possible())
221 ✗ stop_cb.emplace(token, canceller{this});
222 2932 }
223
224 231971 void complete(int err, std::size_t bytes) noexcept
225 {
226 231971 errn = err;
227 231971 bytes_transferred = bytes;
228 231971 }
229
230 ✗ virtual void perform_io() noexcept {}
231 };
232
233
234 struct select_connect_op : select_op
235 {
236 endpoint target_endpoint;
237
238 2930 void reset() noexcept
239 {
240 2930 select_op::reset();
241 2930 target_endpoint = endpoint{};
242 2930 }
243
244 2930 void perform_io() noexcept override
245 {
246 // connect() completion status is retrieved via SO_ERROR, not return value
247 2930 int err = 0;
248 2930 socklen_t len = sizeof(err);
249
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2930 times.
 2930 if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
250 ✗ err = errno;
251 2930 complete(err, 0);
252 2930 }
253
254 // Defined in sockets.cpp where select_socket_impl is complete
255 void operator()() override;
256 void cancel() noexcept override;
257 };
258
259
260 struct select_read_op : select_op
261 {
262 static constexpr std::size_t max_buffers = 16;
263 iovec iovecs[max_buffers];
264 int iovec_count = 0;
265 bool empty_buffer_read = false;
266
267 112977 bool is_read_operation() const noexcept override
268 {
269 112977 return !empty_buffer_read;
270 }
271
272 113175 void reset() noexcept
273 {
274 113175 select_op::reset();
275 113175 iovec_count = 0;
276 113175 empty_buffer_read = false;
277 113175 }
278
279 80 void perform_io() noexcept override
280 {
281 80 ssize_t n = ::readv(fd, iovecs, iovec_count);
282
1/2
✓ Branch 0 taken 80 times.
✗ Branch 1 not taken.
 80 if (n >= 0)
283 80 complete(0, static_cast<std::size_t>(n));
284 else
285 ✗ complete(errno, 0);
286 80 }
287
288 void cancel() noexcept override;
289 };
290
291
292 struct select_write_op : select_op
293 {
294 static constexpr std::size_t max_buffers = 16;
295 iovec iovecs[max_buffers];
296 int iovec_count = 0;
297
298 113054 void reset() noexcept
299 {
300 113054 select_op::reset();
301 113054 iovec_count = 0;
302 113054 }
303
304 ✗ void perform_io() noexcept override
305 {
306 ✗ msghdr msg{};
307 ✗ msg.msg_iov = iovecs;
308 ✗ msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
309
310 ✗ ssize_t n = ::sendmsg(fd, &msg, MSG_NOSIGNAL);
311 ✗ if (n >= 0)
312 ✗ complete(0, static_cast<std::size_t>(n));
313 else
314 ✗ complete(errno, 0);
315 ✗ }
316
317 void cancel() noexcept override;
318 };
319
320
321 struct select_accept_op : select_op
322 {
323 int accepted_fd = -1;
324 io_object::io_object_impl* peer_impl = nullptr;
325 io_object::io_object_impl** impl_out = nullptr;
326
327 2932 void reset() noexcept
328 {
329 2932 select_op::reset();
330 2932 accepted_fd = -1;
331 2932 peer_impl = nullptr;
332 2932 impl_out = nullptr;
333 2932 }
334
335 2927 void perform_io() noexcept override
336 {
337 2927 sockaddr_in addr{};
338 2927 socklen_t addrlen = sizeof(addr);
339
340 // Note: select backend uses accept() + fcntl instead of accept4()
341 // for broader POSIX compatibility
342 2927 int new_fd = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
343
344
1/2
✓ Branch 0 taken 2927 times.
✗ Branch 1 not taken.
 2927 if (new_fd >= 0)
345 {
346 // Reject fds that exceed select()'s FD_SETSIZE limit.
347 // Better to fail now than during later async operations.
348
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2927 times.
 2927 if (new_fd >= FD_SETSIZE)
349 {
350 ✗ ::close(new_fd);
351 ✗ complete(EINVAL, 0);
352 ✗ return;
353 }
354
355 // Set non-blocking and close-on-exec flags.
356 // A non-blocking socket is essential for the async reactor;
357 // if we can't configure it, fail rather than risk blocking.
358 2927 int flags = ::fcntl(new_fd, F_GETFL, 0);
359
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2927 times.
 2927 if (flags == -1)
360 {
361 ✗ int err = errno;
362 ✗ ::close(new_fd);
363 ✗ complete(err, 0);
364 ✗ return;
365 }
366
367
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2927 times.
 2927 if (::fcntl(new_fd, F_SETFL, flags | O_NONBLOCK) == -1)
368 {
369 ✗ int err = errno;
370 ✗ ::close(new_fd);
371 ✗ complete(err, 0);
372 ✗ return;
373 }
374
375
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2927 times.
 2927 if (::fcntl(new_fd, F_SETFD, FD_CLOEXEC) == -1)
376 {
377 ✗ int err = errno;
378 ✗ ::close(new_fd);
379 ✗ complete(err, 0);
380 ✗ return;
381 }
382
383 2927 accepted_fd = new_fd;
384 2927 complete(0, 0);
385 }
386 else
387 {
388 ✗ complete(errno, 0);
389 }
390 }
391
392 // Defined in acceptors.cpp where select_acceptor_impl is complete
393 void operator()() override;
394 void cancel() noexcept override;
395 };
396
397 } // namespace boost::corosio::detail
398
399 #endif // BOOST_COROSIO_HAS_SELECT
400
401 #endif // BOOST_COROSIO_DETAIL_SELECT_OP_HPP
402