Julia Language
Elaborazione parallela
Ricerca…
pmap
pmap prende una funzione (che tu specifichi) e la applica a tutti gli elementi di una matrice. Questo lavoro è suddiviso tra i lavoratori disponibili. pmap restituisce quindi i risultati da tale funzione in un altro array.
addprocs(3)
sqrts = pmap(sqrt, 1:10)
se la tua funzione richiede più argomenti, puoi fornire più vettori a pmap
dots = pmap(dot, 1:10, 11:20)
Come con @parallel , tuttavia, se la funzione assegnata a pmap non è in Julia di base (ovvero è definita dall'utente o definita in un pacchetto), è necessario assicurarsi che tale funzione sia disponibile per tutti i lavoratori per primi:
@everywhere begin
function rand_det(n)
det(rand(n,n))
end
end
determinants = pmap(rand_det, 1:10)
Vedi anche questo SO Q & A.
@parallelo
@parallel può essere usato per parallelizzare un loop, dividendo i passaggi del loop su diversi worker. Come un esempio molto semplice:
addprocs(3)
a = collect(1:10)
for idx = 1:10
println(a[idx])
end
Per un esempio leggermente più complesso, considera:
@time begin
@sync begin
@parallel for idx in 1:length(a)
sleep(a[idx])
end
end
end
27.023411 seconds (13.48 k allocations: 762.532 KB)
julia> sum(a)
55
Quindi, vediamo che se avessimo eseguito questo ciclo senza @parallel ci sarebbero voluti 55 secondi, anziché 27, da eseguire.
Possiamo anche fornire un operatore di riduzione per la macro @parallel . Supponiamo di avere una matrice, vogliamo sommare ogni colonna della matrice e quindi moltiplicare queste somme l'una dall'altra:
A = rand(100,100);
@parallel (*) for idx = 1:size(A,1)
sum(A[:,idx])
end
Ci sono diverse cose importanti da tenere a mente quando si utilizza @parallel per evitare comportamenti imprevisti.
Primo: se si desidera utilizzare qualsiasi funzione nei propri loop che non sono in Julia di base (ad esempio, le funzioni definite nello script o importate dai pacchetti), è necessario rendere tali funzioni accessibili ai lavoratori. Pertanto, ad esempio, quanto segue non funzionerebbe:
myprint(x) = println(x)
for idx = 1:10
myprint(a[idx])
end
Invece, dovremmo usare:
@everywhere begin
function myprint(x)
println(x)
end
end
@parallel for idx in 1:length(a)
myprint(a[idx])
end
Secondo Sebbene ogni lavoratore potrà accedere agli oggetti nel campo di applicazione del controllore, non saranno in grado di modificarli. così
a = collect(1:10)
@parallel for idx = 1:length(a)
a[idx] += 1
end
julia> a'
1x10 Array{Int64,2}:
1 2 3 4 5 6 7 8 9 10
Considerando che, se avessimo eseguito il ciclo con il @parallel, avremmo modificato correttamente l'array a .
PER INDIRIZZARLO, possiamo invece creare a oggetto di tipo SharedArray modo che ogni lavoratore possa accedervi e modificarlo:
a = convert(SharedArray{Float64,1}, collect(1:10))
@parallel for idx = 1:length(a)
a[idx] += 1
end
julia> a'
1x10 Array{Float64,2}:
2.0 3.0 4.0 5.0 6.0 7.0 8.0 9.0 10.0 11.0
@spawn e @spawnat
Le macro @spawn e @spawnat sono due degli strumenti che Julia mette a disposizione per assegnare compiti ai lavoratori. Ecco un esempio:
julia> @spawnat 2 println("hello world")
RemoteRef{Channel{Any}}(2,1,3)
julia> From worker 2: hello world
Entrambe queste macro valuteranno un'espressione su un processo di lavoro. L'unica differenza tra i due è che @spawnat ti permette di scegliere quale lavoratore valuterà l'espressione (nell'esempio sopra è specificato worker 2) mentre con @spawn verrà automaticamente scelto un worker, in base alla disponibilità.
Nell'esempio sopra, abbiamo semplicemente avuto worker 2 per eseguire la funzione println. Non c'era nulla di interessante da restituire o recuperare da questo. Spesso, tuttavia, l'espressione che abbiamo inviato al lavoratore produrrà qualcosa che desideriamo recuperare. Notare nell'esempio sopra, quando abbiamo chiamato @spawnat , prima di ottenere la stampa da worker 2, abbiamo visto quanto segue:
RemoteRef{Channel{Any}}(2,1,3)
Questo indica che la macro @spawnat restituirà un oggetto di tipo RemoteRef . Questo oggetto a sua volta conterrà i valori di ritorno dalla nostra espressione che viene inviata al lavoratore. Se vogliamo recuperare quei valori, possiamo prima assegnare il RemoteRef che @spawnat ritorna ad un oggetto e poi, e poi usa la funzione fetch() che opera su un oggetto di tipo RemoteRef , per recuperare i risultati memorizzati da una valutazione eseguita su un lavoratore.
julia> result = @spawnat 2 2 + 5
RemoteRef{Channel{Any}}(2,1,26)
julia> fetch(result)
7
La chiave per essere in grado di usare efficacemente @spawn è capire la natura dietro le espressioni su cui opera. Usare @spawn per inviare comandi ai lavoratori è un po 'più complicato della semplice digitazione diretta di ciò che si dovrebbe scrivere se si stesse eseguendo un "interprete" su uno dei lavoratori o eseguendo il codice in modo nativo su di essi. Ad esempio, supponiamo di voler utilizzare @spawnat per assegnare un valore a una variabile su un worker. Potremmo provare:
@spawnat 2 a = 5
RemoteRef{Channel{Any}}(2,1,2)
Ha funzionato? Bene, vediamo con il lavoratore 2 provare a stampare a .
julia> @spawnat 2 println(a)
RemoteRef{Channel{Any}}(2,1,4)
julia>
Non è successo niente. Perché? Possiamo investigare di più usando fetch() come sopra. fetch() può essere molto utile perché recupera non solo i risultati di successo ma anche i messaggi di errore. Senza di esso, potremmo anche non sapere che qualcosa è andato storto.
julia> result = @spawnat 2 println(a)
RemoteRef{Channel{Any}}(2,1,5)
julia> fetch(result)
ERROR: On worker 2:
UndefVarError: a not defined
Il messaggio di errore dice che a non è definito su worker 2. Ma perché è questo? Il motivo è che abbiamo bisogno di avvolgere la nostra operazione di assegnazione in un'espressione che usiamo quindi @spawn per dire al lavoratore di valutare. Di seguito è riportato un esempio, con spiegazione seguente:
julia> @spawnat 2 eval(:(a = 2))
RemoteRef{Channel{Any}}(2,1,7)
julia> @spawnat 2 println(a)
RemoteRef{Channel{Any}}(2,1,8)
julia> From worker 2: 2
La sintassi :() è ciò che Julia usa per designare le espressioni . Quindi usiamo la eval() in Julia, che valuta un'espressione, e usiamo la macro @spawnat per @spawnat che l'espressione deve essere valutata su worker 2.
Potremmo anche ottenere lo stesso risultato di:
julia> @spawnat(2, eval(parse("c = 5")))
RemoteRef{Channel{Any}}(2,1,9)
julia> @spawnat 2 println(c)
RemoteRef{Channel{Any}}(2,1,10)
julia> From worker 2: 5
Questo esempio dimostra due nozioni aggiuntive. Innanzitutto, vediamo che possiamo anche creare un'espressione usando la funzione parse() chiamata su una stringa. In secondo luogo, vediamo che possiamo utilizzare le parentesi quando si chiama @spawnat , in situazioni in cui ciò potrebbe rendere la nostra sintassi più chiara e gestibile.
Quando usare @parallel vs pmap
La documentazione di Julia lo consiglia
pmap () è progettato per il caso in cui ogni chiamata di funzione svolge una grande quantità di lavoro. Al contrario, @parallel for può gestire situazioni in cui ogni iterazione è minima, forse semplicemente sommando due numeri.
Ci sono diverse ragioni per questo. Innanzitutto, pmap incorre in costi di avvio maggiori per l'avvio di lavori sui lavoratori. Pertanto, se i lavori sono molto piccoli, questi costi di avvio potrebbero diventare inefficienti. Viceversa, tuttavia, pmap svolge un lavoro "più intelligente" nell'assegnare posti di lavoro tra i lavoratori. In particolare, crea una coda di lavori e invia un nuovo lavoro a ciascun lavoratore ogni volta che quel lavoratore diventa disponibile. @parallel al contrario, divide tutto il lavoro da fare tra i lavoratori quando viene chiamato. Pertanto, se alcuni lavoratori impiegano più tempo a svolgere il proprio lavoro rispetto ad altri, si può finire con una situazione in cui la maggior parte dei lavoratori ha finito e sono inattiva mentre alcuni rimangono attivi per un numero eccessivo di tempo, finendo il proprio lavoro. Tale situazione, tuttavia, è meno probabile che si verifichi con lavori molto piccoli e semplici.
Ciò che segue illustra questo: supponiamo di avere due lavoratori, uno dei quali è lento e l'altro è il doppio più veloce. Idealmente, vorremmo dare al lavoratore veloce il doppio del lavoro del lavoratore lento. (oppure, potremmo avere lavori veloci e lenti, ma il principale è esattamente lo stesso). pmap lo realizzerà, ma @parallel non lo farà.
Per ogni test, inizializziamo quanto segue:
addprocs(2)
@everywhere begin
function parallel_func(idx)
workernum = myid() - 1
sleep(workernum)
println("job $idx")
end
end
Ora, per il test @parallel , eseguiamo quanto segue:
@parallel for idx = 1:12
parallel_func(idx)
end
E torna all'output di stampa:
julia> From worker 2: job 1
From worker 3: job 7
From worker 2: job 2
From worker 2: job 3
From worker 3: job 8
From worker 2: job 4
From worker 2: job 5
From worker 3: job 9
From worker 2: job 6
From worker 3: job 10
From worker 3: job 11
From worker 3: job 12
È quasi dolce. I lavoratori hanno "condiviso" il lavoro in modo uniforme. Nota che ogni lavoratore ha completato 6 lavori, anche se il lavoratore 2 è due volte più veloce del lavoratore 3. Può essere toccante, ma non è efficiente.
Per il test pmap , pmap le seguenti operazioni:
pmap(parallel_func, 1:12)
e ottieni l'output:
From worker 2: job 1
From worker 3: job 2
From worker 2: job 3
From worker 2: job 5
From worker 3: job 4
From worker 2: job 6
From worker 2: job 8
From worker 3: job 7
From worker 2: job 9
From worker 2: job 11
From worker 3: job 10
From worker 2: job 12
Ora, si noti che worker 2 ha eseguito 8 lavori e il worker 3 ha eseguito 4. Questo è esattamente in proporzione alla loro velocità e cosa vogliamo per l'efficienza ottimale. pmap è un master per compiti difficili - da ciascuno secondo le proprie capacità.
@async e @sync
Secondo la documentazione sotto ?@async , " @async wrapping di un'espressione in un'attività." Ciò significa che per qualsiasi cosa rientri nel suo ambito, Julia avvierà questa attività in esecuzione, ma poi procederà a ciò che viene dopo nello script senza attendere il completamento dell'attività. Quindi, ad esempio, senza la macro otterrai:
julia> @time sleep(2)
2.005766 seconds (13 allocations: 624 bytes)
Ma con la macro, ottieni:
julia> @time @async sleep(2)
0.000021 seconds (7 allocations: 657 bytes)
Task (waiting) @0x0000000112a65ba0
julia>
In tal modo, Julia consente allo script di procedere (e alla macro @time di eseguire completamente) senza attendere che l'attività (in questo caso, dormire per due secondi) @time completata.
La macro @sync , al contrario, "Attende fino a quando tutti gli usi dinamicamente chiusi di @async , @spawn , @spawnat e @parallel sono completi." (secondo la documentazione sotto ?@sync ). Quindi, vediamo:
julia> @time @sync @async sleep(2)
2.002899 seconds (47 allocations: 2.986 KB)
Task (done) @0x0000000112bd2e00
In questo semplice esempio, non è necessario includere una singola istanza di @async e @sync insieme. Ma, dove @sync può essere utile, è il caso in cui @async applicato a più operazioni che si desidera consentire a tutti di iniziare subito senza attendere il completamento di ciascuna operazione.
Ad esempio, supponiamo di avere più lavoratori e vorremmo iniziare ognuno di loro a lavorare su un'attività contemporaneamente e quindi recuperare i risultati da tali attività. Un tentativo iniziale (ma errato) potrebbe essere:
addprocs(2)
@time begin
a = cell(nworkers())
for (idx, pid) in enumerate(workers())
a[idx] = remotecall_fetch(pid, sleep, 2)
end
end
## 4.011576 seconds (177 allocations: 9.734 KB)
Il problema qui è che il ciclo attende ogni operazione di remotecall_fetch() per finire, cioè che ogni processo completi il suo lavoro (in questo caso dormendo per 2 secondi) prima di continuare ad avviare la successiva operazione remotecall_fetch() . In termini di situazione pratica, qui non riceviamo i vantaggi del parallelismo, poiché i nostri processi non stanno facendo il loro lavoro (cioè dormendo) simultaneamente.
Possiamo correggere questo, tuttavia, utilizzando una combinazione dei macro @async e @sync :
@time begin
a = cell(nworkers())
@sync for (idx, pid) in enumerate(workers())
@async a[idx] = remotecall_fetch(pid, sleep, 2)
end
end
## 2.009416 seconds (274 allocations: 25.592 KB)
Ora, se contiamo ogni passo del ciclo come operazione separata, vediamo che ci sono due operazioni separate precedute dalla macro @async . La macro consente a ciascuno di questi di avviarsi e il codice per continuare (in questo caso al prossimo passo del ciclo) prima di ogni finitura. Tuttavia, l'uso della macro @sync , il cui ambito comprende l'intero ciclo, significa che non consentiremo che lo script proceda oltre quel ciclo finché tutte le operazioni precedute da @async siano state completate.
È possibile ottenere una comprensione ancora più chiara del funzionamento di queste macro modificando ulteriormente l'esempio precedente per vedere come cambia in determinate modifiche. Ad esempio, supponiamo di avere solo @async senza @sync :
@time begin
a = cell(nworkers())
for (idx, pid) in enumerate(workers())
println("sending work to $pid")
@async a[idx] = remotecall_fetch(pid, sleep, 2)
end
end
## 0.001429 seconds (27 allocations: 2.234 KB)
Qui, la macro @async ci consente di continuare nel nostro ciclo ancor prima che ciascuna operazione di remotecall_fetch() esecuzione. Ma, nel bene o nel male, non abbiamo una macro @sync per impedire che il codice continui oltre questo ciclo fino a quando tutte le operazioni di remotecall_fetch() non terminano.
Ciononostante, ogni operazione di remotecall_fetch() è ancora in esecuzione in parallelo, anche quando andiamo avanti. Possiamo vedere che, se aspettiamo due secondi, l'array a, contenente i risultati, conterrà:
sleep(2)
julia> a
2-element Array{Any,1}:
nothing
nothing
(L'elemento "niente" è il risultato di un recupero riuscito dei risultati della funzione sleep, che non restituisce alcun valore)
Possiamo anche vedere che le due operazioni di remotecall_fetch() iniziano essenzialmente nello stesso momento perché i comandi di print che li precedono vengono eseguiti in rapida successione (l'output di questi comandi non è mostrato qui). Confrontalo con il prossimo esempio in cui i comandi di print vengono eseguiti a intervalli di 2 secondi l'uno dall'altro:
Se mettiamo la macro @async sull'intero loop (invece che sul suo passo interno), il nostro script continuerà immediatamente senza attendere che le operazioni di remotecall_fetch() finiscano. Ora, tuttavia, consentiamo allo script di continuare oltre il ciclo nel suo complesso. Non permettiamo che ogni singola fase del ciclo inizi prima della precedente. Pertanto, a differenza dell'esempio sopra, due secondi dopo che lo script procede dopo il ciclo, l'array dei results ha ancora un elemento come #undef che indica che la seconda operazione remotecall_fetch() non è ancora stata completata.
@time begin
a = cell(nworkers())
@async for (idx, pid) in enumerate(workers())
println("sending work to $pid")
a[idx] = remotecall_fetch(pid, sleep, 2)
end
end
# 0.001279 seconds (328 allocations: 21.354 KB)
# Task (waiting) @0x0000000115ec9120
## This also allows us to continue to
sleep(2)
a
2-element Array{Any,1}:
nothing
#undef
E, non sorprendentemente, se mettiamo @sync e @async accanto all'altro, otteniamo che ogni remotecall_fetch() eseguito in modo sequenziale (anziché simultaneamente) ma non continuiamo nel codice finché ognuno non ha finito. In altre parole, questo sarebbe essenzialmente l'equivalente se non avessimo nessuna macro in posto, proprio come sleep(2) si comporta essenzialmente in modo identico a @sync @async sleep(2)
@time begin
a = cell(nworkers())
@sync @async for (idx, pid) in enumerate(workers())
a[idx] = remotecall_fetch(pid, sleep, 2)
end
end
# 4.019500 seconds (4.20 k allocations: 216.964 KB)
# Task (done) @0x0000000115e52a10
Si noti inoltre che è possibile avere operazioni più complicate nell'ambito della macro @async . La documentazione fornisce un esempio contenente un intero ciclo nell'ambito di @async .
Ricordiamo che l'aiuto per i macro di sincronizzazione afferma che "Attenderà fino a quando tutti gli usi dinamicamente chiusi di @async , @spawn , @spawnat e @parallel saranno completi." Ai fini di ciò che conta come "completo" importa come si definiscono le attività nell'ambito delle macro @sync e @async . Considera l'esempio seguente, che è una leggera variazione su uno degli esempi sopra riportati:
@time begin
a = cell(nworkers())
@sync for (idx, pid) in enumerate(workers())
@async a[idx] = remotecall(pid, sleep, 2)
end
end
## 0.172479 seconds (93.42 k allocations: 3.900 MB)
julia> a
2-element Array{Any,1}:
RemoteRef{Channel{Any}}(2,1,3)
RemoteRef{Channel{Any}}(3,1,4)
L'esempio precedente impiegava circa 2 secondi per indicare che le due attività erano eseguite in parallelo e che lo script attendeva che ciascuna completasse l'esecuzione delle sue funzioni prima di procedere. Questo esempio, tuttavia, ha una valutazione del tempo molto più bassa. Il motivo è che ai fini di @sync l'operazione remotecall() ha "finito" una volta che ha inviato al lavoratore il compito da svolgere. (Si noti che l'array risultante, a, qui, contiene solo tipi di oggetto RemoteRef , che indicano solo che c'è qualcosa in corso con un particolare processo che in teoria potrebbe essere recuperato in qualche punto in futuro). Al contrario, l'operazione remotecall_fetch() ha solo "finito" quando riceve il messaggio dal lavoratore che la sua attività è completa.
Quindi, se stai cercando dei modi per assicurarti che certe operazioni con i lavoratori siano completate prima di proseguire nel tuo script (come ad esempio è discusso in questo post ) è necessario riflettere attentamente su ciò che conta come "completo" e su come misurare e quindi renderlo operativo nel tuo script.
Aggiunta di lavoratori
Quando avvii per la prima volta Julia, per impostazione predefinita, sarà disponibile un solo processo in esecuzione e disponibile per il lavoro. Puoi verificarlo usando:
julia> nprocs()
1
Per trarre vantaggio dall'elaborazione parallela, è necessario innanzitutto aggiungere altri lavoratori che saranno quindi disponibili per svolgere il lavoro che gli viene assegnato. Puoi farlo all'interno del tuo script (o dell'interprete) usando: addprocs(n) dove n è il numero di processi che vuoi usare.
In alternativa, puoi aggiungere processi quando avvii Julia dalla riga di comando usando:
$ julia -p n
dove n è il numero di processi aggiuntivi che desideri aggiungere. Quindi, se iniziamo con Julia
$ julia -p 2
Quando inizieremo Julia otterremo:
julia> nprocs()
3