Futures in Scala

Asynchronous execution

Consider the code given below in scala,which sleeps for 10 seconds and returns the result. Which is 21+21.

Thread.sleep(10000); 21 + 21

What if the code takes 10 minutes sleep instead of 10 seconds. Can you even imagine an API which returns a result after 10 minutes!!. Of course not, we expect the response of an API within seconds, this is the scenario where asynchronous execution comes into play. Java provides multi-threading support for executing threads in the background ,but this mechanism is managed by monitors which provides locks and enables only one thread to enter at time. Python provides celery, which is a completely different process executing tasks in the background. These process can be used to execute different tasks asynchronously in the background.

These tasks execution in background are minefields of deadlocks and race conditions. A thousand things could go wrong, now let us see how we can use scala futures to protect from these deadlocks and race conditions.

Scala futures

When you invoke a function the thread waits while function finishing execution and returns the result, if the result is a future then the execution is carried out asynchronously by a completely different thread.

scala> import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.ExecutionContext.Implicits.global
scala> val fut = Future { Thread.sleep(10000); 21 + 21 }
fut: scala.concurrent.Future[Int] = Future(<not completed>)

The above function will continue to execute the method, often by a completely different thread. we can keep on polling to get the status of the future whether the execution is completed or not , whether the status is success or failure.

scala> fut.isCompleted
res0: Boolean = true
scala> fut.value
res3: Option[scala.util.Try[Int]] = Some(Success(42))

After 10 seconds the result of the future will be as shone above. Before completion the isCompleted result will be false and value will be None.

Transforming future with map

A map function can be used to map the result of a future with another. We can explicitly specify what to do with the result of a future.

scala> val fut = Future { Thread.sleep(10000); 21 + 21 }
fut: scala.concurrent.Future[Int] = ...
scala> val result = fut.map(x => x + 1)
result: scala.concurrent.Future[Int] = ...
//After 10 seconds
scala> result.value
res6: Option[scala.util.Try[Int]] = Some(Success(43))

Transforming futures with for expressions

We can run two futures parallelly and combine their results and create a new future.

scala> val fut1 = Future { Thread.sleep(10000); 21 + 21 }
fut1: scala.concurrent.Future[Int] = Future(<not completed>)
scala> val fut2 = Future { Thread.sleep(10000); 23 + 23 }
fut2: scala.concurrent.Future[Int] = Future(<not completed>)
scala> val rs2=for {
     | x <- fut1
     | y <- fut2
     | } yield x + y
rs2: scala.concurrent.Future[Int] = Future(<not completed>)
scala> rs2.value
res20: Option[scala.util.Try[Int]] = Some(Success(88))

If you don’t want to run futures parallely the specify it in the for loop. For example

scala> for {
x <- Future { Thread.sleep(10000); 21 + 21 }
y <- Future { Thread.sleep(10000); 23 + 23 }
} yield x + y
res9: scala.concurrent.Future[Int] = ...

successful and failed methods can be used to create a future that is already failed or already successful.

scala> Future.successful { 21 + 21 }
res21: scala.concurrent.Future[Int] = Future(Success(42))
scala> Future.failed(new Exception("hackkkkkkkkkkkkk!"))
res23: scala.concurrent.Future[Nothing] = Future(Failure(java.lang.Exception: hackkkkkkkkkkkkk!))

A future can be created and controlled using promises. Promises are most general way to create and control futures. A future will complete when you complete the promise. A promise can be completed as either success,failure or competed.

scala> import scala.concurrent._
import scala.concurrent._
scala> val pro=Promise[Int]
pro: scala.concurrent.Promise[Int] = Future(<not completed>)
scala> val fut = pro.future
fut: scala.concurrent.Future[Int] = Future(<not completed>)
scala> fut.value
res24: Option[scala.util.Try[Int]] = None
scala> pro.success(42)
res25: pro.type = Future(Success(42))

filter, collect and Transforming futures

filter and collect are two methods that are used to filter out from future and perform some operations on them.

The below example shows a filter operation on a future, filtering values greater than zero. If we apply a filter operation to filter values less than zero the result will be None.

scala> val fut = Future { 42 }
fut: scala.concurrent.Future[Int] = Future(<not completed>)
scala> val valid = fut.filter(res => res > 0)
valid: scala.concurrent.Future[Int] = Future(<not completed>)
scala> valid.value
res26: Option[scala.util.Try[Int]] = Some(Success(42))

collect method is used to validate the future’s output and perform some transformation operation on it.

scala> val fut = Future { 42 }
fut: scala.concurrent.Future[Int] = Future(Success(42))
scala> val valid=fut collect { case res if res > 0 => res + 46 }
valid: scala.concurrent.Future[Int] = Future(<not completed>)
scala> valid.value
res31: Option[scala.util.Try[Int]] = Some(Success(88))

transform operation can be applied on a future for a conditional execution to specify what to do if a future is successful and what to do in case of failure. In the below example it will go for first condition if the future is successful and second condition if the future is a failure.

scala>  val success = Future { 42 }
success: scala.concurrent.Future[Int] = Future(<not completed>)
scala> val first = success.transform(
| res => res * -1,
| ex => new Exception("see cause", ex)
| )
first: scala.concurrent.Future[Int] = Future(<not completed>)
scala> first.value
res34: Option[scala.util.Try[Int]] = Some(Success(-42))

Writing test cases for futures

We already saw how to create futures. Now let us see how to write test cases and successfully test our futures. Scala provides a method called await to block our threads for a certain amount of time. After that time we will get the result and can test it. In the below example the await takes a parameter of time delay and waits for that particular amount of time and returns the result. After obtaining the result we can assert the result and test the function.

scala> import scala.concurrent.Await
import scala.concurrent.Await
scala> import scala.concurrent.duration._
import scala.concurrent.duration._
scala> val fut = Future { Thread.sleep(10000); 21 + 21 }
fut: scala.concurrent.Future[Int] = Future(<not completed>)
scala> val x = Await.result(fut, 15.seconds)
x: Int = 42