diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift index 95bf494..e70eac0 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift @@ -235,12 +235,16 @@ final class EndpointFlow: CustomDebugStringConvertible { case .stream(let flow): while true { if let readRequest = self.readRequests.first { - if let content = flow.read( + if let streamRead = flow.read( minimumBytes: readRequest.minimumBytes, maximumBytes: readRequest.maximumBytes ) { - // TODO: Get the actual metadata - readRequest.complete(content: content, isComplete: false, isFinal: true) + // TODO: Get the remaining per-frame metadata + readRequest.complete( + content: streamRead.content, + isComplete: streamRead.isComplete, + isFinal: true + ) // TODO: This is not efficient. Probably better to use an ArraySlice here self.readRequests.removeFirst() } else { diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift index 0dd8bf6..ea0b5c1 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift @@ -528,7 +528,8 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol [UInt8]? { + /// Reads buffered stream bytes, reporting whether the read reached the end of the stream. + func read(minimumBytes: Int, maximumBytes: Int) -> (content: [UInt8], isComplete: Bool)? { fromExternal { do throws(NetworkError) { guard @@ -542,9 +543,16 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol 0 { _ = Deserializer.deserialize(&frame, claim: false) { read throws(DeserializationError) in try read.buffer(&buffer, length: length) @@ -558,7 +566,10 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol) { self._bytes = bytes self.buffer = .bytes diff --git a/Sources/SwiftNetwork/Protocols/FrameArray.swift b/Sources/SwiftNetwork/Protocols/FrameArray.swift index dd0bd45..1618a48 100644 --- a/Sources/SwiftNetwork/Protocols/FrameArray.swift +++ b/Sources/SwiftNetwork/Protocols/FrameArray.swift @@ -238,6 +238,14 @@ public struct FrameArray: ~Copyable { let claimed = frames[0].claim(fromStart: 0, fromEnd: partialBytesToKeep) precondition(claimed) + // The flag belongs with the last byte, which is the tail being kept. `swap` + // exchanges whole frame values, so it would carry the flag out with the prefix + // instead. Move it first, while each name still refers to what it says. + if frames[0].connectionComplete { + frames[0].connectionComplete = false + splitFrame.connectionComplete = true + } + // Swap the new frame with the original frame swap(&splitFrame, &frames[0]) @@ -267,15 +275,15 @@ public struct FrameArray: ~Copyable { return length } + /// Marks the end of the stream, on the last frame only. + public mutating func markEndOfStream() { + if frames.isEmpty { return } + frames[frames.count - 1].connectionComplete = true + } + + /// Whether this array ends the stream, which is a question about its final frame. public var connectionComplete: Bool { - var connectionComplete = false - iterateImmutableFrames { frame in - if frame.connectionComplete { - connectionComplete = true - return false - } - return true - } - return connectionComplete + if frames.isEmpty { return false } + return frames[frames.count - 1].connectionComplete } } diff --git a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift index c598c82..e94a650 100644 --- a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift +++ b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift @@ -429,6 +429,8 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC private var isConnecting = false private var inputSourceSuspended = false private var inputFinished = false + /// Whether the end-of-stream marker has been delivered to the consumer. + private var inputFinishedDelivered = false private var outputFinished = false private var pendingDisconnect = false private var incomingFrames = FrameArray() @@ -571,20 +573,33 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC // 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() + if incomingFrames.isEmpty { + // Nothing buffered and nothing more coming, so report the end of the stream. + // It is synthesized rather than carried on a queued frame, so nothing has to + // be allocated to hold the flag once the queue has drained. + if inputFinished, !inputFinishedDelivered { + inputFinishedDelivered = true + return FrameArray(frame: Frame(count: 0, connectionComplete: true)) } + + resumeReadingIfSuspended() + return nil + } + + // Either hold the consumer until its minimum can be met or if the peer has + // half-closed no more will arrive, so what is buffered is all it is ever going to get. + guard incomingFrames.unclaimedLength >= minimumBytes || inputFinished else { + resumeReadingIfSuspended() return nil } + let result = incomingFrames.drainArray(maximumByteCount: maximumBytes) + // The flag sits on whichever frame carries the last byte, so a drain that takes it counts + // as delivery. Without this the branch above would synthesize a second, empty end of stream + // on the next receive. + if result.connectionComplete { + inputFinishedDelivered = true + } if inputSourceSuspended, incomingFrames.unclaimedLength < maximumInputSize { inputSourceSuspended = false dispatchReadSource?.resume() @@ -592,6 +607,17 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC return result } + /// Resumes the read source if it was suspended on the high-water mark. + /// + /// The consumer wants more than is buffered, so reading has to continue past the soft cap or a + /// large minimum would deadlock. Not once the peer has half-closed: the source is cancelled by + /// then and no further event can arrive. + private func resumeReadingIfSuspended() { + guard inputSourceSuspended, !inputFinished else { return } + inputSourceSuspended = false + dispatchReadSource?.resume() + } + public func getOutboundStreamDataRoomAvailable() throws(NetworkError) -> Int { let pending = pendingOutputFrames.unclaimedLength if pending >= maximumOutputSize { return 0 } @@ -756,7 +782,7 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC switch result { case .processed(let bytesRead): if bytesRead == 0 { - // Stream EOF — mark the next frame as connectionComplete. + // Stream EOF — the queued bytes are marked below. reachedEOF = true } else { let frame = Frame(copyBuffer: UnsafeRawBufferPointer(start: readBuffer, count: bytesRead)) @@ -773,11 +799,10 @@ public final class SocketStreamProtocol: BottomStreamProtocol, ProtocolInstanceC 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) + // Mark the bytes already queued, so the read that returns the last of them also + // reports the end of the stream. With nothing queued, `receiveStreamData` + // synthesizes the marker instead. + incomingFrames.markEndOfStream() receivedAny = true // The stream is finished; stop the read source. EOF keeps the // descriptor readable, so leaving it armed would spin forever. diff --git a/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift index 603e5a0..81535e3 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift @@ -657,6 +657,62 @@ final class SwiftNetworkConnectionTests: NetTestCase { ) } + // A stream send marked `isComplete` must arrive marked complete. + func testStreamReceiveReportsEndOfStream() { + let group = DispatchGroup() + group.enter() + let c1 = NetworkConnection( + to: Endpoint(address: IPv4Address.loopback, port: 7878), + using: .parameters { + NoTransport { + StreamBridge() + } + }.localEndpoint(Endpoint(address: IPv4Address.loopback, port: 7877)) + ) + .onStateUpdate { _, state in + if case .cancelled = state { group.leave() } + } + + group.enter() + let c2 = NetworkConnection( + to: Endpoint(address: IPv4Address.loopback, port: 7877), + using: .parameters { + NoTransport { + StreamBridge() + } + }.localEndpoint(Endpoint(address: IPv4Address.loopback, port: 7878)) + ) + .onStateUpdate { _, state in + if case .cancelled = state { group.leave() } + } + + c1.start() + c2.start() + + c1.send(.message(content: [1, 2, 3], isComplete: true)) { result in + if case .failure(let error) = result { + XCTFail("send failed with error \(error)") + } + } + + c2.receive(atLeast: 1, atMost: Int.max) { result in + switch result { + case .success(let message): + XCTAssertEqual(message.content, [1, 2, 3]) + XCTAssertTrue(message.isComplete, "end of stream was not reported to the receiver") + case .failure(let error): + XCTFail("receive failed with error \(error)") + } + c1.cancel() + c2.cancel() + } + + XCTAssertEqual( + group.wait(timeout: DispatchTime.now() + .seconds(5)), + DispatchTimeoutResult.success + ) + } + #if HAS_SWIFTTLS_RECORD func testTLSNoTransportDataPath() { let group = DispatchGroup() diff --git a/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift index 4572631..f2e9a6c 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkFrameArrayTests.swift @@ -569,6 +569,47 @@ final class SwiftNetworkFrameArrayTests: NetTestCase { array.finalizeAllFramesAsFailed() } + func testMarkEndOfStreamOnlyMarksTheFinalFrame() { + var array = FrameArray(frame: Frame(copyBuffer: Array(repeating: 1, count: 100))) + array.add(frame: Frame(copyBuffer: Array(repeating: 2, count: 100))) + array.add(frame: Frame(copyBuffer: Array(repeating: 3, count: 100))) + array.markEndOfStream() + + var first = array.drainArray(maximumByteCount: 100) + XCTAssertEqual(first.unclaimedLength, 100) + XCTAssertFalse(first.connectionComplete, "end of stream reported with 200 bytes still queued") + + var second = array.drainArray(maximumByteCount: 100) + XCTAssertEqual(second.unclaimedLength, 100) + XCTAssertFalse(second.connectionComplete, "end of stream reported with 100 bytes still queued") + + var third = array.drainArray(maximumByteCount: 100) + XCTAssertEqual(third.unclaimedLength, 100) + XCTAssertTrue(third.connectionComplete, "end of stream not reported with the last bytes") + + first.finalizeAllFramesAsFailed() + second.finalizeAllFramesAsFailed() + third.finalizeAllFramesAsFailed() + } + + func testDrainByteCountSplitLeavesEndOfStreamWithTheTrailingBytes() { + var frame = Frame(copyBuffer: Array(0..<20)) + frame.connectionComplete = true + var array = FrameArray(frame: frame) + + var drained = array.drainArray(maximumByteCount: 12) + XCTAssertEqual(drained.unclaimedLength, 12) + XCTAssertFalse(drained.connectionComplete, "end of stream reported before the last byte") + XCTAssertEqual(array.unclaimedLength, 8) + XCTAssertTrue(array.connectionComplete, "end of stream lost from the remaining bytes") + + var rest = array.drainArray(maximumByteCount: 8) + XCTAssertTrue(rest.connectionComplete) + drained.finalizeAllFramesAsFailed() + rest.finalizeAllFramesAsFailed() + array.finalizeAllFramesAsFailed() + } + func testDrainByteCountSplitMajoritySentToNewArray() { // Split where the majority of the split frame's bytes go to the new array // Drain 18 of 20 bytes from a single frame @@ -664,4 +705,26 @@ final class SwiftNetworkFrameArrayTests: NetTestCase { XCTAssertEqual(collectBytes(array), [4, 5, 6, 1, 2, 3]) array.finalizeAllFramesAsFailed() } + func testConnectionCompleteIsAboutTheFinalFrame() { + var head = Frame(copyBuffer: [1, 2, 3] as [UInt8]) + head.connectionComplete = true + var array = FrameArray(frame: head) + XCTAssertTrue(array.connectionComplete) + + array.add(frame: Frame(copyBuffer: [4, 5, 6] as [UInt8])) + XCTAssertFalse(array.connectionComplete, "bytes follow the marked frame, so the stream has not ended") + + var tail = Frame(copyBuffer: [7, 8, 9] as [UInt8]) + tail.connectionComplete = true + array.add(frame: tail) + XCTAssertTrue(array.connectionComplete) + + array.finalizeAllFramesAsFailed() + } + + func testConnectionCompleteIsFalseWhenEmpty() { + var array = FrameArray() + XCTAssertFalse(array.connectionComplete) + array.finalizeAllFramesAsFailed() + } }