Expose isReady to check if sync will block (#123 #22)

weave.nim

@@ -20,7 +20,7 @@ export
  parallelFor, parallelForStrided,
  init, exit,
  loadBalance,
- isSpawned,
+ isSpawned, isReady,
  getThreadId, getNumThreads,
  # Experimental threadlocal prologue/epilogue
  parallelForStaged, parallelForStagedStrided,

weave/datatypes/flowvars.nim

@@ -83,9 +83,16 @@ EagerFV:
  let resultSent {.used.} = fv.chan[].trySend(childResult)
  postCondition: resultSent
 
- template isFutReady*(fv: Flowvar): bool =
+ template tryComplete*[T](fv: Flowvar, parentResult: var T): bool =
  fv.chan[].tryRecv(parentResult)
 
+ func isReady*[T](fv: Flowvar[T]): bool {.inline.} =
+ ## Returns true if the result of a Flowvar is ready.
+ ## In that case `sync` will not block.
+ ## Otherwise the current will block to help on all the pending tasks
+ ## until the Flowvar is ready.
+ not fv.chan.isEmpty()
+
 LazyFV:
  proc recycleChannel*(fv: Flowvar) {.inline.} =
  recycle(fv.lfv.lazy.chan)
@@ -108,13 +115,23 @@ LazyFV:
  ascertain: not fv.lfv.lazy.chan.isNil
  discard fv.lfv.lazy.chan[].trySend(childResult)
 
- template isFutReady*(fv: Flowvar): bool =
+ template tryComplete*[T](fv: Flowvar, parentResult: var T): bool =
  if fv.lfv.hasChannel:
  ascertain: not fv.lfv.lazy.chan.isNil
  fv.lfv.lazy.chan[].tryRecv(parentResult)
  else:
  fv.lfv.isReady
 
+ func isReady*[T](fv: Flowvar[T]): bool {.inline.} =
+ ## Returns true if the result of a Flowvar is ready.
+ ## In that case `sync` will not block.
+ ## Otherwise the current will block to help on all the pending tasks
+ ## until the Flowvar is ready.
+ if not fv.lfv.hasChannel:
+ fv.lfv.isReady
+ else:
+ not fv.lfv.lazy.chan.isEmpty()
+
  import sync_types
 
  proc convertLazyFlowvar*(task: Task) {.inline.}=

weave/state_machines/sync.dot

@@ -1,27 +1,27 @@
 digraph awaitFSA{
  splines=ortho;
  node [shape = doublecircle]; InitialState AW_Exit;
- node [shape = circle, fontcolor=white, fillcolor=darkslategrey, style="filled"]; AW_Steal AW_SuccessfulTheft AW_CheckTask AW_OutOfChildTasks;
+ node [shape = circle, fontcolor=white, fillcolor=darkslategrey, style="filled"]; AW_Steal AW_SuccessfulTheft AW_CheckTask AW_OutOfDirectChildTasks;
  InitialState -> AW_CheckTask [color="black:invis:black", xlabel="entry point"];
  node [shape = octagon, fontcolor=black, fillcolor=lightsteelblue, style="rounded,filled"]; AW_Steal_AWE_ReceivedTask AW_CheckTask_AWE_HasChildTask ;
- node [shape = diamond, fontcolor=black, fillcolor=coral, style="rounded,filled"]; AW_Steal_AWE_FutureReady AW_CheckTask_AWE_FutureReady AW_OutOfChildTasks_AWE_FutureReady ;
+ node [shape = diamond, fontcolor=black, fillcolor=coral, style="rounded,filled"]; AW_Steal_AWE_FutureReady AW_CheckTask_AWE_FutureReady AW_OutOfDirectChildTasks_AWE_FutureReady ;
  AW_Steal_AWE_ReceivedTask [label="AWE_ReceivedTask\nlootedTask"];
  AW_CheckTask_AWE_HasChildTask [label="AWE_HasChildTask\nnot task.isNil"];
- AW_Steal_AWE_FutureReady [label="AWE_FutureReady\nisFutReady(fv)"];
- AW_CheckTask_AWE_FutureReady [label="AWE_FutureReady\nisFutReady(fv)"];
- AW_OutOfChildTasks_AWE_FutureReady [label="AWE_FutureReady\nisFutReady(fv)"];
+ AW_Steal_AWE_FutureReady [label="AWE_FutureReady\ntryComplete(fv, parentResult)"];
+ AW_CheckTask_AWE_FutureReady [label="AWE_FutureReady\ntryComplete(fv, parentResult)"];
+ AW_OutOfDirectChildTasks_AWE_FutureReady [label="AWE_FutureReady\ntryComplete(fv, parentResult)"];
  AW_Steal -> AW_Steal_AWE_FutureReady[style=bold, xlabel="always"];
  AW_Steal_AWE_FutureReady -> AW_Exit [color="coral", fontcolor="coral", xlabel="interrupted"];
  AW_Steal_AWE_FutureReady -> AW_Steal_AWE_ReceivedTask[xlabel="normal flow"];
  AW_Steal_AWE_ReceivedTask -> AW_SuccessfulTheft [style=dashed, xlabel="true"];
  AW_Steal_AWE_ReceivedTask -> AW_Steal [xlabel="default"];
- AW_SuccessfulTheft -> AW_OutOfChildTasks [xlabel="default"];
+ AW_SuccessfulTheft -> AW_OutOfDirectChildTasks [xlabel="default"];
  AW_CheckTask -> AW_CheckTask_AWE_FutureReady[style=bold, xlabel="always"];
  AW_CheckTask_AWE_FutureReady -> AW_Exit [color="coral", fontcolor="coral", xlabel="interrupted"];
  AW_CheckTask_AWE_FutureReady -> AW_CheckTask_AWE_HasChildTask[xlabel="normal flow"];
  AW_CheckTask_AWE_HasChildTask -> AW_CheckTask [style=dashed, xlabel="true"];
- AW_CheckTask_AWE_HasChildTask -> AW_OutOfChildTasks [xlabel="default"];
- AW_OutOfChildTasks -> AW_OutOfChildTasks_AWE_FutureReady[style=bold, xlabel="always"];
- AW_OutOfChildTasks_AWE_FutureReady -> AW_Exit [color="coral", fontcolor="coral", xlabel="interrupted"];
- AW_OutOfChildTasks_AWE_FutureReady -> AW_Steal [xlabel="default"];
+ AW_CheckTask_AWE_HasChildTask -> AW_OutOfDirectChildTasks [xlabel="default"];
+ AW_OutOfDirectChildTasks -> AW_OutOfDirectChildTasks_AWE_FutureReady[style=bold, xlabel="always"];
+ AW_OutOfDirectChildTasks_AWE_FutureReady -> AW_Exit [color="coral", fontcolor="coral", xlabel="interrupted"];
+ AW_OutOfDirectChildTasks_AWE_FutureReady -> AW_Steal [xlabel="default"];
 }

weave/state_machines/sync.nim

@@ -48,7 +48,7 @@ setTerminalState(awaitFSA, AW_Exit)
 # -------------------------------------------
 
 implEvent(awaitFSA, AWE_FutureReady):
- isFutReady(fv)
+ tryComplete(fv, parentResult)
 
 behavior(awaitFSA):
  # In AW_Steal we might recv tasks and steal requests which get stuck in our queues