diff --git a/.changeset/swift-writer-off-main.md b/.changeset/swift-writer-off-main.md new file mode 100644 index 0000000000..7b114ffb54 --- /dev/null +++ b/.changeset/swift-writer-off-main.md @@ -0,0 +1,5 @@ +--- +'zero-swift': patch +--- + +The client's WebSocket frames, such as mutations and query changes, no longer wait for a busy main thread. The writer that sends them one at a time ran on the main actor and went back to it after every frame, so while the app kept the main thread busy each frame waited its turn: on CI, 107 frames took 30 seconds to go out. The writer now runs off the main actor and still sends the frames, and the close after them, in order. diff --git a/swift/Sources/Zero/Client/Scheduler.swift b/swift/Sources/Zero/Client/Scheduler.swift index 2027e12138..387c171c08 100644 --- a/swift/Sources/Zero/Client/Scheduler.swift +++ b/swift/Sources/Zero/Client/Scheduler.swift @@ -76,7 +76,7 @@ public final class SystemScheduler: Scheduler { return timer } - static func milliseconds(_ d: Duration) -> Double { + nonisolated static func milliseconds(_ d: Duration) -> Double { let (s, attoseconds) = d.components return Double(s) * 1000 + Double(attoseconds) / 1e15 } diff --git a/swift/Sources/Zero/Client/Socket.swift b/swift/Sources/Zero/Client/Socket.swift index 56e58b33fc..a0c719399b 100644 --- a/swift/Sources/Zero/Client/Socket.swift +++ b/swift/Sources/Zero/Client/Socket.swift @@ -89,9 +89,11 @@ final class TransportConnection: SocketConnection { return } // One writer keeps the frames in order; each send in a task of its - // own could overtake the one before. + // own could overtake the one before. It runs off the main actor: on + // it, every frame would wait for the main actor to come back from its + // send, so a busy main thread would hold up the frames. let log = self.log.flatMap { $0.level <= .debug ? $0 : nil } - Task { + Task.detached { for await frame in frames { switch frame { case .text(let text): diff --git a/swift/Tests/AppleTests/URLSessionTransportTests.swift b/swift/Tests/AppleTests/URLSessionTransportTests.swift index beac08c426..de927980f1 100644 --- a/swift/Tests/AppleTests/URLSessionTransportTests.swift +++ b/swift/Tests/AppleTests/URLSessionTransportTests.swift @@ -66,33 +66,161 @@ } } + /// When the client's sends ran and finished and the server got the + /// frames, timed off the main actor, which the client's writer runs on. + struct FrameTimes: Sendable { + let start = ContinuousClock.now + var started = 0 + var sent = 0 + var failures: [String] = [] + var lastSent: Duration = .zero + var slowestSend: Duration = .zero + var received = 0 + var lastReceived: Duration = .zero + var mainActorWait: Duration = .zero + var mainActorWaitAt: Duration = .zero + + var report: String { + func ms(_ d: Duration) -> Int { Int(d / .milliseconds(1)) } + return """ + The server got \(received) of 200 frames, the last at \(ms(lastReceived)) ms. \ + The client started \(started) sends and finished \(sent), the last at \ + \(ms(lastSent)) ms and the slowest in \(ms(slowestSend)) ms; failed: \(failures). \ + The main actor's longest wait was \(ms(mainActorWait)) ms, at \(ms(mainActorWaitAt)) ms. + """ + } + } + + struct TimedTransport: Transport { + let base: any Transport + let times: Locked + + func connect(url: String, secProtocol: String?) async throws -> any Socket { + TimedSocket(base: try await base.connect(url: url, secProtocol: secProtocol), times: times) + } + } + + final class TimedSocket: Socket { + let base: any Socket + let times: Locked + + init(base: any Socket, times: Locked) { + self.base = base + self.times = times + } + + func send(_ message: SocketMessage) async throws { + let began = ContinuousClock.now + times.withLock { $0.started += 1 } + do { + try await base.send(message) + } catch { + times.withLock { $0.failures.append("\(error)") } + throw error + } + let now = ContinuousClock.now + times.withLock { t in + t.sent += 1 + t.lastSent = now - t.start + t.slowestSend = max(t.slowestSend, now - began) + } + } + + func receive() async throws -> SocketMessage? { try await base.receive() } + + func close(code: Int, reason: String) async { await base.close(code: code, reason: reason) } + } + @Test @MainActor func keepsTheOrderOfFramesSentBeforeAndAfterTheOpen() async throws { let server = try WebSocketServer() let port = try await server.start() defer { server.stop() } - var events = server.events.makeAsyncIterator() + let times = Locked(FrameTimes()) let recorder = Recorder() - let factory = TransportSocketFactory(URLSessionTransport()) + let factory = TransportSocketFactory( + TimedTransport(base: URLSessionTransport(), times: times)) let connection = factory.open( url: "ws://127.0.0.1:\(port)/", secProtocol: nil, events: recorder) for i in 0..<200 { connection.send("\(i)") } - guard case .connected(let peer, _)? = await events.next() else { + + // Off the main actor, so that a busy main actor doesn't delay when the + // frames are counted. + let events = server.events + let reading = Task.detached { () -> (peer: WebSocketServer.Peer, texts: [String])? in + var events = events.makeAsyncIterator() + guard case .connected(let peer, _)? = await events.next() else { return nil } + var texts: [String] = [] + while texts.count < 200, case .message(_, .text(let text))? = await events.next() { + texts.append(text) + let now = ContinuousClock.now + times.withLock { t in + t.received += 1 + t.lastReceived = now - t.start + } + } + return (peer, texts) + } + let probe = Task.detached { + while !Task.isCancelled { + let asked = ContinuousClock.now + await MainActor.run {} + let waited = ContinuousClock.now - asked + times.withLock { t in + if waited > t.mainActorWait { + t.mainActorWait = waited + t.mainActorWaitAt = asked - t.start + } + } + try? await Task.sleep(for: .milliseconds(50)) + } + } + let watchdog = Task { + try await Task.sleep(for: .seconds(30)) + Issue.record("After 30 s: \(times.withLock { $0.report })") + } + defer { + probe.cancel() + watchdog.cancel() + } + guard let read = await reading.value else { Issue.record("not connected") return } - var texts: [String] = [] - while texts.count < 200, case .message(_, .text(let text))? = await events.next() { - texts.append(text) - } - #expect(texts == (0..<200).map(String.init)) + #expect(read.texts == (0..<200).map(String.init)) #expect(recorder.opened) - peer.close(code: 4003, reason: "done") + read.peer.close(code: 4003, reason: "done") while recorder.closed == nil { try await Task.sleep(for: .milliseconds(10)) } #expect(recorder.closed == SocketClosed(code: 4003, reason: "done", wasClean: true)) } + @Test @MainActor func sendsWhileTheMainActorIsBusy() async throws { + let server = try WebSocketServer() + let port = try await server.start() + defer { server.stop() } + let events = server.events + let received = Locked(0) + let reading = Task.detached { + for await event in events { + guard case .message = event else { continue } + received.withLock { $0 += 1 } + } + } + defer { reading.cancel() } + + let recorder = Recorder() + let connection = TransportSocketFactory(URLSessionTransport()).open( + url: "ws://127.0.0.1:\(port)/", secProtocol: nil, events: recorder) + defer { connection.close(code: nil) } + while !recorder.opened { try await Task.sleep(for: .milliseconds(10)) } + + for i in 0..<200 { connection.send("\(i)") } + // The main actor stays busy; the frames still go out. + usleep(1_000_000) + #expect(received.withLock { $0 } == 200) + } + @Test @MainActor func connectsAClientThroughTheDefaultFactory() async throws { let server = try WebSocketServer() let port = try await server.start()