Multi-Threading

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

A macro to execute a for loop in parallel. The iteration space is distributed to coarse-grained tasks. This policy can be specified by the schedule argument. The execution of the loop waits for the evaluation of all iterations.

See also: @spawn and pmap in Distributed.

Extended help

Semantics

Unless stronger guarantees are specified by the scheduling option, the loop executed by @threads macro have the following semantics.

The @threads macro executes the loop body in an unspecified order and potentially concurrently. It does not specify the exact assignments of the tasks and the worker threads. The assignments can be different for each execution. The loop body code (including any code transitively called from it) must not make any assumptions about the distribution of iterations to tasks or the worker thread in which they are executed. The loop body for each iteration must be able to make forward progress independent of other iterations and be free from data races. As such, invalid synchronizations across iterations may deadlock while unsynchronized memory accesses may result in undefined behavior.

For example, the above conditions imply that:

• A lock taken in an iteration must be released within the same iteration.
• Communicating between iterations using blocking primitives like Channels is incorrect.
• Write only to locations not shared across iterations (unless a lock or atomic operation is used).
• Unless the :static schedule is used, the value of threadid() may change even within a single iteration. See Task Migration.

Schedulers

Without the scheduler argument, the exact scheduling is unspecified and varies across Julia releases. Currently, :dynamic is used when the scheduler is not specified.

:dynamic (default)

:dynamic scheduler executes iterations dynamically to available worker threads. Current implementation assumes that the workload for each iteration is uniform. However, this assumption may be removed in the future.

This scheduling option is merely a hint to the underlying execution mechanism. However, a few properties can be expected. The number of Tasks used by :dynamic scheduler is bounded by a small constant multiple of the number of available worker threads (Threads.threadpoolsize()). Each task processes contiguous regions of the iteration space. Thus, @threads :dynamic for x in xs; f(x); end is typically more efficient than @sync for x in xs; @spawn f(x); end if length(xs) is significantly larger than the number of the worker threads and the run-time of f(x) is relatively smaller than the cost of spawning and synchronizing a task (typically less than 10 microseconds).

:greedy

:greedy scheduler spawns up to Threads.threadpoolsize() tasks, each greedily working on the given iterated values as they are produced. As soon as one task finishes its work, it takes the next value from the iterator. Work done by any individual task is not necessarily on contiguous values from the iterator. The given iterator may produce values forever, only the iterator interface is required (no indexing).

This scheduling option is generally a good choice if the workload of individual iterations is not uniform/has a large spread.

:static

:static scheduler creates one task per thread and divides the iterations equally among them, assigning each task specifically to each thread. In particular, the value of threadid() is guaranteed to be constant within one iteration. Specifying :static is an error if used from inside another @threads loop or from a thread other than 1.

Examples

To illustrate of the different scheduling strategies, consider the following function busywait containing a non-yielding timed loop that runs for a given number of seconds.

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)

The :dynamic example takes 2 seconds since one of the non-occupied threads is able to run two of the 1-second iterations to complete the for loop.

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

Similar to foreach(f, channel), but iteration over channel and calls to f are split across ntasks tasks spawned by Threads.@spawn. This function will wait for all internally spawned tasks to complete before returning.

If schedule isa FairSchedule, Threads.foreach will attempt to spawn tasks in a manner that enables Julia's scheduler to more freely load-balance work items across threads. This approach generally has higher per-item overhead, but may perform better than StaticSchedule in concurrence with other multithreaded workloads.

If schedule isa StaticSchedule, Threads.foreach will spawn tasks in a manner that incurs lower per-item overhead than FairSchedule, but is less amenable to load-balancing. This approach thus may be more suitable for fine-grained, uniform workloads, but may perform worse than FairSchedule in concurrence with other multithreaded workloads.

Examples

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

Create a Task and schedule it to run on any available thread in the specified threadpool (:default if unspecified). The task is allocated to a thread once one becomes available. To wait for the task to finish, call wait on the result of this macro, or call fetch to wait and then obtain its return value.

Values can be interpolated into @spawn via $, which copies the value directly into the constructed underlying closure. This allows you to insert the value of a variable, isolating the asynchronous code from changes to the variable's value in the current task.

Examples

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([t::Task]) -> Int

Get the ID number of the current thread of execution, or the thread of task t. The master thread has ID 1.

Examples

julia> Threads.threadid()
1

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

julia> Threads.threadid(Threads.@spawn "foo")
2

Threads.maxthreadid() -> Int

Get a lower bound on the number of threads (across all thread pools) available to the Julia process, with atomic-acquire semantics. The result will always be greater than or equal to threadid() as well as threadid(task) for any task you were able to observe before calling maxthreadid.

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

Get the current number of threads within the specified thread pool. The threads in :interactive have id numbers 1:nthreads(:interactive), and the threads in :default have id numbers in nthreads(:interactive) .+ (1:nthreads(:default)).

See also BLAS.get_num_threads and BLAS.set_num_threads in the LinearAlgebra standard library, and nprocs() in the Distributed standard library and Threads.maxthreadid().

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

Returns the specified thread's threadpool; either :default, :interactive, or :foreign.

Threads.nthreadpools() -> Int

Returns the number of threadpools currently configured.

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

Get the number of threads available to the default thread pool (or to the specified thread pool).

See also: BLAS.get_num_threads and BLAS.set_num_threads in the LinearAlgebra standard library, and nprocs() in the Distributed standard library.

Threads.ngcthreads() -> Int

Returns the number of GC threads currently configured. This includes both mark threads and concurrent sweep threads.

Unsafe pointer operations are compatible with loading and storing pointers declared with _Atomic and std::atomic type in C11 and C++23 respectively. An error may be thrown if there is not support for atomically loading the Julia type T.

See also: unsafe_load, unsafe_modify!, unsafe_replace!, unsafe_store!, unsafe_swap!

@atomic var
@atomic order ex

Mark var or ex as being performed atomically, if ex is a supported expression. If no order is specified it defaults to :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
@atomic m[idx] = new
@atomic m[idx] += addend
@atomic :release m[idx] = new
@atomic :acquire_release m[idx] += addend

Perform the store operation expressed on the right atomically and return the new value.

With assignment (=), this operation translates to a setproperty!(a.b, :x, new) or, in case of reference, to a setindex_atomic!(m, order, new, idx) call, with order defaulting to :sequentially_consistent.

With any modifying operator this operation translates to a modifyproperty!(a.b, :x, op, addend)[2] or, in case of reference, to a modifyindex_atomic!(m, order, op, addend, idx...)[2] call, with order defaulting to :sequentially_consistent.

@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
@atomic m[idx] max arg2
@atomic m[idx] + arg2
@atomic max(m[idx], arg2)
@atomic :acquire_release max(m[idx], arg2)
@atomic :acquire_release m[idx] + arg2
@atomic :acquire_release m[idx] max arg2

Perform the binary operation expressed on the right atomically. Store the result into the field or the reference in the first argument, and return the values (old, new).

This operation translates to a modifyproperty!(a.b, :x, func, arg2) or, in case of reference to a modifyindex_atomic!(m, order, func, arg2, idx) call, with order defaulting to :sequentially_consistent.

See Per-field atomics section in the manual for more details.

Examples

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

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

julia> @atomic a.x # fetch field x of a, with sequential consistency
1

julia> @atomic :sequentially_consistent a.x = 2 # set field x of a, with sequential consistency
2

julia> @atomic a.x += 1 # increment field x of a, with sequential consistency
3

julia> @atomic a.x + 1 # increment field x of a, with sequential consistency
3 => 4

julia> @atomic a.x # fetch field x of a, with sequential consistency
4

julia> @atomic max(a.x, 10) # change field x of a to the max value, with sequential consistency
4 => 10

julia> @atomic a.x max 5 # again change field x of a to the max value, with sequential consistency
10 => 10

julia> mem = AtomicMemory{Int}(undef, 2);

julia> @atomic mem[1] = 2 # set mem[1] to value 2 with sequential consistency
2

julia> @atomic :monotonic mem[1] # fetch the first value of mem, with monotonic consistency
2

julia> @atomic mem[1] += 1 # increment the first value of mem, with sequential consistency
3

julia> @atomic mem[1] + 1 # increment the first value of mem, with sequential consistency
3 => 4

julia> @atomic mem[1] # fetch the first value of mem, with sequential consistency
4

julia> @atomic max(mem[1], 10) # change the first value of mem to the max value, with sequential consistency
4 => 10

julia> @atomic mem[1] max 5 # again change the first value of mem to the max value, with sequential consistency
10 => 10

@atomicswap a.b.x = new
@atomicswap :sequentially_consistent a.b.x = new
@atomicswap m[idx] = new
@atomicswap :sequentially_consistent m[idx] = new

Stores new into a.b.x (m[idx] in case of reference) and returns the old value of a.b.x (the old value stored at m[idx], respectively).

This operation translates to a swapproperty!(a.b, :x, new) or, in case of reference, swapindex_atomic!(mem, order, new, idx) call, with order defaulting to :sequentially_consistent.

See Per-field atomics section in the manual for more details.

Examples

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

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

julia> @atomicswap a.x = 2+2 # replace field x of a with 4, with sequential consistency
1

julia> @atomic a.x # fetch field x of a, with sequential consistency
4

julia> mem = AtomicMemory{Int}(undef, 2);

julia> @atomic mem[1] = 1;

julia> @atomicswap mem[1] = 4 # replace the first value of `mem` with 4, with sequential consistency
1

julia> @atomic mem[1] # fetch the first value of mem, with sequential consistency
4

@atomicreplace a.b.x expected => desired
@atomicreplace :sequentially_consistent a.b.x expected => desired
@atomicreplace :sequentially_consistent :monotonic a.b.x expected => desired
@atomicreplace m[idx] expected => desired
@atomicreplace :sequentially_consistent m[idx] expected => desired
@atomicreplace :sequentially_consistent :monotonic m[idx] expected => desired

Perform the conditional replacement expressed by the pair atomically, returning the values (old, success::Bool). Where success indicates whether the replacement was completed.

This operation translates to a replaceproperty!(a.b, :x, expected, desired) or, in case of reference, to a replaceindex_atomic!(mem, success_order, fail_order, expected, desired, idx) call, with both orders defaulting to :sequentially_consistent.

See Per-field atomics section in the manual for more details.

Examples

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

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

julia> @atomicreplace a.x 1 => 2 # replace field x of a with 2 if it was 1, with sequential consistency
(old = 1, success = true)

julia> @atomic a.x # fetch field x of a, with sequential consistency
2

julia> @atomicreplace a.x 1 => 3 # replace field x of a with 2 if it was 1, with sequential consistency
(old = 2, success = false)

julia> xchg = 2 => 0; # replace field x of a with 0 if it was 2, with sequential consistency

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

julia> @atomic a.x # fetch field x of a, with sequential consistency
0

julia> mem = AtomicMemory{Int}(undef, 2);

julia> @atomic mem[1] = 1;

julia> @atomicreplace mem[1] 1 => 2 # replace the first value of mem with 2 if it was 1, with sequential consistency
(old = 1, success = true)

julia> @atomic mem[1] # fetch the first value of mem, with sequential consistency
2

julia> @atomicreplace mem[1] 1 => 3 # replace field x of a with 2 if it was 1, with sequential consistency
(old = 2, success = false)

julia> xchg = 2 => 0; # replace field x of a with 0 if it was 2, with sequential consistency

julia> @atomicreplace mem[1] xchg
(old = 2, success = true)

julia> @atomic mem[1] # fetch the first value of mem, with sequential consistency
0

@atomiconce a.b.x = value
@atomiconce :sequentially_consistent a.b.x = value
@atomiconce :sequentially_consistent :monotonic a.b.x = value
@atomiconce m[idx] = value
@atomiconce :sequentially_consistent m[idx] = value
@atomiconce :sequentially_consistent :monotonic m[idx] = value

Perform the conditional assignment of value atomically if it was previously unset. Returned value success::Bool indicates whether the assignment was completed.

This operation translates to a setpropertyonce!(a.b, :x, value) or, in case of reference, to a setindexonce_atomic!(m, success_order, fail_order, value, idx) call, with both orders defaulting to :sequentially_consistent.

See Per-field atomics section in the manual for more details.

Examples

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

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

julia> @atomiconce a.x = 1 # set field x of a to 1, if unset, with sequential consistency
true

julia> @atomic a.x # fetch field x of a, with sequential consistency
1

julia> @atomiconce :monotonic a.x = 2 # set field x of a to 1, if unset, with monotonic consistence
false

julia> mem = AtomicMemory{Vector{Int}}(undef, 1);

julia> isassigned(mem, 1)
false

julia> @atomiconce mem[1] = [1] # set the first value of mem to [1], if unset, with sequential consistency
true

julia> isassigned(mem, 1)
true

julia> @atomic mem[1] # fetch the first value of mem, with sequential consistency
1-element Vector{Int64}:
 1

julia> @atomiconce :monotonic mem[1] = [2] # set the first value of mem to [2], if unset, with monotonic
false

julia> @atomic mem[1]
1-element Vector{Int64}:
 1

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

Fixed-size DenseVector{T}. Fetching of any of its individual elements is performed atomically (with :monotonic ordering by default).

For details, see Atomic Operations as well as macros @atomic, @atomiconce, @atomicswap, and @atomicreplace.

Threads.Atomic{T}

Holds a reference to an object of type T, ensuring that it is only accessed atomically, i.e. in a thread-safe manner.

Only certain "simple" types can be used atomically, namely the primitive boolean, integer, and float-point types. These are Bool, Int8...Int128, UInt8...UInt128, and Float16...Float64.

New atomic objects can be created from a non-atomic values; if none is specified, the atomic object is initialized with zero.

Atomic objects can be accessed using the [] notation:

Examples

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

julia> x[] = 1
1

julia> x[]
1

Atomic operations use an atomic_ prefix, such as atomic_add!, atomic_xchg!, etc.

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

Atomically compare-and-set x

Atomically compares the value in x with cmp. If equal, write newval to x. Otherwise, leaves x unmodified. Returns the old value in x. By comparing the returned value to cmp (via ===) one knows whether x was modified and now holds the new value newval.

For further details, see LLVM's cmpxchg instruction.

This function can be used to implement transactional semantics. Before the transaction, one records the value in x. After the transaction, the new value is stored only if x has not been modified in the mean time.

Examples

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

Atomically exchange the value in x

Atomically exchanges the value in x with newval. Returns the old value.

For further details, see LLVM's atomicrmw xchg instruction.

Examples

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

Atomically add val to x

Performs x[] += val atomically. Returns the old value. Not defined for Atomic{Bool}.

For further details, see LLVM's atomicrmw add instruction.

Examples

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

Atomically subtract val from x

Performs x[] -= val atomically. Returns the old value. Not defined for Atomic{Bool}.

For further details, see LLVM's atomicrmw sub instruction.

Examples

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

Atomically bitwise-and x with val

Performs x[] &= val atomically. Returns the old value.

For further details, see LLVM's atomicrmw and instruction.

Examples

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

Atomically bitwise-nand (not-and) x with val

Performs x[] = ~(x[] & val) atomically. Returns the old value.

For further details, see LLVM's atomicrmw nand instruction.

Examples

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

Atomically bitwise-or x with val

Performs x[] |= val atomically. Returns the old value.

For further details, see LLVM's atomicrmw or instruction.

Examples

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

Atomically bitwise-xor (exclusive-or) x with val

Performs x[] $= val atomically. Returns the old value.

For further details, see LLVM's atomicrmw xor instruction.

Examples

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

Atomically store the maximum of x and val in x

Performs x[] = max(x[], val) atomically. Returns the old value.

For further details, see LLVM's atomicrmw max instruction.

Examples

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

Atomically store the minimum of x and val in x

Performs x[] = min(x[], val) atomically. Returns the old value.

For further details, see LLVM's atomicrmw min instruction.

Examples

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

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

julia> x[]
5

Threads.atomic_fence()

Insert a sequential-consistency memory fence

Inserts a memory fence with sequentially-consistent ordering semantics. There are algorithms where this is needed, i.e. where an acquire/release ordering is insufficient.

This is likely a very expensive operation. Given that all other atomic operations in Julia already have acquire/release semantics, explicit fences should not be necessary in most cases.

For further details, see LLVM's fence instruction.

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

The @threadcall macro is called in the same way as ccall but does the work in a different thread. This is useful when you want to call a blocking C function without causing the current julia thread to become blocked. Concurrency is limited by size of the libuv thread pool, which defaults to 4 threads but can be increased by setting the UV_THREADPOOL_SIZE environment variable and restarting the julia process.

Note that the called function should never call back into Julia.

SpinLock()

Create a non-reentrant, test-and-test-and-set spin lock. Recursive use will result in a deadlock. This kind of lock should only be used around code that takes little time to execute and does not block (e.g. perform I/O). In general, ReentrantLock should be used instead.

Each lock must be matched with an unlock. If !islocked(lck::SpinLock) holds, trylock(lck) succeeds unless there are other tasks attempting to hold the lock "at the same time."

Test-and-test-and-set spin locks are quickest up to about 30ish contending threads. If you have more contention than that, different synchronization approaches should be considered.

Base.Experimental.task_metrics(::Bool)

Enable or disable the collection of per-task metrics. A Task created when Base.Experimental.task_metrics(true) is in effect will have Base.Experimental.task_running_time_ns and Base.Experimental.task_wall_time_ns timing information available.

Base.Experimental.task_running_time_ns(t::Task) -> Union{UInt64, Nothing}

Return the total nanoseconds that the task t has spent running. This metric is only updated when t yields or completes unless t is the current task, in which it will be updated continuously. See also Base.Experimental.task_wall_time_ns.

Returns nothing if task timings are not enabled. See Base.Experimental.task_metrics.

Base.Experimental.task_wall_time_ns(t::Task) -> Union{UInt64, Nothing}

Return the total nanoseconds that the task t was runnable. This is the time since the task first entered the run queue until the time at which it completed, or until the current time if the task has not yet completed. See also Base.Experimental.task_running_time_ns.

Returns nothing if task timings are not enabled. See Base.Experimental.task_metrics.