Merge pull request #30 from JuliaSIMD/resetonerror

Reset threads on error

Author: chriselrod

src/ThreadingUtilities.jl

@@ -30,22 +30,14 @@ include("atomics.jl")
 include("threadtasks.jl")
 include("warnings.jl")
 
-function initialize_task(tid::Int)
+function initialize_task(tid)
   _atomic_store!(taskpointer(tid), WAIT)
   t = Task(ThreadTask(taskpointer(tid)));
   t.sticky = true # create and pin
   # set to tid, we have tasks 2...nthread, from 1-based ind perspective
   ccall(:jl_set_task_tid, Cvoid, (Any, Cint), t, tid % Cint)
   TASKS[tid] = t
 end
-function reinitialize_tasks!(verbose::Bool = true)
-  for (tid,t) ∈ enumerate(TASKS)
-    if istaskfailed(t)
-      verbose && dump(t)
-      initialize_task(tid)
-    end
-  end
-end
 
 function __init__()
   _print_exclusivity_warning()

src/threadtasks.jl

@@ -51,13 +51,27 @@ end
   ccall(:jl_wakeup_thread, Cvoid, (Int16,), tid % Int16)
 end
 
+@noinline function checktask(tid)
+  t = TASKS[tid]
+  if istaskfailed(t)
+    dump(t)
+    initialize_task(tid)
+    return true
+  end
+  yield()
+  false
+end
 # 1-based tid
-@inline wait(tid::Integer) = wait(taskpointer(tid))
-@inline function wait(p::Ptr{UInt})
+@inline wait(tid::Integer) = wait(taskpointer(tid), tid)
+@inline wait(p::Ptr{UInt}) = wait(p, (p - THREADPOOLPTR[]) ÷ (THREADBUFFERSIZE))
+@inline function wait(p::Ptr{UInt}, tid)
   counter = 0x00000000
   while _atomic_state(p) == TASK
     pause()
-    ((counter += 0x00000001) > 0x00010000) && yield()
+    if ((counter += 0x00000001) > 0x00010000)
+      checktask(tid) && return true
+    end
   end
+  false
 end
 

test/threadingutilities.jl

@@ -38,13 +38,13 @@ function test_copy(tid, N = 100_000)
     GC.@preserve a b c x y z begin
         launch_thread_copy!(tid, x, a)
         yield()
-        ThreadingUtilities.wait(tid)
+        @assert !ThreadingUtilities.wait(tid)
         launch_thread_copy!(tid, y, b)
         yield()
-        ThreadingUtilities.wait(tid)
+        @assert !ThreadingUtilities.wait(tid)
         launch_thread_copy!(tid, z, c)
         yield()
-        ThreadingUtilities.wait(tid)
+        @assert !ThreadingUtilities.wait(ThreadingUtilities.taskpointer(tid))
     end
     @test a == x
     @test b == y
@@ -76,7 +76,7 @@ end
   end
   yield()
   @test all(istaskfailed, ThreadingUtilities.TASKS)
-  ThreadingUtilities.reinitialize_tasks!(false)
+  @test all(ThreadingUtilities.wait, eachindex(ThreadingUtilities.TASKS))
   @test !any(istaskfailed, ThreadingUtilities.TASKS)
   # test copy on the reinitialized tasks
   foreach(test_copy, eachindex(ThreadingUtilities.TASKS))