[finagle-core] Eliminate unnecessary calls in the filter chain

Problem
For extra safety we wrap the next service in the chain with Service.rescue, even though in the majority of cases it's an adapter between Service and Filter interfaces.

Solution
Eliminate this extra call.

Before
```
Benchmark                      (numAndThens)  Mode  Cnt    Score   Error  Units
FilterBenchmark.andThenFilter             20  avgt   10   55.079 ± 0.294  ns/op
FilterBenchmark.andThenFilter             50  avgt   10  163.251 ± 0.903  ns/op
```

After
```
Benchmark                      (numAndThens)  Mode  Cnt   Score   Error  Units
FilterBenchmark.andThenFilter             20  avgt   10  29.767 ± 0.206  ns/op
FilterBenchmark.andThenFilter             50  avgt   10  86.184 ± 0.588  ns/op
```

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

finagle-benchmark/src/main/scala/com/twitter/finagle/FilterBenchmark.scala

@@ -9,7 +9,7 @@ import org.openjdk.jmh.annotations._
 class FilterBenchmark extends StdBenchAnnotations {
   import FilterBenchmark._
 
-  @Param(Array("10"))
+  @Param(Array("20", "50"))
   var numAndThens: Int = _
 
   val mutable = new Mutable(0)

finagle-core/src/main/scala/com/twitter/finagle/Filter.scala

@@ -1,6 +1,8 @@
 package com.twitter.finagle
 
-import com.twitter.util.{Future, Time}
+import com.twitter.util.Future
+import com.twitter.util.Time
+
 import java.util.concurrent.atomic.AtomicBoolean
 import scala.util.control.NonFatal
 
@@ -31,6 +33,8 @@ import scala.util.control.NonFatal
 abstract class Filter[-ReqIn, +RepOut, +ReqOut, -RepIn]
     extends ((ReqIn, Service[ReqOut, RepIn]) => Future[RepOut]) {
   import Filter.AndThen
+  import Filter.Adapter
+  import Filter.SafeService
 
   /**
    * This is the method to override/implement to create your own Filter.
@@ -82,25 +86,18 @@ abstract class Filter[-ReqIn, +RepOut, +ReqOut, -RepIn]
   def andThen(service: Service[ReqOut, RepIn]): Service[ReqIn, RepOut] = {
     // wrap user-supplied Services such that NonFatal exceptions thrown synchronously
     // in their `Service.apply` method are lifted into a `Future.exception`.
-    val rescued = Service.rescue(service)
+    val rescued = service match {
+      case _: SafeService[_, _] =>
+        service
+      case _ =>
+        Service.rescue(service)
+    }
     andThenService(rescued)
   }
 
-  private def andThenService(service: Service[ReqOut, RepIn]): Service[ReqIn, RepOut] = {
-    new Service[ReqIn, RepOut] {
-      def apply(request: ReqIn): Future[RepOut] = {
-        // wrap the user-supplied `Filter.apply`, lifting synchronous exceptions into Futures.
-        try Filter.this.apply(request, service)
-        catch {
-          case NonFatal(e) => Future.exception(e)
-        }
-      }
-      override def close(deadline: Time): Future[Unit] = service.close(deadline)
-      override def status: Status = service.status
-      override def toString: String = {
-        s"${Filter.this.toString}.andThen(${service.toString})"
-      }
-    }
+  private[twitter] def andThenService(next: Service[ReqOut, RepIn]): Service[ReqIn, RepOut] = {
+    val filter: Filter[ReqIn, RepOut, ReqOut, RepIn] = Filter.this
+    new Adapter(filter, next)
   }
 
   /**
@@ -224,6 +221,34 @@ object Filter {
     }
   }
 
+  /**
+   * A marker interface indicating that it's safe to call .apply without wrapping with .rescue
+   * @tparam Req
+   * @tparam Rep
+   */
+  private[twitter] trait SafeService[Req, Rep] extends Service[Req, Rep] {}
+
+  private class Adapter[ReqIn, RepOut, ReqOut, RepIn](
+    filter: Filter[ReqIn, RepOut, ReqOut, RepIn],
+    next: Service[ReqOut, RepIn])
+      extends SafeService[ReqIn, RepOut] {
+    def apply(request: ReqIn): Future[RepOut] = {
+      // wrap the user-supplied `Filter.apply`, lifting synchronous exceptions into Futures.
+      try filter.apply(request, next)
+      catch {
+        case NonFatal(e) => Future.exception(e)
+      }
+    }
+
+    override def close(deadline: Time): Future[Unit] = next.close(deadline)
+
+    override def status: Status = next.status
+
+    override def toString: String = {
+      s"${filter.toString}.andThen(${next.toString})"
+    }
+  }
+
   private case object Identity extends SimpleFilter[Any, Nothing] {
     override def andThen[Req2, Rep2](
       next: Filter[Any, Nothing, Req2, Rep2]