async: minor refactorings (#15354)

5 files changed, 47 insertions, 66 deletions

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index 41d67d172..6f38583d8 100644
--- a/lib/pure/asyncdispatch.nim
+++ b/lib/pure/asyncdispatch.nim
@@ -290,7 +290,7 @@ when defined(windows) or defined(nimdoc):
 
   var gDisp{.threadvar.}: owned PDispatcher ## Global dispatcher
 
-  proc setGlobalDispatcher*(disp: owned PDispatcher) =
+  proc setGlobalDispatcher*(disp: sink PDispatcher) =
     if not gDisp.isNil:
       assert gDisp.callbacks.len == 0
     gDisp = disp
@@ -1217,10 +1217,12 @@ else:
     withData(selector, fd.int, fdData):
       case event
       of Event.Read:
-        shallowCopy(curList, fdData.readList)
+        #shallowCopy(curList, fdData.readList)
+        curList = move fdData.readList
         fdData.readList = newSeqOfCap[Callback](InitCallbackListSize)
       of Event.Write:
-        shallowCopy(curList, fdData.writeList)
+        #shallowCopy(curList, fdData.writeList)
+        curList = move fdData.writeList
         fdData.writeList = newSeqOfCap[Callback](InitCallbackListSize)
       else:
         assert false, "Cannot process callbacks for " & $event
@@ -1232,8 +1234,7 @@ else:
     for cb in curList:
       if eventsExtinguished:
         newList.add(cb)
-        continue
-      if not cb(fd):
+      elif not cb(fd):
         # Callback wants to be called again.
         newList.add(cb)
         # This callback has returned with EAGAIN, so we don't need to
@@ -1259,15 +1260,15 @@ else:
       result.readCbListCount = -1
       result.writeCbListCount = -1
 
-  template processCustomCallbacks(ident: untyped) =
+  proc processCustomCallbacks(p: PDispatcher; fd: AsyncFD) =
     # Process pending custom event callbacks. Custom events are
     # {Event.Timer, Event.Signal, Event.Process, Event.Vnode}.
     # There can be only one callback registered with one descriptor,
     # so there is no need to iterate over list.
     var curList: seq[Callback]
 
-    withData(p.selector, ident.int, adata) do:
-      shallowCopy(curList, adata.readList)
+    withData(p.selector, fd.int, adata) do:
+      curList = move adata.readList
       adata.readList = newSeqOfCap[Callback](InitCallbackListSize)
 
     let newLength = len(curList)
@@ -1277,7 +1278,7 @@ else:
     if not cb(fd.AsyncFD):
       newList.add(cb)
 
-    withData(p.selector, ident.int, adata) do:
+    withData(p.selector, fd.int, adata) do:
       # descriptor still present in queue.
       adata.readList = newList & adata.readList
       if len(adata.readList) == 0:
@@ -1308,10 +1309,6 @@ else:
 
   proc runOnce(timeout = 500): bool =
     let p = getGlobalDispatcher()
-    when ioselSupportedPlatform:
-      let customSet = {Event.Timer, Event.Signal, Event.Process,
-                       Event.Vnode}
-
     if p.selector.isEmpty() and p.timers.len == 0 and p.callbacks.len == 0:
       raise newException(ValueError,
         "No handles or timers registered in dispatcher.")
@@ -1346,9 +1343,11 @@ else:
         result = true
 
       when ioselSupportedPlatform:
+        const customSet = {Event.Timer, Event.Signal, Event.Process,
+                           Event.Vnode}
         if (customSet * events) != {}:
           isCustomEvent = true
-          processCustomCallbacks(fd)
+          processCustomCallbacks(p, fd)
           result = true
 
       # because state `data` can be modified in callback we need to update
@@ -1612,7 +1611,7 @@ proc drain*(timeout = 500) =
   var curTimeout = timeout
   let start = now()
   while hasPendingOperations():
-    discard runOnce(curTimeout) 
+    discard runOnce(curTimeout)
     curTimeout -= (now() - start).inMilliseconds.int
     if curTimeout < 0:
       break
diff --git a/lib/pure/asyncfutures.nim b/lib/pure/asyncfutures.nim
index c1e2770bb..97abf72c9 100644
--- a/lib/pure/asyncfutures.nim
+++ b/lib/pure/asyncfutures.nim
@@ -157,33 +157,15 @@ proc checkFinished[T](future: Future[T]) =
       raise err
 
 proc call(callbacks: var CallbackList) =
-  when not defined(nimV2):
-    # strictly speaking a little code duplication here, but we strive
-    # to minimize regressions and I'm not sure I got the 'nimV2' logic
-    # right:
-    var current = callbacks
-    while true:
-      if not current.function.isNil:
-        callSoon(current.function)
-
-      if current.next.isNil:
-        break
-      else:
-        current = current.next[]
-  else:
-    var currentFunc = unown callbacks.function
-    var currentNext = unown callbacks.next
-
-    while true:
-      if not currentFunc.isNil:
-        callSoon(currentFunc)
-
-      if currentNext.isNil:
-        break
-      else:
-        currentFunc = currentNext.function
-        currentNext = unown currentNext.next
+  var current = callbacks
+  while true:
+    if not current.function.isNil:
+      callSoon(current.function)
 
+    if current.next.isNil:
+      break
+    else:
+      current = current.next[]
   # callback will be called only once, let GC collect them now
   callbacks.next = nil
   callbacks.function = nil
diff --git a/lib/pure/asyncstreams.nim b/lib/pure/asyncstreams.nim
index 393262c4f..7ffde9c10 100644
--- a/lib/pure/asyncstreams.nim
+++ b/lib/pure/asyncstreams.nim
@@ -54,7 +54,8 @@ proc `callback=`*[T](future: FutureStream[T],
   ##
   ## If the future stream already has data or is finished then ``cb`` will be
   ## called immediately.
-  future.cb = proc () = cb(future)
+  proc named() = cb(future)
+  future.cb = named
   if future.queue.len > 0 or future.finished:
     callSoon(future.cb)
 
@@ -90,27 +91,26 @@ proc read*[T](future: FutureStream[T]): owned(Future[(bool, T)]) =
   ## ``FutureStream``.
   var resFut = newFuture[(bool, T)]("FutureStream.take")
   let savedCb = future.cb
-  var newCb =
-    proc (fs: FutureStream[T]) =
-      # Exit early if `resFut` is already complete. (See #8994).
-      if resFut.finished: return
-
-      # We don't want this callback called again.
-      #future.cb = nil
-
-      # The return value depends on whether the FutureStream has finished.
-      var res: (bool, T)
-      if finished(fs):
-        # Remember, this callback is called when the FutureStream is completed.
-        res[0] = false
-      else:
-        res[0] = true
-        res[1] = fs.queue.popFirst()
-
-      resFut.complete(res)
-
-      # If the saved callback isn't nil then let's call it.
-      if not savedCb.isNil: savedCb()
+  proc newCb(fs: FutureStream[T]) =
+    # Exit early if `resFut` is already complete. (See #8994).
+    if resFut.finished: return
+
+    # We don't want this callback called again.
+    #future.cb = nil
+
+    # The return value depends on whether the FutureStream has finished.
+    var res: (bool, T)
+    if finished(fs):
+      # Remember, this callback is called when the FutureStream is completed.
+      res[0] = false
+    else:
+      res[0] = true
+      res[1] = fs.queue.popFirst()
+
+    resFut.complete(res)
+
+    # If the saved callback isn't nil then let's call it.
+    if not savedCb.isNil: savedCb()
 
   if future.queue.len > 0 or future.finished:
     newCb(future)
diff --git a/lib/pure/ioselects/ioselectors_epoll.nim b/lib/pure/ioselects/ioselectors_epoll.nim
index 1b317f8dc..3dcf547bd 100644
--- a/lib/pure/ioselects/ioselectors_epoll.nim
+++ b/lib/pure/ioselects/ioselectors_epoll.nim
@@ -514,7 +514,7 @@ template withData*[T](s: Selector[T], fd: SocketHandle|int, value,
   let fdi = int(fd)
   s.checkFd(fdi)
   if fdi in s:
-    var value = addr(s.getData(fdi))
+    var value = addr(s.fds[fdi].data)
     body
 
 template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1,
@@ -523,7 +523,7 @@ template withData*[T](s: Selector[T], fd: SocketHandle|int, value, body1,
   let fdi = int(fd)
   s.checkFd(fdi)
   if fdi in s:
-    var value = addr(s.getData(fdi))
+    var value = addr(s.fds[fdi].data)
     body1
   else:
     body2
diff --git a/lib/pure/selectors.nim b/lib/pure/selectors.nim
index f3726efaa..e89cc7447 100644
--- a/lib/pure/selectors.nim
+++ b/lib/pure/selectors.nim
@@ -255,7 +255,7 @@ else:
     IOSelectorsException* = object of CatchableError
 
     ReadyKey* = object
-      fd* : int
+      fd*: int
       events*: set[Event]
       errorCode*: OSErrorCode