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..069d6d2ef70c 100644 --- a/src/kernel.cr +++ b/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