Support synchronous socket operations on Windows

src/crystal/system/win32/event_loop_iocp.cr

@@ -22,6 +22,15 @@ class Crystal::Iocp::EventLoop < Crystal::EventLoop
     if iocp.null?
       raise IO::Error.from_winerror("CreateIoCompletionPort")
     end
+    if parent
+      # all overlapped operations may finish synchronously, in which case we do
+      # not reschedule the running fiber; the following call tells Win32 not to
+      # queue an I/O completion packet to the associated IOCP as well, as this
+      # would be done by default
+      if LibC.SetFileCompletionNotificationModes(handle, LibC::FILE_SKIP_COMPLETION_PORT_ON_SUCCESS) == 0
+        raise IO::Error.from_winerror("SetFileCompletionNotificationModes")
+      end
+    end
     iocp
   end
 

src/crystal/system/win32/socket.cr

@@ -110,20 +110,7 @@ module Crystal::System::Socket
 
     error = overlapped_connect(fd, "ConnectEx") do |overlapped|
       # This is: LibC.ConnectEx(fd, addr, addr.size, nil, 0, nil, overlapped)
-      result = Crystal::System::Socket.connect_ex.call(fd, addr.to_unsafe, addr.size, Pointer(Void).null, 0_u32, Pointer(UInt32).null, overlapped)
-
-      if result.zero?
-        wsa_error = WinError.wsa_value
-
-        case wsa_error
-        when .wsa_io_pending?
-          next
-        when .wsaeaddrnotavail?
-          return yield ::Socket::ConnectError.from_os_error("ConnectEx", wsa_error)
-        else
-          return yield ::Socket::Error.from_os_error("ConnectEx", wsa_error)
-        end
-      end
+      Crystal::System::Socket.connect_ex.call(fd, addr.to_unsafe, addr.size, Pointer(Void).null, 0_u32, Pointer(UInt32).null, overlapped)
     end
 
     if error
@@ -145,7 +132,21 @@ module Crystal::System::Socket
 
   private def overlapped_connect(socket, method, &)
     OverlappedOperation.run(socket) do |operation|
-      yield operation.start
+      result = yield operation.start
+
+      if result == 0
+        case error = WinError.wsa_value
+        when .wsa_io_pending?
+          # the operation is running asynchronously; do nothing
+        when .wsaeaddrnotavail?
+          return ::Socket::ConnectError.from_os_error("ConnectEx", error)
+        else
+          return ::Socket::Error.from_os_error("ConnectEx", error)
+        end
+      else
+        operation.synchronous = true
+        return nil
+      end
 
       schedule_overlapped(read_timeout || 1.seconds)
 
@@ -201,19 +202,11 @@ module Crystal::System::Socket
     output_buffer = Bytes.new(address_size * 2 + buffer_size)
 
     success = overlapped_accept(fd, "AcceptEx") do |overlapped|
+      # This is: LibC.AcceptEx(fd, client_socket, output_buffer, buffer_size, address_size, address_size, out received_bytes, overlapped)
       received_bytes = uninitialized UInt32
-
-      result = Crystal::System::Socket.accept_ex.call(fd, client_socket,
+      Crystal::System::Socket.accept_ex.call(fd, client_socket,
         output_buffer.to_unsafe.as(Void*), buffer_size.to_u32!,
         address_size.to_u32!, address_size.to_u32!, pointerof(received_bytes), overlapped)
-
-      if result.zero?
-        error = WinError.wsa_value
-
-        unless error.wsa_io_pending?
-          return false
-        end
-      end
     end
 
     return false unless success
@@ -228,7 +221,19 @@ module Crystal::System::Socket
 
   private def overlapped_accept(socket, method, &)
     OverlappedOperation.run(socket) do |operation|
-      yield operation.start
+      result = yield operation.start
+
+      if result == 0
+        case error = WinError.wsa_value
+        when .wsa_io_pending?
+          # the operation is running asynchronously; do nothing
+        else
+          return false
+        end
+      else
+        operation.synchronous = true
+        return true
+      end
 
       unless schedule_overlapped(read_timeout)
         raise IO::TimeoutError.new("accept timed out")
@@ -256,21 +261,23 @@ module Crystal::System::Socket
     wsabuf = wsa_buffer(message)
 
     bytes = overlapped_write(fd, "WSASend") do |overlapped|
-      LibC.WSASend(fd, pointerof(wsabuf), 1, out bytes_sent, 0, overlapped, nil)
+      ret = LibC.WSASend(fd, pointerof(wsabuf), 1, out bytes_sent, 0, overlapped, nil)
+      {ret, bytes_sent}
     end
 
     bytes.to_i32
   end
 
   private def system_send_to(bytes : Bytes, addr : ::Socket::Address)
     wsabuf = wsa_buffer(bytes)
-    bytes_sent = overlapped_write(fd, "WSASendTo") do |overlapped|
-      LibC.WSASendTo(fd, pointerof(wsabuf), 1, out bytes_sent, 0, addr, addr.size, overlapped, nil)
+    bytes_written = overlapped_write(fd, "WSASendTo") do |overlapped|
+      ret = LibC.WSASendTo(fd, pointerof(wsabuf), 1, out bytes_sent, 0, addr, addr.size, overlapped, nil)
+      {ret, bytes_sent}
     end
-    raise ::Socket::Error.from_errno("Error sending datagram to #{addr}") if bytes_sent == -1
+    raise ::Socket::Error.from_errno("Error sending datagram to #{addr}") if bytes_written == -1
 
     # to_i32 is fine because string/slice sizes are an Int32
-    bytes_sent.to_i32
+    bytes_written.to_i32
   end
 
   private def system_receive(bytes)
@@ -286,7 +293,8 @@ module Crystal::System::Socket
 
     flags = 0_u32
     bytes_read = overlapped_read(fd, "WSARecvFrom") do |overlapped|
-      LibC.WSARecvFrom(fd, pointerof(wsabuf), 1, out bytes_received, pointerof(flags), sockaddr, pointerof(addrlen), overlapped, nil)
+      ret = LibC.WSARecvFrom(fd, pointerof(wsabuf), 1, out bytes_received, pointerof(flags), sockaddr, pointerof(addrlen), overlapped, nil)
+      {ret, bytes_received}
     end
 
     {bytes_read.to_i32, sockaddr, addrlen}
@@ -443,18 +451,21 @@ module Crystal::System::Socket
 
     bytes_read = overlapped_read(fd, "WSARecv", connreset_is_error: false) do |overlapped|
       flags = 0_u32
-      LibC.WSARecv(fd, pointerof(wsabuf), 1, out bytes_received, pointerof(flags), overlapped, nil)
+      ret = LibC.WSARecv(fd, pointerof(wsabuf), 1, out bytes_received, pointerof(flags), overlapped, nil)
+      {ret, bytes_received}
     end
+
     bytes_read.to_i32
   end
 
   private def unbuffered_write(slice : Bytes)
     wsabuf = wsa_buffer(slice)
 
     bytes = overlapped_write(fd, "WSASend") do |overlapped|
-      LibC.WSASend(fd, pointerof(wsabuf), 1, out bytes_sent, 0, overlapped, nil)
+      ret = LibC.WSASend(fd, pointerof(wsabuf), 1, out bytes_sent, 0, overlapped, nil)
+      {ret, bytes_sent}
     end
-    # we could return bytes (from WSAGetOverlappedResult) or bytes_sent
+
     bytes.to_i32
   end
 

src/io/overlapped.cr

@@ -152,9 +152,9 @@ module IO::Overlapped
         # Microsoft documentation:
         # The application must not free or reuse the OVERLAPPED structure
         # associated with the canceled I/O operations until they have completed
-        # (this applies even to asynchronous operations that completed
-        # synchronously)
-        if synchronous? || LibC.CancelIoEx(handle, pointerof(@overlapped)) != 0
+        # (this does not apply to asynchronous operations that finished
+        # synchronously, as nothing would be queued to the IOCP)
+        if !synchronous? && LibC.CancelIoEx(handle, pointerof(@overlapped)) != 0
           @state = :cancelled
           @@canceled.push(self) # to increase lifetime
         end
@@ -217,14 +217,18 @@ module IO::Overlapped
 
   def wsa_overlapped_operation(socket, method, timeout, connreset_is_error = true, &)
     OverlappedOperation.run(socket) do |operation|
-      result = yield operation.start
+      result, value = yield operation.start
 
       if result == LibC::SOCKET_ERROR
-        error = WinError.wsa_value
-
-        unless error.wsa_io_pending?
+        case error = WinError.wsa_value
+        when .wsa_io_pending?
+          # the operation is running asynchronously; do nothing
+        else
           raise IO::Error.from_os_error(method, error)
         end
+      else
+        operation.synchronous = true
+        return value
       end
 
       schedule_overlapped(timeout)

src/lib_c/x86_64-windows-msvc/c/winbase.cr

@@ -25,6 +25,10 @@ lib LibC
   fun CreateNamedPipeA(lpName : LPSTR, dwOpenMode : DWORD, dwPipeMode : DWORD, nMaxInstances : DWORD,
                        nOutBufferSize : DWORD, nInBufferSize : DWORD, nDefaultTimeOut : DWORD, lpSecurityAttributes : SECURITY_ATTRIBUTES*) : HANDLE
 
+  FILE_SKIP_COMPLETION_PORT_ON_SUCCESS = 1_u8
+
+  fun SetFileCompletionNotificationModes(fileHandle : HANDLE, flags : UChar) : BOOL
+
   fun GetEnvironmentVariableW(lpName : LPWSTR, lpBuffer : LPWSTR, nSize : DWORD) : DWORD
   fun GetEnvironmentStringsW : LPWCH
   fun FreeEnvironmentStringsW(lpszEnvironmentBlock : LPWCH) : BOOL