diff --git a/Sources/SwiftNetwork/Connection/Connection.swift b/Sources/SwiftNetwork/Connection/Connection.swift index 51e0852b..3cc0c4c2 100644 --- a/Sources/SwiftNetwork/Connection/Connection.swift +++ b/Sources/SwiftNetwork/Connection/Connection.swift @@ -2106,6 +2106,21 @@ extension NetworkChannel where ApplicationProtocol: StreamProtocol { public let isComplete: Bool } + public struct StreamSpanMessage: ~Escapable { + @_lifetime(borrow content) + init(content: RawSpan? = nil, offset: Int = 0, isComplete: Bool = false, lastChunkOfBatch: Bool = false) { + self.content = content + self.offset = offset + self.isComplete = isComplete + self.lastChunkOfBatch = lastChunkOfBatch + } + + public let content: RawSpan? + public let offset: Int + public let isComplete: Bool + public let lastChunkOfBatch: Bool + } + public func send(_ message: StreamMessage, completion: (@Sendable (Result) -> Void)? = nil) { let endpointFlow = self.endpointFlow endpointFlow.async { @@ -2141,15 +2156,48 @@ extension NetworkChannel where ApplicationProtocol: StreamProtocol { atMost maxBytes: Int, completion: @escaping @Sendable (Result) -> Void ) { - let readRequest = ReadRequest(minimumBytes: minBytes, maximumBytes: maxBytes, maximumFrames: Int.max) { - (content, isComplete, isFinal, error) in - if let error = error { - completion(.failure(error)) - } else { - completion(.success(.message(content: content, isComplete: isComplete))) + let endpointFlow = self.endpointFlow + endpointFlow.async { + let readRequest = ReadRequest(minimumBytes: minBytes, maximumBytes: maxBytes) { + (content, isComplete, isFinal, error) in + if let error = error { + completion(.failure(error)) + } else { + completion(.success(.message(content: content, isComplete: isComplete))) + } } + self.endpointFlow.addReadRequestOnContext(readRequest) + } + } + + public func receive( + atLeast minBytes: Int, + atMost maxBytes: Int, + maximumChunks: Int, + completion: @escaping @Sendable (Result) -> Void + ) { + let endpointFlow = self.endpointFlow + endpointFlow.async { + let readRequest = ReadRequest(minimumBytes: minBytes, maximumBytes: maxBytes, maximumFrames: maximumChunks) + { + (content, offset, isComplete, isFinal, lastChunkOfBatch, error) in + if let error = error { + completion(.failure(error)) + } else { + completion( + .success( + .init( + content: content, + offset: offset, + isComplete: isComplete, + lastChunkOfBatch: lastChunkOfBatch + ) + ) + ) + } + } + self.endpointFlow.addReadRequestOnContext(readRequest) } - self.endpointFlow.addReadRequest(readRequest) } } @@ -2191,14 +2239,17 @@ extension NetworkChannel where ApplicationProtocol: DatagramProtocol { } public func receive(completion: @escaping @Sendable (Result) -> Void) { - let readRequest = ReadRequest(minimumBytes: 1, maximumBytes: Int.max, maximumFrames: 1) { - (content, isComplete, isFinal, error) in - if let error = error { - completion(.failure(error)) - } else { - completion(.success(.message(content: content))) + let endpointFlow = self.endpointFlow + endpointFlow.async { + let readRequest = ReadRequest(maximumFrames: 1) { + (content, isComplete, isFinal, error) in + if let error = error { + completion(.failure(error)) + } else { + completion(.success(.message(content: content))) + } } + self.endpointFlow.addReadRequestOnContext(readRequest) } - self.endpointFlow.addReadRequest(readRequest) } } diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift index 95bf494a..2f6c0ef9 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlow.swift @@ -86,7 +86,7 @@ final class EndpointFlow: CustomDebugStringConvertible { let context: NetworkContext let identifier: UInt64 var writeRequests = NetworkUniqueDeque() - var readRequests = [ReadRequest]() + var readRequests = NetworkUniqueDeque() var stateUpdateHandler: ((State) -> Void)? = nil var cancelRequested = false var teardownComplete = false @@ -233,37 +233,80 @@ final class EndpointFlow: CustomDebugStringConvertible { switch self.flowProtocol { case .stream(let flow): - while true { - if let readRequest = self.readRequests.first { - if let content = flow.read( - minimumBytes: readRequest.minimumBytes, - maximumBytes: readRequest.maximumBytes - ) { - // TODO: Get the actual metadata - readRequest.complete(content: content, isComplete: false, isFinal: true) - // TODO: This is not efficient. Probably better to use an ArraySlice here - self.readRequests.removeFirst() - } else { + while !self.readRequests.isEmpty { + if self.readRequests[0].expectsSpan { + guard + var frames = flow.readFrames( + minimumBytes: self.readRequests[0].minimumBytes, + maximumBytes: self.readRequests[0].maximumBytes + ) + else { flow.waitForInboundDataAvailable(completion: self.inputAvailable) break } + let readRequest = self.readRequests.removeFirst() + var offset = 0 + while var frame = frames.popFirst() { + let isLastFrame = frames.isEmpty + if let bytes = frame.bytes { + readRequest.complete( + bytes: bytes, + offset: offset, + isComplete: frame.metadataComplete, + isFinal: true, + lastChunkOfBatch: isLastFrame + ) + offset += bytes.byteCount + } + frame.finalize(success: true) + } } else { - break + guard + let content = flow.read( + minimumBytes: self.readRequests[0].minimumBytes, + maximumBytes: self.readRequests[0].maximumBytes + ) + else { + flow.waitForInboundDataAvailable(completion: self.inputAvailable) + break + } + // TODO: Get the actual metadata + let readRequest = self.readRequests.removeFirst() + readRequest.complete(content: content, isComplete: false, isFinal: true) } } case .datagram(let flow): - while true { - if let readRequest = self.readRequests.first { - if let content = flow.read() { - readRequest.complete(content: content, isComplete: true, isFinal: false) - // TODO: This is not efficient. Probably better to use an ArraySlice here - self.readRequests.removeFirst() - } else { + while !self.readRequests.isEmpty { + if self.readRequests[0].expectsSpan { + guard var frames = flow.readFrames(maximumFrames: self.readRequests[0].maximumFrames) else { flow.waitForInboundDataAvailable(completion: self.inputAvailable) break } + + let readRequest = self.readRequests.removeFirst() + var offset = 0 + while var frame = frames.popFirst() { + let isLastFrame = frames.isEmpty + if let bytes = frame.bytes { + readRequest.complete( + bytes: bytes, + offset: offset, + isComplete: frame.metadataComplete, + isFinal: false, + lastChunkOfBatch: isLastFrame + ) + offset += bytes.byteCount + } + frame.finalize(success: true) + } } else { - break + guard let content = flow.read() else { + flow.waitForInboundDataAvailable(completion: self.inputAvailable) + break + } + + let readRequest = self.readRequests.removeFirst() + readRequest.complete(content: content, isComplete: true, isFinal: false) } } case .none: @@ -277,24 +320,25 @@ final class EndpointFlow: CustomDebugStringConvertible { func addWriteRequestOnContext(_ writeRequest: consuming WriteRequest) { var writeRequest: WriteRequest? = writeRequest - self.startIfNeeded() + startIfNeeded() if let takenRequest = writeRequest.take() { - self.writeRequests.append(takenRequest) + writeRequests.append(takenRequest) } - if self.state == .ready { - self.write() + if state == .ready { + write() } } - func addReadRequest(_ readRequest: ReadRequest) { - self.parameters.context.async { - self.startIfNeeded() - self.readRequests.append(readRequest) - // If state is ready and this is the first read request, then try to start reading. - // Otherwise, wait for inputAvailable to trigger a call to read() - if self.state == .ready && self.readRequests.count == 1 { - self.read() - } + func addReadRequestOnContext(_ readRequest: consuming ReadRequest) { + var readRequest: ReadRequest? = readRequest + startIfNeeded() + if let takenRequest = readRequest.take() { + readRequests.append(takenRequest) + } + // If state is ready and this is the first read request, then try to start reading. + // Otherwise, wait for inputAvailable to trigger a call to read() + if state == .ready && readRequests.count == 1 { + read() } } @@ -396,7 +440,18 @@ final class EndpointFlow: CustomDebugStringConvertible { } while !self.readRequests.isEmpty { let readRequest = self.readRequests.removeFirst() - readRequest.complete(content: nil, isComplete: false, isFinal: true, error: .posix(ECANCELED)) + if readRequest.expectsSpan { + readRequest.complete( + bytes: nil, + offset: 0, + isComplete: false, + isFinal: true, + lastChunkOfBatch: true, + error: .posix(ECANCELED) + ) + } else { + readRequest.complete(content: nil, isComplete: false, isFinal: true, error: .posix(ECANCELED)) + } } } diff --git a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift index 0dd8bf6d..5b5e7389 100644 --- a/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift +++ b/Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift @@ -364,6 +364,19 @@ final class DatagramEndpointFlowProtocol: EndpointFlowProtocol FrameArray? { + fromExternal { + do throws(NetworkError) { + return try lower.invokeReceiveDatagrams( + reference, + maximumDatagramCount: maximumFrames + ) + } catch { + return nil + } + } + } + func read() -> [UInt8]? { fromExternal { do throws(NetworkError) { @@ -528,6 +541,20 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol FrameArray? { + fromExternal { + do throws(NetworkError) { + return try lower.invokeReceiveStreamData( + reference, + minimumBytes: minimumBytes, + maximumBytes: maximumBytes + ) + } catch { + return nil + } + } + } + func read(minimumBytes: Int, maximumBytes: Int) -> [UInt8]? { fromExternal { do throws(NetworkError) { diff --git a/Sources/SwiftNetwork/EndpointFlow/ReadRequest.swift b/Sources/SwiftNetwork/EndpointFlow/ReadRequest.swift index 266098dc..5449ea2b 100644 --- a/Sources/SwiftNetwork/EndpointFlow/ReadRequest.swift +++ b/Sources/SwiftNetwork/EndpointFlow/ReadRequest.swift @@ -13,13 +13,90 @@ //===----------------------------------------------------------------------===// @available(Network 0.1.0, *) -struct ReadRequest { - let minimumBytes: Int - let maximumBytes: Int - let maximumFrames: Int - let completion: ([UInt8]?, Bool, Bool, NetworkError?) -> Void +struct ReadRequest: ~Copyable { + typealias DataCompletion = ([UInt8]?, Bool, Bool, NetworkError?) -> Void + typealias SpanCompletion = (RawSpan?, Int, Bool, Bool, Bool, NetworkError?) -> Void + + enum ReadRequestType { + case stream(minimumBytes: Int, maximumBytes: Int, dataCompletion: DataCompletion) + case datagram(maximumFrames: Int, dataCompletion: DataCompletion) + case streamSpan(minimumBytes: Int, maximumBytes: Int, maximumFrames: Int, spanCompletion: SpanCompletion) + } + + let type: ReadRequestType + + init(minimumBytes: Int, maximumBytes: Int, completion: @escaping DataCompletion) { + type = ReadRequestType.stream( + minimumBytes: minimumBytes, + maximumBytes: maximumBytes, + dataCompletion: completion + ) + } + + init(maximumFrames: Int, completion: @escaping DataCompletion) { + type = ReadRequestType.datagram(maximumFrames: maximumFrames, dataCompletion: completion) + } + + init(minimumBytes: Int, maximumBytes: Int, maximumFrames: Int, completion: @escaping SpanCompletion) { + type = ReadRequestType.streamSpan( + minimumBytes: minimumBytes, + maximumBytes: maximumBytes, + maximumFrames: maximumFrames, + spanCompletion: completion + ) + } func complete(content: [UInt8]?, isComplete: Bool, isFinal: Bool, error: NetworkError? = nil) { - completion(content, isComplete, isFinal, error) + switch type { + case .stream(_, _, let completion): completion(content, isComplete, isFinal, error) + case .datagram(_, let completion): completion(content, isComplete, isFinal, error) + default: break + } + } + + func complete( + bytes: RawSpan?, + offset: Int, + isComplete: Bool, + isFinal: Bool, + lastChunkOfBatch: Bool, + error: NetworkError? = nil + ) { + switch type { + case .streamSpan(_, _, _, let completion): + completion(bytes, offset, isComplete, isFinal, lastChunkOfBatch, error) + default: break + } + } + + var expectsSpan: Bool { + switch type { + case .streamSpan: return true + default: return false + } + } + + var minimumBytes: Int { + switch type { + case .stream(let minimumBytes, _, _): return minimumBytes + case .datagram(_, _): return 1 + case .streamSpan(let minimumBytes, _, _, _): return minimumBytes + } + } + + var maximumBytes: Int { + switch type { + case .stream(_, let maximumBytes, _): return maximumBytes + case .datagram(_, _): return Int.max + case .streamSpan(_, let maximumBytes, _, _): return maximumBytes + } + } + + var maximumFrames: Int { + switch type { + case .stream(_, _, _): return Int.max + case .datagram(let maximumFrames, _): return maximumFrames + case .streamSpan(_, _, let maximumFrames, _): return maximumFrames + } } } diff --git a/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift index 603e5a07..00e13ecb 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkConnectionTests.swift @@ -657,6 +657,85 @@ final class SwiftNetworkConnectionTests: NetTestCase { ) } + func testNoTransportSpanDataPath() { + let group = DispatchGroup() + group.enter() + let c1 = NetworkConnection( + to: Endpoint(address: IPv4Address.loopback, port: 7778), + using: .parameters { + NoTransport { + StreamBridge() + } + }.localEndpoint(Endpoint(address: IPv4Address.loopback, port: 7777)) + ) + .onStateUpdate { _, state in + print("c1 \(state)") + switch state { + case .cancelled: + group.leave() + default: + break + } + } + XCTAssertNotNil(c1) + + group.enter() + let c2 = NetworkConnection( + to: Endpoint(address: IPv4Address.loopback, port: 7777), + using: .parameters { + NoTransport { + StreamBridge() + } + }.localEndpoint(Endpoint(address: IPv4Address.loopback, port: 7778)) + ) + .onStateUpdate { _, state in + print("c2 \(state)") + switch state { + case .cancelled: + group.leave() + default: + break + } + } + XCTAssertNotNil(c2) + + c1.start() + c2.start() + + c1.send(.message(content: [1, 2, 3])) { result in + switch result { + case .success: + break + case .failure(let error): + XCTFail("send failed with error \(error)") + } + } + + c2.receive(atLeast: 1, atMost: Int.max, maximumChunks: 4) { result in + switch result { + case .success(let message): + let span = message.content + if let span { + XCTAssertEqual(span.byteCount, 3) + for i in 0..<3 { + XCTAssertEqual(span[i], UInt8(exactly: i + 1)) + } + } else { + XCTFail("No span received") + } + c1.cancel() + c2.cancel() + case .failure(let error): + XCTFail("receive failed with error \(error)") + } + } + + XCTAssertEqual( + group.wait(timeout: DispatchTime.now() + .seconds(5)), + DispatchTimeoutResult.success + ) + } + #if HAS_SWIFTTLS_RECORD func testTLSNoTransportDataPath() { let group = DispatchGroup()