Multi-Threading

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

Una macro para ejecutar un bucle for en paralelo. El espacio de iteración se distribuye a tareas de gran grano. Esta política se puede especificar mediante el argumento schedule. La ejecución del bucle espera la evaluación de todas las iteraciones.

Véase también: @spawn y pmap en Distributed.

Ayuda extendida

Semántica

A menos que se especifiquen garantías más fuertes mediante la opción de programación, el bucle ejecutado por la macro @threads tiene la siguiente semántica.

La macro @threads ejecuta el cuerpo del bucle en un orden no especificado y potencialmente de manera concurrente. No especifica las asignaciones exactas de las tareas y los hilos de trabajo. Las asignaciones pueden ser diferentes para cada ejecución. El código del cuerpo del bucle (incluido cualquier código llamado de manera transitiva desde él) no debe hacer suposiciones sobre la distribución de iteraciones a tareas o el hilo de trabajo en el que se ejecutan. El cuerpo del bucle para cada iteración debe ser capaz de avanzar de manera independiente de otras iteraciones y estar libre de condiciones de carrera. Como tal, las sincronizaciones inválidas entre iteraciones pueden provocar un bloqueo, mientras que los accesos a memoria no sincronizados pueden resultar en un comportamiento indefinido.

Por ejemplo, las condiciones anteriores implican que:

• Un bloqueo tomado en una iteración debe ser liberado dentro de la misma iteración.
• Comunicar entre iteraciones utilizando primitivas de bloqueo como Channels es incorrecto.
• Escribir solo en ubicaciones no compartidas entre iteraciones (a menos que se use un bloqueo o una operación atómica).
• A menos que se use la programación :static, el valor de threadid() puede cambiar incluso dentro de una sola iteración. Véase Migración de Tareas.

Programadores

Sin el argumento del programador, la programación exacta no está especificada y varía entre las versiones de Julia. Actualmente, se utiliza :dynamic cuando no se especifica el programador.

:dynamic (predeterminado)

El programador :dynamic ejecuta iteraciones dinámicamente en los hilos de trabajo disponibles. La implementación actual asume que la carga de trabajo para cada iteración es uniforme. Sin embargo, esta suposición puede eliminarse en el futuro.

Esta opción de programación es meramente una sugerencia para el mecanismo de ejecución subyacente. Sin embargo, se pueden esperar algunas propiedades. El número de Tasks utilizados por el programador :dynamic está limitado por un pequeño múltiplo constante del número de hilos de trabajo disponibles (Threads.threadpoolsize()). Cada tarea procesa regiones contiguas del espacio de iteración. Por lo tanto, @threads :dynamic for x in xs; f(x); end es típicamente más eficiente que @sync for x in xs; @spawn f(x); end si length(xs) es significativamente mayor que el número de hilos de trabajo y el tiempo de ejecución de f(x) es relativamente menor que el costo de crear y sincronizar una tarea (típicamente menos de 10 microsegundos).

:greedy

El programador :greedy genera hasta Threads.threadpoolsize() tareas, cada una trabajando de manera codiciosa en los valores iterados dados a medida que se producen. Tan pronto como una tarea termina su trabajo, toma el siguiente valor del iterador. El trabajo realizado por cualquier tarea individual no necesariamente se realiza en valores contiguos del iterador. El iterador dado puede producir valores indefinidamente, solo se requiere la interfaz del iterador (sin indexación).

Esta opción de programación es generalmente una buena elección si la carga de trabajo de las iteraciones individuales no es uniforme/tiene una gran dispersión.

:static

El programador :static crea una tarea por hilo y divide las iteraciones de manera equitativa entre ellas, asignando cada tarea específicamente a cada hilo. En particular, el valor de threadid() está garantizado que sea constante dentro de una iteración. Es un error especificar :static si se usa desde dentro de otro bucle @threads o desde un hilo diferente al 1.

Ejemplos

Para ilustrar las diferentes estrategias de programación, considere la siguiente función busywait que contiene un bucle de tiempo que no cede y que se ejecuta durante un número dado de segundos.

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 seconds (16.33 k allocations: 899.255 KiB, 0.25% compilation time)

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

El ejemplo :dynamic toma 2 segundos ya que uno de los hilos no ocupados puede ejecutar dos de las iteraciones de 1 segundo para completar el bucle for. ```

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

Similar a foreach(f, channel), pero la iteración sobre channel y las llamadas a f se dividen entre ntasks tareas generadas por Threads.@spawn. Esta función esperará a que todas las tareas generadas internamente se completen antes de devolver.

Si schedule isa FairSchedule, Threads.foreach intentará generar tareas de una manera que permita al programador de Julia equilibrar más libremente la carga de trabajo entre los hilos. Este enfoque generalmente tiene un mayor costo por elemento, pero puede funcionar mejor que StaticSchedule en concurrencia con otras cargas de trabajo multihilo.

Si schedule isa StaticSchedule, Threads.foreach generará tareas de una manera que incurre en un menor costo por elemento que FairSchedule, pero es menos adecuada para el equilibrio de carga. Este enfoque puede ser más adecuado para cargas de trabajo uniformes y de grano fino, pero puede funcionar peor que FairSchedule en concurrencia con otras cargas de trabajo multihilo.

Ejemplos

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

Crea una Task y la schedule para que se ejecute en cualquier hilo disponible en el grupo de hilos especificado (:default si no se especifica). La tarea se asigna a un hilo una vez que uno se vuelve disponible. Para esperar a que la tarea termine, llama a wait sobre el resultado de este macro, o llama a fetch para esperar y luego obtener su valor de retorno.

Los valores se pueden interpolar en @spawn a través de $, que copia el valor directamente en el cierre subyacente construido. Esto te permite insertar el valor de una variable, aislando el código asíncrono de los cambios en el valor de la variable en la tarea actual.

Ejemplos

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

Obtiene el número de ID del hilo de ejecución actual. El hilo maestro tiene ID 1.

Ejemplos

julia> Threads.threadid()
1

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

Threads.maxthreadid() -> Int

Obtiene un límite inferior en el número de hilos (a través de todos los grupos de hilos) disponibles para el proceso de Julia, con semántica de adquisición atómica. El resultado siempre será mayor o igual a threadid() así como threadid(task) para cualquier tarea que pudiste observar antes de llamar a maxthreadid.

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

Obtiene el número actual de hilos dentro del grupo de hilos especificado. Los hilos en :interactive tienen números de identificación 1:nthreads(:interactive), y los hilos en :default tienen números de identificación en nthreads(:interactive) .+ (1:nthreads(:default)).

Véase también BLAS.get_num_threads y BLAS.set_num_threads en la biblioteca estándar LinearAlgebra, y nprocs() en la biblioteca estándar Distributed y Threads.maxthreadid().

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

Devuelve el grupo de hilos del hilo especificado; puede ser :default, :interactive o :foreign.

Threads.nthreadpools() -> Int

Devuelve el número de grupos de hilos actualmente configurados.

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

Obtiene el número de hilos disponibles en el grupo de hilos predeterminado (o en el grupo de hilos especificado).

Véase también: BLAS.get_num_threads y BLAS.set_num_threads en la biblioteca estándar LinearAlgebra, y nprocs() en la biblioteca estándar Distributed.

Threads.ngcthreads() -> Int

Devuelve el número de hilos de GC actualmente configurados. Esto incluye tanto los hilos de marcado como los hilos de barrido concurrente.

Las operaciones de punteros inseguros son compatibles con la carga y almacenamiento de punteros declarados con el tipo _Atomic y std::atomic en C11 y C++23 respectivamente. Se puede lanzar un error si no hay soporte para cargar atómicamente el tipo de Julia T.

Véase también: unsafe_load, unsafe_modify!, unsafe_replace!, unsafe_store!, unsafe_swap!

@atomic var
@atomic order ex

Marque var o ex como si se estuviera realizando de manera atómica, si ex es una expresión admitida. Si no se especifica order, por defecto es :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

Realice la operación de almacenamiento expresada a la derecha de manera atómica y devuelva el nuevo valor.

Con =, esta operación se traduce en una llamada a setproperty!(a.b, :x, new). Con cualquier operador también, esta operación se traduce en una llamada a 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

Realice la operación binaria expresada a la derecha de manera atómica. Almacene el resultado en el campo del primer argumento y devuelva los valores (old, new).

Esta operación se traduce en una llamada a modifyproperty!(a.b, :x, func, arg2).

Consulte la sección Per-field atomics en el manual para más detalles.

Ejemplos

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

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

julia> @atomic a.x # obtener el campo x de a, con consistencia secuencial
1

julia> @atomic :sequentially_consistent a.x = 2 # establecer el campo x de a, con consistencia secuencial
2

julia> @atomic a.x += 1 # incrementar el campo x de a, con consistencia secuencial
3

julia> @atomic a.x + 1 # incrementar el campo x de a, con consistencia secuencial
3 => 4

julia> @atomic a.x # obtener el campo x de a, con consistencia secuencial
4

julia> @atomic max(a.x, 10) # cambiar el campo x de a al valor máximo, con consistencia secuencial
4 => 10

julia> @atomic a.x max 5 # nuevamente cambiar el campo x de a al valor máximo, con consistencia secuencial
10 => 10

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

Almacena new en a.b.x y devuelve el valor antiguo de a.b.x.

Esta operación se traduce en una llamada a swapproperty!(a.b, :x, new).

Consulta la sección Per-field atomics en el manual para más detalles.

Ejemplos

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

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

julia> @atomicswap a.x = 2+2 # reemplaza el campo x de a con 4, con consistencia secuencial
1

julia> @atomic a.x # obtiene el campo x de a, con consistencia secuencial
4

@atomicreplace a.b.x esperado => deseado
@atomicreplace :sequentially_consistent a.b.x esperado => deseado
@atomicreplace :sequentially_consistent :monotonic a.b.x esperado => deseado

Realiza el reemplazo condicional expresado por el par de manera atómica, devolviendo los valores (old, success::Bool). Donde success indica si el reemplazo se completó.

Esta operación se traduce en una llamada a replaceproperty!(a.b, :x, esperado, deseado).

Consulta la sección Per-field atomics en el manual para más detalles.

Ejemplos

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

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

julia> @atomicreplace a.x 1 => 2 # reemplaza el campo x de a con 2 si era 1, con consistencia secuencial
(old = 1, success = true)

julia> @atomic a.x # obtiene el campo x de a, con consistencia secuencial
2

julia> @atomicreplace a.x 1 => 2 # reemplaza el campo x de a con 2 si era 1, con consistencia secuencial
(old = 2, success = false)

julia> xchg = 2 => 0; # reemplaza el campo x de a con 0 si era 2, con consistencia secuencial

julia> @atomicreplace a.x xchg
(old = 2, success = true)

julia> @atomic a.x # obtiene el campo x de a, con consistencia secuencial
0

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

Realiza la asignación condicional de value de manera atómica si anteriormente no estaba establecido, devolviendo el valor success::Bool. Donde success indica si la asignación se completó.

Esta operación se traduce en una llamada a setpropertyonce!(a.b, :x, value).

Consulta la sección Per-field atomics en el manual para más detalles.

Ejemplos

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

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

julia> @atomiconce a.x = 1 # establece el campo x de a en 1, si no está establecido, con consistencia secuencial
true

julia> @atomic a.x # obtiene el campo x de a, con consistencia secuencial
1

julia> @atomiconce a.x = 1 # establece el campo x de a en 1, si no está establecido, con consistencia secuencial
false

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

Vector denso de tamaño fijo DenseVector{T}. El acceso a cualquiera de sus elementos se realiza de manera atómica (con ordenamiento :monotonic). Establecer cualquiera de los elementos debe llevarse a cabo utilizando el macro @atomic y especificando explícitamente el ordenamiento.

Para más detalles, consulta Operaciones Atómicas

Threads.Atomic{T}

Mantiene una referencia a un objeto del tipo T, asegurando que solo se acceda a él de manera atómica, es decir, de forma segura para hilos.

Solo ciertos tipos "simples" pueden usarse de manera atómica, a saber, los tipos primitivos booleanos, enteros y de punto flotante. Estos son Bool, Int8...Int128, UInt8...UInt128, y Float16...Float64.

Se pueden crear nuevos objetos atómicos a partir de valores no atómicos; si no se especifica ninguno, el objeto atómico se inicializa con cero.

Los objetos atómicos se pueden acceder utilizando la notación []:

Ejemplos

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

julia> x[] = 1
1

julia> x[]
1

Las operaciones atómicas utilizan un prefijo atomic_, como atomic_add!, atomic_xchg!, etc.

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

Compara y establece x de manera atómica

Compara de manera atómica el valor en x con cmp. Si son iguales, escribe newval en x. De lo contrario, deja x sin modificar. Devuelve el valor antiguo en x. Al comparar el valor devuelto con cmp (a través de ===), se sabe si x fue modificado y ahora tiene el nuevo valor newval.

Para más detalles, consulta la instrucción cmpxchg de LLVM.

Esta función se puede utilizar para implementar semánticas transaccionales. Antes de la transacción, se registra el valor en x. Después de la transacción, el nuevo valor se almacena solo si x no ha sido modificado en el ínterin.

Ejemplos

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

Intercambia atómicamente el valor en x

Intercambia atómicamente el valor en x con newval. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw xchg de LLVM.

Ejemplos

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) donde T <: ArithmeticTypes

Suma val a x de manera atómica

Realiza x[] += val de manera atómica. Devuelve el valor antiguo. No está definido para Atomic{Bool}.

Para más detalles, consulta la instrucción atomicrmw add de LLVM.

Ejemplos

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) donde T <: ArithmeticTypes

Resta val de x de manera atómica.

Realiza x[] -= val de manera atómica. Devuelve el valor antiguo. No está definido para Atomic{Bool}.

Para más detalles, consulta la instrucción atomicrmw sub de LLVM.

Ejemplos

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

Realiza una operación atómica de AND a nivel de bits entre x y val

Realiza x[] &= val de manera atómica. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw and de LLVM.

Ejemplos

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) where T

Realiza un nand (no-y) a nivel atómico de x con val

Realiza x[] = ~(x[] & val) de manera atómica. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw nand de LLVM.

Ejemplos

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) donde T

Realiza un OR a nivel de bits de manera atómica entre x y val

Realiza x[] |= val de manera atómica. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw or de LLVM.

Ejemplos

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

Realiza una operación atómica de xor a nivel de bits (o exclusivo) entre x y val.

Realiza x[] $= val de manera atómica. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw xor de LLVM.

Ejemplos

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

Almacena de manera atómica el máximo de x y val en x

Realiza x[] = max(x[], val) de manera atómica. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw max de LLVM.

Ejemplos

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

Almacena de manera atómica el mínimo de x y val en x

Realiza x[] = min(x[], val) de manera atómica. Devuelve el valor antiguo.

Para más detalles, consulta la instrucción atomicrmw min de LLVM.

Ejemplos

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

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

julia> x[]
5

Threads.atomic_fence()

Inserta una barrera de memoria de consistencia secuencial

Inserta una barrera de memoria con semánticas de orden de consistencia secuencial. Hay algoritmos donde esto es necesario, es decir, donde un orden de adquisición/liberación es insuficiente.

Es probable que esta sea una operación muy costosa. Dado que todas las demás operaciones atómicas en Julia ya tienen semánticas de adquisición/liberación, las barreras explícitas no deberían ser necesarias en la mayoría de los casos.

Para más detalles, consulta la instrucción fence de LLVM.

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

El macro @threadcall se llama de la misma manera que ccall pero realiza el trabajo en un hilo diferente. Esto es útil cuando deseas llamar a una función C bloqueante sin causar que el hilo actual de julia se bloquee. La concurrencia está limitada por el tamaño del grupo de hilos de libuv, que por defecto es de 4 hilos, pero se puede aumentar configurando la variable de entorno UV_THREADPOOL_SIZE y reiniciando el proceso de julia.

Ten en cuenta que la función llamada nunca debe volver a llamar a Julia.

SpinLock()

Crea un bloqueo de giro no reentrante, de prueba-y-prueba-y-establecer. El uso recursivo resultará en un bloqueo. Este tipo de bloqueo solo debe usarse alrededor de código que toma poco tiempo para ejecutarse y no bloquea (por ejemplo, realizar E/S). En general, se debe usar ReentrantLock en su lugar.

Cada lock debe coincidir con un unlock. Si !islocked(lck::SpinLock) se cumple, trylock(lck) tiene éxito a menos que haya otras tareas intentando mantener el bloqueo "al mismo tiempo".

Los bloqueos de giro de prueba-y-prueba-y-establecer son los más rápidos hasta aproximadamente 30 hilos en contención. Si tienes más contención que eso, se deben considerar diferentes enfoques de sincronización.