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_BASIC_SOCKET_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
12 :
13 : #include <boost/corosio/detail/intrusive.hpp>
14 : #include <boost/corosio/detail/native_handle.hpp>
15 : #include <boost/corosio/endpoint.hpp>
16 : #include <boost/corosio/native/detail/native_socket_base.hpp>
17 : #include <boost/corosio/native/detail/reactor/reactor_op_base.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 socket implementations.
33 :
34 : Extracts the shared data members, virtual overrides, and
35 : cancel/close/register logic that is identical across TCP
36 : (reactor_stream_socket) and UDP (reactor_datagram_socket).
37 :
38 : Derived classes provide CRTP callbacks that enumerate their
39 : specific op slots so cancel/close can iterate them generically.
40 :
41 : @tparam Derived The concrete socket type (CRTP).
42 : @tparam ImplBase The public vtable base (tcp_socket::implementation
43 : or udp_socket::implementation).
44 : @tparam Service The backend's service type.
45 : @tparam DescState The backend's descriptor_state type.
46 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
47 : */
48 : template<
49 : class Derived,
50 : class ImplBase,
51 : class Service,
52 : class DescState,
53 : class Endpoint = endpoint>
54 : class reactor_basic_socket
55 : : public native_socket_base<Derived, ImplBase, Endpoint>
56 : , public intrusive_list<Derived>::node
57 : {
58 : friend Derived;
59 :
60 : template<class, class, class, class, class, class, class, class, class>
61 : friend class reactor_stream_socket;
62 :
63 : template<class, class, class, class, class, class, class, class, class, class, class>
64 : friend class reactor_datagram_socket;
65 :
66 HIT 20873 : explicit reactor_basic_socket(Service& svc) noexcept : svc_(svc) {}
67 :
68 : protected:
69 : // fd_ / local_endpoint_ and the synchronous accessors (native_handle,
70 : // is_open, set_option/get_option, set_socket/set_local_endpoint, do_bind)
71 : // live in native_socket_base — the readiness/completion-agnostic base
72 : // shared with io_uring's sockets. The using-declarations make the
73 : // inherited members visible to this template's own unqualified
74 : // references below (two-phase lookup).
75 : using native_socket_base<Derived, ImplBase, Endpoint>::fd_;
76 : using native_socket_base<Derived, ImplBase, Endpoint>::local_endpoint_;
77 :
78 : Service& svc_;
79 :
80 : public:
81 : /// Per-descriptor state for persistent reactor registration.
82 : DescState desc_state_;
83 :
84 20873 : ~reactor_basic_socket() override = default;
85 :
86 : /** Assign the fd, initialize descriptor state, and register with
87 : the reactor.
88 :
89 : @param fd The descriptor to adopt.
90 :
91 : @return The error if the reactor rejects the descriptor, in
92 : which case the implementation is left closed and the caller
93 : retains ownership of @a fd; otherwise a default constructed
94 : error code.
95 : */
96 7410 : std::error_code init_and_register(int fd) noexcept
97 : {
98 7410 : fd_ = fd;
99 7410 : desc_state_.fd = fd;
100 : {
101 7410 : std::lock_guard lock(desc_state_.mutex);
102 7410 : desc_state_.read_op = nullptr;
103 7410 : desc_state_.write_op = nullptr;
104 7410 : desc_state_.connect_op = nullptr;
105 7410 : }
106 7410 : if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
107 : {
108 : // Undo the partial state so a failed adopt is
109 : // indistinguishable from a closed implementation.
110 3 : fd_ = -1;
111 3 : desc_state_.fd = -1;
112 3 : desc_state_.registered_events = 0;
113 3 : return ec;
114 : }
115 7407 : return {};
116 : }
117 :
118 : /** Register an op with the reactor.
119 :
120 : Handles cached edge events. Called on the EAGAIN/EINPROGRESS
121 : path when speculative I/O failed.
122 : */
123 : template<class Op>
124 : void register_op(
125 : Op& op,
126 : reactor_op_base*& desc_slot,
127 : bool& ready_flag,
128 : bool is_write_direction = false) noexcept;
129 :
130 : /** Cancel a single pending operation.
131 :
132 : Claims the operation from its descriptor_state slot under
133 : the mutex and posts it to the scheduler as cancelled.
134 : Derived must implement:
135 : op_to_desc_slot(Op&) -> reactor_op_base**
136 : */
137 : template<class Op>
138 : void cancel_single_op(Op& op) noexcept;
139 :
140 : /** Cancel all pending operations.
141 :
142 : Invoked by the derived class's cancel() override.
143 : Derived must implement:
144 : for_each_op(auto fn)
145 : for_each_desc_entry(auto fn)
146 : */
147 : void do_cancel() noexcept;
148 :
149 : /** Close the socket and cancel pending operations.
150 :
151 : Invoked by the derived class's close_socket(). The
152 : derived class may add backend-specific cleanup after
153 : calling this method.
154 : Derived must implement:
155 : for_each_op(auto fn)
156 : for_each_desc_entry(auto fn)
157 : */
158 : void do_close_socket() noexcept;
159 :
160 : /** Release the socket without closing the fd.
161 :
162 : Like do_close_socket() but does not call ::close().
163 : Returns the fd so the caller can take ownership.
164 : */
165 : native_handle_type do_release_socket() noexcept;
166 : };
167 :
168 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
169 : template<class Op>
170 : void
171 7832 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_op(
172 : Op& op,
173 : reactor_op_base*& desc_slot,
174 : bool& ready_flag,
175 : bool is_write_direction) noexcept
176 : {
177 7832 : svc_.work_started();
178 :
179 7832 : std::lock_guard lock(desc_state_.mutex);
180 7832 : bool io_done = false;
181 7832 : if (ready_flag)
182 : {
183 301 : ready_flag = false;
184 301 : op.perform_io();
185 301 : io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
186 301 : if (!io_done)
187 295 : op.errn = 0;
188 : }
189 :
190 :
191 7832 : if (io_done || op.cancelled.load(std::memory_order_acquire))
192 : {
193 52 : svc_.post(&op);
194 52 : svc_.work_finished();
195 : }
196 : else
197 : {
198 7780 : desc_slot = &op;
199 :
200 : // Select must rebuild its fd_sets when a write-direction op
201 : // is parked, so select() watches for writability. Compiled
202 : // away to nothing for epoll and kqueue.
203 : if constexpr (requires { Service::needs_write_notification; })
204 : {
205 : if constexpr (Service::needs_write_notification)
206 : {
207 2859 : if (is_write_direction)
208 2351 : svc_.scheduler().notify_reactor();
209 : }
210 : }
211 : }
212 7832 : }
213 :
214 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
215 : template<class Op>
216 : void
217 297 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
218 : Op& op) noexcept
219 : {
220 297 : auto self = this->weak_from_this().lock();
221 297 : if (!self)
222 MIS 0 : return;
223 :
224 HIT 297 : op.request_cancel();
225 :
226 297 : auto* d = static_cast<Derived*>(this);
227 297 : reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
228 :
229 297 : if (desc_op_ptr)
230 : {
231 297 : reactor_op_base* claimed = nullptr;
232 : {
233 297 : std::lock_guard lock(desc_state_.mutex);
234 297 : if (*desc_op_ptr == &op)
235 227 : claimed = std::exchange(*desc_op_ptr, nullptr);
236 : // Not in the slot: request_cancel() above already set
237 : // op.cancelled, which register_op consults before parking
238 : // and the completion decode consults on delivery. Latching
239 : // a descriptor flag here instead would outlive this op and
240 : // cancel the next wait in the same direction.
241 297 : }
242 297 : if (claimed)
243 : {
244 227 : op.impl_ptr = self;
245 227 : svc_.post(&op);
246 227 : svc_.work_finished();
247 : }
248 : }
249 297 : }
250 :
251 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
252 : void
253 277 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
254 : do_cancel() noexcept
255 : {
256 277 : auto self = this->weak_from_this().lock();
257 277 : if (!self)
258 MIS 0 : return;
259 :
260 HIT 277 : auto* d = static_cast<Derived*>(this);
261 :
262 2055 : d->for_each_op([](auto& op) { op.request_cancel(); });
263 :
264 : // Claim ops under a single lock acquisition
265 : struct claimed_entry
266 : {
267 : reactor_op_base* op = nullptr;
268 : reactor_op_base* base = nullptr;
269 : };
270 : // Max 8 ops: conn, rd, wr, wait_rd, wait_wr, wait_er, recv_rd, send_wr
271 277 : claimed_entry claimed[8];
272 277 : int count = 0;
273 :
274 : {
275 277 : std::lock_guard lock(desc_state_.mutex);
276 3833 : d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
277 1778 : if (desc_slot == &op)
278 : {
279 188 : claimed[count].op = std::exchange(desc_slot, nullptr);
280 188 : claimed[count].base = &op;
281 188 : ++count;
282 : }
283 : });
284 277 : }
285 :
286 465 : for (int i = 0; i < count; ++i)
287 : {
288 188 : claimed[i].base->impl_ptr = self;
289 188 : svc_.post(claimed[i].base);
290 188 : svc_.work_finished();
291 : }
292 277 : }
293 :
294 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
295 : void
296 63099 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
297 : do_close_socket() noexcept
298 : {
299 63099 : auto self = this->weak_from_this().lock();
300 63099 : if (self)
301 : {
302 63099 : auto* d = static_cast<Derived*>(this);
303 :
304 446323 : d->for_each_op([](auto& op) { op.request_cancel(); });
305 :
306 : struct claimed_entry
307 : {
308 : reactor_op_base* base = nullptr;
309 : };
310 63099 : claimed_entry claimed[8];
311 63099 : int count = 0;
312 :
313 : {
314 63099 : std::lock_guard lock(desc_state_.mutex);
315 63099 : d->for_each_desc_entry(
316 766448 : [&](auto& /*op*/, reactor_op_base*& desc_slot) {
317 383224 : auto* c = std::exchange(desc_slot, nullptr);
318 383224 : if (c)
319 : {
320 56 : claimed[count].base = c;
321 56 : ++count;
322 : }
323 : });
324 63099 : desc_state_.read_ready = false;
325 63099 : desc_state_.write_ready = false;
326 :
327 63099 : if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
328 958 : desc_state_.impl_ref_ = self;
329 63099 : }
330 :
331 63155 : for (int i = 0; i < count; ++i)
332 : {
333 56 : claimed[i].base->impl_ptr = self;
334 56 : svc_.post(claimed[i].base);
335 56 : svc_.work_finished();
336 : }
337 : }
338 :
339 63099 : if (fd_ >= 0)
340 : {
341 13950 : if (desc_state_.registered_events != 0)
342 13948 : svc_.scheduler().deregister_descriptor(fd_);
343 13950 : ::close(fd_);
344 13950 : fd_ = -1;
345 : }
346 :
347 63099 : desc_state_.fd = -1;
348 63099 : desc_state_.registered_events = 0;
349 :
350 63099 : local_endpoint_ = Endpoint{};
351 63099 : }
352 :
353 : template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
354 : native_handle_type
355 16 : reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
356 : do_release_socket() noexcept
357 : {
358 : // Cancel pending ops (same as do_close_socket)
359 16 : auto self = this->weak_from_this().lock();
360 16 : if (self)
361 : {
362 16 : auto* d = static_cast<Derived*>(this);
363 :
364 128 : d->for_each_op([](auto& op) { op.request_cancel(); });
365 :
366 : struct claimed_entry
367 : {
368 : reactor_op_base* base = nullptr;
369 : };
370 16 : claimed_entry claimed[8];
371 16 : int count = 0;
372 :
373 : {
374 16 : std::lock_guard lock(desc_state_.mutex);
375 16 : d->for_each_desc_entry(
376 224 : [&](auto& /*op*/, reactor_op_base*& desc_slot) {
377 112 : auto* c = std::exchange(desc_slot, nullptr);
378 112 : if (c)
379 : {
380 12 : claimed[count].base = c;
381 12 : ++count;
382 : }
383 : });
384 16 : desc_state_.read_ready = false;
385 16 : desc_state_.write_ready = false;
386 :
387 16 : if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
388 3 : desc_state_.impl_ref_ = self;
389 16 : }
390 :
391 28 : for (int i = 0; i < count; ++i)
392 : {
393 12 : claimed[i].base->impl_ptr = self;
394 12 : svc_.post(claimed[i].base);
395 12 : svc_.work_finished();
396 : }
397 : }
398 :
399 16 : native_handle_type released = fd_;
400 :
401 16 : if (fd_ >= 0)
402 : {
403 16 : if (desc_state_.registered_events != 0)
404 16 : svc_.scheduler().deregister_descriptor(fd_);
405 : // Do NOT close -- caller takes ownership
406 16 : fd_ = -1;
407 : }
408 :
409 16 : desc_state_.fd = -1;
410 16 : desc_state_.registered_events = 0;
411 :
412 16 : local_endpoint_ = Endpoint{};
413 :
414 32 : return released;
415 16 : }
416 :
417 : } // namespace boost::corosio::detail
418 :
419 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
|
