Informazioni sul meccanismo Future.firstCompletedOf e Garbage Collect

Question

Ho riscontrato questo problema nel mio progetto di vita reale e dimostrato dal codice di test e dal profiler. Invece di incollare il codice "tl; dr", ti sto mostrando una foto e poi la descrivo.

In poche parole, sto usando Future.firstCompletedOf per ottenere un risultato da 2 Future s, entrambi i quali non hanno cose in comune e non si preoccupano gli uni degli altri. Anche se, quale è la domanda che voglio risolvere, il Garbage Collector non può riciclare il primo oggetto Result fino a quando entrambi i Future sono finiti.

Quindi sono davvero curioso del meccanismo alla base di questo. Qualcuno potrebbe spiegarlo da un livello più basso, o fornire qualche suggerimento per me da esaminare.

Grazie!

PS: è perché condividono lo stesso ExecutionContext?

** Aggiornamento ** codice di prova incolla come richiesto

object Main extends App{ 
    println("Test start") 

    val timeout = 30000 

    trait Result { 
    val id: Int 
    val str = "I'm short" 
    } 
    class BigObject(val id: Int) extends Result{ 
    override val str = "really big str" 
    } 

    def guardian = Future({ 
    Thread.sleep(timeout) 
    new Result { val id = 99999 } 
    }) 

    def worker(i: Int) = Future({ 
    Thread.sleep(100) 
    new BigObject(i) 
    }) 

    for (i <- Range(1, 1000)){ 
    println("round " + i) 
    Thread.sleep(20) 
    Future.firstCompletedOf(Seq(
     guardian, 
     worker(i) 
    )).map(r => println("result" + r.id)) 
    } 

    while (true){ 
    Thread.sleep(2000) 
    } 
} 

Answer 1

Vediamo come firstCompletedOf è implementato:

def firstCompletedOf[T](futures: TraversableOnce[Future[T]])(implicit executor: ExecutionContext): Future[T] = { 
    val p = Promise[T]() 
    val completeFirst: Try[T] => Unit = p tryComplete _ 
    futures foreach { _ onComplete completeFirst } 
    p.future 
} 

Nel fare { futures foreach { _ onComplete completeFirst }, la funzione { _ onComplete completeFirst } viene salvato da qualche parte via ExecutionContext.execute. Dove esattamente è salvata questa funzione è irrilevante, sappiamo solo che deve essere salvato da qualche parte in modo che possa essere selezionato in seguito ed eseguito su un pool di thread quando un thread diventa disponibile.

Questa funzione si chiude su completeFirst che si chiude su p. Finché c'è ancora un futuro (da futures) in attesa di essere completato, c'è un riferimento a p che ne impedisce la raccolta dei dati inutili (anche se a quel punto è probabile che firstCompletedOf sia già stato restituito, rimuovere p dal pila).

Quando il primo futuro viene completato, salva il risultato nella promessa (chiamando lo p.tryComplete). Poiché la promessa p contiene il risultato, il risultato è raggiungibile almeno fino a quando il numero p è raggiungibile e, come abbiamo visto, p è raggiungibile purché almeno una volta futuro da futures non sia stato completato. Questo è il motivo per cui il risultato non può essere raccolto prima che tutti i futures siano stati completati.

UPDATE: Ora la domanda è: potrebbe essere risolto? Penso che potrebbe. Tutto quello che dovremmo fare è garantire che il primo futuro completi "annulla" il riferimento a p in modo thread-safe, che può essere fatto usando un AtomicReference. Qualcosa di simile a questo:

def firstCompletedOf[T](futures: TraversableOnce[Future[T]])(implicit executor: ExecutionContext): Future[T] = { 
    val p = Promise[T]() 
    val pref = new java.util.concurrent.atomic.AtomicReference(p) 
    val completeFirst: Try[T] => Unit = { result: Try[T] => 
    val promise = pref.getAndSet(null) 
    if (promise != null) { 
     promise.tryComplete(result) 
    } 
    } 
    futures foreach { _ onComplete completeFirst } 
    p.future 
} 

ho provato e come previsto se non è permessa il risultato da garbage collection non appena il primo futura completa. Dovrebbe comportarsi allo stesso modo in tutti gli altri aspetti.