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