#37579

pending completion

Build # #37579

Build Type

push

local

Committed by

web-flow

Commit Message

sroa: Mark dead setfields as EFFECT_FREE (#50373)

sroa tries to delete any `setfield!` call for allocations that it
knows it can remove. However, if it does not know that the type is
correct for the allocation, it may not be able to remove the
setfield!. If the type later gets improved (e.g. by irinterp),
the statement becomes eligible for removal, but it currently requires
another sroa pass to actually remove it.

Improve that situation my marking such a statement that is known-dead
as IR_FLAG_EFFECT_FREE, so if we later also prove it nothrow, it
(and the corresponding allocation) immediately become DCE-eligible.

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

73522 of 84317 relevant lines covered (87.2%)

34757350.52 hits per line

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

91.55

/base/threadingconstructs.jl

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

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

"""
    Threads.threadid() -> Int

Get the ID number of the current thread of execution. 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
```

!!! 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()` to index into, say, a vector of buffer 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 default
have id numbers `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
    else
        throw(ArgumentError("Unrecognized threadpool id $tpid"))
    end
end

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

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

Returns the specified thread's threadpool; either `:default` or `:interactive`.
"""
function threadpool(tid = threadid())
    tpid = ccall(:jl_threadpoolid, Int8, (Int16,), tid-1)
    return _tpid_to_sym(tpid)
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)
    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
        static && ccall(:jl_set_task_tid, Cint, (Any, Cint), t, tid_offset + i-1)
        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]
    quote
        local threadsfor_fun
        let range = $(esc(range))
        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
        if $(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

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

### `: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.

## Example

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
            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 _nthreads_in_pool(tpid) == 0
        tpid = _sym_to_tpid(:default)
    end
    ccall(:jl_set_task_threadpoolid, Cint, (Any, Int8), t, tpid)
    nothing
end

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

Create a [`Task`](@ref) and [`schedule`](@ref) 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`](@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.

# 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 ttype !== :interactive && ttype !== :default
                throw(ArgumentError("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
            _spawn_set_thrpool(task, $(esc(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	"""
7	Threads.threadid() -> Int
8
9	Get the ID number of the current thread of execution. The master thread has
10	ID `1`.
11
12	# Examples
13	```julia-repl
14	julia> Threads.threadid()
15	1
16
17	julia> Threads.@threads for i in 1:4
18	println(Threads.threadid())
19	end
20	4
21	2
22	5
23	4
24	```
25
26	!!! note
27	The thread that a task runs on may change if the task yields, which is known as [`Task Migration`](@ref man-task-migration).
28	For this reason in most cases it is not safe to use `threadid()` to index into, say, a vector of buffer or stateful objects.
29
30	"""
31	threadid() = Int(ccall(:jl_threadid, Int16, ())+1)	27,541,206✔
32
33	# lower bound on the largest threadid()
34	"""
35	Threads.maxthreadid() -> Int
36
37	Get a lower bound on the number of threads (across all thread pools) available
38	to the Julia process, with atomic-acquire semantics. The result will always be
39	greater than or equal to [`threadid()`](@ref) as well as `threadid(task)` for
40	any task you were able to observe before calling `maxthreadid`.
41	"""
42	maxthreadid() = Int(Core.Intrinsics.atomic_pointerref(cglobal(:jl_n_threads, Cint), :acquire))	104✔
43
44	"""
45	Threads.nthreads(:default \| :interactive) -> Int
46
47	Get the current number of threads within the specified thread pool. The threads in default
48	have id numbers `1:nthreads(:default)`.
49
50	See also `BLAS.get_num_threads` and `BLAS.set_num_threads` in the [`LinearAlgebra`](@ref
51	man-linalg) standard library, and `nprocs()` in the [`Distributed`](@ref man-distributed)
52	standard library and [`Threads.maxthreadid()`](@ref).
53	"""
54	nthreads(pool::Symbol) = threadpoolsize(pool)	×
55
56	function _nthreads_in_pool(tpid::Int8)	4,368,509✔
57	p = unsafe_load(cglobal(:jl_n_threads_per_pool, Ptr{Cint}))	25,950,085✔
58	return Int(unsafe_load(p, tpid + 1))	25,950,423✔
59	end
60
61	function _tpid_to_sym(tpid::Int8)	7✔
62	if tpid == 0	8,794,543✔
63	return :interactive	5✔
64	elseif tpid == 1	8,795,346✔
65	return :default	8,795,309✔
66	else
67	throw(ArgumentError("Unrecognized threadpool id $tpid"))	×
68	end
69	end
70
71	function _sym_to_tpid(tp::Symbol)	4,368,441✔
72	if tp === :interactive	17,159,907✔
73	return Int8(0)	440,636✔
74	elseif tp === :default	16,719,470✔
75	return Int8(1)	16,719,412✔
76	else
77	throw(ArgumentError("Unrecognized threadpool name `$(repr(tp))`"))	1✔
78	end
79	end
80
81	"""
82	Threads.threadpool(tid = threadid()) -> Symbol
83
84	Returns the specified thread's threadpool; either `:default` or `:interactive`.
85	"""
86	function threadpool(tid = threadid())	13✔
87	tpid = ccall(:jl_threadpoolid, Int8, (Int16,), tid-1)	16✔
88	return _tpid_to_sym(tpid)	10✔
89	end
90
91	"""
92	Threads.nthreadpools() -> Int
93
94	Returns the number of threadpools currently configured.
95	"""
96	nthreadpools() = Int(unsafe_load(cglobal(:jl_n_threadpools, Cint)))	1✔
97
98	"""
99	Threads.threadpoolsize(pool::Symbol = :default) -> Int
100
101	Get the number of threads available to the default thread pool (or to the
102	specified thread pool).
103
104	See also: `BLAS.get_num_threads` and `BLAS.set_num_threads` in the
105	[`LinearAlgebra`](@ref man-linalg) standard library, and `nprocs()` in the
106	[`Distributed`](@ref man-distributed) standard library.
107	"""
108	function threadpoolsize(pool::Symbol = :default)	3,744,594✔
109	if pool === :default \|\| pool === :interactive	12,536,643✔
110	tpid = _sym_to_tpid(pool)	19,677,996✔
111	else
112	error("invalid threadpool specified")	×
113	end
114	return _nthreads_in_pool(tpid)	13,969,692✔
115	end
116
117	"""
118	threadpooltids(pool::Symbol)
119
120	Returns a vector of IDs of threads in the given pool.
121	"""
122	function threadpooltids(pool::Symbol)	2✔
123	ni = _nthreads_in_pool(Int8(0))	2✔
124	if pool === :interactive	2✔
125	return collect(1:ni)	1✔
126	elseif pool === :default	1✔
127	return collect(ni+1:ni+_nthreads_in_pool(Int8(1)))	1✔
128	else
129	error("invalid threadpool specified")	×
130	end
131	end
132
133	"""
134	Threads.ngcthreads() -> Int
135
136	Returns the number of GC threads currently configured.
137	This includes both mark threads and concurrent sweep threads.
138	"""
139	ngcthreads() = Int(unsafe_load(cglobal(:jl_n_gcthreads, Cint))) + 1	10✔
140
141	function threading_run(fun, static)	440,632✔
142	ccall(:jl_enter_threaded_region, Cvoid, ())	440,632✔
143	n = threadpoolsize()	440,632✔
144	tid_offset = threadpoolsize(:interactive)	440,632✔
145	tasks = Vector{Task}(undef, n)	440,632✔
146	for i = 1:n	881,263✔
147	t = Task(() -> fun(i)) # pass in tid	2,421,825✔
148	t.sticky = static	1,211,743✔
149	static && ccall(:jl_set_task_tid, Cint, (Any, Cint), t, tid_offset + i-1)	1,211,741✔
150	tasks[i] = t	1,211,741✔
151	schedule(t)	1,211,740✔
152	end	1,982,852✔
153	for i = 1:n	881,264✔
154	Base._wait(tasks[i])	1,211,741✔
155	end	1,982,845✔
156	ccall(:jl_exit_threaded_region, Cvoid, ())	440,632✔
157	failed_tasks = filter!(istaskfailed, tasks)	440,632✔
158	if !isempty(failed_tasks)	440,632✔
159	throw(CompositeException(map(TaskFailedException, failed_tasks)))	12✔
160	end
161	end
162
163	function _threadsfor(iter, lbody, schedule)	138✔
164	lidx = iter.args[1] # index	138✔
165	range = iter.args[2]	138✔
166	quote	138✔
167	local threadsfor_fun	×
168	let range = $(esc(range))	440,641✔
169	function threadsfor_fun(tid = 1; onethread = false)	2,863,173✔
170	r = range # Load into local variable	1,211,383✔
171	lenr = length(r)	1,211,369✔
172	# divide loop iterations among threads
173	if onethread	1,210,921✔
174	tid = 1	×
175	len, rem = lenr, 0	×
176	else
177	len, rem = divrem(lenr, threadpoolsize())	1,211,154✔
178	end
179	# not enough iterations for all the threads?
180	if len == 0	1,210,960✔
181	if tid > rem	67✔
182	return	40✔
183	end
184	len, rem = 1, 0	27✔
185	end
186	# compute this thread's iterations
187	f = firstindex(r) + ((tid-1) * len)	1,211,021✔
188	l = f + len - 1	1,210,235✔
189	# distribute remaining iterations evenly
190	if rem > 0	1,210,119✔
191	if tid <= rem	58✔
192	f = f + (tid-1)	25✔
193	l = l + tid	25✔
194	else
195	f = f + rem	33✔
196	l = l + rem	33✔
197	end
198	end
199	# run this thread's iterations
200	for i = f:l	2,419,861✔
201	local $(esc(lidx)) = @inbounds r[i]	421,051,830✔
202	$(esc(lbody))	1,084,674✔
203	end	422,175,618✔
204	end
205	end
206	if $(schedule === :dynamic \|\| schedule === :default)	440,635✔
207	threading_run(threadsfor_fun, false)	440,608✔
208	elseif ccall(:jl_in_threaded_region, Cint, ()) != 0 # :static	28✔
209	error("`@threads :static` cannot be used concurrently or nested")	4✔
210	else # :static
211	threading_run(threadsfor_fun, true)	24✔
212	end
213	nothing	440,619✔
214	end
215	end
216
217	"""
218	Threads.@threads [schedule] for ... end
219
220	A macro to execute a `for` loop in parallel. The iteration space is distributed to
221	coarse-grained tasks. This policy can be specified by the `schedule` argument. The
222	execution of the loop waits for the evaluation of all iterations.
223
224	See also: [`@spawn`](@ref Threads.@spawn) and
225	`pmap` in [`Distributed`](@ref man-distributed).
226
227	# Extended help
228
229	## Semantics
230
231	Unless stronger guarantees are specified by the scheduling option, the loop executed by
232	`@threads` macro have the following semantics.
233
234	The `@threads` macro executes the loop body in an unspecified order and potentially
235	concurrently. It does not specify the exact assignments of the tasks and the worker threads.
236	The assignments can be different for each execution. The loop body code (including any code
237	transitively called from it) must not make any assumptions about the distribution of
238	iterations to tasks or the worker thread in which they are executed. The loop body for each
239	iteration must be able to make forward progress independent of other iterations and be free
240	from data races. As such, invalid synchronizations across iterations may deadlock while
241	unsynchronized memory accesses may result in undefined behavior.
242
243	For example, the above conditions imply that:
244
245	- A lock taken in an iteration must be released within the same iteration.
246	- Communicating between iterations using blocking primitives like `Channel`s is incorrect.
247	- Write only to locations not shared across iterations (unless a lock or atomic operation is
248	used).
249	- Unless the `:static` schedule is used, the value of [`threadid()`](@ref Threads.threadid)
250	may change even within a single iteration. See [`Task Migration`](@ref man-task-migration).
251
252	## Schedulers
253
254	Without the scheduler argument, the exact scheduling is unspecified and varies across Julia
255	releases. Currently, `:dynamic` is used when the scheduler is not specified.
256
257	!!! compat "Julia 1.5"
258	The `schedule` argument is available as of Julia 1.5.
259
260	### `:dynamic` (default)
261
262	`:dynamic` scheduler executes iterations dynamically to available worker threads. Current
263	implementation assumes that the workload for each iteration is uniform. However, this
264	assumption may be removed in the future.
265
266	This scheduling option is merely a hint to the underlying execution mechanism. However, a
267	few properties can be expected. The number of `Task`s used by `:dynamic` scheduler is
268	bounded by a small constant multiple of the number of available worker threads
269	([`Threads.threadpoolsize()`](@ref)). Each task processes contiguous regions of the
270	iteration space. Thus, `@threads :dynamic for x in xs; f(x); end` is typically more
271	efficient than `@sync for x in xs; @spawn f(x); end` if `length(xs)` is significantly
272	larger than the number of the worker threads and the run-time of `f(x)` is relatively
273	smaller than the cost of spawning and synchronizing a task (typically less than 10
274	microseconds).
275
276	!!! compat "Julia 1.8"
277	The `:dynamic` option for the `schedule` argument is available and the default as of Julia 1.8.
278
279	### `:static`
280
281	`:static` scheduler creates one task per thread and divides the iterations equally among
282	them, assigning each task specifically to each thread. In particular, the value of
283	[`threadid()`](@ref Threads.threadid) is guaranteed to be constant within one iteration.
284	Specifying `:static` is an error if used from inside another `@threads` loop or from a
285	thread other than 1.
286
287	!!! note
288	`:static` scheduling exists for supporting transition of code written before Julia 1.3.
289	In newly written library functions, `:static` scheduling is discouraged because the
290	functions using this option cannot be called from arbitrary worker threads.
291
292	## Example
293
294	To illustrate of the different scheduling strategies, consider the following function
295	`busywait` containing a non-yielding timed loop that runs for a given number of seconds.
296
297	```julia-repl
298	julia> function busywait(seconds)
299	tstart = time_ns()
300	while (time_ns() - tstart) / 1e9 < seconds
301	end
302	end
303
304	julia> @time begin
305	Threads.@spawn busywait(5)
306	Threads.@threads :static for i in 1:Threads.threadpoolsize()
307	busywait(1)
308	end
309	end
310	6.003001 seconds (16.33 k allocations: 899.255 KiB, 0.25% compilation time)
311
312	julia> @time begin
313	Threads.@spawn busywait(5)
314	Threads.@threads :dynamic for i in 1:Threads.threadpoolsize()
315	busywait(1)
316	end
317	end
318	2.012056 seconds (16.05 k allocations: 883.919 KiB, 0.66% compilation time)
319	```
320
321	The `:dynamic` example takes 2 seconds since one of the non-occupied threads is able
322	to run two of the 1-second iterations to complete the for loop.
323	"""
324	macro threads(args...)	145✔
325	na = length(args)	145✔
326	if na == 2	145✔
327	sched, ex = args	37✔
328	if sched isa QuoteNode	37✔
329	sched = sched.value	36✔
330	elseif sched isa Symbol	1✔
331	# for now only allow quoted symbols
332	sched = nothing	×
333	end
334	if sched !== :static && sched !== :dynamic	37✔
335	throw(ArgumentError("unsupported schedule argument in @threads"))	1✔
336	end
337	elseif na == 1	108✔
338	sched = :default	108✔
339	ex = args[1]	108✔
340	else
341	throw(ArgumentError("wrong number of arguments in @threads"))	×
342	end
343	if !(isa(ex, Expr) && ex.head === :for)	145✔
344	throw(ArgumentError("@threads requires a `for` loop expression"))	2✔
345	end
346	if !(ex.args[1] isa Expr && ex.args[1].head === :(=))	142✔
347	throw(ArgumentError("nested outer loops are not currently supported by @threads"))	4✔
348	end
349	return _threadsfor(ex.args[1], ex.args[2], sched)	138✔
350	end
351
352	function _spawn_set_thrpool(t::Task, tp::Symbol)	6,275,888✔
353	tpid = _sym_to_tpid(tp)	10,276,015✔
354	if _nthreads_in_pool(tpid) == 0	6,277,211✔
355	tpid = _sym_to_tpid(:default)	×
356	end
357	ccall(:jl_set_task_threadpoolid, Cint, (Any, Int8), t, tpid)	6,277,210✔
358	nothing	6,277,162✔
359	end
360
361	"""
362	Threads.@spawn [:default\|:interactive] expr
363
364	Create a [`Task`](@ref) and [`schedule`](@ref) it to run on any available
365	thread in the specified threadpool (`:default` if unspecified). The task is
366	allocated to a thread once one becomes available. To wait for the task to
367	finish, call [`wait`](@ref) on the result of this macro, or call
368	[`fetch`](@ref) to wait and then obtain its return value.
369
370	Values can be interpolated into `@spawn` via `\$`, which copies the value
371	directly into the constructed underlying closure. This allows you to insert
372	the _value_ of a variable, isolating the asynchronous code from changes to
373	the variable's value in the current task.
374
375	!!! note
376	The thread that the task runs on may change if the task yields, therefore `threadid()` should not
377	be treated as constant for a task. See [`Task Migration`](@ref man-task-migration), and the broader
378	[multi-threading](@ref man-multithreading) manual for further important caveats.
379	See also the chapter on [threadpools](@ref man-threadpools).
380
381	!!! compat "Julia 1.3"
382	This macro is available as of Julia 1.3.
383
384	!!! compat "Julia 1.4"
385	Interpolating values via `\$` is available as of Julia 1.4.
386
387	!!! compat "Julia 1.9"
388	A threadpool may be specified as of Julia 1.9.
389
390	# Examples
391	```julia-repl
392	julia> t() = println("Hello from ", Threads.threadid());
393
394	julia> tasks = fetch.([Threads.@spawn t() for i in 1:4]);
395	Hello from 1
396	Hello from 1
397	Hello from 3
398	Hello from 4
399	```
400	"""
401	macro spawn(args...)	122✔
402	tp = QuoteNode(:default)	122✔
403	na = length(args)	122✔
404	if na == 2	122✔
405	ttype, ex = args	5✔
406	if ttype isa QuoteNode	5✔
407	ttype = ttype.value	2✔
408	if ttype !== :interactive && ttype !== :default	2✔
409	throw(ArgumentError("unsupported threadpool in @spawn: $ttype"))	×
410	end
411	tp = QuoteNode(ttype)	2✔
412	else
413	tp = ttype	8✔
414	end
415	elseif na == 1	117✔
416	ex = args[1]	117✔
417	else
418	throw(ArgumentError("wrong number of arguments in @spawn"))	×
419	end
420
421	letargs = Base._lift_one_interp!(ex)	122✔
422
423	thunk = Base.replace_linenums!(:(()->($(esc(ex)))), __source__)	122✔
424	var = esc(Base.sync_varname)	122✔
425	quote	122✔
426	let $(letargs...)	411✔
427	local task = Task($thunk)	6,276,964✔
428	task.sticky = false	6,277,659✔
429	_spawn_set_thrpool(task, $(esc(tp)))	6,277,021✔
430	if $(Expr(:islocal, var))	6,275,859✔
431	put!($var, task)	4,998,751✔
432	end
433	schedule(task)	6,276,153✔
434	task	6,274,646✔
435	end
436	end
437	end
438
439	# This is a stub that can be overloaded for downstream structures like `Channel`
440	function foreach end
441
442	# Scheduling traits that can be employed for downstream overloads
443	abstract type AbstractSchedule end
444	struct StaticSchedule <: AbstractSchedule end	8✔
445	struct FairSchedule <: AbstractSchedule end	12✔

JuliaLang / julia / #37579

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