fork std/tasks, stricter exceptions (#34)

benchmarks/bouncing_producer_consumer/taskpool_bpc.nim

@@ -65,7 +65,7 @@ proc bpc_consume_nopoll(usec: int32) =
 
     dummy_cpt()
 
-proc bpc_produce(n, d: int32) {.gcsafe.} =
+proc bpc_produce(n, d: int32) {.gcsafe, raises: [].} =
   if d > 0:
     # Create producer task
     tp.spawn bpc_produce(n, d-1)

benchmarks/dfs/taskpool_dfs.nim

@@ -17,7 +17,7 @@ when not defined(windows):
 
 var tp: Taskpool
 
-proc dfs(depth, breadth: int): uint32 {.gcsafe.} =
+proc dfs(depth, breadth: int): uint32 {.gcsafe, raises: [].} =
   if depth == 0:
     return 1
 

benchmarks/heat/taskpool_heat.nim

@@ -5,6 +5,8 @@
 #   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
 # at your option. This file may not be copied, modified, or distributed except according to those terms.
 
+{.push raises: [].}
+
 # From fibril
 #
 # Original license
@@ -251,6 +253,7 @@ proc verify() =
 
   me /= nx * ny
 
+  try:
     if mae > 1e-12:
       echo &"Local maximal absolute error {mae:1.3e}"
       quit 1
@@ -260,9 +263,12 @@ proc verify() =
     if me > 1e-12:
       echo &"Global mean absolute error {me:1.3e}"
       quit 1
+  except ValueError: raiseAssert "format strings"
 
   echo "Verification successful"
 
+{.pop.}
+
 proc main() =
   var nthreads: int
   if existsEnv"TASKPOOL_NUM_THREADS":

benchmarks/nqueens/taskpool_nqueens.nim

@@ -88,10 +88,10 @@ func isValid(n: int32, a: CharArray): bool =
   ## Returns true if none of the queens conflict and 0 otherwise.
 
   for i in 0'i32 ..< n:
-    let p = cast[int32](a[i])
+    let p = int32(a[i])
 
     for j in i+1 ..< n:
-      let q = cast[int32](a[j])
+      let q = int32(a[j])
       if q == p or q == p - (j-i) or q == p + (j-i):
         return false
   return true
@@ -111,7 +111,7 @@ proc nqueens_ser(n, j: int32, a: CharArray): int32 =
     if isValid(j+1, a):
       result += nqueens_ser(n, j+1, a)
 
-proc nqueens_par(n, j: int32, a: CharArray): int32 {.gcsafe.} =
+proc nqueens_par(n, j: int32, a: CharArray): int32 {.gcsafe, raises: [].} =
 
   if n == j:
     # Good solution, count it

examples/e01_simple_tasks.nim

@@ -3,8 +3,11 @@ import ../taskpools
 block: # Async without result
 
   proc displayInt(x: int) =
+    try:
       stdout.write(x)
       stdout.write(" - SUCCESS\n")
+    except IOError:
+      quit 1 # can't do anything productive
 
   proc main() =
     echo "\nSanity check 1: Printing 123456 654321 in parallel"
@@ -21,7 +24,7 @@ block: # Async/Await
   var tp: Taskpool
 
 
-  proc asyncFib(n: int): int {.gcsafe.} =
+  proc asyncFib(n: int): int {.gcsafe, raises: [].} =
     if n < 2:
       return n
 

examples/e02_parallel_pi.nim

@@ -1,7 +1,7 @@
 # Demo of API using a very inefficient π approcimation algorithm.
 
 import
-  std/[strutils, math, cpuinfo],
+  std/[strutils, cpuinfo],
   ../taskpools
 
 # From https://github.com/nim-lang/Nim/blob/v1.6.2/tests/parallel/tpi.nim

taskpools/channels_spsc_single.nim

@@ -5,6 +5,8 @@
 #   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
 # at your option. This file may not be copied, modified, or distributed except according to those terms.
 
+{.push raises: [].}
+
 import
   std/atomics,
   ./instrumentation/[contracts, loggers]
@@ -28,8 +30,6 @@ type
     itemSize*: uint8
     buffer*{.align: 8.}: UncheckedArray[byte]
 
-{.push raises: [AssertionDefect].} # Ensure no exceptions can happen
-
 proc `=`(
     dest: var ChannelSPSCSingle,
     source: ChannelSPSCSingle
@@ -78,7 +78,7 @@ func trySend*[T](chan: var ChannelSPSCSingle, src: sink T): bool {.inline.} =
   chan.full.store(true, moRelease)
   return true
 
-{.pop.} # raises: [AssertionDefect]
+{.pop.} # raises: []
 
 # Sanity checks
 # ------------------------------------------------------------------------------

taskpools/chase_lev_deques.nim

@@ -35,6 +35,8 @@
 # To reduce contention, stealing is done on the opposite end from push/pop
 # so that there is a race only for the very last task.
 
+{.push raises: [].} # Ensure no exceptions can happen
+
 import
   system/ansi_c,
   std/atomics,
@@ -65,7 +67,6 @@ type
     buf: Atomic[ptr Buf[T]]
     garbage: ptr Buf[T]
 
-{.push raises: [AssertionDefect].} # Ensure no exceptions can happen
 {.push overflowChecks: off.}       # We don't want exceptions (for Defect) in a multithreaded context
                                    # but we don't to deal with underflow of unsigned int either
                                    # say "if a < b - c" with c > b
@@ -192,4 +193,4 @@ proc steal*[T](deque: var ChaseLevDeque[T]): T =
       return default(T)
 
 {.pop.} # overflowChecks
-{.pop.} # raises: [AssertionDefect]
+{.pop.} # raises: []

taskpools/event_notifiers.nim

@@ -22,6 +22,8 @@
 # but requires the threadpool to be message-passing based.
 # https://github.com/mratsim/weave/blob/a230cce98a8524b2680011e496ec17de3c1039f2/weave/cross_thread_com/event_notifiers.nim
 
+{.push raises: [].} # Ensure no exceptions can happen
+
 import
   std/locks,
   ./instrumentation/contracts
@@ -36,7 +38,6 @@ type
     parked: int
     signals: int
 
-{.push raises: [AssertionDefect].} # Ensure no exceptions can happen
 {.push overflowChecks: off.}       # We don't want exceptions (for Defect) in a multithreaded context
                                    # but we don't to deal with underflow of unsigned int either
                                    # say "if a < b - c" with c > b

taskpools/flowvars.nim

@@ -5,8 +5,9 @@
 #   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
 # at your option. This file may not be copied, modified, or distributed except according to those terms.
 
+{.push raises: [].}
+
 import
-  std/os,
   ./instrumentation/contracts,
   ./channels_spsc_single,
   ./primitives/allocs

taskpools/taskpools.nim

@@ -35,7 +35,7 @@
 # In case a thread is blocked for IO, other threads can steal pending tasks in that thread.
 # If all threads are pending for IO, the threadpool will not make any progress and be soft-locked.
 
-{.push raises: [AssertionDefect].} # Ensure no exceptions can happen
+{.push raises: [].} # Ensure no exceptions can happen
 
 import
   system/ansi_c,
@@ -47,14 +47,13 @@ import
   ./instrumentation/[contracts, loggers],
   ./sparsesets,
   ./flowvars,
-  ./ast_utils
+  ./ast_utils,
+  ./tasks
 
 export
   # flowvars
-  Flowvar, isSpawned, isReady, sync
+  Flowvar, isSpawned, isReady, sync, tasks
 
-import std/[isolation, tasks]
-export isolation
 
 type
   WorkerID = int32
@@ -182,7 +181,7 @@ proc new(T: type TaskNode, parent: TaskNode, task: sink Task): T =
   tn.task = task
   return tn
 
-proc runTask(tn: var TaskNode) {.raises:[Exception], inline.} =
+proc runTask(tn: var TaskNode) {.raises:[], inline.} =
   ## Run a task and consumes the taskNode
   tn.task.invoke()
   {.gcsafe.}: # Upstream missing tagging `=destroy` as gcsafe
@@ -245,7 +244,7 @@ const RootTask = default(Task) # TODO: sentinel value different from null task
 template isRootTask(task: Task): bool =
   task == RootTask
 
-proc forceFuture*[T](fv: Flowvar[T], parentResult: var T) {.raises:[Exception].} =
+proc forceFuture*[T](fv: Flowvar[T], parentResult: var T) {.raises:[].} =
   ## Eagerly complete an awaited FlowVar
 
   template ctx: untyped = workerContext

taskpools/tasks.nim

@@ -0,0 +1,271 @@
+#        (c) Copyright 2021 Nim contributors
+# Copyright (c) 2023- Status Research & Development GmbH
+
+## This module provides basic primitives for creating parallel programs.
+## A `Task` should be only owned by a single Thread, it cannot be shared by threads.
+##
+## The module was forked from std/tasks in Nim 1.6 to add new functionality and
+## tune to the taskpools use case.
+
+import std/[macros, isolation, typetraits]
+import system/ansi_c
+
+export isolation
+
+
+when compileOption("threads"):
+  from std/effecttraits import isGcSafe
+
+
+#
+# proc hello(a: int, b: string) =
+#   echo $a & b
+#
+# let literal = "Nim"
+# let t = toTask(hello(521, literal))
+#
+#
+# is roughly converted to
+#
+# type
+#   ScratchObj_369098780 = object
+#     a: int
+#     b: string
+#
+# let scratch_369098762 = cast[ptr ScratchObj_369098780](c_calloc(csize_t 1,
+#     csize_t sizeof(ScratchObj_369098780)))
+# if scratch_369098762.isNil:
+#   raise newException(OutOfMemDefect, "Could not allocate memory")
+# block:
+#   var isolate_369098776 = isolate(521)
+#   scratch_369098762.a = extract(isolate_369098776)
+#   var isolate_369098778 = isolate(literal)
+#   scratch_369098762.b = extract(isolate_369098778)
+# proc hello_369098781(args`gensym3: pointer) {.nimcall.} =
+#   let objTemp_369098775 = cast[ptr ScratchObj_369098780](args`gensym3)
+#   let :tmp_369098777 = objTemp_369098775.a
+#   let :tmp_369098779 = objTemp_369098775.b
+#   hello(a = :tmp_369098777, b = :tmp_369098779)
+#
+# proc destroyScratch_369098782(args`gensym3: pointer) {.nimcall.} =
+#   let obj_369098783 = cast[ptr ScratchObj_369098780](args`gensym3)
+#   =destroy(obj_369098783[])
+# let t = Task(callback: hello_369098781, args: scratch_369098762, destroy: destroyScratch_369098782)
+#
+
+{.push raises: [].}
+
+type
+  Task* = object ## `Task` contains the callback and its arguments.
+    callback: proc (args: pointer) {.nimcall, gcsafe, raises: [].}
+    args: pointer
+    destroy: proc (args: pointer) {.nimcall, gcsafe, raises: [].}
+
+
+proc `=copy`*(x: var Task, y: Task) {.error.}
+
+proc `=destroy`*(t: var Task) {.inline, gcsafe.} =
+  ## Frees the resources allocated for a `Task`.
+  if t.args != nil:
+    if t.destroy != nil:
+      t.destroy(t.args)
+    c_free(t.args)
+
+proc invoke*(task: Task) {.inline, gcsafe.} =
+  ## Invokes the `task`.
+  assert task.callback != nil
+  task.callback(task.args)
+
+template checkIsolate(scratchAssignList: seq[NimNode], procParam, scratchDotExpr: NimNode) =
+  # block:
+  #   var isoTempA = isolate(521)
+  #   scratch.a = extract(isolateA)
+  #   var isoTempB = isolate(literal)
+  #   scratch.b = extract(isolateB)
+  let isolatedTemp = genSym(nskTemp, "isoTemp")
+  scratchAssignList.add newVarStmt(isolatedTemp, newCall(newIdentNode("isolate"), procParam))
+  scratchAssignList.add newAssignment(scratchDotExpr,
+      newCall(newIdentNode("extract"), isolatedTemp))
+
+template addAllNode(assignParam: NimNode, procParam: NimNode) =
+  let scratchDotExpr = newDotExpr(scratchIdent, formalParams[i][0])
+
+  checkIsolate(scratchAssignList, procParam, scratchDotExpr)
+
+  let tempNode = genSym(kind = nskTemp, ident = formalParams[i][0].strVal)
+  callNode.add nnkExprEqExpr.newTree(formalParams[i][0], tempNode)
+  tempAssignList.add newLetStmt(tempNode, newDotExpr(objTemp, formalParams[i][0]))
+  scratchRecList.add newIdentDefs(newIdentNode(formalParams[i][0].strVal), assignParam)
+
+macro toTask*(e: typed{nkCall | nkInfix | nkPrefix | nkPostfix | nkCommand | nkCallStrLit}): Task =
+  ## Converts the call and its arguments to `Task`.
+  runnableExamples("--gc:orc"):
+    proc hello(a: int) = echo a
+
+    let b = toTask hello(13)
+    assert b is Task
+
+  doAssert getTypeInst(e).typeKind == ntyVoid
+
+  when compileOption("threads"):
+    if not isGcSafe(e[0]):
+      error("'toTask' takes a GC safe call expression", e)
+
+  if hasClosure(e[0]):
+    error("closure call is not allowed", e)
+
+  if e.len > 1:
+    let scratchIdent = genSym(kind = nskTemp, ident = "scratch")
+    let impl = e[0].getTypeInst
+
+    when defined(nimTasksDebug):
+      echo impl.treeRepr
+      echo e.treeRepr
+    let formalParams = impl[0]
+
+    var
+      scratchRecList = newNimNode(nnkRecList)
+      scratchAssignList: seq[NimNode]
+      tempAssignList: seq[NimNode]
+      callNode: seq[NimNode]
+
+    let
+      objTemp = genSym(nskTemp, ident = "objTemp")
+
+    for i in 1 ..< formalParams.len:
+      var param = formalParams[i][1]
+
+      if param.kind == nnkBracketExpr and param[0].eqIdent("sink"):
+        param = param[0]
+
+      if param.typeKind in {ntyExpr, ntyStmt}:
+        error("'toTask'ed function cannot have a 'typed' or 'untyped' parameter", e)
+
+      case param.kind
+      of nnkVarTy:
+        error("'toTask'ed function cannot have a 'var' parameter", e)
+      of nnkBracketExpr:
+        if param[0].typeKind == ntyTypeDesc:
+          callNode.add nnkExprEqExpr.newTree(formalParams[i][0], e[i])
+        elif param[0].typeKind in {ntyVarargs, ntyOpenArray}:
+          if param[1].typeKind in {ntyExpr, ntyStmt}:
+            error("'toTask'ed function cannot have a 'typed' or 'untyped' parameter", e)
+          let
+            seqType = nnkBracketExpr.newTree(newIdentNode("seq"), param[1])
+            seqCallNode = newCall("@", e[i])
+          addAllNode(seqType, seqCallNode)
+        else:
+          addAllNode(param, e[i])
+      of nnkBracket, nnkObjConstr:
+        # passing by static parameters
+        # so we pass them directly instead of passing by scratchObj
+        callNode.add nnkExprEqExpr.newTree(formalParams[i][0], e[i])
+      of nnkSym, nnkPtrTy:
+        addAllNode(param, e[i])
+      of nnkCharLit..nnkNilLit:
+        callNode.add nnkExprEqExpr.newTree(formalParams[i][0], e[i])
+      else:
+        error("'toTask'ed function cannot have a parameter of " & $param.kind & " kind", e)
+
+    let scratchObjType = genSym(kind = nskType, ident = "ScratchObj")
+    let scratchObj = nnkTypeSection.newTree(
+                      nnkTypeDef.newTree(
+                        scratchObjType,
+                        newEmptyNode(),
+                        nnkObjectTy.newTree(
+                          newEmptyNode(),
+                          newEmptyNode(),
+                          scratchRecList
+                        )
+                      )
+                    )
+
+
+    let scratchObjPtrType = quote do:
+      cast[ptr `scratchObjType`](c_calloc(csize_t 1, csize_t sizeof(`scratchObjType`)))
+
+    let scratchLetSection = newLetStmt(
+      scratchIdent,
+      scratchObjPtrType
+    )
+
+    let scratchCheck = quote do:
+      if `scratchIdent`.isNil:
+        raise newException(OutOfMemDefect, "Could not allocate memory")
+
+    var stmtList = newStmtList()
+    stmtList.add(scratchObj)
+    stmtList.add(scratchLetSection)
+    stmtList.add(scratchCheck)
+    stmtList.add(nnkBlockStmt.newTree(newEmptyNode(), newStmtList(scratchAssignList)))
+
+    var functionStmtList = newStmtList()
+    let funcCall = newCall(e[0], callNode)
+    functionStmtList.add tempAssignList
+    functionStmtList.add funcCall
+
+    let funcName = genSym(nskProc, e[0].strVal)
+    let destroyName = genSym(nskProc, "destroyScratch")
+    let objTemp2 = genSym(ident = "obj")
+    let tempNode = quote("@") do:
+        `=destroy`(@objTemp2[])
+
+    result = quote do:
+      `stmtList`
+
+      proc `funcName`(args: pointer) {.gcsafe, nimcall, raises: [].} =
+        let `objTemp` = cast[ptr `scratchObjType`](args)
+        `functionStmtList`
+
+      proc `destroyName`(args: pointer) {.gcsafe, nimcall, raises: [].} =
+        let `objTemp2` = cast[ptr `scratchObjType`](args)
+        `tempNode`
+
+      Task(callback: `funcName`, args: `scratchIdent`, destroy: `destroyName`)
+  else:
+    let funcCall = newCall(e[0])
+    let funcName = genSym(nskProc, e[0].strVal)
+
+    result = quote do:
+      proc `funcName`(args: pointer) {.gcsafe, nimcall, raises: [].} =
+        `funcCall`
+
+      Task(callback: `funcName`, args: nil)
+
+  when defined(nimTasksDebug):
+    echo result.repr
+
+when isMainModule:
+  block:
+    var num = 0
+    proc hello(a: int) = inc num, a
+
+    let b = toTask hello(13)
+    b.invoke()
+    assert num == 13
+    # A task can be invoked multiple times
+    b.invoke()
+    assert num == 26
+
+  block:
+    type
+      Runnable = ref object
+        data: int
+
+    var data: int
+    proc hello(a: Runnable) {.nimcall.} =
+      a.data += 2
+      data = a.data
+
+
+    when false:
+      # the parameters of call must be isolated.
+      let x = Runnable(data: 12)
+      let b = toTask hello(x) # error ----> expression cannot be isolated: x
+      b.invoke()
+
+    let b = toTask(hello(Runnable(data: 12)))
+    b.invoke()
+    assert data == 14
+    b.invoke()
+    assert data == 16