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) }
The solution to this problem is to use akka after pattern which is a non-blocking approach to provide timeout to future. According to Akka documentation, after pattern will return a future with success or failure according to the value specified after the duration specified in after pattern.
Future represents a result of an asynchronous computation that may or may not be available yet. When we create a new Future, Scala spawns a new thread and executes its code. Once the execution is finished, the result of the computation (value or exception) will be assigned to the Future.
Futures can be considered monads if you never construct them with effectful blocks (pure, in-memory computation), or if any effects generated are not considered as part of semantic equivalence (like logging messages).
Handle the result of a future with methods like onComplete , or combinator methods like map , flatMap , filter , andThen , etc. The value in a Future is always an instance of one of the Try types: Success or Failure.
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.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With