asyncmap: Include original backtrace in rethrown exception

Collect the TaskExceptions into a CompositeException.

Author: Richard Palethorpe

base/asyncmap.jl

@@ -175,7 +175,7 @@ function maptwice(wrapped_f, chnl, worker_tasks, c...)
     close(chnl)
 
     # check and throw any exceptions from the worker tasks
-    foreach(x->(v=fetch(x); isa(v, Exception) && throw(v)), worker_tasks)
+    @sync foreach(t -> @sync_add(t), worker_tasks)
 
     # check if there was a genuine problem with asyncrun
     (asyncrun_excp !== nothing) && throw(asyncrun_excp)
@@ -212,34 +212,26 @@ end
 
 function start_worker_task!(worker_tasks, exec_func, chnl, batch_size=nothing)
     t = @async begin
-        retval = nothing
-
-        try
-            if isa(batch_size, Number)
-                while isopen(chnl)
-                    # The mapping function expects an array of input args, as it processes
-                    # elements in a batch.
-                    batch_collection=Any[]
-                    n = 0
-                    for exec_data in chnl
-                        push!(batch_collection, exec_data)
-                        n += 1
-                        (n == batch_size) && break
-                    end
-                    if n > 0
-                        exec_func(batch_collection)
-                    end
-                end
-            else
+        if isa(batch_size, Number)
+            while isopen(chnl)
+                # The mapping function expects an array of input args, as it processes
+                # elements in a batch.
+                batch_collection=Any[]
+                n = 0
                 for exec_data in chnl
-                    exec_func(exec_data...)
+                    push!(batch_collection, exec_data)
+                    n += 1
+                    (n == batch_size) && break
+                end
+                if n > 0
+                    exec_func(batch_collection)
                 end
             end
-        catch e
-            close(chnl)
-            retval = e
+        else
+            for exec_data in chnl
+                exec_func(exec_data...)
+            end
         end
-        retval
     end
     push!(worker_tasks, t)
 end

stdlib/Distributed/test/distributed_exec.jl

@@ -595,7 +595,7 @@ let error_thrown = false
     try
         pmap(x -> x == 50 ? error("foobar") : x, 1:100)
     catch e
-        @test e.captured.ex.msg == "foobar"
+        @test e.exceptions[1].task.exception.captured.ex.msg == "foobar"
         error_thrown = true
     end
     @test error_thrown