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