Adds $-interpolation syntax to @async and Threads.@spawn, to eval…

…uate arguments immediately.

Add the ability to evaluate some parts of a `@spawn`/`@async`
immediately, in the current thread context.

This prevents variables being "boxed" in order to capture them in the
closure, exactly the same as wrapping them in a let-block locally.

NEWS.md

@@ -29,6 +29,8 @@ Language changes
 Multi-threading changes
 -----------------------
 
+* Values can now be interpolated into `@async` and `@spawn` via `$`, which copies the value directly into the constructed
+underlying closure. ([#33119])
 
 Build system changes
 --------------------

base/expr.jl

@@ -330,8 +330,6 @@ function is_short_function_def(ex)
     return false
 end
 
-
-
 function findmeta(ex::Expr)
     if ex.head === :function || is_short_function_def(ex)
         body::Expr = ex.args[2]

base/task.jl

@@ -344,6 +344,13 @@ end
     @async
 
 Wrap an expression in a [`Task`](@ref) and add it to the local machine's scheduler queue.
+
+Values can be interpolated into `@async` via `\$`, which copies the value directly into the
+constructed underlying closure. This allows you to insert the _value_ of a variable,
+isolating the aysnchronous code from changes to the variable's value in the current task.
+
+!!! compat "Julia 1.4"
+    Interpolating values via `\$` is available as of Julia 1.4.
 """
 macro async(expr)
     letargs = Base._lift_one_interp!(expr)
@@ -378,7 +385,6 @@ function _lift_one_interp_helper(expr::Expr, in_quote_context, letargs)
             newarg = gensym()
             push!(letargs, :($(esc(newarg)) = $(esc(expr.args[1]))))
             return newarg  # Don't recurse into the lifted $() exprs
->>>>>>> fb29992... Factor out `$`-lifting; share b/w `@async` & `@spawn`
         end
     elseif expr.head == :quote
         in_quote_context = true   # Don't try to lift $ directly out of quotes

base/threadingconstructs.jl

@@ -107,13 +107,18 @@ Create and run a [`Task`](@ref) on any available thread. To wait for the task to
 finish, call [`wait`](@ref) on the result of this macro, or call [`fetch`](@ref)
 to wait and then obtain its return value.
 
-(Values can be interpolated into `@spawn` via `\$` to evaluate them in the current task.)
+Values can be interpolated into `@spawn` via `\$`, which copies the value directly into the
+constructed underlying closure. This allows you to insert the _value_ of a variable,
+isolating the aysnchronous code from changes to the variable's value in the current task.
 
 !!! note
     This feature is currently considered experimental.
 
 !!! compat "Julia 1.3"
     This macro is available as of Julia 1.3.
+
+!!! compat "Julia 1.4"
+    Interpolating values via `\$` is available as of Julia 1.4.
 """
 macro spawn(expr)
     letargs = Base._lift_one_interp!(expr)

test/threads_exec.jl

@@ -728,9 +728,13 @@ end
     @test fetch(Threads.@spawn sort([3 $2; 1 $0]; dims=$2)) == [2 3; 0 1]
 
     # Supports multiple levels of interpolation
-    @test fetch(Threads.@spawn :($a)) == a
-    @test fetch(Threads.@spawn :($($a))) == a
     @test fetch(Threads.@spawn "$($a)") == "$a"
+    let a = 1
+        # Interpolate the current value of `a` vs the value of `a` in the closure
+        t = Threads.@spawn :(+($$a, $a, a))
+        a = 2  # update `a` after spawning
+        @test fetch(t) == Expr(:call, :+, 1, 2, :a)
+    end
 
     # Test the difference between different levels of interpolation
     let
@@ -750,3 +754,35 @@ end
         @test twointerps == 1:5
     end
 end
+
+@testset "@async interpolation" begin
+    # Args
+    @test fetch(@async 2+$2) == 4
+    @test fetch(@async Int($(2.0))) == 2
+    a = 2
+    @test fetch(@async *($a,$a)) == a^2
+    # kwargs
+    @test fetch(@async sort($([3 2; 1 0]), dims=2)) == [2 3; 0 1]
+    @test fetch(@async sort([3 $2; 1 $0]; dims=$2)) == [2 3; 0 1]
+
+    # Supports multiple levels of interpolation
+    @test fetch(@async :($a)) == a
+    @test fetch(@async :($($a))) == a
+    @test fetch(@async "$($a)") == "$a"
+end
+
+# errors inside @threads
+function _atthreads_with_error(a, err)
+    Threads.@threads for i in eachindex(a)
+        if err
+            error("failed")
+        end
+        a[i] = Threads.threadid()
+    end
+    a
+end
+@test_throws TaskFailedException _atthreads_with_error(zeros(nthreads()), true)
+let a = zeros(nthreads())
+    _atthreads_with_error(a, false)
+    @test a == [1:nthreads();]
+end