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