failed(type: "Connect", errno: 10037, message: "Unknown error") and WSAGetLastError() returns 10035(WSAEWOULDBLOCK)

[forchid]

The version info

Windows 10 x64, FlyingFox 0.26.2.

The test

>.build\release\echoperf
Iterate#0: time 15902ms.
Iterate#1: time 14144ms.
Iterate#2: time 14803ms.
Iterate#3: time 14585ms.
Iterate#4: time 15023ms.
Iterate#5: time 15197ms.
Iterate#6: time 15858ms.
Iterate#7: time 14494ms.
Iterate#8: time 15552ms.
Iterate#9: time 15628ms.
Error: failed(type: "Connect", errno: 10037, message: "Unknown error") - WSAGetLastError 10035
^C

The test source

import FlyingSocks

import Foundation

// rps 600: perf poor!
let startMillis = Int64(Date().timeIntervalSince1970 * 1000)

let closeServer = true
let host = "127.0.0.1"
let port = UInt16(3300)
let address: SocketAddress = try .inet(ip4: host, port: port)
let logger = DisabledLogger.disabled

let serPool: SocketPool<Poll> = .poll(logger: logger)
try await serPool.prepare() // Note: here very import!
let cliPool: SocketPool<Poll> = .poll(logger: logger)
try await cliPool.prepare() // Note: here very import!
Task { try await cliPool.run() }

let server = try makeSocketAndListen(on: address)
defer { try? server.close(); }
let serTask = Task {
    try await start(on: server, pool: serPool)
}

let iters = 100
let conns = 10
let reqs = 1000
let bufsize = 4096
for iter in 0 ..< iters {
    let startMs = Int64(Date().timeIntervalSince1970 * 1000)
    await withThrowingTaskGroup(of: Void.self) { group in
        for _ in 0 ..< conns {
            group.addTask {
                do {
                    let cli = try await AsyncSocket.connected(to: address, pool: cliPool)
                    defer { try? cli.close(); }
                    
                    let buf = Array(repeating: UInt8(96), count: bufsize)
                    let data = Data(buf)
                    for _ in 0 ..< reqs {
                        try await cli.write(data)
                        //print("Client: send ok.")
                        let _ = try await cli.read(bytes: bufsize)
                        //print("Client: read ok.")
                    }
                } catch {
                    print("Error: \(error) - WSAGetLastError \(WSAGetLastError())")
                }
            }
        }
    }
    
    let endMs = Int64(Date().timeIntervalSince1970 * 1000)
    print("Iterate#\(iter): time \(endMs - startMs)ms.")
}
if closeServer {
    serTask.cancel()
}

let endMillis = Int64(Date().timeIntervalSince1970 * 1000)
let reqTotal = Int64(conns * reqs * iters)
let duration = endMillis - startMillis
print("Total: reqs \(reqTotal), time \(duration)ms, rps \(reqTotal/(duration/1000)).")

if !closeServer {
    try await serTask.value
}

func handleConnection(_ socket: AsyncSocket) async throws {
    defer { try? socket.close() }
    do {
        while true {
            //print("Server: reading ..")
            let data = try await socket.read(bytes: bufsize)
            //print("Server: read ok.")
            try await socket.write(Data(data))
            //print("Server: send ok.")
        }
    } catch SocketError.disconnected {
        // Ignore
    }
}

func makeSocketAndListen(on addr: SocketAddress) throws -> Socket {
    let socket = try Socket(domain: Int32(type(of: addr).family))

    if addr.family != AF_UNIX {
        try socket.setValue(true, for: .exclusiveLocalAddressReuse)
    }
    
    try socket.bind(to: addr)
    try socket.listen()
    
    return socket
}

func start(on socket: Socket, pool: some AsyncSocketPool) async throws {
    let asyncSocket = try AsyncSocket(socket: socket, pool: pool)

    try await withThrowingTaskGroup(of: Void.self) { group in
        group.addTask {
            try await pool.run()
        }
        group.addTask {
            try await listenForConnections(on: asyncSocket)
        }
        try await group.next()
    }
}

func listenForConnections(on socket: AsyncSocket) async throws {
    for try await socket in socket.sockets {
        Task {
            try? await handleConnection(socket)
        }
    }
    throw SocketError.disconnected
}

[forchid]

And the FlyingFox current version also can't WSAEALREADY in Sockect.connect().