#37735

Committed 02 Apr 2024 02:31AM UTC coverage: 80.448% (-1.0%) from 81.405%

Build # #37735

Build Type

push

local

Committed by

web-flow

Commit Message

documentation followup for "invert linetable representation (#52415)" (#53781)

- fix up added documents: eb05b4f2a
- ~~set up a specific type to capture the 3-set data of `codelocs`:
6afde4b74~~ (moved to a separate PR)

Run Details

69784 of 86744 relevant lines covered (80.45%)

14187382.51 hits per line

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

82.2

/base/channels.jl

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

"""
    AbstractChannel{T}

Representation of a channel passing objects of type `T`.
"""
abstract type AbstractChannel{T} end

push!(c::AbstractChannel, v) = (put!(c, v); c)
popfirst!(c::AbstractChannel) = take!(c)

"""
    Channel{T=Any}(size::Int=0)

Constructs a `Channel` with an internal buffer that can hold a maximum of `size` objects
of type `T`.
[`put!`](@ref) calls on a full channel block until an object is removed with [`take!`](@ref).

`Channel(0)` constructs an unbuffered channel. `put!` blocks until a matching `take!` is called.
And vice-versa.

Other constructors:

* `Channel()`: default constructor, equivalent to `Channel{Any}(0)`
* `Channel(Inf)`: equivalent to `Channel{Any}(typemax(Int))`
* `Channel(sz)`: equivalent to `Channel{Any}(sz)`

!!! compat "Julia 1.3"
    The default constructor `Channel()` and default `size=0` were added in Julia 1.3.
"""
mutable struct Channel{T} <: AbstractChannel{T}
    cond_take::Threads.Condition                 # waiting for data to become available
    cond_wait::Threads.Condition                 # waiting for data to become maybe available
    cond_put::Threads.Condition                  # waiting for a writeable slot
    @atomic state::Symbol
    excp::Union{Exception, Nothing}      # exception to be thrown when state !== :open

    data::Vector{T}
    @atomic n_avail_items::Int           # Available items for taking, can be read without lock
    sz_max::Int                          # maximum size of channel

    function Channel{T}(sz::Integer = 0) where T
        if sz < 0
            throw(ArgumentError("Channel size must be either 0, a positive integer or Inf"))
        end
        lock = ReentrantLock()
        cond_put, cond_take = Threads.Condition(lock), Threads.Condition(lock)
        cond_wait = (sz == 0 ? Threads.Condition(lock) : cond_take) # wait is distinct from take iff unbuffered
        return new(cond_take, cond_wait, cond_put, :open, nothing, Vector{T}(), 0, sz)
    end
end

function Channel{T}(sz::Float64) where T
    sz = (sz == Inf ? typemax(Int) : convert(Int, sz))
    return Channel{T}(sz)
end
Channel(sz=0) = Channel{Any}(sz)

# special constructors
"""
    Channel{T=Any}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing)

Create a new task from `func`, bind it to a new channel of type
`T` and size `size`, and schedule the task, all in a single call.
The channel is automatically closed when the task terminates.

`func` must accept the bound channel as its only argument.

If you need a reference to the created task, pass a `Ref{Task}` object via
the keyword argument `taskref`.

If `spawn=true`, the `Task` created for `func` may be scheduled on another thread
in parallel, equivalent to creating a task via [`Threads.@spawn`](@ref).

If `spawn=true` and the `threadpool` argument is not set, it defaults to `:default`.

If the `threadpool` argument is set (to `:default` or `:interactive`), this implies
that `spawn=true` and the new Task is spawned to the specified threadpool.

Return a `Channel`.

# Examples
```jldoctest
julia> chnl = Channel() do ch
           foreach(i -> put!(ch, i), 1:4)
       end;

julia> typeof(chnl)
Channel{Any}

julia> for i in chnl
           @show i
       end;
i = 1
i = 2
i = 3
i = 4
```

Referencing the created task:

```jldoctest
julia> taskref = Ref{Task}();

julia> chnl = Channel(taskref=taskref) do ch
           println(take!(ch))
       end;

julia> istaskdone(taskref[])
false

julia> put!(chnl, "Hello");
Hello

julia> istaskdone(taskref[])
true
```

!!! compat "Julia 1.3"
    The `spawn=` parameter was added in Julia 1.3. This constructor was added in Julia 1.3.
    In earlier versions of Julia, Channel used keyword arguments to set `size` and `T`, but
    those constructors are deprecated.

!!! compat "Julia 1.9"
    The `threadpool=` argument was added in Julia 1.9.

```jldoctest
julia> chnl = Channel{Char}(1, spawn=true) do ch
           for c in "hello world"
               put!(ch, c)
           end
       end
Channel{Char}(1) (2 items available)

julia> String(collect(chnl))
"hello world"
```
"""
function Channel{T}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing) where T
    chnl = Channel{T}(size)
    task = Task(() -> func(chnl))
    if threadpool === nothing
        threadpool = :default
    else
        spawn = true
    end
    task.sticky = !spawn
    bind(chnl, task)
    if spawn
        Threads._spawn_set_thrpool(task, threadpool)
        schedule(task) # start it on (potentially) another thread
    else
        yield(task) # immediately start it, yielding the current thread
    end
    isa(taskref, Ref{Task}) && (taskref[] = task)
    return chnl
end
Channel(func::Function, args...; kwargs...) = Channel{Any}(func, args...; kwargs...)

# This constructor is deprecated as of Julia v1.3, and should not be used.
# (Note that this constructor also matches `Channel(::Function)` w/out any kwargs, which is
# of course not deprecated.)
# We use `nothing` default values to check which arguments were set in order to throw the
# deprecation warning if users try to use `spawn=` with `ctype=` or `csize=`.
function Channel(func::Function; ctype=nothing, csize=nothing, taskref=nothing, spawn=nothing, threadpool=nothing)
    # The spawn= keyword argument was added in Julia v1.3, and cannot be used with the
    # deprecated keyword arguments `ctype=` or `csize=`.
    if (ctype !== nothing || csize !== nothing) && (spawn !== nothing || threadpool !== nothing)
        throw(ArgumentError("Cannot set `spawn=` or `threadpool=` in the deprecated constructor `Channel(f; ctype=Any, csize=0)`. Please use `Channel{T=Any}(f, size=0; taskref=nothing, spawn=false, threadpool=nothing)` instead!"))
    end
    # Set the actual default values for the arguments.
    ctype === nothing && (ctype = Any)
    csize === nothing && (csize = 0)
    spawn === nothing && (spawn = false)
    return Channel{ctype}(func, csize; taskref=taskref, spawn=spawn, threadpool=threadpool)
end

closed_exception() = InvalidStateException("Channel is closed.", :closed)

isbuffered(c::Channel) = c.sz_max==0 ? false : true

function check_channel_state(c::Channel)
    if !isopen(c)
        # if the monotonic load succeed, now do an acquire fence
        (@atomic :acquire c.state) === :open && concurrency_violation()
        excp = c.excp
        excp !== nothing && throw(excp)
        throw(closed_exception())
    end
end
"""
    close(c::Channel[, excp::Exception])

Close a channel. An exception (optionally given by `excp`), is thrown by:

* [`put!`](@ref) on a closed channel.
* [`take!`](@ref) and [`fetch`](@ref) on an empty, closed channel.
"""
close(c::Channel) = close(c, closed_exception()) # nospecialize on default arg seems to confuse makedocs
function close(c::Channel, @nospecialize(excp::Exception))
    lock(c)
    try
        c.excp = excp
        @atomic :release c.state = :closed
        notify_error(c.cond_take, excp)
        notify_error(c.cond_wait, excp)
        notify_error(c.cond_put, excp)
    finally
        unlock(c)
    end
    nothing
end

# Use acquire here to pair with release store in `close`, so that subsequent `isready` calls
# are forced to see `isready == true` if they see `isopen == false`. This means users must
# call `isopen` before `isready` if you are using the race-y APIs (or call `iterate`, which
# does this right for you).
isopen(c::Channel) = ((@atomic :acquire c.state) === :open)

"""
    empty!(c::Channel)

Empty a Channel `c` by calling `empty!` on the internal buffer.
Return the empty channel.
"""
function Base.empty!(c::Channel)
    @lock c begin
        ndrop = length(c.data)
        empty!(c.data)
        _increment_n_avail(c, -ndrop)
        notify(c.cond_put)
    end
    return c
end

"""
    bind(chnl::Channel, task::Task)

Associate the lifetime of `chnl` with a task.
`Channel` `chnl` is automatically closed when the task terminates.
Any uncaught exception in the task is propagated to all waiters on `chnl`.

The `chnl` object can be explicitly closed independent of task termination.
Terminating tasks have no effect on already closed `Channel` objects.

When a channel is bound to multiple tasks, the first task to terminate will
close the channel. When multiple channels are bound to the same task,
termination of the task will close all of the bound channels.

# Examples
```jldoctest
julia> c = Channel(0);

julia> task = @async foreach(i->put!(c, i), 1:4);

julia> bind(c,task);

julia> for i in c
           @show i
       end;
i = 1
i = 2
i = 3
i = 4

julia> isopen(c)
false
```

```jldoctest
julia> c = Channel(0);

julia> task = @async (put!(c, 1); error("foo"));

julia> bind(c, task);

julia> take!(c)
1

julia> put!(c, 1);
ERROR: TaskFailedException
Stacktrace:
[...]
    nested task error: foo
[...]
```
"""
function bind(c::Channel, task::Task)
    T = Task(() -> close_chnl_on_taskdone(task, c))
    T.sticky = false
    _wait2(task, T)
    return c
end

"""
    channeled_tasks(n::Int, funcs...; ctypes=fill(Any,n), csizes=fill(0,n))

A convenience method to create `n` channels and bind them to tasks started
from the provided functions in a single call. Each `func` must accept `n` arguments
which are the created channels. Channel types and sizes may be specified via
keyword arguments `ctypes` and `csizes` respectively. If unspecified, all channels are
of type `Channel{Any}(0)`.

Returns a tuple, `(Array{Channel}, Array{Task})`, of the created channels and tasks.
"""
function channeled_tasks(n::Int, funcs...; ctypes=fill(Any,n), csizes=fill(0,n))
    @assert length(csizes) == n
    @assert length(ctypes) == n

    chnls = map(i -> Channel{ctypes[i]}(csizes[i]), 1:n)
    tasks = Task[ Task(() -> f(chnls...)) for f in funcs ]

    # bind all tasks to all channels and schedule them
    foreach(t -> foreach(c -> bind(c, t), chnls), tasks)
    foreach(schedule, tasks)
    yield() # Allow scheduled tasks to run

    return (chnls, tasks)
end

function close_chnl_on_taskdone(t::Task, c::Channel)
    isopen(c) || return
    lock(c)
    try
        isopen(c) || return
        if istaskfailed(t)
            close(c, TaskFailedException(t))
            return
        end
        close(c)
    finally
        unlock(c)
    end
    nothing
end

struct InvalidStateException <: Exception
    msg::String
    state::Symbol
end
showerror(io::IO, ex::InvalidStateException) = print(io, "InvalidStateException: ", ex.msg)

"""
    put!(c::Channel, v)

Append an item `v` to the channel `c`. Blocks if the channel is full.

For unbuffered channels, blocks until a [`take!`](@ref) is performed by a different
task.

!!! compat "Julia 1.1"
    `v` now gets converted to the channel's type with [`convert`](@ref) as `put!` is called.
"""
function put!(c::Channel{T}, v) where T
    check_channel_state(c)
    v = convert(T, v)
    return isbuffered(c) ? put_buffered(c, v) : put_unbuffered(c, v)
end

# Atomically update channel n_avail, *assuming* we hold the channel lock.
function _increment_n_avail(c, inc)
    # We hold the channel lock so it's safe to non-atomically read and
    # increment c.n_avail_items
    newlen = c.n_avail_items + inc
    # Atomically store c.n_avail_items to prevent data races with other threads
    # reading this outside the lock.
    @atomic :monotonic c.n_avail_items = newlen
end

function put_buffered(c::Channel, v)
    lock(c)
    did_buffer = false
    try
        # Increment channel n_avail eagerly (before push!) to count data in the
        # buffer as well as offers from tasks which are blocked in wait().
        _increment_n_avail(c, 1)
        while length(c.data) == c.sz_max
            check_channel_state(c)
            wait(c.cond_put)
        end
        check_channel_state(c)
        push!(c.data, v)
        did_buffer = true
        # notify all, since some of the waiters may be on a "fetch" call.
        notify(c.cond_take, nothing, true, false)
    finally
        # Decrement the available items if this task had an exception before pushing the
        # item to the buffer (e.g., during `wait(c.cond_put)`):
        did_buffer || _increment_n_avail(c, -1)
        unlock(c)
    end
    return v
end

function put_unbuffered(c::Channel, v)
    lock(c)
    taker = try
        _increment_n_avail(c, 1)
        while isempty(c.cond_take.waitq)
            check_channel_state(c)
            notify(c.cond_wait)
            wait(c.cond_put)
        end
        check_channel_state(c)
        # unfair scheduled version of: notify(c.cond_take, v, false, false); yield()
        popfirst!(c.cond_take.waitq)
    finally
        _increment_n_avail(c, -1)
        unlock(c)
    end
    schedule(taker, v)
    yield()  # immediately give taker a chance to run, but don't block the current task
    return v
end

"""
    fetch(c::Channel)

Waits for and returns (without removing) the first available item from the `Channel`.
Note: `fetch` is unsupported on an unbuffered (0-size) `Channel`.

# Examples

Buffered channel:
```jldoctest
julia> c = Channel(3) do ch
           foreach(i -> put!(ch, i), 1:3)
       end;

julia> fetch(c)
1

julia> collect(c)  # item is not removed
3-element Vector{Any}:
 1
 2
 3
```
"""
fetch(c::Channel) = isbuffered(c) ? fetch_buffered(c) : fetch_unbuffered(c)
function fetch_buffered(c::Channel)
    lock(c)
    try
        while isempty(c.data)
            check_channel_state(c)
            wait(c.cond_take)
        end
        return c.data[1]
    finally
        unlock(c)
    end
end
fetch_unbuffered(c::Channel) = throw(ErrorException("`fetch` is not supported on an unbuffered Channel."))


"""
    take!(c::Channel)

Removes and returns a value from a [`Channel`](@ref) in order. Blocks until data is available.
For unbuffered channels, blocks until a [`put!`](@ref) is performed by a different task.

# Examples

Buffered channel:
```jldoctest
julia> c = Channel(1);

julia> put!(c, 1);

julia> take!(c)
1
```

Unbuffered channel:
```jldoctest
julia> c = Channel(0);

julia> task = Task(() -> put!(c, 1));

julia> schedule(task);

julia> take!(c)
1
```
"""
take!(c::Channel) = isbuffered(c) ? take_buffered(c) : take_unbuffered(c)
function take_buffered(c::Channel)
    lock(c)
    try
        while isempty(c.data)
            check_channel_state(c)
            wait(c.cond_take)
        end
        v = popfirst!(c.data)
        _increment_n_avail(c, -1)
        notify(c.cond_put, nothing, false, false) # notify only one, since only one slot has become available for a put!.
        return v
    finally
        unlock(c)
    end
end

# 0-size channel
function take_unbuffered(c::Channel{T}) where T
    lock(c)
    try
        check_channel_state(c)
        notify(c.cond_put, nothing, false, false)
        return wait(c.cond_take)::T
    finally
        unlock(c)
    end
end

"""
    isready(c::Channel)

Determines whether a [`Channel`](@ref) has a value stored in it.
Returns immediately, does not block.

For unbuffered channels, return `true` if there are tasks waiting on a [`put!`](@ref).

# Examples

Buffered channel:
```jldoctest
julia> c = Channel(1);

julia> isready(c)
false

julia> put!(c, 1);

julia> isready(c)
true
```

Unbuffered channel:
```jldoctest
julia> c = Channel();

julia> isready(c)  # no tasks waiting to put!
false

julia> task = Task(() -> put!(c, 1));

julia> schedule(task);  # schedule a put! task

julia> isready(c)
true
```

"""
isready(c::Channel) = n_avail(c) > 0
isempty(c::Channel) = n_avail(c) == 0
function n_avail(c::Channel)
    # Lock-free equivalent to `length(c.data) + length(c.cond_put.waitq)`
    @atomic :monotonic c.n_avail_items
end

"""
    isfull(c::Channel)

Determines if a [`Channel`](@ref) is full, in the sense
that calling `put!(c, some_value)` would have blocked.
Returns immediately, does not block.

Note that it may frequently be the case that `put!` will
not block after this returns `true`. Users must take
precautions not to accidentally create live-lock bugs
in their code by calling this method, as these are
generally harder to debug than deadlocks. It is also
possible that `put!` will block after this call
returns `false`, if there are multiple producer
tasks calling `put!` in parallel.

# Examples

Buffered channel:
```jldoctest
julia> c = Channel(1); # capacity = 1

julia> isfull(c)
false

julia> put!(c, 1);

julia> isfull(c)
true
```

Unbuffered channel:
```jldoctest
julia> c = Channel(); # capacity = 0

julia> isfull(c) # unbuffered channel is always full
true
```
"""
isfull(c::Channel) = n_avail(c) ≥ c.sz_max

lock(c::Channel) = lock(c.cond_take)
lock(f, c::Channel) = lock(f, c.cond_take)
unlock(c::Channel) = unlock(c.cond_take)
trylock(c::Channel) = trylock(c.cond_take)

"""
    wait(c::Channel)

Blocks until the `Channel` [`isready`](@ref).

```jldoctest
julia> c = Channel(1);

julia> isready(c)
false

julia> task = Task(() -> wait(c));

julia> schedule(task);

julia> istaskdone(task)  # task is blocked because channel is not ready
false

julia> put!(c, 1);

julia> istaskdone(task)  # task is now unblocked
true
```
"""
function wait(c::Channel)
    isready(c) && return
    lock(c)
    try
        while !isready(c)
            check_channel_state(c)
            wait(c.cond_wait)
        end
    finally
        unlock(c)
    end
    nothing
end

eltype(::Type{Channel{T}}) where {T} = T

show(io::IO, c::Channel) = print(io, typeof(c), "(", c.sz_max, ")")

function show(io::IO, ::MIME"text/plain", c::Channel)
    show(io, c)
    if !(get(io, :compact, false)::Bool)
        if !isopen(c)
            print(io, " (closed)")
        else
            n = n_avail(c)
            if n == 0
                print(io, " (empty)")
            else
                s = n == 1 ? "" : "s"
                print(io, " (", n, " item$s available)")
            end
        end
    end
end

function iterate(c::Channel, state=nothing)
    if isopen(c) || isready(c)
        try
            return (take!(c), nothing)
        catch e
            if isa(e, InvalidStateException) && e.state === :closed
                return nothing
            else
                rethrow()
            end
        end
    else
        return nothing
    end
end

IteratorSize(::Type{<:Channel}) = SizeUnknown()

1	# This file is a part of Julia. License is MIT: https://julialang.org/license
2
3	"""
4	AbstractChannel{T}
5
6	Representation of a channel passing objects of type `T`.
7	"""
8	abstract type AbstractChannel{T} end
9
10	push!(c::AbstractChannel, v) = (put!(c, v); c)	5,270✔
11	popfirst!(c::AbstractChannel) = take!(c)	×
12
13	"""
14	Channel{T=Any}(size::Int=0)
15
16	Constructs a `Channel` with an internal buffer that can hold a maximum of `size` objects
17	of type `T`.
18	[`put!`](@ref) calls on a full channel block until an object is removed with [`take!`](@ref).
19
20	`Channel(0)` constructs an unbuffered channel. `put!` blocks until a matching `take!` is called.
21	And vice-versa.
22
23	Other constructors:
24
25	* `Channel()`: default constructor, equivalent to `Channel{Any}(0)`
26	* `Channel(Inf)`: equivalent to `Channel{Any}(typemax(Int))`
27	* `Channel(sz)`: equivalent to `Channel{Any}(sz)`
28
29	!!! compat "Julia 1.3"
30	The default constructor `Channel()` and default `size=0` were added in Julia 1.3.
31	"""
32	mutable struct Channel{T} <: AbstractChannel{T}
33	cond_take::Threads.Condition # waiting for data to become available
34	cond_wait::Threads.Condition # waiting for data to become maybe available
35	cond_put::Threads.Condition # waiting for a writeable slot
36	@atomic state::Symbol
37	excp::Union{Exception, Nothing} # exception to be thrown when state !== :open
38
39	data::Vector{T}
40	@atomic n_avail_items::Int # Available items for taking, can be read without lock
41	sz_max::Int # maximum size of channel
42
43	function Channel{T}(sz::Integer = 0) where T	3✔
44	if sz < 0	721✔
45	throw(ArgumentError("Channel size must be either 0, a positive integer or Inf"))	×
46	end
47	lock = ReentrantLock()	1,956✔
48	cond_put, cond_take = Threads.Condition(lock), Threads.Condition(lock)	1,956✔
49	cond_wait = (sz == 0 ? Threads.Condition(lock) : cond_take) # wait is distinct from take iff unbuffered	720✔
50	return new(cond_take, cond_wait, cond_put, :open, nothing, Vector{T}(), 0, sz)	1,956✔
51	end
52	end
53
54	function Channel{T}(sz::Float64) where T
55	sz = (sz == Inf ? typemax(Int) : convert(Int, sz))	406✔
56	return Channel{T}(sz)	1,314✔
57	end
58	Channel(sz=0) = Channel{Any}(sz)	946✔
59
60	# special constructors
61	"""
62	Channel{T=Any}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing)
63
64	Create a new task from `func`, bind it to a new channel of type
65	`T` and size `size`, and schedule the task, all in a single call.
66	The channel is automatically closed when the task terminates.
67
68	`func` must accept the bound channel as its only argument.
69
70	If you need a reference to the created task, pass a `Ref{Task}` object via
71	the keyword argument `taskref`.
72
73	If `spawn=true`, the `Task` created for `func` may be scheduled on another thread
74	in parallel, equivalent to creating a task via [`Threads.@spawn`](@ref).
75
76	If `spawn=true` and the `threadpool` argument is not set, it defaults to `:default`.
77
78	If the `threadpool` argument is set (to `:default` or `:interactive`), this implies
79	that `spawn=true` and the new Task is spawned to the specified threadpool.
80
81	Return a `Channel`.
82
83	# Examples
84	```jldoctest
85	julia> chnl = Channel() do ch
86	foreach(i -> put!(ch, i), 1:4)
87	end;
88
89	julia> typeof(chnl)
90	Channel{Any}
91
92	julia> for i in chnl
93	@show i
94	end;
95	i = 1
96	i = 2
97	i = 3
98	i = 4
99	```
100
101	Referencing the created task:
102
103	```jldoctest
104	julia> taskref = Ref{Task}();
105
106	julia> chnl = Channel(taskref=taskref) do ch
107	println(take!(ch))
108	end;
109
110	julia> istaskdone(taskref[])
111	false
112
113	julia> put!(chnl, "Hello");
114	Hello
115
116	julia> istaskdone(taskref[])
117	true
118	```
119
120	!!! compat "Julia 1.3"
121	The `spawn=` parameter was added in Julia 1.3. This constructor was added in Julia 1.3.
122	In earlier versions of Julia, Channel used keyword arguments to set `size` and `T`, but
123	those constructors are deprecated.
124
125	!!! compat "Julia 1.9"
126	The `threadpool=` argument was added in Julia 1.9.
127
128	```jldoctest
129	julia> chnl = Channel{Char}(1, spawn=true) do ch
130	for c in "hello world"
131	put!(ch, c)
132	end
133	end
134	Channel{Char}(1) (2 items available)
135
136	julia> String(collect(chnl))
137	"hello world"
138	```
139	"""
140	function Channel{T}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing) where T	725✔
141	chnl = Channel{T}(size)	217✔
142	task = Task(() -> func(chnl))	434✔
143	if threadpool === nothing	28✔
144	threadpool = :default	28✔
145	else
146	spawn = true	×
147	end
148	task.sticky = !spawn	217✔
149	bind(chnl, task)	217✔
150	if spawn	217✔
151	Threads._spawn_set_thrpool(task, threadpool)	×
152	schedule(task) # start it on (potentially) another thread	×
153	else
154	yield(task) # immediately start it, yielding the current thread	217✔
155	end
156	isa(taskref, Ref{Task}) && (taskref[] = task)	28✔
157	return chnl	217✔
158	end
159	Channel(func::Function, args...; kwargs...) = Channel{Any}(func, args...; kwargs...)	×
160
161	# This constructor is deprecated as of Julia v1.3, and should not be used.
162	# (Note that this constructor also matches `Channel(::Function)` w/out any kwargs, which is
163	# of course not deprecated.)
164	# We use `nothing` default values to check which arguments were set in order to throw the
165	# deprecation warning if users try to use `spawn=` with `ctype=` or `csize=`.
166	function Channel(func::Function; ctype=nothing, csize=nothing, taskref=nothing, spawn=nothing, threadpool=nothing)	28✔
167	# The spawn= keyword argument was added in Julia v1.3, and cannot be used with the
168	# deprecated keyword arguments `ctype=` or `csize=`.
169	if (ctype !== nothing \|\| csize !== nothing) && (spawn !== nothing \|\| threadpool !== nothing)	28✔
170	throw(ArgumentError("Cannot set `spawn=` or `threadpool=` in the deprecated constructor `Channel(f; ctype=Any, csize=0)`. Please use `Channel{T=Any}(f, size=0; taskref=nothing, spawn=false, threadpool=nothing)` instead!"))	×
171	end
172	# Set the actual default values for the arguments.
173	ctype === nothing && (ctype = Any)	28✔
174	csize === nothing && (csize = 0)	28✔
175	spawn === nothing && (spawn = false)	28✔
176	return Channel{ctype}(func, csize; taskref=taskref, spawn=spawn, threadpool=threadpool)	28✔
177	end
178
179	closed_exception() = InvalidStateException("Channel is closed.", :closed)	1,609✔
180
181	isbuffered(c::Channel) = c.sz_max==0 ? false : true	192,613✔
182
183	function check_channel_state(c::Channel)
184	if !isopen(c)	203,061✔
185	# if the monotonic load succeed, now do an acquire fence
186	(@atomic :acquire c.state) === :open && concurrency_violation()	3✔
187	excp = c.excp	3✔
188	excp !== nothing && throw(excp)	3✔
189	throw(closed_exception())	×
190	end
191	end
192	"""
193	close(c::Channel[, excp::Exception])
194
195	Close a channel. An exception (optionally given by `excp`), is thrown by:
196
197	* [`put!`](@ref) on a closed channel.
198	* [`take!`](@ref) and [`fetch`](@ref) on an empty, closed channel.
199	"""
200	close(c::Channel) = close(c, closed_exception()) # nospecialize on default arg seems to confuse makedocs	1,609✔
201	function close(c::Channel, @nospecialize(excp::Exception))	1,556✔
202	lock(c)	1,556✔
203	try	1,556✔
204	c.excp = excp	1,556✔
205	@atomic :release c.state = :closed	1,556✔
206	notify_error(c.cond_take, excp)	1,556✔
207	notify_error(c.cond_wait, excp)	1,556✔
208	notify_error(c.cond_put, excp)	1,556✔
209	finally
210	unlock(c)	1,556✔
211	end
212	nothing
213	end
214
215	# Use acquire here to pair with release store in `close`, so that subsequent `isready` calls
216	# are forced to see `isready == true` if they see `isopen == false`. This means users must
217	# call `isopen` before `isready` if you are using the race-y APIs (or call `iterate`, which
218	# does this right for you).
219	isopen(c::Channel) = ((@atomic :acquire c.state) === :open)	228,069✔
220
221	"""
222	empty!(c::Channel)
223
224	Empty a Channel `c` by calling `empty!` on the internal buffer.
225	Return the empty channel.
226	"""
227	function Base.empty!(c::Channel)	×
228	@lock c begin	×
229	ndrop = length(c.data)	×
230	empty!(c.data)	×
231	_increment_n_avail(c, -ndrop)	×
232	notify(c.cond_put)	×
233	end
234	return c	×
235	end
236
237	"""
238	bind(chnl::Channel, task::Task)
239
240	Associate the lifetime of `chnl` with a task.
241	`Channel` `chnl` is automatically closed when the task terminates.
242	Any uncaught exception in the task is propagated to all waiters on `chnl`.
243
244	The `chnl` object can be explicitly closed independent of task termination.
245	Terminating tasks have no effect on already closed `Channel` objects.
246
247	When a channel is bound to multiple tasks, the first task to terminate will
248	close the channel. When multiple channels are bound to the same task,
249	termination of the task will close all of the bound channels.
250
251	# Examples
252	```jldoctest
253	julia> c = Channel(0);
254
255	julia> task = @async foreach(i->put!(c, i), 1:4);
256
257	julia> bind(c,task);
258
259	julia> for i in c
260	@show i
261	end;
262	i = 1
263	i = 2
264	i = 3
265	i = 4
266
267	julia> isopen(c)
268	false
269	```
270
271	```jldoctest
272	julia> c = Channel(0);
273
274	julia> task = @async (put!(c, 1); error("foo"));
275
276	julia> bind(c, task);
277
278	julia> take!(c)
279	1
280
281	julia> put!(c, 1);
282	ERROR: TaskFailedException
283	Stacktrace:
284	[...]
285	nested task error: foo
286	[...]
287	```
288	"""
289	function bind(c::Channel, task::Task)
290	T = Task(() -> close_chnl_on_taskdone(task, c))	426✔
291	T.sticky = false	219✔
292	_wait2(task, T)	219✔
293	return c	30✔
294	end
295
296	"""
297	channeled_tasks(n::Int, funcs...; ctypes=fill(Any,n), csizes=fill(0,n))
298
299	A convenience method to create `n` channels and bind them to tasks started
300	from the provided functions in a single call. Each `func` must accept `n` arguments
301	which are the created channels. Channel types and sizes may be specified via
302	keyword arguments `ctypes` and `csizes` respectively. If unspecified, all channels are
303	of type `Channel{Any}(0)`.
304
305	Returns a tuple, `(Array{Channel}, Array{Task})`, of the created channels and tasks.
306	"""
307	function channeled_tasks(n::Int, funcs...; ctypes=fill(Any,n), csizes=fill(0,n))	2✔
308	@assert length(csizes) == n	1✔
309	@assert length(ctypes) == n	1✔
310
311	chnls = map(i -> Channel{ctypes[i]}(csizes[i]), 1:n)	3✔
312	tasks = Task[ Task(() -> f(chnls...)) for f in funcs ]	2✔
313
314	# bind all tasks to all channels and schedule them
315	foreach(t -> foreach(c -> bind(c, t), chnls), tasks)	4✔
316	foreach(schedule, tasks)	1✔
317	yield() # Allow scheduled tasks to run	1✔
318
319	return (chnls, tasks)	1✔
320	end
321
322	function close_chnl_on_taskdone(t::Task, c::Channel)	207✔
323	isopen(c) \|\| return	209✔
324	lock(c)	205✔
325	try	205✔
326	isopen(c) \|\| return	205✔
327	if istaskfailed(t)	205✔
328	close(c, TaskFailedException(t))	2✔
329	return	2✔
330	end
331	close(c)	203✔
332	finally
333	unlock(c)	205✔
334	end
335	nothing
336	end
337
338	struct InvalidStateException <: Exception
339	msg::String	1,609✔
340	state::Symbol
341	end
342	showerror(io::IO, ex::InvalidStateException) = print(io, "InvalidStateException: ", ex.msg)	×
343
344	"""
345	put!(c::Channel, v)
346
347	Append an item `v` to the channel `c`. Blocks if the channel is full.
348
349	For unbuffered channels, blocks until a [`take!`](@ref) is performed by a different
350	task.
351
352	!!! compat "Julia 1.1"
353	`v` now gets converted to the channel's type with [`convert`](@ref) as `put!` is called.
354	"""
355	function put!(c::Channel{T}, v) where T	68,501✔
356	check_channel_state(c)	96,932✔
357	v = convert(T, v)	5,121✔
358	return isbuffered(c) ? put_buffered(c, v) : put_unbuffered(c, v)	100,598✔
359	end
360
361	# Atomically update channel n_avail, assuming we hold the channel lock.
362	function _increment_n_avail(c, inc)
363	# We hold the channel lock so it's safe to non-atomically read and
364	# increment c.n_avail_items
365	newlen = c.n_avail_items + inc	189,655✔
366	# Atomically store c.n_avail_items to prevent data races with other threads
367	# reading this outside the lock.
368	@atomic :monotonic c.n_avail_items = newlen	189,655✔
369	end
370
371	function put_buffered(c::Channel, v)	92,786✔
372	lock(c)	92,786✔
373	did_buffer = false	3,737✔
374	try	92,786✔
375	# Increment channel n_avail eagerly (before push!) to count data in the
376	# buffer as well as offers from tasks which are blocked in wait().
377	_increment_n_avail(c, 1)	92,786✔
378	while length(c.data) == c.sz_max	92,800✔
379	check_channel_state(c)	14✔
380	wait(c.cond_put)	14✔
381	end	14✔
382	check_channel_state(c)	92,786✔
383	push!(c.data, v)	92,786✔
384	did_buffer = true	92,786✔
385	# notify all, since some of the waiters may be on a "fetch" call.
386	notify(c.cond_take, nothing, true, false)	92,786✔
387	finally
388	# Decrement the available items if this task had an exception before pushing the
389	# item to the buffer (e.g., during `wait(c.cond_put)`):
390	did_buffer \|\| _increment_n_avail(c, -1)	92,786✔
391	unlock(c)	92,786✔
392	end
393	return v	92,786✔
394	end
395
396	function put_unbuffered(c::Channel, v)	3,162✔
397	lock(c)	3,163✔
398	taker = try	3,162✔
399	_increment_n_avail(c, 1)	3,162✔
400	while isempty(c.cond_take.waitq)	3,444✔
401	check_channel_state(c)	294✔
402	notify(c.cond_wait)	294✔
403	wait(c.cond_put)	294✔
404	end	282✔
405	check_channel_state(c)	3,150✔
406	# unfair scheduled version of: notify(c.cond_take, v, false, false); yield()
407	popfirst!(c.cond_take.waitq)	4,319✔
408	finally
409	_increment_n_avail(c, -1)	3,150✔
410	unlock(c)	3,150✔
411	end
412	schedule(taker, v)	3,150✔
413	yield() # immediately give taker a chance to run, but don't block the current task	3,150✔
414	return v	3,150✔
415	end
416
417	"""
418	fetch(c::Channel)
419
420	Waits for and returns (without removing) the first available item from the `Channel`.
421	Note: `fetch` is unsupported on an unbuffered (0-size) `Channel`.
422
423	# Examples
424
425	Buffered channel:
426	```jldoctest
427	julia> c = Channel(3) do ch
428	foreach(i -> put!(ch, i), 1:3)
429	end;
430
431	julia> fetch(c)
432	1
433
434	julia> collect(c) # item is not removed
435	3-element Vector{Any}:
436	1
437	2
438	3
439	```
440	"""
441	fetch(c::Channel) = isbuffered(c) ? fetch_buffered(c) : fetch_unbuffered(c)	67✔
442	function fetch_buffered(c::Channel)	67✔
443	lock(c)	67✔
444	try	67✔
445	while isempty(c.data)	122✔
446	check_channel_state(c)	55✔
447	wait(c.cond_take)	55✔
448	end	55✔
449	return c.data[1]	67✔
450	finally
451	unlock(c)	67✔
452	end
453	end
454	fetch_unbuffered(c::Channel) = throw(ErrorException("`fetch` is not supported on an unbuffered Channel."))	×
455
456
457	"""
458	take!(c::Channel)
459
460	Removes and returns a value from a [`Channel`](@ref) in order. Blocks until data is available.
461	For unbuffered channels, blocks until a [`put!`](@ref) is performed by a different task.
462
463	# Examples
464
465	Buffered channel:
466	```jldoctest
467	julia> c = Channel(1);
468
469	julia> put!(c, 1);
470
471	julia> take!(c)
472	1
473	```
474
475	Unbuffered channel:
476	```jldoctest
477	julia> c = Channel(0);
478
479	julia> task = Task(() -> put!(c, 1));
480
481	julia> schedule(task);
482
483	julia> take!(c)
484	1
485	```
486	"""
487	take!(c::Channel) = isbuffered(c) ? take_buffered(c) : take_unbuffered(c)	99,658✔
488	function take_buffered(c::Channel)	90,741✔
489	lock(c)	90,741✔
490	try	90,741✔
491	while isempty(c.data)	94,038✔
492	check_channel_state(c)	5,483✔
493	wait(c.cond_take)	5,483✔
494	end	5,299✔
495	v = popfirst!(c.data)	90,557✔
496	_increment_n_avail(c, -1)	90,557✔
497	notify(c.cond_put, nothing, false, false) # notify only one, since only one slot has become available for a put!.	90,557✔
498	return v	90,740✔
499	finally
500	unlock(c)	90,740✔
501	end
502	end
503
504	# 0-size channel
505	function take_unbuffered(c::Channel{T}) where T	4,341✔
506	lock(c)	4,341✔
507	try	4,341✔
508	check_channel_state(c)	4,341✔
509	notify(c.cond_put, nothing, false, false)	4,340✔
510	return wait(c.cond_take)::T	5,531✔
511	finally
512	unlock(c)	4,341✔
513	end
514	end
515
516	"""
517	isready(c::Channel)
518
519	Determines whether a [`Channel`](@ref) has a value stored in it.
520	Returns immediately, does not block.
521
522	For unbuffered channels, return `true` if there are tasks waiting on a [`put!`](@ref).
523
524	# Examples
525
526	Buffered channel:
527	```jldoctest
528	julia> c = Channel(1);
529
530	julia> isready(c)
531	false
532
533	julia> put!(c, 1);
534
535	julia> isready(c)
536	true
537	```
538
539	Unbuffered channel:
540	```jldoctest
541	julia> c = Channel();
542
543	julia> isready(c) # no tasks waiting to put!
544	false
545
546	julia> task = Task(() -> put!(c, 1));
547
548	julia> schedule(task); # schedule a put! task
549
550	julia> isready(c)
551	true
552	```
553
554	"""
555	isready(c::Channel) = n_avail(c) > 0	68,989✔
556	isempty(c::Channel) = n_avail(c) == 0	×
557	function n_avail(c::Channel)
558	# Lock-free equivalent to `length(c.data) + length(c.cond_put.waitq)`
559	@atomic :monotonic c.n_avail_items	68,989✔
560	end
561
562	"""
563	isfull(c::Channel)
564
565	Determines if a [`Channel`](@ref) is full, in the sense
566	that calling `put!(c, some_value)` would have blocked.
567	Returns immediately, does not block.
568
569	Note that it may frequently be the case that `put!` will
570	not block after this returns `true`. Users must take
571	precautions not to accidentally create live-lock bugs
572	in their code by calling this method, as these are
573	generally harder to debug than deadlocks. It is also
574	possible that `put!` will block after this call
575	returns `false`, if there are multiple producer
576	tasks calling `put!` in parallel.
577
578	# Examples
579
580	Buffered channel:
581	```jldoctest
582	julia> c = Channel(1); # capacity = 1
583
584	julia> isfull(c)
585	false
586
587	julia> put!(c, 1);
588
589	julia> isfull(c)
590	true
591	```
592
593	Unbuffered channel:
594	```jldoctest
595	julia> c = Channel(); # capacity = 0
596
597	julia> isfull(c) # unbuffered channel is always full
598	true
599	```
600	"""
601	isfull(c::Channel) = n_avail(c) ≥ c.sz_max	×
602
603	lock(c::Channel) = lock(c.cond_take)	192,865✔
604	lock(f, c::Channel) = lock(f, c.cond_take)	×
605	unlock(c::Channel) = unlock(c.cond_take)	192,851✔
606	trylock(c::Channel) = trylock(c.cond_take)	×
607
608	"""
609	wait(c::Channel)
610
611	Blocks until the `Channel` [`isready`](@ref).
612
613	```jldoctest
614	julia> c = Channel(1);
615
616	julia> isready(c)
617	false
618
619	julia> task = Task(() -> wait(c));
620
621	julia> schedule(task);
622
623	julia> istaskdone(task) # task is blocked because channel is not ready
624	false
625
626	julia> put!(c, 1);
627
628	julia> istaskdone(task) # task is now unblocked
629	true
630	```
631	"""
632	function wait(c::Channel)	6✔
633	isready(c) && return	6✔
634	lock(c)	6✔
635	try	6✔
636	while !isready(c)	11✔
637	check_channel_state(c)	6✔
638	wait(c.cond_wait)	6✔
639	end	6✔
640	finally
641	unlock(c)	6✔
642	end
643	nothing
644	end
645
646	eltype(::Type{Channel{T}}) where {T} = T	×
647
648	show(io::IO, c::Channel) = print(io, typeof(c), "(", c.sz_max, ")")	×
649
650	function show(io::IO, ::MIME"text/plain", c::Channel)	×
651	show(io, c)	×
652	if !(get(io, :compact, false)::Bool)	×
653	if !isopen(c)	×
654	print(io, " (closed)")	×
655	else
656	n = n_avail(c)	×
657	if n == 0	×
658	print(io, " (empty)")	×
659	else
660	s = n == 1 ? "" : "s"	×
661	print(io, " (", n, " item$s available)")	×
662	end
663	end
664	end
665	end
666
667	function iterate(c::Channel, state=nothing)	24,560✔
668	if isopen(c) \|\| isready(c)	26,831✔
669	try	24,263✔
670	return (take!(c), nothing)	24,263✔
671	catch e
672	if isa(e, InvalidStateException) && e.state === :closed	1,371✔
673	return nothing	1,371✔
674	else
675	rethrow()	×
676	end
677	end
678	else
679	return nothing	291✔
680	end
681	end
682
683	IteratorSize(::Type{<:Channel}) = SizeUnknown()	7✔

JuliaLang / julia / #37735

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