Support hopping event loops with EventLoopFuture (#177)

Motivation:

A somewhat common requirement when working with chains of futures
is needing to hop from one event loop to another. This is particularly
common when relying on the fact that EventLoopFuture will synchronize
with the event loop that created it: you often want to rely on that
implicit locking, rather than use Dispatch or Lock to achieve the
same effect.

While doing this hop requires relatively little code, it's not
necessarily totally apparent to new users how they would do it.
Additionally, the most naive implementation incurs the overhead of
allocations and reference counting in cases where it's not necessary
(e.g. when you have only one event loop, or when both work items are
being scheduled on the same event loop).

For this reason, we should have a nice concise way for a user to
request this behaviour and get a relatively performant implementation
of the behaviour.

Modifications:

Added EventLoopFuture<T>.on(eventLoop:).

Changed AcceptHandler to use the new method rather than its (slower)
alternative.

Result:

Users will have an easier time working with EventLoopFutures.

Author: Lukasa

Sources/NIO/Bootstrap.swift

@@ -199,11 +199,9 @@ public final class ServerBootstrap {
 
         func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
             let accepted = self.unwrapInboundIn(data)
-            let hopEventLoopPromise: EventLoopPromise<()> = ctx.eventLoop.newPromise()
-            self.childChannelOptions.applyAll(channel: accepted).cascade(promise: hopEventLoopPromise)
             let childChannelInit = self.childChannelInit ?? { (_: Channel) in ctx.eventLoop.newSucceededFuture(result: ()) }
 
-            hopEventLoopPromise.futureResult.then {
+            self.childChannelOptions.applyAll(channel: accepted).hopTo(eventLoop: ctx.eventLoop).then {
                 assert(ctx.eventLoop.inEventLoop)
                 return childChannelInit(accepted)
             }.then { () -> EventLoopFuture<()> in

Sources/NIO/EventLoopFuture.swift

@@ -654,14 +654,7 @@ extension EventLoopFuture {
             return CallbackList()
         }
 
-        let hopOver: EventLoopFuture<U>
-        if self.eventLoop === other.eventLoop {
-            hopOver = other
-        } else {
-            let hopOverP = EventLoopPromise<U>(eventLoop: self.eventLoop, file: file, line: line)
-            other.cascade(promise: hopOverP)
-            hopOver = hopOverP.futureResult
-        }
+        let hopOver = other.hopTo(eventLoop: self.eventLoop)
         hopOver._whenComplete { () -> CallbackList in
             assert(self.eventLoop.inEventLoop)
             switch other.value! {
@@ -796,7 +789,29 @@ extension EventLoopFuture {
         p0.succeed(result: ())
         return body
     }
+}
 
+extension EventLoopFuture {
+    /// Returns an `EventLoopFuture` that fires when this future completes, but executes its callbacks on the
+    /// target event loop instead of the original one.
+    ///
+    /// It is common to want to "hop" event loops when you arrange some work: for example, you're closing one channel
+    /// from another, and want to hop back when the close completes. This method lets you spell that requirement
+    /// succinctly. It also contains an optimisation for the case when the loop you're hopping *from* is the same as
+    /// the one you're hopping *to*, allowing you to avoid doing allocations in that case.
+    ///
+    /// - parameters:
+    ///     - target: The `EventLoop` that the returned `EventLoopFuture` will run on.
+    /// - returns: An `EventLoopFuture` whose callbacks run on `target` instead of the original loop.
+    func hopTo(eventLoop target: EventLoop) -> EventLoopFuture<T> {
+        if target === self.eventLoop {
+            // We're already on that event loop, nothing to do here. Save an allocation.
+            return self
+        }
+        let hoppingPromise: EventLoopPromise<T> = target.newPromise()
+        self.cascade(promise: hoppingPromise)
+        return hoppingPromise.futureResult
+    }
 }
 
 /// Execute the given function and synchronously complete the given `EventLoopPromise` (if not `nil`).

Tests/NIOTests/EventLoopFutureTest+XCTest.swift

@@ -41,6 +41,9 @@ extension EventLoopFutureTest {
                 ("testEventLoopHoppingAndAll", testEventLoopHoppingAndAll),
                 ("testEventLoopHoppingAndAllWithFailures", testEventLoopHoppingAndAllWithFailures),
                 ("testFutureInVariousScenarios", testFutureInVariousScenarios),
+                ("testLoopHoppingHelperSuccess", testLoopHoppingHelperSuccess),
+                ("testLoopHoppingHelperFailure", testLoopHoppingHelperFailure),
+                ("testLoopHoppingHelperNoHopping", testLoopHoppingHelperNoHopping),
            ]
    }
 }

Tests/NIOTests/EventLoopFutureTest.swift

@@ -381,4 +381,52 @@ class EventLoopFutureTest : XCTestCase {
 
         XCTAssertNoThrow(try elg.syncShutdownGracefully())
     }
+
+    func testLoopHoppingHelperSuccess() throws {
+        let group = MultiThreadedEventLoopGroup(numThreads: 2)
+        let loop1 = group.next()
+        let loop2 = group.next()
+        XCTAssertFalse(loop1 === loop2)
+
+        let succeedingPromise: EventLoopPromise<Void> = loop1.newPromise()
+        let succeedingFuture = succeedingPromise.futureResult.map {
+            XCTAssertTrue(loop1.inEventLoop)
+        }.hopTo(eventLoop: loop2).map {
+            XCTAssertTrue(loop2.inEventLoop)
+        }
+        succeedingPromise.succeed(result: ())
+        XCTAssertNoThrow(try succeedingFuture.wait())
+    }
+
+    func testLoopHoppingHelperFailure() throws {
+        let group = MultiThreadedEventLoopGroup(numThreads: 2)
+        let loop1 = group.next()
+        let loop2 = group.next()
+        XCTAssertFalse(loop1 === loop2)
+
+        let failingPromise: EventLoopPromise<Void> = loop2.newPromise()
+        let failingFuture = failingPromise.futureResult.thenIfErrorThrowing { error in
+            XCTAssertEqual(error as? EventLoopFutureTestError, EventLoopFutureTestError.example)
+            XCTAssertTrue(loop2.inEventLoop)
+            throw error
+        }.hopTo(eventLoop: loop1).mapIfError { error in
+            XCTAssertEqual(error as? EventLoopFutureTestError, EventLoopFutureTestError.example)
+            XCTAssertTrue(loop1.inEventLoop)
+        }
+
+        failingPromise.fail(error: EventLoopFutureTestError.example)
+        XCTAssertNoThrow(try failingFuture.wait())
+    }
+
+    func testLoopHoppingHelperNoHopping() throws {
+        let group = MultiThreadedEventLoopGroup(numThreads: 2)
+        let loop1 = group.next()
+        let loop2 = group.next()
+        XCTAssertFalse(loop1 === loop2)
+
+        let noHoppingPromise: EventLoopPromise<Void> = loop1.newPromise()
+        let noHoppingFuture = noHoppingPromise.futureResult.hopTo(eventLoop: loop1)
+        XCTAssertTrue(noHoppingFuture === noHoppingPromise.futureResult)
+        noHoppingPromise.succeed(result: ())
+    }
 }