Refactor Win32 iocp evloop

src/crystal/event_loop/iocp.cr

@@ -23,6 +23,47 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
     true
   end
 
+  {% if !flag?(:without_mt) && !flag?(:preview_mt) || flag?(:execution_context) %}
+    # Creates a global IOCP instance, then forwards the completion events to their
+    # original instance (through a dedicated thread), because:
+    #
+    # We can't associate an OVERLAPPED operation to an IOCP instance.
+    # We can only assign a FILE or SOCKET to a single IOCP for its whole lifetime.
+    # Each EventLoop can only receive completion events for its execution context.
+    #
+    # WARNING: the global IOCP instance MUST only receive completion events for
+    # OVERLAPPED operations on FILE and SOCKET handles. Other completion events
+    # must be sent to the local IOCP instances.
+    @@global = System::IOCP.new
+
+    def self.start_forwarder_thread : Nil
+      Thread.new("IOCP") do
+        buffer = uninitialized LibC::OVERLAPPED_ENTRY[64]
+
+        while true
+          ret = LibC.GetQueuedCompletionStatusEx(@@global.handle, buffer, buffer.size, out removed, LibC::INFINITE, 0)
+          System.panic("GetQueuedCompletionStatusEx", WinError.value) if ret == 0
+
+          overlapped_entry = buffer.to_unsafe
+          removed.times do
+            overlapped = overlapped_entry.value.lpOverlapped
+            operation = System::IOCP::OverlappedOperation.unbox(overlapped)
+
+            ret = LibC.PostQueuedCompletionStatus(
+              operation.iocp_handle,
+              overlapped_entry.value.dwNumberOfBytesTransferred,
+              overlapped_entry.value.lpCompletionKey,
+              overlapped,
+            )
+            System.panic "PostQueuedCompletionStatus", WinError.value if ret == 0
+
+            overlapped_entry += 1
+          end
+        end
+      end
+    end
+  {% end %}
+
   @waitable_timer : System::WaitableTimer?
   @timer_packet = LibC::HANDLE.null
   @timer_key : System::IOCP::CompletionKey?
@@ -31,7 +72,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
     @timers_mutex = Thread::Mutex.new
     @timers = Timers(Timer).new
 
-    # the completion port
+    # the local completion port
     @iocp = System::IOCP.new
 
     # custom completion to interrupt a blocking run
@@ -48,13 +89,18 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
     end
   end
 
-  # Returns the base IO Completion Port.
   def iocp_handle : LibC::HANDLE
     @iocp.handle
   end
 
   def create_completion_port(handle : LibC::HANDLE) : LibC::HANDLE
-    iocp = LibC.CreateIoCompletionPort(handle, @iocp.handle, nil, 0)
+    iocp_handle =
+      {% if !flag?(:without_mt) && !flag?(:preview_mt) || flag?(:execution_context) %}
+        @@global.handle
+      {% else %}
+        @iocp.handle
+      {% end %}
+    iocp = LibC.CreateIoCompletionPort(handle, iocp_handle, nil, 0)
     raise IO::Error.from_winerror("CreateIoCompletionPort") if iocp.null?
 
     # all overlapped operations may finish synchronously, in which case we do
@@ -303,7 +349,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   end
 
   def read(file_descriptor : Crystal::System::FileDescriptor, slice : Bytes) : Int32
-    System::IOCP.overlapped_operation(file_descriptor, "ReadFile", file_descriptor.read_timeout) do |overlapped|
+    System::IOCP.overlapped_operation(@iocp.handle, file_descriptor, "ReadFile", file_descriptor.read_timeout) do |overlapped|
       ret = LibC.ReadFile(file_descriptor.windows_handle, slice, slice.size, out byte_count, overlapped)
       {ret, byte_count}
     end.to_i32
@@ -314,14 +360,14 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   end
 
   def pread(file_descriptor : System::FileDescriptor, slice : Bytes, offset : Int64) : Int32
-    System::IOCP.overlapped_operation(file_descriptor, "ReadFile", file_descriptor.read_timeout, offset: offset) do |overlapped|
+    System::IOCP.overlapped_operation(@iocp.handle, file_descriptor, "ReadFile", file_descriptor.read_timeout, offset: offset) do |overlapped|
       ret = LibC.ReadFile(file_descriptor.windows_handle, slice, slice.size, out byte_count, overlapped)
       {ret, byte_count}
     end.to_i32
   end
 
   def write(file_descriptor : Crystal::System::FileDescriptor, slice : Bytes) : Int32
-    bytes_written = System::IOCP.overlapped_operation(file_descriptor, "WriteFile", file_descriptor.write_timeout, writing: true) do |overlapped|
+    bytes_written = System::IOCP.overlapped_operation(@iocp.handle, file_descriptor, "WriteFile", file_descriptor.write_timeout, writing: true) do |overlapped|
       overlapped.offset = UInt64::MAX if file_descriptor.system_append?
 
       ret = LibC.WriteFile(file_descriptor.windows_handle, slice, slice.size, out byte_count, overlapped)
@@ -378,7 +424,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   def read(socket : ::Socket, slice : Bytes) : Int32
     wsabuf = wsa_buffer(slice)
 
-    bytes_read = System::IOCP.wsa_overlapped_operation(socket, socket.fd, "WSARecv", socket.read_timeout, connreset_is_error: false) do |overlapped|
+    bytes_read = System::IOCP.wsa_overlapped_operation(@iocp.handle, socket, socket.fd, "WSARecv", socket.read_timeout, connreset_is_error: false) do |overlapped|
       flags = 0_u32
       ret = LibC.WSARecv(socket.fd, pointerof(wsabuf), 1, out bytes_received, pointerof(flags), overlapped, nil)
       {ret, bytes_received}
@@ -397,7 +443,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   def write(socket : ::Socket, slice : Bytes) : Int32
     wsabuf = wsa_buffer(slice)
 
-    bytes = System::IOCP.wsa_overlapped_operation(socket, socket.fd, "WSASend", socket.write_timeout) do |overlapped|
+    bytes = System::IOCP.wsa_overlapped_operation(@iocp.handle, socket, socket.fd, "WSASend", socket.write_timeout) do |overlapped|
       ret = LibC.WSASend(socket.fd, pointerof(wsabuf), 1, out bytes_sent, 0, overlapped, nil)
       {ret, bytes_sent}
     end
@@ -414,7 +460,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
 
   def send_to(socket : ::Socket, slice : Bytes, address : ::Socket::Address) : Int32
     wsabuf = wsa_buffer(slice)
-    bytes_written = System::IOCP.wsa_overlapped_operation(socket, socket.fd, "WSASendTo", socket.write_timeout) do |overlapped|
+    bytes_written = System::IOCP.wsa_overlapped_operation(@iocp.handle, socket, socket.fd, "WSASendTo", socket.write_timeout) do |overlapped|
       ret = LibC.WSASendTo(socket.fd, pointerof(wsabuf), 1, out bytes_sent, 0, address, address.size, overlapped, nil)
       {ret, bytes_sent}
     end
@@ -440,7 +486,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
     wsabuf = wsa_buffer(slice)
 
     flags = 0_u32
-    bytes_read = System::IOCP.wsa_overlapped_operation(socket, socket.fd, "WSARecvFrom", socket.read_timeout) do |overlapped|
+    bytes_read = System::IOCP.wsa_overlapped_operation(@iocp.handle, socket, socket.fd, "WSARecvFrom", socket.read_timeout) do |overlapped|
       ret = LibC.WSARecvFrom(socket.fd, pointerof(wsabuf), 1, out bytes_received, pointerof(flags), sockaddr, pointerof(addrlen), overlapped, nil)
       {ret, bytes_received}
     end
@@ -449,7 +495,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   end
 
   def connect(socket : ::Socket, address : ::Socket::Addrinfo | ::Socket::Address, timeout : ::Time::Span?) : IO::Error?
-    System::IOCP::WSAOverlappedOperation.run(socket.fd) do |operation|
+    System::IOCP::WSAOverlappedOperation.run(@iocp.handle, socket.fd) do |operation|
       # This is: LibC.ConnectEx(fd, address, address.size, nil, 0, nil, overlapped)
       result = System::Socket.connect_ex.call(socket.fd, address.to_unsafe, address.size, Pointer(Void).null, 0_u32, Pointer(UInt32).null, operation.to_unsafe)
 
@@ -513,7 +559,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
   end
 
   private def overlapped_accept(socket, &)
-    System::IOCP::WSAOverlappedOperation.run(socket.fd) do |operation|
+    System::IOCP::WSAOverlappedOperation.run(@iocp.handle, socket.fd) do |operation|
       result = yield operation
 
       if result == 0
@@ -549,7 +595,7 @@ class Crystal::EventLoop::IOCP < Crystal::EventLoop
     # can't send more than 2,147,483,646 bytes at once
     len = LibC::DWORD.new(count.clamp(..(Int32::MAX - 1)))
 
-    Crystal::System::IOCP::WSAOverlappedOperation.run(socket.fd) do |operation|
+    Crystal::System::IOCP::WSAOverlappedOperation.run(@iocp.handle, socket.fd) do |operation|
       operation.@overlapped.union.offset.offset = LibC::DWORD.new!(offset)
       operation.@overlapped.union.offset.offsetHigh = LibC::DWORD.new!(offset >> 32)
 

src/crystal/system/win32/addrinfo.cr

@@ -46,7 +46,7 @@ module Crystal::System::Addrinfo
     IOCP::GetAddrInfoOverlappedOperation.run(Crystal::EventLoop.current.iocp_handle) do |operation|
       completion_routine = LibC::LPLOOKUPSERVICE_COMPLETION_ROUTINE.new do |dwError, dwBytes, lpOverlapped|
         orig_operation = IOCP::GetAddrInfoOverlappedOperation.unbox(lpOverlapped)
-        LibC.PostQueuedCompletionStatus(orig_operation.iocp, 0, 0, lpOverlapped)
+        LibC.PostQueuedCompletionStatus(orig_operation.iocp_handle, 0, 0, lpOverlapped)
       end
 
       # NOTE: we handle the timeout ourselves so we don't pass a `LibC::Timeval`

src/crystal/system/win32/file_descriptor.cr

@@ -280,7 +280,7 @@ module Crystal::System::FileDescriptor
   end
 
   private def lock_file(handle, flags)
-    IOCP::IOOverlappedOperation.run(handle) do |operation|
+    IOCP::IOOverlappedOperation.run(Crystal::EventLoop.current.iocp_handle, handle) do |operation|
       result = LibC.LockFileEx(handle, flags, 0, 0xFFFF_FFFF, 0xFFFF_FFFF, operation)
 
       if result == 0
@@ -306,7 +306,7 @@ module Crystal::System::FileDescriptor
   end
 
   private def unlock_file(handle)
-    IOCP::IOOverlappedOperation.run(handle) do |operation|
+    IOCP::IOOverlappedOperation.run(Crystal::EventLoop.current.iocp_handle, handle) do |operation|
       result = LibC.UnlockFileEx(handle, 0, 0xFFFF_FFFF, 0xFFFF_FFFF, operation)
 
       if result == 0

src/crystal/system/win32/iocp.cr

@@ -173,6 +173,7 @@ struct Crystal::System::IOCP
     private abstract def try_cancel : Bool
 
     @overlapped = LibC::OVERLAPPED.new
+    getter iocp_handle : LibC::HANDLE = LibC::INVALID_HANDLE_VALUE
     @fiber = ::Fiber.current
     @state : State = :started
 
@@ -182,6 +183,8 @@ struct Crystal::System::IOCP
       yield operation
     end
 
+    # We can't associate user data to OVERLAPPED operations, and must
+    # reconstruct an OverlappedOperation from a Pointer(OVERLAPPED).
     def self.unbox(overlapped : LibC::OVERLAPPED*) : self
       start = overlapped.as(Pointer(UInt8)) - offsetof(self, @overlapped)
       Box(self).unbox(start.as(Pointer(Void)))
@@ -217,7 +220,7 @@ struct Crystal::System::IOCP
   end
 
   class IOOverlappedOperation < OverlappedOperation
-    def initialize(@handle : LibC::HANDLE)
+    def initialize(@iocp_handle : LibC::HANDLE, @handle : LibC::HANDLE)
     end
 
     def offset=(value : UInt64)
@@ -260,7 +263,7 @@ struct Crystal::System::IOCP
   end
 
   class WSAOverlappedOperation < OverlappedOperation
-    def initialize(@handle : LibC::SOCKET)
+    def initialize(@iocp_handle : LibC::HANDLE, @handle : LibC::SOCKET)
     end
 
     def wait_for_result(timeout, & : WinError ->)
@@ -299,10 +302,9 @@ struct Crystal::System::IOCP
   end
 
   class GetAddrInfoOverlappedOperation < OverlappedOperation
-    getter iocp
     setter cancel_handle : LibC::HANDLE = LibC::INVALID_HANDLE_VALUE
 
-    def initialize(@iocp : LibC::HANDLE)
+    def initialize(@iocp_handle : LibC::HANDLE)
     end
 
     def wait_for_result(timeout, & : WinError ->)
@@ -334,11 +336,11 @@ struct Crystal::System::IOCP
     end
   end
 
-  def self.overlapped_operation(file_descriptor, method, timeout, *, offset = nil, writing = false, &)
+  def self.overlapped_operation(iocp_handle, file_descriptor, method, timeout, *, offset = nil, writing = false, &)
     handle = file_descriptor.windows_handle
     seekable = LibC.SetFilePointerEx(handle, 0, out original_offset, IO::Seek::Current) != 0
 
-    IOOverlappedOperation.run(handle) do |operation|
+    IOOverlappedOperation.run(iocp_handle, handle) do |operation|
       overlapped = operation.to_unsafe
       if seekable
         start_offset = offset || original_offset
@@ -391,8 +393,8 @@ struct Crystal::System::IOCP
     end
   end
 
-  def self.wsa_overlapped_operation(target, socket, method, timeout, connreset_is_error = true, &)
-    WSAOverlappedOperation.run(socket) do |operation|
+  def self.wsa_overlapped_operation(iocp_handle, target, socket, method, timeout, connreset_is_error = true, &)
+    WSAOverlappedOperation.run(iocp_handle, socket) do |operation|
       result, value = yield operation
 
       if result == LibC::SOCKET_ERROR

src/kernel.cr

@@ -599,6 +599,10 @@ end
   end
 {% end %}
 
+{% if flag?(:win32) && (!flag?(:without_mt) && !flag?(:preview_mt) || flag?(:execution_context)) %}
+  Crystal::EventLoop::IOCP.start_forwarder_thread
+{% end %}
+
 {% unless flag?(:interpreted) || flag?(:wasm32) %}
   {% if !flag?(:without_mt) && !flag?(:preview_mt) || flag?(:execution_context) %}
     Fiber::ExecutionContext.init_default_context