Handle reentranct reads in ALPNHandler (#2402)
# Motivation I spotted a bug in the ALPNHandler where it doesn't properly unbuffer reentrant reads. This can lead to dropped reads. # Modification Instead of buffering into an array we are now buffering into a Deque and unbuffer as long as there are reads in the Deque. # Result No more dropped reads.
This commit is contained in:
parent
b4ebd5a64a
commit
a2fd8ad077
|
@ -68,7 +68,7 @@ var targets: [PackageDescription.Target] = [
|
|||
name: "CNIOLLHTTP",
|
||||
cSettings: [.define("LLHTTP_STRICT_MODE")]
|
||||
),
|
||||
.target(name: "NIOTLS", dependencies: ["NIO", "NIOCore"]),
|
||||
.target(name: "NIOTLS", dependencies: ["NIO", "NIOCore", swiftCollections]),
|
||||
.executableTarget(name: "NIOChatServer",
|
||||
dependencies: ["NIOPosix", "NIOCore", "NIOConcurrencyHelpers"],
|
||||
exclude: ["README.md"]),
|
||||
|
@ -112,7 +112,7 @@ var targets: [PackageDescription.Target] = [
|
|||
.testTarget(name: "NIOHTTP1Tests",
|
||||
dependencies: ["NIOCore", "NIOEmbedded", "NIOPosix", "NIOHTTP1", "NIOFoundationCompat", "NIOTestUtils"]),
|
||||
.testTarget(name: "NIOTLSTests",
|
||||
dependencies: ["NIOCore", "NIOEmbedded", "NIOTLS", "NIOFoundationCompat"]),
|
||||
dependencies: ["NIOCore", "NIOEmbedded", "NIOTLS", "NIOFoundationCompat", "NIOTestUtils"]),
|
||||
.testTarget(name: "NIOWebSocketTests",
|
||||
dependencies: ["NIOCore", "NIOEmbedded", "NIOWebSocket"]),
|
||||
.testTarget(name: "NIOTestUtilsTests",
|
||||
|
|
|
@ -13,6 +13,7 @@
|
|||
//===----------------------------------------------------------------------===//
|
||||
|
||||
import NIOCore
|
||||
import DequeModule
|
||||
|
||||
/// The result of an ALPN negotiation.
|
||||
///
|
||||
|
@ -62,7 +63,7 @@ public final class ApplicationProtocolNegotiationHandler: ChannelInboundHandler,
|
|||
|
||||
private let completionHandler: (ALPNResult, Channel) -> EventLoopFuture<Void>
|
||||
private var waitingForUser: Bool
|
||||
private var eventBuffer: [NIOAny]
|
||||
private var eventBuffer: Deque<NIOAny>
|
||||
|
||||
/// Create an `ApplicationProtocolNegotiationHandler` with the given completion
|
||||
/// callback.
|
||||
|
@ -125,15 +126,18 @@ public final class ApplicationProtocolNegotiationHandler: ChannelInboundHandler,
|
|||
}
|
||||
|
||||
private func unbuffer(context: ChannelHandlerContext) {
|
||||
for datum in eventBuffer {
|
||||
// First we check if we have anything to unbuffer
|
||||
guard !self.eventBuffer.isEmpty else {
|
||||
return
|
||||
}
|
||||
|
||||
// Now we unbuffer until there is nothing left.
|
||||
// Importantly firing a channel read can lead to new reads being buffered due to reentrancy!
|
||||
while let datum = self.eventBuffer.popFirst() {
|
||||
context.fireChannelRead(datum)
|
||||
}
|
||||
let buffer = eventBuffer
|
||||
eventBuffer = []
|
||||
waitingForUser = false
|
||||
if buffer.count > 0 {
|
||||
context.fireChannelReadComplete()
|
||||
}
|
||||
|
||||
context.fireChannelReadComplete()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -2,7 +2,7 @@
|
|||
//
|
||||
// This source file is part of the SwiftNIO open source project
|
||||
//
|
||||
// Copyright (c) 2017-2021 Apple Inc. and the SwiftNIO project authors
|
||||
// Copyright (c) 2017-2023 Apple Inc. and the SwiftNIO project authors
|
||||
// Licensed under Apache License v2.0
|
||||
//
|
||||
// See LICENSE.txt for license information
|
||||
|
@ -35,6 +35,7 @@ extension ApplicationProtocolNegotiationHandlerTests {
|
|||
("testBufferingWhileWaitingForFuture", testBufferingWhileWaitingForFuture),
|
||||
("testNothingBufferedDoesNotFireReadCompleted", testNothingBufferedDoesNotFireReadCompleted),
|
||||
("testUnbufferingFiresReadCompleted", testUnbufferingFiresReadCompleted),
|
||||
("testUnbufferingHandlesReentrantReads", testUnbufferingHandlesReentrantReads),
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
|
@ -16,6 +16,7 @@ import XCTest
|
|||
import NIOCore
|
||||
import NIOEmbedded
|
||||
import NIOTLS
|
||||
import NIOTestUtils
|
||||
|
||||
private class ReadCompletedHandler: ChannelInboundHandler {
|
||||
public typealias InboundIn = Any
|
||||
|
@ -30,6 +31,26 @@ private class ReadCompletedHandler: ChannelInboundHandler {
|
|||
}
|
||||
}
|
||||
|
||||
final class DuplicatingReadHandler: ChannelInboundHandler {
|
||||
typealias InboundIn = String
|
||||
|
||||
private let channel: EmbeddedChannel
|
||||
|
||||
private var hasDuplicatedRead = false
|
||||
|
||||
init(embeddedChannel: EmbeddedChannel) {
|
||||
self.channel = embeddedChannel
|
||||
}
|
||||
|
||||
func channelRead(context: ChannelHandlerContext, data: NIOAny) {
|
||||
if !self.hasDuplicatedRead {
|
||||
self.hasDuplicatedRead = true
|
||||
try! self.channel.writeInbound(self.unwrapInboundIn(data))
|
||||
}
|
||||
context.fireChannelRead(data)
|
||||
}
|
||||
}
|
||||
|
||||
class ApplicationProtocolNegotiationHandlerTests: XCTestCase {
|
||||
private enum EventType {
|
||||
case basic
|
||||
|
@ -222,4 +243,36 @@ class ApplicationProtocolNegotiationHandlerTests: XCTestCase {
|
|||
|
||||
XCTAssertTrue(try channel.finish().isClean)
|
||||
}
|
||||
|
||||
func testUnbufferingHandlesReentrantReads() throws {
|
||||
let channel = EmbeddedChannel()
|
||||
let continuePromise = channel.eventLoop.makePromise(of: Void.self)
|
||||
|
||||
let handler = ApplicationProtocolNegotiationHandler { result in
|
||||
continuePromise.futureResult
|
||||
}
|
||||
let readCompleteHandler = ReadCompletedHandler()
|
||||
|
||||
try channel.pipeline.addHandler(handler).wait()
|
||||
try channel.pipeline.addHandler(DuplicatingReadHandler(embeddedChannel: channel)).wait()
|
||||
try channel.pipeline.addHandler(readCompleteHandler).wait()
|
||||
|
||||
// Fire in the event.
|
||||
channel.pipeline.fireUserInboundEventTriggered(negotiatedEvent)
|
||||
|
||||
// Send a write, which is buffered.
|
||||
try channel.writeInbound("a write")
|
||||
|
||||
// At this time, readComplete hasn't fired.
|
||||
XCTAssertEqual(readCompleteHandler.readCompleteCount, 1)
|
||||
|
||||
// Now satisfy the future, which forces data unbuffering. This should fire readComplete.
|
||||
continuePromise.succeed(())
|
||||
XCTAssertNoThrow(XCTAssertEqual(try channel.readInbound()!, "a write"))
|
||||
XCTAssertNoThrow(XCTAssertEqual(try channel.readInbound()!, "a write"))
|
||||
|
||||
XCTAssertEqual(readCompleteHandler.readCompleteCount, 3)
|
||||
|
||||
XCTAssertTrue(try channel.finish().isClean)
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue