Scala Futures – built in timeout?

by

You only get timeout behavior when you use blocking to get the results of the Future. If you want to use the non-blocking callbacks onComplete, onSuccess or onFailure, then you would have to roll your own timeout handling. Akka has built in timeout handling for request/response (?) messaging between actors, but not sure if you want to start using Akka. FWIW, in Akka, for timeout handling, they compose two Futures together via Future.firstCompletedOf, one which represents the actual async task and one that represents the timeout. If the timeout timer (via a HashedWheelTimer) pops first, you get a failure on the async callback.

A very simplified example of rolling your own might go something like this. First, an object for scheduling timeouts:

import org.jboss.netty.util.{HashedWheelTimer, TimerTask, Timeout}
import java.util.concurrent.TimeUnit
import scala.concurrent.duration.Duration
import scala.concurrent.Promise
import java.util.concurrent.TimeoutException

object TimeoutScheduler{
  val timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS)
  def scheduleTimeout(promise:Promise[_], after:Duration) = {
    timer.newTimeout(new TimerTask{
      def run(timeout:Timeout){              
        promise.failure(new TimeoutException("Operation timed out after " + after.toMillis + " millis"))        
      }
    }, after.toNanos, TimeUnit.NANOSECONDS)
  }
}

Then a function to take a Future and add timeout behavior to it:

import scala.concurrent.{Future, ExecutionContext, Promise}
import scala.concurrent.duration.Duration

def withTimeout[T](fut:Future[T])(implicit ec:ExecutionContext, after:Duration) = {
  val prom = Promise[T]()
  val timeout = TimeoutScheduler.scheduleTimeout(prom, after)
  val combinedFut = Future.firstCompletedOf(List(fut, prom.future))
  fut onComplete{case result => timeout.cancel()}
  combinedFut
}

Note that the HashedWheelTimer I am using here is from Netty.

More Related Contents:

  1. How to wait for several Futures?
  2. Can Scala actors process multiple messages simultaneously?
  3. Futures do not run before program termination
  4. Futures / Success race
  5. Apache Spark – Scala – ReduceByKey – with keys repeating up to twice only
  6. What is the difference between a var and val definition in Scala?
  7. What is the apply function in Scala?
  8. How do I apply the enrich-my-library pattern to Scala collections?
  9. How to import build.gradle into IntelliJ
  10. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  11. Scala spark, listbuffer is empty
  12. scala slick method I can not understand so far
  13. Difference between this and self in self-type annotations?
  14. Scala constructor overload?
  15. Instantiating immutable paired objects
  16. Change nullable property of column in spark dataframe
  17. What type to use to store an in-memory mutable data table in Scala?
  18. Usefulness (as in practical applications) of Currying v.s. Partial Application in Scala
  19. Left Anti join in Spark?
  20. How to Define Custom partitioner for Spark RDDs of equally sized partition where each partition has equal number of elements?
  21. How does type Dynamic work and how to use it?
  22. Get a TypeTag from a Type?
  23. How to write a proper null-safe coalescing operator in scala?
  24. Can’t access Parent’s Members while dealing with Macro Annotations
  25. Why is join not possible after show operator?
  26. How to read a file as a byte array in Scala
  27. Is there a simple example of how to generate verilog from Chisel3 module?
  28. Run a scala code jar appear NoSuchMethodError:scala.Predef$.refArrayOps
  29. Spark UDF with varargs
  30. How to transpose an RDD in Spark

Leave a Comment