Channel: drop explicit API change, make always threadsafe internally

Author: vtjnash

base/channels.jl

@@ -3,7 +3,7 @@
 abstract type AbstractChannel{T} end
 
 """
-    Channel{T}(sz::Int, threadsafe::Bool)
+    Channel{T}(sz::Int)
 
 Constructs a `Channel` with an internal buffer that can hold a maximum of `sz` objects
 of type `T`.
@@ -12,9 +12,6 @@ of type `T`.
 `Channel(0)` constructs an unbuffered channel. `put!` blocks until a matching `take!` is called.
 And vice-versa.
 
-If `threadsafe` is true, some API operations (specifically `wait`) require first acquiring
-the lock on the Channel object.
-
 Other constructors:
 
 * `Channel(Inf)`: equivalent to `Channel{Any}(typemax(Int))`
@@ -30,22 +27,22 @@ mutable struct Channel{T} <: AbstractChannel{T}
     data::Vector{T}
     sz_max::Int                          # maximum size of channel
 
-    function Channel{T}(sz::Integer, threadsafe::Bool=false) where T
+    function Channel{T}(sz::Integer) where T
         if sz < 0
             throw(ArgumentError("Channel size must be either 0, a positive integer or Inf"))
         end
-        lock = threadsafe ? ReentrantLock() : NotALock()
+        lock = ReentrantLock()
         cond_put, cond_take = Condition(lock), Condition(lock)
         cond_wait = (sz == 0 ? Condition(lock) : cond_take) # wait is distinct from take iff unbuffered
         return new(cond_take, cond_wait, cond_put, :open, nothing, Vector{T}(), sz)
     end
 end
 
-function Channel{T}(sz::Float64, threadsafe::Bool=false) where T
+function Channel{T}(sz::Float64) where T
     sz = (sz == Inf ? typemax(Int) : convert(Int, sz))
     return Channel{T}(sz)
 end
-Channel(sz, threadsafe::Bool=false) = Channel{Any}(sz, threadsafe)
+Channel(sz) = Channel{Any}(sz)
 
 # special constructors
 """
@@ -94,8 +91,8 @@ julia> istaskdone(taskref[])
 true
 ```
 """
-function Channel(func::Function, threadsafe::Bool=false; ctype=Any, csize=0, taskref=nothing)
-    chnl = Channel{ctype}(csize, threadsafe)
+function Channel(func::Function; ctype=Any, csize=0, taskref=nothing)
+    chnl = Channel{ctype}(csize)
     task = Task(() -> func(chnl))
     bind(chnl, task)
     yield(task) # immediately start it
@@ -377,9 +374,15 @@ unlock(c::Channel) = unlock(c.cond_take)
 trylock(c::Channel) = trylock(c.cond_take)
 
 function wait(c::Channel)
-    while !isready(c)
-        check_channel_state(c)
-        wait(c.cond_wait)
+    isready(c) && return
+    lock(c)
+    try
+        while !isready(c)
+            check_channel_state(c)
+            wait(c.cond_wait)
+        end
+    finally
+        unlock(c)
     end
     nothing
 end

test/channels.jl

@@ -56,12 +56,10 @@ end
 
 @testset "type conversion in put!" begin
     c = Channel{Int64}(0)
-    lock(c)
     @async put!(c, Int32(1))
     wait(c)
     @test isa(take!(c), Int64)
     @test_throws MethodError put!(c, "")
-    unlock(c)
     @assert !islocked(c.cond_take)
 end
 
@@ -122,7 +120,6 @@ using Distributed
         while isopen(cs[i])
             yield()
         end
-        i < 3 && foreach(lock, cs)
         @test_throws ErrorException wait(cs[i])
         @test_throws ErrorException take!(cs[i])
         @test_throws ErrorException put!(cs[i], 1)
@@ -145,11 +142,9 @@ using Distributed
     c = Channel(N)
     foreach(t -> bind(c, t), tasks)
     foreach(schedule, tasks)
-    lock(c)
     @test_throws InvalidStateException wait(c)
     @test !isopen(c)
     @test ref[] == nth
-    unlock(c)
     @assert !islocked(c.cond_take)
 
     # channeled_tasks
@@ -158,10 +153,8 @@ using Distributed
         chnls, tasks = Base.channeled_tasks(2, tf_chnls1; ctypes=[T,T], csizes=[N,N])
         put!(chnls[1], 1)
         @test take!(chnls[2]) === 2
-        foreach(lock, chnls)
         @test_throws InvalidStateException wait(chnls[1])
         @test_throws InvalidStateException wait(chnls[2])
-        foreach(unlock, chnls)
         @test istaskdone(tasks[1])
         @test !isopen(chnls[1])
         @test !isopen(chnls[2])
@@ -183,10 +176,8 @@ using Distributed
         yield()
         put!(f, 1) # allow tf4 and tf5 to exit after now, eventually closing the channel
 
-        foreach(lock, chnls)
         @test_throws InvalidStateException wait(chnls[1])
         @test_throws InvalidStateException wait(chnls[2])
-        foreach(unlock, chnls)
         @test istaskdone(tasks[1])
         @test istaskdone(tasks[2])
         @test !isopen(chnls[1])