added tests, added channeled_tasks

Author: amitmurthy

base/channels.jl

@@ -20,7 +20,7 @@ Other constructors:
 type Channel{T} <: AbstractChannel
     cond_take::Condition    # waiting for data to become available
     cond_put::Condition     # waiting for a writeable slot
-    state::Symbol
+    state::Tuple
 
     data::Array{T,1}
     sz_max::Int            # maximum size of channel
@@ -39,7 +39,7 @@ type Channel{T} <: AbstractChannel
         if sz < 0
             throw(ArgumentError("Channel size must be either 0, a positive integer or Inf"))
         end
-        new(Condition(), Condition(), :open, Array{T}(0), sz, Array{Condition}(0))
+        new(Condition(), Condition(), (:open,nothing), Array{T}(0), sz, Array{Condition}(0))
     end
 
     # deprecated empty constructor
@@ -60,6 +60,12 @@ 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)
+        isa(c.state[2], Exception) && throw(c.state[2])
+        throw(closed_exception())
+    end
+end
 """
     close(c::Channel)
 
@@ -69,27 +75,65 @@ Closes a channel. An exception is thrown by:
 * [`take!`](@ref) and [`fetch`](@ref) on an empty, closed channel.
 """
 function close(c::Channel)
-    c.state = :closed
+    c.state = (:closed, nothing)
     notify_error(c::Channel, closed_exception())
     nothing
 end
-isopen(c::Channel) = (c.state == :open)
+isopen(c::Channel) = (c.state[1] == :open)
+
+"""
+    bind(chnl::Channel, task::Task)
 
-# Associates the lifetime of the Channel to a Task.
-# Closes the channel automatically when the task terminates.
-# Any error in the task is also propagated to waiters on the Channel.
-function close(c::Channel, t::Task)
+Associates 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 channels.
+"""
+function bind(c::Channel, task::Task)
     ref = WeakRef(c)
-    register_taskdone_hook(t, tsk->close_chnl_on_taskdone(tsk, ref))
-    nothing
+    register_taskdone_hook(task, tsk->close_chnl_on_taskdone(tsk, ref))
+    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(t->schedule(t), tasks)
+
+    yield()  # Allow scheduled tasks to run
+
+    return (chnls, tasks)
 end
 
 function close_chnl_on_taskdone(t::Task, ref::WeakRef)
     if ref.value !== nothing
         c = ref.value
         !isopen(c) && return
         if istaskfailed(t)
-            c.state = :closed
+            c.state = (:closed, task_result(t))
             notify_error(c, task_result(t))
         else
             close(c)
@@ -111,8 +155,10 @@ For unbuffered channels, blocks until a [`take!`](@ref) is performed by a differ
 task.
 """
 function put!(c::Channel, v)
-    !isopen(c) && throw(closed_exception())
+    check_channel_state(c)
     isbuffered(c) ? put_buffered(c,v) : put_unbuffered(c,v)
+    yield()
+    v
 end
 
 function put_buffered(c::Channel, v)
@@ -170,7 +216,7 @@ shift!(c::Channel) = take!(c)
 
 # 0-size channel
 function take_unbuffered(c::Channel)
-    !isopen(c) && throw(closed_exception())
+    check_channel_state(c)
     cond_taker = Condition()
     push!(c.takers, cond_taker)
     notify(c.cond_put, nothing, false, false)
@@ -200,7 +246,7 @@ n_avail(c::Channel) = isbuffered(c) ? length(c.data) : n_waiters(c.cond_put)
 
 function wait(c::Channel)
     while !isready(c)
-        !isopen(c) && throw(closed_exception())
+        check_channel_state(c)
         wait(c.cond_take)
     end
     nothing
@@ -216,14 +262,14 @@ eltype{T}(::Type{Channel{T}}) = T
 
 show(io::IO, c::Channel) = print(io, "$(typeof(c))(sz_max:$(c.sz_max),sz_curr:$(n_avail(c)))")
 
-type ChannelState{T}
+type ChannelIterState{T}
     hasval::Bool
     val::T
-    ChannelState(x) = new(x)
+    ChannelIterState(x) = new(x)
 end
 
-start{T}(c::Channel{T}) = ChannelState{T}(false)
-function done(c::Channel, state::ChannelState)
+start{T}(c::Channel{T}) = ChannelIterState{T}(false)
+function done(c::Channel, state::ChannelIterState)
     try
         # we are waiting either for more data or channel to be closed
         state.hasval && return false

base/docs/helpdb/Base.jl

@@ -707,14 +707,6 @@ Show every part of the representation of a value.
 """
 dump
 
-"""
-    consume(task, values...)
-
-Receive the next value passed to `produce` by the specified task. Additional arguments may
-be passed, to be returned from the last `produce` call in the producer.
-"""
-consume
-
 """
     cummax(A, [dim])
 
@@ -1886,14 +1878,6 @@ the topmost backend that does not throw a `MethodError`).
 """
 pushdisplay
 
-"""
-    produce(value)
-
-Send the given value to the last `consume` call, switching to the consumer task. If the next
-`consume` call passes any values, they are returned by `produce`.
-"""
-produce
-
 """
     StackOverflowError()
 

base/exports.jl

@@ -950,6 +950,7 @@ export
     sizeof,
 
 # tasks and conditions
+    channeled_tasks,
     Condition,
     consume,
     current_task,

base/file.jl

@@ -470,28 +470,24 @@ function walkdir(root; topdown=true, follow_symlinks=false, onerror=throw)
     end
 
     function _it(chnl)
-        close(chnl, current_task())
         if topdown
             put!(chnl, (root, dirs, files))
-            yield()
         end
         for dir in dirs
             path = joinpath(root,dir)
             if follow_symlinks || !islink(path)
                 for (root_l, dirs_l, files_l) in walkdir(path, topdown=topdown, follow_symlinks=follow_symlinks, onerror=onerror)
                     put!(chnl, (root_l, dirs_l, files_l))
-                    yield()
                 end
             end
         end
         if !topdown
             put!(chnl, (root, dirs, files))
-            yield()
         end
     end
-    chnl = Channel(0)
-    @schedule _it(chnl)
-    return chnl
+
+    chnls, _ = channeled_tasks(1, _it)
+    return chnls[1]
 end
 
 function unlink(p::AbstractString)

base/task.jl

@@ -187,7 +187,7 @@ suppress_excp_printing(t::Task) = isa(t.storage, ObjectIdDict) ? get(get_task_tl
 function register_taskdone_hook(t::Task, hook)
     tls = get_task_tls(t)
     push!(get!(tls, :TASKDONE_HOOKS, []), hook)
-    nothing
+    t
 end
 
 # runtime system hook called when a task finishes

doc/src/stdlib/parallel.md

@@ -8,8 +8,6 @@ Base.yieldto
 Base.current_task
 Base.istaskdone
 Base.istaskstarted
-Base.consume
-Base.produce
 Base.yield
 Base.task_local_storage(::Any)
 Base.task_local_storage(::Any, ::Any)
@@ -26,6 +24,8 @@ Base.take!(::Channel)
 Base.isready(::Channel)
 Base.fetch(::Channel)
 Base.close(::Channel)
+Base.bind(c::Channel, task::Task)
+Base.channeled_tasks
 Base.asyncmap
 Base.asyncmap!
 ```

test/bitarray.jl

@@ -302,11 +302,10 @@ let b1 = bitrand(n1, n2)
         put!(c, (randperm(m1), :, BitMatrix))
         m1, m2 = rand_m1m2()
         put!(c, (randperm(m1), randperm(m2), BitMatrix))
-        close(c)
     end
-    c = Channel(0)
-    @schedule gen_getindex_data(c)
-    for (k1, k2, T) in c
+
+    chnls, _ = channeled_tasks(1, gen_getindex_data)
+    for (k1, k2, T) in chnls[1]
         # println(typeof(k1), " ", typeof(k2), " ", T) # uncomment to debug
         @check_bit_operation getindex(b1, k1, k2) T
     end
@@ -349,12 +348,10 @@ let b1 = bitrand(s1, s2, s3, s4)
 
         m1, m2, m3, m4 = (2:3, 2:7, 1:2, 4:6)
         put!(c, (m1, m2, m3, m4, BitArray{4}))
-        close(c)
     end
 
-    c = Channel(0)
-    @schedule gen_getindex_data4(c)
-    for (k1, k2, k3, k4, T) in c
+    chnls, _ = channeled_tasks(1, gen_getindex_data4)
+    for (k1, k2, k3, k4, T) in chnls[1]
         #println(typeof(k1), " ", typeof(k2), " ", typeof(k3), " ", typeof(k4), " ", T) # uncomment to debug
         @check_bit_operation getindex(b1, k1, k2, k3, k4) T
     end
@@ -393,12 +390,10 @@ let b1 = bitrand(n1, n2)
         m1, m2 = rand_m1m2()
         put!(c, (rand(Bool), randperm(m1), randperm(m2)))
         put!(c, (bitrand(m1,m2), randperm(m1), randperm(m2)))
-        close(c)
     end
 
-    c = Channel(0)
-    @schedule gen_setindex_data(c)
-    for (b2, k1, k2) in c
+    chnls, _ = channeled_tasks(1, gen_setindex_data)
+    for (b2, k1, k2) in chnls[1]
         # println(typeof(b2), " ", typeof(k1), " ", typeof(k2)) # uncomment to debug
         @check_bit_operation setindex!(b1, b2, k1, k2) BitMatrix
     end
@@ -451,12 +446,10 @@ let b1 = bitrand(s1, s2, s3, s4)
 
         m1, m2, m3, m4 = (2:3, 2:7, 1:2, 4:6)
         put!(c, (bitrand(2, 6, 2, 3), m1, m2, m3, m4))
-        close(c)
     end
 
-    c = Channel(0)
-    @schedule gen_setindex_data4(c)
-    for (b2, k1, k2, k3, k4) in c
+    chnls, _ = channeled_tasks(1, gen_setindex_data4)
+    for (b2, k1, k2, k3, k4) in chnls[1]
         # println(typeof(b2), " ", typeof(k1), " ", typeof(k2), " ", typeof(k3), " ", typeof(k4)) # uncomment to debug
         @check_bit_operation setindex!(b1, b2, k1, k2, k3, k4) BitArray{4}
     end