Virtual async events 2

changelog.md

@@ -22,6 +22,8 @@
 - `macros.newLit` now works for ref object types.
 - `system.writeFile` has been overloaded to also support `openarray[byte]`.
 - Added overloaded `strformat.fmt` macro that use specified characters as delimiter instead of '{' and '}'.
+- Added a new Async Event implementation: `asyncdispatch.VirtualAsyncEvent`.
+  - The implementation is designed to enable efficient coordination of async code across threads.
 
 ## Library changes
 

lib/pure/asyncdispatch.nim

@@ -169,17 +169,27 @@
 include "system/inclrtl"
 
 import os, tables, strutils, times, heapqueue, lists, options, asyncstreams
-import options, math, std/monotimes
+import options, math, std/monotimes, hashes
 import asyncfutures except callSoon
 
 import nativesockets, net, deques
 
+when compileOption("threads"):
+  import rlocks
+
 export Port, SocketFlag
 export asyncfutures except callSoon
 export asyncstreams
 
 #{.injectStmt: newGcInvariant().}
 
+template withLockIfThreads(lock, code: untyped) =
+  when compileOption("threads"):
+    withRLock lock:
+      code
+  else:
+    code
+
 # TODO: Check if yielded future is nil and throw a more meaningful exception
 
 type
@@ -246,6 +256,12 @@ when defined(windows) or defined(nimdoc):
     PDispatcher* = ref object of PDispatcherBase
       ioPort: Handle
       handles: HashSet[AsyncFD]
+      vd: VirtualEventDispatcher
+
+    VirtualEventDispatcher = ref object
+      virtualHandles: Table[VirtualFD, VirtualAsyncEvent] # pseudo handles for custom AsyncEvents.
+      nextVirtualHandle: VirtualFD
+      virtualMuxHandle: AsyncEvent # all the virtual handles get multiplexed through a single real handle.
 
     CustomOverlapped = object of OVERLAPPED
       data*: CompletionData
@@ -267,10 +283,26 @@ when defined(windows) or defined(nimdoc):
       pcd: PostCallbackDataPtr
     AsyncEvent* = ptr AsyncEventImpl
 
+    VirtualFD* = distinct int
+
+    VirtualAsyncEventImpl = object
+      triggered: bool
+      when compileOption("threads"):
+        eventLock: RLock
+      vd: VirtualEventDispatcher
+      vFD: VirtualFD
+      cb: Callback
+    VirtualAsyncEvent* = ptr VirtualAsyncEventImpl
+
     Callback = proc (fd: AsyncFD): bool {.closure, gcsafe.}
+    NativeCallback = proc(fd: AsyncFD, bytesCount: DWORD, errcode: OSErrorCode) {.closure,gcsafe.}
 
   proc hash(x: AsyncFD): Hash {.borrow.}
   proc `==`*(x: AsyncFD, y: AsyncFD): bool {.borrow.}
+  proc hash(x: VirtualFD): Hash {.borrow.}
+  proc `==`*(x: VirtualFD, y: VirtualFD): bool {.borrow.}
+
+  const InvalidVirtualFD = VirtualFD(-1)
 
   proc newDispatcher*(): owned PDispatcher =
     ## Creates a new Dispatcher instance.
@@ -279,6 +311,8 @@ when defined(windows) or defined(nimdoc):
     result.handles = initSet[AsyncFD]()
     result.timers.newHeapQueue()
     result.callbacks = initDeque[proc ()](64)
+    result.vd = new VirtualEventDispatcher
+    result.vd.virtualHandles = initTable[VirtualFD, VirtualAsyncEvent]()
 
   var gDisp{.threadvar.}: owned PDispatcher ## Global dispatcher
 
@@ -924,8 +958,8 @@ when defined(windows) or defined(nimdoc):
     ## receiving notifications.
     registerWaitableEvent(fd, cb, FD_WRITE or FD_CONNECT or FD_CLOSE)
 
-  template registerWaitableHandle(p, hEvent, flags, pcd, timeout,
-                                  handleCallback) =
+  proc registerWaitableHandle(p: PDispatcher, hEvent: Handle, flags: DWORD, pcd: PostCallbackDataPtr, timeout: int,
+                                  handleCallback: NativeCallback) =
     let handleFD = AsyncFD(hEvent)
     pcd.ioPort = p.ioPort
     pcd.handleFd = handleFD
@@ -1101,12 +1135,34 @@ else:
 
     AsyncEvent* = distinct SelectEvent
 
+    VirtualFD* = distinct int
+
+    VirtualAsyncEventImpl = object
+      triggered: bool
+      when compileOption("threads"):
+        eventLock: RLock
+      vd: VirtualEventDispatcher
+      vFD: VirtualFD
+      cb: Callback
+    VirtualAsyncEvent* = ptr VirtualAsyncEventImpl
+
     PDispatcher* = ref object of PDispatcherBase
       selector: Selector[AsyncData]
+      vd: VirtualEventDispatcher
+
+    VirtualEventDispatcher = ref object
+      virtualHandles: Table[VirtualFD, VirtualAsyncEvent] # pseudo handles for custom AsyncEvents.
+      nextVirtualHandle: VirtualFD
+      virtualMuxHandle: AsyncEvent # all the virtual handles get multiplexed through a single real handle.
 
   proc `==`*(x, y: AsyncFD): bool {.borrow.}
   proc `==`*(x, y: AsyncEvent): bool {.borrow.}
 
+  proc hash(x: VirtualFD): Hash {.borrow.}
+  proc `==`*(x: VirtualFD, y: VirtualFD): bool {.borrow.}
+
+  const InvalidVirtualFD = VirtualFD(-1)
+
   template newAsyncData(): AsyncData =
     AsyncData(
       readList: newSeqOfCap[Callback](InitCallbackListSize),
@@ -1118,6 +1174,8 @@ else:
     result.selector = newSelector[AsyncData]()
     result.timers.newHeapQueue()
     result.callbacks = initDeque[proc ()](InitDelayedCallbackListSize)
+    result.vd = new VirtualEventDispatcher
+    result.vd.virtualHandles = initTable[VirtualFD, VirtualAsyncEvent]()
 
   var gDisp{.threadvar.}: owned PDispatcher ## Global dispatcher
 
@@ -1851,6 +1909,107 @@ proc send*(socket: AsyncFD, data: string,
 
   return retFuture
 
+proc close(vd: VirtualEventDispatcher) =
+  assert vd.virtualHandles.len == 0, "Cannot close Virtual Event Dispatcher. There are still Pending Events."
+  vd.virtualMuxHandle.unregister()
+  vd.virtualMuxHandle.close()
+  vd.virtualMuxHandle = nil
+  vd.nextVirtualHandle = 0.VirtualFD
+
+proc unregister*(ev: VirtualAsyncEvent) =
+  ## Unregisters event ``ev``.
+  doAssert(ev.vFD != InvalidVirtualFD, "Event is not registered in the queue!")
+
+  let vd = ev.vd
+  withLockIfThreads ev.eventLock:
+    ev.vd.virtualHandles.del ev.vFD
+    ev.vd = nil
+    ev.cb = nil
+    ev.vFD = InvalidVirtualFD
+
+  if vd.virtualHandles.len == 0:
+    # lazy close the physical event with the Dispatcher
+    # The main reason we need this is to make
+    # hasPendingOperations still work
+    # without modifying the ioselector code.
+    vd.close
+
+proc initVirtualEventDispatcher(vd: VirtualEventDispatcher) =
+  vd.virtualMuxHandle = newAsyncEvent()
+
+  proc nativeEventCB(fd: AsyncFD): bool {.gcsafe.} =
+    # find the virtual events that triggered the physical event and
+    # add them to the callback list.
+    # Not very efficient, but requires the least coordination between threads.
+
+    proc vcbFactory(ev: VirtualAsyncEvent): proc() {.gcsafe.} =
+      result =
+        proc() {.gcSafe.} =
+          if ev.cb(ev.vfd.AsyncFD):
+            # the convention is that if the callback returns true,
+            # we unregister the event.
+            ev.unregister
+
+    let disp = getGlobalDispatcher()
+    for vfd, ev in vd.virtualHandles:
+      withLockIfThreads ev.eventLock:
+        if ev.triggered:
+          ev.triggered = false
+          disp.callbacks.addLast(vcbFactory(ev))
+
+    # always return false b/c we never want to unregister this event
+    return false
+
+  vd.virtualMuxHandle.addEvent(nativeEventCB)
+
+proc newVirtualAsyncEvent*(): VirtualAsyncEvent =
+  ## Creates a new thread-safe ``VirtualAsyncEvent`` object.
+  ##
+  ## New ``VirtualAsyncEvent`` object is not automatically registered with
+  ## dispatcher like ``AsyncSocket``.
+  result = cast[VirtualAsyncEvent](allocShared0(sizeof(VirtualAsyncEventImpl)))
+  result.cb = nil
+  result.vFD = InvalidVirtualFD
+  when compileOption("threads"):
+    initRLock result.eventLock
+
+proc trigger*(ev: VirtualAsyncEvent) =
+  ## Sets new ``VirtualAsyncEvent`` to signaled state.
+  withLockIfThreads ev.eventLock:
+    if ev.vFD == InvalidVirtualFD:
+      raise newException(ValueError,
+        "VirtualAsyncEvent is not registered with a dispatcher.")
+    ev.triggered = true
+
+    # send the signal to wake up the dispatcher thread.
+    trigger(ev.vd.virtualMuxHandle)
+
+proc addEvent*(ev: VirtualAsyncEvent, cb: Callback) =
+  ## Registers callback ``cb`` to be called when ``ev`` will be signaled
+  doAssert(ev.vFD == InvalidVirtualFD, "Event is already registered in the queue!")
+
+  let p = getGlobalDispatcher()
+
+  if p.vd.virtualMuxHandle == nil:
+    # lazy register the physical event with the Dispatcher
+    # The main reason we need this is to make hasPendingOperations still work
+    # without modifying the ioselector code.
+    initVirtualEventDispatcher(p.vd)
+
+  withLockIfThreads ev.eventLock:
+    ev.cb = cb
+    ev.vFD = p.vd.nextVirtualHandle
+    p.vd.nextVirtualHandle = VirtualFD(int(p.vd.nextVirtualHandle) + 1)
+    p.vd.virtualHandles[ev.vFD] = ev
+    ev.vd = p.vd
+
+proc close*(ev: VirtualAsyncEvent) =
+  ## Closes event ``ev``.
+  doAssert(ev.vFD != InvalidVirtualFD, "Must unregister Event before you close it!")
+  when compileOption("threads"):
+    deinitRLock(ev.eventLock)
+  deallocShared(cast[pointer](ev))
+
 # -- Await Macro
 include asyncmacro
 

tests/async/testasyncevent.nim

@@ -0,0 +1,13 @@
+discard """
+output: '''
+event triggered!
+'''
+"""
+
+import asyncDispatch
+
+let ev = newAsyncEvent()
+addEvent(ev, proc(fd: AsyncFD): bool {.gcsafe.} = echo "event triggered!"; true)
+ev.trigger()
+
+drain()

tests/async/testmanyasynceventsovertime.nim

@@ -0,0 +1,25 @@
+discard """
+output: '''
+triggerCount: 10
+'''
+"""
+
+import asyncDispatch
+
+var triggerCount = 0
+var evs = newSeq[AsyncEvent]()
+
+for i in 0 ..< 10:
+  var ev = newAsyncEvent()
+  evs.add(ev)
+  addEvent(ev, proc(fd: AsyncFD): bool {.gcsafe,closure.} = triggerCount += 1; true)
+
+proc main() {.async.} =
+  for ev in evs:
+    ev.trigger()
+    await sleepAsync(10)
+    ev.close
+
+asyncCheck main()
+drain()
+echo "triggerCount: ", triggerCount

tests/async/testmanyvirtualasyncevents.nim

@@ -0,0 +1,21 @@
+discard """
+output: '''
+triggerCount: 8000
+'''
+"""
+
+import asyncDispatch
+
+var triggerCount = 0
+var evs = newSeq[VirtualAsyncEvent]()
+
+for i in 0 ..< 8000: # some number way higher than the typical physical fd limit
+  var ev = newVirtualAsyncEvent()
+  evs.add(ev)
+  addEvent(ev, proc(fd: AsyncFD): bool {.gcsafe,closure.} = triggerCount += 1; true)
+
+for ev in evs:
+  ev.trigger()
+
+drain()
+echo "triggerCount: ", triggerCount

tests/async/testmanyvirtualasynceventsovertime.nim

@@ -0,0 +1,28 @@
+discard """
+output: '''
+runForever should throw ValueError, this is expected
+triggerCount: 100
+'''
+"""
+
+import asyncDispatch
+
+var triggerCount = 0
+var evs = newSeq[VirtualAsyncEvent]()
+
+for i in 0 ..< 100:
+  var ev = newVirtualAsyncEvent()
+  evs.add(ev)
+  addEvent(ev, proc(fd: AsyncFD): bool {.gcsafe,closure.} = triggerCount += 1; true)
+
+proc main() {.async.} =
+  for ev in evs:
+    await sleepAsync(10)
+    ev.trigger()
+
+try:
+  asyncCheck main()
+  runForever()
+except ValueError:
+  echo "runForever should throw ValueError, this is expected"
+  echo "triggerCount: ", triggerCount

tests/async/testvirtualasyncevent.nim

@@ -0,0 +1,13 @@
+discard """
+output: '''
+event triggered!
+'''
+"""
+
+import asyncDispatch
+
+let ev = newVirtualAsyncEvent()
+addEvent(ev, proc(fd: AsyncFD): bool {.gcsafe.} = echo "event triggered!"; true)
+ev.trigger()
+
+drain()