diff options
| -rw-r--r-- | lib/pure/asyncdispatch.nim | 31 | ||||
| -rw-r--r-- | lib/pure/asyncfutures.nim (renamed from lib/pure/includes/asyncfutures.nim) | 12 | ||||
| -rw-r--r-- | lib/upcoming/asyncdispatch.nim | 30 | ||||
| -rw-r--r-- | tests/async/tasyncrecursion.nim | 1 |
4 files changed, 52 insertions, 22 deletions
diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim index 8c1cf6b18..59218acce 100644 --- a/lib/pure/asyncdispatch.nim +++ b/lib/pure/asyncdispatch.nim @@ -9,11 +9,11 @@ include "system/inclrtl" -import os, tables, strutils, times, heapqueue, options - +import os, tables, strutils, times, heapqueue, options, asyncfutures import nativesockets, net, deques export Port, SocketFlag +export asyncfutures #{.injectStmt: newGcInvariant().} @@ -159,8 +159,6 @@ export Port, SocketFlag # TODO: Check if yielded future is nil and throw a more meaningful exception -include includes/asyncfutures - type PDispatcherBase = ref object of RootRef timers*: HeapQueue[tuple[finishAt: float, fut: Future[void]]] @@ -190,6 +188,12 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} = result = int((timerTimeout - curTime) * 1000) if result < 0: result = 0 +proc callSoon*(cbproc: proc ()) {.gcsafe.} + +proc initGlobalDispatcher = + if asyncfutures.callSoonProc == nil: + asyncfutures.callSoonProc = callSoon + when defined(windows) or defined(nimdoc): import winlean, sets, hashes type @@ -237,15 +241,17 @@ when defined(windows) or defined(nimdoc): result.callbacks = initDeque[proc ()](64) var gDisp{.threadvar.}: PDispatcher ## Global dispatcher - proc getGlobalDispatcher*(): PDispatcher = - ## Retrieves the global thread-local dispatcher. - if gDisp.isNil: gDisp = newDispatcher() - result = gDisp proc setGlobalDispatcher*(disp: PDispatcher) = if not gDisp.isNil: assert gDisp.callbacks.len == 0 gDisp = disp + initGlobalDispatcher() + + proc getGlobalDispatcher*(): PDispatcher = + if gDisp.isNil: + setGlobalDispatcher(newDispatcher()) + result = gDisp proc register*(fd: AsyncFD) = ## Registers ``fd`` with the dispatcher. @@ -932,14 +938,17 @@ else: result.callbacks = initDeque[proc ()](64) var gDisp{.threadvar.}: PDispatcher ## Global dispatcher - proc getGlobalDispatcher*(): PDispatcher = - if gDisp.isNil: gDisp = newDispatcher() - result = gDisp proc setGlobalDispatcher*(disp: PDispatcher) = if not gDisp.isNil: assert gDisp.callbacks.len == 0 gDisp = disp + initGlobalDispatcher() + + proc getGlobalDispatcher*(): PDispatcher = + if gDisp.isNil: + setGlobalDispatcher(newDispatcher()) + result = gDisp proc update(fd: AsyncFD, events: set[Event]) = let p = getGlobalDispatcher() diff --git a/lib/pure/includes/asyncfutures.nim b/lib/pure/asyncfutures.nim index 6af5bf3cf..8a1538546 100644 --- a/lib/pure/includes/asyncfutures.nim +++ b/lib/pure/asyncfutures.nim @@ -1,3 +1,6 @@ +include "system/inclrtl" + +import os, tables, strutils, times, heapqueue, options, deques # TODO: This shouldn't need to be included, but should ideally be exported. type @@ -30,7 +33,14 @@ type when not defined(release): var currentID = 0 -proc callSoon*(cbproc: proc ()) {.gcsafe.} +var callSoonProc* {.threadvar.}: (proc(cbproc: proc ()) {.gcsafe.}) + +proc callSoon(cbproc: proc ()) = + if callSoonProc == nil: + # Loop not initialized yet. Call the function directly to allow setup code to use futures. + cbproc() + else: + callSoonProc(cbproc) template setupFutureBase(fromProc: string) = new(result) diff --git a/lib/upcoming/asyncdispatch.nim b/lib/upcoming/asyncdispatch.nim index 1623d8375..84c33f3ab 100644 --- a/lib/upcoming/asyncdispatch.nim +++ b/lib/upcoming/asyncdispatch.nim @@ -9,11 +9,12 @@ include "system/inclrtl" -import os, tables, strutils, times, heapqueue, lists, options +import os, tables, strutils, times, heapqueue, lists, options, asyncfutures import nativesockets, net, deques export Port, SocketFlag +export asyncfutures #{.injectStmt: newGcInvariant().} @@ -130,8 +131,6 @@ export Port, SocketFlag # TODO: Check if yielded future is nil and throw a more meaningful exception -include "../includes/asyncfutures" - type PDispatcherBase = ref object of RootRef timers: HeapQueue[tuple[finishAt: float, fut: Future[void]]] @@ -161,6 +160,12 @@ proc adjustedTimeout(p: PDispatcherBase, timeout: int): int {.inline.} = result = int((timerTimeout - curTime) * 1000) if result < 0: result = 0 +proc callSoon*(cbproc: proc ()) {.gcsafe.} + +proc initGlobalDispatcher = + if asyncfutures.callSoonProc == nil: + asyncfutures.callSoonProc = callSoon + when defined(windows) or defined(nimdoc): import winlean, sets, hashes type @@ -214,15 +219,17 @@ when defined(windows) or defined(nimdoc): result.callbacks = initDeque[proc ()](64) var gDisp{.threadvar.}: PDispatcher ## Global dispatcher - proc getGlobalDispatcher*(): PDispatcher = - ## Retrieves the global thread-local dispatcher. - if gDisp.isNil: gDisp = newDispatcher() - result = gDisp proc setGlobalDispatcher*(disp: PDispatcher) = if not gDisp.isNil: assert gDisp.callbacks.len == 0 gDisp = disp + initGlobalDispatcher() + + proc getGlobalDispatcher*(): PDispatcher = + if gDisp.isNil: + setGlobalDispatcher(newDispatcher()) + result = gDisp proc register*(fd: AsyncFD) = ## Registers ``fd`` with the dispatcher. @@ -1081,14 +1088,17 @@ else: result.callbacks = initDeque[proc ()](InitDelayedCallbackListSize) var gDisp{.threadvar.}: PDispatcher ## Global dispatcher - proc getGlobalDispatcher*(): PDispatcher = - if gDisp.isNil: gDisp = newDispatcher() - result = gDisp proc setGlobalDispatcher*(disp: PDispatcher) = if not gDisp.isNil: assert gDisp.callbacks.len == 0 gDisp = disp + initGlobalDispatcher() + + proc getGlobalDispatcher*(): PDispatcher = + if gDisp.isNil: + setGlobalDispatcher(newDispatcher()) + result = gDisp proc register*(fd: AsyncFD) = let p = getGlobalDispatcher() diff --git a/tests/async/tasyncrecursion.nim b/tests/async/tasyncrecursion.nim index 54482edab..1aeebe9b4 100644 --- a/tests/async/tasyncrecursion.nim +++ b/tests/async/tasyncrecursion.nim @@ -17,5 +17,6 @@ proc asyncRecursionTest*(): Future[int] {.async.} = inc(i) when isMainModule: + setGlobalDispatcher(newDispatcher()) var i = waitFor asyncRecursionTest() echo i |