diff --git a/.github/workflows/swift.yml b/.github/workflows/swift.yml index 5cb71af..f39559a 100644 --- a/.github/workflows/swift.yml +++ b/.github/workflows/swift.yml @@ -5,7 +5,9 @@ jobs: macos: name: macOS runs-on: macos-15 + timeout-minutes: 20 strategy: + fail-fast: false matrix: config: ["debug", "release"] options: ["", "SWIFT_BUILD_DYNAMIC_LIBRARY=1"] @@ -17,16 +19,20 @@ jobs: - name: Build run: ${{ matrix.options }} swift build -c ${{ matrix.config }} - name: Test - run: ${{ matrix.options }} swift test -c ${{ matrix.config }} + # serially, sockets share the process file descriptor table and the idle + # benchmark measures CPU for the whole process + run: ${{ matrix.options }} swift test -c ${{ matrix.config }} --no-parallel linux: name: Linux strategy: + fail-fast: false matrix: container: ["swift:6.0.3", "swift:6.1.2", "swift:6.2.3"] config: ["debug", "release"] options: ["", "SWIFT_BUILD_DYNAMIC_LIBRARY=1"] runs-on: ubuntu-latest + timeout-minutes: 20 container: ${{ matrix.container }}-jammy steps: - name: Checkout @@ -35,13 +41,15 @@ jobs: run: swift --version - name: Build run: ${{ matrix.options }} swift build -c ${{ matrix.config }} + - name: Test + run: ${{ matrix.options }} swift test -c ${{ matrix.config }} --no-parallel android-arm: name: Android strategy: fail-fast: false matrix: - swift: ['6.2.3', 'nightly-6.3'] + swift: ['6.3.3'] arch: ['aarch64', 'x86_64', 'armv7'] runs-on: macos-15 timeout-minutes: 30 diff --git a/Package.swift b/Package.swift index d624d30..051f681 100644 --- a/Package.swift +++ b/Package.swift @@ -40,6 +40,9 @@ var package = Package( name: "SystemPackage", package: "swift-system" ), + ], + swiftSettings: [ + .define("ENABLE_MOCKING", .when(configuration: .debug)) ] ), .target( @@ -53,6 +56,9 @@ var package = Package( name: "Logging", package: "swift-log" ) + ], + swiftSettings: [ + .define("ENABLE_MOCKING", .when(configuration: .debug)) ] ) ] diff --git a/Sources/Socket/SocketManager/AsyncSocketManager.swift b/Sources/Socket/SocketManager/AsyncSocketManager.swift index 2fe3ff0..6297507 100644 --- a/Sources/Socket/SocketManager/AsyncSocketManager.swift +++ b/Sources/Socket/SocketManager/AsyncSocketManager.swift @@ -42,15 +42,38 @@ extension AsyncSocketConfiguration: SocketManagerConfiguration { } } +#if os(Linux) || os(Android) +internal typealias PlatformEventQueue = EpollEventQueue +#elseif canImport(Darwin) +internal typealias PlatformEventQueue = KqueueEventQueue +#else +internal typealias PlatformEventQueue = PollEventQueue +#endif + /// Async Socket Manager internal actor AsyncSocketManager: SocketManager { - + // MARK: - Properties - + fileprivate var state = ManagerState() - + + fileprivate var eventQueue: PlatformEventQueue? + + /// Events every socket is registered for. + /// + /// Write readiness is not included. A connected socket is almost always writable, so a + /// standing registration would make every wait return immediately for every idle socket. + /// It is added on demand while a write is pending, see ``addInterest(_:for:)``. + internal static let monitoredEvents: FileEvents = [ + .read, + .readUrgent, + .error, + .hangup, + .invalidRequest + ] + // MARK: - Initialization - + static let shared = AsyncSocketManager() private init() { } @@ -60,9 +83,14 @@ internal actor AsyncSocketManager: SocketManager { func add( _ fileDescriptor: SocketDescriptor ) -> Socket.Event.Stream { - guard state.sockets.keys.contains(fileDescriptor) == false else { - fatalError("Another socket for file descriptor \(fileDescriptor) already exists.") - } + // The kernel only hands back a descriptor number once the previous owner is gone, + // so any existing entry belongs to a socket that was closed without notifying us. + // Kernel event queues drop closed descriptors silently, unlike `poll(2)` reporting `POLLNVAL`. + if state.sockets.keys.contains(fileDescriptor) { + log("Discard stale socket \(fileDescriptor)") + discard(fileDescriptor) + } + state.detached.remove(fileDescriptor) log("Add socket \(fileDescriptor)") // make sure its non blocking do { @@ -76,6 +104,18 @@ internal actor AsyncSocketManager: SocketManager { log("Unable to set non blocking. \(error)") assertionFailure("Unable to set non blocking. \(error)") } + // register with kernel event queue + do { + if eventQueue == nil { + eventQueue = try PlatformEventQueue(maxEvents: 1024) + } + try eventQueue?.add(fileDescriptor, events: Self.monitoredEvents) + state.interests[fileDescriptor] = Self.monitoredEvents + } + catch { + log("Unable to register socket for events. \(error)") + assertionFailure("Unable to register socket for events. \(error)") + } // append socket with events continuation let eventStream = Socket.Event.Stream(bufferingPolicy: .bufferingNewest(1)) { continuation in state.sockets[fileDescriptor] = SocketState( @@ -90,12 +130,32 @@ internal actor AsyncSocketManager: SocketManager { } func remove(_ fileDescriptor: SocketDescriptor) { + if state.sockets[fileDescriptor] != nil { + log("Remove socket \(fileDescriptor)") + // deregister before closing, a closed descriptor cannot be deregistered by number + discard(fileDescriptor) + } else if state.detached.remove(fileDescriptor) == nil { + return // already closed by its owner + } + // close on behalf of the owner, including sockets left open by a hangup + try? fileDescriptor.close() + } + + /// Deregister a file descriptor and tear down its state without closing it. + /// + /// The manager never closes a descriptor it did not open. Closing one that the owner + /// still holds lets the kernel hand the same number to a new socket, and a later + /// ``remove(_:)`` would then close that unrelated socket instead. A descriptor kept + /// open this way is recorded in `detached` so its owner can still close it once. + private func discard(_ fileDescriptor: SocketDescriptor, detach: Bool = false) { guard let socket = state.sockets[fileDescriptor] else { - return // could have been removed previously + return } - log("Remove socket \(fileDescriptor)") - // close underlying socket - try? fileDescriptor.close() + if detach { + state.detached.insert(fileDescriptor) + } + try? eventQueue?.remove(fileDescriptor) + state.interests[fileDescriptor] = nil // cancel all pending actions Task(priority: .userInitiated) { await socket.dequeueAll(Errno.connectionAbort) @@ -228,7 +288,7 @@ private extension AsyncSocketManager { // poll let hasEvents = try poll(&tasks) // stop monitoring if no sockets - if state.pollDescriptors.isEmpty { + if state.sockets.isEmpty { state.isMonitoring = false } // wait for each task to complete @@ -252,7 +312,28 @@ private extension AsyncSocketManager { func contains(_ fileDescriptor: SocketDescriptor) -> Bool { return state.sockets.keys.contains(fileDescriptor) } - + + /// Subscribe to additional events for a registered socket. + func addInterest(_ events: FileEvents, for fileDescriptor: SocketDescriptor) { + guard let current = state.interests[fileDescriptor] else { return } + setInterest(current.union(events), for: fileDescriptor) + } + + /// Stop monitoring events that no longer have a pending operation. + func removeInterest(_ events: FileEvents, for fileDescriptor: SocketDescriptor) { + guard let current = state.interests[fileDescriptor] else { return } + setInterest(current.subtracting(events).union(Self.monitoredEvents), for: fileDescriptor) + } + + private func setInterest(_ events: FileEvents, for fileDescriptor: SocketDescriptor) { + // skip the syscall when the mask is unchanged + guard state.interests[fileDescriptor] != events else { return } + state.interests[fileDescriptor] = events + do { try eventQueue?.update(fileDescriptor, events: events) } + catch { log("Unable to update events for \(fileDescriptor). \(error)") } + } + + nonisolated func wait( for events: FileEvents, fileDescriptor: SocketDescriptor @@ -262,6 +343,8 @@ private extension AsyncSocketManager { guard await socket.pendingEvents.contains(events) == false else { return socket // execute immediately } + // subscribe to events that are not monitored by default, like write readiness + await addInterest(events, for: fileDescriptor) // store continuation to resume when event is polled try await withThrowingContinuation(for: fileDescriptor) { (continuation: SocketContinuation<(), Swift.Error>) -> () in // store pending continuation @@ -285,78 +368,59 @@ private extension AsyncSocketManager { /// Poll for events. @discardableResult func poll(_ tasks: inout [Task]) throws -> Bool { - // build poll descriptor array - let sockets = state.sockets - .lazy - .sorted(by: { $0.key.rawValue < $1.key.rawValue }) - state.pollDescriptors.removeAll(keepingCapacity: true) - state.pollDescriptors.reserveCapacity(sockets.count) - let events: FileEvents = [ - .read, - .readUrgent, - .write, - .error, - .hangup, - .invalidRequest - ] - for (fileDescriptor, _) in sockets { - let poll = SocketDescriptor.Poll( - socket: fileDescriptor, - events: events - ) - state.pollDescriptors.append(poll) - } - assert(state.pollDescriptors.count == sockets.count) - // poll sockets + guard state.sockets.isEmpty == false else { return false } + var hasEvents = false do { - try state.pollDescriptors.poll() + try eventQueue?.wait(timeout: 0) { buffer in + hasEvents = buffer.isEmpty == false + for readiness in buffer { + guard let socket = state.sockets[readiness.fileDescriptor] else { + continue // stale event for a removed descriptor + } + tasks.append(process(readiness, socket: socket)) + } + } } catch { log("Unable to poll for events. \(error.localizedDescription)") throw error } - // wait for concurrent handling - let hasEvents = state.pollDescriptors.contains(where: { $0.returnedEvents.isEmpty == false }) - if hasEvents { - for poll in state.pollDescriptors { - guard let state = state.sockets[poll.socket] else { - preconditionFailure() - continue - } - let task = process(poll, socket: state) - tasks.append(task) - } - } return hasEvents } - - func process(_ poll: SocketDescriptor.Poll, socket: AsyncSocketManager.SocketState) -> Task { + + func process(_ readiness: SocketReadiness, socket: AsyncSocketManager.SocketState) -> Task { Task(priority: state.configuration.monitorPriority) { - if poll.returnedEvents.contains(.read) { + if readiness.events.contains(.read) { await socket.event(.read, notification: socket.isListening ? .connection : .read) } - if poll.returnedEvents.contains(.write) { + if readiness.events.contains(.write) { await socket.event(.write, notification: .write) + // unsubscribe once nothing is waiting to write, the socket stays + // writable and would otherwise report readiness on every wait + if await socket.isWaiting(for: .write) == false { + removeInterest(.write, for: readiness.fileDescriptor) + } } - if poll.returnedEvents.contains(.invalidRequest) { - error(.badFileDescriptor, for: poll.socket) + if readiness.events.contains(.invalidRequest) { + error(.badFileDescriptor, for: readiness.fileDescriptor) } - if poll.returnedEvents.contains(.error) { - error(.connectionReset, for: poll.socket) + if readiness.events.contains(.error) { + error(.connectionReset, for: readiness.fileDescriptor) } - if poll.returnedEvents.contains(.hangup) { - hangup(poll.socket) + if readiness.events.contains(.hangup) { + hangup(readiness.fileDescriptor) } } } func error(_ error: Errno, for fileDescriptor: SocketDescriptor) { state.sockets[fileDescriptor]?.continuation.yield(.error(error)) - remove(fileDescriptor) + // stop monitoring but leave the descriptor open, see `discard(_:detach:)` + discard(fileDescriptor, detach: true) } - + func hangup(_ fileDescriptor: SocketDescriptor) { - remove(fileDescriptor) + discard(fileDescriptor, detach: true) } } @@ -463,6 +527,10 @@ fileprivate extension AsyncSocketManager.SocketState { } } + func isWaiting(for event: FileEvents) -> Bool { + eventContinuation[event, default: []].isEmpty == false + } + func queue(_ event: FileEvents, _ continuation: SocketContinuation<(), Error>) { guard pendingEvents.contains(event) == false else { continuation.resume() @@ -519,9 +587,13 @@ extension AsyncSocketManager { var configuration = AsyncSocketConfiguration() var sockets = [SocketDescriptor: SocketState]() - - var pollDescriptors = [SocketDescriptor.Poll]() - + + /// Events each socket is currently registered for. + var interests = [SocketDescriptor: FileEvents]() + + /// Sockets no longer monitored whose descriptor is still open. + var detached = Set() + var isMonitoring = false } diff --git a/Sources/Socket/SocketManager/EpollEventQueue.swift b/Sources/Socket/SocketManager/EpollEventQueue.swift new file mode 100644 index 0000000..3f43d90 --- /dev/null +++ b/Sources/Socket/SocketManager/EpollEventQueue.swift @@ -0,0 +1,149 @@ +// +// EpollEventQueue.swift +// Socket +// + +#if os(Linux) || os(Android) +import CSocket +#if canImport(Glibc) +import Glibc +#elseif canImport(Musl) +import Musl +#elseif canImport(Android) +import Android +#endif +import SystemPackage + +/// Event queue backed by `epoll(7)`. +/// +/// Registrations are level-triggered to match `poll(2)` semantics. +internal struct EpollEventQueue: EventQueue, @unchecked Sendable { + + static var isStateful: Bool { true } + + private let queueFileDescriptor: CInt + + private let wakeEvent: SocketDescriptor.Event + + private var eventBuffer: [CInterop.EPollEvent] + + private var readiness: [SocketReadiness] + + init(maxEvents: Int) throws(Errno) { + let queueFileDescriptor = system_epoll_create1(_EPOLL_CLOEXEC) + guard queueFileDescriptor != -1 else { + throw Errno(rawValue: system_errno) + } + self.queueFileDescriptor = queueFileDescriptor + do { + self.wakeEvent = try SocketDescriptor.Event(flags: [.nonBlocking, .closeOnExec]) + } catch { + _ = system_close(queueFileDescriptor) + throw error + } + self.eventBuffer = .init(repeating: .init(), count: max(1, maxEvents)) + self.readiness = [] + self.readiness.reserveCapacity(maxEvents) + // register event descriptor to interrupt a blocking wait + do { + try control(_EPOLL_CTL_ADD, wakeEvent.rawValue, mask: _EPOLLIN) + } catch { + _ = system_close(queueFileDescriptor) + try? wakeEvent.close() + throw error + } + } + + mutating func close() { + try? wakeEvent.close() + _ = system_close(queueFileDescriptor) + eventBuffer.removeAll() + readiness.removeAll() + } + + mutating func add(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) { + try control(_EPOLL_CTL_ADD, fileDescriptor.rawValue, mask: Self.mask(for: events)) + } + + mutating func update(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) { + try control(_EPOLL_CTL_MOD, fileDescriptor.rawValue, mask: Self.mask(for: events)) + } + + mutating func remove(_ fileDescriptor: SocketDescriptor) throws(Errno) { + do { + try control(_EPOLL_CTL_DEL, fileDescriptor.rawValue, mask: nil) + } catch { + // tolerate removing descriptors that are missing or already closed + guard error == .noSuchFileOrDirectory || error == .badFileDescriptor else { + throw error + } + } + } + + mutating func wait( + timeout: Int?, + _ body: (UnsafeBufferPointer) throws -> T + ) throws -> T { + var eventCount: CInt + repeat { + eventCount = eventBuffer.withUnsafeMutableBufferPointer { buffer in + system_epoll_wait( + queueFileDescriptor, + buffer.baseAddress!, + CInt(buffer.count), + CInt(timeout ?? -1) + ) + } + if eventCount == -1 { + let error = Errno(rawValue: system_errno) + guard error == .interrupted else { throw error } + } + } while eventCount == -1 + readiness.removeAll(keepingCapacity: true) + for index in 0 ..< Int(eventCount) { + let event = eventBuffer[index] + let fileDescriptor = event.data.fd + guard fileDescriptor != wakeEvent.rawValue else { + _ = try? wakeEvent.read() // drain wake counter + continue + } + var events = FileEvents() + if event.events & _EPOLLIN != 0 { events.insert(.read) } + if event.events & _EPOLLPRI != 0 { events.insert(.readUrgent) } + if event.events & _EPOLLOUT != 0 { events.insert(.write) } + if event.events & _EPOLLERR != 0 { events.insert(.error) } + if event.events & _EPOLLHUP != 0 { events.insert(.hangup) } + readiness.append( + .init( + fileDescriptor: SocketDescriptor(rawValue: fileDescriptor), + events: events + ) + ) + } + return try readiness.withUnsafeBufferPointer(body) + } + + func wake() throws(Errno) { + try wakeEvent.write(1) + } + + private func control(_ operation: CInt, _ fileDescriptor: CInt, mask: UInt32?) throws(Errno) { + var event = CInterop.EPollEvent() + if let mask { + event.events = mask + event.data.fd = fileDescriptor + } + guard system_epoll_ctl(queueFileDescriptor, operation, fileDescriptor, &event) != -1 else { + throw Errno(rawValue: system_errno) + } + } + + private static func mask(for events: FileEvents) -> UInt32 { + var mask: UInt32 = 0 + if events.contains(.read) { mask |= _EPOLLIN } + if events.contains(.readUrgent) { mask |= _EPOLLPRI } + if events.contains(.write) { mask |= _EPOLLOUT } + return mask + } +} +#endif diff --git a/Sources/Socket/SocketManager/EventQueue.swift b/Sources/Socket/SocketManager/EventQueue.swift new file mode 100644 index 0000000..5519421 --- /dev/null +++ b/Sources/Socket/SocketManager/EventQueue.swift @@ -0,0 +1,48 @@ +// +// EventQueue.swift +// Socket +// + +import SystemPackage + +/// A kernel-side registration of file descriptors and interest masks. +internal protocol EventQueue: Sendable { + + /// Whether this queue holds registration state across waits. + static var isStateful: Bool { get } + + /// Creates the underlying kernel queue. + init(maxEvents: Int) throws(Errno) + + /// Closes the underlying kernel queue. + mutating func close() + + /// Registers a file descriptor with the specified interest mask. + mutating func add(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) + + /// Updates the interest mask of a registered file descriptor. + mutating func update(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) + + /// Deregisters a file descriptor. + mutating func remove(_ fileDescriptor: SocketDescriptor) throws(Errno) + + /// Waits up to `timeout` milliseconds for events. + /// + /// A timeout of `0` polls and returns immediately, `nil` blocks indefinitely. + /// The buffer of ready descriptors is only valid for the duration of `body`. + mutating func wait( + timeout: Int?, + _ body: (UnsafeBufferPointer) throws -> T + ) throws -> T + + /// Interrupts an in-flight blocking ``wait(timeout:_:)``. + func wake() throws(Errno) +} + +/// Events reported by an ``EventQueue`` for a single file descriptor. +internal struct SocketReadiness: Equatable, Hashable, Sendable { + + let fileDescriptor: SocketDescriptor + + let events: FileEvents +} diff --git a/Sources/Socket/SocketManager/KqueueEventQueue.swift b/Sources/Socket/SocketManager/KqueueEventQueue.swift new file mode 100644 index 0000000..01e7418 --- /dev/null +++ b/Sources/Socket/SocketManager/KqueueEventQueue.swift @@ -0,0 +1,176 @@ +// +// KqueueEventQueue.swift +// Socket +// + +#if canImport(Darwin) +import Darwin +import SystemPackage + +/// Event queue backed by `kqueue(2)`. +/// +/// Registrations are level-triggered to match `poll(2)` semantics. +internal struct KqueueEventQueue: EventQueue, @unchecked Sendable { + + static var isStateful: Bool { true } + + private let queueFileDescriptor: CInt + + private var eventBuffer: [CInterop.KernelEvent] + + private var readiness: [SocketReadiness] + + private var readinessIndices: [SocketDescriptor: Int] + + init(maxEvents: Int) throws(Errno) { + let queueFileDescriptor = system_kqueue() + guard queueFileDescriptor != -1 else { + throw Errno(rawValue: system_errno) + } + self.queueFileDescriptor = queueFileDescriptor + self.eventBuffer = .init(repeating: .init(), count: max(1, maxEvents)) + self.readiness = [] + self.readinessIndices = [:] + self.readiness.reserveCapacity(maxEvents) + // register user event to interrupt a blocking wait + var wakeEvent = CInterop.KernelEvent( + ident: 0, + filter: Int16(EVFILT_USER), + flags: UInt16(EV_ADD | EV_CLEAR), + fflags: 0, + data: 0, + udata: nil + ) + guard system_kevent(queueFileDescriptor, &wakeEvent, 1, nil, 0, nil) != -1 else { + let error = Errno(rawValue: system_errno) + _ = system_close(queueFileDescriptor) + throw error + } + } + + mutating func close() { + _ = system_close(queueFileDescriptor) + eventBuffer.removeAll() + readiness.removeAll() + readinessIndices.removeAll() + } + + mutating func add(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) { + try apply(fileDescriptor, filter: EVFILT_READ, isDesired: events.contains(.read), isRegistered: false) + try apply(fileDescriptor, filter: EVFILT_WRITE, isDesired: events.contains(.write), isRegistered: false) + } + + mutating func update(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) { + try apply(fileDescriptor, filter: EVFILT_READ, isDesired: events.contains(.read), isRegistered: true) + try apply(fileDescriptor, filter: EVFILT_WRITE, isDesired: events.contains(.write), isRegistered: true) + } + + mutating func remove(_ fileDescriptor: SocketDescriptor) throws(Errno) { + try apply(fileDescriptor, filter: EVFILT_READ, isDesired: false, isRegistered: true) + try apply(fileDescriptor, filter: EVFILT_WRITE, isDesired: false, isRegistered: true) + } + + mutating func wait( + timeout: Int?, + _ body: (UnsafeBufferPointer) throws -> T + ) throws -> T { + var eventCount: CInt + repeat { + eventCount = eventBuffer.withUnsafeMutableBufferPointer { buffer in + if let timeout { + var timeSpec = timespec( + tv_sec: timeout / 1000, + tv_nsec: (timeout % 1000) * 1_000_000 + ) + return system_kevent(queueFileDescriptor, nil, 0, buffer.baseAddress, CInt(buffer.count), &timeSpec) + } else { + return system_kevent(queueFileDescriptor, nil, 0, buffer.baseAddress, CInt(buffer.count), nil) + } + } + if eventCount == -1 { + let error = Errno(rawValue: system_errno) + guard error == .interrupted else { throw error } + } + } while eventCount == -1 + readiness.removeAll(keepingCapacity: true) + readinessIndices.removeAll(keepingCapacity: true) + for index in 0 ..< Int(eventCount) { + let event = eventBuffer[index] + guard CInt(event.filter) != EVFILT_USER else { + continue // wake notification + } + let fileDescriptor = SocketDescriptor(rawValue: CInt(event.ident)) + var events = FileEvents() + switch CInt(event.filter) { + case EVFILT_READ: + events.insert(.read) + // report hangup only once pending data is drained, matching POLLHUP + if event.flags & UInt16(EV_EOF) != 0, event.data == 0 { + events.insert(.hangup) + } + case EVFILT_WRITE: + events.insert(.write) + if event.flags & UInt16(EV_EOF) != 0 { + events.insert(.hangup) + } + default: + break + } + if event.flags & UInt16(EV_ERROR) != 0 { + events.insert(.error) + } + // kqueue reports read and write as separate filters, coalesce per descriptor + if let existingIndex = readinessIndices[fileDescriptor] { + readiness[existingIndex] = .init( + fileDescriptor: fileDescriptor, + events: readiness[existingIndex].events.union(events) + ) + } else { + readinessIndices[fileDescriptor] = readiness.count + readiness.append(.init(fileDescriptor: fileDescriptor, events: events)) + } + } + return try readiness.withUnsafeBufferPointer(body) + } + + func wake() throws(Errno) { + var event = CInterop.KernelEvent( + ident: 0, + filter: Int16(EVFILT_USER), + flags: 0, + fflags: CUnsignedInt(NOTE_TRIGGER), + data: 0, + udata: nil + ) + guard system_kevent(queueFileDescriptor, &event, 1, nil, 0, nil) != -1 else { + throw Errno(rawValue: system_errno) + } + } + + private func apply( + _ fileDescriptor: SocketDescriptor, + filter: CInt, + isDesired: Bool, + isRegistered: Bool + ) throws(Errno) { + guard isDesired || isRegistered else { return } + var event = CInterop.KernelEvent( + ident: UInt(fileDescriptor.rawValue), + filter: Int16(filter), + flags: UInt16(isDesired ? EV_ADD : EV_DELETE), + fflags: 0, + data: 0, + udata: nil + ) + guard system_kevent(queueFileDescriptor, &event, 1, nil, 0, nil) != -1 else { + let error = Errno(rawValue: system_errno) + // tolerate removing filters that are missing or already closed + guard isDesired == false, + error == .noSuchFileOrDirectory || error == .badFileDescriptor else { + throw error + } + return + } + } +} +#endif diff --git a/Sources/Socket/SocketManager/PollEventQueue.swift b/Sources/Socket/SocketManager/PollEventQueue.swift new file mode 100644 index 0000000..0158afd --- /dev/null +++ b/Sources/Socket/SocketManager/PollEventQueue.swift @@ -0,0 +1,78 @@ +// +// PollEventQueue.swift +// Socket +// + +import SystemPackage + +/// Event queue backed by `poll(2)`. +/// +/// The kernel holds no state between waits, so the descriptor set +/// is submitted on every call to ``wait(timeout:_:)``. +internal struct PollEventQueue: EventQueue, Sendable { + + static var isStateful: Bool { false } + + private var descriptors: [SocketDescriptor.Poll] + + private var readiness: [SocketReadiness] + + init(maxEvents: Int) throws(Errno) { + self.descriptors = [] + self.readiness = [] + self.descriptors.reserveCapacity(maxEvents) + self.readiness.reserveCapacity(maxEvents) + } + + mutating func close() { + descriptors.removeAll() + readiness.removeAll() + } + + mutating func add(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) { + guard index(of: fileDescriptor) == nil else { + throw Errno.fileExists + } + descriptors.append(.init(socket: fileDescriptor, events: events)) + } + + mutating func update(_ fileDescriptor: SocketDescriptor, events: FileEvents) throws(Errno) { + guard let index = index(of: fileDescriptor) else { + throw Errno.noSuchFileOrDirectory + } + descriptors[index] = .init(socket: fileDescriptor, events: events) + } + + mutating func remove(_ fileDescriptor: SocketDescriptor) throws(Errno) { + guard let index = index(of: fileDescriptor) else { + throw Errno.noSuchFileOrDirectory + } + descriptors.remove(at: index) + } + + mutating func wait( + timeout: Int?, + _ body: (UnsafeBufferPointer) throws -> T + ) throws -> T { + descriptors.reset() + try descriptors.poll(timeout: timeout ?? -1) + readiness.removeAll(keepingCapacity: true) + for descriptor in descriptors where descriptor.returnedEvents.isEmpty == false { + readiness.append( + .init( + fileDescriptor: descriptor.socket, + events: descriptor.returnedEvents + ) + ) + } + return try readiness.withUnsafeBufferPointer(body) + } + + func wake() throws(Errno) { + // a zero-timeout wait never blocks, nothing to interrupt + } + + private func index(of fileDescriptor: SocketDescriptor) -> Int? { + descriptors.firstIndex(where: { $0.socket == fileDescriptor }) + } +} diff --git a/Sources/Socket/System/CInterop.swift b/Sources/Socket/System/CInterop.swift index fa5669f..7a79d8f 100644 --- a/Sources/Socket/System/CInterop.swift +++ b/Sources/Socket/System/CInterop.swift @@ -94,4 +94,12 @@ public extension CInterop { typealias InterfaceLinkedList = ifaddrs typealias IOControlID = CUnsignedLong + + #if canImport(Darwin) + /// The C `kevent` type + typealias KernelEvent = Darwin.kevent + #elseif os(Linux) || os(Android) + /// The C `epoll_event` type + typealias EPollEvent = epoll_event + #endif } diff --git a/Sources/Socket/System/Constants.swift b/Sources/Socket/System/Constants.swift index d96897d..aba9f1b 100644 --- a/Sources/Socket/System/Constants.swift +++ b/Sources/Socket/System/Constants.swift @@ -164,6 +164,41 @@ internal var _POLLHUP: CInt { POLLHUP } @_alwaysEmitIntoClient internal var _POLLNVAL: CInt { POLLNVAL } +#if os(Linux) || os(Android) +// epoll constants are part of the stable Linux kernel ABI and are hardcoded +// because Glibc, Musl and Bionic import them with different Swift types. + +@_alwaysEmitIntoClient +internal var _EPOLLIN: UInt32 { 0x001 } + +@_alwaysEmitIntoClient +internal var _EPOLLPRI: UInt32 { 0x002 } + +@_alwaysEmitIntoClient +internal var _EPOLLOUT: UInt32 { 0x004 } + +@_alwaysEmitIntoClient +internal var _EPOLLERR: UInt32 { 0x008 } + +@_alwaysEmitIntoClient +internal var _EPOLLHUP: UInt32 { 0x010 } + +@_alwaysEmitIntoClient +internal var _EPOLLRDHUP: UInt32 { 0x2000 } + +@_alwaysEmitIntoClient +internal var _EPOLL_CTL_ADD: CInt { 1 } + +@_alwaysEmitIntoClient +internal var _EPOLL_CTL_DEL: CInt { 2 } + +@_alwaysEmitIntoClient +internal var _EPOLL_CTL_MOD: CInt { 3 } + +@_alwaysEmitIntoClient +internal var _EPOLL_CLOEXEC: CInt { 0o2000000 } +#endif + @_alwaysEmitIntoClient internal var _INET_ADDRSTRLEN: CInt { INET_ADDRSTRLEN } diff --git a/Sources/Socket/System/Mocking.swift b/Sources/Socket/System/Mocking.swift new file mode 100644 index 0000000..c7ba6c6 --- /dev/null +++ b/Sources/Socket/System/Mocking.swift @@ -0,0 +1,142 @@ +// +// Mocking.swift +// Socket +// + +#if canImport(Darwin) +import Darwin +#elseif os(Windows) +import ucrt +#elseif canImport(Glibc) +import Glibc +#elseif canImport(Musl) +import Musl +#elseif canImport(WASILibc) +import WASILibc +#elseif canImport(Android) +import Android +#endif + +// Syscall mocks, modeled after the mechanism in swift-system. + +/// A single recorded syscall invocation. +internal struct Trace { + + internal struct Entry: Hashable { + + internal var name: String + + internal var arguments: [AnyHashable] + + internal init(name: String, _ arguments: [AnyHashable]) { + self.name = name + self.arguments = arguments + } + } + + private var entries: [Entry] = [] + + private var firstUnchecked = 0 + + internal mutating func add(_ entry: Entry) { + entries.append(entry) + } + + internal var isEmpty: Bool { firstUnchecked >= entries.count } + + /// The next unchecked entry, advancing the cursor past it. + internal mutating func dequeue() -> Entry? { + guard isEmpty == false else { return nil } + defer { firstUnchecked += 1 } + return entries[firstUnchecked] + } + + internal var unchecked: ArraySlice { entries[firstUnchecked...] } + + internal var all: [Entry] { entries } +} + +/// An error to force upon the next syscall(s) instead of performing them. +internal enum ForceErrno: Equatable { + + case none + + case always(errno: CInt) + + case counted(errno: CInt, count: Int) +} + +/// Receives the syscall trace and supplies forced errors while mocking is enabled. +internal final class MockingDriver { + + internal var trace = Trace() + + internal var forceErrno = ForceErrno.none + + internal init() { } +} + +#if ENABLE_MOCKING +#if os(Windows) +private let key: _tls_index = { + let raw = FlsAlloc(nil) + guard raw != FLS_OUT_OF_INDEXES else { fatalError("Unable to create thread local storage") } + return raw +}() + +private func getDriver() -> MockingDriver? { + guard let raw = FlsGetValue(key) else { return nil } + return Unmanaged.fromOpaque(raw).takeUnretainedValue() +} + +private func setDriver(_ driver: MockingDriver?) { + let raw = driver.map { Unmanaged.passUnretained($0).toOpaque() } + FlsSetValue(key, raw) +} +#elseif canImport(WASILibc) +// WASI is single threaded, a global is equivalent to thread local storage. +nonisolated(unsafe) private var driver: MockingDriver? + +private func getDriver() -> MockingDriver? { driver } + +private func setDriver(_ newValue: MockingDriver?) { driver = newValue } +#else +private let key: pthread_key_t = { + var raw = pthread_key_t() + guard pthread_key_create(&raw, nil) == 0 else { + fatalError("Unable to create thread local storage") + } + return raw +}() + +private func getDriver() -> MockingDriver? { + guard let raw = pthread_getspecific(key) else { return nil } + return Unmanaged.fromOpaque(raw).takeUnretainedValue() +} + +private func setDriver(_ driver: MockingDriver?) { + let raw = driver.map { Unmanaged.passUnretained($0).toOpaque() } + pthread_setspecific(key, raw) +} +#endif + +/// The driver for the current thread, if mocking is enabled for it. +internal var currentMockingDriver: MockingDriver? { getDriver() } + +internal var contextualMockingEnabled: Bool { getDriver() != nil } + +extension MockingDriver { + + /// Enables mocking for the duration of `body`, which receives the driver + /// recording every syscall made on the current thread. + internal static func withMockingEnabled( + _ body: (MockingDriver) throws -> T + ) rethrows -> T { + let driver = MockingDriver() + let previous = getDriver() + setDriver(driver) + defer { setDriver(previous) } + return try body(driver) + } +} +#endif diff --git a/Sources/Socket/System/Syscalls.swift b/Sources/Socket/System/Syscalls.swift index 56b1b34..d0a4670 100644 --- a/Sources/Socket/System/Syscalls.swift +++ b/Sources/Socket/System/Syscalls.swift @@ -52,7 +52,7 @@ private func mockImpl( } var mockArgs: Array = [] if let p = path { - mockArgs.append(String(_errorCorrectingPlatformString: p)) + mockArgs.append(String(validatingPlatformString: p) ?? "") } mockArgs.append(contentsOf: args) driver.trace.add(Trace.Entry(name: origName, mockArgs)) @@ -384,6 +384,62 @@ internal func system_recvmsg( return recvmsg(socket, message, flags) } +#if canImport(Darwin) +internal func system_kqueue() -> CInt { +#if ENABLE_MOCKING + if mockingEnabled { return _mock() } +#endif + return kqueue() +} + +internal func system_kevent( + _ kq: CInt, + _ changes: UnsafePointer?, + _ changeCount: CInt, + _ events: UnsafeMutablePointer?, + _ eventCount: CInt, + _ timeout: UnsafePointer? +) -> CInt { +#if ENABLE_MOCKING + if mockingEnabled { return _mock(kq, changes, changeCount, events, eventCount, timeout) } +#endif + return kevent(kq, changes, changeCount, events, eventCount, timeout) +} +#endif + +#if os(Linux) || os(Android) +internal func system_epoll_create1(_ flags: CInt) -> CInt { +#if ENABLE_MOCKING + if mockingEnabled { return _mock(flags) } +#endif + return epoll_create1(flags) +} + +internal func system_epoll_ctl( + _ epoll: CInt, + _ operation: CInt, + _ fd: CInt, + _ event: UnsafeMutablePointer? +) -> CInt { +#if ENABLE_MOCKING + if mockingEnabled { return _mock(epoll, operation, fd, event) } +#endif + return epoll_ctl(epoll, operation, fd, event) +} + +internal func system_epoll_wait( + _ epoll: CInt, + _ events: UnsafeMutablePointer, + _ maxEvents: CInt, + _ timeout: CInt +) -> CInt { +#if ENABLE_MOCKING + if mockingEnabled { return _mock(epoll, events, maxEvents, timeout) } +#endif + return epoll_wait(epoll, events, maxEvents, timeout) +} +#endif + #if os(Linux) || os(Android) internal func system_eventfd( _ initval: CUnsignedInt, @@ -466,7 +522,10 @@ internal func system_ioctl( // if_nameindex internal func system_if_nameindex() -> UnsafeMutablePointer? { #if ENABLE_MOCKING - if mockingEnabled { return _mock() } + if mockingEnabled { + _ = _mock() + return nil + } #endif return if_nameindex() } @@ -474,7 +533,10 @@ internal func system_if_nameindex() -> UnsafeMutablePointer?) { #if ENABLE_MOCKING - if mockingEnabled { return _mock(pointer) } + if mockingEnabled { + _ = _mock(pointer) + return + } #endif return if_freenameindex(pointer) } @@ -488,7 +550,10 @@ internal func system_getifaddrs(_ pointer: UnsafeMutablePointer?) { #if ENABLE_MOCKING - if mockingEnabled { return _mock(pointer) } + if mockingEnabled { + _ = _mock(pointer) + return + } #endif return freeifaddrs(pointer) } @@ -496,14 +561,18 @@ internal func system_freeifaddrs(_ pointer: UnsafeMutablePointer, _ address: UnsafeMutablePointer) { #if ENABLE_MOCKING - if mockingEnabled { return _mock(cString) } + if mockingEnabled { + _ = _mock(cString) + return + } #endif return link_addr(cString, address) } internal func system_link_ntoa(_ address: UnsafePointer) -> UnsafeMutablePointer { #if ENABLE_MOCKING - if mockingEnabled { return _mock(cString) } + // records the call but still formats, there is no value to fabricate + if mockingEnabled { _ = _mock(address) } #endif return link_ntoa(address) } diff --git a/Tests/SocketTests/EventQueueTests.swift b/Tests/SocketTests/EventQueueTests.swift new file mode 100644 index 0000000..def7b0d --- /dev/null +++ b/Tests/SocketTests/EventQueueTests.swift @@ -0,0 +1,100 @@ +import Foundation +import Testing +import SystemPackage +@testable import Socket + +@Suite("EventQueue Tests") +struct EventQueueTests { + + @Test("Poll backend conformance") + func pollEventQueue() throws { + try Self.validate(PollEventQueue.self) + } + + #if canImport(Darwin) + @Test("Kqueue backend conformance") + func kqueueEventQueue() throws { + try Self.validate(KqueueEventQueue.self) + } + #endif + + #if os(Linux) || os(Android) + @Test("Epoll backend conformance") + func epollEventQueue() throws { + try Self.validate(EpollEventQueue.self) + } + #endif + + static func validate(_ queueType: Queue.Type) throws { + // connected TCP pair on loopback + let listener = try SocketDescriptor(IPv4Protocol.tcp) + defer { try? listener.close() } + try listener.bind(IPv4SocketAddress(address: .loopback, port: 0)) + try listener.listen(backlog: 1) + let address = try listener.address(IPv4SocketAddress.self) + let client = try SocketDescriptor(IPv4Protocol.tcp) + defer { try? client.close() } + try client.connect(to: address) + let server = try listener.accept() + // closed explicitly below to observe end of file, a second close would + // reap a descriptor number the kernel has since handed to another test + var isServerClosed = false + defer { if isServerClosed == false { try? server.close() } } + + var queue = try Queue(maxEvents: 16) + defer { queue.close() } + try queue.add(client, events: [.read, .write]) + + // connected socket is immediately writable, not readable + var events = try Self.events(for: client, in: &queue, timeout: 1000) + #expect(events?.contains(.write) == true) + #expect(events?.contains(.read) != true) + + // narrow interest mask so write readiness no longer reports + try queue.update(client, events: [.read]) + + // pending data surfaces read readiness + var byte = UInt8(0x42) + try withUnsafeBytes(of: &byte) { + _ = try server.write($0) + } + events = try Self.events(for: client, in: &queue, timeout: 1000) + #expect(events?.contains(.read) == true) + #expect(events?.contains(.write) != true) + + // level-triggered, readiness persists until drained + events = try Self.events(for: client, in: &queue, timeout: 0) + #expect(events?.contains(.read) == true) + var buffer = Data(count: 1) + _ = try buffer.withUnsafeMutableBytes { + try client.read(into: $0) + } + + // restoring the interest mask reports write readiness again + try queue.update(client, events: [.read, .write]) + events = try Self.events(for: client, in: &queue, timeout: 1000) + #expect(events?.contains(.write) == true) + try queue.update(client, events: [.read]) + + // peer close surfaces read readiness for end-of-file + try server.close() + isServerClosed = true + events = try Self.events(for: client, in: &queue, timeout: 1000) + #expect(events?.contains(.read) == true) + + // removed descriptor no longer reports events + try queue.remove(client) + events = try Self.events(for: client, in: &queue, timeout: 0) + #expect(events == nil) + } + + private static func events( + for socket: SocketDescriptor, + in queue: inout Queue, + timeout: Int + ) throws -> FileEvents? { + try queue.wait(timeout: timeout) { buffer in + buffer.first(where: { $0.fileDescriptor == socket })?.events + } + } +} diff --git a/Tests/SocketTests/IdleTests.swift b/Tests/SocketTests/IdleTests.swift new file mode 100644 index 0000000..962a224 --- /dev/null +++ b/Tests/SocketTests/IdleTests.swift @@ -0,0 +1,123 @@ +import Foundation +import Testing +import SystemPackage +@testable import Socket + +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#elseif canImport(Musl) +import Musl +#elseif canImport(Android) +import Android +#endif + +/// Idle sockets must not keep the monitor busy. +/// +/// A connected socket is almost always writable, so registering for write readiness +/// unconditionally makes every wait return every socket on every tick. +@Suite("Idle Tests", .serialized) +struct IdleTests { + + /// Microseconds of CPU allowed per socket per second while idle. + /// + /// Measured at ~2 with on demand write readiness and ~88 with a standing + /// registration, so this fails long before the old behavior returns. + static let budget = 20.0 + + static let duration = 5.0 + + @Test("Idle sockets report no events") + func idleSocketsReportNoEvents() throws { + let pairs = try SocketPairs(count: 8) + defer { pairs.close() } + var queue = try PlatformEventQueue(maxEvents: 64) + defer { queue.close() } + // exactly what the manager registers, write readiness is added on demand + for socket in pairs.sockets { + try queue.add(socket, events: AsyncSocketManager.monitoredEvents) + } + let ready = try queue.wait(timeout: 0) { $0.map(\.fileDescriptor) } + #expect(ready.isEmpty, "Idle sockets reported \(ready.count) events") + } + + @Test("Idle CPU stays within budget") + func idleCPU() async throws { + let pairs = try SocketPairs(count: Self.socketPairCount) + var sockets = [Socket]() + for fileDescriptor in pairs.sockets { + sockets.append(await Socket(fileDescriptor: fileDescriptor)) + } + + // let registration settle before measuring + try await Task.sleep(nanoseconds: 500_000_000) + + let start = Self.cpuSeconds() + try await Task.sleep(nanoseconds: UInt64(Self.duration * 1_000_000_000)) + let elapsed = Self.cpuSeconds() - start + + let perSocketSecond = elapsed / (Double(sockets.count) * Self.duration) * 1_000_000 + print("Idle CPU: \(String(format: "%.3f", elapsed))s for \(sockets.count) sockets, \(String(format: "%.1f", perSocketSecond))µs per socket second") + #expect( + perSocketSecond < Self.budget, + "Idle CPU \(perSocketSecond)µs per socket second exceeds budget of \(Self.budget)µs" + ) + + // close before returning so the next test starts from a clean table + for socket in sockets { + await socket.close() + } + pairs.closeListener() + } + + /// Sized to stay well inside the open file limit. + static let socketPairCount = 100 + + static func cpuSeconds() -> Double { + var usage = rusage() + #if canImport(Glibc) + // Glibc imports the constant as an enum + getrusage(__rusage_who_t(RUSAGE_SELF.rawValue), &usage) + #else + getrusage(RUSAGE_SELF, &usage) + #endif + let user = Double(usage.ru_utime.tv_sec) + Double(usage.ru_utime.tv_usec) / 1_000_000 + let system = Double(usage.ru_stime.tv_sec) + Double(usage.ru_stime.tv_usec) / 1_000_000 + return user + system + } +} + +/// Connected loopback socket pairs. +private struct SocketPairs: ~Copyable { + + let listener: SocketDescriptor + + let sockets: [SocketDescriptor] + + init(count: Int) throws { + let listener = try SocketDescriptor(IPv4Protocol.tcp) + try listener.bind(IPv4SocketAddress(address: .loopback, port: 0)) + try listener.listen(backlog: count) + let address = try listener.address(IPv4SocketAddress.self) + var sockets = [SocketDescriptor]() + sockets.reserveCapacity(count * 2) + for _ in 0 ..< count { + let client = try SocketDescriptor(IPv4Protocol.tcp) + try client.connect(to: address) + sockets.append(client) + sockets.append(try listener.accept()) + } + self.listener = listener + self.sockets = sockets + } + + func closeListener() { + try? listener.close() + } + + func close() { + sockets.forEach { try? $0.close() } + closeListener() + } +} diff --git a/Tests/SocketTests/MockingTests.swift b/Tests/SocketTests/MockingTests.swift new file mode 100644 index 0000000..523bd36 --- /dev/null +++ b/Tests/SocketTests/MockingTests.swift @@ -0,0 +1,51 @@ +import Foundation +import Testing +import SystemPackage +@testable import Socket + +#if ENABLE_MOCKING +@Suite("Mocking Tests") +struct MockingTests { + + @Test("Syscalls are traced while mocking") + func trace() throws { + MockingDriver.withMockingEnabled { driver in + let socket = SocketDescriptor(rawValue: 3) + _ = socket._close() + let entry = driver.trace.dequeue() + #expect(entry?.name == "close") + #expect(entry?.arguments == [AnyHashable(CInt(3))]) + #expect(driver.trace.isEmpty) + } + } + + @Test("Forced errno is thrown instead of performing the syscall") + func forceErrno() throws { + MockingDriver.withMockingEnabled { driver in + driver.forceErrno = .always(errno: EBADF) + let socket = SocketDescriptor(rawValue: 3) + #expect(throws: Errno.badFileDescriptor) { try socket._close().get() } + } + } + + @Test("Counted errno applies to a fixed number of calls") + func countedErrno() throws { + MockingDriver.withMockingEnabled { driver in + driver.forceErrno = .counted(errno: EBADF, count: 1) + let socket = SocketDescriptor(rawValue: 3) + #expect(throws: Errno.badFileDescriptor) { try socket._close().get() } + #expect(throws: Never.self) { try socket._close().get() } + #expect(driver.forceErrno == .none) + } + } + + @Test("Mocking is disabled outside the driver scope") + func scoped() throws { + #expect(contextualMockingEnabled == false) + MockingDriver.withMockingEnabled { _ in + #expect(contextualMockingEnabled) + } + #expect(contextualMockingEnabled == false) + } +} +#endif diff --git a/Tests/SocketTests/SocketTests.swift b/Tests/SocketTests/SocketTests.swift index a6f61d1..2957227 100644 --- a/Tests/SocketTests/SocketTests.swift +++ b/Tests/SocketTests/SocketTests.swift @@ -4,7 +4,8 @@ import SystemPackage import Logging @testable import Socket -@Suite("Socket Tests") +// a socket test that blocks forever should fail, not run out the job timeout +@Suite("Socket Tests", .timeLimit(.minutes(1))) struct SocketTests { static let logger = Logger(label: "logger") { label in @@ -16,30 +17,35 @@ struct SocketTests { #if os(Linux) @Test("Unix Socket Communication") func testUnixSocket() async throws { - let address = UnixSocketAddress(path: FilePath("/tmp/testsocket.sock")) - Self.logger.info("Using path \(address.path.description)") + // each socket needs its own path, a bound path cannot be shared, + // and a unique name avoids colliding with a file left by an earlier run + let name = UUID().uuidString.prefix(8) + let addressA = UnixSocketAddress(path: FilePath("/tmp/testsocket-a-\(name).sock")) + let addressB = UnixSocketAddress(path: FilePath("/tmp/testsocket-b-\(name).sock")) + Self.logger.info("Using paths \(addressA.path.description) and \(addressB.path.description)") + defer { + try? FileManager.default.removeItem(atPath: addressA.path.description) + try? FileManager.default.removeItem(atPath: addressB.path.description) + } let socketA = try await Socket( UnixProtocol.raw ) Self.logger.info("Created socket A") - let option: GenericSocketOption.ReuseAddress = true - try socketA.fileDescriptor.setSocketOption(option) - do { try socketA.fileDescriptor.bind(address) } - catch { } + try socketA.fileDescriptor.bind(addressA) defer { Task { await socketA.close() } } - + let socketB = try await Socket( UnixProtocol.raw ) Self.logger.info("Created socket B") - try socketB.fileDescriptor.setSocketOption(option) - try socketB.fileDescriptor.bind(address) + try socketB.fileDescriptor.bind(addressB) defer { Task { await socketB.close() } } - + let data = Data("Test \(UUID())".utf8) - - try await socketA.write(data) - Self.logger.info("Socket A wrote data") + + // unconnected datagram sockets must name their destination + try await socketA.sendMessage(data, to: addressB) + Self.logger.info("Socket A sent data") let read = try await socketB.read(data.count) Self.logger.info("Socket B read data") #expect(data == read) @@ -48,19 +54,21 @@ struct SocketTests { @Test("IPv4 TCP Socket Communication") func testIPv4TCPSocket() async throws { - let port = UInt16.random(in: 8080 ..< .max) - Self.logger.info("Using port \(port)") - let address = IPv4SocketAddress(address: .any, port: port) + // let the kernel assign a free port, a hardcoded one may already be taken + let address = IPv4SocketAddress(address: .any, port: 0) let data = Data("Test \(UUID())".utf8) let server = try await Socket( IPv4Protocol.tcp, bind: address ) + let serverAddress = try server.fileDescriptor.address(IPv4SocketAddress.self) + Self.logger.info("Using port \(serverAddress.port)") + let destination = IPv4SocketAddress(address: .loopback, port: serverAddress.port) + #expect(serverAddress.port != 0) + Self.logger.info("Server: Created server socket \(server.fileDescriptor)") + // listen before the client connects, otherwise the connection is refused + try await server.listen() let newConnectionTask = Task { - #expect(try server.fileDescriptor.address(IPv4SocketAddress.self) == address) - Self.logger.info("Server: Created server socket \(server.fileDescriptor)") - try await server.listen() - Self.logger.info("Server: Waiting on incoming connection") let newConnection = try await server.accept() Self.logger.info("Server: Got incoming connection \(newConnection.fileDescriptor)") @@ -76,6 +84,10 @@ struct SocketTests { try await Task.sleep(nanoseconds: 10_000_000) let _ = try await newConnection.write(data) Self.logger.info("Server: Wrote outgoing data") + // close once the client is done, waiting on the peer to disconnect + // only ends the stream on platforms that report end of file as a hangup + try await Task.sleep(nanoseconds: 2_500_000_000) + await newConnection.close() return try await eventsTask.value } let serverEventsTask = Task { @@ -102,7 +114,7 @@ struct SocketTests { Self.logger.info("Client: Created client socket \(client.fileDescriptor)") Self.logger.info("Client: Will connect to server") - do { try await client.connect(to: address) } + do { try await client.connect(to: destination) } catch Errno.socketIsConnected { } Self.logger.info("Client: Connected to server") #expect(try client.fileDescriptor.address(IPv4SocketAddress.self).address.rawValue == "127.0.0.1") @@ -112,36 +124,48 @@ struct SocketTests { #expect(data == read) try await Task.sleep(nanoseconds: 2_000_000_000) await client.close() - let clientEvents = try await clientEventsTask.value - #expect(clientEvents.count == 4) - #expect("\(clientEvents)" == "[Socket.Socket.Event.write, Socket.Socket.Event.read, Socket.Socket.Event.didRead(41), Socket.Socket.Event.close]") + // the connect notification races the single element buffer, it is + // replaced by the next event when the consumer has not read it yet + let clientEvents = try await clientEventsTask.value.filter { + if case .connection = $0 { return false } else { return true } + } + // the client never writes, so it is never notified of write readiness + #expect(clientEvents.count == 3) + #expect("\(clientEvents)" == "[Socket.Socket.Event.read, Socket.Socket.Event.didRead(41), Socket.Socket.Event.close]") await server.close() let serverEvents = try await serverEventsTask.value #expect(serverEvents.count == 2) #expect("\(serverEvents)" == "[Socket.Socket.Event.connection, Socket.Socket.Event.close]") let newConnectionEvents = try await newConnectionTask.value - #expect(newConnectionEvents.count == 5) - #expect("\(newConnectionEvents)" == "[Socket.Socket.Event.write, Socket.Socket.Event.didWrite(41), Socket.Socket.Event.write, Socket.Socket.Event.read, Socket.Socket.Event.close]") + // write readiness is only reported for the pending write + #expect(newConnectionEvents.count == 4) + #expect("\(newConnectionEvents)" == "[Socket.Socket.Event.write, Socket.Socket.Event.didWrite(41), Socket.Socket.Event.read, Socket.Socket.Event.close]") } @Test("IPv4 UDP Socket Communication") func testIPv4UDPSocket() async throws { - let port = UInt16.random(in: 8080 ..< .max) - Self.logger.info("Using port \(port)") + // let the kernel assign a free port, a hardcoded one may already be taken let address = IPv4SocketAddress( address: .any, - port: port + port: 0 ) let data = Data("Test \(UUID())".utf8) - + + // bind before sending, a datagram to an unbound port is dropped + let server = try await Socket( + IPv4Protocol.udp, + bind: address + ) + // datagrams are sent to loopback, `.any` is not a destination address + let boundPort = try server.fileDescriptor.address(IPv4SocketAddress.self).port + let serverAddress = IPv4SocketAddress( + address: .loopback, + port: boundPort + ) + Self.logger.info("Server: Created server socket \(server.fileDescriptor) on port \(boundPort)") + Task { - let server = try await Socket( - IPv4Protocol.udp, - bind: address - ) defer { Task { await server.close() } } - Self.logger.info("Server: Created server socket \(server.fileDescriptor)") - do { Self.logger.info("Server: Waiting to receive incoming message") let (read, clientAddress) = try await server.receiveMessage(data.count, fromAddressOf: type(of: address)) @@ -164,7 +188,7 @@ struct SocketTests { Self.logger.info("Client: Created client socket \(client.fileDescriptor)") Self.logger.info("Client: Waiting to send outgoing message") - try await client.sendMessage(data, to: address) + try await client.sendMessage(data, to: serverAddress) Self.logger.info("Client: Sent outgoing message") Self.logger.info("Client: Waiting to receive incoming message")