Skip to content

Commit 1040927

Browse files
FranzBuschFranzBusch
authored
Add closeOnDeinit to the NIOAsyncChannel init (apple#2592)
* Add `closeOnDeinit` to the `NIOAsyncChannel` init # Motivation In my previous PR, I already did the work to add `finishOnDeinit` configuration to the `NIOAsyncWriter` and `NIOAsyncSequenceProducer`. This PR also automatically migrated the `NIOAsyncChanell` to set the `finishOnDeinit = false`. This was intentional since we really want users to not use the deinit based cleanup; however, it also broke all current adopters of this API semantically and they might now run into the preconditions. # Modification This PR reverts the change in `NIOAsyncChannel` and does the usual deprecate + new init dance to provide users to configure this behaviour while still nudging them to check that this is really what they want. # Result Easier migration without semantically breaking current adopters of `NIOAsyncChannel`. * Rename to `wrappingChannelSynchronously`
1 parent 118de50 commit 1040927

File tree

12 files changed

+145
-52
lines changed

12 files changed

+145
-52
lines changed

‎Benchmarks/Benchmarks/NIOPosixBenchmarks/TCPEchoAsyncChannel.swift

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ func runTCPEchoAsyncChannel(numberOfWrites: Int, eventLoop: EventLoop) async thr
2323
) { channel in
2424
channel.eventLoop.makeCompletedFuture {
2525
return try NIOAsyncChannel(
26-
synchronouslyWrapping: channel,
26+
wrappingChannelSynchronously: channel,
2727
configuration: .init(
2828
inboundType: ByteBuffer.self,
2929
outboundType: ByteBuffer.self
@@ -39,7 +39,7 @@ func runTCPEchoAsyncChannel(numberOfWrites: Int, eventLoop: EventLoop) async thr
3939
) { channel in
4040
channel.eventLoop.makeCompletedFuture {
4141
return try NIOAsyncChannel(
42-
synchronouslyWrapping: channel,
42+
wrappingChannelSynchronously: channel,
4343
configuration: .init(
4444
inboundType: ByteBuffer.self,
4545
outboundType: ByteBuffer.self

‎Sources/NIOCore/AsyncChannel/AsyncChannel.swift

Lines changed: 94 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,55 @@ public struct NIOAsyncChannel<Inbound: Sendable, Outbound: Sendable>: Sendable {
100100
/// - channel: The ``Channel`` to wrap.
101101
/// - configuration: The ``NIOAsyncChannel``s configuration.
102102
@inlinable
103+
public init(
104+
wrappingChannelSynchronously channel: Channel,
105+
configuration: Configuration = .init()
106+
) throws {
107+
channel.eventLoop.preconditionInEventLoop()
108+
self.channel = channel
109+
(self._inbound, self._outbound) = try channel._syncAddAsyncHandlers(
110+
backPressureStrategy: configuration.backPressureStrategy,
111+
isOutboundHalfClosureEnabled: configuration.isOutboundHalfClosureEnabled,
112+
closeOnDeinit: false
113+
)
114+
}
115+
116+
/// Initializes a new ``NIOAsyncChannel`` wrapping a ``Channel`` where the outbound type is `Never`.
117+
///
118+
/// This initializer will finish the ``NIOAsyncChannel/outbound`` immediately.
119+
///
120+
/// - Important: This **must** be called on the channel's event loop otherwise this init will crash. This is necessary because
121+
/// we must install the handlers before any other event in the pipeline happens otherwise we might drop reads.
122+
///
123+
/// - Parameters:
124+
/// - channel: The ``Channel`` to wrap.
125+
/// - configuration: The ``NIOAsyncChannel``s configuration.
126+
@inlinable
127+
public init(
128+
wrappingChannelSynchronously channel: Channel,
129+
configuration: Configuration = .init()
130+
) throws where Outbound == Never {
131+
channel.eventLoop.preconditionInEventLoop()
132+
self.channel = channel
133+
(self._inbound, self._outbound) = try channel._syncAddAsyncHandlers(
134+
backPressureStrategy: configuration.backPressureStrategy,
135+
isOutboundHalfClosureEnabled: configuration.isOutboundHalfClosureEnabled,
136+
closeOnDeinit: false
137+
)
138+
139+
self._outbound.finish()
140+
}
141+
142+
/// Initializes a new ``NIOAsyncChannel`` wrapping a ``Channel``.
143+
///
144+
/// - Important: This **must** be called on the channel's event loop otherwise this init will crash. This is necessary because
145+
/// we must install the handlers before any other event in the pipeline happens otherwise we might drop reads.
146+
///
147+
/// - Parameters:
148+
/// - channel: The ``Channel`` to wrap.
149+
/// - configuration: The ``NIOAsyncChannel``s configuration.
150+
@available(*, deprecated, renamed: "init(wrappingChannelSynchronously:configuration:)", message: "This method has been deprecated since it defaults to deinit based resource teardown")
151+
@inlinable
103152
public init(
104153
synchronouslyWrapping channel: Channel,
105154
configuration: Configuration = .init()
@@ -108,7 +157,8 @@ public struct NIOAsyncChannel<Inbound: Sendable, Outbound: Sendable>: Sendable {
108157
self.channel = channel
109158
(self._inbound, self._outbound) = try channel._syncAddAsyncHandlers(
110159
backPressureStrategy: configuration.backPressureStrategy,
111-
isOutboundHalfClosureEnabled: configuration.isOutboundHalfClosureEnabled
160+
isOutboundHalfClosureEnabled: configuration.isOutboundHalfClosureEnabled,
161+
closeOnDeinit: true
112162
)
113163
}
114164

@@ -123,6 +173,7 @@ public struct NIOAsyncChannel<Inbound: Sendable, Outbound: Sendable>: Sendable {
123173
/// - channel: The ``Channel`` to wrap.
124174
/// - configuration: The ``NIOAsyncChannel``s configuration.
125175
@inlinable
176+
@available(*, deprecated, renamed: "init(wrappingChannelSynchronously:configuration:)", message: "This method has been deprecated since it defaults to deinit based resource teardown")
126177
public init(
127178
synchronouslyWrapping channel: Channel,
128179
configuration: Configuration = .init()
@@ -131,7 +182,8 @@ public struct NIOAsyncChannel<Inbound: Sendable, Outbound: Sendable>: Sendable {
131182
self.channel = channel
132183
(self._inbound, self._outbound) = try channel._syncAddAsyncHandlers(
133184
backPressureStrategy: configuration.backPressureStrategy,
134-
isOutboundHalfClosureEnabled: configuration.isOutboundHalfClosureEnabled
185+
isOutboundHalfClosureEnabled: configuration.isOutboundHalfClosureEnabled,
186+
closeOnDeinit: true
135187
)
136188

137189
self._outbound.finish()
@@ -149,12 +201,12 @@ public struct NIOAsyncChannel<Inbound: Sendable, Outbound: Sendable>: Sendable {
149201
self._outbound = outboundWriter
150202
}
151203

152-
153204
/// This method is only used from our server bootstrap to allow us to run the child channel initializer
154205
/// at the right moment.
155206
///
156207
/// - Important: This is not considered stable API and should not be used.
157208
@inlinable
209+
@available(*, deprecated, message: "This method has been deprecated since it defaults to deinit based resource teardown")
158210
public static func _wrapAsyncChannelWithTransformations(
159211
synchronouslyWrapping channel: Channel,
160212
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil,
@@ -165,6 +217,35 @@ public struct NIOAsyncChannel<Inbound: Sendable, Outbound: Sendable>: Sendable {
165217
let (inboundStream, outboundWriter): (NIOAsyncChannelInboundStream<Inbound>, NIOAsyncChannelOutboundWriter<Outbound>) = try channel._syncAddAsyncHandlersWithTransformations(
166218
backPressureStrategy: backPressureStrategy,
167219
isOutboundHalfClosureEnabled: isOutboundHalfClosureEnabled,
220+
closeOnDeinit: true,
221+
channelReadTransformation: channelReadTransformation
222+
)
223+
224+
outboundWriter.finish()
225+
226+
return .init(
227+
channel: channel,
228+
inboundStream: inboundStream,
229+
outboundWriter: outboundWriter
230+
)
231+
}
232+
233+
/// This method is only used from our server bootstrap to allow us to run the child channel initializer
234+
/// at the right moment.
235+
///
236+
/// - Important: This is not considered stable API and should not be used.
237+
@inlinable
238+
public static func _wrapAsyncChannelWithTransformations(
239+
wrappingChannelSynchronously channel: Channel,
240+
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil,
241+
isOutboundHalfClosureEnabled: Bool = false,
242+
channelReadTransformation: @Sendable @escaping (Channel) -> EventLoopFuture<Inbound>
243+
) throws -> NIOAsyncChannel<Inbound, Outbound> where Outbound == Never {
244+
channel.eventLoop.preconditionInEventLoop()
245+
let (inboundStream, outboundWriter): (NIOAsyncChannelInboundStream<Inbound>, NIOAsyncChannelOutboundWriter<Outbound>) = try channel._syncAddAsyncHandlersWithTransformations(
246+
backPressureStrategy: backPressureStrategy,
247+
isOutboundHalfClosureEnabled: isOutboundHalfClosureEnabled,
248+
closeOnDeinit: false,
168249
channelReadTransformation: channelReadTransformation
169250
)
170251

@@ -229,17 +310,20 @@ extension Channel {
229310
@inlinable
230311
func _syncAddAsyncHandlers<Inbound: Sendable, Outbound: Sendable>(
231312
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark?,
232-
isOutboundHalfClosureEnabled: Bool
313+
isOutboundHalfClosureEnabled: Bool,
314+
closeOnDeinit: Bool
233315
) throws -> (NIOAsyncChannelInboundStream<Inbound>, NIOAsyncChannelOutboundWriter<Outbound>) {
234316
self.eventLoop.assertInEventLoop()
235317

236318
let inboundStream = try NIOAsyncChannelInboundStream<Inbound>.makeWrappingHandler(
237319
channel: self,
238-
backPressureStrategy: backPressureStrategy
320+
backPressureStrategy: backPressureStrategy,
321+
closeOnDeinit: closeOnDeinit
239322
)
240323
let writer = try NIOAsyncChannelOutboundWriter<Outbound>(
241324
channel: self,
242-
isOutboundHalfClosureEnabled: isOutboundHalfClosureEnabled
325+
isOutboundHalfClosureEnabled: isOutboundHalfClosureEnabled,
326+
closeOnDeinit: closeOnDeinit
243327
)
244328
return (inboundStream, writer)
245329
}
@@ -249,18 +333,21 @@ extension Channel {
249333
func _syncAddAsyncHandlersWithTransformations<ChannelReadResult>(
250334
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark?,
251335
isOutboundHalfClosureEnabled: Bool,
336+
closeOnDeinit: Bool,
252337
channelReadTransformation: @Sendable @escaping (Channel) -> EventLoopFuture<ChannelReadResult>
253338
) throws -> (NIOAsyncChannelInboundStream<ChannelReadResult>, NIOAsyncChannelOutboundWriter<Never>) {
254339
self.eventLoop.assertInEventLoop()
255340

256341
let inboundStream = try NIOAsyncChannelInboundStream<ChannelReadResult>.makeTransformationHandler(
257342
channel: self,
258343
backPressureStrategy: backPressureStrategy,
344+
closeOnDeinit: closeOnDeinit,
259345
channelReadTransformation: channelReadTransformation
260346
)
261347
let writer = try NIOAsyncChannelOutboundWriter<Never>(
262348
channel: self,
263-
isOutboundHalfClosureEnabled: isOutboundHalfClosureEnabled
349+
isOutboundHalfClosureEnabled: isOutboundHalfClosureEnabled,
350+
closeOnDeinit: closeOnDeinit
264351
)
265352
return (inboundStream, writer)
266353
}

‎Sources/NIOCore/AsyncChannel/AsyncChannelInboundStream.swift

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ public struct NIOAsyncChannelInboundStream<Inbound: Sendable>: Sendable {
8080
init<HandlerInbound: Sendable>(
8181
channel: Channel,
8282
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark?,
83+
closeOnDeinit: Bool,
8384
handler: NIOAsyncChannelInboundStreamChannelHandler<HandlerInbound, Inbound>
8485
) throws {
8586
channel.eventLoop.preconditionInEventLoop()
@@ -95,7 +96,7 @@ public struct NIOAsyncChannelInboundStream<Inbound: Sendable>: Sendable {
9596

9697
let sequence = Producer.makeSequence(
9798
backPressureStrategy: strategy,
98-
finishOnDeinit: false,
99+
finishOnDeinit: closeOnDeinit,
99100
delegate: NIOAsyncChannelInboundStreamChannelHandlerProducerDelegate(handler: handler)
100101
)
101102
handler.source = sequence.source
@@ -107,7 +108,8 @@ public struct NIOAsyncChannelInboundStream<Inbound: Sendable>: Sendable {
107108
@inlinable
108109
static func makeWrappingHandler(
109110
channel: Channel,
110-
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark?
111+
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark?,
112+
closeOnDeinit: Bool
111113
) throws -> NIOAsyncChannelInboundStream {
112114
let handler = NIOAsyncChannelInboundStreamChannelHandler<Inbound, Inbound>.makeHandler(
113115
eventLoop: channel.eventLoop
@@ -116,6 +118,7 @@ public struct NIOAsyncChannelInboundStream<Inbound: Sendable>: Sendable {
116118
return try .init(
117119
channel: channel,
118120
backPressureStrategy: backPressureStrategy,
121+
closeOnDeinit: closeOnDeinit,
119122
handler: handler
120123
)
121124
}
@@ -125,6 +128,7 @@ public struct NIOAsyncChannelInboundStream<Inbound: Sendable>: Sendable {
125128
static func makeTransformationHandler(
126129
channel: Channel,
127130
backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark?,
131+
closeOnDeinit: Bool,
128132
channelReadTransformation: @Sendable @escaping (Channel) -> EventLoopFuture<Inbound>
129133
) throws -> NIOAsyncChannelInboundStream {
130134
let handler = NIOAsyncChannelInboundStreamChannelHandler<Channel, Inbound>.makeHandlerWithTransformations(
@@ -135,6 +139,7 @@ public struct NIOAsyncChannelInboundStream<Inbound: Sendable>: Sendable {
135139
return try .init(
136140
channel: channel,
137141
backPressureStrategy: backPressureStrategy,
142+
closeOnDeinit: closeOnDeinit,
138143
handler: handler
139144
)
140145
}

‎Sources/NIOCore/AsyncChannel/AsyncChannelOutboundWriter.swift

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,8 @@ public struct NIOAsyncChannelOutboundWriter<OutboundOut: Sendable>: Sendable {
8484
@inlinable
8585
init(
8686
channel: Channel,
87-
isOutboundHalfClosureEnabled: Bool
87+
isOutboundHalfClosureEnabled: Bool,
88+
closeOnDeinit: Bool
8889
) throws {
8990
let handler = NIOAsyncChannelOutboundWriterHandler<OutboundOut>(
9091
eventLoop: channel.eventLoop,
@@ -93,7 +94,7 @@ public struct NIOAsyncChannelOutboundWriter<OutboundOut: Sendable>: Sendable {
9394
let writer = _Writer.makeWriter(
9495
elementType: OutboundOut.self,
9596
isWritable: true,
96-
finishOnDeinit: false,
97+
finishOnDeinit: closeOnDeinit,
9798
delegate: .init(handler: handler)
9899
)
99100
handler.sink = writer.sink

‎Sources/NIOCore/Docs.docc/swift-concurrency.md

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ the inbound data and echo it back outbound.
8585

8686
```swift
8787
let channel = ...
88-
let asyncChannel = try NIOAsyncChannel<ByteBuffer, ByteBuffer>(synchronouslyWrapping: channel)
88+
let asyncChannel = try NIOAsyncChannel<ByteBuffer, ByteBuffer>(wrappingChannelSynchronously: channel)
8989

9090
try await asyncChannel.executeThenClose { inbound, outbound in
9191
for try await inboundData in inbound {
@@ -186,7 +186,7 @@ let clientChannel = try await ClientBootstrap(group: eventLoopGroup)
186186
) { channel in
187187
channel.eventLoop.makeCompletedFuture {
188188
return try NIOAsyncChannel<ByteBuffer, ByteBuffer>(
189-
synchronouslyWrapping: channel
189+
wrappingChannelSynchronously: channel
190190
)
191191
}
192192
}
@@ -245,7 +245,7 @@ let upgradeResult: EventLoopFuture<UpgradeResult> = try await ClientBootstrap(gr
245245
// This configures the pipeline after the websocket upgrade was successful.
246246
// We are wrapping the pipeline in a NIOAsyncChannel.
247247
channel.eventLoop.makeCompletedFuture {
248-
let asyncChannel = try NIOAsyncChannel<WebSocketFrame, WebSocketFrame>(synchronouslyWrapping: channel)
248+
let asyncChannel = try NIOAsyncChannel<WebSocketFrame, WebSocketFrame>(wrappingChannelSynchronously: channel)
249249
return UpgradeResult.websocket(asyncChannel)
250250
}
251251
}

‎Sources/NIOPosix/Bootstrap.swift

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -654,7 +654,7 @@ extension ServerBootstrap {
654654
)
655655
let asyncChannel = try NIOAsyncChannel<ChannelInitializerResult, Never>
656656
._wrapAsyncChannelWithTransformations(
657-
synchronouslyWrapping: serverChannel,
657+
wrappingChannelSynchronously: serverChannel,
658658
backPressureStrategy: serverBackPressureStrategy,
659659
channelReadTransformation: { channel -> EventLoopFuture<ChannelInitializerResult> in
660660
// The channelReadTransformation is run on the EL of the server channel

‎Sources/NIOTCPEchoClient/Client.swift

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ struct Client {
5959
try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder()))
6060

6161
return try NIOAsyncChannel(
62-
synchronouslyWrapping: channel,
62+
wrappingChannelSynchronously: channel,
6363
configuration: NIOAsyncChannel.Configuration(
6464
inboundType: String.self,
6565
outboundType: String.self

‎Sources/NIOTCPEchoServer/Server.swift

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ struct Server {
4848
try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder()))
4949

5050
return try NIOAsyncChannel(
51-
synchronouslyWrapping: channel,
51+
wrappingChannelSynchronously: channel,
5252
configuration: NIOAsyncChannel.Configuration(
5353
inboundType: String.self,
5454
outboundType: String.self

‎Sources/NIOWebSocketClient/Client.swift

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ struct Client {
6161
// let upgrader = NIOTypedWebSocketClientUpgrader<UpgradeResult>(
6262
// upgradePipelineHandler: { (channel, _) in
6363
// channel.eventLoop.makeCompletedFuture {
64-
// let asyncChannel = try NIOAsyncChannel<WebSocketFrame, WebSocketFrame>(synchronouslyWrapping: channel)
64+
// let asyncChannel = try NIOAsyncChannel<WebSocketFrame, WebSocketFrame>(wrappingChannelSynchronously: channel)
6565
// return UpgradeResult.websocket(asyncChannel)
6666
// }
6767
// }

‎Sources/NIOWebSocketServer/Server.swift

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ struct Server {
9191
// },
9292
// upgradePipelineHandler: { (channel, _) in
9393
// channel.eventLoop.makeCompletedFuture {
94-
// let asyncChannel = try NIOAsyncChannel<WebSocketFrame, WebSocketFrame>(synchronouslyWrapping: channel)
94+
// let asyncChannel = try NIOAsyncChannel<WebSocketFrame, WebSocketFrame>(wrappingChannelSynchronously: channel)
9595
// return UpgradeResult.websocket(asyncChannel)
9696
// }
9797
// }
@@ -102,7 +102,7 @@ struct Server {
102102
// notUpgradingCompletionHandler: { channel in
103103
// channel.eventLoop.makeCompletedFuture {
104104
// try channel.pipeline.syncOperations.addHandler(HTTPByteBufferResponsePartHandler())
105-
// let asyncChannel = try NIOAsyncChannel<HTTPServerRequestPart, HTTPPart<HTTPResponseHead, ByteBuffer>>(synchronouslyWrapping: channel)
105+
// let asyncChannel = try NIOAsyncChannel<HTTPServerRequestPart, HTTPPart<HTTPResponseHead, ByteBuffer>>(wrappingChannelSynchronously: channel)
106106
// return UpgradeResult.notUpgraded(asyncChannel)
107107
// }
108108
// }

0 commit comments

Comments
 (0)