-
-
Notifications
You must be signed in to change notification settings - Fork 5.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #12264 from JuliaLang/amitm/channels_only
Added inter-task communication channels
- Loading branch information
Showing
13 changed files
with
325 additions
and
70 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| # This file is a part of Julia. License is MIT: http://julialang.org/license | ||
|
|
||
| abstract AbstractChannel{T} | ||
|
|
||
| type Channel{T} <: AbstractChannel{T} | ||
| cond_take::Condition # waiting for data to become available | ||
| cond_put::Condition # waiting for a writeable slot | ||
| state::Symbol | ||
|
|
||
| data::Array{T,1} | ||
| szp1::Int # current channel size plus one | ||
| sz_max::Int # maximum size of channel | ||
| take_pos::Int # read position | ||
| put_pos::Int # write position | ||
|
|
||
| function Channel(sz) | ||
| sz_max = sz == typemax(Int) ? typemax(Int) - 1 : sz | ||
| szp1 = sz > 32 ? 33 : sz+1 | ||
| new(Condition(), Condition(), :open, | ||
| Array(T, szp1), szp1, sz_max, 1, 1) | ||
| end | ||
| end | ||
|
|
||
| const DEF_CHANNEL_SZ=32 | ||
|
|
||
| Channel() = Channel(DEF_CHANNEL_SZ) | ||
| Channel(sz::Int) = Channel{Any}(sz) | ||
|
|
||
| closed_exception() = InvalidStateException("Channel is closed.", :closed) | ||
| function close(c::Channel) | ||
| c.state = :closed | ||
| notify_error(c::Channel, closed_exception()) | ||
| c | ||
| end | ||
| isopen(c::Channel) = (c.state == :open) | ||
|
|
||
| type InvalidStateException <: Exception | ||
| msg::AbstractString | ||
| state | ||
| end | ||
| InvalidStateException() = InvalidStateException("") | ||
| InvalidStateException(msg) = InvalidStateException(msg, 0) | ||
|
|
||
| function put!(c::Channel, v) | ||
| !isopen(c) && throw(closed_exception()) | ||
| d = c.take_pos - c.put_pos | ||
| if (d == 1) || (d == -(c.szp1-1)) | ||
| # grow the channel if possible | ||
| if (c.szp1 - 1) < c.sz_max | ||
| if ((c.szp1-1) * 2) > c.sz_max | ||
| c.szp1 = c.sz_max + 1 | ||
| else | ||
| c.szp1 = ((c.szp1-1) * 2) + 1 | ||
| end | ||
| newdata = Array(eltype(c), c.szp1) | ||
| if c.put_pos > c.take_pos | ||
| copy!(newdata, 1, c.data, c.take_pos, (c.put_pos - c.take_pos)) | ||
| c.put_pos = c.put_pos - c.take_pos + 1 | ||
| else | ||
| len_first_part = length(c.data) - c.take_pos + 1 | ||
| copy!(newdata, 1, c.data, c.take_pos, len_first_part) | ||
| copy!(newdata, len_first_part+1, c.data, 1, c.put_pos-1) | ||
| c.put_pos = len_first_part + c.put_pos | ||
| end | ||
| c.take_pos = 1 | ||
| c.data = newdata | ||
| else | ||
| wait(c.cond_put) | ||
| end | ||
| end | ||
|
|
||
| c.data[c.put_pos] = v | ||
| c.put_pos = (c.put_pos == c.szp1 ? 1 : c.put_pos + 1) | ||
| notify(c.cond_take, nothing, true, false) # notify all, since some of the waiters may be on a "fetch" call. | ||
| v | ||
| end | ||
|
|
||
| function fetch(c::Channel) | ||
| wait(c) | ||
| c.data[c.take_pos] | ||
| end | ||
|
|
||
| function take!(c::Channel) | ||
| !isopen(c) && !isready(c) && throw(closed_exception()) | ||
| while !isready(c) | ||
| wait(c.cond_take) | ||
| end | ||
| v = c.data[c.take_pos] | ||
| c.take_pos = (c.take_pos == c.szp1 ? 1 : c.take_pos + 1) | ||
| notify(c.cond_put, nothing, false, false) # notify only one, since only one slot has become available for a put!. | ||
| v | ||
| end | ||
|
|
||
| isready(c::Channel) = (c.take_pos == c.put_pos ? false : true) | ||
|
|
||
| function wait(c::Channel) | ||
| while !isready(c) | ||
| wait(c.cond_take) | ||
| end | ||
| nothing | ||
| end | ||
|
|
||
| function notify_error(c::Channel, err) | ||
| notify_error(c.cond_take, err) | ||
| notify_error(c.cond_put, err) | ||
| end | ||
|
|
||
| eltype{T}(c::Channel{T}) = T | ||
|
|
||
| function length(c::Channel) | ||
| if c.put_pos >= c.take_pos | ||
| return c.put_pos - c.take_pos | ||
| else | ||
| return c.szp1 - c.take_pos + c.put_pos | ||
| end | ||
| end | ||
|
|
||
| size(c::Channel) = c.sz_max | ||
|
|
||
| show(io::IO, c::Channel) = print(io, "$(typeof(c))(sz_max:$(size(c)),sz_curr:$(length(c)))") | ||
|
|
||
| start{T}(c::Channel{T}) = Ref{Nullable{T}}(Nullable{T}()) | ||
| function done(c::Channel, state::Ref) | ||
| try | ||
| # we are waiting either for more data or channel to be closed | ||
| state.x = take!(c) | ||
| return false | ||
| catch e | ||
| if isa(e, InvalidStateException) && e.state==:closed | ||
| return true | ||
| else | ||
| rethrow(e) | ||
| end | ||
| end | ||
| end | ||
| next{T}(c::Channel{T}, state) = (get(state.x), Ref{Nullable{T}}(Nullable{T}())) | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.