Coverage

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

@@ -135,7 +135,7 @@ class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
 
 private:
     void
-    run_task(lock_type& lock, context_type* ctx,
+    run_task(lock_type& lock, context_type& ctx,
         long timeout_us) override;
     void interrupt_reactor() const override;
     void update_timerfd() const;
@@ -347,7 +347,7 @@ epoll_scheduler::update_timerfd() const
 
 inline void
 epoll_scheduler::run_task(
-    lock_type& lock, context_type* ctx, long timeout_us)
+    lock_type& lock, context_type& ctx, long timeout_us)
 {
     int timeout_ms;
     if (task_interrupted_)

include/boost/corosio/native/detail/io_uring/io_uring_acceptor_ops.hpp

@@ -88,8 +88,8 @@ struct uring_multi_accept_op : io_uring_op
     static void do_retired_cqe(
         io_uring_op* /*base*/, int res, unsigned /*flags*/) noexcept
     {
-        if (res >= 0)
-            ::close(res);
+        if (res >= 0)           // LCOV_EXCL_LINE adopt-over-armed race leak guard
+            ::close(res);       // LCOV_EXCL_LINE adopt-over-armed race leak guard
     }
 
     static void do_cqe(io_uring_op* base, int res, unsigned flags,
@@ -105,15 +105,15 @@ struct uring_multi_accept_op : io_uring_op
         // whether to surface the fd via a waiter or park it.
     }
 
-    /// Never invoked: the multishot op is owned by the acceptor and
-    /// never queued for handler dispatch. Provided so the vtable is
-    /// complete.
+    // LCOV_EXCL_START: never invoked; the multishot op is owned by
+    // the acceptor and never queued for handler dispatch. Provided so
+    // the vtable is complete.
     static void do_handler(
         void* /*owner*/, scheduler_op* /*base*/,
         std::uint32_t /*bytes*/, std::uint32_t /*error*/) noexcept
     {
-        // No-op. The acceptor's per-accept callback handles everything.
     }
+    // LCOV_EXCL_STOP
 };
 
 /** Synthesized accept op — manufactured by the acceptor for parked fds.
@@ -153,11 +153,13 @@ struct uring_accept_op : io_uring_op
         : io_uring_op(&do_handler, &do_cqe)
     {}
 
+    // LCOV_EXCL_START: never receives a CQE; present for vtable
+    // completeness.
     static void do_cqe(io_uring_op*, int, unsigned,
                        ready_queue&) noexcept
     {
-        // Unreachable: this op never receives a CQE.
     }
+    // LCOV_EXCL_STOP
 
     static void do_handler(
         void* owner, scheduler_op* base,
@@ -193,9 +195,12 @@ struct uring_accept_op : io_uring_op
                 self->peer_service, self->accepted_fd,
                 self->peer_storage, self->peer_len);
 
+        // LCOV_EXCL_START: no public accept overload reports the peer
+        // endpoint on this backend yet.
         if (self->peer_endpoint_out)
             *self->peer_endpoint_out =
                 sockaddr_to_endpoint(self->peer_storage);
+        // LCOV_EXCL_STOP
 
         if (self->ec_out)
             *self->ec_out = {};

include/boost/corosio/native/detail/io_uring/io_uring_file_ops.hpp

@@ -180,8 +180,10 @@ struct uring_random_access_read_op : uring_file_read_op_base
 
     Stream files pass `offset == -1` (kernel f_pos); random-access
     files pass an explicit caller-supplied offset. Unlike socket
-    writes, no `MSG_NOSIGNAL` is needed — files don't generate
-    SIGPIPE on closed peers.
+    writes there is no `MSG_NOSIGNAL` equivalent: a write to a pipe
+    or FIFO whose reader has closed raises SIGPIPE when the kernel
+    executes it, so the teardown paths that flush queued writes hold
+    the signal blocked (see `scoped_sigpipe_block`).
 */
 /// Shared state and submission logic for file write ops. Concrete
 /// subclasses pick a `do_handler` matching their storage model.

include/boost/corosio/native/detail/io_uring/io_uring_random_access_file.hpp

@@ -207,6 +207,10 @@ class BOOST_COROSIO_DECL io_uring_random_access_file final
     {
         if (fd_ >= 0)
         {
+            // The kernel may run a queued pipe write as task work at
+            // either kernel entry below; with the reader already gone
+            // that raises SIGPIPE.
+            scoped_sigpipe_block no_sigpipe;
             sched_->cancel_and_flush(fd_);
             ::close(fd_);
             fd_ = -1;

include/boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp

@@ -46,11 +46,63 @@
 
 #include <errno.h>
 #include <poll.h>
+#include <pthread.h>
+#include <signal.h>
 #include <sys/eventfd.h>
+#include <time.h>
 #include <unistd.h>
 
 namespace boost::corosio::detail {
 
+/** Block SIGPIPE on the calling thread for the current scope.
+
+    A queued pipe write can execute inline while this thread is in
+    the kernel submitting SQEs; if the pipe's reader is already gone
+    the kernel raises a thread-directed SIGPIPE, which kills any
+    process that has not ignored the signal. The write's CQE still
+    reports `EPIPE`, so the signal carries no information the
+    completion path does not already deliver. The destructor consumes
+    any SIGPIPE raised while blocked and restores the caller's mask.
+*/
+class scoped_sigpipe_block
+{
+    sigset_t old_{};
+    bool restore_ = false;
+
+public:
+    /// Consume any SIGPIPE raised in scope and restore the mask.
+    ~scoped_sigpipe_block()
+    {
+        if (!restore_)
+            return;
+        if (!sigismember(&old_, SIGPIPE))
+        {
+            // Only consume what this scope could have generated; a
+            // caller who blocked SIGPIPE keeps their pending state.
+            sigset_t set;
+            sigemptyset(&set);
+            sigaddset(&set, SIGPIPE);
+            timespec zero{};
+            while (::sigtimedwait(&set, nullptr, &zero) == SIGPIPE)
+            {
+            }
+        }
+        ::pthread_sigmask(SIG_SETMASK, &old_, nullptr);
+    }
+
+    /// Construct and block SIGPIPE for the calling thread.
+    scoped_sigpipe_block() noexcept
+    {
+        sigset_t set;
+        sigemptyset(&set);
+        sigaddset(&set, SIGPIPE);
+        restore_ = ::pthread_sigmask(SIG_BLOCK, &set, &old_) == 0;
+    }
+
+    scoped_sigpipe_block(scoped_sigpipe_block const&)            = delete;
+    scoped_sigpipe_block& operator=(scoped_sigpipe_block const&) = delete;
+};
+
 // Forward-declared so the out-of-line inline definitions below the class
 // can reference the frame stack without a circular dependency.
 struct io_uring_scheduler_frame;
@@ -556,6 +608,9 @@ io_uring_scheduler::~io_uring_scheduler()
 {
     if (ring_inited_)
     {
+        // Ring teardown can still run a doomed pipe write as task
+        // work; absorb the SIGPIPE it would raise.
+        scoped_sigpipe_block no_sigpipe;
         if (wakeup_eventfd_ >= 0)
             ::close(wakeup_eventfd_);
         ::io_uring_queue_exit(&ring_);
@@ -1493,6 +1548,12 @@ io_uring_op::on_cancel() noexcept
 inline void
 io_uring_scheduler::cancel_and_flush(int fd) noexcept
 {
+    // The flush can execute a queued write on `fd` inline; when the
+    // fd is a pipe whose reader has already closed — service
+    // shutdown closes impls one at a time, so teardown itself
+    // creates that state — the kernel raises SIGPIPE.
+    scoped_sigpipe_block no_sigpipe;
+
     lazy_init_ring();
     interrupt_reactor();
     lock_type lock(ring_mutex_);

include/boost/corosio/native/detail/io_uring/io_uring_stream_file.hpp

@@ -229,6 +229,10 @@ class BOOST_COROSIO_DECL io_uring_stream_file final
     {
         if (fd_ >= 0)
         {
+            // The kernel may run a queued pipe write as task work at
+            // either kernel entry below; with the reader already gone
+            // that raises SIGPIPE.
+            scoped_sigpipe_block no_sigpipe;
             sched_->cancel_and_flush(fd_);
             ::close(fd_);
             fd_ = -1;

include/boost/corosio/native/detail/io_uring/io_uring_types.hpp

@@ -2594,6 +2594,9 @@ class BOOST_COROSIO_DECL io_uring_local_datagram_socket final
         return remote_endpoint_;
     }
 
+    // LCOV_EXCL_START: the public bind routes through the
+    // service's bind_socket; nothing calls the implementation
+    // interface's bind on this backend.
     std::error_code bind(corosio::local_endpoint ep) noexcept override
     {
         sockaddr_storage addr{};
@@ -2609,6 +2612,7 @@ class BOOST_COROSIO_DECL io_uring_local_datagram_socket final
             local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
         return {};
     }
+    // LCOV_EXCL_STOP
 
 private:
     std::coroutine_handle<> submit_send(

include/boost/corosio/native/detail/iocp/win_scheduler.hpp

@@ -96,7 +96,10 @@ class BOOST_COROSIO_DECL win_scheduler final
     /// Return true when scheduler locking is disabled (fully-lockless tier).
     bool scheduler_locking_disabled() const noexcept override
     {
+        // LCOV_EXCL_START: consulted only by the POSIX pool-backed
+        // services; the IOCP services do not read it yet.
         return scheduler_locking_disabled_;
+        // LCOV_EXCL_STOP
     }
 
     /** Signal that an overlapped I/O operation is now pending.
@@ -618,8 +621,8 @@ win_scheduler::do_one(unsigned long timeout_ms)
                 return 1;
             }
 
-            default:
-                continue;
+            default:            // LCOV_EXCL_LINE unreachable: closed key set
+                continue;       // LCOV_EXCL_LINE unreachable: closed key set
             }
         }
 
@@ -646,8 +649,10 @@ win_scheduler::do_one(unsigned long timeout_ms)
                 }
                 continue;
 
-            default:
-                continue;
+            // A key outside the closed set reaches here only if a
+            // third party posts to the port.
+            default:            // LCOV_EXCL_LINE unreachable: closed key set
+                continue;       // LCOV_EXCL_LINE unreachable: closed key set
             }
         }
 

include/boost/corosio/native/detail/iocp/win_signals.hpp

@@ -704,8 +704,8 @@ win_signals::start_wait(win_signal& impl, signal_op* op)
 inline void
 win_signals::deliver_signal(int signal_number)
 {
-    if (signal_number < 0 || signal_number >= max_signal_number)
-        return;
+    if (signal_number < 0 || signal_number >= max_signal_number) // LCOV_EXCL_LINE OS never delivers out-of-range
+        return;                                                  // LCOV_EXCL_LINE OS never delivers out-of-range
 
     signal_detail::signal_state* state = signal_detail::get_signal_state();
     std::lock_guard<std::mutex> lock(state->mutex);

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

@@ -156,7 +156,7 @@ class BOOST_COROSIO_DECL kqueue_scheduler final : public reactor_scheduler
 
 private:
     void
-    run_task(lock_type& lock, context_type* ctx,
+    run_task(lock_type& lock, context_type& ctx,
         long timeout_us) override;
     void interrupt_reactor() const override;
     long calculate_timeout(long requested_timeout_us) const;
@@ -335,7 +335,7 @@ kqueue_scheduler::calculate_timeout(long requested_timeout_us) const
 
 inline void
 kqueue_scheduler::run_task(
-    lock_type& lock, context_type* ctx, long timeout_us)
+    lock_type& lock, context_type& ctx, long timeout_us)
 {
     long effective_timeout_us =
         task_interrupted_ ? 0 : calculate_timeout(timeout_us);

include/boost/corosio/native/detail/reactor/reactor_basic_socket.hpp

@@ -117,16 +117,14 @@ class reactor_basic_socket
 
     /** Register an op with the reactor.
 
-        Handles cached edge events and deferred cancellation.
-        Called on the EAGAIN/EINPROGRESS path when speculative
-        I/O failed.
+        Handles cached edge events. Called on the EAGAIN/EINPROGRESS
+        path when speculative I/O failed.
     */
     template<class Op>
     void register_op(
         Op& op,
         reactor_op_base*& desc_slot,
         bool& ready_flag,
-        bool& cancel_flag,
         bool is_write_direction = false) noexcept;
 
     /** Cancel a single pending operation.
@@ -135,7 +133,6 @@ class reactor_basic_socket
         the mutex and posts it to the scheduler as cancelled.
         Derived must implement:
           op_to_desc_slot(Op&) -> reactor_op_base**
-          op_to_cancel_flag(Op&) -> bool*
     */
     template<class Op>
     void cancel_single_op(Op& op) noexcept;
@@ -175,7 +172,6 @@ reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_
     Op& op,
     reactor_op_base*& desc_slot,
     bool& ready_flag,
-    bool& cancel_flag,
     bool is_write_direction) noexcept
 {
     svc_.work_started();
@@ -191,11 +187,6 @@ reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_
             op.errn = 0;
     }
 
-    if (cancel_flag)
-    {
-        cancel_flag = false;
-        op.cancelled.store(true, std::memory_order_relaxed);
-    }
 
     if (io_done || op.cancelled.load(std::memory_order_acquire))
     {
@@ -332,12 +323,6 @@ reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
                 });
             desc_state_.read_ready             = false;
             desc_state_.write_ready            = false;
-            desc_state_.read_cancel_pending       = false;
-            desc_state_.write_cancel_pending      = false;
-            desc_state_.connect_cancel_pending    = false;
-            desc_state_.wait_read_cancel_pending  = false;
-            desc_state_.wait_write_cancel_pending = false;
-            desc_state_.wait_error_cancel_pending = false;
 
             if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
                 desc_state_.impl_ref_ = self;
@@ -398,12 +383,6 @@ reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
                 });
             desc_state_.read_ready             = false;
             desc_state_.write_ready            = false;
-            desc_state_.read_cancel_pending       = false;
-            desc_state_.write_cancel_pending      = false;
-            desc_state_.connect_cancel_pending    = false;
-            desc_state_.wait_read_cancel_pending  = false;
-            desc_state_.wait_write_cancel_pending = false;
-            desc_state_.wait_error_cancel_pending = false;
 
             if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
                 desc_state_.impl_ref_ = self;