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 26017 : 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 26017 : ~reactor_stream_socket() override = default;
104 :
105 : /// Return the cached remote endpoint.
106 54 : Endpoint remote_endpoint() const noexcept override
107 : {
108 54 : return remote_endpoint_;
109 : }
110 :
111 : // --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
112 :
113 8577 : 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 8577 : return do_connect(h, ex, ep, token, ec);
121 : }
122 :
123 341903 : 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 341903 : return do_read_some(h, ex, param, token, ec, bytes_out);
132 : }
133 :
134 341263 : 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 341263 : return do_write_some(h, ex, param, token, ec, bytes_out);
143 : }
144 :
145 43 : 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 43 : 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 214 : void cancel() noexcept override
162 : {
163 214 : this->do_cancel();
164 214 : }
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 17162 : void set_endpoints(Endpoint local, Endpoint remote) noexcept
202 : {
203 17162 : this->local_endpoint_ = std::move(local);
204 17162 : remote_endpoint_ = std::move(remote);
205 17162 : }
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 : `wait_type::write` completes immediately. Read and error
251 : waits probe the descriptor with a zero-timeout `poll()` and
252 : complete at once if the condition already holds; otherwise
253 : the op re-probes under the descriptor mutex and parks,
254 : completing when a reactor event arrives and a fresh probe
255 : confirms the condition.
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 78044 : void do_close_socket() noexcept
270 : {
271 78044 : base_type::do_close_socket();
272 78044 : remote_endpoint_ = Endpoint{};
273 78044 : }
274 :
275 : private:
276 : // CRTP callbacks for reactor_basic_socket cancel/close
277 :
278 : template<class Op>
279 231 : reactor_op_base** op_to_desc_slot(Op& op) noexcept
280 : {
281 231 : if (&op == static_cast<void*>(&conn_))
282 5 : return &this->desc_state_.connect_op;
283 226 : if (&op == static_cast<void*>(&rd_))
284 213 : return &this->desc_state_.read_op;
285 13 : if (&op == static_cast<void*>(&wr_))
286 4 : return &this->desc_state_.write_op;
287 9 : if (&op == static_cast<void*>(&wait_rd_))
288 7 : return &this->desc_state_.wait_read_op;
289 2 : if (&op == static_cast<void*>(&wait_wr_))
290 MIS 0 : return &this->desc_state_.wait_write_op;
291 HIT 2 : if (&op == static_cast<void*>(&wait_er_))
292 2 : return &this->desc_state_.wait_error_op;
293 MIS 0 : return nullptr;
294 : }
295 :
296 : template<class Op>
297 HIT 18 : bool* op_to_cancel_flag(Op& op) noexcept
298 : {
299 18 : if (&op == static_cast<void*>(&conn_))
300 5 : return &this->desc_state_.connect_cancel_pending;
301 13 : if (&op == static_cast<void*>(&rd_))
302 5 : return &this->desc_state_.read_cancel_pending;
303 8 : if (&op == static_cast<void*>(&wr_))
304 4 : return &this->desc_state_.write_cancel_pending;
305 4 : if (&op == static_cast<void*>(&wait_rd_))
306 4 : return &this->desc_state_.wait_read_cancel_pending;
307 MIS 0 : if (&op == static_cast<void*>(&wait_wr_))
308 0 : return &this->desc_state_.wait_write_cancel_pending;
309 0 : if (&op == static_cast<void*>(&wait_er_))
310 0 : return &this->desc_state_.wait_error_cancel_pending;
311 0 : return nullptr;
312 : }
313 :
314 : template<class Fn>
315 HIT 78260 : void for_each_op(Fn fn) noexcept
316 : {
317 78260 : fn(conn_);
318 78260 : fn(rd_);
319 78260 : fn(wr_);
320 78260 : fn(wait_rd_);
321 78260 : fn(wait_wr_);
322 78260 : fn(wait_er_);
323 78260 : }
324 :
325 : template<class Fn>
326 78260 : void for_each_desc_entry(Fn fn) noexcept
327 : {
328 78260 : fn(conn_, this->desc_state_.connect_op);
329 78260 : fn(rd_, this->desc_state_.read_op);
330 78260 : fn(wr_, this->desc_state_.write_op);
331 78260 : fn(wait_rd_, this->desc_state_.wait_read_op);
332 78260 : fn(wait_wr_, this->desc_state_.wait_write_op);
333 78260 : fn(wait_er_, this->desc_state_.wait_error_op);
334 78260 : }
335 : };
336 :
337 : template<
338 : class Derived,
339 : class Service,
340 : class ConnOp,
341 : class ReadOp,
342 : class WriteOp,
343 : class WaitOp,
344 : class DescState,
345 : class ImplBase,
346 : class Endpoint>
347 : std::coroutine_handle<>
348 8577 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
349 : do_connect(
350 : std::coroutine_handle<> h,
351 : capy::executor_ref ex,
352 : Endpoint const& ep,
353 : std::stop_token const& token,
354 : std::error_code* ec)
355 : {
356 8577 : auto& op = conn_;
357 :
358 8577 : sockaddr_storage storage{};
359 8577 : socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
360 : int result =
361 8577 : ::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
362 :
363 8577 : if (result == 0)
364 : {
365 21 : sockaddr_storage local_storage{};
366 21 : socklen_t local_len = sizeof(local_storage);
367 21 : if (::getsockname(
368 : this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
369 21 : &local_len) == 0)
370 MIS 0 : this->local_endpoint_ =
371 HIT 21 : from_sockaddr_as(local_storage, local_len, Endpoint{});
372 21 : remote_endpoint_ = ep;
373 : }
374 :
375 8577 : if (result == 0 || errno != EINPROGRESS)
376 : {
377 25 : int err = (result < 0) ? errno : 0;
378 25 : if (this->svc_.scheduler().try_consume_inline_budget())
379 : {
380 MIS 0 : *ec = err ? make_err(err) : std::error_code{};
381 0 : op.cont.h = h;
382 0 : return dispatch_coro(ex, op.cont);
383 : }
384 HIT 25 : op.reset();
385 25 : op.h = h;
386 25 : op.ex = ex;
387 25 : op.ec_out = ec;
388 25 : op.fd = this->fd_;
389 25 : op.target_endpoint = ep;
390 25 : op.start(token, static_cast<Derived*>(this));
391 25 : op.impl_ptr = this->shared_from_this();
392 25 : op.complete(err, 0);
393 25 : this->svc_.post(&op);
394 25 : return std::noop_coroutine();
395 : }
396 :
397 : // EINPROGRESS — register with reactor
398 8552 : op.reset();
399 8552 : op.h = h;
400 8552 : op.ex = ex;
401 8552 : op.ec_out = ec;
402 8552 : op.fd = this->fd_;
403 8552 : op.target_endpoint = ep;
404 8552 : op.start(token, static_cast<Derived*>(this));
405 8552 : op.impl_ptr = this->shared_from_this();
406 :
407 8552 : this->register_op(
408 8552 : op, this->desc_state_.connect_op, this->desc_state_.write_ready,
409 8552 : this->desc_state_.connect_cancel_pending, true);
410 8552 : return std::noop_coroutine();
411 : }
412 :
413 : template<
414 : class Derived,
415 : class Service,
416 : class ConnOp,
417 : class ReadOp,
418 : class WriteOp,
419 : class WaitOp,
420 : class DescState,
421 : class ImplBase,
422 : class Endpoint>
423 : std::coroutine_handle<>
424 341903 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
425 : do_read_some(
426 : std::coroutine_handle<> h,
427 : capy::executor_ref ex,
428 : buffer_param param,
429 : std::stop_token const& token,
430 : std::error_code* ec,
431 : std::size_t* bytes_out)
432 : {
433 341903 : auto& op = rd_;
434 341903 : op.reset();
435 :
436 341903 : capy::mutable_buffer bufs[ReadOp::max_buffers];
437 341903 : op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
438 :
439 341903 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
440 : {
441 4 : op.empty_buffer_read = true;
442 4 : op.h = h;
443 4 : op.ex = ex;
444 4 : op.ec_out = ec;
445 4 : op.bytes_out = bytes_out;
446 4 : op.start(token, static_cast<Derived*>(this));
447 4 : op.impl_ptr = this->shared_from_this();
448 4 : op.complete(0, 0);
449 4 : this->svc_.post(&op);
450 4 : return std::noop_coroutine();
451 : }
452 :
453 683810 : for (int i = 0; i < op.iovec_count; ++i)
454 : {
455 341911 : op.iovecs[i].iov_base = bufs[i].data();
456 341911 : op.iovecs[i].iov_len = bufs[i].size();
457 : }
458 :
459 : // Speculative read; for the single-buffer case use recv() so the
460 : // kernel skips the readv iov_iter setup.
461 : ssize_t n;
462 341899 : if (op.iovec_count == 1)
463 : {
464 : do
465 : {
466 341891 : n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
467 : }
468 341891 : while (n < 0 && errno == EINTR);
469 : }
470 : else
471 : {
472 : do
473 : {
474 8 : n = ::readv(this->fd_, op.iovecs, op.iovec_count);
475 : }
476 8 : while (n < 0 && errno == EINTR);
477 : }
478 :
479 341899 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
480 : {
481 341149 : int err = (n < 0) ? errno : 0;
482 341149 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
483 :
484 341149 : if (this->svc_.scheduler().try_consume_inline_budget())
485 : {
486 272960 : if (err)
487 MIS 0 : *ec = make_err(err);
488 HIT 272960 : else if (n == 0)
489 15 : *ec = capy::error::eof;
490 : else
491 272945 : *ec = {};
492 272960 : *bytes_out = bytes;
493 272960 : op.cont.h = h;
494 272960 : return dispatch_coro(ex, op.cont);
495 : }
496 68189 : op.h = h;
497 68189 : op.ex = ex;
498 68189 : op.ec_out = ec;
499 68189 : op.bytes_out = bytes_out;
500 68189 : op.start(token, static_cast<Derived*>(this));
501 68189 : op.impl_ptr = this->shared_from_this();
502 68189 : op.complete(err, bytes);
503 68189 : this->svc_.post(&op);
504 68189 : return std::noop_coroutine();
505 : }
506 :
507 : // EAGAIN — register with reactor
508 750 : op.h = h;
509 750 : op.ex = ex;
510 750 : op.ec_out = ec;
511 750 : op.bytes_out = bytes_out;
512 750 : op.fd = this->fd_;
513 750 : op.start(token, static_cast<Derived*>(this));
514 750 : op.impl_ptr = this->shared_from_this();
515 :
516 750 : this->register_op(
517 750 : op, this->desc_state_.read_op, this->desc_state_.read_ready,
518 750 : this->desc_state_.read_cancel_pending);
519 750 : return std::noop_coroutine();
520 : }
521 :
522 : template<
523 : class Derived,
524 : class Service,
525 : class ConnOp,
526 : class ReadOp,
527 : class WriteOp,
528 : class WaitOp,
529 : class DescState,
530 : class ImplBase,
531 : class Endpoint>
532 : std::coroutine_handle<>
533 341263 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
534 : do_write_some(
535 : std::coroutine_handle<> h,
536 : capy::executor_ref ex,
537 : buffer_param param,
538 : std::stop_token const& token,
539 : std::error_code* ec,
540 : std::size_t* bytes_out)
541 : {
542 341263 : auto& op = wr_;
543 341263 : op.reset();
544 :
545 341263 : capy::mutable_buffer bufs[WriteOp::max_buffers];
546 341263 : op.iovec_count =
547 341263 : static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
548 :
549 341263 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
550 : {
551 4 : op.h = h;
552 4 : op.ex = ex;
553 4 : op.ec_out = ec;
554 4 : op.bytes_out = bytes_out;
555 4 : op.start(token, static_cast<Derived*>(this));
556 4 : op.impl_ptr = this->shared_from_this();
557 4 : op.complete(0, 0);
558 4 : this->svc_.post(&op);
559 4 : return std::noop_coroutine();
560 : }
561 :
562 682528 : for (int i = 0; i < op.iovec_count; ++i)
563 : {
564 341269 : op.iovecs[i].iov_base = bufs[i].data();
565 341269 : op.iovecs[i].iov_len = bufs[i].size();
566 : }
567 :
568 : // Speculative write; the single-buffer case dispatches to a
569 : // backend-specific fast path so the kernel skips msghdr/iov_iter
570 : // setup (and so each backend can pick the right SIGPIPE strategy).
571 : ssize_t n;
572 341259 : if (op.iovec_count == 1)
573 : {
574 682506 : n = WriteOp::write_policy::write_one(
575 341253 : this->fd_, bufs[0].data(), bufs[0].size());
576 : }
577 : else
578 : {
579 6 : n = WriteOp::write_policy::write(
580 6 : this->fd_, op.iovecs, op.iovec_count);
581 : }
582 :
583 341259 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
584 : {
585 341130 : int err = (n < 0) ? errno : 0;
586 341130 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
587 :
588 341130 : if (this->svc_.scheduler().try_consume_inline_budget())
589 : {
590 272871 : *ec = err ? make_err(err) : std::error_code{};
591 272871 : *bytes_out = bytes;
592 272871 : op.cont.h = h;
593 272871 : return dispatch_coro(ex, op.cont);
594 : }
595 68259 : op.h = h;
596 68259 : op.ex = ex;
597 68259 : op.ec_out = ec;
598 68259 : op.bytes_out = bytes_out;
599 68259 : op.start(token, static_cast<Derived*>(this));
600 68259 : op.impl_ptr = this->shared_from_this();
601 68259 : op.complete(err, bytes);
602 68259 : this->svc_.post(&op);
603 68259 : return std::noop_coroutine();
604 : }
605 :
606 : // EAGAIN — register with reactor
607 129 : op.h = h;
608 129 : op.ex = ex;
609 129 : op.ec_out = ec;
610 129 : op.bytes_out = bytes_out;
611 129 : op.fd = this->fd_;
612 129 : op.start(token, static_cast<Derived*>(this));
613 129 : op.impl_ptr = this->shared_from_this();
614 :
615 129 : this->register_op(
616 129 : op, this->desc_state_.write_op, this->desc_state_.write_ready,
617 129 : this->desc_state_.write_cancel_pending, true);
618 129 : return std::noop_coroutine();
619 : }
620 :
621 : template<
622 : class Derived,
623 : class Service,
624 : class ConnOp,
625 : class ReadOp,
626 : class WriteOp,
627 : class WaitOp,
628 : class DescState,
629 : class ImplBase,
630 : class Endpoint>
631 : std::coroutine_handle<>
632 43 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
633 : do_wait(
634 : std::coroutine_handle<> h,
635 : capy::executor_ref ex,
636 : wait_type w,
637 : std::stop_token const& token,
638 : std::error_code* ec)
639 : {
640 : // wait_type::write completes immediately on a connected socket,
641 : // matching asio's behavior on IOCP. Corosio's reactor backends use
642 : // edge-triggered EPOLLOUT, which would never fire on an already-
643 : // writable socket; an immediate completion is also a more useful
644 : // contract than parking until a non-writable -> writable transition.
645 43 : if (w == wait_type::write)
646 : {
647 8 : auto& op = wait_wr_;
648 8 : if (this->svc_.scheduler().try_consume_inline_budget())
649 : {
650 MIS 0 : *ec = std::error_code{};
651 0 : op.cont.h = h;
652 0 : return dispatch_coro(ex, op.cont);
653 : }
654 HIT 8 : op.reset();
655 8 : op.wait_event = reactor_event_write;
656 8 : op.h = h;
657 8 : op.ex = ex;
658 8 : op.ec_out = ec;
659 8 : op.fd = this->fd_;
660 8 : op.start(token, static_cast<Derived*>(this));
661 8 : op.impl_ptr = this->shared_from_this();
662 8 : op.complete(0, 0);
663 8 : this->svc_.post(&op);
664 8 : return std::noop_coroutine();
665 : }
666 :
667 : // Pick refs up-front to avoid duplicating the register_op call.
668 : WaitOp* op_ptr;
669 : reactor_op_base** desc_slot_ptr;
670 : bool* cancel_flag_ptr;
671 : std::uint32_t event;
672 :
673 35 : if (w == wait_type::read)
674 : {
675 21 : op_ptr = &wait_rd_;
676 21 : desc_slot_ptr = &this->desc_state_.wait_read_op;
677 21 : cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
678 21 : event = reactor_event_read;
679 : }
680 : else // wait_type::error
681 : {
682 14 : op_ptr = &wait_er_;
683 14 : desc_slot_ptr = &this->desc_state_.wait_error_op;
684 14 : cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
685 14 : event = reactor_event_error;
686 : }
687 :
688 35 : auto& op = *op_ptr;
689 :
690 : // Speculative probe, mirroring the speculative read: an
691 : // edge-triggered reactor cannot report a condition that already
692 : // holds, so a wait initiated on an already-ready socket would
693 : // otherwise park forever.
694 35 : int perr = 0;
695 35 : if (WaitOp::probe(this->fd_, event, perr))
696 : {
697 8 : if (this->svc_.scheduler().try_consume_inline_budget())
698 : {
699 4 : *ec = perr ? make_err(perr) : std::error_code{};
700 4 : op.cont.h = h;
701 4 : return dispatch_coro(ex, op.cont);
702 : }
703 4 : op.reset();
704 4 : op.wait_event = event;
705 4 : op.h = h;
706 4 : op.ex = ex;
707 4 : op.ec_out = ec;
708 4 : op.fd = this->fd_;
709 4 : op.start(token, static_cast<Derived*>(this));
710 4 : op.impl_ptr = this->shared_from_this();
711 4 : op.complete(perr, 0);
712 4 : this->svc_.post(&op);
713 4 : return std::noop_coroutine();
714 : }
715 :
716 27 : op.reset();
717 27 : op.wait_event = event;
718 27 : op.h = h;
719 27 : op.ex = ex;
720 27 : op.ec_out = ec;
721 27 : op.fd = this->fd_;
722 27 : op.start(token, static_cast<Derived*>(this));
723 27 : op.impl_ptr = this->shared_from_this();
724 :
725 : // Force register_op's ready path so the wait op re-probes under
726 : // the descriptor mutex before parking. An edge consumed between
727 : // the speculative probe above and the park (a concurrent short
728 : // read, or an error event dispatched to an empty slot) would
729 : // otherwise leave the wait parked on a ready socket.
730 27 : bool force_probe = true;
731 27 : this->register_op(op, *desc_slot_ptr, force_probe, *cancel_flag_ptr,
732 : false);
733 27 : return std::noop_coroutine();
734 : }
735 :
736 : } // namespace boost::corosio::detail
737 :
738 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
|
