The primary construct in Scala to drive concurrent programming is Future. Found in the scala.concurrent package, Future can be seen as a container for a value that may not yet exist. Let's look at a simple example to illustrate usage:

scala> import scala.concurrent.Future 
import scala.concurrent.Future 
 
scala> import scala.concurrent.ExecutionContext 
import scala.concurrent.ExecutionContext 
 
scala> val context: ExecutionContext = scala.concurrent.ExecutionContext.global 
context: scala.concurrent.ExecutionContext = scala.concurrent.impl.ExecutionContextImpl@3fce8fd9 
 
scala> def example(){ 
  println("Starting the example") 
  Future{  
    println("Starting the Future") 
    Thread.sleep(1000)  // simulate computation 
    println("Done with the computation")   
 }(context) 
 
println("Ending example") 
} 

The preceding example shows a short method, creating a Future value simulates an expensive computation and prints a couple of lines to make it easier for us to understand the flow of the application. When running example, we see the following output:

scala> example() 
Starting the example 
Ending example 
Starting the future 
// a pause 
Done with the computation 

We can see that Future was executed after the end of the example method. This is because when a Future is created, it starts its computation concurrently. You may be wondering, "What is this context object of the ExecutionContext type that is used when creating the Future?" We will explore ExecutionContext in-depth shortly, but for now, we treat it as the the object that is responsible for the execution of the Future. We import scala.concurrent.ExecutionContext.global, which is a default object that is created by the standard library to be able to execute the Future.

A Future object is a stateful object. It is either not yet complete when the computation is underway or completed once the computation finishes. Furthermore, a completed Future can be either a success when the computation was able to complete, or it can be a failure if an exception was thrown during the computation.

The Future API provides combinators to compose the Future instances and manipulate the result that they contain:

scala> import scala.concurrent.ExecutionContext.Implicits.global 
import scala.concurrent.ExecutionContext.Implicits.global 
 
scala> import scala.concurrent.Future 
import scala.concurrent.Future 
 
scala> Future(1).map(_ + 1).filter(_ % 2 == 0).foreach(println) 
 
2 

This snippet from the Scala console shows construction of a Future data type that wraps a constant integer value. We see that the integer contained in the Future data type is transformed using functions that are similar to the ones that we expect to find on Option and collection data types. These transforms are applied once the preceding Future completes, and return a new Future.

As promised, we now look into ExecutionContext. The ExecutionContext can be thought of as the machinery behind Future that provides runtime asynchrony. In the previous snippet, a Future was created to perform simple addition and modulo division without explicitly providing an ExecutionContext instance at the call site. Instead, only an import of the global object was provided. The snippet executes because global is an implicit value and the signature of map accepts an implicit ExecutionContext. Let's look at the following signature of map to deepen our understanding:

def map[S](f: T => S)(implicit executor: ExecutionContext): Future[S] 

From the signature of map, we see that unlike the map transformation on List, the Future requires a curried, implicit ExecutionContext argument. To understand how an ExecutionContext provides runtime asynchrony, we need to first understand its operations:

trait ExecutionContext { 
  def execute(runnable: Runnable): Unit 
  def reportFailure(cause: Throwable): Unit 
  def prepare(): ExecutionContext = this 
} 

The execute is a side-effecting method that operates on a java.lang.Runnable. For those familiar with concurrency in Java, you most likely recall that Runnable is the commonly-used interface to allow threads and other java.util.concurrent abstractions to execute code concurrently. Although we do not know how Future achieves runtime asynchrony yet, we do know there is a link between Future execution and creation of a Runnable.

The next question we will answer is, "How do I create an ExecutionContext? " By studying the companion object, we discover the following signatures:

def fromExecutorService(e: ExecutorService, reporter: Throwable => Unit): ExecutionContextExecutorService 
def fromExecutorService(e: ExecutorService): ExecutionContextExecutorService 
def fromExecutor(e: Executor, reporter: Throwable => Unit): ExecutionContextExecutor 
def fromExecutor(e: Executor): ExecutionContextExecutor 

The standard library provides convenient ways to create an ExecutionContext from either a java.util.concurrent.Executor or java.util.concurrent.ExecutorService.

The return type of the factory methods is a more specialized version of ExecutionContext. The standard library defines the following inheritance chain for ExecutionContext:

trait ExecutionContextExecutor extends ExecutionContext with java.util.concurrent.Executor 
trait ExecutionContextExecutorService extends ExecutionContextExecutor with java.util.concurrent.ExecutorService 

Also, in the ExecutionContext companion object, we find the implicit context used in our first example, as follows:

def global: ExecutionContextExecutor = Implicits.global 

The documentation for the definition of Implicits.global indicates that this ExecutionContext is backed by a thread pool with a thread count that is equal to the available processor count. Our dive into ExecutionContext shows us how the simple Future example runs. We can illustrate how a Future applies its ExecutionContext to execute on multiple threads:

  Future(1).map(i => { 
    println(Thread.currentThread().getName) 
    i + 1 
  }).filter(i => { 
    println(Thread.currentThread().getName) 
    i % 2 == 0 
  }).foreach(println) 

We extend the original snippet to print the name of the thread performing each transformation. When run on a machine with multiple cores, this snippet yields variable output, depending on which threads pick up the transformations. Here is an example output:

ForkJoinPool-1-worker-3 
ForkJoinPool-1-worker-5 
2 

This example shows that one worker-3 thread performed the map transformation while another worker-5 thread performed the filter transformation. There are two key insights to draw from our simple example about how Future affects control flow. First, Future is a data type for concurrency that enables us to break the control flow of a program into multiple logical threads of processing. Second, our example shows that Future begins execution immediately upon creation. This means that transformations are applied immediately in a different flow of the program. We can use these insights to improve the runtime performance of Dave's crazy ideas.

We apply Future to Dave's set of backtests to improve performance. We believe there is an opportunity for a performance improvement because Dave's laptop has four CPU cores. This means that by adding concurrency to our program, we will be able to benefit from runtime parallelism. Our first attempt utilizes a for-comprehension:

implicit val ec = scala.concurrent.ExecutionContext.Implicits.global 
for { 
      firstDaveAapl <- Future(backtest(Dave1, Ticker("AAPL"), lastMonths(3))) 
      firstDaveGoog <- Future(backtest(Dave1, Ticker("GOOG"), lastMonths(3))) 
      secondDaveAapl <- Future(backtest(Dave2, Ticker("AAPL"), lastMonths(3))) 
      secondDaveGoog <- Future(backtest(Dave2, Ticker("GOOG"), lastMonths(3))) 
    } yield (firstDaveAapl, firstDaveGoog, secondDaveAapl, secondDaveGoog) 

Each backtest invocation is wrapped with the creation of a Future instance by calling Future.apply. This companion object method uses a by-name parameter to defer evaluation of the argument, which, in this case, is the invocation of backtest:

def apply[T](body: =>T)(executor: ExecutionContext): Future[T] 

After running the new version of CrazyIdeas.scala, you are disappointed to see the runtime execution has not improved. You quickly double-check the number of CPUs on your Linux box, as follows:

$ cat /proc/cpuinfo  | grep processor | wc -l 
8 

Having confirmed there are eight cores available on your laptop, you wonder why the execution time matches the original serial execution time. The solution here is to consider how the for-comprehension is compiled. The for-comprehension is equivalent to the following simpler example:

Future(1).flatMap(f1 => Future(2).flatMap(f2 => Future(3).map(f3 => (f1, f2, f3)))) 

In this desugared representation of the for-comprehension, we see that the second Future is created and evaluated within the flatMap transformation of the first Future. Any transformation applied to a Future (for example, flatMap) is only invoked once the value provided to the transform has been computed. This means that the Future in the preceding example and the for-comprehension are executed sequentially. To achieve the concurrency that we are looking for, we must instead modify CrazyIdeas.scala to look like the following:

    val firstDaveAaplF = Future(backtest(Dave1, Ticker("AAPL"), 
      lastMonths(3))) 
    val firstDaveGoogF = Future(backtest(Dave1, Ticker("GOOG"), 
      lastMonths(3))) 
    val secondDaveAaplF = Future(backtest(Dave2, Ticker("AAPL"), 
      lastMonths(3))) 
    val secondDaveGoogF = Future(backtest(Dave2, Ticker("GOOG"), 
      lastMonths(3))) 
    for { 
      firstDaveAapl <- firstDaveAaplF 
      firstDaveGoog <- firstDaveGoogF 
      secondDaveAapl <- secondDaveAaplF 
      secondDaveGoog <- secondDaveGoogF 
    } yield (firstDaveAapl, firstDaveGoog, secondDaveAapl, secondDaveGoog) 

In this snippet, four backtests are kicked off concurrently and the results are transformed into a Future of a Tuple4 consisting of four BacktestPerformanceSummary values. Seeing is believing, and after showing Dave the faster runtime of his backtests, he is excited to iterate quickly on new backtest ideas. Dave never misses a chance to throw around a pun, exclaiming, "Using all my cores is making my laptop fans really whiz. Not sure I'm a fan of the noise, but I sure do like the performance!"

In the previous example, we illustrated the ease of use of the Future API by investigating how to introduce concurrency to the backtester. Like any powerful tool, your usage of Future must be disciplined to ensure correctness and performance. This section evaluates topics that commonly cause confusion and error when using the Future to add concurrency to your program. We will detail performing side-effects, blocking execution, handling failures, choosing an appropriate execution context, and performance considerations.

scala> import scala.concurrent.Future 
import scala.concurrent.Future 
 
scala> import scala.concurrent.ExecutionContext.Implicits.global 
import scala.concurrent.ExecutionContext.Implicits.global 
 
scala> val f = Future{ println("FOO"); 40 + 2} 
FOO 
f: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@5575e0df 
 
scala> f.value 
res3: Option[scala.util.Try[Int]] = Some(Success(42)) 

def ready[T](awaitable: Awaitable[T], atMost: Duration): awaitable.type 
def result[T](awaitable: Awaitable[T], atMost: Duration): T 

    val summariesF = for { 
      firstDaveAapl <- Future(backtest(Dave1, Ticker("AAPL"), lastMonths(3))) 
      firstDaveGoog <- Future(backtest(Dave1, Ticker("GOOG"), lastMonths(3))) 
      secondDaveAapl <- Future(backtest(Dave2, Ticker("AAPL"), lastMonths(3))) 
      secondDaveGoog <- Future(backtest(Dave2, Ticker("GOOG"), lastMonths(3))) 
    } yield (firstDaveAapl, firstDaveGoog, secondDaveAapl, secondDaveGoog) 
 
    Await.result(summariesF, scala.concurrent.duration.Duration(1, java.util.concurrent.TimeUnit.SECONDS)) 

object SafeAwait { 
  def result[T]( 
    awaitable: Awaitable[T], 
    atMost: Duration): Option[T] =

    Try(Await.result(awaitable, atMost)) match { 
      case Success(t) => Some(t) 
      case Failure(_: TimeoutException) => None 
      case Failure(e) => throw e 
    } 
}

Future("not-an-integer").map(_.toInt).map(i => { 
      println("Multiplying") 
      i * 2 
    }) 

def recover[U >: T](pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Future[U] 
def recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]])(implicit executor: ExecutionContext): Future[U] 

Future("not-an-integer").map(_.toInt).recover { 
  case _: NumberFormatException => -2 
}.map(i => { 
  println("Multiplying") 
  i * 2 
}) 

Multiplying 
Multiplication result = -4 

  def submitOrder( 
    ec: ExecutionContext, 
    sendToExchange: ValidatedOrder => Future[OrderSubmitted], 
    updatePositions: OrderSubmitted => Future[AccountPositions], 
    o: RawOrder): Unit = { 
    implicit val iec = ec 
 
    (for { 
      vo <- ValidatedOrder.fromRawOrder(o).fold(
        Future.failed[ValidatedOrder](new Exception(
        "Order failed validation")))(Future.successful) 
      os <- sendToExchange(vo) 
      ap <- updatePositions(os) 
    } yield (os, ap)).onComplete { 
      case Success((os, ap)) => // Marshal order submission info to caller 
      case Failure(e) =>  // Marshal appropriate error response to caller 
    } 
  } 

   (for { 
      vo <- ValidatedOrder.fromRawOrder(o).fold( 
        Future.failed[ValidatedOrder]( 
        new Exception("Order failed validation")))(Future.successful) 
      os <- { 
        val f = sendToExchange(vo) 
        f.onFailure({ case e => incrementExchangeErrorCount() }) 
        f 
      } 
      ap <- updatePositions(os) 
    } yield (os, ap)) 

scalaz./.right[Throwable, Int](1).map(_ * 2) 

As Future provides an expressive and easy-to-use API, it is common to perform numerous transforms to complete a computation in a large-scale system. Reflecting on the order submission web service mentioned in the previous section, you can imagine multiple application layers operating on a Future. A production-ready web service typically composes together multiple layers to service a single request. An example request flow may contain the following stages: request deserialization, authorization, application service invocation, database lookups and/or third-party service callouts, and response translation to a JSON format. If each of these stages in the workflow is modeled with a Future, then it is common to have five or more transformations to handle a single request.

Decomposing your software system into small areas of responsibility in a way that is similar to the preceding example is a good engineering practice to support testing in isolation and improving maintainability. However, this approach to software design comes with a performance cost when working with Future. As we have seen through our example usage, nearly all transforms on a Future require submitting work to an Executor. In our example workflow, most stages in the transformation are small. In this scenario, the overhead of submitting work to the executor dominates the execution time of the computation. If the order submission web service services numerous customers with stringent throughput and latency requirements, then it is possible that engineering practices focusing on testability and maintainability will result in poorly performing software.

If you consider the preceding diagram, you can see a thread pool with four threads being used to apply transforms to a Future. Each transform is submitted to the pool and there is a chance that a different thread is picked for the computation. This diagram visualizes how multiple small transforms may hamper performance due to the overhead of Executor submissions.

Just how large is the overhead of Executor submissions? This is the motivating question to write a benchmark to quantify the overhead of submitting work to an Executor. The benchmark focuses on adding 1 to an integer N-times in two ways. One approach is to perform the addition operation within a single Future, while the second approach is to perform each addition operation with a new Future transformation. The latter approach is a proxy for the stages of order submission request processing that uses multiple Future transformations in a larger software system. Performing integer addition is the proxy operation because it is an extremely cheap computation, which means that the execution time will be dominated by Executor submissions. The benchmarks look like the following:

  @Benchmark 
  def manyTransforms(state: TransformFutureState): Int = { 
    import scala.concurrent.ExecutionContext.Implicits._ 
    val init = Future(0) 
    val res = (1 until state.operations).foldLeft(init)
      ((f, _) => f.map(_ + 1)) 
    Await.result(res, Duration("5 minutes")) 
  } 
 
  @Benchmark 
  def oneTransform(state: TransformFutureState): Int = { 
    import scala.concurrent.ExecutionContext.Implicits._ 
    val res = Future { 
      (1 until state.operations).foldLeft(0)((acc, _) => acc + 1) 
    } 
    Await.result(res, Duration("5 minutes")) 
  } 

TransformFutureState allows the number of operations to be parameterized. manyTransforms represents each addition operation using a map transformation that involves submitting work to an Executor. oneTransform performs all addition operations using a single Executor submission via Future.apply. In this controlled test, Await.result is used as a blocking mechanism to await the completion of the computation. The results of running this test on a two-core machine with five transformations and ten transformations can be seen in the following table:

While both scenarios yield comparable results with five transformations, we can see a clear difference with ten transforms being applied. This benchmark makes it clear that Executor submissions can dominate performance. Although the cost can be high, our advice to you is to model your system without considering this cost up-front. In our experience, it is easier to rework a well-modeled system for performance improvements than it is to extend or to rework a poorly-modeled but performant system. For this reason, we advise against going to great lengths to group Executor submissions when attempting to put together the initial version of a complex system.

Once you have a good design in place, the first step is to benchmark and to profile in order to identify whether Executor submissions are the bottleneck. In the event that you discover that your style of Future usage is causing performance bottlenecks, there are several courses of action you should consider.

The lowest cost development option is to replace unnecessarily costly Future creation with the use of Future.success or Future.failure. The order submission web service took advantage of these factory methods to lift values into a Future. As the value is already computed, these factory methods avoid submitting any tasks to the Executor that are referenced by the provided ExecutionContext. Replacing usages of Future.apply with either Future.successful or Future.failure when the value is already computed can yield cost savings.

A more expensive alternative in terms of development effort is to rework your implementation to group together Future transformations in a way similar to manyTransforms. This tactic involves reviewing each application layer to determine whether transforms within a single layer can be combined. If possible, we recommend that you avoid merging transformations across application layers (for example, between request deserialization or authorization and application service processing) because this weakens your model and increases maintenance cost.

If neither of these options produces acceptable performance, then it may be worthwhile to discuss with the product owners the option of addressing the performance issue with hardware. As your system's design has not been compromised and it reflects solid engineering practices, then it likely can be horizontally scaled or clustered. Depending on the state tracked by your system, this option might be possible without additional development work. Perhaps product owners value a system that can be easily maintained and extended more than performance. If this is the case, adding scale to your system may be a viable way forward.

Provided that you are unable to buy your way out of the performance challenge, then there are three additional possibilities. One option is to investigate an alternative to Future, named Task. This construct, which is provided by the Scalaz library, allows computations to be performed with fewer Executor submissions. This option involves significant development because the Future data type will need to be replaced throughout the application with Task. We will explore Task at the end of this chapter and investigate the performance benefits that it can provide.

Independent of using Task, it can be useful to review your application's model to critically question whether or not there is unnecessary work being done on the critical path. As we saw with MVT's reporting infrastructure and the introduction of stream processing, it is sometimes possible to rethink a design to improve performance. Like the introduction of Task, reconsidering your system's architecture is a large-scale change. The last resort option is to merge application layers in order to support grouping Future transformations. We advise against exercising this option, unless all other suggestions have failed. This option results in a code base that is more difficult to reason about because concerns are no longer separated. In the short-run, you may reap performance benefits, but in our experience, these benefits are outweighed in the long-run by the cost of maintaining and extending such a system.