Scala Parallel Collections - jak wrócić wcześniej?

Question

Mam listę możliwych wartości wejściowych

val inputValues = List(1,2,3,4,5) 

mam naprawdę długo, aby funkcja, która daje mi wynik

def reallyLongFunction(input: Int) : Option[String] = { ..... } 

Korzystanie Scala równoległe kolekcje obliczyć, mogę łatwo zrobić

inputValues.par.map(reallyLongFunction(_)) 

Aby uzyskać wszystkie wyniki, należy jednocześnie. Problem polega na tym, że tak naprawdę nie chcę wszystkich wyników, chcę tylko PIERWSZY wynik. Gdy tylko jedno z moich danych wejściowych zakończy się sukcesem, chcę uzyskać wyniki i chcę kontynuować moje życie. Zrobiło to dużo dodatkowej pracy.

Jak uzyskać najlepsze z obu światów? Chcę

1. Get pierwszy wynik, który zwraca coś z mojej długiej funkcji
2. zatrzymać wszystkie moje inne wątki od bezużytecznej pracy.

Edit - Rozwiązałem go jak niemy programista java poprzez

@volatile var done = false; 

który jest ustawiony i sprawdzony w moim reallyLongFunction. Działa to, ale nie czuje się bardzo scala. Chciałby lepszego sposobu, aby to zrobić ....

Answer 1

Zrobiłem interpretację twojego pytania w taki sam sposób jak huynhjl, ale jeśli chcesz po prostu przeszukać i odrzucić None s, możesz zrobić coś takiego, aby uniknąć powtórzenia obliczeń po znalezieniu odpowiedniego wyniku:

class Computation[A,B](value: A, function: A => B) { 
    lazy val result = function(value) 
} 

def f(x: Int) = {   // your function here 
    Thread.sleep(100 - x) 
    if (x > 5) Some(x * 10) 
    else None 
} 

val list = List.range(1, 20) map (i => new Computation(i, f)) 
val found = list.par find (_.result.isDefined) 
    //found is Option[Computation[Int,Option[Int]]] 
val result = found map (_.result.get) 
    //result is Option[Int] 

Jednak dla kolekcjonerów równoległych wydaje się, że w przypadku zbiorów równoległych wykonuje się dużo niepotrzebnej pracy (patrz this question), więc może to nie działać dobrze, z obecnymi wersjami przynajmniej Scala.

Flagi ulotne używane są w kolekcjach równoległych (spójrz na źródła dla find, exists i forall), więc uważam, że Twój pomysł jest dobry. Właściwie lepiej jest włączyć flagę do samej funkcji. To zabija referencyjną przezroczystość twojej funkcji (np. Dla niektórych wejść twoja funkcja teraz czasami zwraca None zamiast Some), ale ponieważ odrzucasz zatrzymane obliczenia, nie powinno to mieć znaczenia.

Answer 2

(Aktualizacja: nie, to nie działa, nie robi mapie)

będzie działać zrobić coś takiego:

inputValues.par.find({ v => reallyLongFunction(v); true }) 

realizacja wykorzystuje to:

protected[this] class Find[U >: T](pred: T => Boolean, protected[this] val pit: IterableSplitter[T]) extends Accessor[Option[U], Find[U]] { 
    @volatile var result: Option[U] = None 
    def leaf(prev: Option[Option[U]]) = { if (!pit.isAborted) result = pit.find(pred); if (result != None) pit.abort } 
    protected[this] def newSubtask(p: IterableSplitter[T]) = new Find(pred, p) 
    override def merge(that: Find[U]) = if (this.result == None) result = that.result 
    } 

który wygląda całkiem w duchu podobnym do @volatile chyba nie masz na to patrzeć ;-)

Answer 3

Jeśli chcesz korzystać z nie-core biblioteki, myślę, że Futures będzie dobrze pasować do tego zadania.Na przykład:

• Futures Akka obejmują Futures.firstCompletedOf
• Futures Twitter obejmują Future.select

... z których oba wydają się włączyć funkcję, której szukasz.