#38177

Committed 14 Aug 2025 11:50AM UTC coverage: 78.503% (+0.7%) from 77.785%

Build # #38177

Build Type

push

local

Committed by

web-flow

Commit Message

Narrow drive letter matching for `splitdrive` on Windows to one character, fixup docstring for `joinpath` (#58951)

Fixes #58929

Only a single letter followed by a colon is a valid drive prefix for
this format, any other length of string prior to a colon isn't a valid
drive in Windows:
https://learn.microsoft.com/en-us/dotnet/standard/io/file-path-formats

Also the docstring for `joinpath` rendered Windows paths as unescaped,
meaning they could not be copied and pasted on the REPL.

Run Details

48594 of 61901 relevant lines covered (78.5%)

8669158.51 hits per line

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

82.58

/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

Returns 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

Returns 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

Returns 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)

Returns 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

Returns 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)
    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
    ccall(:jl_exit_threaded_region, Cvoid, ())
    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)}(0,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.

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

JuliaLang / julia / #38177

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