Improve IOCP performance

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

@@ -31,6 +31,7 @@
 #include <boost/capy/buffers.hpp>
 
 #include <filesystem>
+#include <utility>
 
 namespace boost::corosio::detail {
 
@@ -216,7 +217,7 @@ win_stream_file::read_some(
     std::error_code* ec,
     std::size_t* bytes)
 {
-    return internal_.read_some(cont, d, buf, token, ec, bytes);
+    return internal_.read_some(cont, d, buf, std::move(token), ec, bytes);
 }
 
 inline std::coroutine_handle<>
@@ -228,7 +229,7 @@ win_stream_file::write_some(
     std::error_code* ec,
     std::size_t* bytes)
 {
-    return internal_.write_some(cont, d, buf, token, ec, bytes);
+    return internal_.write_some(cont, d, buf, std::move(token), ec, bytes);
 }
 
 inline native_handle_type

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

@@ -171,6 +171,13 @@ struct overlapped_op
     DWORD bytes_transferred       = 0;
     cancel_func_type cancel_func_ = nullptr;
 
+    /** Whether the wait reactor holds this op. Set under the reactor's
+        lock when a register queues the op, cleared by the reactor before
+        it completes the op, and read under that lock by a cancel, which
+        has nothing to ask the reactor for while it is clear. reset()
+        leaves it alone: only the reactor knows when it lets go. */
+    std::atomic<bool> reactor_parked_{false};
+
     explicit overlapped_op(func_type func) noexcept : coro_op(func)
     {
         reset_overlapped();

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

@@ -32,6 +32,7 @@
 #include <boost/capy/error.hpp>
 
 #include <filesystem>
+#include <utility>
 
 namespace boost::corosio::detail {
 
@@ -189,7 +190,8 @@ win_random_access_file::read_some_at(
     std::error_code* ec,
     std::size_t* bytes)
 {
-    return internal_.read_some_at(offset, cont, d, buf, token, ec, bytes);
+    return internal_.read_some_at(
+        offset, cont, d, buf, std::move(token), ec, bytes);
 }
 
 inline std::coroutine_handle<>
@@ -202,7 +204,8 @@ win_random_access_file::write_some_at(
     std::error_code* ec,
     std::size_t* bytes)
 {
-    return internal_.write_some_at(offset, cont, d, buf, token, ec, bytes);
+    return internal_.write_some_at(
+        offset, cont, d, buf, std::move(token), ec, bytes);
 }
 
 inline native_handle_type

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

@@ -808,6 +808,13 @@ win_scheduler::update_timeout()
         return;
     if (timers_)
     {
+        // Read and arm under one lock. Two threads arming at once could
+        // otherwise read different earliest expiries and set the timer
+        // in the opposite order, leaving the later deadline set and the
+        // earlier timer firing late. A new earliest expiry is published
+        // before its notification takes this lock, so the last thread
+        // through arms the earliest one.
+        std::lock_guard<win_mutex> lock(dispatch_mutex_);
         timers_->update_timeout(tsvc->nearest_expiry());
     }
     else

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

@@ -46,6 +46,9 @@ struct accept_op : overlapped_op
     SOCKET listen_socket                 = INVALID_SOCKET;
     io_object::implementation** impl_out = nullptr;
     char addr_buf[2 * (sizeof(sockaddr_in6) + 16)];
+    // The size AcceptEx was given for each address in addr_buf, which
+    // GetAcceptExSockaddrs needs to find them again.
+    DWORD addr_len = 0;
 
     static void do_complete(
         void* owner,

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

@@ -33,6 +33,7 @@
 
 #include <ws2tcpip.h>
 
+#include <cstring>
 #include <system_error>
 
 namespace boost::corosio::detail {
@@ -120,6 +121,18 @@ acceptor_wait_op::do_cancel_impl(overlapped_op* base) noexcept
     op->acceptor.socket_service().scheduler().cancel_wait(op);
 }
 
+// An address GetAcceptExSockaddrs pointed into the AcceptEx buffer,
+// as an endpoint; one it could not find stays the default endpoint.
+inline endpoint
+accept_ex_endpoint(sockaddr const* sa, int len) noexcept
+{
+    sockaddr_storage storage{};
+    if (!sa || len <= 0 || len > static_cast<int>(sizeof(storage)))
+        return {};
+    std::memcpy(&storage, sa, static_cast<std::size_t>(len));
+    return from_sockaddr(storage);
+}
+
 // accept_op completion handler
 
 inline void
@@ -172,20 +185,41 @@ accept_op::do_complete(
 
         op->peer_wrapper->set_socket(op->accepted_socket);
 
-        sockaddr_storage local_storage{};
-        int local_len = sizeof(local_storage);
-        sockaddr_storage remote_storage{};
-        int remote_len = sizeof(remote_storage);
-
         endpoint local_ep, remote_ep;
-        if (::getsockname(
-                op->accepted_socket,
-                reinterpret_cast<sockaddr*>(&local_storage), &local_len) == 0)
-            local_ep = from_sockaddr(local_storage);
-        if (::getpeername(
-                op->accepted_socket,
-                reinterpret_cast<sockaddr*>(&remote_storage), &remote_len) == 0)
-            remote_ep = from_sockaddr(remote_storage);
+        if (auto parse =
+                op->acceptor.socket_service().get_accept_ex_sockaddrs())
+        {
+            // AcceptEx already wrote both addresses into addr_buf, so
+            // reading them back costs no syscall, where getsockname and
+            // getpeername cost one each on every accept.
+            sockaddr* local_sa  = nullptr;
+            sockaddr* remote_sa = nullptr;
+            int local_len       = 0;
+            int remote_len      = 0;
+            parse(
+                op->addr_buf, 0, op->addr_len, op->addr_len, &local_sa,
+                &local_len, &remote_sa, &remote_len);
+            local_ep  = accept_ex_endpoint(local_sa, local_len);
+            remote_ep = accept_ex_endpoint(remote_sa, remote_len);
+        }
+        else
+        {
+            sockaddr_storage local_storage{};
+            int local_len = sizeof(local_storage);
+            sockaddr_storage remote_storage{};
+            int remote_len = sizeof(remote_storage);
+
+            if (::getsockname(
+                    op->accepted_socket,
+                    reinterpret_cast<sockaddr*>(&local_storage),
+                    &local_len) == 0)
+                local_ep = from_sockaddr(local_storage);
+            if (::getpeername(
+                    op->accepted_socket,
+                    reinterpret_cast<sockaddr*>(&remote_storage),
+                    &remote_len) == 0)
+                remote_ep = from_sockaddr(remote_storage);
+        }
 
         op->peer_wrapper->set_endpoints(local_ep, remote_ep);
         op->accepted_socket = INVALID_SOCKET;
@@ -369,6 +403,7 @@ win_tcp_acceptor::accept(
     // AcceptEx address buffer sizes must match the socket's address family
     DWORD addr_size = static_cast<DWORD>(
         (af == AF_INET6 ? sizeof(sockaddr_in6) : sizeof(sockaddr_in)) + 16);
+    op.addr_len          = addr_size;
     DWORD bytes_received = 0;
 
     BOOL ok = accept_ex(

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

@@ -181,6 +181,12 @@ class BOOST_COROSIO_DECL win_tcp_service final
         return accept_ex_;
     }
 
+    /** Return the GetAcceptExSockaddrs function pointer, or null. */
+    LPFN_GETACCEPTEXSOCKADDRS get_accept_ex_sockaddrs() const noexcept
+    {
+        return get_accept_ex_sockaddrs_;
+    }
+
     /** Post an overlapped operation for completion. */
     void post(overlapped_op* op);
 
@@ -209,6 +215,9 @@ class BOOST_COROSIO_DECL win_tcp_service final
     void* iocp_;
     LPFN_CONNECTEX connect_ex_ = nullptr;
     LPFN_ACCEPTEX accept_ex_   = nullptr;
+    // Optional: an accept parses its endpoints out of the AcceptEx
+    // buffer with it, and asks the kernel for them without it.
+    LPFN_GETACCEPTEXSOCKADDRS get_accept_ex_sockaddrs_ = nullptr;
     BOOST_COROSIO_MSVC_WARNING_PUSH
     BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // detail:: members, dll-interface
     object_pool<win_tcp_socket> pool_;
@@ -324,14 +333,12 @@ connect_op::do_complete(
             op->internal.socket_, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT,
             nullptr, 0);
 
-        endpoint local_ep;
-        sockaddr_storage local_storage{};
-        int local_len = sizeof(local_storage);
-        if (::getsockname(
-                op->internal.socket_,
-                reinterpret_cast<sockaddr*>(&local_storage), &local_len) == 0)
-            local_ep = from_sockaddr(local_storage);
-        op->internal.set_endpoints(local_ep, op->target_endpoint);
+        // Most connections never read their local endpoint, so it is
+        // left for local_endpoint() to fetch rather than costing every
+        // connect a getsockname.
+        op->internal.remote_endpoint_ = op->target_endpoint;
+        op->internal.local_state_.store(
+            win_tcp_socket::local_state::pending, std::memory_order_release);
     }
 
     auto prevent_premature_destruction = std::move(op->object_ref_);
@@ -432,6 +439,7 @@ win_tcp_socket::reuse() noexcept
     // can still have written these.
     local_endpoint_  = endpoint{};
     remote_endpoint_ = endpoint{};
+    local_state_.store(local_state::cached, std::memory_order_relaxed);
     BOOST_COROSIO_ASSERT(!conn_.stop_cb);
     BOOST_COROSIO_ASSERT(!rd_.stop_cb);
     BOOST_COROSIO_ASSERT(!wr_.stop_cb);
@@ -449,6 +457,7 @@ win_tcp_socket::set_endpoints(endpoint local, endpoint remote) noexcept
 {
     local_endpoint_  = local;
     remote_endpoint_ = remote;
+    local_state_.store(local_state::cached, std::memory_order_release);
 }
 
 inline std::coroutine_handle<>
@@ -472,9 +481,15 @@ win_tcp_socket::connect(
 
     svc_.work_started();
 
-    // ConnectEx requires the socket to be bound. Skip if already bound
-    // (e.g. the caller used tcp_socket::bind() before connect).
-    if (local_endpoint_ == endpoint{})
+    // ConnectEx requires the socket to be bound. A cache that names an
+    // endpoint says it is (e.g. tcp_socket::bind() before connect). An
+    // endpoint an earlier connect left unfetched says it most likely
+    // is, but that mark can outlive the socket it described when a late
+    // completion races a close and reopen, so the bind is tried anyway
+    // and an "already bound" refusal leaves the verdict to ConnectEx.
+    bool const unfetched =
+        local_state_.load(std::memory_order_acquire) != local_state::cached;
+    if (unfetched || local_endpoint_ == endpoint{})
     {
         sockaddr_storage bind_storage{};
         socklen_t bind_len;
@@ -501,8 +516,12 @@ win_tcp_socket::connect(
                 socket_, reinterpret_cast<sockaddr*>(&bind_storage),
                 bind_len) == SOCKET_ERROR)
         {
-            svc_.on_completion(&op, ::WSAGetLastError(), 0);
-            return std::noop_coroutine();
+            int const err = ::WSAGetLastError();
+            if (!unfetched || err != WSAEINVAL)
+            {
+                svc_.on_completion(&op, static_cast<DWORD>(err), 0);
+                return std::noop_coroutine();
+            }
         }
     }
 
@@ -807,6 +826,7 @@ win_tcp_socket::close_socket() noexcept
     // Clear cached endpoints
     local_endpoint_  = endpoint{};
     remote_endpoint_ = endpoint{};
+    local_state_.store(local_state::cached, std::memory_order_release);
 }
 
 inline std::error_code
@@ -853,6 +873,7 @@ win_tcp_socket::release_socket() noexcept
         family_          = AF_UNSPEC;
         local_endpoint_  = endpoint{};
         remote_endpoint_ = endpoint{};
+        local_state_.store(local_state::cached, std::memory_order_release);
     }
     return static_cast<native_handle_type>(s);
 }
@@ -883,7 +904,30 @@ win_tcp_socket::get_option(
 inline endpoint
 win_tcp_socket::local_endpoint() const noexcept
 {
-    return local_endpoint_;
+    auto state = local_state_.load(std::memory_order_acquire);
+    if (state == local_state::cached)
+        return local_endpoint_;
+
+    // A connect left the endpoint for the first read to fetch. One
+    // reader claims the cache and fills it; a reader racing it asks the
+    // kernel for its own copy rather than read the cache mid-write. A
+    // fetch that fails caches nothing, so the next read asks again.
+    sockaddr_storage storage{};
+    int len = sizeof(storage);
+    if (socket_ == INVALID_SOCKET ||
+        ::getsockname(socket_, reinterpret_cast<sockaddr*>(&storage), &len) !=
+            0)
+        return endpoint{};
+    endpoint const ep = from_sockaddr(storage);
+    if (state == local_state::pending &&
+        local_state_.compare_exchange_strong(
+            state, local_state::fetching, std::memory_order_acquire,
+            std::memory_order_relaxed))
+    {
+        local_endpoint_ = ep;
+        local_state_.store(local_state::cached, std::memory_order_release);
+    }
+    return ep;
 }
 
 inline endpoint
@@ -1041,6 +1085,8 @@ win_tcp_service::bind_socket(win_tcp_socket& impl, endpoint ep)
     if (::getsockname(
             sock, reinterpret_cast<sockaddr*>(&local_storage), &local_len) == 0)
         impl.local_endpoint_ = detail::from_sockaddr(local_storage);
+    impl.local_state_.store(
+        win_tcp_socket::local_state::cached, std::memory_order_release);
 
     return {};
 }
@@ -1105,6 +1151,12 @@ win_tcp_service::load_extension_functions()
         sizeof(accept_ex_guid), &accept_ex_, sizeof(accept_ex_), &bytes,
         nullptr, nullptr);
 
+    GUID sockaddrs_guid = WSAID_GETACCEPTEXSOCKADDRS;
+    ::WSAIoctl(
+        sock, SIO_GET_EXTENSION_FUNCTION_POINTER, &sockaddrs_guid,
+        sizeof(sockaddrs_guid), &get_accept_ex_sockaddrs_,
+        sizeof(get_accept_ex_sockaddrs_), &bytes, nullptr, nullptr);
+
     ::closesocket(sock);
 }
 

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

@@ -27,6 +27,7 @@
 #include <boost/corosio/native/detail/iocp/win_overlapped_op.hpp>
 #include <boost/corosio/native/detail/iocp/win_windows.hpp>
 
+#include <atomic>
 #include <coroutine>
 
 #include <mswsock.h>
@@ -144,8 +145,23 @@ class win_tcp_socket final
     wait_op wt_;
     SOCKET socket_ = INVALID_SOCKET;
     int family_    = AF_UNSPEC;
-    endpoint local_endpoint_;
+    // What local_endpoint_ holds. `cached`: it is authoritative, and
+    // empty for a socket with no known endpoint. `pending`: a connect
+    // left the endpoint unfetched, and the first local_endpoint() asks
+    // the kernel for it. `fetching`: the reader that claimed the fetch
+    // is writing the cache, and other readers fetch their own copy
+    // meanwhile. Whatever else writes local_endpoint_ sets `cached`.
+    enum class local_state : unsigned char
+    {
+        cached,
+        pending,
+        fetching
+    };
+
+    // Mutable so local_endpoint() can fill it on first read.
+    mutable endpoint local_endpoint_;
     endpoint remote_endpoint_;
+    mutable std::atomic<local_state> local_state_{local_state::cached};
 
 public:
     explicit win_tcp_socket(win_tcp_service& svc) noexcept;