Add ready_ CAS protocol and restructure IOCP work counting

Implement a thread-safety protocol to coordinate between the I/O
initiator thread and the GQCS dispatch thread:

- Add ready_ flag to overlapped_op with InterlockedCompareExchange
  handshake between do_one() and on_pending()/on_completion()
- Add on_pending() for async paths: CAS coordinates with do_one() to
  ensure the handler only dispatches after the initiator has returned
- Add on_completion() for sync error/noop paths: packs results into
  the op and posts with key_result_stored for deferred dispatch
- Move work_started() to the top of connect/read_some/write_some/accept
  unconditionally so one call covers all error and async paths
- Move work_finished() after complete() in do_one() so the handler
  runs while outstanding_work is still positive

Author: sgerbino

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

@@ -397,9 +397,11 @@ win_socket_internal::connect(
     op.h               = h;
     op.ex              = d;
     op.ec_out          = ec;
-    op.target_endpoint = ep; // Store target for endpoint caching
+    op.target_endpoint = ep;
     op.start(token);
 
+    svc_.work_started();
+
     sockaddr_in bind_addr{};
     bind_addr.sin_family      = AF_INET;
     bind_addr.sin_addr.s_addr = INADDR_ANY;
@@ -409,25 +411,19 @@ win_socket_internal::connect(
             socket_, reinterpret_cast<sockaddr*>(&bind_addr),
             sizeof(bind_addr)) == SOCKET_ERROR)
     {
-        op.dwError = ::WSAGetLastError();
-        svc_.post(&op);
-        // completion is always posted to scheduler queue, never inline.
+        svc_.on_completion(&op, ::WSAGetLastError(), 0);
         return std::noop_coroutine();
     }
 
     auto connect_ex = svc_.connect_ex();
     if (!connect_ex)
     {
-        op.dwError = WSAEOPNOTSUPP;
-        svc_.post(&op);
-        // completion is always posted to scheduler queue, never inline.
+        svc_.on_completion(&op, WSAEOPNOTSUPP, 0);
         return std::noop_coroutine();
     }
 
     sockaddr_in addr = detail::to_sockaddr_in(ep);
 
-    svc_.work_started();
-
     BOOL result = connect_ex(
         socket_, reinterpret_cast<sockaddr*>(&addr), sizeof(addr), nullptr, 0,
         nullptr, &op);
@@ -437,15 +433,12 @@ win_socket_internal::connect(
         DWORD err = ::WSAGetLastError();
         if (err != ERROR_IO_PENDING)
         {
-            svc_.work_finished();
-            op.dwError = err;
-            svc_.post(&op);
-            // completion is always posted to scheduler queue, never inline.
+            svc_.on_completion(&op, err, 0);
             return std::noop_coroutine();
         }
     }
-    // Synchronous completion: IOCP will deliver the completion packet
-    // completion is always posted to scheduler queue, never inline.
+
+    svc_.on_pending(&op);
     return std::noop_coroutine();
 }
 
@@ -470,18 +463,18 @@ win_socket_internal::read_some(
     op.bytes_out = bytes_out;
     op.start(token);
 
-    // Prepare buffers (must happen before initiator runs)
+    svc_.work_started();
+
+    // Prepare buffers
     capy::mutable_buffer bufs[read_op::max_buffers];
     op.wsabuf_count =
         static_cast<DWORD>(param.copy_to(bufs, read_op::max_buffers));
 
-    // Handle empty buffer: complete with 0 bytes via post for consistency
+    // Handle empty buffer: complete with 0 bytes
     if (op.wsabuf_count == 0)
     {
-        op.bytes_transferred = 0;
-        op.dwError           = 0;
-        op.empty_buffer      = true;
-        svc_.post(&op);
+        op.empty_buffer = true;
+        svc_.on_completion(&op, 0, 0);
         return std::noop_coroutine();
     }
 
@@ -491,11 +484,8 @@ win_socket_internal::read_some(
         op.wsabufs[i].len = static_cast<ULONG>(bufs[i].size());
     }
 
-    // Issue WSARecv directly — caller is already suspended
     op.flags = 0;
 
-    svc_.work_started();
-
     int result = ::WSARecv(
         socket_, op.wsabufs, op.wsabuf_count, nullptr, &op.flags, &op,
         nullptr);
@@ -505,15 +495,14 @@ win_socket_internal::read_some(
         DWORD err = ::WSAGetLastError();
         if (err != WSA_IO_PENDING)
         {
-            svc_.work_finished();
-            op.dwError = err;
-            svc_.post(&op);
+            svc_.on_completion(&op, err, 0);
             return std::noop_coroutine();
         }
     }
-    // I/O is now pending. If stop was requested before WSARecv
-    // started, the CancelIoEx in the stop callback had nothing
-    // to cancel. Re-check and cancel the now-pending operation.
+
+    svc_.on_pending(&op);
+
+    // Re-check cancellation after I/O is pending
     if (op.cancelled.load(std::memory_order_acquire))
         ::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
 
@@ -540,17 +529,17 @@ win_socket_internal::write_some(
     op.bytes_out = bytes_out;
     op.start(token);
 
-    // Prepare buffers (must happen before initiator runs)
+    svc_.work_started();
+
+    // Prepare buffers
     capy::mutable_buffer bufs[write_op::max_buffers];
     op.wsabuf_count =
         static_cast<DWORD>(param.copy_to(bufs, write_op::max_buffers));
 
     // Handle empty buffer: complete immediately with 0 bytes
     if (op.wsabuf_count == 0)
     {
-        op.bytes_transferred = 0;
-        op.dwError           = 0;
-        svc_.post(&op);
+        svc_.on_completion(&op, 0, 0);
         return std::noop_coroutine();
     }
 
@@ -560,9 +549,6 @@ win_socket_internal::write_some(
         op.wsabufs[i].len = static_cast<ULONG>(bufs[i].size());
     }
 
-    // Issue WSASend directly — caller is already suspended
-    svc_.work_started();
-
     int result = ::WSASend(
         socket_, op.wsabufs, op.wsabuf_count, nullptr, 0, &op, nullptr);
 
@@ -571,12 +557,13 @@ win_socket_internal::write_some(
         DWORD err = ::WSAGetLastError();
         if (err != WSA_IO_PENDING)
         {
-            svc_.work_finished();
-            op.dwError = err;
-            svc_.post(&op);
+            svc_.on_completion(&op, err, 0);
             return std::noop_coroutine();
         }
     }
+
+    svc_.on_pending(&op);
+
     // Re-check cancellation after I/O is pending
     if (op.cancelled.load(std::memory_order_acquire))
         ::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
@@ -1005,6 +992,18 @@ win_sockets::post(overlapped_op* op)
     sched_.post(op);
 }
 
+inline void
+win_sockets::on_pending(overlapped_op* op) noexcept
+{
+    sched_.on_pending(op);
+}
+
+inline void
+win_sockets::on_completion(overlapped_op* op, DWORD error, DWORD bytes) noexcept
+{
+    sched_.on_completion(op, error, bytes);
+}
+
 inline void
 win_sockets::work_started() noexcept
 {
@@ -1179,6 +1178,8 @@ win_acceptor_internal::accept(
     op.impl_out = impl_out;
     op.start(token);
 
+    svc_.work_started();
+
     // Create wrapper for the peer socket (service owns it)
     auto& peer_wrapper = static_cast<win_socket&>(*svc_.construct());
 
@@ -1189,9 +1190,7 @@ win_acceptor_internal::accept(
     if (accepted == INVALID_SOCKET)
     {
         svc_.destroy(&peer_wrapper);
-        op.dwError = ::WSAGetLastError();
-        svc_.post(&op);
-        // completion is always posted to scheduler queue, never inline.
+        svc_.on_completion(&op, ::WSAGetLastError(), 0);
         return std::noop_coroutine();
     }
 
@@ -1203,9 +1202,7 @@ win_acceptor_internal::accept(
         DWORD err = ::GetLastError();
         ::closesocket(accepted);
         svc_.destroy(&peer_wrapper);
-        op.dwError = err;
-        svc_.post(&op);
-        // completion is always posted to scheduler queue, never inline.
+        svc_.on_completion(&op, err, 0);
         return std::noop_coroutine();
     }
 
@@ -1221,14 +1218,11 @@ win_acceptor_internal::accept(
         svc_.destroy(&peer_wrapper);
         op.peer_wrapper    = nullptr;
         op.accepted_socket = INVALID_SOCKET;
-        op.dwError         = WSAEOPNOTSUPP;
-        svc_.post(&op);
-        // completion is always posted to scheduler queue, never inline.
+        svc_.on_completion(&op, WSAEOPNOTSUPP, 0);
         return std::noop_coroutine();
     }
 
     DWORD bytes_received = 0;
-    svc_.work_started();
 
     BOOL ok = accept_ex(
         socket_, accepted, op.addr_buf, 0, sizeof(sockaddr_in) + 16,
@@ -1239,19 +1233,16 @@ win_acceptor_internal::accept(
         DWORD err = ::WSAGetLastError();
         if (err != ERROR_IO_PENDING)
         {
-            svc_.work_finished();
             ::closesocket(accepted);
             svc_.destroy(&peer_wrapper);
             op.peer_wrapper    = nullptr;
             op.accepted_socket = INVALID_SOCKET;
-            op.dwError         = err;
-            svc_.post(&op);
-            // completion is always posted to scheduler queue, never inline.
+            svc_.on_completion(&op, err, 0);
             return std::noop_coroutine();
         }
     }
-    // Synchronous completion: IOCP will deliver the completion packet
-    // completion is always posted to scheduler queue, never inline.
+
+    svc_.on_pending(&op);
     return std::noop_coroutine();
 }
 

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

@@ -60,6 +60,7 @@ struct overlapped_op
     /** Function pointer type for cancellation hook. */
     using cancel_func_type = void (*)(overlapped_op*) noexcept;
 
+    long ready_             = 0;
     std::coroutine_handle<> h;
     capy::executor_ref ex;
     std::error_code* ec_out = nullptr;
@@ -89,6 +90,7 @@ struct overlapped_op
     void reset() noexcept
     {
         reset_overlapped();
+        ready_            = 0;
         dwError           = 0;
         bytes_transferred = 0;
         empty_buffer      = false;

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

@@ -80,6 +80,14 @@ class BOOST_COROSIO_DECL win_scheduler final
     void work_started() noexcept override;
     void work_finished() noexcept override;
 
+    /** Signal that an overlapped I/O operation is now pending.
+        Coordinates with do_one() via the ready_ CAS protocol. */
+    void on_pending(overlapped_op* op) const;
+
+    /** Post an immediate completion with pre-stored results.
+        Used for sync errors and noop paths. */
+    void on_completion(overlapped_op* op, DWORD error, DWORD bytes) const;
+
     // Timer service integration
     void set_timer_service(timer_service* svc);
     void update_timeout();
@@ -312,6 +320,42 @@ win_scheduler::work_finished() noexcept
         stop();
 }
 
+inline void
+win_scheduler::on_pending(overlapped_op* op) const
+{
+    // CAS: try to set ready_ from 0 to 1.
+    // If the old value was 1, GQCS already grabbed this op and stored
+    // results — we need to re-post so do_one() can dispatch it.
+    if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
+    {
+        if (!::PostQueuedCompletionStatus(
+                iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
+        {
+            std::lock_guard<win_mutex> lock(dispatch_mutex_);
+            completed_ops_.push(op);
+            ::InterlockedExchange(&dispatch_required_, 1);
+        }
+    }
+}
+
+inline void
+win_scheduler::on_completion(
+    overlapped_op* op, DWORD error, DWORD bytes) const
+{
+    // Sync completion: pack results into op and post for dispatch.
+    op->ready_            = 1;
+    op->dwError           = error;
+    op->bytes_transferred = bytes;
+
+    if (!::PostQueuedCompletionStatus(
+            iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
+    {
+        std::lock_guard<win_mutex> lock(dispatch_mutex_);
+        completed_ops_.push(op);
+        ::InterlockedExchange(&dispatch_required_, 1);
+    }
+}
+
 inline void
 win_scheduler::stop()
 {
@@ -493,28 +537,39 @@ win_scheduler::do_one(unsigned long timeout_ms)
             case key_io:
             case key_result_stored:
             {
-                // Actual I/O completion: overlapped is OVERLAPPED* (first base of overlapped_op)
                 auto* ov_op = overlapped_to_op(overlapped);
 
-                // If key_result_stored, results are pre-stored in overlapped fields
+                // If key_result_stored, results are pre-stored in op fields
                 if (key == key_result_stored)
                 {
                     bytes = ov_op->bytes_transferred;
                     err   = ov_op->dwError;
                 }
 
+                // Store GQCS results so on_pending() re-post has valid data
                 ov_op->store_result(bytes, err);
-                work_finished();
-                ov_op->complete(this, bytes, err);
-                return 1;
+
+                // CAS: try to set ready_ from 0 to 1.
+                // If old value was 1, the initiator already returned
+                // (on_pending/on_completion set it) — safe to dispatch.
+                // If old value was 0, the initiator hasn't returned yet —
+                // skip dispatch; on_pending() will re-post.
+                if (::InterlockedCompareExchange(
+                        &ov_op->ready_, 1, 0) == 1)
+                {
+                    ov_op->complete(this, bytes, err);
+                    work_finished();
+                    return 1;
+                }
+                continue;
             }
 
             case key_posted:
             {
                 // Posted scheduler_op*: overlapped is actually a scheduler_op*
                 auto* op = reinterpret_cast<scheduler_op*>(overlapped);
-                work_finished();
                 op->complete(this, bytes, err);
+                work_finished();
                 return 1;
             }
 

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

@@ -129,6 +129,12 @@ class BOOST_COROSIO_DECL win_sockets final
     /** Post an overlapped operation for completion. */
     void post(overlapped_op* op);
 
+    /** Signal that an overlapped I/O is now pending (CAS protocol). */
+    void on_pending(overlapped_op* op) noexcept;
+
+    /** Post an immediate completion with pre-stored results. */
+    void on_completion(overlapped_op* op, DWORD error, DWORD bytes) noexcept;
+
     /** Notify scheduler of pending I/O work. */
     void work_started() noexcept;