[Backport] Fixes callbacks being dropped on Linux/macOS/BSD. (#15012)

Fixes #15003.

This is a serious bug which occurs when data cannot be read/sent
immediately and there are a bunch of other read/write events
pending. What happens is that the new events are dropped which
results in the case of the reported bug resulted in some data not
being sent (!).

(cherry picked from commit 1e3a0ef)

lib/pure/asyncdispatch.nim

@@ -1211,14 +1211,19 @@ else:
     let newLength = max(len(curList), InitCallbackListSize)
     var newList = newSeqOfCap[Callback](newLength)
 
+    var eventsExtinguished = false
     for cb in curList:
+      if eventsExtinguished:
+        newList.add(cb)
+        continue
       if not cb(fd):
         # Callback wants to be called again.
         newList.add(cb)
         # This callback has returned with EAGAIN, so we don't need to
         # call any other callbacks as they are all waiting for the same event
         # on the same fd.
-        break
+        # We do need to ensure they are called again though.
+        eventsExtinguished = true
 
     withData(selector, fd.int, fdData) do:
       # Descriptor is still present in the queue.

tests/async/tasynceagain.nim

@@ -0,0 +1,67 @@
+discard """
+  disabled: "windows"
+  exitcode: 0
+"""
+# AsyncSocketBug.nim
+# Jens Alfke (@snej) -- 16 July 2020
+# Demonstrates data loss by Nim's AsyncSocket.
+# Just run it, and it will raise an assertion failure within a minute.
+
+import asyncdispatch, asyncnet, strformat, strutils, sugar
+
+const FrameSize = 9999   # Exact size not important, but larger sizes fail quicker
+
+proc runServer() {.async.} =
+  # Server side:
+  var server = newAsyncSocket()
+  server.bindAddr(Port(9001))
+  server.listen()
+  let client = await server.accept()
+  echo "Server got client connection"
+  var lastN = 0
+  while true:
+    let frame = await client.recv(FrameSize)
+    assert frame.len == FrameSize
+    let n = frame[0..<6].parseInt()
+    echo "RCVD #", n, ":  ", frame[0..80], "..."
+    if n != lastN + 1:
+      echo &"******** ERROR: Server received #{n}, but last was #{lastN}!"
+    assert n == lastN + 1
+    lastN = n
+    await sleepAsync 100
+
+
+proc main() {.async.} =
+  asyncCheck runServer()
+
+  # Client side:
+  let socket = newAsyncSocket(buffered = false)
+  await socket.connect("localhost", Port(9001))
+  echo "Client socket connected"
+
+  var sentCount = 0
+  var completedCount = 0
+
+  while sentCount < 2000:
+    sentCount += 1
+    let n = sentCount
+
+    var message = &"{n:06} This is message #{n} of ∞. Please stay tuned for more. "
+    #echo ">>> ", message
+    while message.len < FrameSize:
+      message = message & message
+    let frame = message[0..<FrameSize]
+
+    capture n:
+      socket.send(frame).addCallback proc(f: Future[void]) =
+        # Callback when the send completes:
+        assert not f.failed
+        echo "SENT #", n
+        if n != completedCount + 1:
+          echo &"******** ERROR: Client completed #{n}, but last completed was #{completedCount}!"
+        # If this assert is enabled, it will trigger earlier than the server-side assert above:
+        assert n == completedCount + 1
+        completedCount = n
+    await sleepAsync 1
+
+waitFor main()