Come imposto il numero predefinito di thread per le collezioni parallele Scala 2.10?

Question

In Scala prima della 2.10, posso impostare il parallelismo in defaultForkJoinPool (come in questa risposta scala parallel collections degree of parallelism). In Scala 2.10, questa API non esiste più. È ben documentato che possiamo impostare il parallelismo su una singola raccolta (http://docs.scala-lang.org/overviews/parallel-collections/configuration.html) assegnando alla sua proprietà taskSupport.

Tuttavia, utilizzo raccolte parallele su tutta la mia base di codice e non vorrei aggiungere altre due righe a ogni singola istanza di raccolta. C'è un modo per configurare le dimensioni del pool di thread predefinito globale in modo che someCollection.par.map(f(_)) utilizzi automaticamente il numero predefinito di thread?

Answer 1

So che la domanda ha più di un mese, ma ho appena avuto esattamente la stessa domanda. Googling non è stato utile e non ho trovato nulla che apparisse a metà strada nella nuova API.

Impostazione -Dscala.concurrent.context.maxThreads = n come suggerito qui: Set the parallelism level for all collections in Scala 2.10? apparentemente non ha avuto effetto a tutti, ma non sono sicuro se ho usato correttamente (ho eseguito la mia applicazione con 'java' in un ambiente senza 'scala' installata esplicitamente, potrebbe essere la causa).

Non so perché le persone scala hanno rimosso questo setter essenziale dall'oggetto pacchetto appropriato.

Tuttavia, è spesso possibile utilizzare la reflection per aggirare un'interfaccia incompleta/strano:

def setParallelismGlobally(numThreads: Int): Unit = { 
    val parPkgObj = scala.collection.parallel.`package` 
    val defaultTaskSupportField = parPkgObj.getClass.getDeclaredFields.find{ 
    _.getName == "defaultTaskSupport" 
    }.get 

    defaultTaskSupportField.setAccessible(true) 
    defaultTaskSupportField.set(
    parPkgObj, 
    new scala.collection.parallel.ForkJoinTaskSupport(
     new scala.concurrent.forkjoin.ForkJoinPool(numThreads) 
    ) 
) 
} 

Per chi non ha familiarità con le caratteristiche più oscuri della Scala, ecco un breve spiegazione:

scala.collection.parallel.`package` 

accede all'oggetto pacchetto con la variabile defaultTaskSupport (assomiglia alla variabile statica di Java, ma in realtà è una variabile membro dell'oggetto pacchetto). I backtick sono necessari per l'identificatore, perché package è una parola chiave riservata. Quindi otteniamo il campo finale privato che vogliamo (getField ("defaultTaskSupport") non ha funzionato per qualche motivo? ...), digli di essere accessibile per poterlo modificare, e quindi sostituirlo con il valore di il nostro ForkJoinTaskSupport.

Non ho ancora capito il meccanismo esatto della creazione di raccolte parallele, ma il codice sorgente del tratto Combiner suggerisce che il valore di defaultTaskSupport dovrebbe in qualche modo ricadere sulle raccolte parallele.

Si noti che la domanda è qualitativamente dello stesso tipo di una domanda molto più vecchia: "Ho Math.random() su tutta la mia base di codice, come posso impostare il seme su un numero fisso per scopi di debug?" (Vedi ad esempio: Set seed on Math.random()). In entrambi i casi, abbiamo una sorta di variabile "statica" globale che utilizziamo implicitamente in un milione di luoghi diversi, vogliamo cambiarla, ma non ci sono setter per questa variabile => usiamo la riflessione.

Brutto da morire, ma sembra funzionare bene. Se è necessario limitare il numero totale di thread, non dimenticare che il garbage collector viene eseguito su un thread separato.