finagle-netty4: Do Not Set Auto Read When Using LocalChannel

Problem

Turning off the AUTO_READ channel option causes SSL/TLS to not
work properly when using Finagle with a Netty `LocalChannel`.

Solution

Do not set the AUTO_READ channel option and let it use the default
value instead.

Result

SSL/TLS works in Finagle when using a Netty `LocalChannel`.

JIRA Issues: CSL-8197

Differential Revision: https://phabricator.twitter.biz/D319011

CHANGELOG.rst

@@ -20,6 +20,10 @@ Runtime Behavior Changes
 * finagle: Upgrade to Netty 4.1.35.Final and netty-tcnative 2.0.25.Final.
   ``PHAB_ID=D312439``
 
+* finagle-netty4: When using a Netty `LocalChannel`, the value of the `BackPressure`
+  stack param is effectively changed to `backPressureDisabled` so that other functionality
+  (e.g. SSL/TLS) works as expected. ``PHAB_ID=D319011``
+
 19.5.1
 ------
 

finagle-netty4/src/main/scala/com/twitter/finagle/netty4/ConnectionBuilder.scala

@@ -179,15 +179,19 @@ private[finagle] object ConnectionBuilder {
       .group(params[param.WorkerPool].eventLoopGroup)
       .channel(channelClass)
       .option(ChannelOption.ALLOCATOR, allocator)
-      .option[JBool](ChannelOption.AUTO_READ, !backpressure) // backpressure! no reads on transport => no reads on the socket
       .option[JInt](ChannelOption.CONNECT_TIMEOUT_MILLIS, compensatedConnectTimeoutMs.toInt)
       .handler(init)
 
-    // Trying to set these options results in an 'Unknown channel option' warning for `LocalChannel`
     if (!isLocal(addr)) {
+      // Trying to set SO_REUSEADDR and TCP_NODELAY gives 'Unkonwn channel option' warnings
+      // when used with `LocalServerChannel`. So skip setting them at all.
+      bootstrap.option[JBool](ChannelOption.TCP_NODELAY, noDelay)
+      bootstrap.option[JBool](ChannelOption.SO_REUSEADDR, reuseAddr)
+
+      // Turning off AUTO_READ causes SSL/TLS errors in Finagle when using `LocalChannel`.
+      // So skip setting it at all.
       bootstrap
-        .option[JBool](ChannelOption.TCP_NODELAY, noDelay)
-        .option[JBool](ChannelOption.SO_REUSEADDR, reuseAddr)
+        .option[JBool](ChannelOption.AUTO_READ, !backpressure) // backpressure! no reads on transport => no reads on the socket
     }
 
     val Transport.Liveness(_, _, keepAlive) = params[Transport.Liveness]

finagle-netty4/src/main/scala/com/twitter/finagle/netty4/ListeningServerBuilder.scala

@@ -93,9 +93,13 @@ private class ListeningServerBuilder(
           bootstrap.channel(classOf[NioServerSocketChannel])
         }
         // Trying to set SO_REUSEADDR and TCP_NODELAY gives 'Unkonwn channel option' warnings
-        // when used with `LocalServerChannel`.
+        // when used with `LocalServerChannel`. So skip setting them at all.
         bootstrap.option[JBool](ChannelOption.SO_REUSEADDR, reuseAddr)
         bootstrap.childOption[JBool](ChannelOption.TCP_NODELAY, noDelay)
+
+        // Turning off AUTO_READ causes SSL/TLS errors in Finagle when using `LocalChannel`.
+        // So skip setting it at all.
+        bootstrap.childOption[JBool](ChannelOption.AUTO_READ, !backPressureEnabled)
       }
 
       bootstrap.group(bossLoop, params[param.WorkerPool].eventLoopGroup)
@@ -106,7 +110,6 @@ private class ListeningServerBuilder(
       sendBufSize.foreach(bootstrap.childOption[JInt](ChannelOption.SO_SNDBUF, _))
       recvBufSize.foreach(bootstrap.childOption[JInt](ChannelOption.SO_RCVBUF, _))
       keepAlive.foreach(bootstrap.childOption[JBool](ChannelOption.SO_KEEPALIVE, _))
-      bootstrap.childOption[JBool](ChannelOption.AUTO_READ, !backPressureEnabled)
       params[Listener.TrafficClass].value.foreach { tc =>
         bootstrap.option[JInt](Netty4Listener.TrafficClass, tc)
         bootstrap.childOption[JInt](Netty4Listener.TrafficClass, tc)

finagle-netty4/src/test/scala/com/twitter/finagle/netty4/LocalChannelTest.scala

@@ -2,7 +2,9 @@ package com.twitter.finagle.netty4
 
 import com.twitter.finagle.{Service, Stack}
 import com.twitter.finagle.client.utils.StringClient.StringClientPipeline
+import com.twitter.finagle.netty4.ssl.Netty4SslTestComponents._
 import com.twitter.finagle.server.utils.StringServer
+import com.twitter.finagle.transport.Transport
 import com.twitter.util.{Await, Awaitable, Future, Duration}
 import io.netty.channel.local.LocalAddress
 import org.scalatest.FunSuite
@@ -16,16 +18,18 @@ import org.scalatest.FunSuite
  */
 class LocalChannelTest extends FunSuite {
 
-  def await[T](a: Awaitable[T]): T = Await.result(a, Duration.fromSeconds(1))
+  private[this] def await[T](a: Awaitable[T]): T = Await.result(a, Duration.fromSeconds(1))
+
+  private[this] val service = Service.mk[String, String](Future.value)
 
   // To use a full Finagle client (i.e. StringClient) here
   // requires additional changes to `c.t.f.Address` to allow
   // it to accept a `SocketAddress` and not just an `InetSocketAddress`.
   // So, we test components of the client here instead as a substitute
   // for the time being.
   test("client components and server can communicate") {
-    val addr = new LocalAddress("LocalChannelTest1")
-    val server = StringServer.server.serve(addr, Service.mk[String, String](Future.value))
+    val addr = new LocalAddress("LocalChannelTestPlain")
+    val server = StringServer.server.serve(addr, service)
     try {
       val transporter =
         Netty4Transporter.raw[String, String](StringClientPipeline, addr, Stack.Params.empty)
@@ -42,4 +46,26 @@ class LocalChannelTest extends FunSuite {
     }
   }
 
+  test("client components and server can communicate over SSL/TLS") {
+    val addr = new LocalAddress("LocalChannelTestTls")
+    val server = StringServer.server.withTransport.tls(serverConfig).serve(addr, service)
+    try {
+      val transporter =
+        Netty4Transporter.raw[String, String](
+          StringClientPipeline,
+          addr,
+          Stack.Params.empty + Transport.ClientSsl(Some(clientConfig)))
+      val transport = await(transporter())
+      transport.write("Finagle over TLS")
+      try {
+        val result = await(transport.read())
+        assert(result == "Finagle over TLS")
+      } finally {
+        transport.close()
+      }
+    } finally {
+      server.close()
+    }
+  }
+
 }

finagle-netty4/src/test/scala/com/twitter/finagle/netty4/ssl/Netty4SslTestComponents.scala

@@ -40,6 +40,17 @@ object Netty4SslTestComponents {
       false
   }
 
+  val serverConfig = SslServerConfiguration(
+    keyCredentials = KeyCredentials.CertAndKey(serverCert, serverKey),
+    trustCredentials = TrustCredentials.CertCollection(chainCert),
+    clientAuth = ClientAuth.Needed
+  )
+
+  val clientConfig = SslClientConfiguration(
+    keyCredentials = KeyCredentials.CertAndKey(clientCert, clientKey),
+    trustCredentials = TrustCredentials.CertCollection(chainCert)
+  )
+
   def getPort(server: ListeningServer): Int =
     server.boundAddress.asInstanceOf[InetSocketAddress].getPort
 
@@ -50,11 +61,6 @@ object Netty4SslTestComponents {
     sessionVerifier: SslClientSessionVerifier = SslClientSessionVerifier.AlwaysValid,
     onHandshakeComplete: Try[Unit] => Unit = _ => ()
   ): Service[String, String] = {
-    val clientConfig = SslClientConfiguration(
-      keyCredentials = KeyCredentials.CertAndKey(clientCert, clientKey),
-      trustCredentials = TrustCredentials.CertCollection(chainCert)
-    )
-
     // inject a handler which is called when the ssl handshake is complete.
     // Note, this isn't something which we expose outside of finagle and thus,
     // we don't have a "friendly" with* API for it.
@@ -74,12 +80,6 @@ object Netty4SslTestComponents {
     statsReceiver: StatsReceiver = NullStatsReceiver,
     sessionVerifier: SslServerSessionVerifier = SslServerSessionVerifier.AlwaysValid
   ): ListeningServer = {
-    val serverConfig = SslServerConfiguration(
-      keyCredentials = KeyCredentials.CertAndKey(serverCert, serverKey),
-      trustCredentials = TrustCredentials.CertCollection(chainCert),
-      clientAuth = ClientAuth.Needed
-    )
-
     StringServer.server.withTransport
       .tls(serverConfig, sessionVerifier)
       .withLabel(label)