diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowExtension.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowExtension.swift index e803959..420ba6c 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowExtension.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowExtension.swift @@ -76,26 +76,49 @@ extension EndpointFlow { switch transport { case .tcp(let options): - guard let reference = TCPProtocol().newProtocolInstance(context: context) else { - throw NetworkError.posix(EINVAL) + if case .custom(let linkOptions) = stack.link, + linkOptions.identifier == BridgeDatagramProtocol.identifier + { + guard let reference = TCPProtocol().newProtocolInstance(context: context) else { + throw NetworkError.posix(EINVAL) + } + options.setProtocolInstance(reference) + let linkage = OutboundStreamLinkage(reference: reference) + let flow = try StreamEndpointFlowProtocol( + identifier: String(self.identifier), + local: effectiveLocalEndpoint, + remote: effectiveRemoteEndpoint, + parameters: self.parameters, + path: path, + context: context, + lowerStreamProtocol: linkage + ) + self.flowProtocol = .stream(flow) + options.setLogID( + prefix: "C", + parent: String(self.identifier), + protocolLogIDNumber: Int(self.identifier) + ) + } else { + let socketReference = SocketStreamProtocol.instance(context: context) + options.setProtocolInstance(socketReference) + let linkage = OutboundStreamLinkage(reference: socketReference) + let flow = try StreamEndpointFlowProtocol( + identifier: String(self.identifier), + local: effectiveLocalEndpoint, + remote: effectiveRemoteEndpoint, + parameters: self.parameters, + path: path, + context: context, + lowerStreamProtocol: linkage + ) + self.flowProtocol = .stream(flow) + options.setLogID( + prefix: "C", + parent: String(self.identifier), + protocolLogIDNumber: Int(self.identifier) + ) } - options.setProtocolInstance(reference) - let linkage = OutboundStreamLinkage(reference: reference) - let flow = try StreamEndpointFlowProtocol( - identifier: String(self.identifier), - local: effectiveLocalEndpoint, - remote: effectiveRemoteEndpoint, - parameters: self.parameters, - path: path, - context: context, - lowerStreamProtocol: linkage - ) - self.flowProtocol = .stream(flow) - options.setLogID( - prefix: "C", - parent: String(self.identifier), - protocolLogIDNumber: Int(self.identifier) - ) case .udp(let options): if case .custom(let linkOptions) = stack.link, linkOptions.identifier == BridgeDatagramProtocol.identifier diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift index 17fa399..6f43954 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift @@ -165,9 +165,13 @@ class EndpointFlowProtocol: ProtocolInstanceCon func handleInboundDataAvailableEvent(_ from: ProtocolInstanceReference) { log.debug("Received inbound data available event") + // Clear the slot before invoking: the completion may synchronously + // re-arm the waiter (when receiveStreamData returns nil because the + // requested minimum spans more than one segment). Clearing afterwards + // would clobber that re-registration and drop later notifications. if let inboundDataAvailableCompletion = self.completions.inboundDataAvailable { - inboundDataAvailableCompletion(true) self.completions.inboundDataAvailable = nil + inboundDataAvailableCompletion(true) } } diff --git a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift index 072ce5d..b7f6221 100644 --- a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift +++ b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift @@ -25,6 +25,8 @@ internal import os #if canImport(Dispatch) import Dispatch +// MARK: - SocketDatagramProtocol + @_spi(Essentials) @available(Network 0.1.0, *) public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInstanceContainer { @@ -259,6 +261,8 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta // - When the write source fires: call serviceWrites again to retry // - When all writes succeed: suspend the write source, notify upper protocol + // MARK: - Write source + private func setupWriteSource() { socket?.withFileDescriptor { fileDescriptor -> Void in dispatchWriteSource = DispatchSource.makeWriteSource(fileDescriptor: fileDescriptor, queue: context.queue) @@ -361,4 +365,850 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta SocketDatagramProtocol(context: context).reference } } -#endif // canImport(Dispatch) + +// MARK: - SocketStreamProtocol + +@_spi(Essentials) +@available(Network 0.1.0, *) +public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceContainer { + + public private(set) var context: NetworkContext + public var reference: ProtocolInstanceReference { ProtocolInstanceReference(custom: self) } + public var eventManager = ProtocolEventManager() + public var upper = InboundStreamLinkage() + var log = NetworkLoggerState() + + private var socket: SystemSocket? = nil + #if canImport(Darwin) + // Socket option constants that the Swift Darwin overlay does not surface. + // Values match and the Darwin xnu socket headers. + private static let socketOptionIPV6UseMinMTU: CInt = 42 // IPV6_USE_MIN_MTU + private static let socketOptionIPV6DontFrag: CInt = 62 // IPV6_DONTFRAG + #endif + private var dispatchReadSource: (any DispatchSourceRead)? = nil + private var dispatchWriteSource: (any DispatchSourceWrite)? = nil + private var waitingForWritable = false + private var isConnecting = false + private var inputSourceSuspended = false + private var inputFinished = false + private var outputFinished = false + private var pendingDisconnect = false + private var incomingFrames = FrameArray() + private var pendingOutputFrames = FrameArray() + var localEndpoint: Endpoint? + var remoteEndpoint: Endpoint? + + private let maximumInputSize = 65536 + private let maximumOutputSize = 65536 + + // Cap dynamic input sizing so a flood of pending bytes can't make us + // allocate an arbitrarily large temporary buffer. + private static let maximumDynamicInputSize = 256 * 1024 + + // TCPMetadata wired up so the upper layer can query and modify socket + // state via the standard TCP option callbacks. + private var protocolMetadata: ProtocolMetadata? = nil + private var socketHandle: UnsafeMutableRawPointer? = nil + + init(context: NetworkContext) { + self.context = context + } + + deinit { + releaseSocketHandle() + socket = nil + incomingFrames.finalizeAllFramesAsFailed() + pendingOutputFrames.finalizeAllFramesAsFailed() + } + + public func setup( + remote: Endpoint?, + local: Endpoint?, + parameters: Parameters?, + path: PathProperties? + ) throws(NetworkError) { + guard let remote else { + throw NetworkError.posix(EINVAL) + } + + self.localEndpoint = local + self.remoteEndpoint = remote + + guard case .address(let address) = remote.type else { + throw NetworkError.posix(EINVAL) + } + let socket = try createSocket(for: address) + self.socket = socket + + applyDefaultSocketOptions(socket: socket) + + if let stack = parameters?.defaultStack { + if let ipOptions = stack.internetOptionsAsIPOptions(mutable: false)?.perProtocolOptions { + applyIPOptions(socket: socket, opts: ipOptions) + } + if let tcpOptions = stack.transport?.options as? TCPProtocol.Options { + applyTCPOptions(socket: socket, opts: tcpOptions) + } + } + + setupProtocolMetadata(socket: socket) + + setupReadSource() + setupWriteSource() + } + + public func teardown() { + cancelReadSource() + cancelWriteSource() + releaseSocketHandle() + socket = nil + incomingFrames.finalizeAllFramesAsFailed() + pendingOutputFrames.finalizeAllFramesAsFailed() + } + + public func connect() { + guard let socket, let remoteEndpoint, + case .address(let address) = remoteEndpoint.type + else { + log.error("Cannot connect: no socket or remote endpoint") + deliverDisconnectedEvent(error: .posix(ENOTCONN)) + return + } + + do { + let isIPv6Dst = { + if case .v6 = address.type { return true } + return false + }() + if let localEndpoint, case .address(let localAddress) = localEndpoint.type, + !isIPv6Dst || localAddress.addressFamily == .ipv6 + { + try bindSocket(to: localAddress, port: localEndpoint.port) + } + if case .v6 = address.type { + // connectx(2) on Darwin requires the socket to be bound before + // connecting an IPv6 socket with no explicit source address. + try socket.bindSocket(address: IPv6Address.any, port: 0) + } + + let ip: any IPAddress + switch address.type { + case .v4(let addr, _): ip = addr + case .v6(let addr, _): ip = addr + default: + log.error("Unsupported address family for connect") + deliverDisconnectedEvent(error: .posix(EAFNOSUPPORT)) + return + } + + // For non-blocking TCP, connectSocket returns false on EINPROGRESS. + // Wait for the socket to become writable, then treat as connected. + let connectedNow = try socket.connectSocket(to: ip, port: remoteEndpoint.port) + if connectedNow { + deliverConnectedEvent() + } else { + isConnecting = true + if !waitingForWritable { + waitingForWritable = true + dispatchWriteSource?.resume() + } + } + } catch let error { + log.error("Failed to connect: \(error)") + deliverDisconnectedEvent(error: .posix(ECONNREFUSED)) + } + } + + public func disconnect() { + if pendingOutputFrames.isEmpty { + shutdownWrites() + deliverDisconnectedEvent(error: nil) + return + } + pendingDisconnect = true + serviceWrites() + } + + // MARK: - BottomStreamProtocol + + public func receiveStreamData(minimumBytes: Int, maximumBytes: Int) throws(NetworkError) -> FrameArray? { + guard !incomingFrames.isEmpty, + incomingFrames.unclaimedLength >= minimumBytes || incomingFrames.connectionComplete + else { + // We don't have enough buffered to satisfy the consumer yet. If we + // had suspended on the high-water mark, resume — the consumer needs + // more than we're currently holding, so reading must continue even + // past the soft cap. Otherwise a large minimum would deadlock. + if inputSourceSuspended && !inputFinished { + inputSourceSuspended = false + dispatchReadSource?.resume() + } + return nil + } + let result = incomingFrames.drainArray(maximumByteCount: maximumBytes) + if inputSourceSuspended, incomingFrames.unclaimedLength < maximumInputSize { + inputSourceSuspended = false + dispatchReadSource?.resume() + } + return result + } + + public func getOutboundStreamDataRoomAvailable() throws(NetworkError) -> Int { + let pending = pendingOutputFrames.unclaimedLength + if pending >= maximumOutputSize { return 0 } + return maximumOutputSize - pending + } + + public func sendStreamData(_ streamData: consuming FrameArray) throws(NetworkError) { + pendingOutputFrames.add(frames: streamData) + serviceWrites() + } + + #if !NETWORK_EMBEDDED + public var metadata: AbstractProtocolMetadata? { protocolMetadata } + #endif + + private func createSocket(for address: AddressEndpoint) throws(NetworkError) -> SystemSocket { + switch address.type { + case .v4: + return try SystemSocket( + protocolFamily: .ipv4, + sockType: .stream, + protocolSubType: 0, + nonBlocking: true + ) + case .v6: + return try SystemSocket( + protocolFamily: .ipv6, + sockType: .stream, + protocolSubType: 0, + nonBlocking: true + ) + default: + throw NetworkError.posix(EAFNOSUPPORT) + } + } + + private func bindSocket(to address: AddressEndpoint, port: UInt16) throws(NetworkError) { + do { + switch address.type { + case .v4(let ip, _): + try socket?.bindSocket(address: ip, port: port) + case .v6(let ip, _): + try socket?.bindSocket(address: ip, port: port) + default: + break + } + } catch { + throw NetworkError.posix(EADDRNOTAVAIL) + } + } + + // MARK: - Read source + + private func setupReadSource() { + socket?.withFileDescriptor { fileDescriptor in + dispatchReadSource = DispatchSource.makeReadSource(fileDescriptor: fileDescriptor, queue: context.queue) + dispatchReadSource?.setEventHandler { + self.handleSocketReadEvent() + } + dispatchReadSource?.resume() + } + } + + private func cancelReadSource() { + // A read source must be resumed before it can be released, and it must + // not be left armed once we're done reading — an EOF condition stays + // readable, so an armed source would spin the handler forever. + if inputSourceSuspended { + dispatchReadSource?.resume() + inputSourceSuspended = false + } + dispatchReadSource?.setEventHandler(handler: nil) + dispatchReadSource?.cancel() + dispatchReadSource = nil + } + + private func handleSocketReadEvent() { + guard !inputFinished else { return } + + let pending = socket?.availableBytesToRead() ?? 0 + let readSize: Int + if pending > 0 { + readSize = min(max(pending, maximumInputSize), Self.maximumDynamicInputSize) + } else { + readSize = maximumInputSize + } + + withUnsafeTemporaryAllocation(byteCount: readSize, alignment: 1) { buffer in + let readBuffer = buffer.baseAddress! + var receivedAny = false + var reachedEOF = false + var fatalError: NetworkError? = nil + + repeat { + let result: IOResult? + do { + result = try socket?.readIOResult(buffer: readBuffer, size: readSize) + } catch let error as NetworkError { + fatalError = error + break + } catch { + fatalError = .posix(EIO) + break + } + guard let result else { break } + switch result { + case .processed(let bytesRead): + if bytesRead == 0 { + // Stream EOF — mark the next frame as connectionComplete. + reachedEOF = true + } else { + let frame = Frame(copyBuffer: UnsafeRawBufferPointer(start: readBuffer, count: bytesRead)) + incomingFrames.add(frame: frame) + receivedAny = true + } + case .wouldBlock: + break + } + if reachedEOF { break } + if case .wouldBlock = result { break } + if incomingFrames.unclaimedLength >= maximumInputSize { break } + } while true + + if reachedEOF { + inputFinished = true + // Tag the last incoming frame (or an empty one) with connectionComplete + // so the upper protocol sees stream completion. + var sentinel = Frame(count: 0) + sentinel.connectionComplete = true + incomingFrames.add(frame: sentinel) + receivedAny = true + // The stream is finished; stop the read source. EOF keeps the + // descriptor readable, so leaving it armed would spin forever. + cancelReadSource() + } + + if let fatalError { + deliverDisconnectedEvent(error: fatalError) + return + } + + if receivedAny { + fromExternal { + upper.deliverInboundDataAvailableEvent(reference) + } + } + // Backpressure on buffered volume: suspend whenever we're over the + // limit, regardless of whether new frames arrived this call. The + // read source is level-triggered, so if we exit the loop due to the + // cap without suspending, it would fire again immediately. + if !inputFinished, + !inputSourceSuspended, + incomingFrames.unclaimedLength >= maximumInputSize + { + inputSourceSuspended = true + dispatchReadSource?.suspend() + } + } + } + + // MARK: - Write source + + private func setupWriteSource() { + socket?.withFileDescriptor { fileDescriptor -> Void in + dispatchWriteSource = DispatchSource.makeWriteSource(fileDescriptor: fileDescriptor, queue: context.queue) + dispatchWriteSource?.setEventHandler { + self.handleSocketWriteEvent() + } + // Starts suspended — resumed on EINPROGRESS connect or EAGAIN on write. + } + } + + private func handleSocketWriteEvent() { + if isConnecting { + isConnecting = false + if waitingForWritable { + waitingForWritable = false + dispatchWriteSource?.suspend() + } + let connectError = socket?.getSocketError() ?? 0 + if connectError != 0 { + log.error("Async connect failed: \(connectError)") + deliverDisconnectedEvent(error: .posix(connectError)) + return + } + deliverConnectedEvent() + + // Try any pending writes that arrived before connect completed. + serviceWrites() + triggerOutboundRoomAvailable() + return + } + serviceWrites() + triggerOutboundRoomAvailable() + } + + private func triggerOutboundRoomAvailable() { + fromExternal { + upper.deliverOutboundRoomAvailableEvent(reference) + } + } + + private func cancelWriteSource() { + guard let dispatchWriteSource else { return } + if !waitingForWritable { + dispatchWriteSource.resume() + } + dispatchWriteSource.setEventHandler(handler: nil) + dispatchWriteSource.cancel() + self.dispatchWriteSource = nil + waitingForWritable = false + } + + // Drains pendingOutputFrames, includes partial writes (unlike the datagram + // version). On EAGAIN, resumes the write source to retry when writable. + // On fatal errors (EPIPE, ECONNRESET, etc.), delivers a disconnected event. + // When a frame with connectionComplete is fully written, issues SHUT_WR. + private func serviceWrites() { + guard !isConnecting else { return } + + var shouldShutdownWrite = false + var fatalError: NetworkError? = nil + var remaining = FrameArray() + + while var frame = pendingOutputFrames.popFirst() { + // Once we've hit backpressure or a fatal error, preserve the remaining + // frames in order for retry (on EAGAIN) or failure (on a fatal error). + if fatalError != nil { + frame.finalize(success: false) + continue + } + + var madeProgress = true + while frame.unclaimedLength > 0 { + let length = frame.unclaimedLength + let bytesWritten = writeFrameToSocket(&frame, length: length) + if bytesWritten == length { + break + } + if bytesWritten > 0 { + // Partial write — claim the bytes that made it out and retry. + _ = frame.claim(fromStart: bytesWritten) + continue + } + let err: CInt = bytesWritten < 0 ? CInt(-bytesWritten) : EIO + switch err { + case EAGAIN, EWOULDBLOCK, ENOBUFS: + self.log.datapath("Send buffer full, waiting for writable event") + madeProgress = false + case EPIPE: + self.log.info("Socket has been closed") + fatalError = .posix(EPIPE) + case ECONNRESET: + self.log.info("Connection reset") + fatalError = .posix(ECONNRESET) + default: + self.log.datapath("write failed: \(err)") + fatalError = .posix(err) + } + break + } + + if fatalError != nil { + frame.finalize(success: false) + continue + } + + if !madeProgress { + // Keep this frame (and any still-queued frames) for a later attempt. + remaining.add(frame: frame) + while let next = pendingOutputFrames.popFirst() { + remaining.add(frame: next) + } + break + } + + if frame.connectionComplete { + shouldShutdownWrite = true + } + frame.finalize(success: true) + } + + pendingOutputFrames = remaining + + if let fatalError { + if waitingForWritable { + waitingForWritable = false + dispatchWriteSource?.suspend() + } + deliverDisconnectedEvent(error: fatalError) + return + } + + if !pendingOutputFrames.isEmpty { + if !waitingForWritable { + waitingForWritable = true + dispatchWriteSource?.resume() + } + return + } + + if waitingForWritable { + waitingForWritable = false + dispatchWriteSource?.suspend() + } + if shouldShutdownWrite { + shutdownWrites() + } + if pendingDisconnect { + pendingDisconnect = false + shutdownWrites() + deliverDisconnectedEvent(error: nil) + } + } + + // Returns bytes written on success (≥ 0), or -errno on failure (< 0). + // Using a negative errno avoids relying on the C errno global after Swift + // error-handling boundaries, which can clobber it. + private func writeFrameToSocket(_ frame: inout Frame, length: Int) -> Int { + guard let socket else { return -Int(EINVAL) } + if length == 0 { return 0 } + guard let bytes = frame.bytes else { return -Int(EINVAL) } + var result: Int = 0 + bytes.withUnsafeBytes { rawBytes in + guard let baseAddress = rawBytes.baseAddress else { + result = -Int(EINVAL) + return + } + do { + result = try socket.write(buffer: baseAddress, size: length) + } catch let error as NetworkError { + switch error.domainSpecificError { + case .some(let (_, code)): + result = -Int(code) + case .none: + result = -Int(EIO) + } + } catch { + result = -Int(EIO) + } + } + return result + } + + private func shutdownWrites() { + guard !outputFinished, let socket else { return } + outputFinished = true + socket.withFileDescriptor { fd in + #if canImport(Glibc) + _ = Glibc.shutdown(fd, CInt(SHUT_WR)) + #else + _ = shutdown(fd, CInt(SHUT_WR)) + #endif + } + } + + // MARK: - Socket option / metadata wiring + + // Defaults applied to every stream socket. SO_RCVLOWAT / SO_SNDLOWAT are + // dropped to 1 byte so the read/write dispatch sources fire as soon as any + // data or any send-buffer space is available — the default macOS kernel + // values are larger and would delay async-connect completion behind the + // write low-watermark. + private func applyDefaultSocketOptions(socket: SystemSocket) { + do { try socket.setReceiveLowWatermark(1) } catch { log.info("Failed to set SO_RCVLOWAT: \(error)") } + do { try socket.setSendLowWatermark(1) } catch { log.info("Failed to set SO_SNDLOWAT: \(error)") } + } + + private func applyTCPOptions(socket: SystemSocket, opts: TCPProtocol.Options) { + if opts.noDelay { + do { try socket.setNoDelay(true) } catch { log.info("Failed to set TCP_NODELAY: \(error)") } + } + if opts.enableKeepalive { + do { + try socket.setKeepalive( + enabled: true, + idleTime: opts.keepaliveIdleTime, + interval: opts.keepaliveInterval, + count: opts.keepaliveCount + ) + } catch { + log.info("Failed to enable keepalive: \(error)") + } + } + if opts.maximumSegmentSize > 0 { + do { try socket.setMaximumSegmentSize(opts.maximumSegmentSize) } catch { + log.info("Failed to set TCP_MAXSEG: \(error)") + } + } + if opts.reduceBuffering { + // Match the C++ reduce_buffering behavior: cap unsent bytes at 16 KiB. + do { try socket.setNotSentLowWatermark(16 * 1024) } catch { + log.info("Failed to set TCP_NOTSENT_LOWAT: \(error)") + } + } + if opts.resetLocalPort { + do { try socket.setReusableLocalPort(true) } catch { log.info("Failed to set SO_REUSEPORT: \(error)") } + } + #if canImport(Darwin) + if opts.noPush { + do { try socket.setNoPush(true) } catch { log.info("Failed to set TCP_NOPUSH: \(error)") } + } + if opts.noOptions { + do { try socket.setNoOptions(true) } catch { log.info("Failed to set TCP_NOOPT: \(error)") } + } + if opts.disableAckStretching { + do { try socket.setSendMoreAcks(true) } catch { log.info("Failed to set TCP_SENDMOREACKS: \(error)") } + } + if opts.retransmitFinDrop { + do { try socket.setRetransmitFinDrop(true) } catch { log.info("Failed to set TCP_RXT_FINDROP: \(error)") } + } + #endif + } + + private func applyIPOptions(socket: SystemSocket, opts: IPProtocol.Options) { + guard let remoteEndpoint, case .address(let address) = remoteEndpoint.type else { + return + } + let isIPv6: Bool + switch address.type { + case .v6: isIPv6 = true + case .v4: isIPv6 = false + default: return + } + + if let hopLimit = opts.hopLimit { + do { + if isIPv6 { + try socket.setSocketOption( + level: CInt(IPPROTO_IPV6), + name: IPV6_UNICAST_HOPS, + value: CInt(hopLimit) + ) + } else { + try socket.setSocketOption( + level: CInt(IPPROTO_IP), + name: IP_TTL, + value: CInt(hopLimit) + ) + } + } catch { + log.info("Failed to set hop limit: \(error)") + } + } + + if let dscpValue = opts.dscpValue { + // DSCP occupies the upper 6 bits of the IPv4 ToS / IPv6 Traffic Class byte. + let tos = CInt(dscpValue) << 2 + do { + if isIPv6 { + try socket.setSocketOption( + level: CInt(IPPROTO_IPV6), + name: IPV6_TCLASS, + value: tos + ) + } else { + try socket.setSocketOption( + level: CInt(IPPROTO_IP), + name: IP_TOS, + value: tos + ) + } + } catch { + log.info("Failed to set DSCP: \(error)") + } + } + + #if canImport(Darwin) + if isIPv6 && opts.flags.contains(.useMinimumMTU) { + do { + try socket.setSocketOption( + level: CInt(IPPROTO_IPV6), + name: Self.socketOptionIPV6UseMinMTU, + value: CInt(1) + ) + } catch { + log.info("Failed to set IPV6_USE_MIN_MTU: \(error)") + } + } + + if let fragmentationEnabled = opts.fragmentationEnabled { + let dontFragment: CInt = fragmentationEnabled ? 0 : 1 + do { + if isIPv6 { + try socket.setSocketOption( + level: CInt(IPPROTO_IPV6), + name: Self.socketOptionIPV6DontFrag, + value: dontFragment + ) + } else { + try socket.setSocketOption( + level: CInt(IPPROTO_IP), + name: IP_DONTFRAG, + value: dontFragment + ) + } + } catch { + log.info("Failed to set DONTFRAG: \(error)") + } + } + #elseif canImport(Glibc) || canImport(Musl) + if let fragmentationEnabled = opts.fragmentationEnabled { + let disc: CInt = fragmentationEnabled ? CInt(IP_PMTUDISC_DONT) : CInt(IP_PMTUDISC_DO) + do { + if isIPv6 { + try socket.setSocketOption( + level: CInt(IPPROTO_IPV6), + name: IPV6_MTU_DISCOVER, + value: disc + ) + } else { + try socket.setSocketOption( + level: CInt(IPPROTO_IP), + name: IP_MTU_DISCOVER, + value: disc + ) + } + } catch { + log.info("Failed to set MTU_DISCOVER: \(error)") + } + } + #endif + } + + // Wires up TCPMetadata so the upper layer can query buffer sizes and modify + // socket state through the standard TCP option callbacks. Only callbacks + // that map to public socket APIs are populated. + private func setupProtocolMetadata(socket: SystemSocket) { + let box = SocketHandleBox(socket: socket) + let handle = UnsafeMutableRawPointer(Unmanaged.passRetained(box).toOpaque()) + socketHandle = handle + + let metadata = TCPProtocol.TCPMetadata() + metadata.handle = handle + metadata.callbacks = TCPProtocol.TCPMetadata.TCPOptionCallbacks( + get_receive_buffer_size: socketGetReceiveBufferSize, + get_send_buffer_size: socketGetSendBufferSize, + reset_keepalives: socketResetKeepalives, + set_no_delay: socketSetNoDelay, + set_no_push: socketSetNoPush, + set_no_wake_from_sleep: nil, + set_max_pacing_rate: socketSetMaxPacingRate + ) + protocolMetadata = ProtocolMetadata( + protocolIdentifier: TCPProtocol.identifier, + perProtocolMetadata: metadata, + messageIdentifier: SystemUUID() + ) + } + + private func releaseSocketHandle() { + if let perMetadata = protocolMetadata?.perProtocolMetadata { + perMetadata.handle = nil + perMetadata.callbacks = nil + } + protocolMetadata = nil + if let handle = socketHandle { + Unmanaged.fromOpaque(handle).release() + socketHandle = nil + } + } + + static public func instance(context: NetworkContext) -> ProtocolInstanceReference { + SocketStreamProtocol(context: context).reference + } +} + +// MARK: - SocketHandleBox (private helpers for TCPMetadata callbacks) + +// Strong-reference holder bridged through an opaque pointer so the C-convention +// TCPMetadata callbacks can reach back to the SystemSocket. The protocol owns +// the box's lifetime via Unmanaged.passRetained / .release in setup/teardown. +@available(Network 0.1.0, *) +private final class SocketHandleBox { + let socket: SystemSocket + init(socket: SystemSocket) { self.socket = socket } +} + +@available(Network 0.1.0, *) +private let socketGetReceiveBufferSize: @convention(c) (UnsafeMutableRawPointer?) -> UInt32 = { handle in + guard let handle else { return 0 } + let box = Unmanaged.fromOpaque(handle).takeUnretainedValue() + return UInt32(max(0, Int(box.socket.getReceiveBufferSize()))) +} + +@available(Network 0.1.0, *) +private let socketGetSendBufferSize: @convention(c) (UnsafeMutableRawPointer?) -> UInt32 = { handle in + guard let handle else { return 0 } + let box = Unmanaged.fromOpaque(handle).takeUnretainedValue() + return UInt32(max(0, Int(box.socket.getSendBufferSize()))) +} + +@available(Network 0.1.0, *) +private let socketResetKeepalives: @convention(c) (UnsafeMutableRawPointer?, Bool, UInt32, UInt32, UInt32) -> Int32 = { + handle, + enabled, + count, + idleTime, + interval in + guard let handle else { return -1 } + let box = Unmanaged.fromOpaque(handle).takeUnretainedValue() + do { + try box.socket.setKeepalive(enabled: enabled, idleTime: idleTime, interval: interval, count: count) + return 0 + } catch { + return -1 + } +} + +@available(Network 0.1.0, *) +private let socketSetNoDelay: @convention(c) (UnsafeMutableRawPointer?, Bool) -> Int32 = { handle, enabled in + guard let handle else { return -1 } + let box = Unmanaged.fromOpaque(handle).takeUnretainedValue() + do { + try box.socket.setNoDelay(enabled) + return 0 + } catch { + return -1 + } +} + +@available(Network 0.1.0, *) +private let socketSetNoPush: @convention(c) (UnsafeMutableRawPointer?, Bool) -> Int32 = { handle, enabled in + guard let handle else { return -1 } + let box = Unmanaged.fromOpaque(handle).takeUnretainedValue() + #if canImport(Darwin) + do { + try box.socket.setNoPush(enabled) + return 0 + } catch { + return -1 + } + #else + _ = box + return -1 + #endif +} + +@available(Network 0.1.0, *) +private let socketSetMaxPacingRate: @convention(c) (UnsafeMutableRawPointer?, UInt64) -> Int32 = { handle, rate in + guard let handle else { return -1 } + let box = Unmanaged.fromOpaque(handle).takeUnretainedValue() + #if canImport(Glibc) + do { + try box.socket.setSocketOption( + level: SOL_SOCKET, + name: SO_MAX_PACING_RATE, + value: UInt32(min(rate, UInt64(UInt32.max))) + ) + return 0 + } catch { + return -1 + } + #else + _ = (box, rate) + return -1 + #endif +} +#endif diff --git a/Sources/SwiftNetwork/System/SystemSocket.swift b/Sources/SwiftNetwork/System/SystemSocket.swift index 7b838f5..6c133ca 100644 --- a/Sources/SwiftNetwork/System/SystemSocket.swift +++ b/Sources/SwiftNetwork/System/SystemSocket.swift @@ -244,6 +244,156 @@ class SystemSocket { return try System.recvmsg(descriptor: self.sockfd, msgHdr: msgHdr, flags: flags).result } + // MARK: - Socket options + + /// Set a fixed-size socket option value. + public func setSocketOption(level: CInt, name: CInt, value: T) throws(NetworkError) { + guard self.sockfd > 0 else { throw NetworkError.posix(EINVAL) } + var copy = value + let size = socklen_t(MemoryLayout.size) + let result = withUnsafePointer(to: ©) { ptr -> CInt in + ptr.withMemoryRebound(to: UInt8.self, capacity: Int(size)) { rebound in + setsockopt(self.sockfd, level, name, rebound, size) + } + } + if result != 0 { + throw NetworkError.posix(errno) + } + } + + /// Get a fixed-size socket option value. + public func getSocketOption(level: CInt, name: CInt, defaultValue: T) throws(NetworkError) -> T { + guard self.sockfd > 0 else { throw NetworkError.posix(EINVAL) } + var value = defaultValue + var size = socklen_t(MemoryLayout.size) + let result = withUnsafeMutablePointer(to: &value) { ptr -> CInt in + ptr.withMemoryRebound(to: UInt8.self, capacity: Int(size)) { rebound in + getsockopt(self.sockfd, level, name, rebound, &size) + } + } + if result != 0 { + throw NetworkError.posix(errno) + } + return value + } + + /// Read pending socket error via SO_ERROR (and clear it). Returns 0 if no error. + public func getSocketError() -> CInt { + (try? getSocketOption(level: SOL_SOCKET, name: SO_ERROR, defaultValue: CInt(0))) ?? 0 + } + + #if canImport(Darwin) + /// Number of bytes available to read in the socket receive buffer (Darwin only). + public func availableBytesToRead() -> Int { + let value: CInt = (try? getSocketOption(level: SOL_SOCKET, name: SO_NREAD, defaultValue: 0)) ?? 0 + return Int(value) + } + #else + /// Number of bytes available to read in the socket receive buffer (Linux: FIONREAD). + public func availableBytesToRead() -> Int { + guard self.sockfd > 0 else { return 0 } + var value: CInt = 0 + let result = ioctl(self.sockfd, UInt(FIONREAD), &value) + return result == 0 ? Int(value) : 0 + } + #endif + + public func getReceiveBufferSize() -> CInt { + (try? getSocketOption(level: SOL_SOCKET, name: SO_RCVBUF, defaultValue: CInt(0))) ?? 0 + } + + public func getSendBufferSize() -> CInt { + (try? getSocketOption(level: SOL_SOCKET, name: SO_SNDBUF, defaultValue: CInt(0))) ?? 0 + } + + public func setReceiveBufferSize(_ bytes: CInt) throws(NetworkError) { + try setSocketOption(level: SOL_SOCKET, name: SO_RCVBUF, value: bytes) + } + + public func setSendBufferSize(_ bytes: CInt) throws(NetworkError) { + try setSocketOption(level: SOL_SOCKET, name: SO_SNDBUF, value: bytes) + } + + public func setReceiveLowWatermark(_ bytes: CInt) throws(NetworkError) { + try setSocketOption(level: SOL_SOCKET, name: SO_RCVLOWAT, value: bytes) + } + + public func setSendLowWatermark(_ bytes: CInt) throws(NetworkError) { + try setSocketOption(level: SOL_SOCKET, name: SO_SNDLOWAT, value: bytes) + } + + public func setNoDelay(_ enabled: Bool) throws(NetworkError) { + let value: CInt = enabled ? 1 : 0 + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_NODELAY, value: value) + } + + public func setKeepalive(enabled: Bool, idleTime: UInt32, interval: UInt32, count: UInt32) throws(NetworkError) { + let on: CInt = enabled ? 1 : 0 + try setSocketOption(level: SOL_SOCKET, name: SO_KEEPALIVE, value: on) + if !enabled { return } + if idleTime > 0 { + #if canImport(Darwin) + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_KEEPALIVE, value: CInt(idleTime)) + #else + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_KEEPIDLE, value: CInt(idleTime)) + #endif + } + if interval > 0 { + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_KEEPINTVL, value: CInt(interval)) + } + if count > 0 { + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_KEEPCNT, value: CInt(count)) + } + } + + /// Set the maximum segment size (TCP_MAXSEG). Portable. + public func setMaximumSegmentSize(_ mss: UInt32) throws(NetworkError) { + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_MAXSEG, value: CInt(mss)) + } + + /// Set TCP_NOTSENT_LOWAT — reduce send-side buffering for latency-sensitive + /// flows. Portable (Darwin and Linux both define TCP_NOTSENT_LOWAT). + public func setNotSentLowWatermark(_ bytes: UInt32) throws(NetworkError) { + #if canImport(Darwin) || canImport(Glibc) + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_NOTSENT_LOWAT, value: CInt(bytes)) + #else + throw NetworkError.posix(ENOTSUP) + #endif + } + + #if canImport(Darwin) + /// TCP_NOPUSH — delay sending small segments (Darwin-only; Linux has TCP_CORK + /// with different semantics, not wired here). + public func setNoPush(_ enabled: Bool) throws(NetworkError) { + let value: CInt = enabled ? 1 : 0 + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_NOPUSH, value: value) + } + + /// TCP_NOOPT — disable all TCP options in outgoing segments. Darwin-only. + public func setNoOptions(_ enabled: Bool) throws(NetworkError) { + let value: CInt = enabled ? 1 : 0 + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_NOOPT, value: value) + } + + /// TCP_SENDMOREACKS — disable ACK stretching. Darwin-only. + public func setSendMoreAcks(_ enabled: Bool) throws(NetworkError) { + let value: CInt = enabled ? 1 : 0 + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_SENDMOREACKS, value: value) + } + + /// TCP_RXT_FINDROP — drop connection after unacked FIN. Darwin-only. + public func setRetransmitFinDrop(_ enabled: Bool) throws(NetworkError) { + let value: CInt = enabled ? 1 : 0 + try setSocketOption(level: CInt(IPPROTO_TCP), name: TCP_RXT_FINDROP, value: value) + } + #endif + + /// Reuse the local port via SO_REUSEPORT. + public func setReusableLocalPort(_ enabled: Bool) throws(NetworkError) { + let value: CInt = enabled ? 1 : 0 + try setSocketOption(level: SOL_SOCKET, name: SO_REUSEPORT, value: value) + } + deinit { close(self.sockfd) } diff --git a/Sources/Tools/SocketTransfer/README.md b/Sources/Tools/SocketTransfer/README.md index dc3559b..1185079 100644 --- a/Sources/Tools/SocketTransfer/README.md +++ b/Sources/Tools/SocketTransfer/README.md @@ -1,6 +1,6 @@ # Socket Transfer Benchmark -SwiftNetwork benchmark for measuring the performance of round-trip datagram transfers over the SocketDatagramProtocol, which uses real system UDP sockets on loopback. +SwiftNetwork benchmark for measuring the performance of round-trip transfers over the kernel-socket protocols — `SocketDatagramProtocol` (UDP) and `SocketStreamProtocol` (TCP) — on loopback. ## Usage of the Benchmark @@ -12,11 +12,47 @@ To use this benchmark for performance measurements, make sure that you build it # Navigate into the ./build/release directory and run the executable ./SocketTransfer ``` -NOTE: Never run and measure this benchmark as a debug build the performance information will not be valid. +NOTE: Never run and measure this benchmark as a debug build; the performance information will not be valid. + +## Command line arguments + +General: +``` +-proto udp|tcp : Transport to use. Default: udp. +-tcp / -udp : Shortcuts for -proto tcp / -proto udp. +-iterations N : Number of transfers to perform. Default: 100. +-size N : Size of each payload in bytes. Default: 1000. +-oneway : One-way send mode (no echo). Measures pure write throughput. +-ip STRING : Remote IP (IPv4 or IPv6) to send to. When set, runs a client-only + send loop instead of the in-process echo setup. +-port N : Remote port (required with -ip). +-localport N : Local source port to bind to. Default: 0 (ephemeral for TCP). +``` + +TCP-specific (applied when `-tcp` / `-proto tcp` is active): +``` +-no-delay : Disable Nagle's algorithm (TCP_NODELAY). +-keepalive : Enable SO_KEEPALIVE with kernel defaults. +-keepalive-idle N : TCP_KEEPIDLE / TCP_KEEPALIVE — idle seconds before the + first probe. Implies -keepalive. +-keepalive-interval N : TCP_KEEPINTVL — seconds between probes. Implies -keepalive. +-keepalive-count N : TCP_KEEPCNT — unacked probes before drop. Implies -keepalive. +``` + +Only the TCP options that actually translate through to `SocketStreamProtocolOptions` are exposed here. Other `TCP()` builder knobs (`noPush`, `maximumSegmentSize`, fast-open, etc.) currently no-op on the kernel-socket path and are deliberately omitted from this tool. + +## Examples -There are a few command line arguments that are accepted: ``` --iterations : The number of transfers to perform. Default: 100. --size : The size of each datagram payload in bytes. Default: 1000. --oneway : Run in one-way send mode (no echo). Measures pure write throughput. +# 1000 UDP echo round-trips, 1400-byte payloads: +./SocketTransfer -iterations 1000 -size 1400 + +# TCP echo with Nagle disabled: +./SocketTransfer -tcp -no-delay -iterations 1000 -size 1400 + +# TCP one-way send with keepalive tuned: +./SocketTransfer -tcp -oneway -keepalive -keepalive-idle 30 -keepalive-interval 5 -keepalive-count 3 + +# Send TCP to a remote listener: +./SocketTransfer -tcp -ip 127.0.0.1 -port 5555 -iterations 500 -size 2048 ``` diff --git a/Sources/Tools/SocketTransfer/main.swift b/Sources/Tools/SocketTransfer/main.swift index b8bda09..9cf0809 100644 --- a/Sources/Tools/SocketTransfer/main.swift +++ b/Sources/Tools/SocketTransfer/main.swift @@ -28,7 +28,7 @@ internal import os // MARK: - IP address parsing @available(Network 0.1.0, *) -func parseIPv4(_ string: String) -> IPv4Address? { +private func parseIPv4(_ string: String) -> IPv4Address? { let parts = string.split(separator: ".") guard parts.count == 4 else { return nil } var bytes = [UInt8]() @@ -40,7 +40,7 @@ func parseIPv4(_ string: String) -> IPv4Address? { } @available(Network 0.1.0, *) -func parseIPv6(_ string: String) -> IPv6Address? { +private func parseIPv6(_ string: String) -> IPv6Address? { var addr = in6_addr() guard string.withCString({ inet_pton(AF_INET6, $0, &addr) }) == 1 else { return nil @@ -49,22 +49,78 @@ func parseIPv6(_ string: String) -> IPv6Address? { return IPv6Address(bytes) } -// MARK: - Helpers +// MARK: - TCP settings (CLI-configured) + +// Only the knobs that actually translate through to SocketStreamProtocolOptions +// today. noPush / maximumSegmentSize / fast-open / etc. are accepted on the +// TCP() builder but silently no-op on the kernel-socket path, so they're not +// exposed here to avoid misleading users. +struct TCPSettings { + var noDelay: Bool = false + var enableKeepalive: Bool = false + var keepaliveIdle: UInt32 = 0 + var keepaliveInterval: UInt32 = 0 + var keepaliveCount: UInt32 = 0 + + func describe() -> String { + var parts: [String] = [] + if noDelay { parts.append("noDelay") } + if enableKeepalive { + parts.append("keepalive(idle=\(keepaliveIdle)s, interval=\(keepaliveInterval)s, count=\(keepaliveCount))") + } + return parts.isEmpty ? "defaults" : parts.joined(separator: ", ") + } +} -@available(Network 0.1.0, *) -func makeParams(localEndpoint: Endpoint) -> ParametersBuilder { +// MARK: - UDP helpers + +private func makeUDPParams(localEndpoint: Endpoint) -> ParametersBuilder { let builder = ParametersBuilder.parameters { UDP() } .localEndpoint(localEndpoint) return builder } -@available(Network 0.1.0, *) -func makeConnection(to remote: Endpoint, localEndpoint: Endpoint) -> NetworkConnection { - NetworkConnection(to: remote, using: makeParams(localEndpoint: localEndpoint)) +private func makeUDPConnection(to remote: Endpoint, localEndpoint: Endpoint) -> NetworkConnection { + NetworkConnection(to: remote, using: makeUDPParams(localEndpoint: localEndpoint)) +} + +// MARK: - TCP helpers + +private func configureTCP(_ tcp: TCP, from settings: TCPSettings) -> TCP { + var mutableTCP = tcp + if settings.noDelay { + mutableTCP = mutableTCP.noDelay(true) + } + if settings.enableKeepalive { + // Values of 0 tell the kernel to use its defaults, so passing them through is safe. + mutableTCP = mutableTCP.keepalive( + idleTime: settings.keepaliveIdle, + count: settings.keepaliveCount, + interval: settings.keepaliveInterval + ) + } + return mutableTCP +} + +private func makeTCPParams(localEndpoint: Endpoint?, settings: TCPSettings) -> ParametersBuilder { + let tcp = configureTCP(TCP(), from: settings) + var builder = ParametersBuilder.parameters { tcp } + if let localEndpoint { + builder = builder.localEndpoint(localEndpoint) + } + return builder +} + +private func makeTCPConnection( + to remote: Endpoint, + localEndpoint: Endpoint?, + settings: TCPSettings +) -> NetworkConnection { + NetworkConnection(to: remote, using: makeTCPParams(localEndpoint: localEndpoint, settings: settings)) } @available(Network 0.1.0, *) -func parseEndpoints(ipString: String, port: UInt16, localPort: UInt16) -> (remote: Endpoint, local: Endpoint)? { +private func parseEndpoints(ipString: String, port: UInt16, localPort: UInt16) -> (remote: Endpoint, local: Endpoint)? { if let v4 = parseIPv4(ipString) { return ( Endpoint(address: v4, port: port), @@ -79,6 +135,102 @@ func parseEndpoints(ipString: String, port: UInt16, localPort: UInt16) -> (remot return nil } +// MARK: - Minimal POSIX TCP echo server +// +// Used for in-process TCP transfer mode. Listens on loopback, accepts a single +// connection, and either echoes or drains (depending on `echo`) until the +// client half-closes. + +final class TCPEchoServer: @unchecked Sendable { + struct ServerError: Error { + let step: String + let errno: Int32 + } + + private let port: UInt16 + private let echo: Bool + private let queue: DispatchQueue + private var listenFd: Int32 = -1 + + init(port: UInt16, echo: Bool) { + self.port = port + self.echo = echo + self.queue = DispatchQueue(label: "socket-transfer-tcp-echo-\(port)", qos: .userInitiated) + } + + func start() throws { + #if canImport(Glibc) + let type = CInt(SOCK_STREAM.rawValue) + #else + let type = SOCK_STREAM + #endif + let fd = socket(CInt(AF_INET), type, 0) + guard fd >= 0 else { throw ServerError(step: "socket", errno: errno) } + + var yes: CInt = 1 + _ = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout.size)) + + var addr = sockaddr_in() + addr.sin_family = sa_family_t(AF_INET) + addr.sin_port = port.bigEndian + addr.sin_addr = in_addr(s_addr: UInt32(0x7f00_0001).bigEndian) + #if canImport(Darwin) + addr.sin_len = UInt8(MemoryLayout.size) + #endif + let bound = withUnsafePointer(to: &addr) { ptr in + ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in + bind(fd, sa, socklen_t(MemoryLayout.size)) + } + } + guard bound == 0 else { + close(fd) + throw ServerError(step: "bind", errno: errno) + } + guard listen(fd, 1) == 0 else { + close(fd) + throw ServerError(step: "listen", errno: errno) + } + self.listenFd = fd + queue.async { [weak self] in self?.acceptLoop() } + } + + func stop() { + let fd = listenFd + listenFd = -1 + if fd >= 0 { + _ = shutdown(fd, CInt(SHUT_RDWR)) + close(fd) + } + } + + private func acceptLoop() { + let listen = listenFd + guard listen >= 0 else { return } + let connFd = accept(listen, nil, nil) + if connFd < 0 { return } + defer { close(connFd) } + + var buffer = [UInt8](repeating: 0, count: 64 * 1024) + while true { + let n = buffer.withUnsafeMutableBytes { raw -> Int in + guard let base = raw.baseAddress else { return -1 } + return read(connFd, base, raw.count) + } + if n <= 0 { break } + guard echo else { continue } + var written = 0 + while written < n { + let w = buffer.withUnsafeBytes { raw -> Int in + guard let base = raw.baseAddress else { return -1 } + return write(connFd, base.advanced(by: written), n - written) + } + if w <= 0 { return } + written += w + } + } + } +} + // MARK: - SocketTransfer @available(Network 0.1.0, *) @@ -89,7 +241,9 @@ final class SocketTransfer { let clientPort: UInt16 = 9100 let serverPort: UInt16 = 9101 - func run(iterations: Int, sendSize: Int, echo: Bool) -> Double { + // MARK: UDP round-trip / one-way (existing behavior) + + func runUDP(iterations: Int, sendSize: Int, echo: Bool) -> Double { let clientLocal = Endpoint(address: IPv4Address.loopback, port: clientPort) let clientRemote = Endpoint(address: IPv4Address.loopback, port: serverPort) let serverLocal = Endpoint(address: IPv4Address.loopback, port: serverPort) @@ -98,14 +252,14 @@ final class SocketTransfer { let payload = (0.. 1 ? "s" : "")") + print("Running UDP \(mode), transferring \(iterations) datagram\(iterations > 1 ? "s" : "")") let timestart = DispatchTime.now().uptimeNanoseconds let group = DispatchGroup() group.enter() - let client = makeConnection(to: clientRemote, localEndpoint: clientLocal) - let server = makeConnection(to: serverRemote, localEndpoint: serverLocal) + let client = makeUDPConnection(to: clientRemote, localEndpoint: clientLocal) + let server = makeUDPConnection(to: serverRemote, localEndpoint: serverLocal) nonisolated(unsafe) let done = { [group] in client.cancel() @@ -188,7 +342,7 @@ final class SocketTransfer { return Double(endTime - timestart) / Double(SocketTransfer.NSEC_PER_MSEC) / 1000.0 } - func sendToRemote( + func sendToRemoteUDP( ipString: String, port: UInt16, localPort: UInt16, @@ -202,13 +356,13 @@ final class SocketTransfer { let payload = (0.. 1 ? "s" : "") to \(ipString):\(port)") + print("Sending \(iterations) UDP datagram\(iterations > 1 ? "s" : "") to \(ipString):\(port)") let timestart = DispatchTime.now().uptimeNanoseconds let group = DispatchGroup() group.enter() - let client = makeConnection(to: endpoints.remote, localEndpoint: endpoints.local) + let client = makeUDPConnection(to: endpoints.remote, localEndpoint: endpoints.local) client.start() for i in 0.. Double { + let serverEndpoint = Endpoint(address: IPv4Address.loopback, port: serverPort) + + let mode = echo ? "round-trip echo" : "one-way send" + print( + "Running TCP \(mode), transferring \(iterations) message\(iterations > 1 ? "s" : ""), options: \(settings.describe())" + ) -func parseArg(_ arguments: [String], _ flag: String, parse: (String) -> T?) -> T? { - guard let index = arguments.firstIndex(of: flag), - arguments.count > index + 1 - else { return nil } - return parse(arguments[index + 1]) + let server = TCPEchoServer(port: serverPort, echo: echo) + do { + try server.start() + } catch { + print("TCPEchoServer failed to start on port \(serverPort): \(error)") + return 0 + } + defer { server.stop() } + + let payload = (0..= sendSize { + if collected == payload { + successCount += 1 + } else { + print("Echo mismatch at \(i)") + } + echoIteration(i + 1) + } else { + drain() + } + case .failure(let error): + print("Client echo receive failed at \(i): \(error)") + done() + } + } + } + drain() + } + echoIteration(0) + } else { + for i in 0.. Double { + guard let endpoints = parseEndpoints(ipString: ipString, port: port, localPort: localPort) else { + print("Invalid IP address: \(ipString)") + return 0 + } + + let payload = (0.. 1 ? "s" : "") to \(ipString):\(port), options: \(settings.describe())" + ) + let timestart = DispatchTime.now().uptimeNanoseconds + + let client = makeTCPConnection( + to: endpoints.remote, + localEndpoint: localPort != 0 ? endpoints.local : nil, + settings: settings + ) + client.start() + + for i in 0..(_ flag: String, parse: (String) -> T?) -> T? { + guard let index = arguments.firstIndex(of: flag), + arguments.count > index + 1 + else { return nil } + return parse(arguments[index + 1]) + } + + if let v: String = parseArg("-proto", parse: { $0 }) { + guard let parsed = TransferProtocol(rawValue: v.lowercased()) else { + print("Error: -proto must be 'udp' or 'tcp' (got '\(v)')") + exit(1) + } + proto = parsed + } + if arguments.contains("-tcp") { proto = .tcp } + if arguments.contains("-udp") { proto = .udp } + + if let v: Int = parseArg("-iterations", parse: { Int($0) }) { iterations = v } + if let v: Int = parseArg("-size", parse: { Int($0) }) { sendSize = v } + if let v: String = parseArg("-ip", parse: { $0 }) { remoteIP = v } + if let v: UInt16 = parseArg("-port", parse: { UInt16($0) }) { remotePort = v } + if let v: UInt16 = parseArg("-localport", parse: { UInt16($0) }) { localPort = v } if arguments.contains("-oneway") { echo = false } + // TCP-specific options — only effective when -proto tcp (or -tcp) is set. + if arguments.contains("-no-delay") { tcpSettings.noDelay = true } + if arguments.contains("-keepalive") { tcpSettings.enableKeepalive = true } + if let v: UInt32 = parseArg("-keepalive-idle", parse: { UInt32($0) }) { + tcpSettings.keepaliveIdle = v + tcpSettings.enableKeepalive = true + } + if let v: UInt32 = parseArg("-keepalive-interval", parse: { UInt32($0) }) { + tcpSettings.keepaliveInterval = v + tcpSettings.enableKeepalive = true + } + if let v: UInt32 = parseArg("-keepalive-count", parse: { UInt32($0) }) { + tcpSettings.keepaliveCount = v + tcpSettings.enableKeepalive = true + } + // MARK: - Run let socketTransfer = SocketTransfer() @@ -273,14 +613,27 @@ if #available(anyAppleOS 26, *) { print("Error: -port is required when using -ip") exit(1) } - print("Starting \(iterations) sends to \(remoteIP):\(remotePort)") - let totalTime = socketTransfer.sendToRemote( - ipString: remoteIP, - port: remotePort, - localPort: localPort, - iterations: iterations, - sendSize: sendSize - ) + print("Starting \(iterations) \(proto.rawValue.uppercased()) sends to \(remoteIP):\(remotePort)") + let totalTime: Double + switch proto { + case .udp: + totalTime = socketTransfer.sendToRemoteUDP( + ipString: remoteIP, + port: remotePort, + localPort: localPort, + iterations: iterations, + sendSize: sendSize + ) + case .tcp: + totalTime = socketTransfer.sendToRemoteTCP( + ipString: remoteIP, + port: remotePort, + localPort: localPort, + iterations: iterations, + sendSize: sendSize, + settings: tcpSettings + ) + } if totalTime > 0 { print("Finished all (\(iterations)) sends in \(totalTime) seconds") } else { @@ -288,8 +641,19 @@ if #available(anyAppleOS 26, *) { } } else { let mode = echo ? "round-trip echo" : "one-way send" - print("Starting \(iterations) \(mode) transfers") - let totalTime = socketTransfer.run(iterations: iterations, sendSize: sendSize, echo: echo) + print("Starting \(iterations) \(proto.rawValue.uppercased()) \(mode) transfers") + let totalTime: Double + switch proto { + case .udp: + totalTime = socketTransfer.runUDP(iterations: iterations, sendSize: sendSize, echo: echo) + case .tcp: + totalTime = socketTransfer.runTCP( + iterations: iterations, + sendSize: sendSize, + echo: echo, + settings: tcpSettings + ) + } if totalTime > 0 { print("Finished all (\(iterations)) transfers in \(totalTime) seconds") } else { diff --git a/Tests/SwiftNetworkTests/SwiftNetworkSocketTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkSocketTests.swift index 6a912cc..3c40555 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkSocketTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkSocketTests.swift @@ -14,14 +14,18 @@ import XCTest +#if canImport(Glibc) +import Glibc +#elseif canImport(Darwin) +internal import Darwin +#endif + #if !NETWORK_NO_SWIFT_QUIC -#if canImport(Network_Internal) -@_spi(Essentials) @_spi(ProtocolProvider) import Network -#else #if canImport(SwiftNetwork) @_spi(Essentials) @_spi(ProtocolProvider) @testable import SwiftNetwork -#endif +#elseif canImport(Network) +@_spi(Essentials) @_spi(ProtocolProvider) @testable import Network #endif @available(Network 0.1.0, *) @@ -1026,6 +1030,1070 @@ final class SwiftNetworkSocketTests: NetTestCase { c2.cancel() wait(for: [c1Done, c2Done], timeout: 5.0) } + + // MARK: - TCP / SocketStreamProtocol tests + // + // The stream tests need an actual server side because TCP is connection- + // oriented; the loopback datagram pattern of "two peers binding to each + // other's port" doesn't work. Each test spins up a TCPEchoServer (a tiny + // POSIX listener that accepts one connection and echoes everything back + // until EOF), then uses NetworkConnection as the client. + + private func makeTCPParams(localPort: UInt16 = 0, ipv6: Bool = false) -> ParametersBuilder { + var builder = ParametersBuilder.parameters { TCP() } + if localPort != 0 { + if ipv6 { + builder.parameters.localAddress = Endpoint(address: IPv6Address.loopback, port: localPort) + } else { + builder.parameters.localAddress = Endpoint(address: IPv4Address.loopback, port: localPort) + } + } + return builder + } + + private func makeTCPConnection( + toPort: UInt16, + localPort: UInt16 = 0, + ipv6: Bool = false, + readyExpectation: XCTestExpectation? = nil, + cancelExpectation: XCTestExpectation? = nil + ) -> NetworkConnection { + let remote: Endpoint + if ipv6 { + remote = Endpoint(address: IPv6Address.loopback, port: toPort) + } else { + remote = Endpoint(address: IPv4Address.loopback, port: toPort) + } + return NetworkConnection(to: remote, using: makeTCPParams(localPort: localPort, ipv6: ipv6)) + .onStateUpdate { _, state in + if case .ready = state { readyExpectation?.fulfill() } + if case .cancelled = state { cancelExpectation?.fulfill() } + } + } + + // MARK: - Connection lifecycle (TCP) + + func testTCPConnectionStateLifecycle() { + let server = TCPEchoServer(port: 12000) + try? server.start() + defer { server.stop() } + + let ready = XCTestExpectation(description: "tcp ready") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection( + toPort: 12000, + readyExpectation: ready, + cancelExpectation: cancelled + ) + conn.start() + wait(for: [ready], timeout: 5.0) + + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPConnectionRefusedDeliversFailure() { + // No listener on this port: connect should fail. + let failed = XCTestExpectation(description: "tcp failed") + let cancelled = XCTestExpectation(description: "tcp cancelled") + let ready = XCTestExpectation(description: "tcp ready") + ready.isInverted = true + + let remote = Endpoint(address: IPv4Address.loopback, port: 12001) + let conn = NetworkConnection(to: remote, using: makeTCPParams()) + .onStateUpdate { _, state in + if case .failed = state { failed.fulfill() } + if case .cancelled = state { cancelled.fulfill() } + if case .ready = state { ready.fulfill() } + } + conn.start() + wait(for: [failed, ready], timeout: 5.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // MARK: - Basic data path (TCP) + + func testTCPRoundTripEchoIPv4() { + let server = TCPEchoServer(port: 12010) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "tcp echo done") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12010, cancelExpectation: cancelled) + conn.start() + + let payload: [UInt8] = [1, 2, 3, 4, 5] + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + } + + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + switch result { + case .success(let msg): + XCTAssertEqual(msg.content, payload) + done.fulfill() + case .failure(let error): + XCTFail("receive failed: \(error)") + } + } + + wait(for: [done], timeout: 10.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPRoundTripEchoIPv6() { + let server = TCPEchoServer(port: 12120, ipv6: true) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "tcp ipv6 echo") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12120, ipv6: true, cancelExpectation: cancelled) + conn.start() + + let payload: [UInt8] = [10, 20, 30] + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + } + + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + switch result { + case .success(let msg): + XCTAssertEqual(msg.content, payload) + done.fulfill() + case .failure(let error): + XCTFail("ipv6 receive failed: \(error)") + } + } + + wait(for: [done], timeout: 10.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPSendSingleByte() { + let server = TCPEchoServer(port: 12030) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "tcp single byte") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12030, cancelExpectation: cancelled) + conn.start() + + conn.send(.message(content: [0xFF])) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + } + + conn.receive(atLeast: 1, atMost: 1) { result in + if case .success(let msg) = result { + XCTAssertEqual(msg.content, [0xFF]) + } else { + XCTFail("receive failed") + } + done.fulfill() + } + + wait(for: [done], timeout: 10.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPLargePayloadEcho() { + let server = TCPEchoServer(port: 12040) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "tcp large echo") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12040, cancelExpectation: cancelled) + conn.start() + + let payload = [UInt8](repeating: 0xAB, count: 8192) + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + } + + // Stream may deliver the echo across multiple receives — keep reading + // until we've collected the full payload. + nonisolated(unsafe) var collected: [UInt8] = [] + @Sendable func readMore() { + let need = payload.count - collected.count + conn.receive(atLeast: 1, atMost: need) { result in + switch result { + case .success(let msg): + if let bytes = msg.content { collected.append(contentsOf: bytes) } + if collected.count >= payload.count { + XCTAssertEqual(collected, payload) + done.fulfill() + } else { + readMore() + } + case .failure(let error): + XCTFail("receive failed: \(error)") + done.fulfill() + } + } + } + readMore() + + wait(for: [done], timeout: 15.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPMultipleSequentialMessages() { + let server = TCPEchoServer(port: 12050) + try? server.start() + defer { server.stop() } + + let allDone = XCTestExpectation(description: "all tcp messages") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12050, cancelExpectation: cancelled) + conn.start() + + let messages: [[UInt8]] = [ + [1], [2, 3], [4, 5, 6], [7, 8, 9, 10], [11, 12, 13, 14, 15], + ] + nonisolated(unsafe) var received = 0 + + @Sendable func sendAndReceive(_ i: Int) { + guard i < messages.count else { + allDone.fulfill() + return + } + let payload = messages[i] + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send \(i) failed: \(error)") } + } + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + if case .success(let msg) = result { + XCTAssertEqual(msg.content, payload) + received += 1 + } else { + XCTFail("receive \(i) failed") + } + sendAndReceive(i + 1) + } + } + + sendAndReceive(0) + + wait(for: [allDone], timeout: 15.0) + XCTAssertEqual(received, messages.count) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // MARK: - Volume / stress (TCP) + + func testTCPHighVolumeEcho100Messages() { + let server = TCPEchoServer(port: 12060) + try? server.start() + defer { server.stop() } + + let allDone = XCTestExpectation(description: "100 tcp echoes") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12060, cancelExpectation: cancelled) + conn.start() + + let total = 100 + nonisolated(unsafe) var success = 0 + + @Sendable func echoOnce(_ i: Int) { + guard i < total else { + allDone.fulfill() + return + } + let payload: [UInt8] = [UInt8(i % 256), UInt8((i + 1) % 256)] + + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send \(i): \(error)") } + } + + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + guard case .success(let msg) = result else { + XCTFail("recv \(i)") + return + } + XCTAssertEqual(msg.content, payload) + success += 1 + echoOnce(i + 1) + } + } + echoOnce(0) + + wait(for: [allDone], timeout: 60.0) + XCTAssertEqual(success, total) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPVaryingPayloadSizes() { + let server = TCPEchoServer(port: 12070) + try? server.start() + defer { server.stop() } + + let allDone = XCTestExpectation(description: "varying tcp sizes") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12070, cancelExpectation: cancelled) + conn.start() + + let sizes = [1, 100, 500, 1000, 1400, 4096, 10] + let messages = sizes.map { [UInt8](repeating: 0xAA, count: $0) } + nonisolated(unsafe) var received = 0 + + @Sendable func sendAndReceive(_ i: Int) { + guard i < messages.count else { + allDone.fulfill() + return + } + let payload = messages[i] + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send \(i): \(error)") } + } + // Drain exactly payload.count bytes back from the echo server. + nonisolated(unsafe) var collected: [UInt8] = [] + @Sendable func drain() { + let need = payload.count - collected.count + conn.receive(atLeast: 1, atMost: need) { result in + if case .success(let msg) = result, let bytes = msg.content { + collected.append(contentsOf: bytes) + if collected.count >= payload.count { + XCTAssertEqual(collected, payload) + received += 1 + sendAndReceive(i + 1) + } else { + drain() + } + } else { + XCTFail("receive \(i) failed") + sendAndReceive(i + 1) + } + } + } + drain() + } + sendAndReceive(0) + + wait(for: [allDone], timeout: 30.0) + XCTAssertEqual(received, messages.count) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // MARK: - Backpressure / lifecycle (TCP) + + func testTCPBurstSendsThenDrain() { + let server = TCPEchoServer(port: 12080) + try? server.start() + defer { server.stop() } + + let allDrained = XCTestExpectation(description: "tcp drained") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12080, cancelExpectation: cancelled) + conn.start() + + let messageCount = 50 + let perMessage: [UInt8] = [0xDE, 0xAD] + for i in 0..= totalBytes { + allDrained.fulfill() + } else { + drain() + } + } else { + XCTFail("receive failed at \(collected.count)") + } + } + } + drain() + + wait(for: [allDrained], timeout: 30.0) + XCTAssertEqual(collected.count, totalBytes) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPDelayedConsumer() { + let server = TCPEchoServer(port: 12090) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "delayed tcp consumer") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12090, cancelExpectation: cancelled) + conn.start() + + let payload: [UInt8] = [0xDE, 0xAD, 0xBE, 0xEF] + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + } + + // Delay the receive by 500ms — bytes should still be buffered. + DispatchQueue.main.asyncAfter(deadline: .now() + 0.5) { + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + if case .success(let msg) = result { + XCTAssertEqual(msg.content, payload) + } else { + XCTFail("delayed receive failed") + } + done.fulfill() + } + } + + wait(for: [done], timeout: 10.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPPayloadWithAllByteValues() { + let server = TCPEchoServer(port: 12100) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "tcp all bytes") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12100, cancelExpectation: cancelled) + conn.start() + + let payload = (0...255).map { UInt8($0) } + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + } + + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + if case .success(let msg) = result { + XCTAssertEqual(msg.content, payload) + } else { + XCTFail("receive failed") + } + done.fulfill() + } + + wait(for: [done], timeout: 10.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPCancelBeforeStart() { + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let remote = Endpoint(address: IPv4Address.loopback, port: 12110) + let conn = NetworkConnection(to: remote, using: makeTCPParams()) + .onStateUpdate { _, state in + if case .cancelled = state { cancelled.fulfill() } + } + + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // MARK: - Echo round trips (TCP) + + func testTCPEcho10RoundTrips() { + let server = TCPEchoServer(port: 12130) + try? server.start() + defer { server.stop() } + + let allDone = XCTestExpectation(description: "10 tcp echoes") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 12130, cancelExpectation: cancelled) + conn.start() + + nonisolated(unsafe) var success = 0 + @Sendable func echoOnce(_ i: Int) { + guard i < 10 else { + allDone.fulfill() + return + } + let payload: [UInt8] = [UInt8(i), UInt8(i &* 2), UInt8(i &* 3)] + + conn.send(.message(content: payload)) { result in + if case .failure(let error) = result { XCTFail("send \(i): \(error)") } + } + + conn.receive(atLeast: payload.count, atMost: payload.count) { result in + guard case .success(let msg) = result else { + XCTFail("recv \(i)") + return + } + XCTAssertEqual(msg.content, payload) + success += 1 + echoOnce(i + 1) + } + } + echoOnce(0) + + wait(for: [allDone], timeout: 30.0) + XCTAssertEqual(success, 10) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // MARK: - Shared base-class behaviour (SocketProtocolBase) + // + // These tests exercise paths that run through the shared base: + // bindSocket, cancelWriteSource, setupReadSource/WriteSource, + // and the teardown/reinit lifecycle — verified through both + // SocketDatagramProtocol (UDP) and SocketStreamProtocol (TCP). + + // Verifies that a cancelled connection can be garbage-collected promptly + // (exercises deinit / teardown paths on both concrete types). + func testUDPTeardownIsClean() { + let cancelled = XCTestExpectation(description: "udp cancelled") + let conn = makeConnection(toPort: 13000, localPort: 13001, cancelExpectation: cancelled) + conn.start() + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + // If teardown is wrong (e.g. write source not resumed before cancel) + // the DispatchSource will trap on dealloc — reaching here means it is clean. + } + + // Cancelling before the socket even has a chance to become writable must + // not crash (exercises the write-source suspend path in cancelWriteSource). + func testUDPCancelImmediatelyAfterStart() { + let cancelled = XCTestExpectation(description: "cancelled immediately") + let conn = makeConnection(toPort: 13020, localPort: 13021, cancelExpectation: cancelled) + conn.start() + // Cancel on the very next runloop tick — write source may still be + // suspended (never received EAGAIN). + DispatchQueue.main.async { conn.cancel() } + wait(for: [cancelled], timeout: 5.0) + } + + func testTCPCancelImmediatelyAfterStart() { + let cancelled = XCTestExpectation(description: "tcp cancelled immediately") + let remote = Endpoint(address: IPv4Address.loopback, port: 13030) + let conn = NetworkConnection(to: remote, using: makeTCPParams()) + .onStateUpdate { _, state in + if case .cancelled = state { cancelled.fulfill() } + } + conn.start() + DispatchQueue.main.async { conn.cancel() } + wait(for: [cancelled], timeout: 5.0) + } + + // Exercises bindSocket through the base class for both directions. + func testUDPBindToExplicitLocalPort() { + let ready = XCTestExpectation(description: "bound udp ready") + let cancelled = XCTestExpectation(description: "bound udp cancelled") + + // makeConnection already sets a localPort — this verifies bindSocket + // in the base class succeeds without error. + let remote = Endpoint(address: IPv4Address.loopback, port: 13040) + let c1 = NetworkConnection(to: remote, using: makeUDPParams(localPort: 13041)) + .onStateUpdate { _, state in + if case .ready = state { ready.fulfill() } + if case .cancelled = state { cancelled.fulfill() } + } + c1.start() + wait(for: [ready], timeout: 5.0) + c1.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // Verifies that the write-source fires and triggerOutboundRoomAvailable + // is called after a datagram send — the upper layer gets the room event. + func testUDPOutboundRoomAvailableAfterSend() { + let sent = XCTestExpectation(description: "sent") + let c1Done = XCTestExpectation(description: "c1 cancelled") + let c2Done = XCTestExpectation(description: "c2 cancelled") + + let c1 = makeConnection(toPort: 13050, localPort: 13051, cancelExpectation: c1Done) + let c2 = makeConnection(toPort: 13051, localPort: 13050, cancelExpectation: c2Done) + c1.start() + c2.start() + + c1.send(.message(content: [0x01, 0x02])) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + sent.fulfill() + } + + wait(for: [sent], timeout: 5.0) + c1.cancel() + c2.cancel() + wait(for: [c1Done, c2Done], timeout: 5.0) + } + + // Verifies write-source / triggerOutboundRoomAvailable for TCP. + func testTCPOutboundRoomAvailableAfterSend() { + let server = TCPEchoServer(port: 13060) + try? server.start() + defer { server.stop() } + + let sent = XCTestExpectation(description: "tcp sent") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: 13060, cancelExpectation: cancelled) + conn.start() + + conn.send(.message(content: [0xAA])) { result in + if case .failure(let error) = result { XCTFail("send failed: \(error)") } + sent.fulfill() + } + + wait(for: [sent], timeout: 5.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // Multiple sequential start/cancel cycles — verifies repeated teardown + // goes through the base-class cleanup without double-freeing sources. + func testUDPMultipleCancelCycles() { + for cycle in 0..<3 { + let cancelled = XCTestExpectation(description: "cycle \(cycle) cancelled") + let port = UInt16(13070 + cycle * 2) + let conn = makeConnection(toPort: port, localPort: port + 1, cancelExpectation: cancelled) + conn.start() + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + } + + func testTCPMultipleCancelCycles() { + for cycle in 0..<3 { + let port = UInt16(13080 + cycle) + let server = TCPEchoServer(port: port) + try? server.start() + + let ready = XCTestExpectation(description: "tcp cycle \(cycle) ready") + let cancelled = XCTestExpectation(description: "tcp cycle \(cycle) cancelled") + let conn = makeTCPConnection( + toPort: port, + readyExpectation: ready, + cancelExpectation: cancelled + ) + conn.start() + wait(for: [ready], timeout: 5.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + server.stop() + } + } + + // MARK: - Receive across segment boundaries (regression) + // + // Reproduces a stall: when receive(atLeast:) requests more bytes than + // arrive in the first TCP segment, the bottom delivers the first segment, + // the consumer's receiveStreamData returns nil (< minimumBytes), and the + // read source is left suspended. If the read source is only re-armed on a + // *successful* drain, the bytes that arrive in a later segment are never + // read and the receive never completes. + // + // SplitSendServer pushes `total` bytes as two NODELAY segments separated by + // a gap, so the first segment is processed (and the source suspended) + // before the rest arrives. With a correct implementation this completes + // promptly; with the stall it times out. + func testTCPReceiveAtLeastSpanningTwoSegments() { + let port: UInt16 = 12200 + let firstChunk = 4 + let total = 16 // atLeast spans both segments + + let server = SplitSendServer(port: port, firstChunk: firstChunk, total: total, gap: 0.5) + try? server.start() + defer { server.stop() } + + let done = XCTestExpectation(description: "received full payload across segments") + let cancelled = XCTestExpectation(description: "tcp cancelled") + + let conn = makeTCPConnection(toPort: port, cancelExpectation: cancelled) + conn.start() + + conn.receive(atLeast: total, atMost: total) { result in + switch result { + case .success(let msg): + XCTAssertEqual(msg.content?.count, total) + done.fulfill() + case .failure(let error): + XCTFail("receive failed: \(error)") + } + } + + wait(for: [done], timeout: 10.0) + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } + + // MARK: - EOF handling (regression) + + private static func processCPUSeconds() -> Double { + var usage = rusage() + #if canImport(Darwin) + getrusage(RUSAGE_SELF, &usage) + #else + getrusage(RUSAGE_SELF.rawValue, &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 + } + + // Regression: after the peer half-closes (EOF), the read source must stop + // firing. Previously it stayed armed and EOF keeps the descriptor readable, + // so the read handler re-fired forever and pegged a CPU core. We detect the + // spin by measuring process CPU consumed during an idle window after EOF — + // a busy-loop burns ~a full core; correct behavior consumes ~nothing. + func testTCPNoBusyLoopAfterEOF() { + let server = ClosingServer(port: 12210, payload: [1, 2, 3, 4]) + try? server.start() + defer { server.stop() } + + let received = XCTestExpectation(description: "received payload") + let cancelled = XCTestExpectation(description: "tcp cancelled") + let conn = makeTCPConnection(toPort: 12210, cancelExpectation: cancelled) + conn.start() + + conn.receive(atLeast: 4, atMost: 4) { result in + if case .success(let msg) = result { + XCTAssertEqual(msg.content, [1, 2, 3, 4]) + } else { + XCTFail("receive failed") + } + received.fulfill() + } + wait(for: [received], timeout: 5.0) + + // Give the bottom a moment to observe the peer's FIN (EOF), then sample + // CPU across an otherwise-idle second. + Thread.sleep(forTimeInterval: 0.3) + let before = Self.processCPUSeconds() + Thread.sleep(forTimeInterval: 1.0) + let cpu = Self.processCPUSeconds() - before + XCTAssertLessThan( + cpu, + 0.3, + "read source appears to busy-loop after EOF (consumed \(cpu)s CPU while idle)" + ) + + conn.cancel() + wait(for: [cancelled], timeout: 5.0) + } +} + +// MARK: - TCP Echo Server helper +// +// Minimal POSIX TCP echo server used by the SocketStreamProtocol tests. +// Listens on loopback, accepts a single connection, and echoes everything +// it receives back to the sender until the peer half-closes (FIN). + +private final class TCPEchoServer: @unchecked Sendable { + private let port: UInt16 + private let ipv6: Bool + private let queue: DispatchQueue + private var listenFd: Int32 = -1 + + init(port: UInt16, ipv6: Bool = false) { + self.port = port + self.ipv6 = ipv6 + self.queue = DispatchQueue(label: "tcp-echo-server-\(port)", qos: .userInitiated) + } + + func start() throws { + #if canImport(Glibc) + let family = ipv6 ? CInt(AF_INET6) : CInt(AF_INET) + let type = CInt(SOCK_STREAM.rawValue) + #else + let family = ipv6 ? CInt(AF_INET6) : CInt(AF_INET) + let type = SOCK_STREAM + #endif + let fd = socket(family, type, 0) + guard fd >= 0 else { throw NSError(domain: "TCPEchoServer", code: Int(errno)) } + + var yes: CInt = 1 + _ = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout.size)) + + let bound: CInt + if ipv6 { + var addr = sockaddr_in6() + addr.sin6_family = sa_family_t(AF_INET6) + addr.sin6_port = port.bigEndian + addr.sin6_addr = in6addr_loopback + #if canImport(Darwin) + addr.sin6_len = UInt8(MemoryLayout.size) + #endif + bound = withUnsafePointer(to: &addr) { ptr in + ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in + bind(fd, sa, socklen_t(MemoryLayout.size)) + } + } + } else { + var addr = sockaddr_in() + addr.sin_family = sa_family_t(AF_INET) + addr.sin_port = port.bigEndian + addr.sin_addr = in_addr(s_addr: UInt32(0x7f00_0001).bigEndian) + #if canImport(Darwin) + addr.sin_len = UInt8(MemoryLayout.size) + #endif + bound = withUnsafePointer(to: &addr) { ptr in + ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in + bind(fd, sa, socklen_t(MemoryLayout.size)) + } + } + } + guard bound == 0 else { + close(fd) + throw NSError(domain: "TCPEchoServer.bind", code: Int(errno)) + } + + guard listen(fd, 1) == 0 else { + close(fd) + throw NSError(domain: "TCPEchoServer.listen", code: Int(errno)) + } + + self.listenFd = fd + queue.async { [weak self] in self?.acceptLoop() } + } + + func stop() { + let fd = listenFd + listenFd = -1 + if fd >= 0 { + _ = shutdown(fd, CInt(SHUT_RDWR)) + close(fd) + } + } + + private func acceptLoop() { + let listen = listenFd + guard listen >= 0 else { return } + let connFd = accept(listen, nil, nil) + if connFd < 0 { return } + defer { close(connFd) } + + var buffer = [UInt8](repeating: 0, count: 16 * 1024) + while true { + let n = buffer.withUnsafeMutableBytes { raw -> Int in + guard let base = raw.baseAddress else { return -1 } + return read(connFd, base, raw.count) + } + if n <= 0 { break } + var written = 0 + while written < n { + let w = buffer.withUnsafeBytes { raw -> Int in + guard let base = raw.baseAddress else { return -1 } + return write(connFd, base.advanced(by: written), n - written) + } + if w <= 0 { return } + written += w + } + } + } +} + +// MARK: - Split-send Server helper +// +// POSIX TCP server that pushes a payload as two separate NODELAY segments with +// a gap between them, without reading anything from the client. Used to force +// a receive(atLeast:) to span more than one read event. + +private final class SplitSendServer: @unchecked Sendable { + private let port: UInt16 + private let firstChunk: Int + private let total: Int + private let gap: TimeInterval + private let queue: DispatchQueue + private var listenFd: Int32 = -1 + + init(port: UInt16, firstChunk: Int, total: Int, gap: TimeInterval) { + self.port = port + self.firstChunk = firstChunk + self.total = total + self.gap = gap + self.queue = DispatchQueue(label: "split-send-server-\(port)", qos: .userInitiated) + } + + func start() throws { + #if canImport(Glibc) + let type = CInt(SOCK_STREAM.rawValue) + #else + let type = SOCK_STREAM + #endif + let fd = socket(CInt(AF_INET), type, 0) + guard fd >= 0 else { throw NSError(domain: "SplitSendServer", code: Int(errno)) } + + var yes: CInt = 1 + _ = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout.size)) + + var addr = sockaddr_in() + addr.sin_family = sa_family_t(AF_INET) + addr.sin_port = port.bigEndian + addr.sin_addr = in_addr(s_addr: UInt32(0x7f00_0001).bigEndian) + #if canImport(Darwin) + addr.sin_len = UInt8(MemoryLayout.size) + #endif + let bound = withUnsafePointer(to: &addr) { ptr in + ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in + bind(fd, sa, socklen_t(MemoryLayout.size)) + } + } + guard bound == 0 else { + close(fd) + throw NSError(domain: "SplitSendServer.bind", code: Int(errno)) + } + guard listen(fd, 1) == 0 else { + close(fd) + throw NSError(domain: "SplitSendServer.listen", code: Int(errno)) + } + + self.listenFd = fd + queue.async { [weak self] in self?.serve() } + } + + func stop() { + let fd = listenFd + listenFd = -1 + if fd >= 0 { + _ = shutdown(fd, CInt(SHUT_RDWR)) + close(fd) + } + } + + private func serve() { + let listen = listenFd + guard listen >= 0 else { return } + let connFd = accept(listen, nil, nil) + if connFd < 0 { return } + defer { close(connFd) } + + // Disable Nagle so each write leaves as its own segment. + var one: CInt = 1 + _ = setsockopt(connFd, CInt(IPPROTO_TCP), TCP_NODELAY, &one, socklen_t(MemoryLayout.size)) + + let payload = [UInt8](repeating: 0xAB, count: total) + + func sendAll(_ bytes: ArraySlice) { + let array = Array(bytes) + var off = 0 + while off < array.count { + let w = array.withUnsafeBytes { raw -> Int in + write(connFd, raw.baseAddress!.advanced(by: off), raw.count - off) + } + if w <= 0 { break } + off += w + } + } + + // First segment, then a gap so the client processes it (and, if buggy, + // suspends its read source) before the remainder arrives. + sendAll(payload.prefix(firstChunk)) + Thread.sleep(forTimeInterval: gap) + sendAll(payload.suffix(total - firstChunk)) + + // Hold the connection open long enough for the client to finish. + Thread.sleep(forTimeInterval: 1.0) + } +} + +// MARK: - Closing Server helper +// +// POSIX TCP server that sends a payload and immediately closes the connection +// (sending FIN), so the client observes EOF. Used to exercise the bottom's +// end-of-stream handling. + +private final class ClosingServer: @unchecked Sendable { + private let port: UInt16 + private let payload: [UInt8] + private let queue: DispatchQueue + private var listenFd: Int32 = -1 + + init(port: UInt16, payload: [UInt8]) { + self.port = port + self.payload = payload + self.queue = DispatchQueue(label: "closing-server-\(port)", qos: .userInitiated) + } + + func start() throws { + #if canImport(Glibc) + let type = CInt(SOCK_STREAM.rawValue) + #else + let type = SOCK_STREAM + #endif + let fd = socket(CInt(AF_INET), type, 0) + guard fd >= 0 else { throw NSError(domain: "ClosingServer", code: Int(errno)) } + + var yes: CInt = 1 + _ = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, socklen_t(MemoryLayout.size)) + + var addr = sockaddr_in() + addr.sin_family = sa_family_t(AF_INET) + addr.sin_port = port.bigEndian + addr.sin_addr = in_addr(s_addr: UInt32(0x7f00_0001).bigEndian) + #if canImport(Darwin) + addr.sin_len = UInt8(MemoryLayout.size) + #endif + let bound = withUnsafePointer(to: &addr) { ptr in + ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { sa in + bind(fd, sa, socklen_t(MemoryLayout.size)) + } + } + guard bound == 0 else { + close(fd) + throw NSError(domain: "ClosingServer.bind", code: Int(errno)) + } + guard listen(fd, 1) == 0 else { + close(fd) + throw NSError(domain: "ClosingServer.listen", code: Int(errno)) + } + + self.listenFd = fd + queue.async { [weak self] in self?.serve() } + } + + func stop() { + let fd = listenFd + listenFd = -1 + if fd >= 0 { + _ = shutdown(fd, CInt(SHUT_RDWR)) + close(fd) + } + } + + private func serve() { + let listen = listenFd + guard listen >= 0 else { return } + let connFd = accept(listen, nil, nil) + if connFd < 0 { return } + + var off = 0 + while off < payload.count { + let w = payload.withUnsafeBytes { raw -> Int in + write(connFd, raw.baseAddress!.advanced(by: off), raw.count - off) + } + if w <= 0 { break } + off += w + } + // Close immediately, sending FIN so the client sees EOF. + close(connFd) + } } #endif