make at-sync thread safe

base/Base.jl

@@ -216,11 +216,12 @@ end
 
 include("env.jl")
 
-# Scheduling
+# Concurrency
 include("linked_list.jl")
 include("condition.jl")
 include("threads.jl")
 include("lock.jl")
+include("channels.jl")
 include("task.jl")
 include("weakkeydict.jl")
 
@@ -315,9 +316,6 @@ using .MathConstants: ℯ, π, pi
 # metaprogramming
 include("meta.jl")
 
-# concurrency and parallelism
-include("channels.jl")
-
 # utilities
 include("deepcopy.jl")
 include("download.jl")

base/experimental.jl

@@ -52,30 +52,34 @@ macro aliasscope(body)
 end
 
 
-function sync_end(refs)
-    local c_ex
-    defined = false
-    t = current_task()
-    cond = Threads.Condition()
-    lock(cond)
-    nremaining = length(refs)
-    for r in refs
-        schedule(Task(()->begin
-            try
-                wait(r)
-                lock(cond)
-                nremaining -= 1
-                nremaining == 0 && notify(cond)
-                unlock(cond)
-            catch e
-                lock(cond)
-                notify(cond, e; error=true)
-                unlock(cond)
+function sync_end(c::Channel{Any})
+    if !isready(c)
+        # there must be at least one item to begin with
+        close(c)
+        return
+    end
+    nremaining::Int = 0
+    while true
+        event = take!(c)
+        if event === :__completion__
+            nremaining -= 1
+            if nremaining == 0
+                break
             end
-        end))
+        else
+            nremaining += 1
+            schedule(Task(()->begin
+                try
+                    wait(event)
+                    put!(c, :__completion__)
+                catch e
+                    close(c, e)
+                end
+            end))
+        end
     end
-    wait(cond)
-    unlock(cond)
+    close(c)
+    nothing
 end
 
 """
@@ -92,7 +96,7 @@ during error handling.
 macro sync(block)
     var = esc(sync_varname)
     quote
-        let $var = Any[]
+        let $var = Channel(Inf)
             v = $(esc(block))
             sync_end($var)
             v

base/task.jl

@@ -286,15 +286,14 @@ end
 
 ## lexically-scoped waiting for multiple items
 
-function sync_end(refs)
+function sync_end(c::Channel{Any})
     local c_ex
-    defined = false
-    for r in refs
+    while isready(c)
+        r = take!(c)
         if isa(r, Task)
             _wait(r)
             if istaskfailed(r)
-                if !defined
-                    defined = true
+                if !@isdefined(c_ex)
                     c_ex = CompositeException()
                 end
                 push!(c_ex, TaskFailedException(r))
@@ -303,16 +302,15 @@ function sync_end(refs)
             try
                 wait(r)
             catch e
-                if !defined
-                    defined = true
+                if !@isdefined(c_ex)
                     c_ex = CompositeException()
                 end
                 push!(c_ex, e)
             end
         end
     end
-
-    if defined
+    close(c)
+    if @isdefined(c_ex)
         throw(c_ex)
     end
     nothing
@@ -330,7 +328,7 @@ a `CompositeException`.
 macro sync(block)
     var = esc(sync_varname)
     quote
-        let $var = Any[]
+        let $var = Channel(Inf)
             v = $(esc(block))
             sync_end($var)
             v
@@ -361,7 +359,7 @@ macro async(expr)
         let $(letargs...)
             local task = Task($thunk)
             if $(Expr(:islocal, var))
-                push!($var, task)
+                put!($var, task)
             end
             schedule(task)
             task
@@ -403,7 +401,7 @@ macro sync_add(expr)
     var = esc(sync_varname)
     quote
         local ref = $(esc(expr))
-        push!($var, ref)
+        put!($var, ref)
         ref
     end
 end

base/threadingconstructs.jl

@@ -130,7 +130,7 @@ macro spawn(expr)
             local task = Task($thunk)
             task.sticky = false
             if $(Expr(:islocal, var))
-                push!($var, task)
+                put!($var, task)
             end
             schedule(task)
             task

stdlib/Distributed/src/macros.jl

@@ -49,7 +49,7 @@ macro spawn(expr)
     quote
         local ref = spawn_somewhere($thunk)
         if $(Expr(:islocal, var))
-            push!($var, ref)
+            put!($var, ref)
         end
         ref
     end
@@ -94,7 +94,7 @@ macro spawnat(p, expr)
     quote
         local ref = $spawncall
         if $(Expr(:islocal, var))
-            push!($var, ref)
+            put!($var, ref)
         end
         ref
     end
@@ -345,7 +345,7 @@ macro distributed(args...)
         return quote
             local ref = pfor($(make_pfor_body(var, body)), $(esc(r)))
             if $(Expr(:islocal, syncvar))
-                push!($syncvar, ref)
+                put!($syncvar, ref)
             end
             ref
         end

test/threads_exec.jl

@@ -821,3 +821,16 @@ end
     x = 2
     @test @eval(fetch(@async 2+$x)) == 4
 end
+
+# issue #34666
+fib34666(x) =
+    @sync begin
+        function f(x)
+            x in (0, 1) && return x
+            a = Threads.@spawn f(x - 2)
+            b = Threads.@spawn f(x - 1)
+            return fetch(a) + fetch(b)
+        end
+        f(x)
+    end
+@test fib34666(25) == 75025