include/boost/corosio/native/detail/epoll/epoll_socket_service.hpp

81.0% Lines (336/415) 93.3% List of functions (28/30)

[ ] prev/next file n p prev/next uncovered line

f(x) Functions (30)

Function Calls Lines Branches Blocks

Line	Hits	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_EPOLL_EPOLL_SOCKET_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SOCKET_SERVICE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_EPOLL
16
17		#include <boost/corosio/detail/config.hpp>
18		#include <boost/capy/ex/execution_context.hpp>
19		#include <boost/corosio/detail/socket_service.hpp>
20
21		#include <boost/corosio/native/detail/epoll/epoll_socket.hpp>
22		#include <boost/corosio/native/detail/epoll/epoll_scheduler.hpp>
23
24		#include <boost/corosio/native/detail/endpoint_convert.hpp>
25		#include <boost/corosio/native/detail/make_err.hpp>
26		#include <boost/corosio/detail/dispatch_coro.hpp>
27		#include <boost/corosio/detail/except.hpp>
28		#include <boost/capy/buffers.hpp>
29
30		#include <coroutine>
31		#include <mutex>
32		#include <unordered_map>
33		#include <utility>
34
35		#include <errno.h>
36		#include <netinet/in.h>
37		#include <netinet/tcp.h>
38		#include <sys/epoll.h>
39		#include <sys/socket.h>
40		#include <unistd.h>
41
42		/*
43		epoll Socket Implementation
44		===========================
45
46		Each I/O operation follows the same pattern:
47		1. Try the syscall immediately (non-blocking socket)
48		2. If it succeeds or fails with a real error, post to completion queue
49		3. If EAGAIN/EWOULDBLOCK, register with epoll and wait
50
51		This "try first" approach avoids unnecessary epoll round-trips for
52		operations that can complete immediately (common for small reads/writes
53		on fast local connections).
54
55		One-Shot Registration
56		---------------------
57		We use one-shot epoll registration: each operation registers, waits for
58		one event, then unregisters. This simplifies the state machine since we
59		don't need to track whether an fd is currently registered or handle
60		re-arming. The tradeoff is slightly more epoll_ctl calls, but the
61		simplicity is worth it.
62
63		Cancellation
64		------------
65		See op.hpp for the completion/cancellation race handling via the
66		`registered` atomic. cancel() must complete pending operations (post
67		them with cancelled flag) so coroutines waiting on them can resume.
68		close_socket() calls cancel() first to ensure this.
69
70		Impl Lifetime with shared_ptr
71		-----------------------------
72		Socket impls use enable_shared_from_this. The service owns impls via
73		shared_ptr maps (socket_ptrs_) keyed by raw pointer for O(1) lookup and
74		removal. When a user calls close(), we call cancel() which posts pending
75		ops to the scheduler.
76
77		CRITICAL: The posted ops must keep the impl alive until they complete.
78		Otherwise the scheduler would process a freed op (use-after-free). The
79		cancel() method captures shared_from_this() into op.impl_ptr before
80		posting. When the op completes, impl_ptr is cleared, allowing the impl
81		to be destroyed if no other references exist.
82
83		Service Ownership
84		-----------------
85		epoll_socket_service owns all socket impls. destroy_impl() removes the
86		shared_ptr from the map, but the impl may survive if ops still hold
87		impl_ptr refs. shutdown() closes all sockets and clears the map; any
88		in-flight ops will complete and release their refs.
89		*/
90
91		namespace boost::corosio::detail {
92
93		/** State for epoll socket service. */
94		class epoll_socket_state
95		{
96		public:
97	244x	explicit epoll_socket_state(epoll_scheduler& sched) noexcept : sched_(sched)
98		{
99	244x	}
100
101		epoll_scheduler& sched_;
102		std::mutex mutex_;
103		intrusive_list<epoll_socket> socket_list_;
104		std::unordered_map<epoll_socket*, std::shared_ptr<epoll_socket>>
105		socket_ptrs_;
106		};
107
108		/** epoll socket service implementation.
109
110		Inherits from socket_service to enable runtime polymorphism.
111		Uses key_type = socket_service for service lookup.
112		*/
113		class BOOST_COROSIO_DECL epoll_socket_service final : public socket_service
114		{
115		public:
116		explicit epoll_socket_service(capy::execution_context& ctx);
117		~epoll_socket_service() override;
118
119		epoll_socket_service(epoll_socket_service const&) = delete;
120		epoll_socket_service& operator=(epoll_socket_service const&) = delete;
121
122		void shutdown() override;
123
124		io_object::implementation* construct() override;
125		void destroy(io_object::implementation*) override;
126		void close(io_object::handle&) override;
127		std::error_code open_socket(
128		tcp_socket::implementation& impl,
129		int family,
130		int type,
131		int protocol) override;
132
133	155165x	epoll_scheduler& scheduler() const noexcept
134		{
135	155165x	return state_->sched_;
136		}
137		void post(epoll_op* op);
138		void work_started() noexcept;
139		void work_finished() noexcept;
140
141		private:
142		std::unique_ptr<epoll_socket_state> state_;
143		};
144
145		//--------------------------------------------------------------------------
146		//
147		// Implementation
148		//
149		//--------------------------------------------------------------------------
150
151		// Register an op with the reactor, handling cached edge events.
152		// Called under the EAGAIN/EINPROGRESS path when speculative I/O failed.
153		inline void
154	3335x	epoll_socket::register_op(
155		epoll_op& op,
156		epoll_op*& desc_slot,
157		bool& ready_flag,
158		bool& cancel_flag) noexcept
159		{
160	3335x	svc_.work_started();
161
162	3335x	std::lock_guard lock(desc_state_.mutex);
163	3335x	bool io_done = false;
164	3335x	if (ready_flag)
165		{
166	132x	ready_flag = false;
167	132x	op.perform_io();
168	132x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
169	132x	if (!io_done)
170	132x	op.errn = 0;
171		}
172
173	3335x	if (cancel_flag)
174		{
175	87x	cancel_flag = false;
176	87x	op.cancelled.store(true, std::memory_order_relaxed);
177		}
178
179	3335x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
180		{
181	87x	svc_.post(&op);
182	87x	svc_.work_finished();
183		}
184		else
185		{
186	3248x	desc_slot = &op;
187		}
188	3335x	}
189
190		inline void
191	101x	epoll_op::canceller::operator()() const noexcept
192		{
193	101x	op->cancel();
194	101x	}
195
196		inline void
197	✗	epoll_connect_op::cancel() noexcept
198		{
199	✗	if (socket_impl_)
200	✗	socket_impl_->cancel_single_op(*this);
201		else
202	✗	request_cancel();
203	✗	}
204
205		inline void
206	95x	epoll_read_op::cancel() noexcept
207		{
208	95x	if (socket_impl_)
209	95x	socket_impl_->cancel_single_op(*this);
210		else
211	✗	request_cancel();
212	95x	}
213
214		inline void
215	✗	epoll_write_op::cancel() noexcept
216		{
217	✗	if (socket_impl_)
218	✗	socket_impl_->cancel_single_op(*this);
219		else
220	✗	request_cancel();
221	✗	}
222
223		inline void
224	23340x	epoll_op::operator()()
225		{
226	23340x	stop_cb.reset();
227
228	23340x	socket_impl_->svc_.scheduler().reset_inline_budget();
229
230	23340x	if (cancelled.load(std::memory_order_acquire))
231	194x	*ec_out = capy::error::canceled;
232	23146x	else if (errn != 0)
233	✗	*ec_out = make_err(errn);
234	23146x	else if (is_read_operation() && bytes_transferred == 0)
235	✗	*ec_out = capy::error::eof;
236		else
237	23146x	*ec_out = {};
238
239	23340x	*bytes_out = bytes_transferred;
240
241		// Move to stack before resuming coroutine. The coroutine might close
242		// the socket, releasing the last wrapper ref. If impl_ptr were the
243		// last ref and we destroyed it while still in operator(), we'd have
244		// use-after-free. Moving to local ensures destruction happens at
245		// function exit, after all member accesses are complete.
246	23340x	capy::executor_ref saved_ex(ex);
247	23340x	std::coroutine_handle<> saved_h(h);
248	23340x	auto prevent_premature_destruction = std::move(impl_ptr);
249	23340x	dispatch_coro(saved_ex, saved_h).resume();
250	23340x	}
251
252		inline void
253	3145x	epoll_connect_op::operator()()
254		{
255	3145x	stop_cb.reset();
256
257	3145x	socket_impl_->svc_.scheduler().reset_inline_budget();
258
259	3145x	bool success = (errn == 0 && !cancelled.load(std::memory_order_acquire));
260
261		// Cache endpoints on successful connect
262	3145x	if (success && socket_impl_)
263		{
264	3143x	endpoint local_ep;
265	3143x	sockaddr_storage local_storage{};
266	3143x	socklen_t local_len = sizeof(local_storage);
267	3143x	if (::getsockname(
268	3143x	fd, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
269		0)
270	3143x	local_ep = from_sockaddr(local_storage);
271	3143x	static_cast<epoll_socket*>(socket_impl_)
272	3143x	->set_endpoints(local_ep, target_endpoint);
273		}
274
275	3145x	if (cancelled.load(std::memory_order_acquire))
276	✗	*ec_out = capy::error::canceled;
277	3145x	else if (errn != 0)
278	2x	*ec_out = make_err(errn);
279		else
280	3143x	*ec_out = {};
281
282		// Move to stack before resuming. See epoll_op::operator()() for rationale.
283	3145x	capy::executor_ref saved_ex(ex);
284	3145x	std::coroutine_handle<> saved_h(h);
285	3145x	auto prevent_premature_destruction = std::move(impl_ptr);
286	3145x	dispatch_coro(saved_ex, saved_h).resume();
287	3145x	}
288
289	9492x	inline epoll_socket::epoll_socket(epoll_socket_service& svc) noexcept
290	9492x	: svc_(svc)
291		{
292	9492x	}
293
294	9492x	inline epoll_socket::~epoll_socket() = default;
295
296		inline std::coroutine_handle<>
297	3145x	epoll_socket::connect(
298		std::coroutine_handle<> h,
299		capy::executor_ref ex,
300		endpoint ep,
301		std::stop_token token,
302		std::error_code* ec)
303		{
304	3145x	auto& op = conn_;
305
306	3145x	sockaddr_storage storage{};
307		socklen_t addrlen =
308	3145x	detail::to_sockaddr(ep, detail::socket_family(fd_), storage);
309	3145x	int result = ::connect(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
310
311	3145x	if (result == 0)
312		{
313	✗	sockaddr_storage local_storage{};
314	✗	socklen_t local_len = sizeof(local_storage);
315	✗	if (::getsockname(
316	✗	fd_, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
317		0)
318	✗	local_endpoint_ = detail::from_sockaddr(local_storage);
319	✗	remote_endpoint_ = ep;
320		}
321
322	3145x	if (result == 0 \|\| errno != EINPROGRESS)
323		{
324	✗	int err = (result < 0) ? errno : 0;
325	✗	if (svc_.scheduler().try_consume_inline_budget())
326		{
327	✗	*ec = err ? make_err(err) : std::error_code{};
328	✗	return dispatch_coro(ex, h);
329		}
330	✗	op.reset();
331	✗	op.h = h;
332	✗	op.ex = ex;
333	✗	op.ec_out = ec;
334	✗	op.fd = fd_;
335	✗	op.target_endpoint = ep;
336	✗	op.start(token, this);
337	✗	op.impl_ptr = shared_from_this();
338	✗	op.complete(err, 0);
339	✗	svc_.post(&op);
340	✗	return std::noop_coroutine();
341		}
342
343		// EINPROGRESS — register with reactor
344	3145x	op.reset();
345	3145x	op.h = h;
346	3145x	op.ex = ex;
347	3145x	op.ec_out = ec;
348	3145x	op.fd = fd_;
349	3145x	op.target_endpoint = ep;
350	3145x	op.start(token, this);
351	3145x	op.impl_ptr = shared_from_this();
352
353	3145x	register_op(
354	3145x	op, desc_state_.connect_op, desc_state_.write_ready,
355	3145x	desc_state_.connect_cancel_pending);
356	3145x	return std::noop_coroutine();
357		}
358
359		inline std::coroutine_handle<>
360	58227x	epoll_socket::read_some(
361		std::coroutine_handle<> h,
362		capy::executor_ref ex,
363		buffer_param param,
364		std::stop_token token,
365		std::error_code* ec,
366		std::size_t* bytes_out)
367		{
368	58227x	auto& op = rd_;
369	58227x	op.reset();
370
371	58227x	capy::mutable_buffer bufs[epoll_read_op::max_buffers];
372	58227x	op.iovec_count =
373	58227x	static_cast<int>(param.copy_to(bufs, epoll_read_op::max_buffers));
374
375	58227x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
376		{
377	1x	op.empty_buffer_read = true;
378	1x	op.h = h;
379	1x	op.ex = ex;
380	1x	op.ec_out = ec;
381	1x	op.bytes_out = bytes_out;
382	1x	op.start(token, this);
383	1x	op.impl_ptr = shared_from_this();
384	1x	op.complete(0, 0);
385	1x	svc_.post(&op);
386	1x	return std::noop_coroutine();
387		}
388
389	116452x	for (int i = 0; i < op.iovec_count; ++i)
390		{
391	58226x	op.iovecs[i].iov_base = bufs[i].data();
392	58226x	op.iovecs[i].iov_len = bufs[i].size();
393		}
394
395		// Speculative read
396		ssize_t n;
397		do
398		{
399	58226x	n = ::readv(fd_, op.iovecs, op.iovec_count);
400		}
401	58226x	while (n < 0 && errno == EINTR);
402
403	58226x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
404		{
405	58036x	int err = (n < 0) ? errno : 0;
406	58036x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
407
408	58036x	if (svc_.scheduler().try_consume_inline_budget())
409		{
410	46471x	if (err)
411	✗	*ec = make_err(err);
412	46471x	else if (n == 0)
413	5x	*ec = capy::error::eof;
414		else
415	46466x	*ec = {};
416	46471x	*bytes_out = bytes;
417	46471x	return dispatch_coro(ex, h);
418		}
419	11565x	op.h = h;
420	11565x	op.ex = ex;
421	11565x	op.ec_out = ec;
422	11565x	op.bytes_out = bytes_out;
423	11565x	op.start(token, this);
424	11565x	op.impl_ptr = shared_from_this();
425	11565x	op.complete(err, bytes);
426	11565x	svc_.post(&op);
427	11565x	return std::noop_coroutine();
428		}
429
430		// EAGAIN — register with reactor
431	190x	op.h = h;
432	190x	op.ex = ex;
433	190x	op.ec_out = ec;
434	190x	op.bytes_out = bytes_out;
435	190x	op.fd = fd_;
436	190x	op.start(token, this);
437	190x	op.impl_ptr = shared_from_this();
438
439	190x	register_op(
440	190x	op, desc_state_.read_op, desc_state_.read_ready,
441	190x	desc_state_.read_cancel_pending);
442	190x	return std::noop_coroutine();
443		}
444
445		inline std::coroutine_handle<>
446	58039x	epoll_socket::write_some(
447		std::coroutine_handle<> h,
448		capy::executor_ref ex,
449		buffer_param param,
450		std::stop_token token,
451		std::error_code* ec,
452		std::size_t* bytes_out)
453		{
454	58039x	auto& op = wr_;
455	58039x	op.reset();
456
457	58039x	capy::mutable_buffer bufs[epoll_write_op::max_buffers];
458	58039x	op.iovec_count =
459	58039x	static_cast<int>(param.copy_to(bufs, epoll_write_op::max_buffers));
460
461	58039x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
462		{
463	1x	op.h = h;
464	1x	op.ex = ex;
465	1x	op.ec_out = ec;
466	1x	op.bytes_out = bytes_out;
467	1x	op.start(token, this);
468	1x	op.impl_ptr = shared_from_this();
469	1x	op.complete(0, 0);
470	1x	svc_.post(&op);
471	1x	return std::noop_coroutine();
472		}
473
474	116076x	for (int i = 0; i < op.iovec_count; ++i)
475		{
476	58038x	op.iovecs[i].iov_base = bufs[i].data();
477	58038x	op.iovecs[i].iov_len = bufs[i].size();
478		}
479
480		// Speculative write
481	58038x	msghdr msg{};
482	58038x	msg.msg_iov = op.iovecs;
483	58038x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
484
485		ssize_t n;
486		do
487		{
488	58038x	n = ::sendmsg(fd_, &msg, MSG_NOSIGNAL);
489		}
490	58038x	while (n < 0 && errno == EINTR);
491
492	58038x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
493		{
494	58038x	int err = (n < 0) ? errno : 0;
495	58038x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
496
497	58038x	if (svc_.scheduler().try_consume_inline_budget())
498		{
499	46455x	*ec = err ? make_err(err) : std::error_code{};
500	46455x	*bytes_out = bytes;
501	46455x	return dispatch_coro(ex, h);
502		}
503	11583x	op.h = h;
504	11583x	op.ex = ex;
505	11583x	op.ec_out = ec;
506	11583x	op.bytes_out = bytes_out;
507	11583x	op.start(token, this);
508	11583x	op.impl_ptr = shared_from_this();
509	11583x	op.complete(err, bytes);
510	11583x	svc_.post(&op);
511	11583x	return std::noop_coroutine();
512		}
513
514		// EAGAIN — register with reactor
515	✗	op.h = h;
516	✗	op.ex = ex;
517	✗	op.ec_out = ec;
518	✗	op.bytes_out = bytes_out;
519	✗	op.fd = fd_;
520	✗	op.start(token, this);
521	✗	op.impl_ptr = shared_from_this();
522
523	✗	register_op(
524	✗	op, desc_state_.write_op, desc_state_.write_ready,
525	✗	desc_state_.write_cancel_pending);
526	✗	return std::noop_coroutine();
527		}
528
529		inline std::error_code
530	3x	epoll_socket::shutdown(tcp_socket::shutdown_type what) noexcept
531		{
532		int how;
533	3x	switch (what)
534		{
535	1x	case tcp_socket::shutdown_receive:
536	1x	how = SHUT_RD;
537	1x	break;
538	1x	case tcp_socket::shutdown_send:
539	1x	how = SHUT_WR;
540	1x	break;
541	1x	case tcp_socket::shutdown_both:
542	1x	how = SHUT_RDWR;
543	1x	break;
544	✗	default:
545	✗	return make_err(EINVAL);
546		}
547	3x	if (::shutdown(fd_, how) != 0)
548	✗	return make_err(errno);
549	3x	return {};
550		}
551
552		inline std::error_code
553	32x	epoll_socket::set_option(
554		int level, int optname, void const* data, std::size_t size) noexcept
555		{
556	32x	if (::setsockopt(fd_, level, optname, data, static_cast<socklen_t>(size)) !=
557		0)
558	✗	return make_err(errno);
559	32x	return {};
560		}
561
562		inline std::error_code
563	31x	epoll_socket::get_option(
564		int level, int optname, void* data, std::size_t* size) const noexcept
565		{
566	31x	socklen_t len = static_cast<socklen_t>(*size);
567	31x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
568	✗	return make_err(errno);
569	31x	*size = static_cast<std::size_t>(len);
570	31x	return {};
571		}
572
573		inline void
574	171x	epoll_socket::cancel() noexcept
575		{
576	171x	auto self = weak_from_this().lock();
577	171x	if (!self)
578	✗	return;
579
580	171x	conn_.request_cancel();
581	171x	rd_.request_cancel();
582	171x	wr_.request_cancel();
583
584	171x	epoll_op* conn_claimed = nullptr;
585	171x	epoll_op* rd_claimed = nullptr;
586	171x	epoll_op* wr_claimed = nullptr;
587		{
588	171x	std::lock_guard lock(desc_state_.mutex);
589	171x	if (desc_state_.connect_op == &conn_)
590	✗	conn_claimed = std::exchange(desc_state_.connect_op, nullptr);
591		else
592	171x	desc_state_.connect_cancel_pending = true;
593	171x	if (desc_state_.read_op == &rd_)
594	3x	rd_claimed = std::exchange(desc_state_.read_op, nullptr);
595		else
596	168x	desc_state_.read_cancel_pending = true;
597	171x	if (desc_state_.write_op == &wr_)
598	✗	wr_claimed = std::exchange(desc_state_.write_op, nullptr);
599		else
600	171x	desc_state_.write_cancel_pending = true;
601	171x	}
602
603	171x	if (conn_claimed)
604		{
605	✗	conn_.impl_ptr = self;
606	✗	svc_.post(&conn_);
607	✗	svc_.work_finished();
608		}
609	171x	if (rd_claimed)
610		{
611	3x	rd_.impl_ptr = self;
612	3x	svc_.post(&rd_);
613	3x	svc_.work_finished();
614		}
615	171x	if (wr_claimed)
616		{
617	✗	wr_.impl_ptr = self;
618	✗	svc_.post(&wr_);
619	✗	svc_.work_finished();
620		}
621	171x	}
622
623		inline void
624	95x	epoll_socket::cancel_single_op(epoll_op& op) noexcept
625		{
626	95x	auto self = weak_from_this().lock();
627	95x	if (!self)
628	✗	return;
629
630	95x	op.request_cancel();
631
632	95x	epoll_op** desc_op_ptr = nullptr;
633	95x	if (&op == &conn_)
634	✗	desc_op_ptr = &desc_state_.connect_op;
635	95x	else if (&op == &rd_)
636	95x	desc_op_ptr = &desc_state_.read_op;
637	✗	else if (&op == &wr_)
638	✗	desc_op_ptr = &desc_state_.write_op;
639
640	95x	if (desc_op_ptr)
641		{
642	95x	epoll_op* claimed = nullptr;
643		{
644	95x	std::lock_guard lock(desc_state_.mutex);
645	95x	if (*desc_op_ptr == &op)
646	95x	claimed = std::exchange(*desc_op_ptr, nullptr);
647	✗	else if (&op == &conn_)
648	✗	desc_state_.connect_cancel_pending = true;
649	✗	else if (&op == &rd_)
650	✗	desc_state_.read_cancel_pending = true;
651	✗	else if (&op == &wr_)
652	✗	desc_state_.write_cancel_pending = true;
653	95x	}
654	95x	if (claimed)
655		{
656	95x	op.impl_ptr = self;
657	95x	svc_.post(&op);
658	95x	svc_.work_finished();
659		}
660		}
661	95x	}
662
663		inline void
664	28447x	epoll_socket::close_socket() noexcept
665		{
666	28447x	auto self = weak_from_this().lock();
667	28447x	if (self)
668		{
669	28447x	conn_.request_cancel();
670	28447x	rd_.request_cancel();
671	28447x	wr_.request_cancel();
672
673	28447x	epoll_op* conn_claimed = nullptr;
674	28447x	epoll_op* rd_claimed = nullptr;
675	28447x	epoll_op* wr_claimed = nullptr;
676		{
677	28447x	std::lock_guard lock(desc_state_.mutex);
678	28447x	conn_claimed = std::exchange(desc_state_.connect_op, nullptr);
679	28447x	rd_claimed = std::exchange(desc_state_.read_op, nullptr);
680	28447x	wr_claimed = std::exchange(desc_state_.write_op, nullptr);
681	28447x	desc_state_.read_ready = false;
682	28447x	desc_state_.write_ready = false;
683	28447x	desc_state_.read_cancel_pending = false;
684	28447x	desc_state_.write_cancel_pending = false;
685	28447x	desc_state_.connect_cancel_pending = false;
686	28447x	}
687
688	28447x	if (conn_claimed)
689		{
690	✗	conn_.impl_ptr = self;
691	✗	svc_.post(&conn_);
692	✗	svc_.work_finished();
693		}
694	28447x	if (rd_claimed)
695		{
696	1x	rd_.impl_ptr = self;
697	1x	svc_.post(&rd_);
698	1x	svc_.work_finished();
699		}
700	28447x	if (wr_claimed)
701		{
702	✗	wr_.impl_ptr = self;
703	✗	svc_.post(&wr_);
704	✗	svc_.work_finished();
705		}
706
707	28447x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
708	97x	desc_state_.impl_ref_ = self;
709		}
710
711	28447x	if (fd_ >= 0)
712		{
713	6303x	if (desc_state_.registered_events != 0)
714	6303x	svc_.scheduler().deregister_descriptor(fd_);
715	6303x	::close(fd_);
716	6303x	fd_ = -1;
717		}
718
719	28447x	desc_state_.fd = -1;
720	28447x	desc_state_.registered_events = 0;
721
722	28447x	local_endpoint_ = endpoint{};
723	28447x	remote_endpoint_ = endpoint{};
724	28447x	}
725
726	244x	inline epoll_socket_service::epoll_socket_service(capy::execution_context& ctx)
727	244x	: state_(
728		std::make_unique<epoll_socket_state>(
729	244x	ctx.use_service<epoll_scheduler>()))
730		{
731	244x	}
732
733	488x	inline epoll_socket_service::~epoll_socket_service() {}
734
735		inline void
736	244x	epoll_socket_service::shutdown()
737		{
738	244x	std::lock_guard lock(state_->mutex_);
739
740	244x	while (auto* impl = state_->socket_list_.pop_front())
741	✗	impl->close_socket();
742
743		// Don't clear socket_ptrs_ here. The scheduler shuts down after us and
744		// drains completed_ops_, calling destroy() on each queued op. If we
745		// released our shared_ptrs now, an epoll_op::destroy() could free the
746		// last ref to an impl whose embedded descriptor_state is still linked
747		// in the queue — use-after-free on the next pop(). Letting ~state_
748		// release the ptrs (during service destruction, after scheduler
749		// shutdown) keeps every impl alive until all ops have been drained.
750	244x	}
751
752		inline io_object::implementation*
753	9492x	epoll_socket_service::construct()
754		{
755	9492x	auto impl = std::make_shared<epoll_socket>(*this);
756	9492x	auto* raw = impl.get();
757
758		{
759	9492x	std::lock_guard lock(state_->mutex_);
760	9492x	state_->socket_list_.push_back(raw);
761	9492x	state_->socket_ptrs_.emplace(raw, std::move(impl));
762	9492x	}
763
764	9492x	return raw;
765	9492x	}
766
767		inline void
768	9492x	epoll_socket_service::destroy(io_object::implementation* impl)
769		{
770	9492x	auto* epoll_impl = static_cast<epoll_socket*>(impl);
771	9492x	epoll_impl->close_socket();
772	9492x	std::lock_guard lock(state_->mutex_);
773	9492x	state_->socket_list_.remove(epoll_impl);
774	9492x	state_->socket_ptrs_.erase(epoll_impl);
775	9492x	}
776
777		inline std::error_code
778	3160x	epoll_socket_service::open_socket(
779		tcp_socket::implementation& impl, int family, int type, int protocol)
780		{
781	3160x	auto* epoll_impl = static_cast<epoll_socket*>(&impl);
782	3160x	epoll_impl->close_socket();
783
784	3160x	int fd = ::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
785	3160x	if (fd < 0)
786	✗	return make_err(errno);
787
788	3160x	if (family == AF_INET6)
789		{
790	5x	int one = 1;
791	5x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
792		}
793
794	3160x	epoll_impl->fd_ = fd;
795
796		// Register fd with epoll (edge-triggered mode)
797	3160x	epoll_impl->desc_state_.fd = fd;
798		{
799	3160x	std::lock_guard lock(epoll_impl->desc_state_.mutex);
800	3160x	epoll_impl->desc_state_.read_op = nullptr;
801	3160x	epoll_impl->desc_state_.write_op = nullptr;
802	3160x	epoll_impl->desc_state_.connect_op = nullptr;
803	3160x	}
804	3160x	scheduler().register_descriptor(fd, &epoll_impl->desc_state_);
805
806	3160x	return {};
807		}
808
809		inline void
810	15795x	epoll_socket_service::close(io_object::handle& h)
811		{
812	15795x	static_cast<epoll_socket*>(h.get())->close_socket();
813	15795x	}
814
815		inline void
816	23336x	epoll_socket_service::post(epoll_op* op)
817		{
818	23336x	state_->sched_.post(op);
819	23336x	}
820
821		inline void
822	3335x	epoll_socket_service::work_started() noexcept
823		{
824	3335x	state_->sched_.work_started();
825	3335x	}
826
827		inline void
828	186x	epoll_socket_service::work_finished() noexcept
829		{
830	186x	state_->sched_.work_finished();
831	186x	}
832
833		} // namespace boost::corosio::detail
834
835		#endif // BOOST_COROSIO_HAS_EPOLL
836
837		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SOCKET_SERVICE_HPP
838