iteratore parallelo in Scala

Question

È possibile in qualche modo utilizzare le raccolte parallele di Scala per parallelizzare uno Iteratorsenza valutandolo completamente prima?

Qui sto parlando di parallelizzare le trasformazioni funzionali su un Iterator, ovvero map e flatMap. Penso che ciò richieda in anticipo alcuni elementi dello Iterator, e quindi il calcolo di più, una volta che alcuni vengono consumati tramite next.

Tutto quello che riuscivo a trovare richiederebbe che l'iteratore fosse convertito in un Iterable o un Stream nella migliore delle ipotesi. Il Stream viene quindi completamente valutato quando si chiama .par su di esso.

Accolgo anche proposte di implementazione se questo non è facilmente disponibile. Le implementazioni devono supportare il parallelo map e flatMap.

Answer 1

La cosa migliore con la libreria standard non è probabilmente utilizza collezioni parallele ma concurrent.Future.traverse:

import concurrent._ 
import ExecutionContext.Implicits.global 
Future.traverse(Iterator(1,2,3))(i => Future{ i*i }) 

anche se credo che questo verrà eseguito il tutto a partire non appena possibile.

Answer 2

Dalla ML, Movimento degli elementi iterazione in parallelo:

https://groups.google.com/d/msg/scala-user/q2NVdE6MAGE/KnutOq3iT3IJ

ho spostato fuori Future.traverse per una ragione simile. Per il mio caso d'uso, mantenendo N lavori in esecuzione, mi sono ritrovato con il codice per limitare l'alimentazione del contesto di esecuzione dalla coda dei lavori.

Il mio primo tentativo ha riguardato il blocco del thread di feeder, ma questo ha rischiato anche il blocco delle attività che desideravano generare attività nel contesto di esecuzione. Che cosa sai, bloccare è malvagio.

Answer 3

È un po 'difficile da seguire esattamente quello che stai dopo, ma forse si tratta di qualcosa di simile:

val f = (x: Int) => x + 1 
val s = (0 to 9).toStream map f splitAt(6) match { 
    case (left, right) => left.par; right 
} 

Ciò eveluate f sui primi 6 elementi in parallelo e poi tornare un ruscello sul resto .

Answer 4

Mi rendo conto che questa è una vecchia domanda, ma l'implementazione ParIterator nella libreria iterata fa ciò che stavi cercando?

scala> import com.timgroup.iterata.ParIterator.Implicits._ 
scala> val it = (1 to 100000).toIterator.par().map(n => (n + 1, Thread.currentThread.getId)) 
scala> it.map(_._2).toSet.size 
res2: Int = 8 // addition was distributed over 8 threads