TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Steve Gerbino
3 : // Copyright (c) 2026 Michael Vandeberg
4 : //
5 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 : //
8 : // Official repository: https://github.com/cppalliance/corosio
9 : //
10 :
11 : #ifndef BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
12 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
13 :
14 : #include <boost/corosio/tcp_acceptor.hpp>
15 : #include <boost/corosio/wait_type.hpp>
16 : #include <boost/corosio/detail/intrusive.hpp>
17 : #include <boost/corosio/detail/config.hpp>
18 : #include <boost/corosio/detail/object_ref.hpp>
19 : #include <boost/corosio/native/detail/reactor/reactor_io_core.hpp>
20 : #include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
21 : #include <boost/corosio/native/detail/make_err.hpp>
22 : #include <boost/corosio/native/detail/endpoint_convert.hpp>
23 :
24 : #include <cstring>
25 : #include <memory>
26 : #include <mutex>
27 : #include <utility>
28 :
29 : #include <errno.h>
30 : #include <netinet/in.h>
31 : #include <sys/socket.h>
32 : #include <unistd.h>
33 :
34 : namespace boost::corosio::detail {
35 :
36 : /** CRTP base for reactor-backed acceptor implementations.
37 :
38 : Provides shared data members, trivial virtual overrides, and
39 : non-virtual helper methods for cancellation and close. Concrete
40 : backends inherit and add `cancel()`, `close_socket()`, and
41 : `accept()` overrides that delegate to the `do_*` helpers.
42 :
43 : @tparam Derived The concrete acceptor type (CRTP).
44 : @tparam Service The backend's acceptor service type.
45 : @tparam Op The backend's base op type.
46 : @tparam AcceptOp The backend's accept op type.
47 : @tparam WaitOp The backend's wait op type.
48 : @tparam DescState The backend's descriptor_state type.
49 : @tparam ImplBase The public vtable base
50 : (tcp_acceptor::implementation or
51 : local_stream_acceptor::implementation).
52 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
53 : */
54 : template<
55 : class Derived,
56 : class Service,
57 : class Op,
58 : class AcceptOp,
59 : class WaitOp,
60 : class DescState,
61 : class ImplBase = tcp_acceptor::implementation,
62 : class Endpoint = endpoint>
63 : class reactor_acceptor
64 : : public ImplBase
65 : , public reactor_io_core<Derived, Service, DescState>
66 : , public intrusive_list<Derived>::node
67 : {
68 : friend Derived;
69 :
70 : using core_type = reactor_io_core<Derived, Service, DescState>;
71 : friend core_type;
72 :
73 : protected:
74 : // NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
75 HIT 1776 : explicit reactor_acceptor(Service& svc) noexcept : core_type(svc) {}
76 :
77 : protected:
78 : using core_type::svc_;
79 : int fd_ = -1;
80 : Endpoint local_endpoint_;
81 :
82 : public:
83 : /// Pending accept operation slot.
84 : AcceptOp acc_;
85 :
86 : /// Pending wait-for-read operation slot.
87 : WaitOp wait_rd_;
88 :
89 : /// Pending wait-for-write operation slot.
90 : WaitOp wait_wr_;
91 :
92 : /// Pending wait-for-error operation slot.
93 : WaitOp wait_er_;
94 :
95 : using core_type::desc_state_;
96 :
97 1776 : ~reactor_acceptor() override = default;
98 :
99 : /// Return the underlying file descriptor.
100 46 : native_handle_type native_handle() const noexcept override
101 : {
102 46 : return fd_;
103 : }
104 :
105 1411 : corosio::family family() const noexcept override
106 : {
107 1411 : return to_family(socket_family(fd_));
108 : }
109 :
110 : /// Release and return the native handle without closing it.
111 20 : native_handle_type release_socket() noexcept override
112 : {
113 20 : return do_release_socket();
114 : }
115 :
116 : /// Return the cached local endpoint.
117 7258 : Endpoint local_endpoint() const noexcept override
118 : {
119 7258 : return local_endpoint_;
120 : }
121 :
122 : /// Return true if the acceptor has an open file descriptor.
123 21269 : bool is_open() const noexcept override
124 : {
125 21269 : return fd_ >= 0;
126 : }
127 :
128 : /// Set a socket option.
129 1386 : std::error_code set_option(
130 : int level,
131 : int optname,
132 : void const* data,
133 : std::size_t size) noexcept override
134 : {
135 1386 : if (::setsockopt(
136 1386 : fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
137 10 : return make_err(errno);
138 1376 : return {};
139 : }
140 :
141 : /// Get a socket option.
142 : std::error_code
143 25 : get_option(int level, int optname, void* data, std::size_t* size)
144 : const noexcept override
145 : {
146 25 : socklen_t len = static_cast<socklen_t>(*size);
147 25 : if (::getsockopt(fd_, level, optname, data, &len) != 0)
148 10 : return make_err(errno);
149 15 : *size = static_cast<std::size_t>(len);
150 15 : return {};
151 : }
152 :
153 : /// Cache the local endpoint.
154 2298 : void set_local_endpoint(Endpoint ep) noexcept
155 : {
156 2298 : local_endpoint_ = std::move(ep);
157 2298 : }
158 :
159 : /// Assign the fd and initialize descriptor state for the acceptor.
160 2329 : void init_acceptor_fd(int fd) noexcept
161 : {
162 2329 : fd_ = fd;
163 2329 : desc_state_.fd = fd;
164 : {
165 2329 : std::lock_guard lock(desc_state_.mutex);
166 2329 : desc_state_.read_op = nullptr;
167 2329 : desc_state_.wait_read_op = nullptr;
168 2329 : desc_state_.wait_write_op = nullptr;
169 2329 : desc_state_.wait_error_op = nullptr;
170 2329 : }
171 2329 : }
172 :
173 : /** Assign the fd, initialize descriptor state, and register with
174 : the reactor.
175 :
176 : Adoption skips `do_listen`, so the registration it performs
177 : has to happen here instead.
178 :
179 : @param fd The already-listening descriptor to adopt.
180 :
181 : @return The error if the reactor rejects the descriptor, in
182 : which case the implementation is left closed and the caller
183 : retains ownership of @a fd; otherwise a default constructed
184 : error code.
185 : */
186 14 : std::error_code init_and_register(int fd) noexcept
187 : {
188 14 : fd_ = fd;
189 14 : if (auto ec = this->register_fd(fd))
190 : {
191 1 : fd_ = -1;
192 1 : return ec;
193 : }
194 13 : return {};
195 : }
196 :
197 : /// Return a reference to the owning service.
198 5927 : Service& service() noexcept
199 : {
200 5927 : return svc_;
201 : }
202 :
203 33 : void cancel() noexcept override
204 : {
205 33 : do_cancel();
206 33 : }
207 :
208 : /// Close the acceptor (non-virtual, called by the service).
209 9687 : void close_socket() noexcept
210 : {
211 9687 : do_close_socket();
212 9687 : }
213 :
214 41 : std::coroutine_handle<> wait(
215 : capy::continuation& cont,
216 : capy::executor_ref ex,
217 : wait_type w,
218 : std::stop_token token,
219 : std::error_code* ec) override
220 : {
221 41 : return do_wait(cont, ex, w, token, ec);
222 : }
223 :
224 : /** Wait for readiness on the listen socket.
225 :
226 : For `wait_type::read`, completion signals that an incoming
227 : connection is pending and a subsequent accept succeeds
228 : without blocking; a connection already queued when the wait
229 : begins completes it immediately via an initiation probe.
230 :
231 : `wait_type::write` fails with `operation_not_supported` on
232 : every backend: writability carries no meaning for a
233 : listening socket.
234 : */
235 : std::coroutine_handle<> do_wait(
236 : capy::continuation&,
237 : capy::executor_ref,
238 : wait_type,
239 : std::stop_token const&,
240 : std::error_code*);
241 :
242 : /** Cancel the pending accept operation. */
243 33 : void do_cancel() noexcept
244 : {
245 33 : this->cancel_all();
246 33 : }
247 :
248 : /** Close the acceptor and cancel pending operations.
249 :
250 : Invoked by the derived class's close_socket(). The
251 : derived class may add backend-specific cleanup after
252 : calling this method.
253 : */
254 9687 : void do_close_socket() noexcept
255 : {
256 9687 : this->abandon_all();
257 9687 : if (fd_ >= 0)
258 : {
259 2322 : ::close(fd_);
260 2322 : fd_ = -1;
261 : }
262 9687 : local_endpoint_ = Endpoint{};
263 9687 : }
264 :
265 : /** Release the acceptor without closing the fd. */
266 20 : native_handle_type do_release_socket() noexcept
267 : {
268 20 : this->abandon_all();
269 20 : native_handle_type released = fd_;
270 20 : fd_ = -1;
271 20 : local_endpoint_ = Endpoint{};
272 20 : return released;
273 : }
274 :
275 : /** Reset descriptor state for recycling.
276 :
277 : Actively re-initializes every field it owns rather than
278 : trusting `close_socket()`'s prior writes to still be there —
279 : see reactor_basic_socket::reuse() for the same rationale.
280 : `object_ref_` and `is_enqueued_` are asserted instead: `poison()`
281 : never touches them, so they must already hold their
282 : zero-action-complete state.
283 :
284 : @pre refs_ == 0, fd closed and deregistered, no op in flight.
285 : */
286 721 : void reuse() noexcept
287 : {
288 721 : fd_ = -1;
289 721 : local_endpoint_ = Endpoint{};
290 721 : this->reset_desc_state();
291 721 : BOOST_COROSIO_ASSERT(!acc_.stop_cb);
292 721 : BOOST_COROSIO_ASSERT(!wait_rd_.stop_cb);
293 721 : BOOST_COROSIO_ASSERT(!wait_wr_.stop_cb);
294 721 : BOOST_COROSIO_ASSERT(!wait_er_.stop_cb);
295 721 : }
296 :
297 : #if !defined(NDEBUG)
298 : /** Poison the fields `reuse()` re-initializes, before the pool
299 : parks this impl on the free list. See
300 : reactor_basic_socket::poison() for the full rationale and the
301 : list of fields deliberately left untouched.
302 :
303 : @pre refs_ == 0, fd closed and deregistered, no op in flight.
304 : */
305 2497 : void poison() noexcept
306 : {
307 4994 : auto smash = [](auto& field) {
308 4994 : std::memset(static_cast<void*>(&field), 0xDB, sizeof(field));
309 : };
310 2497 : smash(fd_);
311 2497 : smash(local_endpoint_);
312 2497 : this->poison_desc_state();
313 2497 : }
314 : #endif
315 :
316 : /** Bind the acceptor socket to an endpoint.
317 :
318 : Caches the resolved local endpoint (including ephemeral
319 : port) after a successful bind.
320 :
321 : @param ep The endpoint to bind to.
322 : @return The error code from bind(), or success.
323 : */
324 : std::error_code do_bind(Endpoint const& ep);
325 :
326 : /** Start listening on the acceptor socket.
327 :
328 : Registers the file descriptor with the reactor after
329 : a successful listen() call.
330 :
331 : @param backlog The listen backlog.
332 : @return The error code from listen() or from reactor
333 : registration, or success.
334 : */
335 : std::error_code do_listen(int backlog);
336 :
337 : private:
338 : // CRTP callbacks for reactor_io_core cancel/close
339 :
340 : template<class AnyOp>
341 338 : reactor_op_base** op_to_desc_slot(AnyOp& op) noexcept
342 : {
343 338 : if (&op == static_cast<void*>(&acc_))
344 338 : return &desc_state_.read_op;
345 MIS 0 : if (&op == static_cast<void*>(&wait_rd_))
346 0 : return &desc_state_.wait_read_op;
347 0 : if (&op == static_cast<void*>(&wait_wr_))
348 0 : return &desc_state_.wait_write_op;
349 0 : if (&op == static_cast<void*>(&wait_er_))
350 0 : return &desc_state_.wait_error_op;
351 0 : return nullptr;
352 : }
353 :
354 : template<class Fn>
355 HIT 9740 : void for_each_op(Fn fn) noexcept
356 : {
357 9740 : fn(acc_);
358 9740 : fn(wait_rd_);
359 9740 : fn(wait_wr_);
360 9740 : fn(wait_er_);
361 9740 : }
362 :
363 : template<class Fn>
364 9754 : void for_each_desc_entry(Fn fn) noexcept
365 : {
366 9754 : fn(acc_, desc_state_.read_op);
367 9754 : fn(wait_rd_, desc_state_.wait_read_op);
368 9754 : fn(wait_wr_, desc_state_.wait_write_op);
369 9754 : fn(wait_er_, desc_state_.wait_error_op);
370 9754 : }
371 : };
372 :
373 : template<
374 : class Derived,
375 : class Service,
376 : class Op,
377 : class AcceptOp,
378 : class WaitOp,
379 : class DescState,
380 : class ImplBase,
381 : class Endpoint>
382 : std::error_code
383 2301 : reactor_acceptor<
384 : Derived,
385 : Service,
386 : Op,
387 : AcceptOp,
388 : WaitOp,
389 : DescState,
390 : ImplBase,
391 : Endpoint>::do_bind(Endpoint const& ep)
392 : {
393 2301 : sockaddr_storage storage{};
394 2301 : socklen_t addrlen = to_sockaddr(ep, storage);
395 2301 : if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
396 16 : return make_err(errno);
397 :
398 : // Cache local endpoint (resolves ephemeral port / path)
399 2285 : sockaddr_storage local{};
400 2285 : socklen_t local_len = sizeof(local);
401 2285 : if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
402 : 0)
403 2285 : set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
404 :
405 2285 : return {};
406 : }
407 :
408 : template<
409 : class Derived,
410 : class Service,
411 : class Op,
412 : class AcceptOp,
413 : class WaitOp,
414 : class DescState,
415 : class ImplBase,
416 : class Endpoint>
417 : std::error_code
418 2257 : reactor_acceptor<
419 : Derived,
420 : Service,
421 : Op,
422 : AcceptOp,
423 : WaitOp,
424 : DescState,
425 : ImplBase,
426 : Endpoint>::do_listen(int backlog)
427 : {
428 2257 : if (::listen(fd_, backlog) < 0)
429 12 : return make_err(errno);
430 :
431 : // A re-listen only changes the backlog; the descriptor is already
432 : // registered and re-adding it would fail on epoll.
433 2245 : if (desc_state_.registered_events != 0)
434 2 : return {};
435 :
436 2243 : return svc_.scheduler().register_descriptor(fd_, &desc_state_);
437 : }
438 :
439 : template<
440 : class Derived,
441 : class Service,
442 : class Op,
443 : class AcceptOp,
444 : class WaitOp,
445 : class DescState,
446 : class ImplBase,
447 : class Endpoint>
448 : std::coroutine_handle<>
449 41 : reactor_acceptor<
450 : Derived,
451 : Service,
452 : Op,
453 : AcceptOp,
454 : WaitOp,
455 : DescState,
456 : ImplBase,
457 : Endpoint>::
458 : do_wait(
459 : capy::continuation& cont,
460 : capy::executor_ref ex,
461 : wait_type w,
462 : std::stop_token const& token,
463 : std::error_code* ec)
464 : {
465 : // Writability carries no meaning for a listening socket; some
466 : // backends could only lie about it and others could never report
467 : // it, so the wait fails the same way everywhere instead.
468 41 : if (w == wait_type::write)
469 : {
470 6 : auto& op = wait_wr_;
471 6 : op.reset();
472 6 : op.wait_event = reactor_event_write;
473 6 : op.cont = &cont;
474 6 : op.ex = ex;
475 6 : op.ec_out = ec;
476 6 : op.fd = this->fd_;
477 6 : op.start(token, static_cast<Derived*>(this));
478 6 : op.object_ref_ = detail::object_ref(this);
479 6 : op.complete(ENOTSUP, 0);
480 6 : svc_.post(&op);
481 6 : return std::noop_coroutine();
482 : }
483 :
484 : WaitOp* op_ptr;
485 : reactor_op_base** desc_slot_ptr;
486 : std::uint32_t event;
487 :
488 35 : if (w == wait_type::read)
489 : {
490 29 : op_ptr = &wait_rd_;
491 29 : desc_slot_ptr = &desc_state_.wait_read_op;
492 29 : event = reactor_event_read;
493 : }
494 : else // wait_type::error
495 : {
496 6 : op_ptr = &wait_er_;
497 6 : desc_slot_ptr = &desc_state_.wait_error_op;
498 6 : event = reactor_event_error;
499 : }
500 :
501 35 : auto& op = *op_ptr;
502 35 : op.reset();
503 35 : op.wait_event = event;
504 35 : op.cont = &cont;
505 35 : op.ex = ex;
506 35 : op.ec_out = ec;
507 35 : op.fd = this->fd_;
508 35 : op.start(token, static_cast<Derived*>(this));
509 35 : op.object_ref_ = detail::object_ref(this);
510 :
511 : // A listener's readiness can predate the wait: an adopted or
512 : // shared descriptor has history the reactor never saw, and an
513 : // edge already dispatched will not be re-announced. Probe before
514 : // parking.
515 35 : int perr = 0;
516 35 : if (WaitOp::probe(this->fd_, event, perr))
517 : {
518 12 : op.complete(perr, 0);
519 12 : svc_.post(&op);
520 12 : return std::noop_coroutine();
521 : }
522 :
523 23 : svc_.work_started();
524 :
525 : // Posted after the descriptor lock is released: the scheduler
526 : // takes its own lock before descriptor locks.
527 23 : bool post_now = false;
528 : {
529 23 : std::lock_guard lock(desc_state_.mutex);
530 23 : if (op.cancelled.load(std::memory_order_acquire))
531 : {
532 MIS 0 : post_now = true;
533 : }
534 HIT 23 : else if (WaitOp::probe(this->fd_, event, perr))
535 : {
536 : // Close the probe-to-park window: an edge that landed after
537 : // the first probe was consumed, so re-check under the mutex
538 : // the dispatch path holds.
539 MIS 0 : op.complete(perr, 0);
540 0 : post_now = true;
541 : }
542 : else
543 : {
544 HIT 23 : *desc_slot_ptr = &op;
545 : // Select watches an fd only while an op is parked; see
546 : // park_op.
547 : if constexpr (Service::needs_park_notification)
548 11 : svc_.scheduler().notify_reactor();
549 : }
550 23 : }
551 23 : if (post_now)
552 MIS 0 : svc_.scheduler().post_counted(&op);
553 HIT 23 : return std::noop_coroutine();
554 : }
555 :
556 : } // namespace boost::corosio::detail
557 :
558 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
|
