Show full stacktrace when Channel Task throws (#35177)

This reuses the machinery that gives good stacktraces when a user
calls wait on a Task. Since bind internally uses _wait2, this
machinery is bypassed currently.

Currently, julia throws an uninformative stacktrace in this situation:

julia> c = Channel(_ -> error("foo"))
Channel{Any}(sz_max:0,sz_curr:0)

julia> for i in c end
ERROR: foo
Stacktrace:
 [1] iterate(::Channel{Any}) at ./channels.jl:459
 [2] top-level scope at ./REPL[2]:1

With this change, the stacktrace is much more informative:

julia> for i in c end
ERROR: TaskFailedException:
foo
Stacktrace:
 [1] error(::String) at ./error.jl:33
 [2] (::var"#1#2")(::Channel{Any}) at ./REPL[4]:1
 [3] (::Base.var"#652#653"{var"#1#2",Channel{Any}})() at ./channels.jl:129
Stacktrace:
 [1] check_channel_state at ./channels.jl:167 [inlined]
 [2] take_unbuffered(::Channel{Any}) at ./channels.jl:405
 [3] take! at ./channels.jl:383 [inlined]
 [4] iterate(::Channel{Any}, ::Nothing) at ./channels.jl:449
 [5] iterate(::Channel{Any}) at ./channels.jl:448
 [6] top-level scope at ./REPL[5]:1

Author: non-Jedi

base/channels.jl

@@ -228,15 +228,16 @@ false
 ```jldoctest
 julia> c = Channel(0);
 
-julia> task = @async (put!(c,1);error("foo"));
+julia> task = @async (put!(c, 1); error("foo"));
 
-julia> bind(c,task);
+julia> bind(c, task);
 
 julia> take!(c)
 1
 
-julia> put!(c,1);
-ERROR: foo
+julia> put!(c, 1);
+ERROR: TaskFailedException:
+foo
 Stacktrace:
 [...]
 ```
@@ -281,7 +282,7 @@ function close_chnl_on_taskdone(t::Task, c::Channel)
         if istaskfailed(t)
             excp = task_result(t)
             if excp isa Exception
-                close(c, excp)
+                close(c, TaskFailedException(t))
                 return
             end
         end

test/channels.jl

@@ -143,8 +143,9 @@ using Distributed
 
     # Error exception in task
     c = Channel(N)
-    bind(c, @async (GC.gc(); yield(); error("foo")))
-    @test_throws ErrorException take!(c)
+    task = @async (GC.gc(); yield(); error("foo"))
+    bind(c, task)
+    @test_throws TaskFailedException(task) take!(c)
     @test !isopen(c)
 
     # Multiple channels closed by the same bound task
@@ -170,10 +171,11 @@ using Distributed
         while isopen(cs[i])
             yield()
         end
-        @test_throws ErrorException wait(cs[i])
-        @test_throws ErrorException take!(cs[i])
-        @test_throws ErrorException put!(cs[i], 1)
-        @test_throws ErrorException fetch(cs[i])
+        @test_throws TaskFailedException(task) wait(cs[i])
+        @test_throws TaskFailedException(task) take!(cs[i])
+        @test_throws TaskFailedException(task) put!(cs[i], 1)
+        N == 0 || @test_throws TaskFailedException(task) fetch(cs[i])
+        N == 0 && @test_throws ErrorException fetch(cs[i])
     end
 
     # Multiple tasks, first one to terminate closes the channel
@@ -245,10 +247,10 @@ using Distributed
         chnl = Channel{T}(tf6, N, taskref=taskref)
         put!(chnl, 2)
         yield()
-        @test_throws ErrorException wait(chnl)
+        @test_throws TaskFailedException(taskref[]) wait(chnl)
         @test istaskdone(taskref[])
         @test !isopen(chnl)
-        @test_throws ErrorException take!(chnl)
+        @test_throws TaskFailedException(taskref[]) take!(chnl)
     end
 end
 

test/file.jl

@@ -1298,7 +1298,7 @@ cd(dirwalk) do
     @test files == ["file1", "file2"]
 
     rm(joinpath("sub_dir1"), recursive=true)
-    @test_throws SystemError take!(chnl_error) # throws an error because sub_dir1 do not exist
+    @test_throws TaskFailedException take!(chnl_error) # throws an error because sub_dir1 do not exist
 
     root, dirs, files = take!(chnl_noerror)
     @test root == "."

test/worlds.jl

@@ -138,7 +138,7 @@ f265(::Int) = 1
 h265() = true
 loc_h265 = "$(@__FILE__):$(@__LINE__() - 1)"
 @test h265()
-@test_throws MethodError put_n_take!(h265, ())
+@test_throws TaskFailedException(t265) put_n_take!(h265, ())
 @test_throws TaskFailedException(t265) fetch(t265)
 @test istaskdone(t265)
 let ex = t265.exception