Skip to content

Commit 472cac0

Browse files
glbrnttglbrntt
committed
Add a server-idle handler
Motivation: Connections should be dropped to save resources if there are not RPCs for an idle timeout. This was added for clients in grpc#798; this adds similart functionality to the server. Modifications: - Rename a `ClientConnectivityHandler` to `GRPCIdleHandler` - Add a 'mode' to the idle handler Note: timeout configuration will be added for client and server in a followup PR. Result: - Server's will drop connections if the client isn't doing anything. - Resolves grpc#706
1 parent 97be024 commit 472cac0

File tree

4 files changed

+87
-48
lines changed

4 files changed

+87
-48
lines changed

‎Sources/GRPC/ClientConnection.swift

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -339,7 +339,7 @@ extension Channel {
339339
}.flatMap { _ in
340340
return self.pipeline.handler(type: NIOHTTP2Handler.self).flatMap { http2Handler in
341341
self.pipeline.addHandler(
342-
ClientConnectivityHandler(connectionManager: connectionManager),
342+
GRPCIdleHandler(mode: .client(connectionManager)),
343343
position: .after(http2Handler)
344344
)
345345
}.flatMap {

‎Sources/GRPC/ClientConnectivityHandler.swift renamed to ‎Sources/GRPC/GRPCIdleHandler.swift

Lines changed: 73 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,29 @@
1616
import NIO
1717
import NIOHTTP2
1818

19-
internal class ClientConnectivityHandler: ChannelInboundHandler {
19+
internal class GRPCIdleHandler: ChannelInboundHandler {
2020
typealias InboundIn = HTTP2Frame
2121

22-
private var connectionManager: ConnectionManager
22+
/// The amount of time to wait before closing the channel when there are no active streams.
2323
private let idleTimeout: TimeAmount
2424

25+
/// The number of active streams.
2526
private var activeStreams = 0
27+
28+
/// The scheduled task which will close the channel.
2629
private var scheduledIdle: Scheduled<Void>? = nil
30+
31+
/// Client and server have slightly different behaviours; track which we are following.
32+
private var mode: Mode
33+
34+
/// The mode of operation: the client tracks additional connection state in the connection
35+
/// manager.
36+
internal enum Mode {
37+
case client(ConnectionManager)
38+
case server
39+
}
40+
41+
/// The current connection state.
2742
private var state: State = .notReady
2843

2944
private enum State {
@@ -37,8 +52,8 @@ internal class ClientConnectivityHandler: ChannelInboundHandler {
3752
case closed
3853
}
3954

40-
init(connectionManager: ConnectionManager, idleTimeout: TimeAmount = .minutes(5)) {
41-
self.connectionManager = connectionManager
55+
init(mode: Mode, idleTimeout: TimeAmount = .minutes(5)) {
56+
self.mode = mode
4257
self.idleTimeout = idleTimeout
4358
}
4459

@@ -67,58 +82,74 @@ internal class ClientConnectivityHandler: ChannelInboundHandler {
6782
}
6883

6984
func channelActive(context: ChannelHandlerContext) {
70-
switch self.state {
71-
case .notReady:
72-
self.connectionManager.channelActive(channel: context.channel)
73-
case .ready, .closed:
85+
switch (self.mode, self.state) {
86+
case (.client(let manager), .notReady):
87+
manager.channelActive(channel: context.channel)
88+
89+
case (.server, .notReady),
90+
(_, .ready),
91+
(_, .closed):
7492
()
7593
}
7694

95+
self.scheduleIdleTimeout(context: context)
7796
context.fireChannelActive()
7897
}
7998

8099
func channelInactive(context: ChannelHandlerContext) {
81100
self.scheduledIdle?.cancel()
82101
self.scheduledIdle = nil
83102

84-
switch self.state {
85-
case .notReady, .ready:
86-
self.connectionManager.channelInactive()
87-
case .closed:
103+
switch (self.mode, self.state) {
104+
case (.client(let manager), .notReady),
105+
(.client(let manager), .ready):
106+
manager.channelInactive()
107+
108+
case (.server, .notReady),
109+
(.server, .ready),
110+
(_, .closed):
88111
()
112+
89113
}
90114

91115
context.fireChannelInactive()
92116
}
93117

94118
func channelRead(context: ChannelHandlerContext, data: NIOAny) {
95-
let frame = self.unwrapInboundIn(data)
96-
97-
if frame.streamID == .rootStream {
98-
switch (self.state, frame.payload) {
99-
// We only care about SETTINGS as long as we are in state `.notReady`.
100-
case (.notReady, .settings):
101-
self.state = .ready
102-
103-
let remoteAddressDescription = context.channel.remoteAddress.map { "\($0)" } ?? "n/a"
104-
self.connectionManager.logger.info("gRPC connection ready", metadata: [
105-
"remote_address": "\(remoteAddressDescription)",
106-
"event_loop": "\(context.eventLoop)"
107-
])
108-
109-
// Start the idle timeout.
110-
self.scheduleIdleTimeout(context: context)
119+
switch self.mode {
120+
// The client has some connection state transitions we need to deal with here.
121+
case .client(let manager):
122+
let frame = self.unwrapInboundIn(data)
123+
124+
if frame.streamID == .rootStream {
125+
switch (self.state, frame.payload) {
126+
// We only care about SETTINGS as long as we are in state `.notReady`.
127+
case (.notReady, .settings):
128+
self.state = .ready
129+
130+
let remoteAddressDescription = context.channel.remoteAddress.map { "\($0)" } ?? "n/a"
131+
manager.logger.info("gRPC connection ready", metadata: [
132+
"remote_address": "\(remoteAddressDescription)",
133+
"event_loop": "\(context.eventLoop)"
134+
])
135+
136+
// Start the idle timeout.
137+
self.scheduleIdleTimeout(context: context)
111138

112-
// Let the manager know we're ready.
113-
self.connectionManager.ready()
139+
// Let the manager know we're ready.
140+
manager.ready()
114141

115-
case (.notReady, .goAway),
116-
(.ready, .goAway):
117-
self.idle(context: context)
142+
case (.notReady, .goAway),
143+
(.ready, .goAway):
144+
self.idle(context: context)
118145

119-
default:
120-
()
146+
default:
147+
()
148+
}
121149
}
150+
151+
case .server:
152+
()
122153
}
123154

124155
context.fireChannelRead(data)
@@ -140,7 +171,13 @@ internal class ClientConnectivityHandler: ChannelInboundHandler {
140171
}
141172

142173
self.state = .closed
143-
self.connectionManager.idle()
174+
switch self.mode {
175+
case .client(let manager):
176+
manager.idle()
177+
case .server:
178+
()
179+
}
180+
144181
context.close(mode: .all, promise: nil)
145182
}
146183
}

‎Sources/GRPC/HTTPProtocolSwitcher.swift

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,8 +135,10 @@ extension HTTPProtocolSwitcher: ChannelInboundHandler, RemovableChannelHandler {
135135
context.channel.configureHTTP2Pipeline(mode: .server) { (streamChannel, streamID) in
136136
streamChannel.pipeline.addHandler(HTTP2ToHTTP1ServerCodec(streamID: streamID, normalizeHTTPHeaders: true))
137137
.flatMap { self.handlersInitializer(streamChannel) }
138+
}.flatMap { multiplexer in
139+
// Add an idle handler between the two HTTP2 handlers.
140+
context.channel.pipeline.addHandler(GRPCIdleHandler(mode: .server), position: .before(multiplexer))
138141
}
139-
.map { _ in }
140142
.cascade(to: pipelineConfigured)
141143
}
142144

‎Tests/GRPCTests/ConnectionManagerTests.swift

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@ extension ConnectionManagerTests {
116116

117117
// Setup the real channel and activate it.
118118
let channel = EmbeddedChannel(
119-
handler: ClientConnectivityHandler(connectionManager: manager),
119+
handler: GRPCIdleHandler(mode: .client(manager)),
120120
loop: self.loop
121121
)
122122
channelPromise.succeed(channel)
@@ -150,7 +150,7 @@ extension ConnectionManagerTests {
150150

151151
// Setup the channel.
152152
let channel = EmbeddedChannel(
153-
handler: ClientConnectivityHandler(connectionManager: manager),
153+
handler: GRPCIdleHandler(mode: .client(manager)),
154154
loop: self.loop
155155
)
156156
channelPromise.succeed(channel)
@@ -193,7 +193,7 @@ extension ConnectionManagerTests {
193193

194194
// Setup the channel.
195195
let channel = EmbeddedChannel(
196-
handler: ClientConnectivityHandler(connectionManager: manager),
196+
handler: GRPCIdleHandler(mode: .client(manager)),
197197
loop: self.loop
198198
)
199199
channelPromise.succeed(channel)
@@ -249,7 +249,7 @@ extension ConnectionManagerTests {
249249

250250
// Setup the channel.
251251
let channel = EmbeddedChannel(
252-
handler: ClientConnectivityHandler(connectionManager: manager),
252+
handler: GRPCIdleHandler(mode: .client(manager)),
253253
loop: self.loop
254254
)
255255
channelPromise.succeed(channel)
@@ -305,7 +305,7 @@ extension ConnectionManagerTests {
305305

306306
// Setup the actual channel and complete the promise.
307307
let channel = EmbeddedChannel(
308-
handler: ClientConnectivityHandler(connectionManager: manager),
308+
handler: GRPCIdleHandler(mode: .client(manager)),
309309
loop: self.loop
310310
)
311311
channelPromise.succeed(channel)
@@ -401,7 +401,7 @@ extension ConnectionManagerTests {
401401

402402
// Prepare the channel
403403
let channel = EmbeddedChannel(
404-
handler: ClientConnectivityHandler(connectionManager: manager),
404+
handler: GRPCIdleHandler(mode: .client(manager)),
405405
loop: self.loop
406406
)
407407
channelPromise.succeed(channel)
@@ -461,7 +461,7 @@ extension ConnectionManagerTests {
461461

462462
// Prepare the channel
463463
let firstChannel = EmbeddedChannel(
464-
handler: ClientConnectivityHandler(connectionManager: manager),
464+
handler: GRPCIdleHandler(mode: .client(manager)),
465465
loop: self.loop
466466
)
467467
channelPromise.succeed(firstChannel)
@@ -521,7 +521,7 @@ extension ConnectionManagerTests {
521521

522522
// Prepare the first channel
523523
let firstChannel = EmbeddedChannel(
524-
handler: ClientConnectivityHandler(connectionManager: manager),
524+
handler: GRPCIdleHandler(mode: .client(manager)),
525525
loop: self.loop
526526
)
527527
firstChannelPromise.succeed(firstChannel)
@@ -548,7 +548,7 @@ extension ConnectionManagerTests {
548548

549549
// Prepare the second channel
550550
let secondChannel = EmbeddedChannel(
551-
handler: ClientConnectivityHandler(connectionManager: manager),
551+
handler: GRPCIdleHandler(mode: .client(manager)),
552552
loop: self.loop
553553
)
554554
secondChannelPromise.succeed(secondChannel)
@@ -582,7 +582,7 @@ extension ConnectionManagerTests {
582582

583583
// Setup the channel.
584584
let channel = EmbeddedChannel(
585-
handler: ClientConnectivityHandler(connectionManager: manager),
585+
handler: GRPCIdleHandler(mode: .client(manager)),
586586
loop: self.loop
587587
)
588588
channelPromise.succeed(channel)

0 commit comments

Comments
 (0)