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_ACCEPTOR_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
12 :
13 : #include <boost/corosio/tcp_acceptor.hpp>
14 : #include <boost/corosio/wait_type.hpp>
15 : #include <boost/corosio/detail/intrusive.hpp>
16 : #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
17 : #include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
18 : #include <boost/corosio/native/detail/make_err.hpp>
19 : #include <boost/corosio/native/detail/endpoint_convert.hpp>
20 :
21 : #include <memory>
22 : #include <mutex>
23 : #include <utility>
24 :
25 : #include <errno.h>
26 : #include <netinet/in.h>
27 : #include <sys/socket.h>
28 : #include <unistd.h>
29 :
30 : namespace boost::corosio::detail {
31 :
32 : /** CRTP base for reactor-backed acceptor implementations.
33 :
34 : Provides shared data members, trivial virtual overrides, and
35 : non-virtual helper methods for cancellation and close. Concrete
36 : backends inherit and add `cancel()`, `close_socket()`, and
37 : `accept()` overrides that delegate to the `do_*` helpers.
38 :
39 : @tparam Derived The concrete acceptor type (CRTP).
40 : @tparam Service The backend's acceptor service type.
41 : @tparam Op The backend's base op type.
42 : @tparam AcceptOp The backend's accept op type.
43 : @tparam WaitOp The backend's wait op type.
44 : @tparam DescState The backend's descriptor_state type.
45 : @tparam ImplBase The public vtable base
46 : (tcp_acceptor::implementation or
47 : local_stream_acceptor::implementation).
48 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
49 : */
50 : template<
51 : class Derived,
52 : class Service,
53 : class Op,
54 : class AcceptOp,
55 : class WaitOp,
56 : class DescState,
57 : class ImplBase = tcp_acceptor::implementation,
58 : class Endpoint = endpoint>
59 : class reactor_acceptor
60 : : public ImplBase
61 : , public std::enable_shared_from_this<Derived>
62 : , public intrusive_list<Derived>::node
63 : {
64 : friend Derived;
65 :
66 : protected:
67 : // NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
68 HIT 772 : explicit reactor_acceptor(Service& svc) noexcept : svc_(svc) {}
69 :
70 : protected:
71 : Service& svc_;
72 : int fd_ = -1;
73 : Endpoint local_endpoint_;
74 :
75 : public:
76 : /// Pending accept operation slot.
77 : AcceptOp acc_;
78 :
79 : /// Pending wait-for-read operation slot.
80 : WaitOp wait_rd_;
81 :
82 : /// Pending wait-for-write operation slot.
83 : WaitOp wait_wr_;
84 :
85 : /// Pending wait-for-error operation slot.
86 : WaitOp wait_er_;
87 :
88 : /// Per-descriptor state for persistent reactor registration.
89 : DescState desc_state_;
90 :
91 772 : ~reactor_acceptor() override = default;
92 :
93 : /// Return the underlying file descriptor.
94 59 : native_handle_type native_handle() const noexcept override
95 : {
96 59 : return fd_;
97 : }
98 :
99 636 : corosio::family family() const noexcept override
100 : {
101 636 : return to_family(socket_family(fd_));
102 : }
103 :
104 : /// Release and return the native handle without closing it.
105 22 : native_handle_type release_socket() noexcept override
106 : {
107 22 : return do_release_socket();
108 : }
109 :
110 : /// Return the cached local endpoint.
111 4876 : Endpoint local_endpoint() const noexcept override
112 : {
113 4876 : return local_endpoint_;
114 : }
115 :
116 : /// Return true if the acceptor has an open file descriptor.
117 9155 : bool is_open() const noexcept override
118 : {
119 9155 : return fd_ >= 0;
120 : }
121 :
122 : /// Set a socket option.
123 611 : std::error_code set_option(
124 : int level,
125 : int optname,
126 : void const* data,
127 : std::size_t size) noexcept override
128 : {
129 611 : if (::setsockopt(
130 611 : fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
131 10 : return make_err(errno);
132 601 : return {};
133 : }
134 :
135 : /// Get a socket option.
136 : std::error_code
137 25 : get_option(int level, int optname, void* data, std::size_t* size)
138 : const noexcept override
139 : {
140 25 : socklen_t len = static_cast<socklen_t>(*size);
141 25 : if (::getsockopt(fd_, level, optname, data, &len) != 0)
142 10 : return make_err(errno);
143 15 : *size = static_cast<std::size_t>(len);
144 15 : return {};
145 : }
146 :
147 : /// Cache the local endpoint.
148 685 : void set_local_endpoint(Endpoint ep) noexcept
149 : {
150 685 : local_endpoint_ = std::move(ep);
151 685 : }
152 :
153 : /// Assign the fd and initialize descriptor state for the acceptor.
154 726 : void init_acceptor_fd(int fd) noexcept
155 : {
156 726 : fd_ = fd;
157 726 : desc_state_.fd = fd;
158 : {
159 726 : std::lock_guard lock(desc_state_.mutex);
160 726 : desc_state_.read_op = nullptr;
161 726 : desc_state_.wait_read_op = nullptr;
162 726 : desc_state_.wait_write_op = nullptr;
163 726 : desc_state_.wait_error_op = nullptr;
164 726 : }
165 726 : }
166 :
167 : /** Assign the fd, initialize descriptor state, and register with
168 : the reactor.
169 :
170 : Adoption skips `do_listen`, so the registration it performs
171 : has to happen here instead.
172 :
173 : @param fd The already-listening descriptor to adopt.
174 :
175 : @return The error if the reactor rejects the descriptor, in
176 : which case the implementation is left closed and the caller
177 : retains ownership of @a fd; otherwise a default constructed
178 : error code.
179 : */
180 18 : std::error_code init_and_register(int fd) noexcept
181 : {
182 18 : init_acceptor_fd(fd);
183 18 : if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
184 : {
185 1 : fd_ = -1;
186 1 : desc_state_.fd = -1;
187 1 : desc_state_.registered_events = 0;
188 1 : return ec;
189 : }
190 17 : return {};
191 : }
192 :
193 : /// Return a reference to the owning service.
194 4318 : Service& service() noexcept
195 : {
196 4318 : return svc_;
197 : }
198 :
199 19 : void cancel() noexcept override
200 : {
201 19 : do_cancel();
202 19 : }
203 :
204 : /// Close the acceptor (non-virtual, called by the service).
205 2972 : void close_socket() noexcept
206 : {
207 2972 : do_close_socket();
208 2972 : }
209 :
210 38 : std::coroutine_handle<> wait(
211 : std::coroutine_handle<> h,
212 : capy::executor_ref ex,
213 : wait_type w,
214 : std::stop_token token,
215 : std::error_code* ec) override
216 : {
217 38 : return do_wait(h, ex, w, token, ec);
218 : }
219 :
220 : /** Wait for readiness on the listen socket.
221 :
222 : For `wait_type::read`, completion signals that an incoming
223 : connection is pending and a subsequent accept will succeed
224 : without blocking; a connection already queued when the wait
225 : begins completes it immediately via an initiation probe.
226 :
227 : `wait_type::write` fails with `operation_not_supported` on
228 : every backend: writability carries no meaning for a
229 : listening socket.
230 : */
231 : std::coroutine_handle<> do_wait(
232 : std::coroutine_handle<>,
233 : capy::executor_ref,
234 : wait_type,
235 : std::stop_token const&,
236 : std::error_code*);
237 :
238 : /** Cancel a single pending operation.
239 :
240 : Claims the operation from the read_op descriptor slot
241 : under the mutex and posts it to the scheduler as cancelled.
242 :
243 : @param op The operation to cancel.
244 : */
245 : void cancel_single_op(Op& op) noexcept;
246 :
247 : /** Cancel the pending accept operation. */
248 : void do_cancel() noexcept;
249 :
250 : /** Close the acceptor and cancel pending operations.
251 :
252 : Invoked by the derived class's close_socket(). The
253 : derived class may add backend-specific cleanup after
254 : calling this method.
255 : */
256 : void do_close_socket() noexcept;
257 :
258 : /** Release the acceptor without closing the fd. */
259 : native_handle_type do_release_socket() noexcept;
260 :
261 : /** Bind the acceptor socket to an endpoint.
262 :
263 : Caches the resolved local endpoint (including ephemeral
264 : port) after a successful bind.
265 :
266 : @param ep The endpoint to bind to.
267 : @return The error code from bind(), or success.
268 : */
269 : std::error_code do_bind(Endpoint const& ep);
270 :
271 : /** Start listening on the acceptor socket.
272 :
273 : Registers the file descriptor with the reactor after
274 : a successful listen() call.
275 :
276 : @param backlog The listen backlog.
277 : @return The error code from listen() or from reactor
278 : registration, or success.
279 : */
280 : std::error_code do_listen(int backlog);
281 : };
282 :
283 : template<
284 : class Derived,
285 : class Service,
286 : class Op,
287 : class AcceptOp,
288 : class WaitOp,
289 : class DescState,
290 : class ImplBase,
291 : class Endpoint>
292 : void
293 146 : reactor_acceptor<
294 : Derived,
295 : Service,
296 : Op,
297 : AcceptOp,
298 : WaitOp,
299 : DescState,
300 : ImplBase,
301 : Endpoint>::cancel_single_op(Op& op) noexcept
302 : {
303 146 : auto self = this->weak_from_this().lock();
304 146 : if (!self)
305 MIS 0 : return;
306 :
307 HIT 146 : op.request_cancel();
308 :
309 146 : reactor_op_base* claimed = nullptr;
310 : {
311 146 : std::lock_guard lock(desc_state_.mutex);
312 1314 : auto try_claim = [&](reactor_op_base*& slot) {
313 584 : if (!claimed && slot == &op)
314 87 : claimed = std::exchange(slot, nullptr);
315 : };
316 146 : try_claim(desc_state_.read_op);
317 146 : try_claim(desc_state_.wait_read_op);
318 146 : try_claim(desc_state_.wait_write_op);
319 146 : try_claim(desc_state_.wait_error_op);
320 146 : }
321 146 : if (claimed)
322 : {
323 87 : op.impl_ptr = self;
324 87 : svc_.post(&op);
325 87 : svc_.work_finished();
326 : }
327 146 : }
328 :
329 : template<
330 : class Derived,
331 : class Service,
332 : class Op,
333 : class AcceptOp,
334 : class WaitOp,
335 : class DescState,
336 : class ImplBase,
337 : class Endpoint>
338 : void
339 19 : reactor_acceptor<
340 : Derived,
341 : Service,
342 : Op,
343 : AcceptOp,
344 : WaitOp,
345 : DescState,
346 : ImplBase,
347 : Endpoint>::do_cancel() noexcept
348 : {
349 19 : cancel_single_op(acc_);
350 19 : cancel_single_op(wait_rd_);
351 19 : cancel_single_op(wait_wr_);
352 19 : cancel_single_op(wait_er_);
353 19 : }
354 :
355 : template<
356 : class Derived,
357 : class Service,
358 : class Op,
359 : class AcceptOp,
360 : class WaitOp,
361 : class DescState,
362 : class ImplBase,
363 : class Endpoint>
364 : void
365 2972 : reactor_acceptor<
366 : Derived,
367 : Service,
368 : Op,
369 : AcceptOp,
370 : WaitOp,
371 : DescState,
372 : ImplBase,
373 : Endpoint>::do_close_socket() noexcept
374 : {
375 2972 : auto self = this->weak_from_this().lock();
376 2972 : if (self)
377 : {
378 2972 : acc_.request_cancel();
379 2972 : wait_rd_.request_cancel();
380 2972 : wait_wr_.request_cancel();
381 2972 : wait_er_.request_cancel();
382 :
383 2972 : reactor_op_base* claimed_acc = nullptr;
384 2972 : reactor_op_base* claimed_wr = nullptr;
385 2972 : reactor_op_base* claimed_ww = nullptr;
386 2972 : reactor_op_base* claimed_we = nullptr;
387 : {
388 2972 : std::lock_guard lock(desc_state_.mutex);
389 2972 : claimed_acc = std::exchange(desc_state_.read_op, nullptr);
390 2972 : claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
391 2972 : claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
392 2972 : claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
393 2972 : desc_state_.read_ready = false;
394 2972 : desc_state_.write_ready = false;
395 :
396 2972 : if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
397 32 : desc_state_.impl_ref_ = self;
398 2972 : }
399 :
400 26748 : auto repost = [&](reactor_op_base* claimed, reactor_op_base& op) {
401 11888 : if (claimed)
402 : {
403 19 : op.impl_ptr = self;
404 19 : svc_.post(&op);
405 19 : svc_.work_finished();
406 : }
407 : };
408 2972 : repost(claimed_acc, acc_);
409 2972 : repost(claimed_wr, wait_rd_);
410 2972 : repost(claimed_ww, wait_wr_);
411 2972 : repost(claimed_we, wait_er_);
412 : }
413 :
414 2972 : if (fd_ >= 0)
415 : {
416 703 : if (desc_state_.registered_events != 0)
417 620 : svc_.scheduler().deregister_descriptor(fd_);
418 703 : ::close(fd_);
419 703 : fd_ = -1;
420 : }
421 :
422 2972 : desc_state_.fd = -1;
423 2972 : desc_state_.registered_events = 0;
424 :
425 2972 : local_endpoint_ = Endpoint{};
426 2972 : }
427 :
428 : template<
429 : class Derived,
430 : class Service,
431 : class Op,
432 : class AcceptOp,
433 : class WaitOp,
434 : class DescState,
435 : class ImplBase,
436 : class Endpoint>
437 : native_handle_type
438 22 : reactor_acceptor<
439 : Derived,
440 : Service,
441 : Op,
442 : AcceptOp,
443 : WaitOp,
444 : DescState,
445 : ImplBase,
446 : Endpoint>::do_release_socket() noexcept
447 : {
448 22 : auto self = this->weak_from_this().lock();
449 22 : if (self)
450 : {
451 22 : acc_.request_cancel();
452 22 : wait_rd_.request_cancel();
453 22 : wait_wr_.request_cancel();
454 22 : wait_er_.request_cancel();
455 :
456 22 : reactor_op_base* claimed_acc = nullptr;
457 22 : reactor_op_base* claimed_wr = nullptr;
458 22 : reactor_op_base* claimed_ww = nullptr;
459 22 : reactor_op_base* claimed_we = nullptr;
460 : {
461 22 : std::lock_guard lock(desc_state_.mutex);
462 22 : claimed_acc = std::exchange(desc_state_.read_op, nullptr);
463 22 : claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
464 22 : claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
465 22 : claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
466 22 : desc_state_.read_ready = false;
467 22 : desc_state_.write_ready = false;
468 :
469 22 : if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
470 1 : desc_state_.impl_ref_ = self;
471 22 : }
472 :
473 198 : auto repost = [&](reactor_op_base* claimed, reactor_op_base& op) {
474 88 : if (claimed)
475 : {
476 5 : op.impl_ptr = self;
477 5 : svc_.post(&op);
478 5 : svc_.work_finished();
479 : }
480 : };
481 22 : repost(claimed_acc, acc_);
482 22 : repost(claimed_wr, wait_rd_);
483 22 : repost(claimed_ww, wait_wr_);
484 22 : repost(claimed_we, wait_er_);
485 : }
486 :
487 22 : native_handle_type released = fd_;
488 :
489 22 : if (fd_ >= 0)
490 : {
491 22 : if (desc_state_.registered_events != 0)
492 22 : svc_.scheduler().deregister_descriptor(fd_);
493 22 : fd_ = -1;
494 : }
495 :
496 22 : desc_state_.fd = -1;
497 22 : desc_state_.registered_events = 0;
498 :
499 22 : local_endpoint_ = Endpoint{};
500 :
501 44 : return released;
502 22 : }
503 :
504 : template<
505 : class Derived,
506 : class Service,
507 : class Op,
508 : class AcceptOp,
509 : class WaitOp,
510 : class DescState,
511 : class ImplBase,
512 : class Endpoint>
513 : std::error_code
514 684 : reactor_acceptor<
515 : Derived,
516 : Service,
517 : Op,
518 : AcceptOp,
519 : WaitOp,
520 : DescState,
521 : ImplBase,
522 : Endpoint>::do_bind(Endpoint const& ep)
523 : {
524 684 : sockaddr_storage storage{};
525 684 : socklen_t addrlen = to_sockaddr(ep, storage);
526 684 : if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
527 16 : return make_err(errno);
528 :
529 : // Cache local endpoint (resolves ephemeral port / path)
530 668 : sockaddr_storage local{};
531 668 : socklen_t local_len = sizeof(local);
532 668 : if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
533 : 0)
534 668 : set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
535 :
536 668 : return {};
537 : }
538 :
539 : template<
540 : class Derived,
541 : class Service,
542 : class Op,
543 : class AcceptOp,
544 : class WaitOp,
545 : class DescState,
546 : class ImplBase,
547 : class Endpoint>
548 : std::error_code
549 640 : reactor_acceptor<
550 : Derived,
551 : Service,
552 : Op,
553 : AcceptOp,
554 : WaitOp,
555 : DescState,
556 : ImplBase,
557 : Endpoint>::do_listen(int backlog)
558 : {
559 640 : if (::listen(fd_, backlog) < 0)
560 12 : return make_err(errno);
561 :
562 : // A re-listen only changes the backlog; the descriptor is already
563 : // registered and re-adding it would fail on epoll.
564 628 : if (desc_state_.registered_events != 0)
565 2 : return {};
566 :
567 626 : return svc_.scheduler().register_descriptor(fd_, &desc_state_);
568 : }
569 :
570 : template<
571 : class Derived,
572 : class Service,
573 : class Op,
574 : class AcceptOp,
575 : class WaitOp,
576 : class DescState,
577 : class ImplBase,
578 : class Endpoint>
579 : std::coroutine_handle<>
580 38 : reactor_acceptor<
581 : Derived,
582 : Service,
583 : Op,
584 : AcceptOp,
585 : WaitOp,
586 : DescState,
587 : ImplBase,
588 : Endpoint>::
589 : do_wait(
590 : std::coroutine_handle<> h,
591 : capy::executor_ref ex,
592 : wait_type w,
593 : std::stop_token const& token,
594 : std::error_code* ec)
595 : {
596 : // Writability carries no meaning for a listening socket; some
597 : // backends could only lie about it and others could never report
598 : // it, so the wait fails the same way everywhere instead.
599 38 : if (w == wait_type::write)
600 : {
601 6 : auto& op = wait_wr_;
602 6 : op.reset();
603 6 : op.wait_event = reactor_event_write;
604 6 : op.h = h;
605 6 : op.ex = ex;
606 6 : op.ec_out = ec;
607 6 : op.fd = this->fd_;
608 6 : op.start(token, static_cast<Derived*>(this));
609 6 : op.impl_ptr = this->shared_from_this();
610 6 : op.complete(ENOTSUP, 0);
611 6 : svc_.post(&op);
612 6 : return std::noop_coroutine();
613 : }
614 :
615 : WaitOp* op_ptr;
616 : reactor_op_base** desc_slot_ptr;
617 : std::uint32_t event;
618 :
619 32 : if (w == wait_type::read)
620 : {
621 26 : op_ptr = &wait_rd_;
622 26 : desc_slot_ptr = &desc_state_.wait_read_op;
623 26 : event = reactor_event_read;
624 : }
625 : else // wait_type::error
626 : {
627 6 : op_ptr = &wait_er_;
628 6 : desc_slot_ptr = &desc_state_.wait_error_op;
629 6 : event = reactor_event_error;
630 : }
631 :
632 32 : auto& op = *op_ptr;
633 32 : op.reset();
634 32 : op.wait_event = event;
635 32 : op.h = h;
636 32 : op.ex = ex;
637 32 : op.ec_out = ec;
638 32 : op.fd = this->fd_;
639 32 : op.start(token, static_cast<Derived*>(this));
640 32 : op.impl_ptr = this->shared_from_this();
641 :
642 : // A listener's readiness can predate the wait: an adopted or
643 : // shared descriptor has history the reactor never saw, and an
644 : // edge already dispatched will not be re-announced. Probe before
645 : // parking.
646 32 : int perr = 0;
647 32 : if (WaitOp::probe(this->fd_, event, perr))
648 : {
649 11 : op.complete(perr, 0);
650 11 : svc_.post(&op);
651 11 : return std::noop_coroutine();
652 : }
653 :
654 21 : svc_.work_started();
655 :
656 21 : std::lock_guard lock(desc_state_.mutex);
657 21 : if (op.cancelled.load(std::memory_order_acquire))
658 : {
659 MIS 0 : svc_.post(&op);
660 0 : svc_.work_finished();
661 : }
662 HIT 21 : else if (WaitOp::probe(this->fd_, event, perr))
663 : {
664 : // Close the probe-to-park window: an edge that landed after
665 : // the first probe was consumed, so re-check under the mutex
666 : // the dispatch path holds.
667 MIS 0 : op.complete(perr, 0);
668 0 : svc_.post(&op);
669 0 : svc_.work_finished();
670 : }
671 : else
672 : {
673 HIT 21 : *desc_slot_ptr = &op;
674 : }
675 21 : return std::noop_coroutine();
676 21 : }
677 :
678 : } // namespace boost::corosio::detail
679 :
680 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
|
