How to wait for several Futures?

Question

Suppose I have several futures and need to wait until either any of them fails or all of them succeed.

For example: Let there are 3 futures: f1, f2, f3.

• If f1 succeeds and f2 fails I do not wait for f3 (and return failure to the client).

• If f2 fails while f1 and f3 are still running I do not wait for them (and return failure)

• If f1 succeeds and then f2 succeeds I continue waiting for f3.

How would you implement it?

Answer 1

You could use a for-comprehension as follows instead:

val fut1 = Future{...} val fut2 = Future{...} val fut3 = Future{...}  val aggFut = for{   f1Result <- fut1   f2Result <- fut2   f3Result <- fut3 } yield (f1Result, f2Result, f3Result) 

In this example, futures 1, 2 and 3 are kicked off in parallel. Then, in the for comprehension, we wait until the results 1 and then 2 and then 3 are available. If either 1 or 2 fails, we will not wait for 3 anymore. If all 3 succeed, then the aggFut val will hold a tuple with 3 slots, corresponding to the results of the 3 futures.

Now if you need the behavior where you want to stop waiting if say fut2 fails first, things get a little trickier. In the above example, you would have to wait for fut1 to complete before realizing fut2 failed. To solve that, you could try something like this:

  val fut1 = Future{Thread.sleep(3000);1}   val fut2 = Promise.failed(new RuntimeException("boo")).future   val fut3 = Future{Thread.sleep(1000);3}    def processFutures(futures:Map[Int,Future[Int]], values:List[Any], prom:Promise[List[Any]]):Future[List[Any]] = {     val fut = if (futures.size == 1) futures.head._2     else Future.firstCompletedOf(futures.values)      fut onComplete{       case Success(value) if (futures.size == 1)=>          prom.success(value :: values)        case Success(value) =>         processFutures(futures - value, value :: values, prom)        case Failure(ex) => prom.failure(ex)     }     prom.future   }    val aggFut = processFutures(Map(1 -> fut1, 2 -> fut2, 3 -> fut3), List(), Promise[List[Any]]())   aggFut onComplete{     case value => println(value)   } 

Now this works correctly, but the issue comes from knowing which Future to remove from the Map when one has been successfully completed. As long as you have some way to properly correlate a result with the Future that spawned that result, then something like this works. It just recursively keeps removing completed Futures from the Map and then calling Future.firstCompletedOf on the remaining Futures until there are none left, collecting the results along the way. It's not pretty, but if you really need the behavior you are talking about, then this, or something similar could work.

Answer 2

You can use a promise, and send to it either the first failure, or the final completed aggregated success:

def sequenceOrBailOut[A, M[_] <: TraversableOnce[_]](in: M[Future[A]] with TraversableOnce[Future[A]])(implicit cbf: CanBuildFrom[M[Future[A]], A, M[A]], executor: ExecutionContext): Future[M[A]] = {   val p = Promise[M[A]]()    // the first Future to fail completes the promise   in.foreach(_.onFailure{case i => p.tryFailure(i)})    // if the whole sequence succeeds (i.e. no failures)   // then the promise is completed with the aggregated success   Future.sequence(in).foreach(p trySuccess _)    p.future } 

Then you can Await on that resulting Future if you want to block, or just map it into something else.

The difference with for comprehension is that here you get the error of the first to fail, whereas with for comprehension you get the first error in traversal order of the input collection (even if another one failed first). For example:

val f1 = Future { Thread.sleep(1000) ; 5 / 0 } val f2 = Future { 5 } val f3 = Future { None.get }  Future.sequence(List(f1,f2,f3)).onFailure{case i => println(i)} // this waits one second, then prints "java.lang.ArithmeticException: / by zero" // the first to fail in traversal order 

And:

val f1 = Future { Thread.sleep(1000) ; 5 / 0 } val f2 = Future { 5 } val f3 = Future { None.get }  sequenceOrBailOut(List(f1,f2,f3)).onFailure{case i => println(i)} // this immediately prints "java.util.NoSuchElementException: None.get" // the 'actual' first to fail (usually...) // and it returns early (it does not wait 1 sec)