From eb780d38af302fcc04ca9a7bc6c8c202d759daa5 Mon Sep 17 00:00:00 2001 From: Julien Portalier Date: Fri, 26 Jun 2026 18:41:25 +0200 Subject: [PATCH 1/2] Global IOCP for file and socket overlapped operations We must assign every File and Socket to an IOCP instance for its whole lifetime, but we can start multiple event loops, each with its own IOCP instance, meaning that opening a file or socket in evloop A then reading and writing in evloop B would keep sending the events to evloop A. With execution contexts this isn't acceptable: 1. a context can shutdown (e.g. the isolated context) and evloop A disappear: the would no more valid evloop to receive IOCP notifications for the file or socket; 2. every evloop is expected to recieve events for its context only (not others) so we can enqueue locally; 3. if the threads running evloop A are busy they can block the fibers of evloop B. --- src/crystal/event_loop/iocp.cr | 72 +++++++++++++++++---- src/crystal/system/win32/addrinfo.cr | 2 +- src/crystal/system/win32/file_descriptor.cr | 4 +- src/crystal/system/win32/iocp.cr | 18 +++--- src/kernel.cr | 4 ++ 5 files changed, 76 insertions(+), 24 deletions(-) diff --git a/src/crystal/event_loop/iocp.cr b/src/crystal/event_loop/iocp.cr index 23b9669233c8..29fa0273fc34 100644 --- a/src/crystal/event_loop/iocp.cr +++ b/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) diff --git a/src/crystal/system/win32/addrinfo.cr b/src/crystal/system/win32/addrinfo.cr index 465cc7b7c2da..e3464d3fb614 100644 --- a/src/crystal/system/win32/addrinfo.cr +++ b/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` diff --git a/src/crystal/system/win32/file_descriptor.cr b/src/crystal/system/win32/file_descriptor.cr index 93a4f099a0c7..8123d4be9591 100644 --- a/src/crystal/system/win32/file_descriptor.cr +++ b/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 diff --git a/src/crystal/system/win32/iocp.cr b/src/crystal/system/win32/iocp.cr index 121875c1870f..bf2d1940010a 100644 --- a/src/crystal/system/win32/iocp.cr +++ b/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 diff --git a/src/kernel.cr b/src/kernel.cr index 8ef25d54345f..93707e729d77 100644 --- a/src/kernel.cr +++ b/src/kernel.cr @@ -599,6 +599,10 @@ end end {% end %} +{% if flag?(:win32) && 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 From d25394be21f687b7ff2ef6194ceacbad2e5a656f Mon Sep 17 00:00:00 2001 From: Julien Portalier Date: Thu, 2 Jul 2026 14:13:03 +0200 Subject: [PATCH 2/2] Fix: compatibility with released execution contexts --- src/kernel.cr | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/kernel.cr b/src/kernel.cr index 93707e729d77..069d6d2ef70c 100644 --- a/src/kernel.cr +++ b/src/kernel.cr @@ -599,7 +599,7 @@ end end {% end %} -{% if flag?(:win32) && flag?(:execution_context) %} +{% if flag?(:win32) && (!flag?(:without_mt) && !flag?(:preview_mt) || flag?(:execution_context)) %} Crystal::EventLoop::IOCP.start_forwarder_thread {% end %}