fix EventLoops that Bootstrap channel initialiser's call out on (#424)

Motivation:

Previously we were not running the (child/server)channelInitializers on the
event loop associated to the `Channel` we're initialising. Almost all
operations in there are pipeline modifications which are thread safe so
it presumably wasn't a massive correctness issue. However it's very
expensive to hop threads that often and it is also very unexpected. This
addresses this issue.

Modifications:

made all (child/server)channelInitializers run on the event loop that is
associated to their `Channel`

Result:

- more correctness
- less unexpected behaviour

Author: weissi

Sources/NIO/Bootstrap.swift

@@ -205,18 +205,22 @@ public final class ServerBootstrap {
             return eventLoop.newFailedFuture(error: error)
         }
 
-        return serverChannelInit(serverChannel).then {
-            serverChannel.pipeline.add(handler: AcceptHandler(childChannelInitializer: childChannelInit,
-                                                              childChannelOptions: childChannelOptions))
-        }.then {
-            serverChannelOptions.applyAll(channel: serverChannel)
+        return eventLoop.submit {
+            return serverChannelInit(serverChannel).then {
+                serverChannel.pipeline.add(handler: AcceptHandler(childChannelInitializer: childChannelInit,
+                                                                  childChannelOptions: childChannelOptions))
+            }.then {
+                serverChannelOptions.applyAll(channel: serverChannel)
+            }.then {
+                register(eventLoop, serverChannel)
+            }.map {
+                serverChannel as Channel
+            }.thenIfError { error in
+                serverChannel.close0(error: error, mode: .all, promise: nil)
+                return eventLoop.newFailedFuture(error: error)
+            }
         }.then {
-            register(eventLoop, serverChannel)
-        }.map {
-            serverChannel
-        }.thenIfError { error in
-            serverChannel.close0(error: error, mode: .all, promise: nil)
-            return eventLoop.newFailedFuture(error: error)
+            $0
         }
     }
 
@@ -240,23 +244,42 @@ public final class ServerBootstrap {
 
         func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
             let accepted = self.unwrapInboundIn(data)
-            let childChannelInit = self.childChannelInit ?? { (_: Channel) in ctx.eventLoop.newSucceededFuture(result: ()) }
-
-            self.childChannelOptions.applyAll(channel: accepted).hopTo(eventLoop: ctx.eventLoop).then {
-                assert(ctx.eventLoop.inEventLoop)
-                return childChannelInit(accepted)
-            }.then { () -> EventLoopFuture<Void> in
-                assert(ctx.eventLoop.inEventLoop)
-                guard !ctx.pipeline.destroyed else {
-                    return accepted.close().thenThrowing {
-                        throw ChannelError.ioOnClosedChannel
+            let ctxEventLoop = ctx.eventLoop
+            let childEventLoop = accepted.eventLoop
+            let childChannelInit = self.childChannelInit ?? { (_: Channel) in childEventLoop.newSucceededFuture(result: ()) }
+
+            @inline(__always)
+            func setupChildChannel() -> EventLoopFuture<Void> {
+                return self.childChannelOptions.applyAll(channel: accepted).then { () -> EventLoopFuture<Void> in
+                    assert(childEventLoop.inEventLoop)
+                    return childChannelInit(accepted)
+                }
+            }
+
+            @inline(__always)
+            func fireThroughPipeline(_ future: EventLoopFuture<Void>) {
+                assert(ctxEventLoop.inEventLoop)
+                future.then { (_) -> EventLoopFuture<Void> in
+                    assert(ctxEventLoop.inEventLoop)
+                    guard !ctx.pipeline.destroyed else {
+                        return accepted.close().thenThrowing {
+                            throw ChannelError.ioOnClosedChannel
+                        }
                     }
+                    ctx.fireChannelRead(data)
+                    return ctx.eventLoop.newSucceededFuture(result: ())
+                }.whenFailure { error in
+                    assert(ctx.eventLoop.inEventLoop)
+                    self.closeAndFire(ctx: ctx, accepted: accepted, err: error)
                 }
-                ctx.fireChannelRead(data)
-                return ctx.eventLoop.newSucceededFuture(result: ())
-            }.whenFailure { error in
-                assert(ctx.eventLoop.inEventLoop)
-                self.closeAndFire(ctx: ctx, accepted: accepted, err: error)
+            }
+
+            if childEventLoop === ctxEventLoop {
+                fireThroughPipeline(setupChildChannel())
+            } else {
+                fireThroughPipeline(childEventLoop.submit {
+                    return setupChildChannel()
+                }.then { $0 }.hopTo(eventLoop: ctxEventLoop))
             }
         }
 
@@ -463,18 +486,27 @@ public final class ClientBootstrap {
             return promise.futureResult
         }
 
-        channelInitializer(channel).then {
-            channelOptions.applyAll(channel: channel)
-        }.then {
-            channel.registerAndDoSynchronously(body)
-        }.map {
-            channel
-        }.thenIfError { error in
-            channel.close0(error: error, mode: .all, promise: nil)
-            return channel.eventLoop.newFailedFuture(error: error)
-        }.cascade(promise: promise)
+        @inline(__always)
+        func setupChannel() -> EventLoopFuture<Channel> {
+            assert(eventLoop.inEventLoop)
+            channelInitializer(channel).then {
+                channelOptions.applyAll(channel: channel)
+            }.then {
+                channel.registerAndDoSynchronously(body)
+            }.map {
+                channel
+            }.thenIfError { error in
+                channel.close0(error: error, mode: .all, promise: nil)
+                return channel.eventLoop.newFailedFuture(error: error)
+            }.cascade(promise: promise)
+            return promise.futureResult
+        }
 
-        return promise.futureResult
+        if eventLoop.inEventLoop {
+            return setupChannel()
+        } else {
+            return eventLoop.submit(setupChannel).then { $0 }
+        }
     }
 }
 

Tests/LinuxMain.swift

@@ -36,6 +36,7 @@ import XCTest
          testCase(Base64Test.allTests),
          testCase(BaseObjectTest.allTests),
          testCase(BlockingIOThreadPoolTest.allTests),
+         testCase(BootstrapTest.allTests),
          testCase(ByteBufferTest.allTests),
          testCase(ByteToMessageDecoderTest.allTests),
          testCase(ChannelNotificationTest.allTests),

Tests/NIOTests/BootstrapTest+XCTest.swift

@@ -0,0 +1,33 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+//
+// BootstrapTest+XCTest.swift
+//
+import XCTest
+
+///
+/// NOTE: This file was generated by generate_linux_tests.rb
+///
+/// Do NOT edit this file directly as it will be regenerated automatically when needed.
+///
+
+extension BootstrapTest {
+
+   static var allTests : [(String, (BootstrapTest) -> () throws -> Void)] {
+      return [
+                ("testBootstrapsCallInitializersOnCorrectEventLoop", testBootstrapsCallInitializersOnCorrectEventLoop),
+           ]
+   }
+}
+

Tests/NIOTests/BootstrapTest.swift

@@ -0,0 +1,59 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIO
+import XCTest
+
+class BootstrapTest: XCTestCase {
+    func testBootstrapsCallInitializersOnCorrectEventLoop() throws {
+        for numThreads in [1 /* everything on one event loop */,
+                           2 /* some stuff has shared event loops */,
+                           5 /* everything on a different event loop */] {
+            let group = MultiThreadedEventLoopGroup(numThreads: numThreads)
+            defer {
+                XCTAssertNoThrow(try group.syncShutdownGracefully())
+            }
+
+            let childChannelDone: EventLoopPromise<Void> = group.next().newPromise()
+            let serverChannelDone: EventLoopPromise<Void> = group.next().newPromise()
+            let serverChannel = try ServerBootstrap(group: group)
+                .childChannelInitializer { channel in
+                    XCTAssert(channel.eventLoop.inEventLoop)
+                    childChannelDone.succeed(result: ())
+                    return channel.eventLoop.newSucceededFuture(result: ())
+                }
+                .serverChannelInitializer { channel in
+                    XCTAssert(channel.eventLoop.inEventLoop)
+                    serverChannelDone.succeed(result: ())
+                    return channel.eventLoop.newSucceededFuture(result: ())
+                }
+                .bind(host: "localhost", port: 0)
+                .wait()
+            defer {
+                XCTAssertNoThrow(try serverChannel.close().wait())
+            }
+
+            let client = try ClientBootstrap(group: group)
+                .channelInitializer { channel in
+                    XCTAssert(channel.eventLoop.inEventLoop)
+                    return channel.eventLoop.newSucceededFuture(result: ())
+                }
+                .connect(to: serverChannel.localAddress!)
+                .wait()
+            defer {
+                XCTAssertNoThrow(try client.syncCloseAcceptingAlreadyClosed())
+            }
+        }
+    }
+}

docker/docker-compose.1404.41.yaml

@@ -17,7 +17,7 @@ services:
     image: swift-nio:14.04-4.1
     environment:
       - MAX_ALLOCS_ALLOWED_1000_reqs_1_conn=47000
-      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=792100
+      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=718100
       - MAX_ALLOCS_ALLOWED_ping_pong_1000_reqs_1_conn=4600
       - MAX_ALLOCS_ALLOWED_bytebuffer_lots_of_rw=3100
       - MAX_ALLOCS_ALLOWED_future_lots_of_callbacks=99100
@@ -26,7 +26,7 @@ services:
     image: swift-nio:14.04-4.1
     environment:
       - MAX_ALLOCS_ALLOWED_1000_reqs_1_conn=47000
-      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=792100
+      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=718100
       - MAX_ALLOCS_ALLOWED_ping_pong_1000_reqs_1_conn=4600
       - MAX_ALLOCS_ALLOWED_bytebuffer_lots_of_rw=3100
       - MAX_ALLOCS_ALLOWED_future_lots_of_callbacks=99100

docker/docker-compose.1604.41.yaml

@@ -16,7 +16,7 @@ services:
     image: swift-nio:16.04-4.1
     environment:
       - MAX_ALLOCS_ALLOWED_1000_reqs_1_conn=47000
-      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=793100
+      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=718100
       - MAX_ALLOCS_ALLOWED_ping_pong_1000_reqs_1_conn=4600
       - MAX_ALLOCS_ALLOWED_bytebuffer_lots_of_rw=3100
       - MAX_ALLOCS_ALLOWED_future_lots_of_callbacks=99100
@@ -25,7 +25,7 @@ services:
     image: swift-nio:16.04-4.1
     environment:
       - MAX_ALLOCS_ALLOWED_1000_reqs_1_conn=47000
-      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=793100
+      - MAX_ALLOCS_ALLOWED_1_reqs_1000_conn=718100
       - MAX_ALLOCS_ALLOWED_ping_pong_1000_reqs_1_conn=4600
       - MAX_ALLOCS_ALLOWED_bytebuffer_lots_of_rw=3100
       - MAX_ALLOCS_ALLOWED_future_lots_of_callbacks=99100