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