Skip to content

Commit 56defe6

Browse files
glbrnttglbrntt
authored
Adopt new bag-of-bytes protocol (#55)
Motivation: The core package added a bag-of-bytes protocol which transports must be generic over. We need to adopt those changes here. Modifications: - Add a `GRPCNIOTransportBytes` type which wraps `ByteBuffer` and implements the `GRPCContiguousBytes` protocol. - Use this as the bytes type for client/server transports. - Update the appropriate state machines to deal in terms of `ByteBuffer` - Update the compressor/decompressor to deal in terms of `ByteBuffer` - Update tests Result: Avoid unconditonal copying of messages to/from `[UInt8]`/`ByteBuffer`
1 parent 64874f2 commit 56defe6

File tree

42 files changed

+789
-354
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

42 files changed

+789
-354
lines changed

‎IntegrationTests/grpc-interop-tests/Sources/InteroperabilityTestsExecutable.swift‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ struct InteroperabilityTestsExecutable: AsyncParsableCommand {
9696

9797
try await withThrowingDiscardingTaskGroup { group in
9898
group.addTask {
99-
try await client.run()
99+
try await client.runConnections()
100100
}
101101

102102
for testName in testNames {
@@ -111,7 +111,10 @@ struct InteroperabilityTestsExecutable: AsyncParsableCommand {
111111
}
112112
}
113113

114-
private func buildClient(host: String, port: Int) throws -> GRPCClient {
114+
private func buildClient(
115+
host: String,
116+
port: Int
117+
) throws -> GRPCClient<HTTP2ClientTransport.Posix> {
115118
let serviceConfig = ServiceConfig(loadBalancingConfig: [.roundRobin])
116119
return GRPCClient(
117120
transport: try .http2NIOPosix(
@@ -127,7 +130,7 @@ struct InteroperabilityTestsExecutable: AsyncParsableCommand {
127130

128131
private func runTest(
129132
_ testCase: InteroperabilityTestCase,
130-
using client: GRPCClient
133+
using client: GRPCClient<HTTP2ClientTransport.Posix>
131134
) async {
132135
print("Running '\(testCase.name)' ... ", terminator: "")
133136
do {

‎IntegrationTests/grpc-performance-tests/Sources/BenchmarkClient.swift‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
import Foundation
1818
import GRPCCore
19+
import GRPCNIOTransportHTTP2
1920
import NIOConcurrencyHelpers
2021
import Synchronization
2122

@@ -29,7 +30,7 @@ final class BenchmarkClient: Sendable {
2930
}
3031

3132
/// The underlying client.
32-
private let client: GRPCClient
33+
private let client: GRPCClient<HTTP2ClientTransport.Posix>
3334

3435
/// The number of concurrent RPCs to run.
3536
private let concurrentRPCs: Int
@@ -49,7 +50,7 @@ final class BenchmarkClient: Sendable {
4950
private let rpcStats: NIOLockedValueBox<RPCStats>
5051

5152
init(
52-
client: GRPCClient,
53+
client: GRPCClient<HTTP2ClientTransport.Posix>,
5354
concurrentRPCs: Int,
5455
rpcType: RPCType,
5556
messagesPerStream: Int,
@@ -96,7 +97,7 @@ final class BenchmarkClient: Sendable {
9697
return try await withThrowingTaskGroup(of: Void.self) { clientGroup in
9798
// Start the client.
9899
clientGroup.addTask {
99-
try await self.client.run()
100+
try await self.client.runConnections()
100101
}
101102

102103
try await withThrowingTaskGroup(of: Void.self) { rpcsGroup in
@@ -148,7 +149,9 @@ final class BenchmarkClient: Sendable {
148149
return (result, nanoseconds: Double(endTime - startTime))
149150
}
150151

151-
private func unary(benchmark: Grpc_Testing_BenchmarkService.Client) async {
152+
private func unary(
153+
benchmark: Grpc_Testing_BenchmarkService.Client<HTTP2ClientTransport.Posix>
154+
) async {
152155
let (errorCode, nanoseconds): (RPCError.Code?, Double) = await self.timeIt {
153156
do {
154157
try await benchmark.unaryCall(request: ClientRequest(message: self.message)) {
@@ -165,7 +168,9 @@ final class BenchmarkClient: Sendable {
165168
self.record(latencyNanos: nanoseconds, errorCode: errorCode)
166169
}
167170

168-
private func streaming(benchmark: Grpc_Testing_BenchmarkService.Client) async {
171+
private func streaming(
172+
benchmark: Grpc_Testing_BenchmarkService.Client<HTTP2ClientTransport.Posix>
173+
) async {
169174
// Streaming RPCs ping-pong messages back and forth. To achieve this the response message
170175
// stream is sent to the request closure, and the request closure indicates the outcome back
171176
// to the response handler to keep the RPC alive for the appropriate amount of time.

‎IntegrationTests/grpc-performance-tests/Sources/Generated/grpc_testing_benchmark_service.grpc.swift‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@
2626

2727
import GRPCCore
2828
import GRPCProtobuf
29-
import SwiftProtobuf
3029

3130
// MARK: - grpc.testing.BenchmarkService
3231

@@ -437,7 +436,7 @@ extension Grpc_Testing_BenchmarkService {
437436

438437
// Default implementation of 'registerMethods(with:)'.
439438
extension Grpc_Testing_BenchmarkService.StreamingServiceProtocol {
440-
internal func registerMethods(with router: inout GRPCCore.RPCRouter) {
439+
internal func registerMethods<Transport>(with router: inout GRPCCore.RPCRouter<Transport>) where Transport: GRPCCore.ServerTransport {
441440
router.registerHandler(
442441
forMethod: Grpc_Testing_BenchmarkService.Method.UnaryCall.descriptor,
443442
deserializer: GRPCProtobuf.ProtobufDeserializer<Grpc_Testing_SimpleRequest>(),
@@ -747,14 +746,14 @@ extension Grpc_Testing_BenchmarkService {
747746
/// The ``Client`` provides an implementation of ``ClientProtocol`` which wraps
748747
/// a `GRPCCore.GRPCCClient`. The underlying `GRPCClient` provides the long-lived
749748
/// means of communication with the remote peer.
750-
internal struct Client: ClientProtocol {
751-
private let client: GRPCCore.GRPCClient
749+
internal struct Client<Transport>: ClientProtocol where Transport: GRPCCore.ClientTransport {
750+
private let client: GRPCCore.GRPCClient<Transport>
752751

753752
/// Creates a new client wrapping the provided `GRPCCore.GRPCClient`.
754753
///
755754
/// - Parameters:
756755
/// - client: A `GRPCCore.GRPCClient` providing a communication channel to the service.
757-
internal init(wrapping client: GRPCCore.GRPCClient) {
756+
internal init(wrapping client: GRPCCore.GRPCClient<Transport>) {
758757
self.client = client
759758
}
760759

‎IntegrationTests/grpc-performance-tests/Sources/Generated/grpc_testing_control.grpc.swift‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
// Copyright 2015 gRPC authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// DO NOT EDIT.
16+
// swift-format-ignore-file
17+
//
18+
// Generated by the gRPC Swift generator plugin for the protocol buffer compiler.
19+
// Source: grpc/testing/control.proto
20+
//
21+
// For information on using the generated types, please see the documentation:
22+
// https://github.com/grpc/grpc-swift
23+
24+
// This file contained no services.

‎IntegrationTests/grpc-performance-tests/Sources/Generated/grpc_testing_messages.grpc.swift‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
// Copyright 2015-2016 gRPC authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Message definitions to be used by integration test service definitions.
16+
17+
// DO NOT EDIT.
18+
// swift-format-ignore-file
19+
//
20+
// Generated by the gRPC Swift generator plugin for the protocol buffer compiler.
21+
// Source: grpc/testing/messages.proto
22+
//
23+
// For information on using the generated types, please see the documentation:
24+
// https://github.com/grpc/grpc-swift
25+
26+
// This file contained no services.

‎IntegrationTests/grpc-performance-tests/Sources/Generated/grpc_testing_payloads.grpc.swift‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
// Copyright 2015 gRPC authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// DO NOT EDIT.
16+
// swift-format-ignore-file
17+
//
18+
// Generated by the gRPC Swift generator plugin for the protocol buffer compiler.
19+
// Source: grpc/testing/payloads.proto
20+
//
21+
// For information on using the generated types, please see the documentation:
22+
// https://github.com/grpc/grpc-swift
23+
24+
// This file contained no services.

‎IntegrationTests/grpc-performance-tests/Sources/Generated/grpc_testing_stats.grpc.swift‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
// Copyright 2015 gRPC authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// DO NOT EDIT.
16+
// swift-format-ignore-file
17+
//
18+
// Generated by the gRPC Swift generator plugin for the protocol buffer compiler.
19+
// Source: grpc/testing/stats.proto
20+
//
21+
// For information on using the generated types, please see the documentation:
22+
// https://github.com/grpc/grpc-swift
23+
24+
// This file contained no services.

‎IntegrationTests/grpc-performance-tests/Sources/Generated/grpc_testing_worker_service.grpc.swift‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@
2626

2727
import GRPCCore
2828
import GRPCProtobuf
29-
import SwiftProtobuf
3029

3130
// MARK: - grpc.testing.WorkerService
3231

@@ -381,7 +380,7 @@ extension Grpc_Testing_WorkerService {
381380

382381
// Default implementation of 'registerMethods(with:)'.
383382
extension Grpc_Testing_WorkerService.StreamingServiceProtocol {
384-
internal func registerMethods(with router: inout GRPCCore.RPCRouter) {
383+
internal func registerMethods<Transport>(with router: inout GRPCCore.RPCRouter<Transport>) where Transport: GRPCCore.ServerTransport {
385384
router.registerHandler(
386385
forMethod: Grpc_Testing_WorkerService.Method.RunServer.descriptor,
387386
deserializer: GRPCProtobuf.ProtobufDeserializer<Grpc_Testing_ServerArgs>(),

‎IntegrationTests/grpc-performance-tests/Sources/WorkerService.swift‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ final class WorkerService: Sendable {
3737
}
3838

3939
struct Server {
40-
var server: GRPCServer
40+
var server: GRPCServer<HTTP2ServerTransport.Posix>
4141
var stats: ServerStats
4242
var eventLoopGroup: MultiThreadedEventLoopGroup
4343
}
@@ -96,7 +96,7 @@ final class WorkerService: Sendable {
9696
}
9797

9898
mutating func startedServer(
99-
_ server: GRPCServer,
99+
_ server: GRPCServer<HTTP2ServerTransport.Posix>,
100100
stats: ServerStats,
101101
eventLoopGroup: MultiThreadedEventLoopGroup
102102
) -> OnStartedServer {
@@ -167,7 +167,7 @@ final class WorkerService: Sendable {
167167
}
168168

169169
enum OnStopListening {
170-
case stopListening(GRPCServer)
170+
case stopListening(GRPCServer<HTTP2ServerTransport.Posix>)
171171
case nothing
172172
}
173173

@@ -200,7 +200,7 @@ final class WorkerService: Sendable {
200200
}
201201

202202
enum OnQuitWorker {
203-
case shutDownServer(GRPCServer)
203+
case shutDownServer(GRPCServer<HTTP2ServerTransport.Posix>)
204204
case shutDownClients([BenchmarkClient])
205205
case nothing
206206
}
@@ -377,7 +377,7 @@ extension WorkerService: Grpc_Testing_WorkerService.ServiceProtocol {
377377
extension WorkerService {
378378
private func startServer(
379379
_ serverConfig: Grpc_Testing_ServerConfig
380-
) async throws -> (GRPCServer, HTTP2ServerTransport.Posix) {
380+
) async throws -> (GRPCServer<HTTP2ServerTransport.Posix>, HTTP2ServerTransport.Posix) {
381381
// Prepare an ELG, the test might require more than the default of one.
382382
let numberOfThreads: Int
383383
if serverConfig.asyncServerThreads > 0 {

‎Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift‎

Lines changed: 20 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -239,8 +239,8 @@ package final class Connection: Sendable {
239239
wrappingChannelSynchronously: channel,
240240
configuration: NIOAsyncChannel.Configuration(
241241
isOutboundHalfClosureEnabled: true,
242-
inboundType: RPCResponsePart.self,
243-
outboundType: RPCRequestPart.self
242+
inboundType: RPCResponsePart<GRPCNIOTransportBytes>.self,
243+
outboundType: RPCRequestPart<GRPCNIOTransportBytes>.self
244244
)
245245
)
246246
}
@@ -389,23 +389,32 @@ package final class Connection: Sendable {
389389

390390
extension Connection {
391391
package struct Stream {
392-
package typealias Inbound = NIOAsyncChannelInboundStream<RPCResponsePart>
392+
package typealias Inbound = NIOAsyncChannelInboundStream<RPCResponsePart<GRPCNIOTransportBytes>>
393+
394+
typealias RequestWriter = NIOAsyncChannelOutboundWriter<
395+
RPCRequestPart<GRPCNIOTransportBytes>
396+
>
397+
398+
typealias HTTP2Stream = NIOAsyncChannel<
399+
RPCResponsePart<GRPCNIOTransportBytes>,
400+
RPCRequestPart<GRPCNIOTransportBytes>
401+
>
393402

394403
package struct Outbound: ClosableRPCWriterProtocol {
395-
package typealias Element = RPCRequestPart
404+
package typealias Element = RPCRequestPart<GRPCNIOTransportBytes>
396405

397-
private let requestWriter: NIOAsyncChannelOutboundWriter<RPCRequestPart>
398-
private let http2Stream: NIOAsyncChannel<RPCResponsePart, RPCRequestPart>
406+
private let requestWriter: RequestWriter
407+
private let http2Stream: HTTP2Stream
399408

400409
fileprivate init(
401-
requestWriter: NIOAsyncChannelOutboundWriter<RPCRequestPart>,
402-
http2Stream: NIOAsyncChannel<RPCResponsePart, RPCRequestPart>
410+
requestWriter: RequestWriter,
411+
http2Stream: HTTP2Stream
403412
) {
404413
self.requestWriter = requestWriter
405414
self.http2Stream = http2Stream
406415
}
407416

408-
package func write(_ element: RPCRequestPart) async throws {
417+
package func write(_ element: RPCRequestPart<GRPCNIOTransportBytes>) async throws {
409418
try await self.requestWriter.write(element)
410419
}
411420

@@ -425,10 +434,10 @@ extension Connection {
425434

426435
let context: ClientContext
427436

428-
private let http2Stream: NIOAsyncChannel<RPCResponsePart, RPCRequestPart>
437+
private let http2Stream: HTTP2Stream
429438

430439
init(
431-
wrapping stream: NIOAsyncChannel<RPCResponsePart, RPCRequestPart>,
440+
wrapping stream: HTTP2Stream,
432441
context: ClientContext
433442
) {
434443
self.http2Stream = stream

0 commit comments

Comments
 (0)