Always use the same event loop for a client connection (#562)

* Always use the same event loop for a client connection

Motivation:

It might be useful for other applications interacting with gRPC to know
that a client connection will always use the same event loop, event if
it reconnects.

Modifications:

Use the same event loop when attempting reconnections.

Result:

A connection will only ever use one event loop.

* fix endif

Package.swift

@@ -34,9 +34,8 @@ let package = Package(
     .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.5.0"),
     // TLS via SwiftNIO
     .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.4.0"),
-    // Support for Network.framework where possible. Note: from 1.0.2 the package
-    // is essentially an empty import on platforms where it isn't supported.
-    .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.0.2"),
+    // Support for Network.framework where possible.
+    .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.1.0"),
 
     // Official SwiftProtobuf library, for [de]serializing data to send on the wire.
     .package(url: "https://github.com/apple/swift-protobuf.git", from: "1.5.0"),

Sources/GRPC/ClientConnection.swift

@@ -109,6 +109,7 @@ public class ClientConnection {
 
     self.channel = ClientConnection.makeChannel(
       configuration: self.configuration,
+      eventLoop: self.configuration.eventLoopGroup.next(),
       connectivity: self.connectivity,
       backoffIterator: self.configuration.connectionBackoff?.makeIterator(),
       logger: self.logger
@@ -230,6 +231,7 @@ extension ClientConnection {
       self.logger.debug("client connection channel closed, creating a new one")
       self.channel = ClientConnection.makeChannel(
         configuration: self.configuration,
+        eventLoop: channel.eventLoop,
         connectivity: self.connectivity,
         backoffIterator: self.configuration.connectionBackoff?.makeIterator(),
         logger: self.logger
@@ -258,11 +260,13 @@ extension ClientConnection {
   /// `ConnectionBackoffIterator` is provided.
   ///
   /// - Parameter configuration: The configuration to start the connection with.
+  /// - Parameter eventLoop: The event loop to use for this connection.
   /// - Parameter connectivity: A connectivity state monitor.
   /// - Parameter backoffIterator: An `Iterator` for `ConnectionBackoff` providing a sequence of
   ///     connection timeouts and backoff to use when attempting to create a connection.
   private class func makeChannel(
     configuration: Configuration,
+    eventLoop: EventLoop,
     connectivity: ConnectivityStateMonitor,
     backoffIterator: ConnectionBackoffIterator?,
     logger: Logger
@@ -277,7 +281,7 @@ extension ClientConnection {
 
     let bootstrap = self.makeBootstrap(
       configuration: configuration,
-      group: configuration.eventLoopGroup,
+      eventLoop: eventLoop,
       timeout: timeoutAndBackoff?.timeout,
       connectivityMonitor: connectivity,
       logger: logger
@@ -330,6 +334,7 @@ extension ClientConnection {
     return eventLoop.scheduleTask(in: .seconds(timeInterval: timeout)) {
       ClientConnection.makeChannel(
         configuration: configuration,
+        eventLoop: eventLoop,
         connectivity: connectivity,
         backoffIterator: backoffIterator,
         logger: logger
@@ -345,12 +350,12 @@ extension ClientConnection {
   /// handlers detailed in the documentation for `ClientConnection`.
   ///
   /// - Parameter configuration: The configuration to prepare the bootstrap with.
-  /// - Parameter group: The `EventLoopGroup` to use for the bootstrap.
+  /// - Parameter eventLoop: The `EventLoop` to use for the bootstrap.
   /// - Parameter timeout: The connection timeout in seconds.
   /// - Parameter connectivityMonitor: The connectivity state monitor for the created channel.
   private class func makeBootstrap(
     configuration: Configuration,
-    group: EventLoopGroup,
+    eventLoop: EventLoop,
     timeout: TimeInterval?,
     connectivityMonitor: ConnectivityStateMonitor,
     logger: Logger
@@ -367,7 +372,7 @@ extension ClientConnection {
       }
     }
 
-    let bootstrap = PlatformSupport.makeClientBootstrap(group: group)
+    let bootstrap = PlatformSupport.makeClientBootstrap(group: eventLoop)
       // Enable SO_REUSEADDR and TCP_NODELAY.
       .channelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
       .channelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
@@ -385,6 +390,7 @@ extension ClientConnection {
       logger.info("setting connect timeout to \(timeout) seconds")
       return bootstrap.connectTimeout(.seconds(timeInterval: timeout))
     } else {
+      logger.info("no connect timeout provided")
       return bootstrap
     }
   }

Sources/GRPC/PlatformSupport.swift

@@ -165,6 +165,9 @@ public enum PlatformSupport {
       if let tsGroup = group as? NIOTSEventLoopGroup {
         logger.debug("Network.framework is available and the group is correctly typed, creating a NIOTSConnectionBootstrap")
         return NIOTSConnectionBootstrap(group: tsGroup)
+      } else if let qosEventLoop = group as? QoSEventLoop {
+        logger.debug("Network.framework is available and the group is correctly typed, creating a NIOTSConnectionBootstrap")
+        return NIOTSConnectionBootstrap(group: qosEventLoop)
       }
       logger.debug("Network.framework is available but the group is not typed for NIOTS, falling back to ClientBootstrap")
     }
@@ -186,6 +189,9 @@ public enum PlatformSupport {
       if let tsGroup = group as? NIOTSEventLoopGroup {
         logger.debug("Network.framework is available and the group is correctly typed, creating a NIOTSListenerBootstrap")
         return NIOTSListenerBootstrap(group: tsGroup)
+      } else if let qosEventLoop = group as? QoSEventLoop {
+        logger.debug("Network.framework is available and the group is correctly typed, creating a NIOTSListenerBootstrap")
+        return NIOTSListenerBootstrap(group: qosEventLoop)
       }
       logger.debug("Network.framework is available but the group is not typed for NIOTS, falling back to ServerBootstrap")
     }

Tests/GRPCTests/PlatformSupportTests.swift

@@ -0,0 +1,115 @@
+import Foundation
+import GRPC
+import NIO
+import NIOTransportServices
+import XCTest
+
+class PlatformSupportTests: GRPCTestCase {
+  var group: EventLoopGroup!
+
+  override func tearDown() {
+    XCTAssertNoThrow(try self.group?.syncShutdownGracefully())
+  }
+
+  func testMakeEventLoopGroupReturnsMultiThreadedGroupForPosix() {
+    self.group = PlatformSupport.makeEventLoopGroup(
+      loopCount: 1,
+      networkPreference: .userDefined(.posix)
+    )
+
+    XCTAssertTrue(self.group is MultiThreadedEventLoopGroup)
+  }
+
+  func testMakeEventLoopGroupReturnsNIOTSGroupForNetworkFramework() {
+    // If we don't have Network.framework then we can't test this.
+    #if canImport(Network)
+    guard #available(macOS 10.14, *) else { return }
+
+    self.group = PlatformSupport.makeEventLoopGroup(
+      loopCount: 1,
+      networkPreference: .userDefined(.networkFramework)
+    )
+
+    XCTAssertTrue(self.group is NIOTSEventLoopGroup)
+    #endif
+  }
+
+  func testMakeClientBootstrapReturnsClientBootstrapForMultiThreadedGroup() {
+    self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    let bootstrap = PlatformSupport.makeClientBootstrap(group: self.group)
+    XCTAssertTrue(bootstrap is ClientBootstrap)
+  }
+
+  func testMakeClientBootstrapReturnsClientBootstrapForEventLoop() {
+    self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    let eventLoop = self.group.next()
+    let bootstrap = PlatformSupport.makeClientBootstrap(group: eventLoop)
+    XCTAssertTrue(bootstrap is ClientBootstrap)
+  }
+
+  func testMakeClientBootstrapReturnsNIOTSConnectionBootstrapForNIOTSGroup() {
+    // If we don't have Network.framework then we can't test this.
+    #if canImport(Network)
+    guard #available(macOS 10.14, *) else { return }
+
+    self.group = NIOTSEventLoopGroup(loopCount: 1)
+    let bootstrap = PlatformSupport.makeClientBootstrap(group: self.group)
+    XCTAssertTrue(bootstrap is NIOTSConnectionBootstrap)
+    #endif
+  }
+
+  func testMakeClientBootstrapReturnsNIOTSConnectionBootstrapForQoSEventLoop() {
+    // If we don't have Network.framework then we can't test this.
+    #if canImport(Network)
+    guard #available(macOS 10.14, *) else { return }
+
+    self.group = NIOTSEventLoopGroup(loopCount: 1)
+
+    let eventLoop = self.group.next()
+    XCTAssertTrue(eventLoop is QoSEventLoop)
+
+    let bootstrap = PlatformSupport.makeClientBootstrap(group: eventLoop)
+    XCTAssertTrue(bootstrap is NIOTSConnectionBootstrap)
+    #endif
+  }
+
+  func testMakeServerBootstrapReturnsServerBootstrapForMultiThreadedGroup() {
+    self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    let bootstrap = PlatformSupport.makeServerBootstrap(group: self.group)
+    XCTAssertTrue(bootstrap is ServerBootstrap)
+  }
+
+  func testMakeServerBootstrapReturnsServerBootstrapForEventLoop() {
+    self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+
+    let eventLoop = self.group.next()
+    let bootstrap = PlatformSupport.makeServerBootstrap(group: eventLoop)
+    XCTAssertTrue(bootstrap is ServerBootstrap)
+  }
+
+  func testMakeServerBootstrapReturnsNIOTSListenerBootstrapForNIOTSGroup() {
+    // If we don't have Network.framework then we can't test this.
+    #if canImport(Network)
+    guard #available(macOS 10.14, *) else { return }
+
+    self.group = NIOTSEventLoopGroup(loopCount: 1)
+    let bootstrap = PlatformSupport.makeServerBootstrap(group: self.group)
+    XCTAssertTrue(bootstrap is NIOTSListenerBootstrap)
+    #endif
+  }
+
+  func testMakeServerBootstrapReturnsNIOTSListenerBootstrapForQoSEventLoop() {
+    // If we don't have Network.framework then we can't test this.
+    #if canImport(Network)
+    guard #available(macOS 10.14, *) else { return }
+
+    self.group = NIOTSEventLoopGroup(loopCount: 1)
+
+    let eventLoop = self.group.next()
+    XCTAssertTrue(eventLoop is QoSEventLoop)
+
+    let bootstrap = PlatformSupport.makeServerBootstrap(group: eventLoop)
+    XCTAssertTrue(bootstrap is NIOTSListenerBootstrap)
+    #endif
+  }
+}

Tests/GRPCTests/XCTestManifests.swift

@@ -384,6 +384,24 @@ extension LengthPrefixedMessageReaderTests {
     ]
 }
 
+extension PlatformSupportTests {
+    // DO NOT MODIFY: This is autogenerated, use:
+    //   `swift test --generate-linuxmain`
+    // to regenerate.
+    static let __allTests__PlatformSupportTests = [
+        ("testMakeClientBootstrapReturnsClientBootstrapForEventLoop", testMakeClientBootstrapReturnsClientBootstrapForEventLoop),
+        ("testMakeClientBootstrapReturnsClientBootstrapForMultiThreadedGroup", testMakeClientBootstrapReturnsClientBootstrapForMultiThreadedGroup),
+        ("testMakeClientBootstrapReturnsNIOTSConnectionBootstrapForNIOTSGroup", testMakeClientBootstrapReturnsNIOTSConnectionBootstrapForNIOTSGroup),
+        ("testMakeClientBootstrapReturnsNIOTSConnectionBootstrapForQoSEventLoop", testMakeClientBootstrapReturnsNIOTSConnectionBootstrapForQoSEventLoop),
+        ("testMakeEventLoopGroupReturnsMultiThreadedGroupForPosix", testMakeEventLoopGroupReturnsMultiThreadedGroupForPosix),
+        ("testMakeEventLoopGroupReturnsNIOTSGroupForNetworkFramework", testMakeEventLoopGroupReturnsNIOTSGroupForNetworkFramework),
+        ("testMakeServerBootstrapReturnsNIOTSListenerBootstrapForNIOTSGroup", testMakeServerBootstrapReturnsNIOTSListenerBootstrapForNIOTSGroup),
+        ("testMakeServerBootstrapReturnsNIOTSListenerBootstrapForQoSEventLoop", testMakeServerBootstrapReturnsNIOTSListenerBootstrapForQoSEventLoop),
+        ("testMakeServerBootstrapReturnsServerBootstrapForEventLoop", testMakeServerBootstrapReturnsServerBootstrapForEventLoop),
+        ("testMakeServerBootstrapReturnsServerBootstrapForMultiThreadedGroup", testMakeServerBootstrapReturnsServerBootstrapForMultiThreadedGroup),
+    ]
+}
+
 extension ServerDelayedThrowingTests {
     // DO NOT MODIFY: This is autogenerated, use:
     //   `swift test --generate-linuxmain`
@@ -468,6 +486,7 @@ public func __allTests() -> [XCTestCaseEntry] {
         testCase(HTTP1ToRawGRPCServerCodecTests.__allTests__HTTP1ToRawGRPCServerCodecTests),
         testCase(ImmediatelyFailingProviderTests.__allTests__ImmediatelyFailingProviderTests),
         testCase(LengthPrefixedMessageReaderTests.__allTests__LengthPrefixedMessageReaderTests),
+        testCase(PlatformSupportTests.__allTests__PlatformSupportTests),
         testCase(ServerDelayedThrowingTests.__allTests__ServerDelayedThrowingTests),
         testCase(ServerErrorTransformingTests.__allTests__ServerErrorTransformingTests),
         testCase(ServerThrowingTests.__allTests__ServerThrowingTests),