1572

Committed 01 Feb 2026 09:55PM UTC coverage: 76.677% (-0.07%) from 76.749%

Build # 1572

Build Type

push

buildkite

Committed by

web-flow

Commit Message

docs: clarify 'using A, B' semantics (#60856)

Resolves #36090

Run Details

62889 of 82018 relevant lines covered (76.68%)

23269256.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.5

/base/threadingconstructs.jl

# This file is a part of Julia. License is MIT: https://julialang.org/license

export threadid, nthreads, @threads, @spawn,
       threadpool, nthreadpools

public Condition, threadpoolsize, ngcthreads

"""
    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-repl
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
```

!!! note
    The thread that a task runs on may change if the task yields, which is known as [`Task Migration`](@ref man-task-migration).
    For this reason in most cases it is not safe to use `threadid([task])` to index into, say, a vector of buffers or stateful
    objects.
"""
threadid() = Int(ccall(:jl_threadid, Int16, ())+1)

# lower bound on the largest threadid()
"""
    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()`](@ref) as well as `threadid(task)` for
any task you were able to observe before calling `maxthreadid`.
"""
maxthreadid() = Int(Core.Intrinsics.atomic_pointerref(cglobal(:jl_n_threads, Cint), :acquire))

"""
    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`](@ref
man-linalg) standard library, and `nprocs()` in the [`Distributed`](@ref man-distributed)
standard library and [`Threads.maxthreadid()`](@ref).
"""
nthreads(pool::Symbol) = threadpoolsize(pool)

function _nthreads_in_pool(tpid::Int8)
    p = unsafe_load(cglobal(:jl_n_threads_per_pool, Ptr{Cint}))
    return Int(unsafe_load(p, tpid + 1))
end

function _tpid_to_sym(tpid::Int8)
    if tpid == 0
        return :interactive
    elseif tpid == 1
        return :default
    elseif tpid == -1
        return :foreign
    else
        throw(ArgumentError(LazyString("Unrecognized threadpool id ", tpid)))
    end
end

function _sym_to_tpid(tp::Symbol)
    if tp === :interactive
        return Int8(0)
    elseif tp === :default
        return Int8(1)
    elseif tp == :foreign
        return Int8(-1)
    else
        throw(ArgumentError(LazyString("Unrecognized threadpool name `", tp, "`")))
    end
end

"""
    Threads.threadpool(tid = threadid())::Symbol

Return the specified thread's threadpool; either `:default`, `:interactive`, or `:foreign`.
"""
function threadpool(tid = threadid())
    tpid = ccall(:jl_threadpoolid, Int8, (Int16,), tid-1)
    return _tpid_to_sym(tpid)
end

"""
    Threads.threadpooldescription(tid = threadid())::String

Return the specified thread's threadpool name with extended description where appropriate.
"""
function threadpooldescription(tid = threadid())
    threadpool_name = threadpool(tid)
    if threadpool_name == :foreign
        # TODO: extend tls to include a field to add a description to a foreign thread and make this more general
        n_others = nthreads(:interactive) + nthreads(:default)
        # Assumes GC threads come first in the foreign thread pool
        if tid > n_others && tid <= n_others + ngcthreads()
            return "foreign: gc"
        end
    end
    return string(threadpool_name)
end

"""
    Threads.nthreadpools()::Int

Return the number of threadpools currently configured.
"""
nthreadpools() = Int(unsafe_load(cglobal(:jl_n_threadpools, Cint)))

"""
    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`](@ref man-linalg) standard library, and `nprocs()` in the
[`Distributed`](@ref man-distributed) standard library.
"""
function threadpoolsize(pool::Symbol = :default)
    if pool === :default || pool === :interactive
        tpid = _sym_to_tpid(pool)
    elseif pool == :foreign
        error("Threadpool size of `:foreign` is indeterminant")
    else
        error("invalid threadpool specified")
    end
    return _nthreads_in_pool(tpid)
end

"""
    threadpooltids(pool::Symbol)

Return a vector of IDs of threads in the given pool.
"""
function threadpooltids(pool::Symbol)
    ni = _nthreads_in_pool(Int8(0))
    if pool === :interactive
        return collect(1:ni)
    elseif pool === :default
        return collect(ni+1:ni+_nthreads_in_pool(Int8(1)))
    else
        error("invalid threadpool specified")
    end
end

"""
    Threads.ngcthreads()::Int

Return the number of GC threads currently configured.
This includes both mark threads and concurrent sweep threads.
"""
ngcthreads() = Int(unsafe_load(cglobal(:jl_n_gcthreads, Cint))) + 1

function threading_run(fun, static)
    ccall(:jl_enter_threaded_region, Cvoid, ())
    n = threadpoolsize()
    tid_offset = threadpoolsize(:interactive)
    tasks = Vector{Task}(undef, n)
    try
        for i = 1:n
            t = Task(() -> fun(i)) # pass in tid
            t.sticky = static
            if static
                ccall(:jl_set_task_tid, Cint, (Any, Cint), t, tid_offset + i-1)
            else
                # TODO: this should be the current pool (except interactive) if there
                # are ever more than two pools.
                _result = ccall(:jl_set_task_threadpoolid, Cint, (Any, Int8), t, _sym_to_tpid(:default))
                @assert _result == 1
            end
            tasks[i] = t
            schedule(t)
        end
        for i = 1:n
            Base._wait(tasks[i])
        end
    finally
        ccall(:jl_exit_threaded_region, Cvoid, ())
    end
    failed_tasks = filter!(istaskfailed, tasks)
    if !isempty(failed_tasks)
        throw(CompositeException(map(TaskFailedException, failed_tasks)))
    end
end

function _threadsfor(iter, lbody, schedule)
    lidx = iter.args[1]         # index
    range = iter.args[2]
    esc_range = esc(range)
    func = if schedule === :greedy
        greedy_func(esc_range, lidx, lbody)
    else
        default_func(esc_range, lidx, lbody)
    end
    quote
        local threadsfor_fun
        $func
        if $(schedule === :greedy || schedule === :dynamic || schedule === :default)
            threading_run(threadsfor_fun, false)
        elseif ccall(:jl_in_threaded_region, Cint, ()) != 0 # :static
            error("`@threads :static` cannot be used concurrently or nested")
        else # :static
            threading_run(threadsfor_fun, true)
        end
        nothing
    end
end

function greedy_func(itr, lidx, lbody)
    quote
        let c = Channel{eltype($itr)}(threadpoolsize(), spawn=true) do ch
            for item in $itr
                put!(ch, item)
            end
        end
        function threadsfor_fun(tid)
            for item in c
                local $(esc(lidx)) = item
                $(esc(lbody))
            end
        end
        end
    end
end

function default_func(itr, lidx, lbody)
    quote
        let range = $itr
        function threadsfor_fun(tid = 1; onethread = false)
            r = range # Load into local variable
            lenr = length(r)
            # divide loop iterations among threads
            if onethread
                tid = 1
                len, rem = lenr, 0
            else
                len, rem = divrem(lenr, threadpoolsize())
            end
            # not enough iterations for all the threads?
            if len == 0
                if tid > rem
                    return
                end
                len, rem = 1, 0
            end
            # compute this thread's iterations
            f = firstindex(r) + ((tid-1) * len)
            l = f + len - 1
            # distribute remaining iterations evenly
            if rem > 0
                if tid <= rem
                    f = f + (tid-1)
                    l = l + tid
                else
                    f = f + rem
                    l = l + rem
                end
            end
            # run this thread's iterations
            for i = f:l
                local $(esc(lidx)) = @inbounds r[i]
                $(esc(lbody))
            end
        end
        end
    end
end

"""
    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.

Tasks spawned by `@threads` are scheduled on the `:default` threadpool. This means that
`@threads` will not use threads from the `:interactive` threadpool, even if called from
the main thread or from a task in the interactive pool. The `:default` threadpool is
intended for compute-intensive parallel workloads.

See also: [`@spawn`](@ref Threads.@spawn) and
`pmap` in [`Distributed`](@ref man-distributed).
For more information on threadpools, see the chapter on [threadpools](@ref man-threadpools).

# 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 `Channel`s 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()`](@ref Threads.threadid)
  may change even within a single iteration. See [`Task Migration`](@ref man-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.

!!! compat "Julia 1.5"
    The `schedule` argument is available as of Julia 1.5.

### `: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 `Task`s used by `:dynamic` scheduler is
bounded by a small constant multiple of the number of available worker threads
([`Threads.threadpoolsize()`](@ref)). 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).

!!! compat "Julia 1.8"
    The `:dynamic` option for the `schedule` argument is available and the default as of Julia 1.8.

### `:greedy`

`:greedy` scheduler spawns up to [`Threads.threadpoolsize()`](@ref) 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.

!!! compat "Julia 1.11"
    The `:greedy` option for the `schedule` argument is available as of Julia 1.11.

### `: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()`](@ref Threads.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.

!!! note
    `:static` scheduling exists for supporting transition of code written before Julia 1.3.
    In newly written library functions, `:static` scheduling is discouraged because the
    functions using this option cannot be called from arbitrary worker threads.

## 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-repl
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.
"""
macro threads(args...)
    na = length(args)
    if na == 2
        sched, ex = args
        if sched isa QuoteNode
            sched = sched.value
        elseif sched isa Symbol
            # for now only allow quoted symbols
            sched = nothing
        end
        if sched !== :static && sched !== :dynamic && sched !== :greedy
            throw(ArgumentError("unsupported schedule argument in @threads"))
        end
    elseif na == 1
        sched = :default
        ex = args[1]
    else
        throw(ArgumentError("wrong number of arguments in @threads"))
    end
    if !(isa(ex, Expr) && ex.head === :for)
        throw(ArgumentError("@threads requires a `for` loop expression"))
    end
    if !(ex.args[1] isa Expr && ex.args[1].head === :(=))
        throw(ArgumentError("nested outer loops are not currently supported by @threads"))
    end
    return _threadsfor(ex.args[1], ex.args[2], sched)
end

function _spawn_set_thrpool(t::Task, tp::Symbol)
    tpid = _sym_to_tpid(tp)
    if tpid == -1 || _nthreads_in_pool(tpid) == 0
        tpid = _sym_to_tpid(:default)
    end
    _result = ccall(:jl_set_task_threadpoolid, Cint, (Any, Int8), t, tpid)
    @assert _result == 1
    nothing
end

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

Create a [`Task`](@ref) and [`schedule`](@ref) it to run on any available
thread in the specified threadpool: `:default`, `:interactive`, or `:samepool`
to use the same as the caller. `:default` is used if unspecified. The task is
allocated to a thread once one becomes available. To wait for the task to
finish, call [`wait`](@ref) on the result of this macro, or call
[`fetch`](@ref) 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.

!!! note
    The thread that the task runs on may change if the task yields, therefore `threadid()` should not
    be treated as constant for a task. See [`Task Migration`](@ref man-task-migration), and the broader
    [multi-threading](@ref man-multithreading) manual for further important caveats.
    See also the chapter on [threadpools](@ref man-threadpools).

!!! compat "Julia 1.3"
    This macro is available as of Julia 1.3.

!!! compat "Julia 1.4"
    Interpolating values via `\$` is available as of Julia 1.4.

!!! compat "Julia 1.9"
    A threadpool may be specified as of Julia 1.9.

!!! compat "Julia 1.12"
    The same threadpool may be specified as of Julia 1.12.

# Examples
```julia-repl
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
```
"""
macro spawn(args...)
    tp = QuoteNode(:default)
    na = length(args)
    if na == 2
        ttype, ex = args
        if ttype isa QuoteNode
            ttype = ttype.value
            if !in(ttype, (:interactive, :default, :samepool))
                throw(ArgumentError(LazyString("unsupported threadpool in @spawn: ", ttype)))
            end
            tp = QuoteNode(ttype)
        else
            tp = ttype
        end
    elseif na == 1
        ex = args[1]
    else
        throw(ArgumentError("wrong number of arguments in @spawn"))
    end

    letargs = Base._lift_one_interp!(ex)

    thunk = Base.replace_linenums!(:(()->($(esc(ex)))), __source__)
    var = esc(Base.sync_varname)
    quote
        let $(letargs...)
            local task = Task($thunk)
            task.sticky = false
            local tp = $(esc(tp))
            if tp == :samepool
                tp = Threads.threadpool()
            end
            _spawn_set_thrpool(task, tp)
            if $(Expr(:islocal, var))
                put!($var, task)
            end
            schedule(task)
            task
        end
    end
end

# This is a stub that can be overloaded for downstream structures like `Channel`
function foreach end

# Scheduling traits that can be employed for downstream overloads
abstract type AbstractSchedule end
struct StaticSchedule <: AbstractSchedule end
struct FairSchedule <: AbstractSchedule end

1	# This file is a part of Julia. License is MIT: https://julialang.org/license
2
3	export threadid, nthreads, @threads, @spawn,
4	threadpool, nthreadpools
5
6	public Condition, threadpoolsize, ngcthreads
7
8	"""
9	Threads.threadid([t::Task])::Int
10
11	Get the ID number of the current thread of execution, or the thread of task
12	`t`. The master thread has ID `1`.
13
14	# Examples
15	```julia-repl
16	julia> Threads.threadid()
17	1
18
19	julia> Threads.@threads for i in 1:4
20	println(Threads.threadid())
21	end
22	4
23	2
24	5
25	4
26
27	julia> Threads.threadid(Threads.@spawn "foo")
28	2
29	```
30
31	!!! note
32	The thread that a task runs on may change if the task yields, which is known as [`Task Migration`](@ref man-task-migration).
33	For this reason in most cases it is not safe to use `threadid([task])` to index into, say, a vector of buffers or stateful
34	objects.
35	"""
36	threadid() = Int(ccall(:jl_threadid, Int16, ())+1)	171,002,915✔
37
38	# lower bound on the largest threadid()
39	"""
40	Threads.maxthreadid()::Int
41
42	Get a lower bound on the number of threads (across all thread pools) available
43	to the Julia process, with atomic-acquire semantics. The result will always be
44	greater than or equal to [`threadid()`](@ref) as well as `threadid(task)` for
45	any task you were able to observe before calling `maxthreadid`.
46	"""
47	maxthreadid() = Int(Core.Intrinsics.atomic_pointerref(cglobal(:jl_n_threads, Cint), :acquire))	3,245✔
48
49	"""
50	Threads.nthreads(:default \| :interactive)::Int
51
52	Get the current number of threads within the specified thread pool. The threads in `:interactive`
53	have id numbers `1:nthreads(:interactive)`, and the threads in `:default` have id numbers in
54	`nthreads(:interactive) .+ (1:nthreads(:default))`.
55
56	See also `BLAS.get_num_threads` and `BLAS.set_num_threads` in the [`LinearAlgebra`](@ref
57	man-linalg) standard library, and `nprocs()` in the [`Distributed`](@ref man-distributed)
58	standard library and [`Threads.maxthreadid()`](@ref).
59	"""
60	nthreads(pool::Symbol) = threadpoolsize(pool)	66✔
61
62	function _nthreads_in_pool(tpid::Int8)	16✔
63	p = unsafe_load(cglobal(:jl_n_threads_per_pool, Ptr{Cint}))	118,202,346✔
64	return Int(unsafe_load(p, tpid + 1))	118,202,346✔
65	end
66
67	function _tpid_to_sym(tpid::Int8)	8✔
68	if tpid == 0	41,513,947✔
69	return :interactive	5,926✔
70	elseif tpid == 1	41,508,021✔
71	return :default	41,508,021✔
72	elseif tpid == -1	×
73	return :foreign	×
74	else
75	throw(ArgumentError(LazyString("Unrecognized threadpool id ", tpid)))	×
76	end
77	end
78
79	function _sym_to_tpid(tp::Symbol)	16✔
80	if tp === :interactive	96,887,493✔
81	return Int8(0)	17,675,177✔
82	elseif tp === :default	79,212,316✔
83	return Int8(1)	79,212,310✔
84	elseif tp == :foreign	6✔
85	return Int8(-1)	×
86	else
87	throw(ArgumentError(LazyString("Unrecognized threadpool name `", tp, "`")))	6✔
88	end
89	end
90
91	"""
92	Threads.threadpool(tid = threadid())::Symbol
93
94	Return the specified thread's threadpool; either `:default`, `:interactive`, or `:foreign`.
95	"""
96	function threadpool(tid = threadid())	30✔
97	tpid = ccall(:jl_threadpoolid, Int8, (Int16,), tid-1)	394✔
98	return _tpid_to_sym(tpid)	233✔
99	end
100
101	"""
102	Threads.threadpooldescription(tid = threadid())::String
103
104	Return the specified thread's threadpool name with extended description where appropriate.
105	"""
106	function threadpooldescription(tid = threadid())	4✔
107	threadpool_name = threadpool(tid)	8✔
108	if threadpool_name == :foreign	4✔
109	# TODO: extend tls to include a field to add a description to a foreign thread and make this more general
110	n_others = nthreads(:interactive) + nthreads(:default)	×
111	# Assumes GC threads come first in the foreign thread pool
112	if tid > n_others && tid <= n_others + ngcthreads()	×
113	return "foreign: gc"	×
114	end
115	end
116	return string(threadpool_name)	4✔
117	end
118
119	"""
120	Threads.nthreadpools()::Int
121
122	Return the number of threadpools currently configured.
123	"""
124	nthreadpools() = Int(unsafe_load(cglobal(:jl_n_threadpools, Cint)))	3✔
125
126	"""
127	Threads.threadpoolsize(pool::Symbol = :default)::Int
128
129	Get the number of threads available to the default thread pool (or to the
130	specified thread pool).
131
132	See also: `BLAS.get_num_threads` and `BLAS.set_num_threads` in the
133	[`LinearAlgebra`](@ref man-linalg) standard library, and `nprocs()` in the
134	[`Distributed`](@ref man-distributed) standard library.
135	"""
136	function threadpoolsize(pool::Symbol = :default)	983✔
137	if pool === :default \|\| pool === :interactive	72,413,183✔
138	tpid = _sym_to_tpid(pool)	107,304,543✔
139	elseif pool == :foreign	×
140	error("Threadpool size of `:foreign` is indeterminant")	×
141	else
142	error("invalid threadpool specified")	×
143	end
144	return _nthreads_in_pool(tpid)	65,795,993✔
145	end
146
147	"""
148	threadpooltids(pool::Symbol)
149
150	Return a vector of IDs of threads in the given pool.
151	"""
152	function threadpooltids(pool::Symbol)	6✔
153	ni = _nthreads_in_pool(Int8(0))	6✔
154	if pool === :interactive	6✔
155	return collect(1:ni)	3✔
156	elseif pool === :default	3✔
157	return collect(ni+1:ni+_nthreads_in_pool(Int8(1)))	3✔
158	else
159	error("invalid threadpool specified")	×
160	end
161	end
162
163	"""
164	Threads.ngcthreads()::Int
165
166	Return the number of GC threads currently configured.
167	This includes both mark threads and concurrent sweep threads.
168	"""
169	ngcthreads() = Int(unsafe_load(cglobal(:jl_n_gcthreads, Cint))) + 1	25✔
170
171	function threading_run(fun, static)	1,983,184✔
172	ccall(:jl_enter_threaded_region, Cvoid, ())	1,983,184✔
173	n = threadpoolsize()	1,983,184✔
174	tid_offset = threadpoolsize(:interactive)	1,983,184✔
175	tasks = Vector{Task}(undef, n)	1,983,184✔
176	try	1,983,184✔
177	for i = 1:n	3,966,365✔
178	t = Task(() -> fun(i)) # pass in tid	9,255,200✔
179	t.sticky = static	4,627,600✔
180	if static	4,627,600✔
181	ccall(:jl_set_task_tid, Cint, (Any, Cint), t, tid_offset + i-1)	280✔
182	else
183	# TODO: this should be the current pool (except interactive) if there
184	# are ever more than two pools.
185	_result = ccall(:jl_set_task_threadpoolid, Cint, (Any, Int8), t, _sym_to_tpid(:default))	4,627,320✔
186	@assert _result == 1	4,627,320✔
187	end
188	tasks[i] = t	4,627,600✔
189	schedule(t)	4,627,600✔
190	end	7,272,016✔
191	for i = 1:n	3,966,365✔
192	Base._wait(tasks[i])	4,627,600✔
193	end	7,272,016✔
194	finally
195	ccall(:jl_exit_threaded_region, Cvoid, ())	1,983,184✔
196	end
197	failed_tasks = filter!(istaskfailed, tasks)	1,983,187✔
198	if !isempty(failed_tasks)	1,983,184✔
199	throw(CompositeException(map(TaskFailedException, failed_tasks)))	72✔
200	end
201	end
202
203	function _threadsfor(iter, lbody, schedule)	838✔
204	lidx = iter.args[1] # index	838✔
205	range = iter.args[2]	838✔
206	esc_range = esc(range)	838✔
207	func = if schedule === :greedy	838✔
208	greedy_func(esc_range, lidx, lbody)	126✔
209	else
210	default_func(esc_range, lidx, lbody)	1,550✔
211	end
212	quote	838✔
213	local threadsfor_fun
214	$func
215	if $(schedule === :greedy \|\| schedule === :dynamic \|\| schedule === :default)	1,983,181✔
216	threading_run(threadsfor_fun, false)	1,983,064✔
217	elseif ccall(:jl_in_threaded_region, Cint, ()) != 0 # :static	138✔
218	error("`@threads :static` cannot be used concurrently or nested")	18✔
219	else # :static
220	threading_run(threadsfor_fun, true)	120✔
221	end
222	nothing	1,983,091✔
223	end
224	end
225
226	function greedy_func(itr, lidx, lbody)	126✔
227	quote	126✔
228	let c = Channel{eltype($itr)}(threadpoolsize(), spawn=true) do ch	252✔
229	for item in $itr	504✔
230	put!(ch, item)	828✔
231	end	828✔
232	end
233	function threadsfor_fun(tid)	882✔
234	for item in c	672✔
235	local $(esc(lidx)) = item	828✔
236	$(esc(lbody))	828✔
237	end	786✔
238	end
239	end
240	end
241	end
242
243	function default_func(itr, lidx, lbody)	712✔
244	quote	712✔
245	let range = $itr	1,982,986✔
246	function threadsfor_fun(tid = 1; onethread = false)	11,236,764✔
247	r = range # Load into local variable	4,626,928✔
248	lenr = length(r)	4,627,012✔
249	# divide loop iterations among threads
250	if onethread	4,626,928✔
251	tid = 1	×
252	len, rem = lenr, 0	×
253	else
254	len, rem = divrem(lenr, threadpoolsize())	4,626,928✔
255	end
256	# not enough iterations for all the threads?
257	if len == 0	4,626,928✔
258	if tid > rem	234✔
259	return	138✔
260	end
261	len, rem = 1, 0	96✔
262	end
263	# compute this thread's iterations
264	f = firstindex(r) + ((tid-1) * len)	4,626,790✔
265	l = f + len - 1	4,626,790✔
266	# distribute remaining iterations evenly
267	if rem > 0	4,626,790✔
268	if tid <= rem	228✔
269	f = f + (tid-1)	108✔
270	l = l + tid	108✔
271	else
272	f = f + rem	120✔
273	l = l + rem	120✔
274	end
275	end
276	# run this thread's iterations
277	for i = f:l	9,253,577✔
278	local $(esc(lidx)) = @inbounds r[i]	2,025,352,235✔
279	$(esc(lbody))	20,510,585✔
280	end	2,025,352,133✔
281	end
282	end
283	end
284	end
285
286	"""
287	Threads.@threads [schedule] for ... end
288
289	A macro to execute a `for` loop in parallel. The iteration space is distributed to
290	coarse-grained tasks. This policy can be specified by the `schedule` argument. The
291	execution of the loop waits for the evaluation of all iterations.
292
293	Tasks spawned by `@threads` are scheduled on the `:default` threadpool. This means that
294	`@threads` will not use threads from the `:interactive` threadpool, even if called from
295	the main thread or from a task in the interactive pool. The `:default` threadpool is
296	intended for compute-intensive parallel workloads.
297
298	See also: [`@spawn`](@ref Threads.@spawn) and
299	`pmap` in [`Distributed`](@ref man-distributed).
300	For more information on threadpools, see the chapter on [threadpools](@ref man-threadpools).
301
302	# Extended help
303
304	## Semantics
305
306	Unless stronger guarantees are specified by the scheduling option, the loop executed by
307	`@threads` macro have the following semantics.
308
309	The `@threads` macro executes the loop body in an unspecified order and potentially
310	concurrently. It does not specify the exact assignments of the tasks and the worker threads.
311	The assignments can be different for each execution. The loop body code (including any code
312	transitively called from it) must not make any assumptions about the distribution of
313	iterations to tasks or the worker thread in which they are executed. The loop body for each
314	iteration must be able to make forward progress independent of other iterations and be free
315	from data races. As such, invalid synchronizations across iterations may deadlock while
316	unsynchronized memory accesses may result in undefined behavior.
317
318	For example, the above conditions imply that:
319
320	- A lock taken in an iteration must be released within the same iteration.
321	- Communicating between iterations using blocking primitives like `Channel`s is incorrect.
322	- Write only to locations not shared across iterations (unless a lock or atomic operation is
323	used).
324	- Unless the `:static` schedule is used, the value of [`threadid()`](@ref Threads.threadid)
325	may change even within a single iteration. See [`Task Migration`](@ref man-task-migration).
326
327	## Schedulers
328
329	Without the scheduler argument, the exact scheduling is unspecified and varies across Julia
330	releases. Currently, `:dynamic` is used when the scheduler is not specified.
331
332	!!! compat "Julia 1.5"
333	The `schedule` argument is available as of Julia 1.5.
334
335	### `:dynamic` (default)
336
337	`:dynamic` scheduler executes iterations dynamically to available worker threads. Current
338	implementation assumes that the workload for each iteration is uniform. However, this
339	assumption may be removed in the future.
340
341	This scheduling option is merely a hint to the underlying execution mechanism. However, a
342	few properties can be expected. The number of `Task`s used by `:dynamic` scheduler is
343	bounded by a small constant multiple of the number of available worker threads
344	([`Threads.threadpoolsize()`](@ref)). Each task processes contiguous regions of the
345	iteration space. Thus, `@threads :dynamic for x in xs; f(x); end` is typically more
346	efficient than `@sync for x in xs; @spawn f(x); end` if `length(xs)` is significantly
347	larger than the number of the worker threads and the run-time of `f(x)` is relatively
348	smaller than the cost of spawning and synchronizing a task (typically less than 10
349	microseconds).
350
351	!!! compat "Julia 1.8"
352	The `:dynamic` option for the `schedule` argument is available and the default as of Julia 1.8.
353
354	### `:greedy`
355
356	`:greedy` scheduler spawns up to [`Threads.threadpoolsize()`](@ref) tasks, each greedily working on
357	the given iterated values as they are produced. As soon as one task finishes its work, it takes
358	the next value from the iterator. Work done by any individual task is not necessarily on
359	contiguous values from the iterator. The given iterator may produce values forever, only the
360	iterator interface is required (no indexing).
361
362	This scheduling option is generally a good choice if the workload of individual iterations
363	is not uniform/has a large spread.
364
365	!!! compat "Julia 1.11"
366	The `:greedy` option for the `schedule` argument is available as of Julia 1.11.
367
368	### `:static`
369
370	`:static` scheduler creates one task per thread and divides the iterations equally among
371	them, assigning each task specifically to each thread. In particular, the value of
372	[`threadid()`](@ref Threads.threadid) is guaranteed to be constant within one iteration.
373	Specifying `:static` is an error if used from inside another `@threads` loop or from a
374	thread other than 1.
375
376	!!! note
377	`:static` scheduling exists for supporting transition of code written before Julia 1.3.
378	In newly written library functions, `:static` scheduling is discouraged because the
379	functions using this option cannot be called from arbitrary worker threads.
380
381	## Examples
382
383	To illustrate of the different scheduling strategies, consider the following function
384	`busywait` containing a non-yielding timed loop that runs for a given number of seconds.
385
386	```julia-repl
387	julia> function busywait(seconds)
388	tstart = time_ns()
389	while (time_ns() - tstart) / 1e9 < seconds
390	end
391	end
392
393	julia> @time begin
394	Threads.@spawn busywait(5)
395	Threads.@threads :static for i in 1:Threads.threadpoolsize()
396	busywait(1)
397	end
398	end
399	6.003001 seconds (16.33 k allocations: 899.255 KiB, 0.25% compilation time)
400
401	julia> @time begin
402	Threads.@spawn busywait(5)
403	Threads.@threads :dynamic for i in 1:Threads.threadpoolsize()
404	busywait(1)
405	end
406	end
407	2.012056 seconds (16.05 k allocations: 883.919 KiB, 0.66% compilation time)
408	```
409
410	The `:dynamic` example takes 2 seconds since one of the non-occupied threads is able
411	to run two of the 1-second iterations to complete the for loop.
412	"""
413	macro threads(args...)	865✔
414	na = length(args)	865✔
415	if na == 2	865✔
416	sched, ex = args	345✔
417	if sched isa QuoteNode	345✔
418	sched = sched.value	342✔
419	elseif sched isa Symbol	3✔
420	# for now only allow quoted symbols
421	sched = nothing	×
422	end
423	if sched !== :static && sched !== :dynamic && sched !== :greedy	345✔
424	throw(ArgumentError("unsupported schedule argument in @threads"))	3✔
425	end
426	elseif na == 1	520✔
427	sched = :default	520✔
428	ex = args[1]	520✔
429	else
430	throw(ArgumentError("wrong number of arguments in @threads"))	×
431	end
432	if !(isa(ex, Expr) && ex.head === :for)	865✔
433	throw(ArgumentError("@threads requires a `for` loop expression"))	6✔
434	end
435	if !(ex.args[1] isa Expr && ex.args[1].head === :(=))	856✔
436	throw(ArgumentError("nested outer loops are not currently supported by @threads"))	18✔
437	end
438	return _threadsfor(ex.args[1], ex.args[2], sched)	838✔
439	end
440
441	function _spawn_set_thrpool(t::Task, tp::Symbol)	26,464,340✔
442	tpid = _sym_to_tpid(tp)	52,928,646✔
443	if tpid == -1 \|\| _nthreads_in_pool(tpid) == 0	52,928,668✔
444	tpid = _sym_to_tpid(:default)	4✔
445	end
446	_result = ccall(:jl_set_task_threadpoolid, Cint, (Any, Int8), t, tpid)	26,464,334✔
447	@assert _result == 1	26,464,334✔
448	nothing	26,464,334✔
449	end
450
451	"""
452	Threads.@spawn [:default\|:interactive\|:samepool] expr
453
454	Create a [`Task`](@ref) and [`schedule`](@ref) it to run on any available
455	thread in the specified threadpool: `:default`, `:interactive`, or `:samepool`
456	to use the same as the caller. `:default` is used if unspecified. The task is
457	allocated to a thread once one becomes available. To wait for the task to
458	finish, call [`wait`](@ref) on the result of this macro, or call
459	[`fetch`](@ref) to wait and then obtain its return value.
460
461	Values can be interpolated into `@spawn` via `\$`, which copies the value
462	directly into the constructed underlying closure. This allows you to insert
463	the _value_ of a variable, isolating the asynchronous code from changes to
464	the variable's value in the current task.
465
466	!!! note
467	The thread that the task runs on may change if the task yields, therefore `threadid()` should not
468	be treated as constant for a task. See [`Task Migration`](@ref man-task-migration), and the broader
469	[multi-threading](@ref man-multithreading) manual for further important caveats.
470	See also the chapter on [threadpools](@ref man-threadpools).
471
472	!!! compat "Julia 1.3"
473	This macro is available as of Julia 1.3.
474
475	!!! compat "Julia 1.4"
476	Interpolating values via `\$` is available as of Julia 1.4.
477
478	!!! compat "Julia 1.9"
479	A threadpool may be specified as of Julia 1.9.
480
481	!!! compat "Julia 1.12"
482	The same threadpool may be specified as of Julia 1.12.
483
484	# Examples
485	```julia-repl
486	julia> t() = println("Hello from ", Threads.threadid());
487
488	julia> tasks = fetch.([Threads.@spawn t() for i in 1:4]);
489	Hello from 1
490	Hello from 1
491	Hello from 3
492	Hello from 4
493	```
494	"""
495	macro spawn(args...)	1,099✔
496	tp = QuoteNode(:default)	1,099✔
497	na = length(args)	1,099✔
498	if na == 2	1,099✔
499	ttype, ex = args	66✔
500	if ttype isa QuoteNode	66✔
501	ttype = ttype.value	54✔
502	if !in(ttype, (:interactive, :default, :samepool))	54✔
503	throw(ArgumentError(LazyString("unsupported threadpool in @spawn: ", ttype)))	×
504	end
505	tp = QuoteNode(ttype)	54✔
506	else
507	tp = ttype	12✔
508	end
509	elseif na == 1	1,033✔
510	ex = args[1]	1,033✔
511	else
512	throw(ArgumentError("wrong number of arguments in @spawn"))	×
513	end
514
515	letargs = Base._lift_one_interp!(ex)	1,099✔
516
517	thunk = Base.replace_linenums!(:(()->($(esc(ex)))), __source__)	1,099✔
518	var = esc(Base.sync_varname)	1,099✔
519	quote	1,099✔
520	let $(letargs...)	1,134✔
521	local task = Task($thunk)	26,477,030✔
522	task.sticky = false	26,477,030✔
523	local tp = $(esc(tp))	26,477,063✔
524	if tp == :samepool	26,477,030✔
525	tp = Threads.threadpool()	12✔
526	end
527	_spawn_set_thrpool(task, tp)	26,477,030✔
528	if $(Expr(:islocal, var))	26,477,030✔
529	put!($var, task)	22,501,360✔
530	end
531	schedule(task)	26,469,346✔
532	task	26,469,346✔
533	end
534	end
535	end
536
537	# This is a stub that can be overloaded for downstream structures like `Channel`
538	function foreach end
539
540	# Scheduling traits that can be employed for downstream overloads
541	abstract type AbstractSchedule end
542	struct StaticSchedule <: AbstractSchedule end	36✔
543	struct FairSchedule <: AbstractSchedule end	54✔

JuliaLang / julia / 1572

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous