Scala Futures - délai d'attente intégré ?

Question

Il y a un aspect des futures que je ne comprends pas exactement à partir du tutoriel officiel réf. http://docs.scala-lang.org/overviews/core/futures.html

Les contrats à terme en Scala ont-ils un mécanisme de temporisation intégré ? Disons que l'exemple ci-dessous était un fichier texte de 5 gigaoctets... est-ce que la portée implicite de "Implicits.global" fait que onFailure se déclenche de manière non bloquante ou est-ce que cela peut être défini ? Et sans une sorte de délai d'attente par défaut, cela ne signifierait-il pas qu'il est possible que ni le succès ni l'échec ne se déclenchent jamais ?

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)
}

Answer 1

Vous n'obtenez un comportement de dépassement de délai que si vous utilisez le blocage pour obtenir les résultats de la méthode Future . Si vous voulez utiliser les callbacks non bloquants onComplete , onSuccess o onFailure vous devrez alors mettre en place votre propre gestion des délais d'attente. Akka a intégré la gestion des délais d'attente pour les requêtes/réponses ( ? ) entre acteurs, mais vous n'êtes pas certain de vouloir commencer à utiliser Akka. FWIW, dans Akka, pour la gestion du timeout, ils composent deux Futures ensemble via Future.firstCompletedOf L'un représente la tâche asynchrone actuelle et l'autre le délai d'attente. Si le délai d'attente (via un HashedWheelTimer ) pops d'abord, vous obtenez un échec sur le callback asynchrone.

Un exemple très simplifié de l'utilisation d'un rouleau de papier pourrait ressembler à ceci. Tout d'abord, un objet pour planifier les délais d'attente :

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)
  }
}

Puis une fonction pour prendre un futur et lui ajouter un comportement de temporisation :

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
}

Notez que le HashedWheelTimer J'utilise ici est de Netty.

Answer 2

Toutes ces réponses nécessitent des dépendances supplémentaires. J'ai décidé d'écrire une version utilisant java.util.Timer qui est un moyen efficace d'exécuter une fonction dans le futur, dans ce cas pour déclencher un timeout.

Blog post avec plus de détails ici

En utilisant ceci avec Promise de Scala, nous pouvons créer un Future avec timeout comme suit :

package justinhj.concurrency

import java.util.concurrent.TimeoutException
import java.util.{Timer, TimerTask}

import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.language.postfixOps

object FutureUtil {

  // All Future's that use futureWithTimeout will use the same Timer object
  // it is thread safe and scales to thousands of active timers
  // The true parameter ensures that timeout timers are daemon threads and do not stop
  // the program from shutting down

  val timer: Timer = new Timer(true)

  /**
    * Returns the result of the provided future within the given time or a timeout exception, whichever is first
    * This uses Java Timer which runs a single thread to handle all futureWithTimeouts and does not block like a
    * Thread.sleep would
    * @param future Caller passes a future to execute
    * @param timeout Time before we return a Timeout exception instead of future's outcome
    * @return Future[T]
    */
  def futureWithTimeout[T](future : Future[T], timeout : FiniteDuration)(implicit ec: ExecutionContext): Future[T] = {

    // Promise will be fulfilled with either the callers Future or the timer task if it times out
    val p = Promise[T]

    // and a Timer task to handle timing out

    val timerTask = new TimerTask() {
      def run() : Unit = {
            p.tryFailure(new TimeoutException())
        }
      }

    // Set the timeout to check in the future
    timer.schedule(timerTask, timeout.toMillis)

    future.map {
      a =>
        if(p.trySuccess(a)) {
          timerTask.cancel()
        }
    }
    .recover {
      case e: Exception =>
        if(p.tryFailure(e)) {
          timerTask.cancel()
        }
    }

    p.future
  }

}

Answer 3

Je viens de créer un TimeoutFuture pour un collègue de travail :

TimeoutFuture

package model

import scala.concurrent._
import scala.concurrent.duration._
import play.libs.Akka
import play.api.libs.concurrent.Execution.Implicits._

object TimeoutFuture {
  def apply[A](timeout: FiniteDuration)(block: => A): Future[A] = {

    val prom = promise[A]

    // timeout logic
    Akka.system.scheduler.scheduleOnce(timeout) {
      prom tryFailure new java.util.concurrent.TimeoutException
    }

    // business logic
    Future { 
      prom success block
    }

    prom.future
  } 
}

Utilisation

val future = TimeoutFuture(10 seconds) { 
  // do stuff here
}

future onComplete {
  case Success(stuff) => // use "stuff"
  case Failure(exception) => // catch exception (either TimeoutException or an exception inside the given block)
}

Notes :

• Suppose le cadre Play ! (mais il est assez facile de l'adapter).
• Chaque morceau de code s'exécute dans le même ExecutionContext ce qui n'est pas forcément idéal.

Answer 4

Le framework Play contient Promise.timeout, vous pouvez donc écrire un code comme le suivant

private def get(): Future[Option[Boolean]] = {
  val timeoutFuture = Promise.timeout(None, Duration("1s"))
  val mayBeHaveData = Future{
    // do something
    Some(true)
  }

  // if timeout occurred then None will be result of method
  Future.firstCompletedOf(List(mayBeHaveData, timeoutFuture))
}

Answer 5

Je suis assez surpris que cela ne soit pas standard en Scala. Mes versions sont courtes et n'ont pas de dépendances

import scala.concurrent.Future

sealed class TimeoutException extends RuntimeException

object FutureTimeout {

  import scala.concurrent.ExecutionContext.Implicits.global

  implicit class FutureTimeoutLike[T](f: Future[T]) {
    def withTimeout(ms: Long): Future[T] = Future.firstCompletedOf(List(f, Future {
      Thread.sleep(ms)
      throw new TimeoutException
    }))

    lazy val withTimeout: Future[T] = withTimeout(2000) // default 2s timeout
  }

}

Exemple d'utilisation

import FutureTimeout._
Future { /* do smth */ } withTimeout