disallow reentrancy

Reentrancy causes event reordering and stack explosions, in addition to
leading to hard-to-debug scenarios.

Since none of chronos is actually tested under reentrant conditions (ie
that all features such as exceptions, cancellations, buffer operations,
timers etc work reliably when the loop is reentered), this is hard to
support over time and prevents useful optimizations - this PR simply
detects and disallows the behaviour to remove the uncertainty,
simplifying reasoning about the event loop in general.
This commit is contained in:
Jacek Sieka 2023-04-25 16:34:55 +02:00
parent 1306170255
commit 50abdc4954
No known key found for this signature in database
GPG Key ID: A1B09461ABB656B8
3 changed files with 36 additions and 75 deletions

View File

@ -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

View File

@ -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

View File

@ -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