Skip to content

Strict concurrency for perf tests and the h2 server #509

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
glbrntt merged 1 commit into from
Mar 31, 2025
Merged

Strict concurrency for perf tests and the h2 server #509

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions Package.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ let package = Package(
.product(name: "NIOCore", package: "swift-nio"),
.product(name: "NIOPosix", package: "swift-nio"),
.product(name: "NIOHTTP1", package: "swift-nio"),
]
],
swiftSettings: strictConcurrencySettings
),
.executableTarget(
name: "NIOHTTP2PerformanceTester",
Expand All @@ -59,7 +60,8 @@ let package = Package(
.product(name: "NIOCore", package: "swift-nio"),
.product(name: "NIOPosix", package: "swift-nio"),
.product(name: "NIOEmbedded", package: "swift-nio"),
]
],
swiftSettings: strictConcurrencySettings
),
.target(
name: "NIOHTTP2",
Expand Down
83 changes: 45 additions & 38 deletions Sources/NIOHTTP2PerformanceTester/Bench1Conn10kRequests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,14 +21,11 @@ final class Bench1Conn10kRequests: Benchmark {
var group: MultiThreadedEventLoopGroup!
var server: Channel!
var client: Channel!
var clientMultiplexer: HTTP2StreamMultiplexer!

func setUp() throws {
self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
self.server = try setupServer(group: self.group)
let (client, multiplexer) = try setupClient(group: self.group, address: self.server.localAddress!)
self.client = client
self.clientMultiplexer = multiplexer
self.client = try setupClient(group: self.group, address: self.server.localAddress!)
}

func tearDown() {
Expand All @@ -41,7 +38,7 @@ final class Bench1Conn10kRequests: Benchmark {
func run() throws -> Int {
var bodyByteCount = 0
for _ in 0..<10_000 {
bodyByteCount += try sendOneRequest(channel: self.client, multiplexer: self.clientMultiplexer)
bodyByteCount += try sendOneRequest(channel: self.client)
}
return bodyByteCount
}
Expand All @@ -55,15 +52,22 @@ func setupServer(group: EventLoopGroup) throws -> Channel {

// Set the handlers that are applied to the accepted Channels
.childChannelInitializer { channel in
channel.configureHTTP2Pipeline(mode: .server) { streamChannel -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(HTTP2FramePayloadToHTTP1ServerCodec()).flatMap {
() -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(HTTP1TestServer())
}.flatMap { () -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(ErrorHandler())
channel.eventLoop.makeCompletedFuture {
let sync = channel.pipeline.syncOperations
let _ = try sync.configureHTTP2Pipeline(
mode: .server,
connectionConfiguration: .init(),
streamConfiguration: .init()
) { streamChannel -> EventLoopFuture<Void> in
streamChannel.eventLoop.makeCompletedFuture {
let sync = streamChannel.pipeline.syncOperations
try sync.addHandler(HTTP2FramePayloadToHTTP1ServerCodec())
try sync.addHandler(HTTP1TestServer())
try sync.addHandler(ErrorHandler())
}
}
}.flatMap { (_: HTTP2StreamMultiplexer) in
channel.pipeline.addHandler(ErrorHandler())

try sync.addHandler(ErrorHandler())
}
}

Expand All @@ -75,13 +79,16 @@ func setupServer(group: EventLoopGroup) throws -> Channel {
return try bootstrap.bind(host: "127.0.0.1", port: 12345).wait()
}

func sendOneRequest(channel: Channel, multiplexer: HTTP2StreamMultiplexer) throws -> Int {
func sendOneRequest(channel: Channel) throws -> Int {
let responseReceivedPromise = channel.eventLoop.makePromise(of: Int.self)
let requestStreamInitializer: NIOChannelInitializer = { channel in
channel.pipeline.addHandlers(
[
HTTP2FramePayloadToHTTP1ClientCodec(httpProtocol: .https),
SendRequestHandler(

channel.pipeline.handler(type: HTTP2StreamMultiplexer.self).whenSuccess { multiplexer in
multiplexer.createStreamChannel(promise: nil) { streamChannel in
streamChannel.eventLoop.makeCompletedFuture {
let sync = streamChannel.pipeline.syncOperations
try sync.addHandler(HTTP2FramePayloadToHTTP1ClientCodec(httpProtocol: .https))

let requestHandler = SendRequestHandler(
host: "127.0.0.1",
request: .init(
version: .init(major: 2, minor: 0),
Expand All @@ -90,28 +97,31 @@ func sendOneRequest(channel: Channel, multiplexer: HTTP2StreamMultiplexer) throw
headers: ["host": "localhost"]
),
responseReceivedPromise: responseReceivedPromise
),
ErrorHandler(),
],
position: .last
)
}
channel.pipeline.handler(type: HTTP2StreamMultiplexer.self).whenSuccess { multiplexer in
multiplexer.createStreamChannel(promise: nil, requestStreamInitializer)
)
try sync.addHandler(requestHandler)
try sync.addHandler(ErrorHandler())
}
}
}

return try responseReceivedPromise.futureResult.wait()
}

func setupClient(group: EventLoopGroup, address: SocketAddress) throws -> (Channel, HTTP2StreamMultiplexer) {
func setupClient(
group: EventLoopGroup,
address: SocketAddress
) throws -> Channel {
try ClientBootstrap(group: group)
.channelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
.channelInitializer { channel in
channel.pipeline.addHandler(ErrorHandler())
channel.eventLoop.makeCompletedFuture {
try channel.pipeline.syncOperations.addHandler(ErrorHandler())
}
}
.connect(to: address).flatMap { channel in
channel.configureHTTP2Pipeline(mode: .client, position: .first) { channel -> EventLoopFuture<Void> in
channel.configureHTTP2Pipeline(mode: .client, position: .first) { channel in
channel.eventLoop.makeSucceededFuture(())
}.map { (channel, $0) }
}.map { _ in channel }
}.wait()
}

Expand All @@ -124,7 +134,6 @@ final class HTTP1TestServer: ChannelInboundHandler {
return
}

let loopBoundSelf = NIOLoopBound(self, eventLoop: context.eventLoop)
let loopBoundContext = NIOLoopBound(context, eventLoop: context.eventLoop)

// Insert an event loop tick here. This more accurately represents real workloads in SwiftNIO, which will not
Expand All @@ -136,21 +145,19 @@ final class HTTP1TestServer: ChannelInboundHandler {
headers.add(name: "content-length", value: "5")
headers.add(name: "x-stream-id", value: String(Int(streamID)))
channel.write(
loopBoundSelf.value.wrapOutboundOut(
HTTPServerResponsePart.head(
HTTPResponseHead(version: .init(major: 2, minor: 0), status: .ok, headers: headers)
)
HTTPServerResponsePart.head(
HTTPResponseHead(version: .init(major: 2, minor: 0), status: .ok, headers: headers)
),
promise: nil
)

var buffer = channel.allocator.buffer(capacity: 12)
buffer.writeStaticString("hello")
channel.write(
loopBoundSelf.value.wrapOutboundOut(HTTPServerResponsePart.body(.byteBuffer(buffer))),
HTTPServerResponsePart.body(.byteBuffer(buffer)),
promise: nil
)
return channel.writeAndFlush(loopBoundSelf.value.wrapOutboundOut(HTTPServerResponsePart.end(nil)))
return channel.writeAndFlush(HTTPServerResponsePart.end(nil))
}.whenComplete { _ in
loopBoundContext.value.close(promise: nil)
}
Expand Down
9 changes: 6 additions & 3 deletions Sources/NIOHTTP2PerformanceTester/ServerOnly10KRequestsBenchmark.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,9 +117,12 @@ final class ServerOnly10KRequestsBenchmark: Benchmark {

func setUp() throws {
let channel = EmbeddedChannel()
_ = try channel.configureHTTP2Pipeline(mode: .server) { streamChannel -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(TestServer())
}.wait()

try channel.configureHTTP2Pipeline(mode: .server) { streamChannel -> EventLoopFuture<Void> in
streamChannel.eventLoop.makeCompletedFuture {
try streamChannel.pipeline.syncOperations.addHandler(TestServer())
}
}.map { _ in }.wait()

try channel.connect(to: .init(unixDomainSocketPath: "/fake"), promise: nil)

Expand Down
11 changes: 7 additions & 4 deletions Sources/NIOHTTP2PerformanceTester/StreamTeardownBenchmark.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,10 +96,13 @@ final class StreamTeardownBenchmark: Benchmark {

func createChannel() throws -> EmbeddedChannel {
let channel = EmbeddedChannel()
_ = try channel.configureHTTP2Pipeline(mode: .server) { streamChannel -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(DoNothingServer())
}.wait()
try channel.pipeline.addHandler(SendGoawayHandler(expectedStreams: self.concurrentStreams)).wait()

try channel.configureHTTP2Pipeline(mode: .server) { streamChannel -> EventLoopFuture<Void> in
streamChannel.eventLoop.makeCompletedFuture {
try streamChannel.pipeline.syncOperations.addHandler(DoNothingServer())
}
}.map { _ in }.wait()
try channel.pipeline.syncOperations.addHandler(SendGoawayHandler(expectedStreams: self.concurrentStreams))

try channel.connect(to: .init(unixDomainSocketPath: "/fake"), promise: nil)

Expand Down
28 changes: 12 additions & 16 deletions Sources/NIOHTTP2Server/main.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,18 +46,14 @@ final class HTTP1TestServer: ChannelInboundHandler {
// re-entrantly write their response frames.
let channel = context.channel
let loopBoundContext = NIOLoopBound(context, eventLoop: context.eventLoop)
let loopBoundSelf = NIOLoopBound(self, eventLoop: context.eventLoop)
context.eventLoop.execute {
channel.getOption(HTTP2StreamChannelOptions.streamID).flatMap { (streamID) -> EventLoopFuture<Void> in
let http1TestServer = loopBoundSelf.value
var headers = HTTPHeaders()
headers.add(name: "content-length", value: "5")
headers.add(name: "x-stream-id", value: String(Int(streamID)))
channel.write(
http1TestServer.wrapOutboundOut(
HTTPServerResponsePart.head(
HTTPResponseHead(version: .init(major: 2, minor: 0), status: .ok, headers: headers)
)
HTTPServerResponsePart.head(
HTTPResponseHead(version: .init(major: 2, minor: 0), status: .ok, headers: headers)
),
promise: nil
)
Expand All @@ -66,20 +62,20 @@ final class HTTP1TestServer: ChannelInboundHandler {
var buffer = channel.allocator.buffer(capacity: 12)
buffer.writeStaticString("hello")
channel.write(
http1TestServer.wrapOutboundOut(HTTPServerResponsePart.body(.byteBuffer(buffer))),
HTTPServerResponsePart.body(.byteBuffer(buffer)),
promise: nil
)
}

return channel.writeAndFlush(http1TestServer.wrapOutboundOut(HTTPServerResponsePart.end(nil)))
return channel.writeAndFlush(HTTPServerResponsePart.end(nil))
}.whenComplete { _ in
loopBoundContext.value.close(promise: nil)
}
}
}
}

final class ErrorHandler: ChannelInboundHandler {
final class ErrorHandler: ChannelInboundHandler, Sendable {
typealias InboundIn = Never

func errorCaught(context: ChannelHandlerContext, error: Error) {
Expand Down Expand Up @@ -131,14 +127,14 @@ let bootstrap = ServerBootstrap(group: group)

// Set the handlers that are applied to the accepted Channels
.childChannelInitializer { channel in
channel.configureHTTP2Pipeline(mode: .server) { streamChannel -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(HTTP2FramePayloadToHTTP1ServerCodec()).flatMap {
() -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(HTTP1TestServer())
}.flatMap { () -> EventLoopFuture<Void> in
streamChannel.pipeline.addHandler(ErrorHandler())
channel.configureHTTP2Pipeline(mode: .server) { streamChannel in
streamChannel.eventLoop.makeCompletedFuture {
let sync = streamChannel.pipeline.syncOperations
try sync.addHandler(HTTP2FramePayloadToHTTP1ServerCodec())
try sync.addHandler(HTTP1TestServer())
try sync.addHandler(ErrorHandler())
}
}.flatMap { (_: HTTP2StreamMultiplexer) in
}.flatMap { _ in
channel.pipeline.addHandler(ErrorHandler())
}
}
Expand Down