TLA Line data 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_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
12 :
13 : #include <boost/corosio/tcp_socket.hpp>
14 : #include <boost/corosio/shutdown_type.hpp>
15 : #include <boost/corosio/wait_type.hpp>
16 : #include <boost/corosio/native/detail/reactor/reactor_basic_socket.hpp>
17 : #include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
18 : #include <boost/corosio/detail/dispatch_coro.hpp>
19 : #include <boost/capy/buffers.hpp>
20 :
21 : #include <coroutine>
22 :
23 : #include <errno.h>
24 : #include <sys/socket.h>
25 : #include <sys/uio.h>
26 :
27 : namespace boost::corosio::detail {
28 :
29 : /** CRTP base for reactor-backed stream socket implementations.
30 :
31 : Inherits shared data members and cancel/close/register logic
32 : from reactor_basic_socket. Adds the stream-specific remote
33 : endpoint, shutdown, and I/O dispatch (connect, read, write, wait).
34 :
35 : @tparam Derived The concrete socket type (CRTP).
36 : @tparam Service The backend's socket service type.
37 : @tparam ConnOp The backend's connect op type.
38 : @tparam ReadOp The backend's read op type.
39 : @tparam WriteOp The backend's write op type.
40 : @tparam WaitOp The backend's wait op type.
41 : @tparam DescState The backend's descriptor_state type.
42 : @tparam ImplBase The public vtable base
43 : (tcp_socket::implementation or
44 : local_stream_socket::implementation).
45 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
46 : */
47 : template<
48 : class Derived,
49 : class Service,
50 : class ConnOp,
51 : class ReadOp,
52 : class WriteOp,
53 : class WaitOp,
54 : class DescState,
55 : class ImplBase = tcp_socket::implementation,
56 : class Endpoint = endpoint>
57 : class reactor_stream_socket
58 : : public reactor_basic_socket<
59 : Derived,
60 : ImplBase,
61 : Service,
62 : DescState,
63 : Endpoint>
64 : {
65 : using base_type = reactor_basic_socket<
66 : Derived,
67 : ImplBase,
68 : Service,
69 : DescState,
70 : Endpoint>;
71 : using self_type = reactor_stream_socket<
72 : Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp,
73 : DescState, ImplBase, Endpoint>;
74 : friend base_type;
75 : friend Derived;
76 :
77 : protected:
78 : // NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
79 HIT 21512 : explicit reactor_stream_socket(Service& svc) noexcept : base_type(svc) {}
80 :
81 : protected:
82 : Endpoint remote_endpoint_;
83 :
84 : public:
85 : /// Pending connect operation slot.
86 : ConnOp conn_;
87 :
88 : /// Pending read operation slot.
89 : ReadOp rd_;
90 :
91 : /// Pending write operation slot.
92 : WriteOp wr_;
93 :
94 : /// Pending wait-for-read operation slot.
95 : WaitOp wait_rd_;
96 :
97 : /// Pending wait-for-write operation slot.
98 : WaitOp wait_wr_;
99 :
100 : /// Pending wait-for-error operation slot.
101 : WaitOp wait_er_;
102 :
103 21512 : ~reactor_stream_socket() override = default;
104 :
105 : /// Return the cached remote endpoint.
106 60 : Endpoint remote_endpoint() const noexcept override
107 : {
108 60 : return remote_endpoint_;
109 : }
110 :
111 : // --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
112 :
113 7052 : std::coroutine_handle<> connect(
114 : std::coroutine_handle<> h,
115 : capy::executor_ref ex,
116 : Endpoint ep,
117 : std::stop_token token,
118 : std::error_code* ec) override
119 : {
120 7052 : return do_connect(h, ex, ep, token, ec);
121 : }
122 :
123 337754 : std::coroutine_handle<> read_some(
124 : std::coroutine_handle<> h,
125 : capy::executor_ref ex,
126 : buffer_param param,
127 : std::stop_token token,
128 : std::error_code* ec,
129 : std::size_t* bytes_out) override
130 : {
131 337754 : return do_read_some(h, ex, param, token, ec, bytes_out);
132 : }
133 :
134 337096 : std::coroutine_handle<> write_some(
135 : std::coroutine_handle<> h,
136 : capy::executor_ref ex,
137 : buffer_param param,
138 : std::stop_token token,
139 : std::error_code* ec,
140 : std::size_t* bytes_out) override
141 : {
142 337096 : return do_write_some(h, ex, param, token, ec, bytes_out);
143 : }
144 :
145 57 : std::coroutine_handle<> wait(
146 : std::coroutine_handle<> h,
147 : capy::executor_ref ex,
148 : wait_type w,
149 : std::stop_token token,
150 : std::error_code* ec) override
151 : {
152 57 : return do_wait(h, ex, w, token, ec);
153 : }
154 :
155 : std::error_code
156 23 : shutdown(corosio::shutdown_type what) noexcept override
157 : {
158 23 : return do_shutdown(static_cast<int>(what));
159 : }
160 :
161 216 : void cancel() noexcept override
162 : {
163 216 : this->do_cancel();
164 216 : }
165 :
166 : // --- End virtual overrides ---
167 :
168 : /// Close the socket (non-virtual, called by the service).
169 : void close_socket() noexcept
170 : {
171 : this->do_close_socket();
172 : }
173 :
174 : /** Shut down part or all of the full-duplex connection.
175 :
176 : @param what 0 = receive, 1 = send, 2 = both.
177 : */
178 23 : std::error_code do_shutdown(int what) noexcept
179 : {
180 : int how;
181 23 : switch (what)
182 : {
183 4 : case 0: // shutdown_receive
184 4 : how = SHUT_RD;
185 4 : break;
186 15 : case 1: // shutdown_send
187 15 : how = SHUT_WR;
188 15 : break;
189 4 : case 2: // shutdown_both
190 4 : how = SHUT_RDWR;
191 4 : break;
192 MIS 0 : default:
193 0 : return make_err(EINVAL);
194 : }
195 HIT 23 : if (::shutdown(this->fd_, how) != 0)
196 MIS 0 : return make_err(errno);
197 HIT 23 : return {};
198 : }
199 :
200 : /// Cache local and remote endpoints.
201 14147 : void set_endpoints(Endpoint local, Endpoint remote) noexcept
202 : {
203 14147 : this->local_endpoint_ = std::move(local);
204 14147 : remote_endpoint_ = std::move(remote);
205 14147 : }
206 :
207 : /** Shared connect dispatch.
208 :
209 : Tries the connect syscall speculatively. On synchronous
210 : completion, returns via inline budget or posts through queue.
211 : On EINPROGRESS, registers with the reactor.
212 : */
213 : std::coroutine_handle<> do_connect(
214 : std::coroutine_handle<>,
215 : capy::executor_ref,
216 : Endpoint const&,
217 : std::stop_token const&,
218 : std::error_code*);
219 :
220 : /** Shared scatter-read dispatch.
221 :
222 : Tries readv() speculatively. On success or hard error,
223 : returns via inline budget or posts through queue.
224 : On EAGAIN, registers with the reactor.
225 : */
226 : std::coroutine_handle<> do_read_some(
227 : std::coroutine_handle<>,
228 : capy::executor_ref,
229 : buffer_param,
230 : std::stop_token const&,
231 : std::error_code*,
232 : std::size_t*);
233 :
234 : /** Shared gather-write dispatch.
235 :
236 : Tries the write via WriteOp::write_policy speculatively.
237 : On success or hard error, returns via inline budget or
238 : posts through queue. On EAGAIN, registers with the reactor.
239 : */
240 : std::coroutine_handle<> do_write_some(
241 : std::coroutine_handle<>,
242 : capy::executor_ref,
243 : buffer_param,
244 : std::stop_token const&,
245 : std::error_code*,
246 : std::size_t*);
247 :
248 : /** Shared readiness-wait dispatch.
249 :
250 : Every wait type probes the descriptor with a zero-timeout
251 : `poll()` and completes at once if the condition already
252 : holds; otherwise the op re-probes under the descriptor mutex
253 : and parks, completing when a reactor event arrives and a
254 : fresh probe confirms the condition. A write wait therefore
255 : completes only while a non-blocking write can make progress.
256 : */
257 : std::coroutine_handle<> do_wait(
258 : std::coroutine_handle<>,
259 : capy::executor_ref,
260 : wait_type,
261 : std::stop_token const&,
262 : std::error_code*);
263 :
264 : /** Close the socket and cancel pending operations.
265 :
266 : Extends the base do_close_socket() to also reset
267 : the remote endpoint.
268 : */
269 64516 : void do_close_socket() noexcept
270 : {
271 64516 : base_type::do_close_socket();
272 64516 : remote_endpoint_ = Endpoint{};
273 64516 : }
274 :
275 : /// Release ownership of the descriptor and drop the cached peer.
276 6 : native_handle_type do_release_socket() noexcept
277 : {
278 6 : auto fd = base_type::do_release_socket();
279 6 : remote_endpoint_ = Endpoint{};
280 6 : return fd;
281 : }
282 :
283 : private:
284 : // CRTP callbacks for reactor_basic_socket cancel/close
285 :
286 : template<class Op>
287 232 : reactor_op_base** op_to_desc_slot(Op& op) noexcept
288 : {
289 232 : if (&op == static_cast<void*>(&conn_))
290 5 : return &this->desc_state_.connect_op;
291 227 : if (&op == static_cast<void*>(&rd_))
292 212 : return &this->desc_state_.read_op;
293 15 : if (&op == static_cast<void*>(&wr_))
294 4 : return &this->desc_state_.write_op;
295 11 : if (&op == static_cast<void*>(&wait_rd_))
296 7 : return &this->desc_state_.wait_read_op;
297 4 : if (&op == static_cast<void*>(&wait_wr_))
298 2 : return &this->desc_state_.wait_write_op;
299 2 : if (&op == static_cast<void*>(&wait_er_))
300 2 : return &this->desc_state_.wait_error_op;
301 MIS 0 : return nullptr;
302 : }
303 :
304 : template<class Op>
305 : bool* op_to_cancel_flag(Op& op) noexcept
306 : {
307 : if (&op == static_cast<void*>(&conn_))
308 : return &this->desc_state_.connect_cancel_pending;
309 : if (&op == static_cast<void*>(&rd_))
310 : return &this->desc_state_.read_cancel_pending;
311 : if (&op == static_cast<void*>(&wr_))
312 : return &this->desc_state_.write_cancel_pending;
313 : if (&op == static_cast<void*>(&wait_rd_))
314 : return &this->desc_state_.wait_read_cancel_pending;
315 : if (&op == static_cast<void*>(&wait_wr_))
316 : return &this->desc_state_.wait_write_cancel_pending;
317 : if (&op == static_cast<void*>(&wait_er_))
318 : return &this->desc_state_.wait_error_cancel_pending;
319 : return nullptr;
320 : }
321 :
322 : template<class Fn>
323 HIT 64738 : void for_each_op(Fn fn) noexcept
324 : {
325 64738 : fn(conn_);
326 64738 : fn(rd_);
327 64738 : fn(wr_);
328 64738 : fn(wait_rd_);
329 64738 : fn(wait_wr_);
330 64738 : fn(wait_er_);
331 64738 : }
332 :
333 : template<class Fn>
334 64738 : void for_each_desc_entry(Fn fn) noexcept
335 : {
336 64738 : fn(conn_, this->desc_state_.connect_op);
337 64738 : fn(rd_, this->desc_state_.read_op);
338 64738 : fn(wr_, this->desc_state_.write_op);
339 64738 : fn(wait_rd_, this->desc_state_.wait_read_op);
340 64738 : fn(wait_wr_, this->desc_state_.wait_write_op);
341 64738 : fn(wait_er_, this->desc_state_.wait_error_op);
342 64738 : }
343 : };
344 :
345 : template<
346 : class Derived,
347 : class Service,
348 : class ConnOp,
349 : class ReadOp,
350 : class WriteOp,
351 : class WaitOp,
352 : class DescState,
353 : class ImplBase,
354 : class Endpoint>
355 : std::coroutine_handle<>
356 7052 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
357 : do_connect(
358 : std::coroutine_handle<> h,
359 : capy::executor_ref ex,
360 : Endpoint const& ep,
361 : std::stop_token const& token,
362 : std::error_code* ec)
363 : {
364 7052 : auto& op = conn_;
365 :
366 7052 : sockaddr_storage storage{};
367 7052 : socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
368 : int result =
369 7052 : ::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
370 :
371 7052 : if (result == 0)
372 : {
373 27 : sockaddr_storage local_storage{};
374 27 : socklen_t local_len = sizeof(local_storage);
375 27 : if (::getsockname(
376 : this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
377 27 : &local_len) == 0)
378 MIS 0 : this->local_endpoint_ =
379 HIT 27 : from_sockaddr_as(local_storage, local_len, Endpoint{});
380 27 : remote_endpoint_ = ep;
381 : }
382 :
383 7052 : if (result == 0 || errno != EINPROGRESS)
384 : {
385 31 : int err = (result < 0) ? errno : 0;
386 31 : if (this->svc_.scheduler().try_consume_inline_budget())
387 : {
388 MIS 0 : *ec = err ? make_err(err) : std::error_code{};
389 0 : op.cont.h = h;
390 0 : return dispatch_coro(ex, op.cont);
391 : }
392 HIT 31 : op.reset();
393 31 : op.h = h;
394 31 : op.ex = ex;
395 31 : op.ec_out = ec;
396 31 : op.fd = this->fd_;
397 31 : op.target_endpoint = ep;
398 31 : op.start(token, static_cast<Derived*>(this));
399 31 : op.impl_ptr = this->shared_from_this();
400 31 : op.complete(err, 0);
401 31 : this->svc_.post(&op);
402 31 : return std::noop_coroutine();
403 : }
404 :
405 : // EINPROGRESS — register with reactor
406 7021 : op.reset();
407 7021 : op.h = h;
408 7021 : op.ex = ex;
409 7021 : op.ec_out = ec;
410 7021 : op.fd = this->fd_;
411 7021 : op.target_endpoint = ep;
412 7021 : op.start(token, static_cast<Derived*>(this));
413 7021 : op.impl_ptr = this->shared_from_this();
414 :
415 7021 : this->register_op(
416 7021 : op, this->desc_state_.connect_op, this->desc_state_.write_ready,
417 7021 : this->desc_state_.connect_cancel_pending, true);
418 7021 : return std::noop_coroutine();
419 : }
420 :
421 : template<
422 : class Derived,
423 : class Service,
424 : class ConnOp,
425 : class ReadOp,
426 : class WriteOp,
427 : class WaitOp,
428 : class DescState,
429 : class ImplBase,
430 : class Endpoint>
431 : std::coroutine_handle<>
432 337754 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
433 : do_read_some(
434 : std::coroutine_handle<> h,
435 : capy::executor_ref ex,
436 : buffer_param param,
437 : std::stop_token const& token,
438 : std::error_code* ec,
439 : std::size_t* bytes_out)
440 : {
441 337754 : auto& op = rd_;
442 337754 : op.reset();
443 :
444 : // Closed-object contract: complete with bad_file_descriptor without
445 : // touching the kernel or the unregistered descriptor state.
446 337754 : if (this->fd_ < 0)
447 : {
448 8 : op.h = h;
449 8 : op.ex = ex;
450 8 : op.ec_out = ec;
451 8 : op.bytes_out = bytes_out;
452 8 : op.start(token, static_cast<Derived*>(this));
453 8 : op.impl_ptr = this->shared_from_this();
454 8 : op.complete(EBADF, 0);
455 8 : this->svc_.post(&op);
456 8 : return std::noop_coroutine();
457 : }
458 :
459 337746 : capy::mutable_buffer bufs[ReadOp::max_buffers];
460 337746 : op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
461 :
462 337746 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
463 : {
464 4 : op.empty_buffer_read = true;
465 4 : op.h = h;
466 4 : op.ex = ex;
467 4 : op.ec_out = ec;
468 4 : op.bytes_out = bytes_out;
469 4 : op.start(token, static_cast<Derived*>(this));
470 4 : op.impl_ptr = this->shared_from_this();
471 4 : op.complete(0, 0);
472 4 : this->svc_.post(&op);
473 4 : return std::noop_coroutine();
474 : }
475 :
476 675496 : for (int i = 0; i < op.iovec_count; ++i)
477 : {
478 337754 : op.iovecs[i].iov_base = bufs[i].data();
479 337754 : op.iovecs[i].iov_len = bufs[i].size();
480 : }
481 :
482 : // Speculative read; for the single-buffer case use recv() so the
483 : // kernel skips the readv iov_iter setup.
484 : ssize_t n;
485 337742 : if (op.iovec_count == 1)
486 : {
487 : do
488 : {
489 337734 : n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
490 : }
491 337734 : while (n < 0 && errno == EINTR);
492 : }
493 : else
494 : {
495 : do
496 : {
497 8 : n = ::readv(this->fd_, op.iovecs, op.iovec_count);
498 : }
499 8 : while (n < 0 && errno == EINTR);
500 : }
501 :
502 337742 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
503 : {
504 336983 : int err = (n < 0) ? errno : 0;
505 336983 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
506 :
507 336983 : if (this->svc_.scheduler().try_consume_inline_budget())
508 : {
509 269626 : if (err)
510 MIS 0 : *ec = make_err(err);
511 HIT 269626 : else if (n == 0)
512 15 : *ec = capy::error::eof;
513 : else
514 269611 : *ec = {};
515 269626 : *bytes_out = bytes;
516 269626 : op.cont.h = h;
517 269626 : return dispatch_coro(ex, op.cont);
518 : }
519 67357 : op.h = h;
520 67357 : op.ex = ex;
521 67357 : op.ec_out = ec;
522 67357 : op.bytes_out = bytes_out;
523 67357 : op.start(token, static_cast<Derived*>(this));
524 67357 : op.impl_ptr = this->shared_from_this();
525 67357 : op.complete(err, bytes);
526 67357 : this->svc_.post(&op);
527 67357 : return std::noop_coroutine();
528 : }
529 :
530 : // EAGAIN — register with reactor
531 759 : op.h = h;
532 759 : op.ex = ex;
533 759 : op.ec_out = ec;
534 759 : op.bytes_out = bytes_out;
535 759 : op.fd = this->fd_;
536 759 : op.start(token, static_cast<Derived*>(this));
537 759 : op.impl_ptr = this->shared_from_this();
538 :
539 759 : this->register_op(
540 759 : op, this->desc_state_.read_op, this->desc_state_.read_ready,
541 759 : this->desc_state_.read_cancel_pending);
542 759 : return std::noop_coroutine();
543 : }
544 :
545 : template<
546 : class Derived,
547 : class Service,
548 : class ConnOp,
549 : class ReadOp,
550 : class WriteOp,
551 : class WaitOp,
552 : class DescState,
553 : class ImplBase,
554 : class Endpoint>
555 : std::coroutine_handle<>
556 337096 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
557 : do_write_some(
558 : std::coroutine_handle<> h,
559 : capy::executor_ref ex,
560 : buffer_param param,
561 : std::stop_token const& token,
562 : std::error_code* ec,
563 : std::size_t* bytes_out)
564 : {
565 337096 : auto& op = wr_;
566 337096 : op.reset();
567 :
568 : // Closed-object contract: complete with bad_file_descriptor without
569 : // touching the kernel or the unregistered descriptor state.
570 337096 : if (this->fd_ < 0)
571 : {
572 8 : op.h = h;
573 8 : op.ex = ex;
574 8 : op.ec_out = ec;
575 8 : op.bytes_out = bytes_out;
576 8 : op.start(token, static_cast<Derived*>(this));
577 8 : op.impl_ptr = this->shared_from_this();
578 8 : op.complete(EBADF, 0);
579 8 : this->svc_.post(&op);
580 8 : return std::noop_coroutine();
581 : }
582 :
583 337088 : capy::mutable_buffer bufs[WriteOp::max_buffers];
584 337088 : op.iovec_count =
585 337088 : static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
586 :
587 337088 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
588 : {
589 4 : op.h = h;
590 4 : op.ex = ex;
591 4 : op.ec_out = ec;
592 4 : op.bytes_out = bytes_out;
593 4 : op.start(token, static_cast<Derived*>(this));
594 4 : op.impl_ptr = this->shared_from_this();
595 4 : op.complete(0, 0);
596 4 : this->svc_.post(&op);
597 4 : return std::noop_coroutine();
598 : }
599 :
600 674178 : for (int i = 0; i < op.iovec_count; ++i)
601 : {
602 337094 : op.iovecs[i].iov_base = bufs[i].data();
603 337094 : op.iovecs[i].iov_len = bufs[i].size();
604 : }
605 :
606 : // Speculative write; the single-buffer case dispatches to a
607 : // backend-specific fast path so the kernel skips msghdr/iov_iter
608 : // setup (and so each backend can pick the right SIGPIPE strategy).
609 : ssize_t n;
610 337084 : if (op.iovec_count == 1)
611 : {
612 674156 : n = WriteOp::write_policy::write_one(
613 337078 : this->fd_, bufs[0].data(), bufs[0].size());
614 : }
615 : else
616 : {
617 6 : n = WriteOp::write_policy::write(
618 6 : this->fd_, op.iovecs, op.iovec_count);
619 : }
620 :
621 337084 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
622 : {
623 336955 : int err = (n < 0) ? errno : 0;
624 336955 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
625 :
626 336955 : if (this->svc_.scheduler().try_consume_inline_budget())
627 : {
628 269533 : *ec = err ? make_err(err) : std::error_code{};
629 269533 : *bytes_out = bytes;
630 269533 : op.cont.h = h;
631 269533 : return dispatch_coro(ex, op.cont);
632 : }
633 67422 : op.h = h;
634 67422 : op.ex = ex;
635 67422 : op.ec_out = ec;
636 67422 : op.bytes_out = bytes_out;
637 67422 : op.start(token, static_cast<Derived*>(this));
638 67422 : op.impl_ptr = this->shared_from_this();
639 67422 : op.complete(err, bytes);
640 67422 : this->svc_.post(&op);
641 67422 : return std::noop_coroutine();
642 : }
643 :
644 : // EAGAIN — register with reactor
645 129 : op.h = h;
646 129 : op.ex = ex;
647 129 : op.ec_out = ec;
648 129 : op.bytes_out = bytes_out;
649 129 : op.fd = this->fd_;
650 129 : op.start(token, static_cast<Derived*>(this));
651 129 : op.impl_ptr = this->shared_from_this();
652 :
653 129 : this->register_op(
654 129 : op, this->desc_state_.write_op, this->desc_state_.write_ready,
655 129 : this->desc_state_.write_cancel_pending, true);
656 129 : return std::noop_coroutine();
657 : }
658 :
659 : template<
660 : class Derived,
661 : class Service,
662 : class ConnOp,
663 : class ReadOp,
664 : class WriteOp,
665 : class WaitOp,
666 : class DescState,
667 : class ImplBase,
668 : class Endpoint>
669 : std::coroutine_handle<>
670 57 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
671 : do_wait(
672 : std::coroutine_handle<> h,
673 : capy::executor_ref ex,
674 : wait_type w,
675 : std::stop_token const& token,
676 : std::error_code* ec)
677 : {
678 : // Pick refs up-front to avoid duplicating the register_op call.
679 : WaitOp* op_ptr;
680 : reactor_op_base** desc_slot_ptr;
681 : bool* cancel_flag_ptr;
682 : std::uint32_t event;
683 :
684 57 : if (w == wait_type::read)
685 : {
686 27 : op_ptr = &wait_rd_;
687 27 : desc_slot_ptr = &this->desc_state_.wait_read_op;
688 27 : cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
689 27 : event = reactor_event_read;
690 : }
691 30 : else if (w == wait_type::write)
692 : {
693 16 : op_ptr = &wait_wr_;
694 16 : desc_slot_ptr = &this->desc_state_.wait_write_op;
695 16 : cancel_flag_ptr = &this->desc_state_.wait_write_cancel_pending;
696 16 : event = reactor_event_write;
697 : }
698 : else // wait_type::error
699 : {
700 14 : op_ptr = &wait_er_;
701 14 : desc_slot_ptr = &this->desc_state_.wait_error_op;
702 14 : cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
703 14 : event = reactor_event_error;
704 : }
705 :
706 57 : auto& op = *op_ptr;
707 :
708 : // Speculative probe, mirroring the speculative read: an
709 : // edge-triggered reactor cannot report a condition that already
710 : // holds, so a wait initiated on an already-ready socket would
711 : // otherwise park forever.
712 57 : int perr = 0;
713 57 : if (WaitOp::probe(this->fd_, event, perr))
714 : {
715 24 : if (this->svc_.scheduler().try_consume_inline_budget())
716 : {
717 4 : *ec = perr ? make_err(perr) : std::error_code{};
718 4 : op.cont.h = h;
719 4 : return dispatch_coro(ex, op.cont);
720 : }
721 20 : op.reset();
722 20 : op.wait_event = event;
723 20 : op.h = h;
724 20 : op.ex = ex;
725 20 : op.ec_out = ec;
726 20 : op.fd = this->fd_;
727 20 : op.start(token, static_cast<Derived*>(this));
728 20 : op.impl_ptr = this->shared_from_this();
729 20 : op.complete(perr, 0);
730 20 : this->svc_.post(&op);
731 20 : return std::noop_coroutine();
732 : }
733 :
734 33 : op.reset();
735 33 : op.wait_event = event;
736 33 : op.h = h;
737 33 : op.ex = ex;
738 33 : op.ec_out = ec;
739 33 : op.fd = this->fd_;
740 33 : op.start(token, static_cast<Derived*>(this));
741 33 : op.impl_ptr = this->shared_from_this();
742 :
743 : // Force register_op's ready path so the wait op re-probes under
744 : // the descriptor mutex before parking. An edge consumed between
745 : // the speculative probe above and the park (a concurrent short
746 : // read, or an error event dispatched to an empty slot) would
747 : // otherwise leave the wait parked on a ready socket.
748 33 : bool force_probe = true;
749 33 : this->register_op(op, *desc_slot_ptr, force_probe, *cancel_flag_ptr,
750 : event == reactor_event_write);
751 33 : return std::noop_coroutine();
752 : }
753 :
754 : } // namespace boost::corosio::detail
755 :
756 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
|
