From 1eed4850b075b1add54e809ef390c0e130323bf5 Mon Sep 17 00:00:00 2001 From: Tony Arnold Date: Sat, 3 Oct 2026 13:24:00 +1000 Subject: [PATCH 1/4] fix: support aliasing FlyingFox and FlyingSocks together --- Package.swift | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/Package.swift b/Package.swift index 6006052..63f4185 100644 --- a/Package.swift +++ b/Package.swift @@ -20,7 +20,11 @@ let package = Package( targets: [ .target( name: "FlyingFox", - dependencies: ["FlyingSocks"], + dependencies: [ + "FlyingSocks", + // Workaround for SwiftPM omitting dependent aliases when both Swift modules are aliased. + .target(name: "CSystemLinux", condition: .when(platforms: [.linux, .android])) + ], path: "FlyingFox/Sources", swiftSettings: .upcomingFeatures ), From aee3a820ed90328d8929618260cb0fde1d73b0c9 Mon Sep 17 00:00:00 2001 From: Tony Arnold Date: Sat, 3 Oct 2026 17:53:24 +1000 Subject: [PATCH 2/4] fix: initialize packet-info source addresses Apply the proposed fix from swhitty/FlyingFox#243 (49356ee87776c8e52f57af70ebe6f337d096c7ee) for Linux CI verification. --- FlyingSocks/Sources/Socket.swift | 32 ++++++++++++-------- FlyingSocks/Tests/SocketTests.swift | 46 +++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 12 deletions(-) diff --git a/FlyingSocks/Sources/Socket.swift b/FlyingSocks/Sources/Socket.swift index 2e63ff8..b458444 100644 --- a/FlyingSocks/Sources/Socket.swift +++ b/FlyingSocks/Sources/Socket.swift @@ -765,16 +765,20 @@ extension Socket { // match getPacketInfoControl above. header.pointee.cmsg_level = Socket.ipproto_ip header.pointee.cmsg_type = Socket.ip_pktinfo + // ManagedBuffer leaves the element uninitialized: zero it, so that no + // field is sent holding whatever was in memory. + element.initialize(to: in_pktinfo()) element.pointee.ipi_ifindex = IPv4InterfaceIndexType(interfaceIndex ?? 0) + // ip(7): "If IP_PKTINFO is passed to sendmsg(2) and ipi_spec_dst is not + // zero, then it is used as the local source address for the routing + // table lookup". ipi_addr is the destination of a received packet. if let address { - var address = address - withUnsafePointer(to: &address) { - $0.withMemoryRebound(to: sockaddr_in.self, capacity: 1) { - element.pointee.ipi_addr = $0.pointee.sin_addr + let storage = address.makeStorage() + if storage.ss_family == sa_family_t(AF_INET) { + element.pointee.ipi_spec_dst = withUnsafeBytes(of: storage) { + $0.load(as: sockaddr_in.self).sin_addr } } - } else { - element.pointee.ipi_addr.s_addr = 0 } return header.pointee @@ -791,16 +795,20 @@ extension Socket { // match getPacketInfoControl above. header.pointee.cmsg_level = Socket.ipproto_ipv6 header.pointee.cmsg_type = Socket.ipv6_pktinfo + // memberwise: Socket+Glibc declares its own in6_pktinfo, which has no + // zero initializer + element.initialize(to: in6_pktinfo(ipi6_addr: in6_addr(), ipi6_ifindex: 0)) element.pointee.ipi6_ifindex = IPv6InterfaceIndexType(interfaceIndex ?? 0) + // RFC 3542 §6.1: on output, ipi6_addr is the source address. Read it + // from the address's storage: a sockaddr_in6 is too large to be held + // inline in an existential, so the existential's own bytes are not it. if let address { - var address = address - withUnsafePointer(to: &address) { - $0.withMemoryRebound(to: sockaddr_in6.self, capacity: 1) { - element.pointee.ipi6_addr = $0.pointee.sin6_addr + let storage = address.makeStorage() + if storage.ss_family == sa_family_t(AF_INET6) { + element.pointee.ipi6_addr = withUnsafeBytes(of: storage) { + $0.load(as: sockaddr_in6.self).sin6_addr } } - } else { - element.pointee.ipi6_addr = in6_addr() } return header.pointee diff --git a/FlyingSocks/Tests/SocketTests.swift b/FlyingSocks/Tests/SocketTests.swift index bab2737..2b724e0 100644 --- a/FlyingSocks/Tests/SocketTests.swift +++ b/FlyingSocks/Tests/SocketTests.swift @@ -419,6 +419,52 @@ struct SocketTests { #expect(type == Int32(IPV6_PKTINFO)) #endif } + + @Test + func withPacketInfoControl_IP4_SetsSourceAddressInSpecDst() throws { + // ip(7): sendmsg(2) takes the source address from ipi_spec_dst; ipi_addr is the + // destination of a received packet. The address is also passed as a + // sockaddr_storage, which an existential holds boxed rather than inline. + let loopback = try sockaddr_in.inet(ip4: "127.0.0.1", port: 0) + for address in [loopback as any SocketAddress, loopback.makeStorage()] { + let info = Socket.withPacketInfoControl( + family: sa_family_t(AF_INET), + interfaceIndex: 7, + address: address + ) { header, _ in + UnsafeRawPointer(header!) + .advanced(by: MemoryLayout.size) + .loadUnaligned(as: in_pktinfo.self) + } + #expect(info.ipi_spec_dst.s_addr == loopback.sin_addr.s_addr) + #expect(info.ipi_addr.s_addr == 0) + #expect(info.ipi_ifindex == 7) + } + } + + @Test + func withPacketInfoControl_IP6_SetsSourceAddress() { + // RFC 3542 §6.1: on output, ipi6_addr is the source address. A sockaddr_in6 is too + // large to be held inline in an existential, so this also checks the address is + // read from the value, not from the existential's own storage. + let loopback = sockaddr_in6.loopback(port: 0) + for address in [loopback as any SocketAddress, loopback.makeStorage()] { + let info = Socket.withPacketInfoControl( + family: sa_family_t(AF_INET6), + interfaceIndex: 7, + address: address + ) { header, _ in + UnsafeRawPointer(header!) + .advanced(by: MemoryLayout.size) + .loadUnaligned(as: in6_pktinfo.self) + } + #expect( + withUnsafeBytes(of: info.ipi6_addr) { Array($0) } == + withUnsafeBytes(of: loopback.sin6_addr) { Array($0) } + ) + #expect(info.ipi6_ifindex == 7) + } + } #endif } From 92e96431dfa1fc3092590acd5988554a4efb1b08 Mon Sep 17 00:00:00 2001 From: Tony Arnold Date: Sat, 3 Oct 2026 17:53:25 +1000 Subject: [PATCH 3/4] fix: refresh reused socket descriptor registrations Apply the proposed fix from swhitty/FlyingFox#244 (f38b81227648e70cfc529af2677bae5fed2334ad) for Linux CI verification. --- FlyingSocks/Sources/SocketPool+ePoll.swift | 52 ++++++++------ FlyingSocks/Sources/SocketPool+kQueue.swift | 30 ++++----- FlyingSocks/Tests/SocketPoolTests.swift | 75 +++++++++++++++++++++ 3 files changed, 120 insertions(+), 37 deletions(-) diff --git a/FlyingSocks/Sources/SocketPool+ePoll.swift b/FlyingSocks/Sources/SocketPool+ePoll.swift index 1a134ec..2e9e842 100644 --- a/FlyingSocks/Sources/SocketPool+ePoll.swift +++ b/FlyingSocks/Sources/SocketPool+ePoll.swift @@ -107,7 +107,10 @@ public struct ePoll: EventQueue { public mutating func addEvents(_ events: Socket.Events, for socket: Socket.FileDescriptor) throws { var socketEvents = existing[socket] ?? [] socketEvents.formUnion(events) - try setEvents(socketEvents, for: socket) + // Tell the kernel even when the events are recorded as registered: a descriptor + // closed while registered has left the epoll set, and a socket since given the + // descriptor would otherwise never be registered. + try setEvents(socketEvents, for: socket, evenIfRecorded: true) } public mutating func removeEvents(_ events: Socket.Events, for socket: Socket.FileDescriptor) throws { @@ -118,32 +121,39 @@ public struct ePoll: EventQueue { try setEvents(socketEvents, for: socket) } - mutating func setEvents(_ events: Socket.Events, for socket: Socket.FileDescriptor) throws { - guard existing[socket] != events else { return } + mutating func setEvents( + _ events: Socket.Events, + for socket: Socket.FileDescriptor, + evenIfRecorded: Bool = false + ) throws { + guard evenIfRecorded || existing[socket] != events else { return } var event = CSystemLinux.epoll_event() event.events = events.epollEvents(triggerMode: triggerMode).rawValue event.data.fd = socket.rawValue - if existing[socket] != nil { - if events.isEmpty { - guard epoll_ctl(file.rawValue, EPOLL_CTL_DEL, socket.rawValue, &event) != -1 else { - throw SocketError.makeFailed("epoll_ctl EPOLL_CTL_DEL") - } - } else { - guard epoll_ctl(file.rawValue, EPOLL_CTL_MOD, socket.rawValue, &event) != -1 else { - throw SocketError.makeFailed("epoll_ctl EPOLL_CTL_MOD") - } - } - } else if !events.isEmpty { - guard epoll_ctl(file.rawValue, EPOLL_CTL_ADD, socket.rawValue, &event) != -1 else { - throw SocketError.makeFailed("epoll_ctl EPOLL_CTL_ADD") - } - } + let recorded = existing[socket] != nil + // Record the change before making it. If it fails because the descriptor was closed + // while registered, the epoll set has dropped it already. + existing[socket] = events.isEmpty ? nil : events if events.isEmpty { - existing[socket] = nil - } else { - existing[socket] = events + guard recorded else { return } + guard epoll_ctl(file.rawValue, EPOLL_CTL_DEL, socket.rawValue, &event) != -1 else { + throw SocketError.makeFailed("epoll_ctl EPOLL_CTL_DEL") + } + } else if epoll_ctl(file.rawValue, recorded ? EPOLL_CTL_MOD : EPOLL_CTL_ADD, socket.rawValue, &event) == -1 { + // The epoll set can disagree with the record once a descriptor closed while + // registered is reused: modifying it then fails with ENOENT, adding with EEXIST. + var retry: Int32? + if recorded, errno == ENOENT { + retry = EPOLL_CTL_ADD + } else if !recorded, errno == EEXIST { + retry = EPOLL_CTL_MOD + } + guard let retry, epoll_ctl(file.rawValue, retry, socket.rawValue, &event) != -1 else { + existing[socket] = nil + throw SocketError.makeFailed(recorded ? "epoll_ctl EPOLL_CTL_MOD" : "epoll_ctl EPOLL_CTL_ADD") + } } } diff --git a/FlyingSocks/Sources/SocketPool+kQueue.swift b/FlyingSocks/Sources/SocketPool+kQueue.swift index fff33f5..5faf3ee 100644 --- a/FlyingSocks/Sources/SocketPool+kQueue.swift +++ b/FlyingSocks/Sources/SocketPool+kQueue.swift @@ -76,12 +76,14 @@ public struct kQueue: EventQueue { public mutating func addEvents(_ events: Socket.Events, for socket: Socket.FileDescriptor) throws { for event in events { + // Add the event even when it is recorded as added: EV_ADD of a filter the kqueue + // has just updates it, but a descriptor closed while registered has left the + // kqueue, and skipping the add would leave a socket since given the descriptor + // never woken. + try addEvent(event, for: socket) var socketEvents = existing[socket] ?? [] - if !socketEvents.contains(event) { - try addEvent(event, for: socket) - socketEvents.insert(event) - existing[socket] = socketEvents - } + socketEvents.insert(event) + existing[socket] = socketEvents } } @@ -101,17 +103,13 @@ public struct kQueue: EventQueue { public mutating func removeEvents(_ events: Socket.Events, for socket: Socket.FileDescriptor) throws { for event in events { - if var entries = existing[socket] { - if entries.contains(event) { - try removeEvent(event, for: socket) - entries.remove(event) - if entries.isEmpty { - existing[socket] = nil - } else { - existing[socket] = entries - } - } - } + guard var entries = existing[socket], entries.contains(event) else { continue } + // Forget the event before removing it: a socket closed while registered has left + // the kqueue already, so the removal fails, and remembering the event would stop + // it being added for a socket later given the same descriptor. + entries.remove(event) + existing[socket] = entries.isEmpty ? nil : entries + try removeEvent(event, for: socket) } } diff --git a/FlyingSocks/Tests/SocketPoolTests.swift b/FlyingSocks/Tests/SocketPoolTests.swift index 7dcb94e..d99d369 100644 --- a/FlyingSocks/Tests/SocketPoolTests.swift +++ b/FlyingSocks/Tests/SocketPoolTests.swift @@ -175,6 +175,65 @@ struct SocketPoolTests { } } +#if canImport(Darwin) || canImport(CSystemLinux) + @Test + func reusedDescriptor_IsWoken_AfterTheSuspendedSocketIsReplacedAndCancelled() async throws { + let pool = SocketPool.make(maxEvents: 4) + try await pool.prepare() + let run = Task { try await pool.run() } + defer { run.cancel() } + + let (old, oldPeer) = try Socket.makeNonBlockingPair() + defer { try? oldPeer.close() } + let suspended = Task { try await pool.suspendSocket(old, untilReadyFor: .read) } + try await Task.sleep(seconds: 0.1) + + // closing the socket drops it from the queue; cancelling its suspension then fails + // to remove it, and a socket given the same descriptor must still be woken + let (reused, peer) = try Socket.makeNonBlockingPair(replacing: old) + defer { + try? reused.close() + try? peer.close() + } + suspended.cancel() + _ = try? await suspended.value + + _ = try peer.write(Data([1])) + let woken = Task(timeout: 1) { try await pool.suspendSocket(reused, untilReadyFor: .read) } + await #expect(throws: Never.self) { + try await woken.value + } + } + + @Test + func reusedDescriptor_IsWoken_WhileTheReplacedSocketIsStillSuspended() async throws { + let pool = SocketPool.make(maxEvents: 4) + try await pool.prepare() + let run = Task { try await pool.run() } + defer { run.cancel() } + + let (old, oldPeer) = try Socket.makeNonBlockingPair() + defer { try? oldPeer.close() } + let suspended = Task { try await pool.suspendSocket(old, untilReadyFor: .read) } + defer { suspended.cancel() } + try await Task.sleep(seconds: 0.1) + + // the socket is closed and its descriptor given to another before its suspension + // ends; the new socket's own suspension must still register with the queue + let (reused, peer) = try Socket.makeNonBlockingPair(replacing: old) + defer { + try? reused.close() + try? peer.close() + } + + _ = try peer.write(Data([1])) + let woken = Task(timeout: 1) { try await pool.suspendSocket(reused, untilReadyFor: .read) } + await #expect(throws: Never.self) { + try await woken.value + } + } +#endif + @Test func waiting_IsEmpty() async { let cn = await Continuation.make() @@ -253,6 +312,22 @@ struct SocketPoolTests { } } +#if canImport(Darwin) || canImport(CSystemLinux) +private extension Socket { + /// A connected pair whose first socket takes over `socket`'s descriptor, closing it, as + /// a new socket can be given the descriptor of one just closed. `dup2` closes and + /// replaces in one step, so no other socket can be given the descriptor in between. + static func makeNonBlockingPair(replacing socket: Socket) throws -> (Socket, Socket) { + let (first, second) = try Socket.makeNonBlockingPair() + guard dup2(first.file.rawValue, socket.file.rawValue) != -1 else { + throw SocketError.makeFailed("dup2") + } + try first.close() + return (Socket(file: socket.file), second) + } +} +#endif + private extension SocketPool where Queue == MockEventQueue { static func make() -> Self { .init(queue: MockEventQueue()) From 8a7181c854d037ff22f0bcc367bab0f39a1a3722 Mon Sep 17 00:00:00 2001 From: Tony Arnold Date: Sat, 3 Oct 2026 20:13:22 +1000 Subject: [PATCH 4/4] test: synchronize continuation cancellation scenarios --- .../Tests/IdentifiableContinuationTests.swift | 60 +++++++++++++++---- 1 file changed, 50 insertions(+), 10 deletions(-) diff --git a/FlyingSocks/Tests/IdentifiableContinuationTests.swift b/FlyingSocks/Tests/IdentifiableContinuationTests.swift index 7d62e1d..61fd802 100644 --- a/FlyingSocks/Tests/IdentifiableContinuationTests.swift +++ b/FlyingSocks/Tests/IdentifiableContinuationTests.swift @@ -68,7 +68,7 @@ struct IdentifiableContinuationAsyncTests { let waiter = Waiter() let task = await waiter.makeTask(onCancel: nil) - try? await Task.sleep(seconds: 0.1) + await waiter.waitUntilCreated() var isEmpty = await waiter.isEmpty #expect(!isEmpty) task.cancel() @@ -84,11 +84,12 @@ struct IdentifiableContinuationAsyncTests { func cancels_Before_Created() async { let waiter = Waiter() - let task = await waiter.makeTask(delay: 1.0, onCancel: nil) - try? await Task.sleep(seconds: 0.1) + let task = await waiter.makeTask(pauseBeforeCreating: true, onCancel: nil) + await waiter.waitUntilPaused() let isEmpty = await waiter.isEmpty #expect(isEmpty) task.cancel() + await waiter.resumeCreation() let val = await task.value #expect(val == nil) @@ -140,7 +141,7 @@ struct IdentifiableContinuationAsyncTests { let waiter = Waiter() let task = await waiter.makeTask(onCancel: .failure(CancellationError())) - try? await Task.sleep(seconds: 0.5) + await waiter.waitUntilCreated() var isEmpty = await waiter.isEmpty #expect(!isEmpty) task.cancel() @@ -158,11 +159,12 @@ struct IdentifiableContinuationAsyncTests { func throwingCancels_Before_Created() async { let waiter = Waiter() - let task = await waiter.makeTask(delay: 1.0, onCancel: .failure(CancellationError())) - try? await Task.sleep(seconds: 0.5) + let task = await waiter.makeTask(pauseBeforeCreating: true, onCancel: .failure(CancellationError())) + await waiter.waitUntilPaused() let isEmpty = await waiter.isEmpty #expect(isEmpty) task.cancel() + await waiter.resumeCreation() let result = await task.result #expect(throws: CancellationError.self) { @@ -175,14 +177,19 @@ private actor Waiter { typealias Continuation = IdentifiableContinuation private var waiting = [Continuation.ID: Continuation]() + private var creationObserver: CheckedContinuation? + private var pauseObserver: CheckedContinuation? + private var creationGate: CheckedContinuation? var isEmpty: Bool { waiting.isEmpty } - func makeTask(delay: TimeInterval = 0, onCancel: T) -> Task where E == Never { + func makeTask(pauseBeforeCreating: Bool = false, onCancel: T) -> Task where E == Never { Task { - try? await Task.sleep(seconds: delay) + if pauseBeforeCreating { + await pauseCreation() + } return await withIdentifiableContinuation { addContinuation($0) } onCancel: { id in @@ -191,9 +198,11 @@ private actor Waiter { } } - func makeTask(delay: TimeInterval = 0, onCancel: Result) -> Task where E == any Error { + func makeTask(pauseBeforeCreating: Bool = false, onCancel: Result) -> Task where E == any Error { Task { - try? await Task.sleep(seconds: delay) + if pauseBeforeCreating { + await pauseCreation() + } return try await withIdentifiableThrowingContinuation { addContinuation($0) } onCancel: { id in @@ -202,9 +211,40 @@ private actor Waiter { } } + func waitUntilCreated() async { + guard waiting.isEmpty else { return } + await withCheckedContinuation { + creationObserver = $0 + } + } + + func waitUntilPaused() async { + guard creationGate == nil else { return } + await withCheckedContinuation { + pauseObserver = $0 + } + } + + func resumeCreation() { + creationGate?.resume() + creationGate = nil + } + + private func pauseCreation() async { + // Cancellation must not release this gate: the before-creation tests resume + // the already-cancelled task so it still invokes the continuation API. + await withCheckedContinuation { + creationGate = $0 + pauseObserver?.resume() + pauseObserver = nil + } + } + private func addContinuation(_ continuation: Continuation) { assertIsolated() waiting[continuation.id] = continuation + creationObserver?.resume() + creationObserver = nil } private func resumeID(_ id: Continuation.ID, returning value: T) {