Eseguire più futures in parallelo, restituire il valore predefinito sul timeout

Question

Devo eseguire più future in parallelo e il programma non dovrebbe bloccarsi o bloccarsi.

Per ora aspetto i futures uno per uno e utilizzo il valore di fallback se c'è TimeoutException.

val future1 = // start future1 
val future2 = // start future2 
val future3 = // start future3 

// <- at this point all 3 futures are running 

// waits for maximum of timeout1 seconds 
val res1 = toFallback(future1, timeout1, Map[String, Int]()) 
// .. timeout2 seconds 
val res2 = toFallback(future2, timeout2, List[Int]()) 
// ... timeout3 seconds 
val res3 = toFallback(future3, timeout3, Map[String, BigInt]()) 

def toFallback[T](f: Future[T], to: Int, default: T) = { 
    Try(Await.result(f, to seconds)) 
    .recover { case to: TimeoutException => default } 
} 

Come posso vedere, il tempo massimo di attesa di questo frammento è timeout1 + timeout2 + timeout3

La mia domanda è: come posso aspettare tutte quelle future in una sola volta, in modo da poter ridurre il tempo di attesa per max(timeout1, timeout2, timeout3)?

EDIT: Alla fine ho usato modifica @Jatin e @senia risposte:

private def composeWaitingFuture[T](fut: Future[T], 
            timeout: Int, default: T) = 
    future { Await.result(fut, timeout seconds) } recover { 
    case e: Exception => default 
    } 

e in seguito viene utilizzato come segue:

// starts futures immediately and waits for maximum of timeoutX seconds 
val res1 = composeWaitingFuture(future1, timeout1, Map[String, Int]()) 
val res2 = composeWaitingFuture(future2, timeout2, List[Int]()) 
val res3 = composeWaitingFuture(future3, timeout3, Map[String, BigInt]()) 

// takes the maximum of max(timeout1, timeout2, timeout3) to complete 
val combinedFuture = 
    for { 
    r1 <- res1 
    r2 <- res2 
    r3 <- res3 
    } yield (r1, r2, r3) 

e poi io uso combinedFuture come vedo in forma .

Answer 1

def toFallback[T](f: Future[T], to: Int, default: T) = { 
    future{ 
    try{ 
     Await.result(f, to seconds) 
    }catch{ 
     case e:TimeoutException => default 
    } 
} 

È anche possibile rendere questo blocco asincrono e ogni richiesta attende il suo tempo massimo. Se ci sono troppi thread, probabilmente hanno un singolo thread che continua a controllare altri futures usando il codice system scheduler di Akka. @Senia ha risposto qui sotto su questo.

Answer 2

Si potrebbe creare future che restituisce i risultati di tutti e 3 i futures utilizzando flatMap o-di comprensione:

val combinedFuture = 
    for { 
    r1 <- future1 
    r2 <- future2 
    r3 <- future3 
    } yield (r1, r2, r3) 

val (r1, r2, r3) = Await.result(combinedFuture , Seq(timeout1, timeout2, timeout3).max) 

Se si utilizza akka si potrebbe completare il vostro futuro con valore predefinito dopo timeout:

implicit class FutureHelper[T](f: Future[T]) extends AnyVal{ 
    import akka.pattern.after 
    def orDefault(t: Timeout, default: => T)(implicit system: ActorSystem): Future[T] = { 
    val delayed = after(t.duration, system.scheduler)(Future.successful(default)) 
    Future firstCompletedOf Seq(f, delayed) 
    } 
} 

val combinedFuture = 
    for { 
    r1 <- future1.orDefault(timeout1, Map()) 
    r2 <- future2.orDefault(timeout2, List()) 
    r3 <- future3.orDefault(timeout3, Map()) 
    } yield (r1, r2, r3) 

val (r1, r2, r3) = Await.result(combinedFuture , allowance + Seq(timeout1, timeout2, timeout3).max) 

Answer 3

Eviterei di utilizzare Await.result poiché utilizza un thread solo per il blocco. Una possibilità per implementare timeout per i futures sarebbe questo:

val timer = new Timer() 

def toFallback[T](f: Future[T], timeout: Int, default: T) = { 
    val p = Promise[T]() 
    f.onComplete(result => p.tryComplete(result)) 
    timer.schedule(new TimerTask { 
    def run() { 
     p.tryComplete(Success(default)) 
    } 
    }, timeout) 
    p.future 
} 

Questo crea una promessa che sarà completato o da un futuro o da un risultato di default dopo il timeout specificato - si verifica per primo.

Per eseguire le query contemporaneamente si dovrebbe fare in questo modo:

val future1 = // start future1 
val future2 = // start future2 
val future3 = // start future3 

val res1 = toFallback(future1, timeout1, Map[String, Int]()) 
val res2 = toFallback(future2, timeout2, List[Int]()) 
val res3 = toFallback(future3, timeout3, Map[String, BigInt]()) 

val resultF = for { 
    r1 <- res1 
    r2 <- res2 
    r3 <- res3 
} yield (r1, r2, r3) 

val (r1, r2, r3) = Await.result(resultF, Duration.Inf) 
println(s"$r1, $r2, $r3") 

//or 
resultF.onSuccess { 
    case (r1, r2, r3) => println(s"$r1, $r2, $r3") 
} 

Answer 4

perché non ottenere il Futuresé per eseguire la cattura e il ritorno del difetto eccezione? Quindi puoi semplicemente Await su ciascun futuro a turno, e non devi preoccuparti della gestione delle eccezioni al di fuori del futuro.

Answer 5

Questo è forse un po 'hacky, ma è sufficiente misurare il tempo trascorso e modificare i timeout di conseguenza. Assumendo timeout1 <= timeout2 <= timeout3:

def now  = System.currentTimeMillis(); 
val start = now; 
def remains(timeout: Long): Long 
      = math.max(0, timeout + start - now) 

def toFallback[T](f: Future[T], to: Int, default: T) = { 
    Try(Await.result(f, remains(to) seconds)) 
    .recover { case to: TimeoutException => default } 
} 

In questo modo ogni timeout si basa al momento start = now è stato chiamato, in modo che il tempo complessivo di esecuzione è al massimo timeout3. Se i timeout non sono ordinati, funziona ancora, ma alcune attività possono essere lasciate in esecuzione più a lungo del loro timeout designato.

Answer 6

Ecco una risposta più lunga (unakka) che affronta quello che potrebbe essere il caso d'uso, vale a dire, se uno dei valori "scade" si desidera utilizzare il valore predefinito per quel risultato e anche fare qualcosa con esso (come annullare il calcolo a lungo termine o I/O o qualsiasi altra cosa).

Inutile dire che l'altra storia è di minimizzare il blocco.

L'idea di base è quella di sedersi ad anello in attesa degli articoli firstCompletedOf non ancora completati. Il timeout su ready è il timeout minimo rimanente.

Questo codice utilizza scadenze anziché durate, ma l'utilizzo di una durata come "tempo rimanente" è semplice.

import scala.language.postfixOps 
import scala.concurrent._ 
import scala.concurrent.duration._ 
import ExecutionContext.Implicits._ 
import scala.reflect._ 
import scala.util._ 
import java.lang.System.{ nanoTime => now } 

import Test.time 

class Test { 

    type WorkUnit[A] = (Promise[A], Future[A], Deadline, A) 
    type WorkQ[A] = Seq[WorkUnit[A]] 

    def await[A: ClassTag](work: Seq[(Future[A], Deadline, A)]): Seq[A] = { 
    // check for timeout; if using Duration instead of Deadline, decrement here 
    def ticktock(w: WorkUnit[A]): WorkUnit[A] = w match { 
     case (p, f, t, v) if !p.isCompleted && t.isOverdue => p trySuccess v ; w 
     case _ => w 
    } 
    def await0(work: WorkQ[A]): WorkQ[A] = { 
     val live = work filterNot (_._1.isCompleted) 
     val t0 = (live map (_._3)).min 
     Console println s"Next deadline in ${t0.timeLeft.toMillis}" 
     val f0 = Future firstCompletedOf (live map (_._2)) 
     Try(Await ready (f0, t0.timeLeft)) 
     val next = work map (w => ticktock(w)) 
     if (next exists (!_._1.isCompleted)) { 
     await0(next) 
     } else { 
     next 
     } 
    } 
    val wq = work map (_ match { 
     case (f, t, v) => 
     val p = Promise[A] 
     p.future onComplete (x => Console println s"Value available: $x: $time") 
     f onSuccess { 
      case a: A => p trySuccess a // doesn't match on primitive A 
      case x => p trySuccess x.asInstanceOf[A] 
     } 
     f onFailure { case _ => p trySuccess v } 
     (p, f, t, v) 
    }) 
    await0(wq) map (_ match { 
     case (p, f, t, v) => p.future.value.get.get 
    }) 
    } 
} 

object Test { 
    val start = now 
    def time = s"The time is ${ Duration fromNanos (now - start) toMillis }" 

    def main(args: Array[String]): Unit = { 
    // #2 times out 
    def calc(i: Int) = { 
     val t = if (args.nonEmpty && i == 2) 6 else i 
     Thread sleep t * 1000L 
     Console println s"Calculate $i: $time" 
     i 
    } 
    // futures to be completed by a timeout deadline 
    // or else use default and let other work happen 
    val work = List(
     (future(calc(1)), 3 seconds fromNow, 10), 
     (future(calc(2)), 5 seconds fromNow, 20), 
     (future(calc(3)), 7 seconds fromNow, 30) 
    ) 
    Console println new Test().await(work) 
    } 
} 

esempio di esecuzione:

apm@mara:~/tmp$ skalac nextcompleted.scala ; skala nextcompleted.Test 
Next deadline in 2992 
Calculate 1: The time is 1009 
Value available: Success(1): The time is 1012 
Next deadline in 4005 
Calculate 2: The time is 2019 
Value available: Success(2): The time is 2020 
Next deadline in 4999 
Calculate 3: The time is 3020 
Value available: Success(3): The time is 3020 
List(1, 2, 3) 
apm@mara:~/tmp$ skala nextcompleted.Test arg 
Next deadline in 2992 
Calculate 1: The time is 1009 
Value available: Success(1): The time is 1012 
Next deadline in 4005 
Calculate 3: The time is 3020 
Value available: Success(3): The time is 3020 
Next deadline in 1998 
Value available: Success(20): The time is 5020 
List(1, 20, 3)