Index/include / boost / corosio / native / detail / kqueue / kqueue_acceptor_service.hpp

include/boost/corosio/native/detail/kqueue/kqueue_acceptor_service.hpp

61.7% Lines (229/371) 100.0% Functions (21/21) 35.9% Branches (66/184)
include/boost/corosio/native/detail/kqueue/kqueue_acceptor_service.hpp
Prev Index Next
Line Branch TLA Hits Source Code
1 //
2 // Copyright (c) 2026 Michael Vandeberg
3 // Copyright (c) 2026 Steve Gerbino
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_KQUEUE_KQUEUE_ACCEPTOR_SERVICE_HPP
12 #define BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_ACCEPTOR_SERVICE_HPP
13
14 #include <boost/corosio/detail/platform.hpp>
15
16 #if BOOST_COROSIO_HAS_KQUEUE
17
18 #include <boost/corosio/detail/config.hpp>
19 #include <boost/capy/ex/execution_context.hpp>
20 #include <boost/corosio/detail/acceptor_service.hpp>
21
22 #include <boost/corosio/native/detail/kqueue/kqueue_acceptor.hpp>
23 #include <boost/corosio/native/detail/kqueue/kqueue_socket_service.hpp>
24 #include <boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp>
25
26 #include <boost/corosio/detail/endpoint_convert.hpp>
27 #include <boost/corosio/detail/dispatch_coro.hpp>
28 #include <boost/corosio/detail/make_err.hpp>
29
30 #include <memory>
31 #include <mutex>
32 #include <unordered_map>
33 #include <utility>
34
35 #include <errno.h>
36 #include <fcntl.h>
37 #include <netinet/in.h>
38 #include <sys/socket.h>
39 #include <unistd.h>
40
41 namespace boost::corosio::detail {
42
43 /** State for kqueue acceptor service. */
44 class kqueue_acceptor_state
45 {
46 friend class kqueue_acceptor_service;
47
48 public:
49 897 explicit kqueue_acceptor_state(kqueue_scheduler& sched) noexcept
50 299 : sched_(sched)
51 299 {
52 598 }
53
54 private:
55 kqueue_scheduler& sched_;
56 std::mutex mutex_;
57 intrusive_list<kqueue_acceptor> acceptor_list_;
58 std::unordered_map<kqueue_acceptor*, std::shared_ptr<kqueue_acceptor>>
59 acceptor_ptrs_;
60 };
61
62 /** kqueue acceptor service implementation.
63
64 Inherits from acceptor_service to enable runtime polymorphism.
65 Uses key_type = acceptor_service for service lookup.
66 */
67 class BOOST_COROSIO_DECL kqueue_acceptor_service final : public acceptor_service
68 {
69 public:
70 explicit kqueue_acceptor_service(capy::execution_context& ctx);
71 ~kqueue_acceptor_service();
72
73 kqueue_acceptor_service(kqueue_acceptor_service const&) = delete;
74 kqueue_acceptor_service& operator=(kqueue_acceptor_service const&) = delete;
75
76 void shutdown() override;
77 io_object::implementation* construct() override;
78 void destroy(io_object::implementation*) override;
79 void close(io_object::handle&) override;
80 std::error_code open_acceptor(
81 tcp_acceptor::implementation& impl, endpoint ep, int backlog) override;
82
83 6407 kqueue_scheduler& scheduler() const noexcept
84 {
85 6407 return state_->sched_;
86 }
87 void post(kqueue_op* op);
88 void work_started() noexcept;
89 void work_finished() noexcept;
90
91 /** Get the socket service for creating peer sockets during accept. */
92 kqueue_socket_service* socket_service() const noexcept;
93
94 private:
95 capy::execution_context& ctx_;
96 std::unique_ptr<kqueue_acceptor_state> state_;
97 };
98
99 inline void
100 6 kqueue_accept_op::cancel() noexcept
101 {
102
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (acceptor_impl_)
103 6 acceptor_impl_->cancel_single_op(*this);
104 else
105 ✗ request_cancel();
106 6 }
107
108 inline void
109 5445 kqueue_accept_op::operator()()
110 {
111 5445 stop_cb.reset();
112
113 5445 static_cast<kqueue_acceptor*>(acceptor_impl_)
114 5445 ->service()
115 5445 .scheduler()
116 5445 .reset_inline_budget();
117
118
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5445 times.
 5445 bool success = (errn == 0 && !cancelled.load(std::memory_order_acquire));
119
120
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5445 times.
 5445 if (ec_out)
121 {
122
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 5436 times.
 5445 if (cancelled.load(std::memory_order_acquire))
123 9 *ec_out = capy::error::canceled;
124
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5436 times.
 5436 else if (errn != 0)
125 ✗ *ec_out = make_err(errn);
126 else
127 5436 *ec_out = {};
128 5445 }
129
130
3/4
✓ Branch 0 taken 5436 times.
✓ Branch 1 taken 9 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 5436 times.
 5445 if (success && accepted_fd >= 0)
131 {
132
1/2
✓ Branch 0 taken 5436 times.
✗ Branch 1 not taken.
 5436 if (acceptor_impl_)
133 {
134 10872 auto* socket_svc = static_cast<kqueue_acceptor*>(acceptor_impl_)
135 5436 ->service()
136 5436 .socket_service();
137
1/2
✓ Branch 0 taken 5436 times.
✗ Branch 1 not taken.
 5436 if (socket_svc)
138 {
139 5436 auto& impl =
140 5436 static_cast<kqueue_socket&>(*socket_svc->construct());
141 5436 impl.set_socket(accepted_fd);
142
143 // Register accepted socket with kqueue (edge-triggered via EV_CLEAR)
144 5436 impl.desc_state_.fd = accepted_fd;
145 {
146 5436 std::lock_guard lock(impl.desc_state_.mutex);
147 5436 impl.desc_state_.read_op = nullptr;
148 5436 impl.desc_state_.write_op = nullptr;
149 5436 impl.desc_state_.connect_op = nullptr;
150 5436 }
151 10872 socket_svc->scheduler().register_descriptor(
152 5436 accepted_fd, &impl.desc_state_);
153
154 // Suppress SIGPIPE on the accepted socket; macOS lacks MSG_NOSIGNAL
155 5436 int one = 1;
156
2/4
✓ Branch 0 taken 5436 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5436 times.
✗ Branch 3 not taken.
 10872 if (::setsockopt(
157 5436 accepted_fd, SOL_SOCKET, SO_NOSIGPIPE, &one,
158 5436 sizeof(one)) == -1)
159 {
160 ✗ if (ec_out)
161 ✗ *ec_out = make_err(errno);
162 ✗ socket_svc->destroy(&impl);
163 ✗ accepted_fd = -1;
164 ✗ if (impl_out)
165 ✗ *impl_out = nullptr;
166 ✗ }
167 else
168 {
169 5436 sockaddr_in local_addr{};
170 5436 socklen_t local_len = sizeof(local_addr);
171 5436 sockaddr_in remote_addr{};
172 5436 socklen_t remote_len = sizeof(remote_addr);
173
174 5436 endpoint local_ep, remote_ep;
175
2/4
✗ Branch 0 not taken.
✓ Branch 1 taken 5436 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 5436 times.
 10872 if (::getsockname(
176 5436 accepted_fd,
177 reinterpret_cast<sockaddr*>(&local_addr),
178 5436 &local_len) == 0)
179 5436 local_ep = from_sockaddr_in(local_addr);
180
2/4
✗ Branch 0 not taken.
✓ Branch 1 taken 5436 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 5436 times.
 10872 if (::getpeername(
181 5436 accepted_fd,
182 reinterpret_cast<sockaddr*>(&remote_addr),
183 5436 &remote_len) == 0)
184 5436 remote_ep = from_sockaddr_in(remote_addr);
185
186 5436 impl.set_endpoints(local_ep, remote_ep);
187
188
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5436 times.
 5436 if (impl_out)
189 5436 *impl_out = &impl;
190
191 5436 accepted_fd = -1;
192 }
193 5436 }
194 else
195 {
196 ✗ if (ec_out && !*ec_out)
197 ✗ *ec_out = make_err(ENOENT);
198 ✗ ::close(accepted_fd);
199 ✗ accepted_fd = -1;
200 ✗ if (impl_out)
201 ✗ *impl_out = nullptr;
202 }
203 5436 }
204 else
205 {
206 ✗ ::close(accepted_fd);
207 ✗ accepted_fd = -1;
208 ✗ if (impl_out)
209 ✗ *impl_out = nullptr;
210 }
211 5436 }
212 else
213 {
214
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 9 if (accepted_fd >= 0)
215 {
216 ✗ ::close(accepted_fd);
217 ✗ accepted_fd = -1;
218 ✗ }
219
220
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 9 if (peer_impl)
221 {
222 ✗ auto* socket_svc_cleanup =
223 ✗ static_cast<kqueue_acceptor*>(acceptor_impl_)
224 ✗ ->service()
225 ✗ .socket_service();
226 ✗ if (socket_svc_cleanup)
227 ✗ socket_svc_cleanup->destroy(peer_impl);
228 ✗ peer_impl = nullptr;
229 ✗ }
230
231
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9 times.
 9 if (impl_out)
232 9 *impl_out = nullptr;
233 }
234
235 // Move to stack before resuming. See kqueue_op::operator()() for rationale.
236 5445 capy::executor_ref saved_ex(std::move(ex));
237 5445 std::coroutine_handle<> saved_h(std::move(h));
238 5445 auto prevent_premature_destruction = std::move(impl_ptr);
239
2/4
✓ Branch 0 taken 5445 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5445 times.
✗ Branch 3 not taken.
 5445 dispatch_coro(saved_ex, saved_h).resume();
240 5445 }
241
242 3864 inline kqueue_acceptor::kqueue_acceptor(kqueue_acceptor_service& svc) noexcept
243 966 : svc_(svc)
244 2898 {
245 966 }
246
247 inline std::coroutine_handle<>
248 5445 kqueue_acceptor::accept(
249 std::coroutine_handle<> h,
250 capy::executor_ref ex,
251 std::stop_token token,
252 std::error_code* ec,
253 io_object::implementation** impl_out)
254 {
255 5445 auto& op = acc_;
256 5445 op.reset();
257 5445 op.h = h;
258 5445 op.ex = ex;
259 5445 op.ec_out = ec;
260 5445 op.impl_out = impl_out;
261 5445 op.fd = fd_;
262
1/2
✓ Branch 0 taken 5445 times.
✗ Branch 1 not taken.
 5445 op.start(token, this);
263
264 5445 sockaddr_in addr{};
265 5445 socklen_t addrlen = sizeof(addr);
266
267 // FreeBSD: Can use accept4(fd_, addr, addrlen, SOCK_NONBLOCK | SOCK_CLOEXEC)
268 5445 int accepted = ::accept(fd_, reinterpret_cast<sockaddr*>(&addr), &addrlen);
269
270
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5445 times.
 5445 if (accepted >= 0)
271 {
272 // Set non-blocking and close-on-exec on the accepted socket
273 ✗ int flags = ::fcntl(accepted, F_GETFL, 0);
274 ✗ if (flags == -1 || ::fcntl(accepted, F_SETFL, flags | O_NONBLOCK) == -1)
275 {
276 ✗ int errn = errno;
277 ✗ ::close(accepted);
278 ✗ op.complete(errn, 0);
279 ✗ op.impl_ptr = shared_from_this();
280 ✗ svc_.post(&op);
281 ✗ return std::noop_coroutine();
282 }
283 ✗ if (::fcntl(accepted, F_SETFD, FD_CLOEXEC) == -1)
284 {
285 ✗ int errn = errno;
286 ✗ ::close(accepted);
287 ✗ op.complete(errn, 0);
288 ✗ op.impl_ptr = shared_from_this();
289 ✗ svc_.post(&op);
290 ✗ return std::noop_coroutine();
291 }
292
293 {
294 ✗ std::lock_guard lock(desc_state_.mutex);
295 ✗ desc_state_.read_ready = false;
296 ✗ }
297
298 ✗ if (svc_.scheduler().try_consume_inline_budget())
299 {
300 ✗ auto* socket_svc = svc_.socket_service();
301 ✗ if (socket_svc)
302 {
303 ✗ auto& impl = static_cast<kqueue_socket&>(*socket_svc->construct());
304 ✗ impl.set_socket(accepted);
305
306 ✗ impl.desc_state_.fd = accepted;
307 {
308 ✗ std::lock_guard lock(impl.desc_state_.mutex);
309 ✗ impl.desc_state_.read_op = nullptr;
310 ✗ impl.desc_state_.write_op = nullptr;
311 ✗ impl.desc_state_.connect_op = nullptr;
312 ✗ }
313 ✗ socket_svc->scheduler().register_descriptor(accepted, &impl.desc_state_);
314
315 // Suppress SIGPIPE on the accepted socket; macOS lacks MSG_NOSIGNAL
316 ✗ int one = 1;
317 ✗ if (::setsockopt(accepted, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one)) == -1)
318 {
319 ✗ int saved_errno = errno;
320 ✗ socket_svc->destroy(&impl);
321 ✗ if (ec)
322 ✗ *ec = make_err(saved_errno);
323 ✗ if (impl_out)
324 ✗ *impl_out = nullptr;
325 ✗ }
326 else
327 {
328 ✗ sockaddr_in local_addr{};
329 ✗ socklen_t local_len = sizeof(local_addr);
330 ✗ endpoint local_ep;
331 ✗ if (::getsockname(
332 ✗ accepted,
333 reinterpret_cast<sockaddr*>(&local_addr),
334 ✗ &local_len) == 0)
335 ✗ local_ep = from_sockaddr_in(local_addr);
336 ✗ impl.set_endpoints(local_ep, from_sockaddr_in(addr));
337 ✗ if (ec)
338 ✗ *ec = {};
339 ✗ if (impl_out)
340 ✗ *impl_out = &impl;
341 }
342 ✗ return dispatch_coro(ex, h);
343 }
344 else
345 {
346 ✗ ::close(accepted);
347 ✗ if (ec)
348 ✗ *ec = make_err(ENOENT);
349 ✗ if (impl_out)
350 ✗ *impl_out = nullptr;
351 ✗ return dispatch_coro(ex, h);
352 }
353 }
354
355 ✗ op.accepted_fd = accepted;
356 ✗ op.complete(0, 0);
357 ✗ op.impl_ptr = shared_from_this();
358 ✗ svc_.post(&op);
359 ✗ return std::noop_coroutine();
360 }
361
362
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 5445 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 5445 if (errno == EAGAIN || errno == EWOULDBLOCK)
363 {
364 5445 svc_.work_started();
365
1/2
✓ Branch 0 taken 5445 times.
✗ Branch 1 not taken.
 5445 op.impl_ptr = shared_from_this();
366
367 5445 bool perform_now = false;
368 {
369 5445 std::lock_guard lock(desc_state_.mutex);
370
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5445 times.
 5445 if (desc_state_.read_ready)
371 {
372 ✗ desc_state_.read_ready = false;
373 ✗ perform_now = true;
374 ✗ }
375 else
376 {
377 5445 desc_state_.read_op = &op;
378 }
379 5445 }
380
381
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5445 times.
 5445 if (perform_now)
382 {
383 ✗ for (;;)
384 {
385 ✗ op.perform_io();
386 ✗ if (op.errn != EAGAIN && op.errn != EWOULDBLOCK)
387 {
388 ✗ svc_.post(&op);
389 ✗ svc_.work_finished();
390 ✗ break;
391 }
392 ✗ op.errn = 0;
393 ✗ std::lock_guard lock(desc_state_.mutex);
394 ✗ if (desc_state_.read_ready)
395 {
396 ✗ desc_state_.read_ready = false;
397 ✗ continue;
398 }
399 ✗ desc_state_.read_op = &op;
400 ✗ break;
401 ✗ }
402 ✗ return std::noop_coroutine();
403 }
404
405
1/2
✓ Branch 0 taken 5445 times.
✗ Branch 1 not taken.
 5445 if (op.cancelled.load(std::memory_order_acquire))
406 {
407 ✗ kqueue_op* claimed = nullptr;
408 {
409 ✗ std::lock_guard lock(desc_state_.mutex);
410 ✗ if (desc_state_.read_op == &op)
411 ✗ claimed = std::exchange(desc_state_.read_op, nullptr);
412 ✗ }
413 ✗ if (claimed)
414 {
415 ✗ svc_.post(claimed);
416 ✗ svc_.work_finished();
417 ✗ }
418 ✗ }
419 5445 return std::noop_coroutine();
420 }
421
422 ✗ op.complete(errno, 0);
423 ✗ op.impl_ptr = shared_from_this();
424 ✗ svc_.post(&op);
425 ✗ return std::noop_coroutine();
426 5445 }
427
428 inline void
429 1 kqueue_acceptor::cancel() noexcept
430 {
431 1 auto self = weak_from_this().lock();
432
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1 if (!self)
433 ✗ return;
434
435 1 acc_.request_cancel();
436
437 1 kqueue_op* claimed = nullptr;
438 {
439
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1 std::lock_guard lock(desc_state_.mutex);
440
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 time.
 1 if (desc_state_.read_op == &acc_)
441 1 claimed = std::exchange(desc_state_.read_op, nullptr);
442 1 }
443
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1 if (claimed)
444 {
445 1 acc_.impl_ptr = self;
446
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1 svc_.post(&acc_);
447 1 svc_.work_finished();
448 1 }
449 1 }
450
451 inline void
452 6 kqueue_acceptor::cancel_single_op(kqueue_op& op) noexcept
453 {
454 6 auto self = weak_from_this().lock();
455
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (!self)
456 ✗ return;
457
458 6 op.request_cancel();
459
460 6 kqueue_op* claimed = nullptr;
461 {
462
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 std::lock_guard lock(desc_state_.mutex);
463
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (desc_state_.read_op == &op)
464 6 claimed = std::exchange(desc_state_.read_op, nullptr);
465 6 }
466
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 if (claimed)
467 {
468 6 op.impl_ptr = self;
469
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 svc_.post(&op);
470 6 svc_.work_finished();
471 6 }
472 6 }
473
474 inline void
475 3858 kqueue_acceptor::close_socket() noexcept
476 {
477 3858 auto self = weak_from_this().lock();
478
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3858 times.
 3858 if (self)
479 {
480 3858 acc_.request_cancel();
481
482 3858 kqueue_op* claimed = nullptr;
483 {
484
1/2
✓ Branch 0 taken 3858 times.
✗ Branch 1 not taken.
 3858 std::lock_guard lock(desc_state_.mutex);
485 3858 claimed = std::exchange(desc_state_.read_op, nullptr);
486 3858 desc_state_.read_ready = false;
487 3858 desc_state_.write_ready = false;
488 3858 }
489
490
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 3856 times.
 3858 if (claimed)
491 {
492 2 acc_.impl_ptr = self;
493
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 svc_.post(&acc_);
494 2 svc_.work_finished();
495 2 }
496
497
1/2
✓ Branch 0 taken 3858 times.
✗ Branch 1 not taken.
 3858 if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
498 ✗ desc_state_.impl_ref_ = self;
499 3858 }
500
501
2/2
✓ Branch 0 taken 962 times.
✓ Branch 1 taken 2896 times.
 3858 if (fd_ >= 0)
502 {
503
1/2
✓ Branch 0 taken 962 times.
✗ Branch 1 not taken.
 962 ::close(fd_);
504 962 fd_ = -1;
505 962 }
506
507 3858 desc_state_.fd = -1;
508 3858 desc_state_.registered_events = 0;
509
510 3858 local_endpoint_ = endpoint{};
511 3858 }
512
513 598 inline kqueue_acceptor_service::kqueue_acceptor_service(
514 capy::execution_context& ctx)
515 299 : ctx_(ctx)
516 299 , state_(
517
1/2
✓ Branch 0 taken 299 times.
✗ Branch 1 not taken.
 299 std::make_unique<kqueue_acceptor_state>(
518
1/2
✓ Branch 0 taken 299 times.
✗ Branch 1 not taken.
 299 ctx.use_service<kqueue_scheduler>()))
519 598 {
520 598 }
521
522 897 inline kqueue_acceptor_service::~kqueue_acceptor_service() = default;
523
524 inline void
525 299 kqueue_acceptor_service::shutdown()
526 {
527 299 std::lock_guard lock(state_->mutex_);
528
529
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 299 times.
 299 while (auto* impl = state_->acceptor_list_.pop_front())
530 ✗ impl->close_socket();
531 299 }
532
533 inline io_object::implementation*
534 966 kqueue_acceptor_service::construct()
535 {
536 966 auto impl = std::make_shared<kqueue_acceptor>(*this);
537 966 auto* raw = impl.get();
538
539
1/2
✓ Branch 0 taken 966 times.
✗ Branch 1 not taken.
 966 std::lock_guard lock(state_->mutex_);
540 966 state_->acceptor_list_.push_back(raw);
541
1/2
✓ Branch 0 taken 966 times.
✗ Branch 1 not taken.
 966 state_->acceptor_ptrs_.emplace(raw, std::move(impl));
542
543 966 return raw;
544 966 }
545
546 inline void
547 966 kqueue_acceptor_service::destroy(io_object::implementation* impl)
548 {
549 966 auto* kq_impl = static_cast<kqueue_acceptor*>(impl);
550 966 kq_impl->close_socket();
551 966 std::lock_guard lock(state_->mutex_);
552 966 state_->acceptor_list_.remove(kq_impl);
553
1/2
✓ Branch 0 taken 966 times.
✗ Branch 1 not taken.
 966 state_->acceptor_ptrs_.erase(kq_impl);
554 966 }
555
556 inline void
557 1928 kqueue_acceptor_service::close(io_object::handle& h)
558 {
559 1928 static_cast<kqueue_acceptor*>(h.get())->close_socket();
560 1928 }
561
562 inline std::error_code
563 964 kqueue_acceptor_service::open_acceptor(
564 tcp_acceptor::implementation& impl, endpoint ep, int backlog)
565 {
566 964 auto* kq_impl = static_cast<kqueue_acceptor*>(&impl);
567 964 kq_impl->close_socket();
568
569 964 int fd = ::socket(AF_INET, SOCK_STREAM, 0);
570
1/2
✓ Branch 0 taken 964 times.
✗ Branch 1 not taken.
 964 if (fd < 0)
571 ✗ return make_err(errno);
572
573 964 int flags = ::fcntl(fd, F_GETFL, 0);
574
1/2
✓ Branch 0 taken 964 times.
✗ Branch 1 not taken.
 964 if (flags == -1)
575 {
576 ✗ int errn = errno;
577 ✗ ::close(fd);
578 ✗ return make_err(errn);
579 }
580
1/2
✓ Branch 0 taken 964 times.
✗ Branch 1 not taken.
 964 if (::fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1)
581 {
582 ✗ int errn = errno;
583 ✗ ::close(fd);
584 ✗ return make_err(errn);
585 }
586
1/2
✓ Branch 0 taken 964 times.
✗ Branch 1 not taken.
 964 if (::fcntl(fd, F_SETFD, FD_CLOEXEC) == -1)
587 {
588 ✗ int errn = errno;
589 ✗ ::close(fd);
590 ✗ return make_err(errn);
591 }
592
593 964 int reuse = 1;
594 964 (void)::setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
595
596 964 sockaddr_in addr = detail::to_sockaddr_in(ep);
597
2/2
✓ Branch 0 taken 962 times.
✓ Branch 1 taken 2 times.
 964 if (::bind(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) < 0)
598 {
599 2 int errn = errno;
600 2 ::close(fd);
601 2 return make_err(errn);
602 }
603
604
1/2
✓ Branch 0 taken 962 times.
✗ Branch 1 not taken.
 962 if (::listen(fd, backlog) < 0)
605 {
606 ✗ int errn = errno;
607 ✗ ::close(fd);
608 ✗ return make_err(errn);
609 }
610
611 962 kq_impl->fd_ = fd;
612
613 962 kq_impl->desc_state_.fd = fd;
614 {
615 962 std::lock_guard lock(kq_impl->desc_state_.mutex);
616 962 kq_impl->desc_state_.read_op = nullptr;
617 962 }
618 962 scheduler().register_descriptor(fd, &kq_impl->desc_state_);
619
620 962 sockaddr_in local_addr{};
621 962 socklen_t local_len = sizeof(local_addr);
622
2/4
✗ Branch 0 not taken.
✓ Branch 1 taken 962 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 962 times.
 1924 if (::getsockname(
623 1924 fd, reinterpret_cast<sockaddr*>(&local_addr), &local_len) == 0)
624 962 kq_impl->set_local_endpoint(detail::from_sockaddr_in(local_addr));
625
626 962 return {};
627 964 }
628
629 inline void
630 9 kqueue_acceptor_service::post(kqueue_op* op)
631 {
632 9 state_->sched_.post(op);
633 9 }
634
635 inline void
636 5445 kqueue_acceptor_service::work_started() noexcept
637 {
638 5445 state_->sched_.work_started();
639 5445 }
640
641 inline void
642 9 kqueue_acceptor_service::work_finished() noexcept
643 {
644 9 state_->sched_.work_finished();
645 9 }
646
647 inline kqueue_socket_service*
648 5436 kqueue_acceptor_service::socket_service() const noexcept
649 {
650 5436 auto* svc = ctx_.find_service<detail::socket_service>();
651
2/4
✓ Branch 0 taken 5436 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5436 times.
✗ Branch 3 not taken.
 5436 return svc ? dynamic_cast<kqueue_socket_service*>(svc) : nullptr;
652 }
653
654 } // namespace boost::corosio::detail
655
656 #endif // BOOST_COROSIO_HAS_KQUEUE
657
658 #endif // BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_ACCEPTOR_SERVICE_HPP
659