#37861

Committed 05 Aug 2024 06:07AM UTC coverage: 87.594% (+0.004%) from 87.59%

Build # #37861

Build Type

push

local

Committed by

web-flow

Commit Message

Round-trippable show for `Bidiagonal` (#55347)

This changes how a `Bidiagonal` is displayed using the 2-arg `show` to a
form that may be parsed.
After this,
```julia
julia> B = Bidiagonal([1,2,3], [1,2], :U)
3×3 Bidiagonal{Int64, Vector{Int64}}:
 1  1  ⋅
 ⋅  2  2
 ⋅  ⋅  3

julia> show(B)
Bidiagonal([1, 2, 3], [1, 2], :U)
```
The displayed form is a valid constructor now.

Run Details

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

112 existing lines in 5 files now uncovered.

77773 of 88788 relevant lines covered (87.59%)

16184721.47 hits per line

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

61.9

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

JuliaLang / julia / #37861

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