Executor fixes (#157)

* Add more sanity checks

* workaround #155

tests/README.md

@@ -0,0 +1,7 @@
+# Weave tests
+
+Testing multithreading runtime is difficult as bugs may only surface after particular thread patterns.
+
+The main approach to testing Weave is therefore to have extensive internal invariant checking which is enabled in Nim debug mode or with -d:WV_Asserts.
+
+The whole benchmarking suite is run with that flag across a wide range of threading strategies and intensive stress-testing.

weave/datatypes/prell_deques.nim

@@ -173,7 +173,7 @@ func steal*[T](dq: var PrellDeque[T]): T =
 
   if dq.tail.prev.isNil:
     # Stealing last task of the deque
-    # ascertain: dq.head == result # Concept are buggy with repr, TODO
+    ascertain: cast[pointer](dq.head) == cast[pointer](result)
     dq.head = dq.tail.addr # isEmpty() condition
   else:
     dq.tail.prev.next = dq.tail.addr # last task points to dummy

weave/executor.nim

@@ -96,6 +96,10 @@ proc processAllandTryPark*(_: typedesc[Weave]) =
   ## This `syncRoot` then put the Weave runtime to sleep
   ## if no job submission was received concurrently
   ##
+  ## This is only valid on Weave root thread.
+  ## Note that if you use `runInBackground`
+  ## this is what the background manager thread does automaticallu
+  ##
   ## This should be used if Weave root thread (that called init(Weave))
   ## is on a dedicated long-running thread
   ## in an event loop:
@@ -104,6 +108,7 @@ proc processAllandTryPark*(_: typedesc[Weave]) =
   ##   park(Weave)
   ##
   ## New job submissions will automatically wakeup the runtime
+  preCondition: onWeaveThread() and myTask().isRootTask()
 
   manager.jobNotifier[].prepareToPark()
   syncRoot(Weave)
@@ -148,8 +153,8 @@ proc runInBackground*(
     init(Weave)
     Weave.runUntil(shutdown)
     exit(Weave)
-  {.gcsafe.}: # Workaround regression - https://github.com/nim-lang/Nim/issues/14370
-    thr.createThread(eventLoop, signalShutdown)
+
+  thr.createThread(eventLoop, signalShutdown)
 
 proc runInBackground*(thr: var Thread[void], _: typedesc[Weave])  =
   ## Start the Weave runtime on a background thread.
@@ -159,6 +164,7 @@ proc runInBackground*(thr: var Thread[void], _: typedesc[Weave])  =
   proc eventLoop() {.thread.} =
     init(Weave)
     Weave.runForever()
+
   thr.createThread(eventLoop)
 
 proc submitJob*(job: sink Job) =

weave/scheduler.nim

@@ -129,6 +129,7 @@ proc setupWorker*() =
   postCondition: myWorker().workSharingRequests.isEmpty()
   postCondition: not ctx.signaledTerminate
   postCondition: not myWorker().isWaiting
+  postCondition: localThreadKind == WorkerThread
 
   # Thread-Local Profiling
   # -----------------------------------------------------------

weave/state_machines/sync_root.nim

@@ -49,7 +49,7 @@ setPrologue(syncRootFSA):
   Worker: return
   debugTermination:
     log(">>> Worker %2d enters barrier <<<\n", myID())
-  preCondition: myTask().isRootTask()
+  preCondition: onWeaveThread() and myTask().isRootTask()
 
   debug: log("Worker %2d: syncRoot 1 - task from local deque\n", myID())
   var task: Task

weave/victims.nim

@@ -313,11 +313,13 @@ proc distributeWork*(req: sink StealRequest, workSharing: bool): bool =
     var firstJob, lastJob: Job
     let count = managerJobQueue.tryRecvBatch(firstJob, lastJob)
     if count != 0:
-      myWorker().deque.addListFirst(
-        cast[Task](firstJob),
-        cast[Task](lastJob),
-        count
-      )
+      # TODO: https://github.com/mratsim/weave/issues/155
+      # myWorker().deque.addListFirst(
+      #   cast[Task](firstJob),
+      #   cast[Task](lastJob),
+      #   count
+      # )
+      myWorker().deque.addListFirst(cast[Task](lastJob))
       req.dispatchElseDecline()
       return true