О Future.firstCompletedOf и Механизм сбора мусора

Question

Я столкнулся с этой проблемой в своем реальном проекте и подтвердил свой тестовый код и профайлер. Вместо того, чтобы вставлять код «tl; dr», я показываю вам фотографию, а затем описываю ее.

Проще говоря, я использую Future.firstCompletedOf, чтобы получить результат от 2 Future с, оба из которых не имеют общих вещей, и не заботятся друг о друге. Несмотря на то, что вопрос, который я хочу рассмотреть, Сборщик мусора не может перерабатывать первый объект Result, пока оба из Future s не закончили.

Так что мне действительно интересно, какой механизм стоит за этим. Может кто-нибудь объяснить это с более низкого уровня или дать мне какой-то намек.

Спасибо!

PS: Это потому что у них одинаковые ExecutionContext?

** Обновление ** код теста паста в соответствии с просьбой

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

Давайте посмотрим, как firstCompletedOf реализуется:

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 
} 

При выполнении { futures foreach { _ onComplete completeFirst } функция { _ onComplete completeFirst } сохраняется где-то через ExecutionContext.execute. Где точно сохраняется эта функция, мы просто знаем, что ее нужно сохранить где-то , чтобы ее можно было выбрать позже и выполнить в пуле потоков, когда поток станет доступным.

Эта функция закрывается над completeFirst, которая закрывается над p. До тех пор, пока есть еще одно будущее (от), которое должно быть завершено, есть ссылка на p, что предотвращает его сбор мусора (хотя, возможно, это то, что firstCompletedOf уже вернулся, удалив p из стек).

Когда первое будущее завершается, оно сохраняет результат в обещание (путем вызова p.tryComplete). Поскольку результат p обещает результат, результат будет достигнут, по крайней мере, до тех пор, пока p не достигнет уровня, и, как мы видели, p достигнут, поскольку по крайней мере один раз в будущем от futures еще не завершен. Именно по этой причине результат не может быть собран до того, как все фьючерсы будут завершены.

ОБНОВЛЕНИЕ: Вопрос теперь: не может быть исправлено? Я думаю, это возможно. Все, что нам нужно сделать, это обеспечить, чтобы первое будущее завершило «вычеркивание» ссылки на p поточно-безопасным способом, что можно сделать с помощью примера с использованием AtomicReference. Что-то вроде этого:

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 
} 

Я проверил его и, как ожидается, это позволит результат быть мусора, как только завершает первый будущее. Он должен вести себя одинаково во всех других отношениях.