finagle: Use ImmediateValueFuture in service acquisition

Problem

Before we even begin to send a finagle request over the wire, we
perform "service acquisition" -- this is acquiring the connection over
which we will send the request. For simplicity, this can generally be viewed
as getting a cached connection -- that is, no "real" work happens. At multiple
points during this acquisition, we create a `Future.value` and then perform
some transformation on it. These continuations are not executed immediately--
they are instead submitted to the scheduler and executed in FIFO order for
that thread.

This is cheap -- but not free. When many such requests must be made to
backends (for example, many cache lookups for a given request), this results
in the following behaviour:

Each `Future.const`-returning stage of service acquisition happens serially
for all these outgoing request (e.g. endpoint name binding, load balancer node
picking). After *all* requests have completed (assuming "no-op" acquisition) service
acquisition, then requests are sent over the wire.

To understand why this is undesirable, let's consider latency distributions.
Assume that we need to issue 100 requests to backends. Assume p99 latency of
sending the request is 100ms and service acquisition is 1ms.

With the current behaviour, this means that the total latency of issuing the
requests is:

```
 (100 * 1ms) +  100ms (assuming perfect distribution so we hit the p99)
= 200ms avg p99
```

Now imagine that did not submit the aforementioned continuations to the
scheduler and instead executed them immediately. We would then be able to
issue each request immediately after its service acquisition. The total latency
would then be:

```
(100 * 1ms) + 50ms (assuming we hit the p99 on the middle request on avg)
= 150ms avg p99
```

We've managed to save 1/4 of the latency!

Solution

Use `ImmediateValueFuture` in service acquisition so we can accomplish the above.

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

Author: noctella

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

@@ -1,6 +1,8 @@
 package com.twitter.finagle
 
-import com.twitter.util.{Future, Time}
+import com.twitter.util.ImmediateValueFuture
+import com.twitter.util.Future
+import com.twitter.util.Time
 
 /**
  * Turns a [[com.twitter.finagle.ServiceFactory]] into a
@@ -66,9 +68,10 @@ object FactoryToService {
            *
            * This is too complicated.
            */
-          val service = Future.value(new ServiceProxy[Req, Rep](new FactoryToService(next)) {
-            override def close(deadline: Time): Future[Unit] = Future.Done
-          })
+          val service = new ImmediateValueFuture(
+            new ServiceProxy[Req, Rep](new FactoryToService(next)) {
+              override def close(deadline: Time): Future[Unit] = Future.Done
+            })
           new ServiceFactoryProxy(next) {
             override def apply(conn: ClientConnection): Future[ServiceProxy[Req, Rep]] = service
           }

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

@@ -2,6 +2,7 @@ package com.twitter.finagle
 
 import com.twitter.util.Closable
 import com.twitter.util.Future
+import com.twitter.util.ImmediateValueFuture
 import com.twitter.util.Return
 import com.twitter.util.Throw
 import com.twitter.util.Time
@@ -59,7 +60,7 @@ abstract class ServiceFactory[-Req, +Rep]
    * styles of factory wrappers.
    */
   def map[Req1, Rep1](f: Service[Req, Rep] => Service[Req1, Rep1]): ServiceFactory[Req1, Rep1] =
-    flatMap { s => Future.value(f(s)) }
+    flatMap { s => new ImmediateValueFuture(f(s)) }
 
   /**
    * Make a service that after dispatching a request on that service,
@@ -87,7 +88,7 @@ object ServiceFactory {
    */
   def const[Req, Rep](service: Service[Req, Rep]): ServiceFactory[Req, Rep] =
     new ServiceFactory[Req, Rep] {
-      private[this] val noRelease = Future.value(new ServiceProxy[Req, Rep](service) {
+      private[this] val noRelease = new ImmediateValueFuture(new ServiceProxy[Req, Rep](service) {
         // close() is meaningless on connectionless services.
         override def close(deadline: Time): Future[Unit] = Future.Done
       })

finagle-core/src/main/scala/com/twitter/finagle/factory/ServiceFactoryCache.scala

@@ -2,7 +2,8 @@ package com.twitter.finagle.factory
 
 import com.twitter.conversions.DurationOps._
 import com.twitter.finagle._
-import com.twitter.finagle.stats.{NullStatsReceiver, StatsReceiver}
+import com.twitter.finagle.stats.NullStatsReceiver
+import com.twitter.finagle.stats.StatsReceiver
 import com.twitter.util._
 import java.util
 import java.util.concurrent.atomic.AtomicInteger
@@ -24,7 +25,7 @@ private class IdlingFactory[Req, Rep](self: ServiceFactory[Req, Rep])
       Future.exception(exc)
 
     case Return(service) =>
-      Future.value(new ServiceProxy(service) {
+      new ImmediateValueFuture(new ServiceProxy(service) {
         override def close(deadline: Time): Future[Unit] = {
           decr()
           super.close(deadline)

finagle-core/src/main/scala/com/twitter/finagle/liveness/FailureAccrualFactory.scala

@@ -428,7 +428,7 @@ class FailureAccrualFactory[Req, Rep](
   }
 
   private[this] val acquireService: Try[Service[Req, Rep]] => Future[Service[Req, Rep]] = {
-    case Return(svc) => Future.value(makeService(svc))
+    case Return(svc) => new ImmediateValueFuture(makeService(svc))
     case t @ Throw(f: FailureFlags[_]) if f.isFlagged(FailureFlags.ClientDiscarded) =>
       didReceiveIgnorable()
       Future.const(t.cast[Service[Req, Rep]])

finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/LeastLoaded.scala

@@ -4,6 +4,7 @@ import com.twitter.finagle._
 import com.twitter.util.Throw
 import com.twitter.util.Time
 import com.twitter.util.Future
+import com.twitter.util.ImmediateValueFuture
 import com.twitter.util.Return
 import java.util.concurrent.atomic.AtomicInteger
 
@@ -28,7 +29,7 @@ private trait LeastLoaded[Req, Rep] extends BalancerNode[Req, Rep] { self: Balan
       counter.incrementAndGet()
       super.apply(conn).transform {
         case Return(svc) =>
-          Future.value(new ServiceProxy(svc) {
+          new ImmediateValueFuture(new ServiceProxy(svc) {
             override def close(deadline: Time): Future[Unit] =
               super.close(deadline).ensure {
                 counter.decrementAndGet()

finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/aperture/Expiration.scala

@@ -1,7 +1,8 @@
 package com.twitter.finagle.loadbalancer.aperture
 
 import com.twitter.finagle._
-import com.twitter.finagle.loadbalancer.{BalancerNode, NodeT}
+import com.twitter.finagle.loadbalancer.BalancerNode
+import com.twitter.finagle.loadbalancer.NodeT
 import com.twitter.util._
 
 private[loadbalancer] trait Expiration[Req, Rep] extends BalancerNode[Req, Rep] {
@@ -93,7 +94,7 @@ private[loadbalancer] trait Expiration[Req, Rep] extends BalancerNode[Req, Rep]
       onRequest()
       super.apply(conn).transform {
         case Return(svc) =>
-          Future.value(new ServiceProxy(svc) {
+          new ImmediateValueFuture(new ServiceProxy(svc) {
             override def close(deadline: Time) =
               super.close(deadline).ensure {
                 onResponse()

finagle-core/src/main/scala/com/twitter/finagle/pool/SingletonPool.scala

@@ -3,8 +3,14 @@ package com.twitter.finagle.pool
 import com.twitter.finagle._
 import com.twitter.finagle.client.StackClient
 import com.twitter.finagle.stats.StatsReceiver
-import com.twitter.util.{Future, Return, Throw, Time, Promise}
-import java.util.concurrent.atomic.{AtomicReference, AtomicInteger}
+import com.twitter.util.ImmediateValueFuture
+import com.twitter.util.Future
+import com.twitter.util.Promise
+import com.twitter.util.Return
+import com.twitter.util.Throw
+import com.twitter.util.Time
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.AtomicReference
 import scala.annotation.tailrec
 
 private[finagle] object SingletonPool {
@@ -41,7 +47,7 @@ private[finagle] object SingletonPool {
   private class RefcountedService[Req, Rep](underlying: Service[Req, Rep])
       extends ServiceProxy[Req, Rep](underlying) {
     private[this] val count = new AtomicInteger(1)
-    private[this] val future = Future.value(this)
+    private[this] val future = new ImmediateValueFuture(this)
 
     def open(): Future[Service[Req, Rep]] = {
       count.incrementAndGet()

finagle-mux/src/main/scala/com/twitter/finagle/Mux.scala

@@ -41,6 +41,7 @@ import com.twitter.finagle.transport.Transport
 import com.twitter.io.Buf
 import com.twitter.io.ByteReader
 import com.twitter.util.Future
+import com.twitter.util.ImmediateValueFuture
 import com.twitter.util.StorageUnit
 import io.netty.channel.Channel
 import io.netty.channel.ChannelPipeline
@@ -277,7 +278,7 @@ object Mux extends Client[mux.Request, mux.Response] with Server[mux.Request, mu
         params[CompressionPreferences].compressionPreferences
       )
 
-      Future.value(
+      new ImmediateValueFuture(
         new MuxClientNegotiatingSession(
           handle = handle,
           version = Mux.LatestVersion,

finagle-mux/src/main/scala/com/twitter/finagle/mux/pushsession/MuxClientSession.scala

@@ -102,9 +102,9 @@ private[finagle] final class MuxClientSession(
             h_canDispatch = CanDispatch.No
             apply(request)
 
-          case m @ Return(_) =>
+          case Return(response) =>
             h_canDispatch = CanDispatch.Yes
-            Future.const(m)
+            new ImmediateValueFuture(response)
 
           case t @ Throw(_) =>
             Future.const(t.cast[Response])
@@ -198,7 +198,7 @@ private[finagle] final class MuxClientSession(
     pp
   }
 
-  def asService: Future[Service[Request, Response]] = Future.value(MuxClientServiceImpl)
+  def asService: Future[Service[Request, Response]] = new ImmediateValueFuture(MuxClientServiceImpl)
 
   def status: Status = {
     // Return the worst status reported among