disallow reentrancy

chronos/internal/asyncengine.nim

@@ -64,20 +64,23 @@ type
     ticks*: Deque[AsyncCallback]
     trackers*: Table[string, TrackerBase]
     counters*: Table[string, TrackerCounter]
-
-proc sentinelCallbackImpl(arg: pointer) {.gcsafe, noreturn.} =
-  raiseAssert "Sentinel callback MUST not be scheduled"
-
-const
-  SentinelCallback = AsyncCallback(function: sentinelCallbackImpl,
-                                   udata: nil)
-
-proc isSentinel(acb: AsyncCallback): bool =
-  acb == SentinelCallback
+    polling*: bool
+      ## The event loop is currently running
 
 proc `<`(a, b: TimerCallback): bool =
   result = a.finishAt < b.finishAt
 
+template preparePoll(loop: PDispatcherBase) =
+  # If you hit this assert, you've called `poll`, `runForever` or `waitFor`
+  # from within an async function which is not supported due to the difficulty
+  # to control stack depth and event ordering
+  # If you're using `waitFor`, switch to `await` and / or propagate the
+  # up the call stack.
+  doAssert not loop.polling, "The event loop and chronos functions in general are not reentrant"
+
+  loop.polling = true
+  defer: loop.polling = false
+
 func getAsyncTimestamp*(a: Duration): auto {.inline.} =
   ## Return rounded up value of duration with milliseconds resolution.
   ##
@@ -142,10 +145,10 @@ template processTicks(loop: untyped) =
     loop.callbacks.addLast(loop.ticks.popFirst())
 
 template processCallbacks(loop: untyped) =
-  while true:
-    let callable = loop.callbacks.popFirst()  # len must be > 0 due to sentinel
-    if isSentinel(callable):
-      break
+  # Process existing callbacks but not those that follow, to allow the network
+  # to regain control regularly
+  for _ in 0..<loop.callbacks.len():
+    let callable = loop.callbacks.popFirst()
     if not(isNil(callable.function)):
       callable.function(callable.udata)
 
@@ -337,7 +340,6 @@ elif defined(windows):
       trackers: initTable[string, TrackerBase](),
       counters: initTable[string, TrackerCounter]()
     )
-    res.callbacks.addLast(SentinelCallback)
     initAPI(res)
     res
 
@@ -585,16 +587,13 @@ elif defined(windows):
 
   proc poll*() =
     let loop = getThreadDispatcher()
+    loop.preparePoll()
+
     var
       curTime = Moment.now()
       curTimeout = DWORD(0)
       events: array[MaxEventsCount, osdefs.OVERLAPPED_ENTRY]
 
-    # On reentrant `poll` calls from `processCallbacks`, e.g., `waitFor`,
-    # complete pending work of the outer `processCallbacks` call.
-    # On non-reentrant `poll` calls, this only removes sentinel element.
-    processCallbacks(loop)
-
     # Moving expired timers to `loop.callbacks` and calculate timeout
     loop.processTimersGetTimeout(curTimeout)
 
@@ -660,14 +659,10 @@ elif defined(windows):
     # We move tick callbacks to `loop.callbacks` always.
     processTicks(loop)
 
-    # All callbacks which will be added during `processCallbacks` will be
-    # scheduled after the sentinel and are processed on next `poll()` call.
-    loop.callbacks.addLast(SentinelCallback)
+    # Process the callbacks currently scheduled - new callbacks scheduled during
+    # callback execution will run in the next poll iteration
     processCallbacks(loop)
 
-    # All callbacks done, skip `processCallbacks` at start.
-    loop.callbacks.addFirst(SentinelCallback)
-
   proc closeSocket*(fd: AsyncFD, aftercb: CallbackFunc = nil) =
     ## Closes a socket and ensures that it is unregistered.
     let loop = getThreadDispatcher()
@@ -758,7 +753,6 @@ elif defined(macosx) or defined(freebsd) or defined(netbsd) or
       trackers: initTable[string, TrackerBase](),
       counters: initTable[string, TrackerCounter]()
     )
-    res.callbacks.addLast(SentinelCallback)
     initAPI(res)
     res
 
@@ -1014,14 +1008,11 @@ elif defined(macosx) or defined(freebsd) or defined(netbsd) or
   proc poll*() {.gcsafe.} =
     ## Perform single asynchronous step.
     let loop = getThreadDispatcher()
+    loop.preparePoll()
+
     var curTime = Moment.now()
     var curTimeout = 0
 
-    # On reentrant `poll` calls from `processCallbacks`, e.g., `waitFor`,
-    # complete pending work of the outer `processCallbacks` call.
-    # On non-reentrant `poll` calls, this only removes sentinel element.
-    processCallbacks(loop)
-
     # Moving expired timers to `loop.callbacks` and calculate timeout.
     loop.processTimersGetTimeout(curTimeout)
 
@@ -1068,14 +1059,10 @@ elif defined(macosx) or defined(freebsd) or defined(netbsd) or
     # We move tick callbacks to `loop.callbacks` always.
     processTicks(loop)
 
-    # All callbacks which will be added during `processCallbacks` will be
-    # scheduled after the sentinel and are processed on next `poll()` call.
-    loop.callbacks.addLast(SentinelCallback)
+    # Process the callbacks currently scheduled - new callbacks scheduled during
+    # callback execution will run in the next poll iteration
     processCallbacks(loop)
 
-    # All callbacks done, skip `processCallbacks` at start.
-    loop.callbacks.addFirst(SentinelCallback)
-
 else:
   proc initAPI() = discard
   proc globalInit() = discard

tests/testbugs.nim

@@ -101,40 +101,6 @@ suite "Asynchronous issues test suite":
 
     result = r1 and r2
 
-  proc createBigMessage(size: int): seq[byte] =
-    var message = "MESSAGE"
-    var res = newSeq[byte](size)
-    for i in 0 ..< len(result):
-      res[i] = byte(message[i mod len(message)])
-    res
-
-  proc testIndexError(): Future[bool] {.async.} =
-    var server = createStreamServer(initTAddress("127.0.0.1:0"),
-                                    flags = {ReuseAddr})
-    let messageSize = DefaultStreamBufferSize * 4
-    var buffer = newSeq[byte](messageSize)
-    let msg = createBigMessage(messageSize)
-    let address = server.localAddress()
-    let afut = server.accept()
-    let outTransp = await connect(address)
-    let inpTransp = await afut
-    let bytesSent = await outTransp.write(msg)
-    check bytesSent == messageSize
-    var rfut {.used.} = inpTransp.readExactly(addr buffer[0], messageSize)
-
-    proc waiterProc(udata: pointer) {.raises: [], gcsafe.} =
-      try:
-        waitFor(sleepAsync(0.milliseconds))
-      except CatchableError:
-        raiseAssert "Unexpected exception happened"
-    let timer {.used.} = setTimer(Moment.fromNow(0.seconds), waiterProc, nil)
-    await sleepAsync(100.milliseconds)
-
-    await inpTransp.closeWait()
-    await outTransp.closeWait()
-    await server.closeWait()
-    return true
-
   test "Issue #6":
     check waitFor(issue6()) == true
 
@@ -149,6 +115,3 @@ suite "Asynchronous issues test suite":
 
   test "Defer for asynchronous procedures test [Nim's issue #13899]":
     check waitFor(testDefer()) == true
-
-  test "IndexError crash test":
-    check waitFor(testIndexError()) == true

tests/testmacro.nim

@@ -555,3 +555,14 @@ suite "Exceptions tracking":
         await raiseException()
 
     waitFor(callCatchAll())
+
+  test "No poll re-entrancy allowed":
+    proc testReentrancy() {.async.} =
+      await sleepAsync(1.milliseconds)
+      poll()
+
+    let reenter = testReentrancy()
+    expect(Defect):
+      waitFor reenter
+
+    waitFor reenter.cancelAndWait() # avoid pending future leaks