kqueue adaptive pump

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

@@ -159,9 +159,8 @@ kqueue_accept_op::operator()()
                 {
                     if (ec_out)
                         *ec_out = make_err(errno);
-                    ::close(accepted_fd);
-                    accepted_fd = -1;
                     socket_svc->destroy(&impl);
+                    accepted_fd = -1;
                     if (impl_out)
                         *impl_out = nullptr;
                 }
@@ -295,6 +294,64 @@ kqueue_acceptor::accept(
             std::lock_guard lock(desc_state_.mutex);
             desc_state_.read_ready = false;
         }
+
+        if (svc_.scheduler().try_consume_inline_budget())
+        {
+            auto* socket_svc = svc_.socket_service();
+            if (socket_svc)
+            {
+                auto& impl = static_cast<kqueue_socket&>(*socket_svc->construct());
+                impl.set_socket(accepted);
+
+                impl.desc_state_.fd = accepted;
+                {
+                    std::lock_guard lock(impl.desc_state_.mutex);
+                    impl.desc_state_.read_op = nullptr;
+                    impl.desc_state_.write_op = nullptr;
+                    impl.desc_state_.connect_op = nullptr;
+                }
+                socket_svc->scheduler().register_descriptor(accepted, &impl.desc_state_);
+
+                // Suppress SIGPIPE on the accepted socket; macOS lacks MSG_NOSIGNAL
+                int one = 1;
+                if (::setsockopt(accepted, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one)) == -1)
+                {
+                    int saved_errno = errno;
+                    socket_svc->destroy(&impl);
+                    if (ec)
+                        *ec = make_err(saved_errno);
+                    if (impl_out)
+                        *impl_out = nullptr;
+                }
+                else
+                {
+                    sockaddr_in local_addr{};
+                    socklen_t local_len = sizeof(local_addr);
+                    endpoint local_ep;
+                    if (::getsockname(
+                            accepted,
+                            reinterpret_cast<sockaddr*>(&local_addr),
+                            &local_len) == 0)
+                        local_ep = from_sockaddr_in(local_addr);
+                    impl.set_endpoints(local_ep, from_sockaddr_in(addr));
+                    if (ec)
+                        *ec = {};
+                    if (impl_out)
+                        *impl_out = &impl;
+                }
+                return dispatch_coro(ex, h);
+            }
+            else
+            {
+                ::close(accepted);
+                if (ec)
+                    *ec = make_err(ENOENT);
+                if (impl_out)
+                    *impl_out = nullptr;
+                return dispatch_coro(ex, h);
+            }
+        }
+
         op.accepted_fd = accepted;
         op.complete(0, 0);
         op.impl_ptr = shared_from_this();

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

@@ -173,6 +173,18 @@ class BOOST_COROSIO_DECL kqueue_scheduler final
 
         Called at the start of each posted completion handler to
         grant a fresh budget for speculative inline completions.
+        Operates in two modes depending on whether another thread
+        absorbed queued work from the previous dispatch cycle:
+
+        - **Adaptive** (default): the effective cap ramps up when
+          the previous cycle fully consumed its budget (doubles up
+          to 16) and ramps down to the floor (2) when budget was
+          only partially consumed, tracking actual inline demand.
+        - **Unassisted**: entered when no other thread was available
+          to signal (unlock_and_signal_one returned false). Applies
+          a fixed conservative cap (4) to amortize scheduling
+          overhead for small buffers while avoiding bursty I/O that
+          fills socket buffers and stalls large transfers.
     */
     void reset_inline_budget() const noexcept;
 
@@ -308,8 +320,11 @@ class BOOST_COROSIO_DECL kqueue_scheduler final
         Mutex must be held.
 
         @param lock The held mutex lock.
+
+        @return `true` if at least one waiter was signaled,
+        `false` if no waiters existed.
     */
-    void unlock_and_signal_one(std::unique_lock<std::mutex>& lock) const;
+    bool unlock_and_signal_one(std::unique_lock<std::mutex>& lock) const;
 
     /** Clear the signaled state before waiting.
 
@@ -343,7 +358,6 @@ class BOOST_COROSIO_DECL kqueue_scheduler final
         std::unique_lock<std::mutex>& lock, long timeout_us) const;
 
     int kq_fd_;
-    int max_inline_budget_ = 2;
     mutable std::mutex mutex_;
     mutable std::condition_variable cond_;
     mutable op_queue completed_ops_;
@@ -391,12 +405,16 @@ struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context
     op_queue private_queue;
     std::int64_t private_outstanding_work;
     int inline_budget;
+    int inline_budget_max;
+    bool unassisted;
 
     scheduler_context(kqueue_scheduler const* k, scheduler_context* n)
         : key(k)
         , next(n)
         , private_outstanding_work(0)
         , inline_budget(0)
+        , inline_budget_max(2)
+        , unassisted(false)
     {
     }
 };
@@ -468,7 +486,24 @@ inline void
 kqueue_scheduler::reset_inline_budget() const noexcept
 {
     if (auto* ctx = kqueue::find_context(this))
-        ctx->inline_budget = max_inline_budget_;
+    {
+        // Cap when no other thread absorbed queued work. A moderate
+        // cap (4) amortizes scheduling for small buffers while avoiding
+        // bursty I/O that fills socket buffers and stalls large transfers.
+        if (ctx->unassisted)
+        {
+            ctx->inline_budget_max = 4;
+            ctx->inline_budget = 4;
+            return;
+        }
+        // Ramp up when previous cycle fully consumed budget.
+        // Reset on partial consumption (EAGAIN hit or peer got scheduled).
+        if (ctx->inline_budget == 0)
+            ctx->inline_budget_max = (std::min)(ctx->inline_budget_max * 2, 16);
+        else if (ctx->inline_budget < ctx->inline_budget_max)
+            ctx->inline_budget_max = 2;
+        ctx->inline_budget = ctx->inline_budget_max;
+    }
 }
 
 inline bool
@@ -1073,7 +1108,7 @@ kqueue_scheduler::maybe_unlock_and_signal_one(
     return false;
 }
 
-inline void
+inline bool
 kqueue_scheduler::unlock_and_signal_one(
     std::unique_lock<std::mutex>& lock) const
 {
@@ -1082,6 +1117,7 @@ kqueue_scheduler::unlock_and_signal_one(
     lock.unlock();
     if (have_waiters)
         cond_.notify_one();
+    return have_waiters;
 }
 
 inline void
@@ -1378,10 +1414,15 @@ kqueue_scheduler::do_one(
         // Handle operation
         if (op != nullptr)
         {
-            if (!completed_ops_.empty())
-                unlock_and_signal_one(lock);
+            bool more = !completed_ops_.empty();
+
+            if (more)
+                ctx->unassisted = !unlock_and_signal_one(lock);
             else
+            {
+                ctx->unassisted = false;
                 lock.unlock();
+            }
 
             work_cleanup on_exit{this, &lock, ctx};
             (void)on_exit;

perf/bench/corosio/socket_throughput_bench.cpp

@@ -88,7 +88,6 @@ bench_throughput(std::size_t chunk_size, double duration_s)
             total_written += n;
         }
         writer.shutdown( corosio::tcp_socket::shutdown_send );
-        writer.close();
     };
 
     auto read_task = [&]() -> capy::task<> {
@@ -125,6 +124,9 @@ bench_throughput(std::size_t chunk_size, double duration_s)
     std::cout << "    Throughput: " << perf::format_throughput(throughput)
               << "\n\n";
 
+    writer.close();
+    reader.close();
+
     return bench::benchmark_result("throughput_" + std::to_string(chunk_size))
         .add("chunk_size", static_cast<double>(chunk_size))
         .add("bytes_written", static_cast<double>(total_written))