Scala Futures - built in timeout?

concurrency, scala

Solution

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.

Problem

there is an aspect of futures that I do not exactly understand from the official tutorial ref. http://docs.scala-lang.org/overviews/core/futures.html Do futures in scala have a built in time-out mechanism of some kind? Let's say the example below was a 5 gigabyte text file... does the implied scope of "Implicits.global" eventually cause onFailure to fire in a non-blocking way or can that be defined? And without a default time-out of some kind, wouldn't that imply it's possible neither success nor failure would ever fire? ``` import scala.concurrent._ import ExecutionContext.Implicits.global val firstOccurence: Future[Int] = future { val source = scala.io.Source.fromFile("myText.txt") source.toSeq.indexOfSlice("myKeyword") } firstOccurence onSuccess { case idx => println("The keyword first appears at position: " + idx) } firstOccurence onFailure { case t => println("Could not process file: " + t.getMessage) } ```

Original source

Related problems