Throw CapturedExceptions from asyncmap to avoid dropping the stackt…

…race (#42105)

base/asyncmap.jl

@@ -236,7 +236,7 @@ function start_worker_task!(worker_tasks, exec_func, chnl, batch_size=nothing)
  end
  catch e
  close(chnl)
- retval = e
+ retval = capture_exception(e, catch_backtrace())
  end
  retval
  end

base/task.jl

@@ -25,6 +25,16 @@ function showerror(io::IO, ce::CapturedException)
  showerror(io, ce.ex, ce.processed_bt, backtrace=true)
 end
 
+"""
+ capture_exception(ex, bt) -> Exception
+
+Returns an exception, possibly incorporating information from a backtrace `bt`. Defaults to returning [`CapturedException(ex, bt)`](@ref).
+
+Used in [`asyncmap`](@ref) and [`asyncmap!`](@ref) to capture exceptions thrown during
+the user-supplied function call.
+"""
+capture_exception(ex, bt) = CapturedException(ex, bt)
+
 """
  CompositeException
 

stdlib/Distributed/src/process_messages.jl

@@ -44,6 +44,13 @@ struct RemoteException <: Exception
  captured::CapturedException
 end
 
+"""
+ capture_exception(ex::RemoteException, bt)
+
+Returns `ex::RemoteException` which has already captured a backtrace (via it's [`CapturedException`](@ref) field `captured`).
+"""
+Base.capture_exception(ex::RemoteException, bt) = ex
+
 """
  RemoteException(captured)
 

test/asyncmap.jl

@@ -54,6 +54,19 @@ len_only_iterable = (1,2,3,4,5)
 @test_throws ArgumentError asyncmap(identity, 1:10; batch_size="10")
 @test_throws ArgumentError asyncmap(identity, 1:10; ntasks="10")
 
+# Check that we throw a `CapturedException` holding the stacktrace if `f` throws
+f42105(i) = i == 5 ? error("captured") : i
+let
+ e = try
+ asyncmap(f42105, 1:5)
+ catch e
+ e
+ end
+ @test e isa CapturedException
+ @test e.ex == ErrorException("captured")
+ @test e.processed_bt[2][1].func == :f42105
+end
+
 include("generic_map_tests.jl")
 generic_map_tests(asyncmap, asyncmap!)