Julia Language
Traitement parallèle
Recherche…
pmap
pmap prend une fonction (que vous spécifiez) et l'applique à tous les éléments d'un tableau. Ce travail est réparti entre les travailleurs disponibles. pmap renvoie alors les résultats de cette fonction dans un autre tableau.
addprocs(3)
sqrts = pmap(sqrt, 1:10)
Si vous pmap plusieurs arguments, vous pouvez fournir plusieurs vecteurs à pmap
dots = pmap(dot, 1:10, 11:20)
Comme avec @parallel , cependant, si la fonction donnée à pmap n'est pas dans la base Julia (c'est-à-dire définie par l'utilisateur ou définie dans un package), vous devez d'abord vous assurer que cette fonction est accessible à tous:
@everywhere begin
function rand_det(n)
det(rand(n,n))
end
end
determinants = pmap(rand_det, 1:10)
Voir aussi ce SO Q & A.
@parallèle
@parallel peut être utilisé pour paralléliser une boucle, en divisant les étapes de la boucle en différentes parties. Comme exemple très simple:
addprocs(3)
a = collect(1:10)
for idx = 1:10
println(a[idx])
end
Pour un exemple légèrement plus complexe, considérez:
@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
Ainsi, nous voyons que si nous avions exécuté cette boucle sans @parallel 55 secondes au lieu de 27 auraient été nécessaires.
Nous pouvons également fournir un opérateur de réduction pour la macro @parallel . Supposons que nous ayons un tableau, nous voulons additionner chaque colonne du tableau et ensuite multiplier ces sommes les unes par les autres:
A = rand(100,100);
@parallel (*) for idx = 1:size(A,1)
sum(A[:,idx])
end
Il y a plusieurs choses importantes à garder à l'esprit lorsque vous utilisez @parallel pour éviter les comportements inattendus.
Premièrement: si vous souhaitez utiliser des fonctions de vos boucles qui ne sont pas dans la base Julia (par exemple, des fonctions que vous définissez dans votre script ou que vous importez depuis des packages), vous devez les rendre accessibles aux utilisateurs. Ainsi, par exemple, ce qui suit ne fonctionnerait pas :
myprint(x) = println(x)
for idx = 1:10
myprint(a[idx])
end
Au lieu de cela, nous devrions utiliser:
@everywhere begin
function myprint(x)
println(x)
end
end
@parallel for idx in 1:length(a)
myprint(a[idx])
end
Seconde Bien que chaque travailleur puisse accéder aux objets dans la portée du contrôleur, il ne pourra pas les modifier. Ainsi
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
Alors que si nous avions exécuté la boucle sans le @parallel, cela aurait pu modifier le tableau a .
POUR ADRESSER CECI, nous pouvons créer a objet de type SharedArray afin que chaque agent puisse y accéder et le modifier:
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 et @spawnat
Les macros @spawn et @spawnat sont deux des outils mis à disposition par Julia pour attribuer des tâches aux travailleurs. Voici un exemple:
julia> @spawnat 2 println("hello world")
RemoteRef{Channel{Any}}(2,1,3)
julia> From worker 2: hello world
Ces deux macros évalueront une expression sur un processus de travail. La seule différence entre les deux est que @spawnat vous permet de choisir quel travailleur évaluera l'expression (dans l'exemple ci-dessus, le travailleur 2 est spécifié) alors qu'avec @spawn un travailleur sera automatiquement choisi, en fonction de la disponibilité.
Dans l'exemple ci-dessus, nous avons simplement demandé au travailleur 2 d'exécuter la fonction println. Il n'y avait rien d'intéressant à retourner ou à en récupérer. Souvent, cependant, l'expression que nous envoyons au travailleur donnera quelque chose que nous souhaitons récupérer. Notez que dans l'exemple ci-dessus, lorsque nous avons appelé @spawnat , avant que nous ayons obtenu l'impression de l'ouvrier 2, nous avons vu ce qui suit:
RemoteRef{Channel{Any}}(2,1,3)
Cela indique que la macro @spawnat renvoie un objet de type RemoteRef . Cet objet contiendra à son tour les valeurs de retour de notre expression envoyée au travailleur. Si nous voulons récupérer ces valeurs, nous pouvons d'abord attribuer le RemoteRef que @spawnat renvoie à un objet, puis utiliser la fonction fetch() qui opère sur un objet de type RemoteRef , pour récupérer les résultats stockés dans une évaluation effectuée sur un travailleur.
julia> result = @spawnat 2 2 + 5
RemoteRef{Channel{Any}}(2,1,26)
julia> fetch(result)
7
La clé pour pouvoir utiliser efficacement @spawn est de comprendre la nature des expressions sur lesquelles il opère. Utiliser @spawn pour envoyer des commandes aux travailleurs est un peu plus compliqué que de simplement taper directement ce que vous tapez si vous exécutez un "interpréteur" sur l'un des travailleurs ou si vous exécutez du code en mode natif. Par exemple, supposons que nous voulions utiliser @spawnat pour assigner une valeur à une variable sur un agent. Nous pourrions essayer:
@spawnat 2 a = 5
RemoteRef{Channel{Any}}(2,1,2)
A-t-il fonctionné? Eh bien, nous allons voir en ayant 2 travailleurs essayer d'imprimer a .
julia> @spawnat 2 println(a)
RemoteRef{Channel{Any}}(2,1,4)
julia>
Rien ne s'est passé. Pourquoi? Nous pouvons étudier cela plus en utilisant fetch() comme ci-dessus. fetch() peut être très utile car il récupère non seulement les résultats réussis, mais également les messages d'erreur. Sans cela, nous ne pourrions même pas savoir que quelque chose a mal tourné.
julia> result = @spawnat 2 println(a)
RemoteRef{Channel{Any}}(2,1,5)
julia> fetch(result)
ERROR: On worker 2:
UndefVarError: a not defined
Le message d'erreur indique que a n'est pas défini sur worker 2. Mais pourquoi est-ce? La raison en est que nous devons envelopper notre opération d'affectation dans une expression que nous utilisons ensuite avec @spawn pour @spawn à l'ouvrier d'évaluer. Voici un exemple, avec une explication ci-dessous:
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 syntaxe :() est ce que Julia utilise pour désigner des expressions . Nous utilisons ensuite la fonction eval() dans Julia, qui évalue une expression, et nous utilisons la macro @spawnat pour indiquer que l'expression doit être évaluée sur worker 2.
Nous pourrions également atteindre le même résultat que:
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
Cet exemple montre deux notions supplémentaires. Tout d'abord, nous voyons que nous pouvons également créer une expression à l'aide de la fonction parse() appelée sur une chaîne. Deuxièmement, nous voyons que nous pouvons utiliser des parenthèses lorsque nous appelons @spawnat , dans des situations où cela pourrait rendre notre syntaxe plus claire et plus facile à gérer.
Quand utiliser @parallel vs pmap
La documentation de Julia indique que
pmap () est conçu pour le cas où chaque appel de fonction effectue une grande quantité de travail. En revanche, @parallel for peut gérer des situations où chaque itération est minuscule, en faisant simplement la somme de deux nombres.
Il y a plusieurs raisons à cela. Premièrement, pmap engendre des coûts de démarrage plus élevés en pmap emplois pour les travailleurs. Ainsi, si les tâches sont très petites, ces coûts de démarrage peuvent devenir inefficaces. Inversement, pmap accomplit un travail "plus intelligent" en répartissant les emplois entre les travailleurs. En particulier, il crée une file d'attente de travaux et envoie un nouveau travail à chaque travailleur chaque fois que ce travailleur est disponible. @parallel en revanche, divise tout le travail à faire parmi les travailleurs quand il est appelé. En tant que tel, si certains travailleurs prennent plus de temps que d'autres, vous pouvez vous retrouver avec une situation où la plupart de vos travailleurs ont fini et sont inactifs, tandis que quelques-uns restent actifs pour une durée excessive, terminant leur travail. Une telle situation est toutefois moins susceptible de se produire avec des emplois très petits et simples.
Ce qui suit illustre ceci: supposons que nous ayons deux travailleurs, dont l'un est lent et l'autre est deux fois plus rapide. Idéalement, nous voudrions donner au travailleur rapide deux fois plus de travail que le travailleur lent. (ou nous pourrions avoir des travaux rapides et lents, mais le principe est exactement le même). pmap parviendra, mais @parallel ne le fera pas.
Pour chaque test, nous initialisons les éléments suivants:
addprocs(2)
@everywhere begin
function parallel_func(idx)
workernum = myid() - 1
sleep(workernum)
println("job $idx")
end
end
Maintenant, pour le test @parallel , nous @parallel les opérations suivantes:
@parallel for idx = 1:12
parallel_func(idx)
end
Et récupérer l'impression:
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
C'est presque bon Les travailleurs ont "partagé" le travail de manière égale. Notez que chaque travailleur a effectué 6 tâches, même si le travailleur 2 est deux fois plus rapide que le travailleur 3. Il peut être touchant, mais il est inefficace.
Pour le test pmap , je lance ce qui suit:
pmap(parallel_func, 1:12)
et obtenir la sortie:
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
Maintenant, notez que le travailleur 2 a effectué 8 travaux et que le travailleur 3 a effectué 4 tâches. Ceci est exactement proportionnel à leur vitesse et à ce que nous souhaitons pour une efficacité optimale. pmap est un maître de tâche difficile - de chacun selon ses capacités.
@ async et @sync
Selon la documentation sous ?@async , " @async une expression dans une tâche". Cela signifie que pour tout ce qui relève de sa portée, Julia lance cette tâche, mais passe ensuite à la prochaine étape du script sans attendre la fin de la tâche. Ainsi, par exemple, sans la macro, vous obtiendrez:
julia> @time sleep(2)
2.005766 seconds (13 allocations: 624 bytes)
Mais avec la macro, vous obtenez:
julia> @time @async sleep(2)
0.000021 seconds (7 allocations: 657 bytes)
Task (waiting) @0x0000000112a65ba0
julia>
Julia permet donc au script de continuer (et à la macro @time de s’exécuter complètement) sans attendre que la tâche (dans ce cas, dormir pendant deux secondes) se termine.
La macro @sync , en revanche, "attendra que toutes les utilisations de @async , @spawn , @spawnat et @parallel @spawn dans une @spawnat @parallel soient terminées." (selon la documentation sous ?@sync ). Ainsi, on voit:
julia> @time @sync @async sleep(2)
2.002899 seconds (47 allocations: 2.986 KB)
Task (done) @0x0000000112bd2e00
Dans cet exemple simple, il ne sert à rien d'inclure une seule instance de @async et @sync . Mais, où @sync peut être utile, vous @async appliquer @async à plusieurs opérations que vous souhaitez autoriser à démarrer sans attendre que chacune soit terminée.
Par exemple, supposons que nous ayons plusieurs employés et que nous souhaitons commencer chacun d'entre eux à travailler sur une tâche simultanément, puis à récupérer les résultats de ces tâches. Une tentative initiale (mais incorrecte) pourrait être:
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)
Le problème ici est que la boucle attend que chaque opération remotecall_fetch() se termine, c'est-à-dire que chaque processus termine son travail (dans ce cas, veille pendant 2 secondes) avant de continuer à lancer l'opération remotecall_fetch() . En termes de situation pratique, nous n'obtenons pas les avantages du parallélisme, car nos processus ne font pas leur travail (c.-à-d. Dormir) simultanément.
Nous pouvons corriger cela en utilisant une combinaison des macros @async et @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)
Maintenant, si nous comptons chaque étape de la boucle comme une opération distincte, nous voyons qu'il y a deux opérations distinctes précédées par la macro @async . La macro permet à chacun d'entre eux de démarrer, et le code à continuer (dans ce cas à l'étape suivante de la boucle) avant que chacun ne se termine. Mais, l'utilisation de la macro @sync , dont la portée englobe la boucle entière, signifie que nous ne permettrons pas au script de passer au-delà de cette boucle tant que toutes les opérations précédées de @async sont pas terminées.
Il est possible d'obtenir une compréhension encore plus claire du fonctionnement de ces macros en peaufinant l'exemple ci-dessus pour voir comment il change sous certaines modifications. Par exemple, supposons que nous ayons simplement le @async sans @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)
Ici, la macro @async nous permet de continuer dans notre boucle avant même que chaque opération remotecall_fetch() ne se termine. Mais, pour le meilleur ou pour le pire, nous n'avons pas de macro @sync pour empêcher le code de continuer au-delà de cette boucle jusqu'à ce que toutes les opérations remotecall_fetch() terminées.
Néanmoins, chaque opération remotecall_fetch() est toujours en cours d'exécution en parallèle, même une fois que nous continuons. Nous pouvons voir cela parce que si nous attendons deux secondes, le tableau a, contenant les résultats, contiendra:
sleep(2)
julia> a
2-element Array{Any,1}:
nothing
nothing
(L'élément "nothing" est le résultat d'une récupération réussie des résultats de la fonction sleep, qui ne renvoie aucune valeur)
Nous pouvons également voir que les deux opérations remotecall_fetch() démarrent essentiellement au même moment car les commandes d’ print qui les précèdent s’exécutent également rapidement (sorties de ces commandes non affichées ici). Comparez cela avec l'exemple suivant où les commandes d' print s'exécutent à un intervalle de 2 secondes:
Si nous plaçons la macro @async sur toute la boucle (au lieu de simplement l’étape interne), notre script continuera à nouveau sans attendre la remotecall_fetch() opérations remotecall_fetch() . Maintenant, cependant, nous autorisons uniquement le script à continuer au-delà de la boucle dans son ensemble. Nous ne permettons pas à chaque étape de la boucle de démarrer avant la précédente. En tant que tel, contrairement à l'exemple ci-dessus, deux secondes après le #undef du script après la boucle, le tableau de results contient toujours un élément comme #undef indiquant que la seconde opération remotecall_fetch() n'est toujours pas terminée.
@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
Et, sans surprise, si nous plaçons @sync et @async juste à côté l'un de l'autre, nous obtenons que chaque remotecall_fetch() s'exécute séquentiellement (plutôt que simultanément) mais nous ne continuons pas dans le code tant que chacun n'est pas terminé. En d'autres termes, ce serait essentiellement l'équivalent de si aucune macro n'était en place, tout comme sleep(2) se comporte essentiellement de manière identique à @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
Notez également qu'il est possible d'avoir des opérations plus compliquées dans la portée de la macro @async . La documentation donne un exemple contenant une boucle entière dans le cadre de @async .
Rappelez-vous que l'aide pour les macros de synchronisation indique qu'elle "Attend que toutes les utilisations enfermées dynamiquement de @async , @spawn , @spawnat et @parallel soient complètes." Pour ce qui est considéré comme "complet", il est important de définir les tâches dans le cadre des macros @sync et @async . Considérons l'exemple ci-dessous, qui est une légère variation sur l'un des exemples donnés ci-dessus:
@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'exemple précédent a mis environ 2 secondes à s'exécuter, indiquant que les deux tâches étaient exécutées en parallèle et que le script attendait que chacune exécute ses fonctions avant de continuer. Cet exemple, cependant, a une évaluation beaucoup moins longue. La raison en est que, pour les besoins de @sync l'opération remotecall() a "fini" une fois que le travail a été envoyé au travailleur. (Notez que le tableau résultant, a, ici, ne contient RemoteRef types d'objets RemoteRef , qui indiquent simplement qu'il se passe quelque chose avec un processus particulier qui pourrait en théorie être récupéré à un moment donné). En revanche, l'opération remotecall_fetch() n'a que "fini" lorsqu'elle reçoit le message de l'agent indiquant que sa tâche est terminée.
Ainsi, si vous cherchez des moyens de vous assurer que certaines opérations avec les travailleurs sont terminées avant de passer à votre script (comme cela est discuté dans cet article ), il est nécessaire de bien réfléchir à ce qui compte comme "complet" et comment vous le ferez. mesurez puis opérationnalisez cela dans votre script.
Ajouter des travailleurs
Lorsque vous démarrez Julia pour la première fois, par défaut, il n'y aura qu'un seul processus en cours d'exécution et disponible pour donner du travail. Vous pouvez vérifier cela en utilisant:
julia> nprocs()
1
Afin de tirer parti du traitement parallèle, vous devez d’abord ajouter des travailleurs supplémentaires qui seront alors disponibles pour effectuer le travail que vous leur assignez. Vous pouvez le faire dans votre script (ou à partir de l'interpréteur) en utilisant: addprocs(n) où n est le nombre de processus que vous souhaitez utiliser.
Alternativement, vous pouvez ajouter des processus lorsque vous démarrez Julia à partir de la ligne de commande en utilisant:
$ julia -p n
où n est le nombre de processus supplémentaires à ajouter. Donc, si on commence Julia avec
$ julia -p 2
Quand Julia commencera, nous aurons:
julia> nprocs()
3