Multi-Threading

Threads.@threads [schedule] for ... end

Une macro pour exécuter une boucle for en parallèle. L'espace d'itération est distribué à des tâches à grain grossier. Cette politique peut être spécifiée par l'argument schedule. L'exécution de la boucle attend l'évaluation de toutes les itérations.

Voir aussi : @spawn et pmap dans Distributed.

Aide étendue

Sémantique

À moins que des garanties plus fortes ne soient spécifiées par l'option de planification, la boucle exécutée par la macro @threads a les sémantiques suivantes.

La macro @threads exécute le corps de la boucle dans un ordre non spécifié et potentiellement de manière concurrente. Elle ne spécifie pas les affectations exactes des tâches et des threads de travail. Les affectations peuvent être différentes pour chaque exécution. Le code du corps de la boucle (y compris tout code appelé de manière transitive à partir de celui-ci) ne doit pas faire d'hypothèses sur la distribution des itérations aux tâches ou sur le thread de travail dans lequel elles sont exécutées. Le corps de la boucle pour chaque itération doit être capable de progresser indépendamment des autres itérations et être exempt de courses de données. En tant que tel, des synchronisations invalides entre les itérations peuvent entraîner un blocage, tandis que des accès mémoire non synchronisés peuvent entraîner un comportement indéfini.

Par exemple, les conditions ci-dessus impliquent que :

• Un verrou pris dans une itération doit être libéré dans la même itération.
• Communiquer entre les itérations en utilisant des primitives de blocage comme les Channels est incorrect.
• Écrire uniquement dans des emplacements non partagés entre les itérations (à moins qu'un verrou ou une opération atomique ne soit utilisée).
• À moins que le plan :static ne soit utilisé, la valeur de threadid() peut changer même au sein d'une seule itération. Voir Migration de tâche.

Planificateurs

Sans l'argument de planification, la planification exacte n'est pas spécifiée et varie selon les versions de Julia. Actuellement, :dynamic est utilisé lorsque le planificateur n'est pas spécifié.

:dynamic (par défaut)

Le planificateur :dynamic exécute les itérations dynamiquement sur les threads de travail disponibles. L'implémentation actuelle suppose que la charge de travail pour chaque itération est uniforme. Cependant, cette hypothèse peut être supprimée à l'avenir.

Cette option de planification est simplement un indice pour le mécanisme d'exécution sous-jacent. Cependant, quelques propriétés peuvent être attendues. Le nombre de Tasks utilisées par le planificateur :dynamic est limité par un petit multiple constant du nombre de threads de travail disponibles (Threads.threadpoolsize()). Chaque tâche traite des régions contiguës de l'espace d'itération. Ainsi, @threads :dynamic for x in xs; f(x); end est généralement plus efficace que @sync for x in xs; @spawn f(x); end si length(xs) est significativement plus grand que le nombre de threads de travail et que le temps d'exécution de f(x) est relativement plus petit que le coût de création et de synchronisation d'une tâche (généralement moins de 10 microsecondes).

:greedy

Le planificateur :greedy crée jusqu'à Threads.threadpoolsize() tâches, chacune travaillant de manière avide sur les valeurs itérées données au fur et à mesure qu'elles sont produites. Dès qu'une tâche termine son travail, elle prend la valeur suivante de l'itérateur. Le travail effectué par une tâche individuelle n'est pas nécessairement sur des valeurs contiguës de l'itérateur. L'itérateur donné peut produire des valeurs indéfiniment, seule l'interface de l'itérateur est requise (pas d'indexation).

Cette option de planification est généralement un bon choix si la charge de travail des itérations individuelles n'est pas uniforme/a une grande dispersion.

:static

Le planificateur :static crée une tâche par thread et divise les itérations également entre elles, assignant chaque tâche spécifiquement à chaque thread. En particulier, la valeur de threadid() est garantie d'être constante au sein d'une itération. Spécifier :static est une erreur si utilisé à l'intérieur d'une autre boucle @threads ou depuis un thread autre que 1.

Exemples

Pour illustrer les différentes stratégies de planification, considérons la fonction suivante busywait contenant une boucle temporelle non cédante qui s'exécute pendant un certain nombre de secondes.

julia> function busywait(seconds)
            tstart = time_ns()
            while (time_ns() - tstart) / 1e9 < seconds
            end
        end

julia> @time begin
            Threads.@spawn busywait(5)
            Threads.@threads :static for i in 1:Threads.threadpoolsize()
                busywait(1)
            end
        end
6.003001 secondes (16.33 k allocations: 899.255 KiB, 0.25% temps de compilation)

julia> @time begin
            Threads.@spawn busywait(5)
            Threads.@threads :dynamic for i in 1:Threads.threadpoolsize()
                busywait(1)
            end
        end
2.012056 secondes (16.05 k allocations: 883.919 KiB, 0.66% temps de compilation)

L'exemple :dynamic prend 2 secondes car l'un des threads non occupés est capable d'exécuter deux des itérations de 1 seconde pour compléter la boucle for. ```

Threads.foreach(f, channel::Channel;
                schedule::Threads.AbstractSchedule=Threads.FairSchedule(),
                ntasks=Threads.threadpoolsize())

Semblable à foreach(f, channel), mais l'itération sur channel et les appels à f sont répartis sur ntasks tâches lancées par Threads.@spawn. Cette fonction attend que toutes les tâches lancées en interne soient terminées avant de retourner.

Si schedule isa FairSchedule, Threads.foreach tentera de lancer des tâches de manière à permettre à l'ordonnanceur de Julia de mieux équilibrer la charge de travail entre les threads. Cette approche a généralement un coût par élément plus élevé, mais peut mieux performer que StaticSchedule en concurrence avec d'autres charges de travail multithreadées.

Si schedule isa StaticSchedule, Threads.foreach lancera des tâches d'une manière qui entraîne un coût par élément inférieur à celui de FairSchedule, mais qui est moins propice à l'équilibrage de la charge. Cette approche peut donc être plus adaptée aux charges de travail uniformes et de petite taille, mais peut moins bien performer que FairSchedule en concurrence avec d'autres charges de travail multithreadées.

Exemples

julia> n = 20

julia> c = Channel{Int}(ch -> foreach(i -> put!(ch, i), 1:n), 1)

julia> d = Channel{Int}(n) do ch
           f = i -> put!(ch, i^2)
           Threads.foreach(f, c)
       end

julia> collect(d)
collect(d) = [1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361, 400]

Threads.@spawn [:default|:interactive] expr

Créez une Task et schedule pour l'exécuter sur n'importe quel thread disponible dans le pool de threads spécifié (:default si non spécifié). La tâche est allouée à un thread une fois qu'un devient disponible. Pour attendre que la tâche se termine, appelez wait sur le résultat de cette macro, ou appelez fetch pour attendre puis obtenir sa valeur de retour.

Des valeurs peuvent être interpolées dans @spawn via $, ce qui copie la valeur directement dans la fermeture sous-jacente construite. Cela vous permet d'insérer la valeur d'une variable, isolant le code asynchrone des changements de la valeur de la variable dans la tâche actuelle.

Exemples

julia> t() = println("Hello from ", Threads.threadid());

julia> tasks = fetch.([Threads.@spawn t() for i in 1:4]);
Hello from 1
Hello from 1
Hello from 3
Hello from 4

Threads.threadid() -> Int

Obtenez le numéro d'identification du thread d'exécution actuel. Le thread maître a l'ID 1.

Exemples

julia> Threads.threadid()
1

julia> Threads.@threads for i in 1:4
          println(Threads.threadid())
       end
4
2
5
4

Threads.maxthreadid() -> Int

Obtenez une limite inférieure sur le nombre de threads (dans tous les pools de threads) disponibles pour le processus Julia, avec des sémantiques d'acquisition atomique. Le résultat sera toujours supérieur ou égal à threadid() ainsi qu'à threadid(task) pour toute tâche que vous avez pu observer avant d'appeler maxthreadid.

Threads.nthreads(:default | :interactive) -> Int

Obtenez le nombre actuel de threads dans le pool de threads spécifié. Les threads dans :interactive ont des numéros d'identification 1:nthreads(:interactive), et les threads dans :default ont des numéros d'identification dans nthreads(:interactive) .+ (1:nthreads(:default)).

Voir aussi BLAS.get_num_threads et BLAS.set_num_threads dans la bibliothèque standard LinearAlgebra, et nprocs() dans la bibliothèque standard Distributed et Threads.maxthreadid().

Threads.threadpool(tid = threadid()) -> Symbol

Renvoie le pool de threads du thread spécifié ; soit :default, :interactive, ou :foreign.

Threads.nthreadpools() -> Int

Renvoie le nombre de pools de threads actuellement configurés.

Threads.threadpoolsize(pool::Symbol = :default) -> Int

Obtenez le nombre de threads disponibles pour le pool de threads par défaut (ou pour le pool de threads spécifié).

Voir aussi : BLAS.get_num_threads et BLAS.set_num_threads dans la bibliothèque standard LinearAlgebra, et nprocs() dans la bibliothèque standard Distributed.

Threads.ngcthreads() -> Int

Renvoie le nombre de threads GC actuellement configurés. Cela inclut à la fois les threads de marquage et les threads de balayage concurrent.

Les opérations de pointeur non sécurisées sont compatibles avec le chargement et le stockage de pointeurs déclarés avec le type _Atomic et std::atomic en C11 et C++23 respectivement. Une erreur peut être générée s'il n'y a pas de support pour le chargement atomique du type Julia T.

Voir aussi : unsafe_load, unsafe_modify!, unsafe_replace!, unsafe_store!, unsafe_swap!

@atomic var
@atomic order ex

Marquez var ou ex comme étant exécuté de manière atomique, si ex est une expression supportée. Si aucun order n'est spécifié, il par défaut à :sequentially_consistent.

@atomic a.b.x = new
@atomic a.b.x += addend
@atomic :release a.b.x = new
@atomic :acquire_release a.b.x += addend

Effectuez l'opération de stockage exprimée à droite de manière atomique et renvoyez la nouvelle valeur.

Avec =, cette opération se traduit par un appel à setproperty!(a.b, :x, new). Avec n'importe quel opérateur également, cette opération se traduit par un appel à modifyproperty!(a.b, :x, +, addend)[2].

@atomic a.b.x max arg2
@atomic a.b.x + arg2
@atomic max(a.b.x, arg2)
@atomic :acquire_release max(a.b.x, arg2)
@atomic :acquire_release a.b.x + arg2
@atomic :acquire_release a.b.x max arg2

Effectuez l'opération binaire exprimée à droite de manière atomique. Stockez le résultat dans le champ du premier argument et renvoyez les valeurs (old, new).

Cette opération se traduit par un appel à modifyproperty!(a.b, :x, func, arg2).

Voir la section Per-field atomics dans le manuel pour plus de détails.

Exemples

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomic a.x # récupère le champ x de a, avec une cohérence séquentielle
1

julia> @atomic :sequentially_consistent a.x = 2 # définit le champ x de a, avec une cohérence séquentielle
2

julia> @atomic a.x += 1 # incrémente le champ x de a, avec une cohérence séquentielle
3

julia> @atomic a.x + 1 # incrémente le champ x de a, avec une cohérence séquentielle
3 => 4

julia> @atomic a.x # récupère le champ x de a, avec une cohérence séquentielle
4

julia> @atomic max(a.x, 10) # change le champ x de a à la valeur maximale, avec une cohérence séquentielle
4 => 10

julia> @atomic a.x max 5 # change à nouveau le champ x de a à la valeur maximale, avec une cohérence séquentielle
10 => 10

@atomicswap a.b.x = new
@atomicswap :sequentially_consistent a.b.x = new

Stocke new dans a.b.x et renvoie l'ancienne valeur de a.b.x.

Cette opération se traduit par un appel à swapproperty!(a.b, :x, new).

Voir la section Per-field atomics dans le manuel pour plus de détails.

Exemples

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomicswap a.x = 2+2 # remplacer le champ x de a par 4, avec cohérence séquentielle
1

julia> @atomic a.x # récupérer le champ x de a, avec cohérence séquentielle
4

@atomicreplace a.b.x attendu => désiré
@atomicreplace :sequentially_consistent a.b.x attendu => désiré
@atomicreplace :sequentially_consistent :monotonic a.b.x attendu => désiré

Effectuez le remplacement conditionnel exprimé par la paire de manière atomique, en retournant les valeurs (ancien, succès::Bool). Où succès indique si le remplacement a été effectué.

Cette opération se traduit par un appel à replaceproperty!(a.b, :x, attendu, désiré).

Voir la section Per-field atomics dans le manuel pour plus de détails.

Exemples

julia> mutable struct Atomic{T}; @atomic x::T; end

julia> a = Atomic(1)
Atomic{Int64}(1)

julia> @atomicreplace a.x 1 => 2 # remplacer le champ x de a par 2 s'il était 1, avec cohérence séquentielle
(ancien = 1, succès = true)

julia> @atomic a.x # récupérer le champ x de a, avec cohérence séquentielle
2

julia> @atomicreplace a.x 1 => 2 # remplacer le champ x de a par 2 s'il était 1, avec cohérence séquentielle
(ancien = 2, succès = false)

julia> xchg = 2 => 0; # remplacer le champ x de a par 0 s'il était 2, avec cohérence séquentielle

julia> @atomicreplace a.x xchg
(ancien = 2, succès = true)

julia> @atomic a.x # récupérer le champ x de a, avec cohérence séquentielle
0

@atomiconce a.b.x = valeur
@atomiconce :sequentially_consistent a.b.x = valeur
@atomiconce :sequentially_consistent :monotonic a.b.x = valeur

Effectuez l'assignation conditionnelle de valeur de manière atomique si elle n'était pas précédemment définie, en retournant la valeur `success::Bool`. Où `success` indique si l'assignation a été complétée.

Cette opération se traduit par un appel à `setpropertyonce!(a.b, :x, valeur)`.

Voir la section [Per-field atomics](@ref man-atomics) dans le manuel pour plus de détails.

# Exemples

jldoctest julia> mutable struct AtomicOnce @atomic x AtomicOnce() = new() end

julia> a = AtomicOnce() AtomicOnce(#undef)

julia> @atomiconce a.x = 1 # définir le champ x de a à 1, si non défini, avec cohérence séquentielle true

julia> @atomic a.x # récupérer le champ x de a, avec cohérence séquentielle 1

julia> @atomiconce a.x = 1 # définir le champ x de a à 1, si non défini, avec cohérence séquentielle false

!!! compat "Julia 1.11"
    Cette fonctionnalité nécessite au moins Julia 1.11.

AtomicMemory{T} == GenericMemory{:atomic, T, Core.CPU}

Vecteur dense de taille fixe DenseVector{T}. L'accès à n'importe quel de ses éléments se fait de manière atomique (avec un ordre :monotonic). La définition de n'importe lequel des éléments doit être réalisée en utilisant le macro @atomic et en spécifiant explicitement l'ordre.

Pour plus de détails, voir Opérations atomiques

Threads.Atomic{T}

Contient une référence à un objet de type T, garantissant qu'il n'est accédé qu'atomiquement, c'est-à-dire de manière sécurisée pour les threads.

Seuls certains types "simples" peuvent être utilisés de manière atomique, à savoir les types primitifs booléen, entier et à virgule flottante. Ceux-ci sont Bool, Int8...Int128, UInt8...UInt128, et Float16...Float64.

De nouveaux objets atomiques peuvent être créés à partir de valeurs non atomiques ; si aucune n'est spécifiée, l'objet atomique est initialisé à zéro.

Les objets atomiques peuvent être accédés en utilisant la notation [] :

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> x[] = 1
1

julia> x[]
1

Les opérations atomiques utilisent un préfixe atomic_, tel que atomic_add!, atomic_xchg!, etc.

Threads.atomic_cas!(x::Atomic{T}, cmp::T, newval::T) where T

Compare et définit x de manière atomique

Compare atomiquement la valeur dans x avec cmp. Si elles sont égales, écrit newval dans x. Sinon, laisse x non modifié. Renvoie l'ancienne valeur dans x. En comparant la valeur renvoyée à cmp (via ===), on sait si x a été modifié et contient maintenant la nouvelle valeur newval.

Pour plus de détails, voir l'instruction cmpxchg de LLVM.

Cette fonction peut être utilisée pour implémenter des sémantiques transactionnelles. Avant la transaction, on enregistre la valeur dans x. Après la transaction, la nouvelle valeur n'est stockée que si x n'a pas été modifié entre-temps.

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_cas!(x, 4, 2);

julia> x
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_cas!(x, 3, 2);

julia> x
Base.Threads.Atomic{Int64}(2)

Threads.atomic_xchg!(x::Atomic{T}, newval::T) where T

Échange atomiquement la valeur dans x

Échange atomiquement la valeur dans x avec newval. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw xchg de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_xchg!(x, 2)
3

julia> x[]
2

Threads.atomic_add!(x::Atomic{T}, val::T) where T <: ArithmeticTypes

Ajoute atomiquement val à x

Effectue x[] += val de manière atomique. Renvoie la ancienne valeur. Non défini pour Atomic{Bool}.

Pour plus de détails, voir l'instruction atomicrmw add de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_add!(x, 2)
3

julia> x[]
5

Threads.atomic_sub!(x::Atomic{T}, val::T) where T <: ArithmeticTypes

Soustrait atomiquement val de x

Effectue x[] -= val de manière atomique. Renvoie la ancienne valeur. Non défini pour Atomic{Bool}.

Pour plus de détails, voir l'instruction atomicrmw sub de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_sub!(x, 2)
3

julia> x[]
1

Threads.atomic_and!(x::Atomic{T}, val::T) where T

Effectue un ET bit à bit atomique de x avec val

Effectue x[] &= val de manière atomique. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw and de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_and!(x, 2)
3

julia> x[]
2

Threads.atomic_nand!(x::Atomic{T}, val::T) où T

Effectue un nand (non-et) bit à bit atomique de x avec val

Effectue x[] = ~(x[] & val) de manière atomique. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw nand de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(3)
Base.Threads.Atomic{Int64}(3)

julia> Threads.atomic_nand!(x, 2)
3

julia> x[]
-3

Threads.atomic_or!(x::Atomic{T}, val::T) where T

Effectue un ou bit à bit atomique de x avec val

Effectue x[] |= val de manière atomique. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw or de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(5)
Base.Threads.Atomic{Int64}(5)

julia> Threads.atomic_or!(x, 7)
5

julia> x[]
7

Threads.atomic_xor!(x::Atomic{T}, val::T) where T

Effectue un xor (ou exclusif) au niveau des bits de manière atomique entre x et val.

Effectue x[] $= val de manière atomique. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw xor de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(5)
Base.Threads.Atomic{Int64}(5)

julia> Threads.atomic_xor!(x, 7)
5

julia> x[]
2

Threads.atomic_max!(x::Atomic{T}, val::T) where T

Stocke de manière atomique le maximum de x et val dans x

Effectue x[] = max(x[], val) de manière atomique. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw max de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(5)
Base.Threads.Atomic{Int64}(5)

julia> Threads.atomic_max!(x, 7)
5

julia> x[]
7

Threads.atomic_min!(x::Atomic{T}, val::T) where T

Stocke de manière atomique le minimum de x et val dans x

Effectue x[] = min(x[], val) de manière atomique. Renvoie la ancienne valeur.

Pour plus de détails, voir l'instruction atomicrmw min de LLVM.

Exemples

julia> x = Threads.Atomic{Int}(7)
Base.Threads.Atomic{Int64}(7)

julia> Threads.atomic_min!(x, 5)
7

julia> x[]
5

Threads.atomic_fence()

Insérer une barrière de mémoire à cohérence séquentielle

Insère une barrière de mémoire avec des sémantiques d'ordre à cohérence séquentielle. Il existe des algorithmes où cela est nécessaire, c'est-à-dire où un ordre d'acquisition/libération est insuffisant.

C'est probablement une opération très coûteuse. Étant donné que toutes les autres opérations atomiques en Julia ont déjà des sémantiques d'acquisition/libération, des barrières explicites ne devraient pas être nécessaires dans la plupart des cas.

Pour plus de détails, voir l'instruction fence de LLVM.

@threadcall((cfunc, clib), rettype, (argtypes...), argvals...)

Le macro @threadcall est appelé de la même manière que ccall mais effectue le travail dans un thread différent. Cela est utile lorsque vous souhaitez appeler une fonction C bloquante sans provoquer le blocage du thread julia actuel. La concurrence est limitée par la taille du pool de threads libuv, qui par défaut est de 4 threads mais peut être augmentée en définissant la variable d'environnement UV_THREADPOOL_SIZE et en redémarrant le processus julia.

Notez que la fonction appelée ne doit jamais rappeler Julia.

SpinLock()

Créez un verrou à spin non réentré, test-et-teste-et-établit. Une utilisation récursive entraînera un blocage. Ce type de verrou ne doit être utilisé que autour de code qui prend peu de temps à s'exécuter et ne bloque pas (par exemple, effectuer des E/S). En général, ReentrantLock devrait être utilisé à la place.

Chaque lock doit être associé à un unlock. Si !islocked(lck::SpinLock) est vrai, trylock(lck) réussit à moins qu'il n'y ait d'autres tâches tentant de détenir le verrou "en même temps".

Les verrous à spin test-et-teste-et-établit sont les plus rapides jusqu'à environ 30 threads en concurrence. Si vous avez plus de concurrence que cela, d'autres approches de synchronisation devraient être envisagées.