Perché in questo esempio non vi è alcun aumento significativo dell'utilizzo dei riduttori?

Question

(require '[clojure.core.reducers :as r]) 

(def data (into [] (take 10000000 (repeatedly #(rand-int 1000))))) 

(defn frequencies [coll] 
    (reduce (fn [counts x] 
    (merge-with + counts {x 1})) 
    {} coll)) 

(defn pfrequencies [coll] 
    (r/reduce (fn [counts x] 
    (merge-with + counts {x 1})) 
    {} coll)) 


user=> (time (do (frequencies data) nil)) 
"Elapsed time: 29697.183 msecs" 

user=> (time (do (pfrequencies data) nil)) 
"Elapsed time: 25273.794 msecs" 

user=> (time (do (frequencies data) nil)) 
"Elapsed time: 25384.086 msecs" 

user=> (time (do (pfrequencies data) nil)) 
"Elapsed time: 25778.502 msecs" 

E chi può mostrarmi un esempio con una notevole velocità?

Sono in esecuzione su Mac OS X 10.7.5 con Java 1.7 su un Intel Core i7 (2 core, http://ark.intel.com/products/54617).

Answer 1

hai chiamato lo pfrequencies, che, insieme con il vostro parallel-processing tag sulla questione, suggerisce si pensa che qualcosa sta usando più thread qui. Questo non è il caso, e non è nemmeno l'obiettivo "principale" della libreria dei riduttori.

La cosa principale che i riduttori comprano è che non è necessario allocare molte celle di controllo intermedie per le sequenze pigre. Prima dell'introduzione dei riduttori, frequencies allocava 10000000 celle cons per creare una vista sequenziale del vettore da utilizzare per reduce. Ora che esistono i riduttori, i vettori sanno come ridurli senza creare oggetti temporanei. Ma questa funzionalità è stata trasferita in clojure.core/reduce, che si comporta esattamente come r/reduce (ignorando alcune caratteristiche minori che sono irrilevanti qui). Quindi stai solo confrontando la tua funzione con un identico clone di se stesso.

La libreria di riduttori include anche la nozione di fold, che può eseguire alcuni lavori in parallelo e quindi unire insieme i risultati intermedi. Per utilizzare questo, è necessario fornire più informazioni rispetto alle esigenze di reduce: è necessario definire come avviare un "blocco" dal nulla; la tua funzione deve essere associativa; e devi specificare come combinare i blocchi. A. Webb's answer dimostra come usare fold correttamente, per lavorare su più thread.

Tuttavia, è improbabile che si possa trarre alcun vantaggio dal piegamento: oltre al motivo per cui nota (si rinuncia ai transienti, rispetto a clojure.core/frequencies), la creazione di una mappa non è facilmente parallelizzabile. Se la maggior parte del lavoro in frequencies è stato aggiunto (come sarebbe in qualcosa come (frequencies (repeat 1e6 1))), quindi fold sarebbe di aiuto; ma la maggior parte del lavoro è nella gestione delle chiavi nella hashmap, che alla fine deve essere single-threaded alla fine. Puoi costruire mappe in parallelo, ma poi devi unirle insieme; dal momento che questa combinazione ha un tempo proporzionale alla dimensione del chunk, piuttosto che al tempo costante, guadagni poco facendo comunque i pezzi su un thread separato.

Answer 2

Una versione fold della funzione frequenze sarebbe simile

(defn pfrequencies [coll] 
    (r/fold 
    (fn combinef 
     ([] {}) 
     ([x y] (merge-with + x y))) 
    (fn reducef 
     ([counts x] (merge-with + counts {x 1}))) 
    coll)) 

Su 2 core, sarà probabilmente molto più lento di clojure.core/frequencies che utilizza transitori. Almeno su 4 core, è più veloce (2x) rispetto alla prima implementazione, ma è ancora più lento di clojure.core/frequencies.

Si potrebbe anche sperimentare

(defn p2frequencies [coll] 
    (apply merge-with + (pmap clojure.core/frequencies (partition-all 512 coll)))) 

Answer 3

Alcuni seri spunti di riflessione nelle risposte qui. In questo caso specifico le mappe non dovrebbero essere necessarie, dal momento che il dominio dei risultati può essere facilmente previsto e inserito in un vettore in cui è possibile utilizzare l'indice. Quindi, un'implementazione ingenuo di un problema ingenuo sarebbe qualcosa di simile:

(defn freqs 
    [coll] 
    (reduce (fn [counts x] (assoc counts x (inc (get counts x)))) 
      (vec (int-array 1000 0)) 
      coll)) 

(defn rfreqs 
    [coll] 
    (r/fold 
     (fn combinef 
     ([] (vec (int-array 1000 0))) 
     ([& cols] (apply mapv + cols))) 
     (fn reducef 
     [counts x] (assoc counts x (inc (get counts x)))) 
     coll)) 

Qui il combinef sarebbe una semplice mappa aggiunta sopra le 1000 colonne delle collezioni risultanti, che dovrebbe essere trascurabile.

Questo dà alla versione del riduttore un'accelerazione di circa 2-3x rispetto alla versione normale, specialmente su dataset più grandi (10x-100x). Qualche giochino con la dimensione della partizione di r/fold (parametro 'n' opzionale) può essere fatto come finecorsa. Sembrava ottimale da usare (* 16 1024) con una dimensione dei dati di 1E8 (almeno 6GB JVM minimo).

Si potrebbe anche utilizzare i transienti in entrambe le versioni, ma non ho notato molti miglioramenti.

So che questa versione non è appropriata per l'utilizzo generico, ma potrebbe mostrare il miglioramento della velocità senza il sovraccarico di gestione hash.