Merge pull request #1369 from JuliaGPU/tb/async_errors

Make nonblocking synchronization robust to errors.

Author: maleadt

lib/cudadrv/execution.jl

@@ -147,7 +147,8 @@ function launch(f::Base.Callable; stream::CuStream=stream())
         f()
         close(async_cond)
     end
-    # FIXME: protect this from GC
+
+    # the condition object is embedded in a task, so the Julia scheduler keeps it alive
 
     # callback = @cfunction(async_send, Cint, (Ptr{Cvoid},))
     # See https://github.com/JuliaGPU/CUDA.jl/issues/1314.

lib/cudadrv/stream.jl

@@ -153,13 +153,37 @@ end
         spins += 1
     end
 
-    # minimize CPU usage of long-running kernels
-    # by waiting for an event signalled by CUDA
-    event = Threads.Event()
+    # minimize CPU usage of long-running kernels by waiting for an event signalled by CUDA
+    event = Base.Event()
     launch(; stream) do
         notify(event)
     end
-    Base.wait(event)
+    # if an error occurs, the callback may never fire, so use a timer to detect such cases
+    dev = device()
+    timer = Timer(0; interval=1)
+    Base.@sync begin
+        Threads.@spawn try
+            device!(dev)
+            while true
+                try
+                    Base.wait(timer)
+                catch err
+                    err isa EOFError && break
+                    rethrow()
+                end
+                if unsafe_cuStreamQuery(stream) != ERROR_NOT_READY
+                    break
+                end
+            end
+        finally
+            notify(event)
+        end
+
+        Threads.@spawn begin
+            Base.wait(event)
+            close(timer)
+        end
+    end
 
     return
 end

src/compiler/execution.jl

@@ -311,7 +311,7 @@ function run_and_collect(cmd)
     proc = run(pipeline(ignorestatus(cmd); stdout, stderr=stdout), wait=false)
     close(stdout.in)
 
-    reader = @async String(read(stdout))
+    reader = Threads.@spawn String(read(stdout))
     Base.wait(proc)
     log = strip(fetch(reader))
 

src/pool.jl

@@ -89,7 +89,7 @@ function pool_mark(dev::CuDevice)
 
       # launch a task to periodically trim the pool
       if isinteractive() && !isassigned(__pool_cleanup)
-        __pool_cleanup[] = @async pool_cleanup()
+        __pool_cleanup[] = errormonitor(Threads.@spawn pool_cleanup())
       end
   end
   status[] = true