diff --git a/GraphcodeKit/Sources/IPC/OutboundChannel.swift b/GraphcodeKit/Sources/IPC/OutboundChannel.swift index 23479667..cda9805d 100644 --- a/GraphcodeKit/Sources/IPC/OutboundChannel.swift +++ b/GraphcodeKit/Sources/IPC/OutboundChannel.swift @@ -74,6 +74,7 @@ final class OutboundChannel: @unchecked Sendable { isSocket = getsockopt(fileDescriptor, SOL_SOCKET, SO_TYPE, &socketType, &typeSize) == 0 Self.armAgainstSIGPIPE(fileDescriptor) + if isSocket { Self.boundSends(on: fileDescriptor) } // Captured strongly on purpose: the channel must outlive the registry's reference to // it, because the descriptor is closed by `pump` on its way out. A weak capture would // let a channel released without `close()` take its thread — and its unclosed @@ -92,6 +93,25 @@ final class OutboundChannel: @unchecked Sendable { /// The channel arms the descriptor itself rather than trusting whoever handed it over: /// `graphcoded` does set this on the sockets it accepts, but a writer that only survives /// because its caller remembered is a crash waiting for the one caller that does not. + /// `SO_SNDTIMEO`, one poll slice long — what actually makes a send return. + /// + /// On macOS a `send(2)` with `MSG_DONTWAIT` on a blocking AF_UNIX stream socket + /// **blocks anyway**: the flag is ignored (60 KB into a 4 KB peer sits inside the + /// syscall until the peer drains; Linux honours the flag). So the writer below parked + /// inside `send` on a peer that stopped reading and never reached its poll loop — and + /// `closeAndWait`, which relies on the writer noticing `isClosing` between slices, was + /// not bounded after all. With a send timeout the call returns what it wrote, or + /// `EAGAIN` when nothing went, after the slice: the shape the loop was written for. + /// It bounds sends only — the reader sharing this descriptor keeps its blocking reads, + /// which `O_NONBLOCK` could not promise since that flag lives on the open file + /// description both halves share. + private static func boundSends(on fileDescriptor: Int32) { + var slice = timeval( + tv_sec: 0, tv_usec: suseconds_t(Int(writabilityPollMilliseconds) * 1000)) + setsockopt( + fileDescriptor, SOL_SOCKET, SO_SNDTIMEO, &slice, socklen_t(MemoryLayout.size)) + } + private static func armAgainstSIGPIPE(_ fileDescriptor: Int32) { #if canImport(Darwin) var enabled: Int32 = 1 @@ -254,18 +274,21 @@ final class OutboundChannel: @unchecked Sendable { /// Writes one length-prefixed frame, returning false if the peer failed or the channel /// was closed part-way through. /// - /// Every write is non-blocking *per call* (`MSG_DONTWAIT`), waiting for writability - /// with a bounded `poll` rather than parking inside `write(2)` until the peer drains. - /// That is not a refinement, it is what lets `closeAndWait` promise to return: a thread - /// already blocked inside a write on a unix socket is **not** reliably woken by another - /// thread's `shutdown`, so a wedged peer could hang the disconnecting caller for ever — - /// the original bug moved one layer down, where it showed up as the full test suite - /// hanging on a channel whose client never read. Polling in slices lets the writer - /// notice `isClosing` by itself instead of depending on being interrupted. + /// Every call returns within one slice — because of the socket's send timeout + /// (`boundSends`), not the `MSG_DONTWAIT` it also passes, which macOS ignores on a + /// blocking unix socket — and then waits for writability with a bounded `poll` rather + /// than parking inside the syscall until the peer drains. That is what lets + /// `closeAndWait` promise to return: a thread already blocked inside a write on a unix + /// socket is **not** reliably woken by another thread's `shutdown`, so a wedged peer + /// could hang the disconnecting caller for ever — the original bug moved one layer + /// down, where it showed up as the full test suite hanging on a channel whose client + /// never read. Returning between slices lets the writer notice `isClosing` by itself + /// instead of depending on being interrupted. /// - /// `MSG_DONTWAIT` per call rather than `O_NONBLOCK` on the descriptor: that flag lives - /// on the open file description, which the daemon's *reader* shares, and a reader that - /// started returning `EAGAIN` would tear down every connection. + /// The flag stays for Linux, where it is honoured and where `SO_SNDTIMEO` is then + /// merely redundant. Neither is `O_NONBLOCK` on the descriptor: that lives on the open + /// file description, which the daemon's *reader* shares, and a reader that started + /// returning `EAGAIN` would tear down every connection. private func writeFrame(_ data: Data) -> Bool { let length = UInt32(data.count) var buffer = Data(capacity: 4 + data.count) diff --git a/graphcode/Tests/OutboundChannelBoundedSendTests.swift b/graphcode/Tests/OutboundChannelBoundedSendTests.swift new file mode 100644 index 00000000..1724230c --- /dev/null +++ b/graphcode/Tests/OutboundChannelBoundedSendTests.swift @@ -0,0 +1,81 @@ +import Foundation +import Testing + +@testable import GraphcodeKit + +#if canImport(Darwin) + import Darwin +#endif + +/// A writer on a peer that never reads must still return between slices, or closing +/// that connection hangs the caller — the promise #291 made for `closeAndWait`. On macOS +/// `MSG_DONTWAIT` does not keep it (the flag is ignored on a blocking unix socket); the +/// socket's send timeout does. +@Suite +struct OutboundChannelBoundedSendTests { + private func deafPair() -> (daemon: Int32, peer: Int32) { + var pair: [Int32] = [0, 0] + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0) + var small: Int32 = 4096 + setsockopt(pair[0], SOL_SOCKET, SO_SNDBUF, &small, socklen_t(MemoryLayout.size)) + setsockopt(pair[1], SOL_SOCKET, SO_RCVBUF, &small, socklen_t(MemoryLayout.size)) + return (pair[0], pair[1]) + } + + /// `closeAndWait` after `shutdown` is bounded either way — `shutdown` wakes a send + /// parked on a unix socket. The path that cannot shut the socket down is `detach`, + /// for a descriptor number a new connection has already taken over: there the writer + /// must notice `isClosing` between slices on its own, and a writer parked inside + /// `send(2)` never does. Without the send timeout this waits for the peer's lifetime. + @Test + func aDetachedWriterWedgedOnADeafPeerRetiresWithinABoundedTime() async throws { + let (daemon, peer) = deafPair() + defer { + close(daemon) + close(peer) + } + let channel = OutboundChannel(fileDescriptor: daemon) + #expect(channel.send(Data(repeating: 0x78, count: 60_000))) + // Let the writer fill the peer and park. + try await Task.sleep(for: .milliseconds(300)) + + let started = Date() + let retired = Task.detached { + channel.detach() + channel.closeAndWait() + } + let outcome = await withTaskGroup(of: Bool.self) { group in + group.addTask { + await retired.value + return true + } + group.addTask { + try? await Task.sleep(for: .seconds(3)) + return false + } + let first = await group.next() ?? false + group.cancelAll() + return first + } + #expect(outcome, "a detached writer did not retire within 3 s of a wedged peer") + #expect(Date().timeIntervalSince(started) < 3) + } + + @Test + func aSlowReaderStillGetsTheWholeFrame() async throws { + let (daemon, peer) = deafPair() + defer { + OutboundChannels.close(daemon) + close(peer) + } + OutboundChannels.open(daemon) + let payload = Data(repeating: 0x79, count: 20_000) + #expect(OutboundChannels.send(payload, to: daemon)) + // Drain slowly, off the writer's thread, and check the frame arrives intact. + let received = await Task.detached { () -> Data? in + try? await Task.sleep(for: .milliseconds(200)) + return try? FramedMessageIO.readFrame(from: peer) + }.value + #expect(received == payload) + } +}