Fix epoll async cancellation, TLS test corruption, and SIGPIPE handling

This commit addresses several correctness and portability issues in the
epoll backend and TLS implementations.

epoll completion/cancellation race:
- Add atomic `registered` flag to epoll_op as a claim token
- Scheduler and cancel() now race to atomically exchange the flag
- Whoever wins owns the completion, preventing double-completion bugs
- cancel() properly unregisters fd from epoll and posts to completion queue
- close_socket() calls cancel() first to ensure pending ops complete

SIGPIPE handling:
- Replace writev() with sendmsg() using MSG_NOSIGNAL flag
- Matches Boost.Asio's per-syscall approach instead of global signal handler
- Avoids process-wide side effects from signal(SIGPIPE, SIG_IGN)

Empty buffer handling:
- Extend checks to handle single zero-length buffers, not just empty iovec

WolfSSL deferred context initialization:
- Cache separate client_ctx_ and server_ctx_ in wolfssl_native_context
- Defer SSL object creation until handshake when role is known
- Works with standard WolfSSL builds without --enable-opensslextra
- Single tls::context can now be shared across client and server streams

TLS test lambda capture fix:
- Store coroutine lambdas in named variables before invocation
- Fixes dangling reference when temporary lambda is destroyed before
  coroutine completes, which caused client/server to share captures

Add high-level overview comments to epoll implementation files explaining
architecture, completion/cancellation race handling, and design rationale.

Author: sgerbino

include/boost/corosio/tls/openssl_stream.hpp

@@ -57,6 +57,12 @@ class BOOST_COROSIO_DECL openssl_stream : public tls_stream
         @param ctx The TLS context containing configuration.
     */
     openssl_stream( io_stream& stream, tls::context ctx );
+
+    /** Destructor.
+
+        Releases the underlying OpenSSL resources.
+    */
+    ~openssl_stream();
 };
 
 } // namespace corosio

include/boost/corosio/tls/wolfssl_stream.hpp

@@ -59,6 +59,12 @@ class BOOST_COROSIO_DECL
         @param ctx The TLS context containing configuration.
     */
     wolfssl_stream(io_stream& stream, tls::context ctx);
+
+    /** Destructor.
+
+        Releases the underlying WolfSSL resources.
+    */
+    ~wolfssl_stream();
 };
 
 } // namespace corosio

src/corosio/src/detail/epoll/op.hpp

@@ -31,23 +31,57 @@
 #include <atomic>
 #include <cstddef>
 #include <cstdint>
+#include <memory>
 #include <optional>
 #include <stop_token>
 
 #include <netinet/in.h>
 #include <sys/socket.h>
 #include <sys/uio.h>
 
+/*
+    epoll Operation State
+    =====================
+
+    Each async I/O operation has a corresponding epoll_op-derived struct that
+    holds the operation's state while it's in flight. The socket impl owns
+    fixed slots for each operation type (conn_, rd_, wr_), so only one
+    operation of each type can be pending per socket at a time.
+
+    Completion vs Cancellation Race
+    -------------------------------
+    The `registered` atomic handles the race between epoll signaling ready
+    and cancel() being called. Whoever atomically exchanges it from true to
+    false "claims" the operation and is responsible for completing it. The
+    loser sees false and does nothing. This avoids double-completion bugs
+    without requiring a mutex in the hot path.
+
+    Impl Lifetime Management
+    ------------------------
+    When cancel() posts an op to the scheduler's ready queue, the socket impl
+    might be destroyed before the scheduler processes the op. The `impl_ptr`
+    member holds a shared_ptr to the impl, keeping it alive until the op
+    completes. This is set by cancel() in sockets.hpp and cleared in operator()
+    after the coroutine is resumed. Without this, closing a socket with pending
+    operations causes use-after-free.
+
+    EOF Detection
+    -------------
+    For reads, 0 bytes with no error means EOF. But an empty user buffer also
+    returns 0 bytes. The `empty_buffer_read` flag distinguishes these cases
+    so we don't spuriously report EOF when the user just passed an empty buffer.
+
+    SIGPIPE Prevention
+    ------------------
+    Writes use sendmsg() with MSG_NOSIGNAL instead of writev() to prevent
+    SIGPIPE when the peer has closed. This is the same approach Boost.Asio
+    uses on Linux.
+*/
+
 namespace boost {
 namespace corosio {
 namespace detail {
 
-/** Base class for epoll async operations.
-
-    This class is analogous to overlapped_op on Windows.
-    It stores the coroutine handle, executor, and result
-    pointers needed to complete an async operation.
-*/
 struct epoll_op : scheduler_op
 {
     struct canceller
@@ -67,8 +101,13 @@ struct epoll_op : scheduler_op
     std::size_t bytes_transferred = 0;
 
     std::atomic<bool> cancelled{false};
+    std::atomic<bool> registered{false};
     std::optional<std::stop_callback<canceller>> stop_cb;
 
+    // Prevents use-after-free when socket is closed with pending ops.
+    // See "Impl Lifetime Management" in file header.
+    std::shared_ptr<void> impl_ptr;
+
     epoll_op()
     {
         data_ = this;
@@ -81,6 +120,8 @@ struct epoll_op : scheduler_op
         errn = 0;
         bytes_transferred = 0;
         cancelled.store(false, std::memory_order_relaxed);
+        registered.store(false, std::memory_order_relaxed);
+        impl_ptr.reset();
     }
 
     void operator()() override
@@ -94,23 +135,24 @@ struct epoll_op : scheduler_op
             else if (errn != 0)
                 *ec_out = make_err(errn);
             else if (is_read_operation() && bytes_transferred == 0)
-            {
-                // EOF: 0 bytes transferred with no error indicates end of stream
                 *ec_out = capy::error::eof;
-            }
         }
 
         if (bytes_out)
             *bytes_out = bytes_transferred;
 
-        d.dispatch(h).resume();
+        auto saved_d = d;
+        auto saved_h = std::move(h);
+        impl_ptr.reset();
+        saved_d.dispatch(saved_h).resume();
     }
 
     virtual bool is_read_operation() const noexcept { return false; }
 
     void destroy() override
     {
         stop_cb.reset();
+        impl_ptr.reset();
     }
 
     void request_cancel() noexcept
@@ -133,10 +175,6 @@ struct epoll_op : scheduler_op
         bytes_transferred = bytes;
     }
 
-    /** Called when epoll signals the fd is ready.
-        Derived classes override this to perform the actual I/O.
-        Sets error and bytes_transferred appropriately.
-    */
     virtual void perform_io() noexcept {}
 };
 
@@ -148,12 +186,11 @@ get_epoll_op(scheduler_op* h) noexcept
 
 //------------------------------------------------------------------------------
 
-/** Connect operation state. */
 struct epoll_connect_op : epoll_op
 {
     void perform_io() noexcept override
     {
-        // For connect, check SO_ERROR to see if connection succeeded
+        // connect() completion status is retrieved via SO_ERROR, not return value
         int err = 0;
         socklen_t len = sizeof(err);
         if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
@@ -164,17 +201,13 @@ struct epoll_connect_op : epoll_op
 
 //------------------------------------------------------------------------------
 
-/** Read operation state with buffer descriptors. */
 struct epoll_read_op : epoll_op
 {
     static constexpr std::size_t max_buffers = 16;
     iovec iovecs[max_buffers];
     int iovec_count = 0;
-
-    // True when 0 bytes is due to empty buffer, not EOF
     bool empty_buffer_read = false;
 
-    // EOF only applies when we actually tried to read something
     bool is_read_operation() const noexcept override
     {
         return !empty_buffer_read;
@@ -199,7 +232,6 @@ struct epoll_read_op : epoll_op
 
 //------------------------------------------------------------------------------
 
-/** Write operation state with buffer descriptors. */
 struct epoll_write_op : epoll_op
 {
     static constexpr std::size_t max_buffers = 16;
@@ -214,7 +246,11 @@ struct epoll_write_op : epoll_op
 
     void perform_io() noexcept override
     {
-        ssize_t n = ::writev(fd, iovecs, iovec_count);
+        msghdr msg{};
+        msg.msg_iov = iovecs;
+        msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
+
+        ssize_t n = ::sendmsg(fd, &msg, MSG_NOSIGNAL);
         if (n >= 0)
             complete(0, static_cast<std::size_t>(n));
         else
@@ -224,7 +260,6 @@ struct epoll_write_op : epoll_op
 
 //------------------------------------------------------------------------------
 
-/** Accept operation state. */
 struct epoll_accept_op : epoll_op
 {
     int accepted_fd = -1;
@@ -241,7 +276,6 @@ struct epoll_accept_op : epoll_op
         accepted_fd = -1;
         peer_impl = nullptr;
         impl_out = nullptr;
-        // Don't reset create_peer and service_ptr - they're set once
     }
 
     void perform_io() noexcept override
@@ -302,7 +336,10 @@ struct epoll_accept_op : epoll_op
                 *impl_out = nullptr;
         }
 
-        d.dispatch(h).resume();
+        auto saved_d = d;
+        auto saved_h = std::move(h);
+        impl_ptr.reset();
+        saved_d.dispatch(saved_h).resume();
     }
 };
 

src/corosio/src/detail/epoll/scheduler.cpp

@@ -29,6 +29,47 @@
 #include <sys/socket.h>
 #include <unistd.h>
 
+/*
+    epoll Scheduler
+    ===============
+
+    The scheduler is the heart of the I/O event loop. It multiplexes I/O
+    readiness notifications from epoll with a completion queue for operations
+    that finished synchronously or were cancelled.
+
+    Event Loop Structure (do_one)
+    -----------------------------
+    1. Check completion queue first (mutex-protected)
+    2. If empty, call epoll_wait with calculated timeout
+    3. Process timer expirations
+    4. For each ready fd, claim the operation and perform I/O
+    5. Push completed operations to completion queue
+    6. Pop one and invoke its handler
+
+    The completion queue exists because handlers must run outside the epoll
+    processing loop. This allows handlers to safely start new operations
+    on the same fd without corrupting iteration state.
+
+    Wakeup Mechanism
+    ----------------
+    An eventfd allows other threads (or cancel/post calls) to wake the
+    event loop from epoll_wait. We distinguish wakeup events from I/O by
+    storing nullptr in epoll_event.data.ptr for the eventfd.
+
+    Work Counting
+    -------------
+    outstanding_work_ tracks pending operations. When it hits zero, run()
+    returns. This is how io_context knows there's nothing left to do.
+    Each operation increments on start, decrements on completion.
+
+    Timer Integration
+    -----------------
+    Timers are handled by timer_service. The scheduler adjusts epoll_wait
+    timeout to wake in time for the nearest timer expiry. When a new timer
+    is scheduled earlier than current, timer_service calls wakeup() to
+    re-evaluate the timeout.
+*/
+
 namespace boost {
 namespace corosio {
 namespace detail {
@@ -84,7 +125,6 @@ epoll_scheduler(
         detail::throw_system_error(make_err(errn), "eventfd");
     }
 
-    // data.ptr = nullptr distinguishes wakeup events from I/O completions
     epoll_event ev{};
     ev.events = EPOLLIN;
     ev.data.ptr = nullptr;
@@ -119,16 +159,13 @@ shutdown()
     std::unique_lock lock(mutex_);
     shutdown_ = true;
 
-    // Drain all completed operations without invoking handlers
     while (auto* h = completed_ops_.pop())
     {
         lock.unlock();
         h->destroy();
         lock.lock();
     }
 
-    // Reset outstanding work count - any pending I/O operations
-    // will be cleaned up when their owning objects are destroyed
     outstanding_work_.store(0, std::memory_order_release);
 }
 
@@ -357,7 +394,6 @@ void
 epoll_scheduler::
 unregister_fd(int fd) const
 {
-    // EPOLL_CTL_DEL ignores the event parameter (can be NULL on Linux 2.6.9+)
     ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
 }
 
@@ -379,12 +415,10 @@ void
 epoll_scheduler::
 wakeup() const
 {
-    // Write cannot fail: eventfd counter won't overflow with single increments
     std::uint64_t val = 1;
     [[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val));
 }
 
-// RAII guard - work_finished called even if handler throws
 struct work_guard
 {
     epoll_scheduler const* self;
@@ -459,30 +493,30 @@ do_one(long timeout_us)
         {
             if (errno == EINTR)
             {
-                // EINTR: retry for infinite waits, return for timed waits
                 if (timeout_us < 0)
                     continue;
                 return 0;
             }
             detail::throw_system_error(make_err(errno), "epoll_wait");
         }
 
-        // May dispatch timer handlers inline
         timer_svc_->process_expired();
 
         for (int i = 0; i < nfds; ++i)
         {
             if (events[i].data.ptr == nullptr)
             {
-                // Drain eventfd; read cannot fail since epoll signaled readiness
                 std::uint64_t val;
                 [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
                 continue;
             }
 
             auto* op = static_cast<epoll_op*>(events[i].data.ptr);
 
-            // One-shot: unregister before I/O
+            bool was_registered = op->registered.exchange(false, std::memory_order_acq_rel);
+            if (!was_registered)
+                continue;
+
             unregister_fd(op->fd);
 
             if (events[i].events & (EPOLLERR | EPOLLHUP))
@@ -521,7 +555,6 @@ do_one(long timeout_us)
             return 1;
         }
 
-        // Finite timeout: return on timeout; infinite: keep looping
         if (timeout_us >= 0)
             return 0;
     }