diff --git a/base/task.jl b/base/task.jl index db586d59d02ca..f9adfb9d418fe 100644 --- a/base/task.jl +++ b/base/task.jl @@ -406,6 +406,12 @@ isolating the asynchronous code from changes to the variable's value in the curr Interpolating values via `\$` is available as of Julia 1.4. """ macro async(expr) + do_async_macro(expr) +end + +# generate the code for @async, possibly wrapping the task in something before +# pushing it to the wait queue. +function do_async_macro(expr; wrap=identity) letargs = Base._lift_one_interp!(expr) thunk = esc(:(()->($expr))) @@ -414,7 +420,7 @@ macro async(expr) let $(letargs...) local task = Task($thunk) if $(Expr(:islocal, var)) - put!($var, task) + put!($var, $(wrap(:task))) end schedule(task) task @@ -422,6 +428,35 @@ macro async(expr) end end +# task wrapper that doesn't create exceptions wrapped in TaskFailedException +struct UnwrapTaskFailedException + task::Task +end + +# common code for wait&fetch for UnwrapTaskFailedException +function unwrap_task_failed(f::Function, t::UnwrapTaskFailedException) + try + f(t.task) + catch ex + if ex isa TaskFailedException + throw(ex.task.exception) + else + rethrow() + end + end +end + +# the unwrapping for above task wrapper (gets triggered in sync_end()) +wait(t::UnwrapTaskFailedException) = unwrap_task_failed(wait, t) + +# same for fetching the tasks, for convenience +fetch(t::UnwrapTaskFailedException) = unwrap_task_failed(fetch, t) + +# macro for running async code that doesn't throw wrapped exceptions +macro async_unwrap(expr) + do_async_macro(expr, wrap=task->:(Base.UnwrapTaskFailedException($task))) +end + # Capture interpolated variables in $() and move them to let-block function _lift_one_interp!(e) letargs = Any[] # store the new gensymed arguments diff --git a/stdlib/Distributed/src/Distributed.jl b/stdlib/Distributed/src/Distributed.jl index 50108f05eed26..4c7427e265afa 100644 --- a/stdlib/Distributed/src/Distributed.jl +++ b/stdlib/Distributed/src/Distributed.jl @@ -10,7 +10,7 @@ import Base: getindex, wait, put!, take!, fetch, isready, push!, length, hash, ==, kill, close, isopen, showerror # imports for use -using Base: Process, Semaphore, JLOptions, buffer_writes, @sync_add, +using Base: Process, Semaphore, JLOptions, buffer_writes, @async_unwrap, VERSION_STRING, binding_module, atexit, julia_exename, julia_cmd, AsyncGenerator, acquire, release, invokelatest, shell_escape_posixly, shell_escape_wincmd, escape_microsoft_c_args, @@ -75,7 +75,7 @@ function _require_callback(mod::Base.PkgId) # broadcast top-level (e.g. from Main) import/using from node 1 (only) @sync for p in procs() p == 1 && continue - @sync_add remotecall(p) do + @async_unwrap remotecall_wait(p) do Base.require(mod) nothing end diff --git a/stdlib/Distributed/src/clusterserialize.jl b/stdlib/Distributed/src/clusterserialize.jl index e37987c5bf875..28025ae867c78 100644 --- a/stdlib/Distributed/src/clusterserialize.jl +++ b/stdlib/Distributed/src/clusterserialize.jl @@ -243,7 +243,7 @@ An exception is raised if a global constant is requested to be cleared. """ function clear!(syms, pids=workers(); mod=Main) @sync for p in pids - @sync_add remotecall(clear_impl!, p, syms, mod) + @async_unwrap remotecall_wait(clear_impl!, p, syms, mod) end end clear!(sym::Symbol, pid::Int; mod=Main) = clear!([sym], [pid]; mod=mod) diff --git a/stdlib/Distributed/src/macros.jl b/stdlib/Distributed/src/macros.jl index b53890017d4de..f14aa7f42e3c1 100644 --- a/stdlib/Distributed/src/macros.jl +++ b/stdlib/Distributed/src/macros.jl @@ -226,10 +226,10 @@ function remotecall_eval(m::Module, procs, ex) if pid == myid() run_locally += 1 else - @sync_add remotecall(Core.eval, pid, m, ex) + @async_unwrap remotecall_wait(Core.eval, pid, m, ex) end end - yield() # ensure that the remotecall_fetch have had a chance to start + yield() # ensure that the remotecalls have had a chance to start # execute locally last as we do not want local execution to block serialization # of the request to remote nodes.