wip: send many

.gitignore

@@ -1,2 +1,5 @@
 tests/**
 !tests/**/*.nim
+nimble.develop
+nimble.paths
+

config.nims

@@ -9,3 +9,7 @@ when not defined(windows):
   usages
 --styleCheck:
   error
+# begin Nimble config (version 2)
+when withDir(thisDir(), system.fileExists("nimble.paths")):
+  include "nimble.paths"
+# end Nimble config

lsquic.nimble

@@ -10,7 +10,7 @@ requires "nim >= 2.0.0"
 requires "zlib"
 requires "nim >= 2.0.0"
 requires "stew >= 0.4.0"
-requires "chronos >= 4.0.4"
+requires "https://github.com/status-im/nim-chronos#685ecd7f4c3826fd5fd36b4d0b6f7aaa680d35b5"
 requires "nimcrypto >= 0.6.0"
 requires "unittest2"
 requires "chronicles >= 0.11.0"

lsquic/api.nim

@@ -43,7 +43,7 @@ proc listen*(
 proc new*(
     t: typedesc[QuicClient], tlsConfig: TLSConfig
 ): QuicClient {.raises: [QuicError, TransportOsError].} =
-  let outgoing = newAsyncQueue[Datagram]()
+  let outgoing = ManyQueue[Datagram]()
 
   let clientCtx = ClientContext.new(tlsConfig, outgoing).valueOr:
     raise newException(QuicError, error)

lsquic/connectionmanager.nim

@@ -1,4 +1,4 @@
-import chronicles
+import chronicles, sequtils
 import chronos
 import chronos/osdefs
 import results
@@ -10,7 +10,7 @@ type ConnectionManager* = ref object of RootObj
   tlsConfig*: TLSConfig
   udp*: DatagramTransport
   quicContext*: QuicContext
-  outgoing*: AsyncQueue[Datagram]
+  outgoing*: ManyQueue[Datagram]
   connections: seq[Connection]
   loop*: Future[void]
   closed*: Future[void]
@@ -20,7 +20,7 @@ proc init*(
     tlsConfig: TLSConfig,
     udp: DatagramTransport,
     quicContext: QuicContext,
-    outgoing: AsyncQueue[Datagram],
+    outgoing: ManyQueue[Datagram],
 ) =
   c.tlsConfig = tlsConfig
   c.quicContext = quicContext
@@ -34,7 +34,7 @@ proc new*(
     tlsConfig: TLSConfig,
     udp: DatagramTransport,
     quicContext: QuicContext,
-    outgoing: AsyncQueue[Datagram],
+    outgoing: ManyQueue[Datagram],
 ): T =
   let ret = ConnectionManager()
   ret.init(tlsConfig, udp, quicContext, outgoing)
@@ -50,12 +50,17 @@ proc startSending*(connman: ConnectionManager) =
 
   proc send() {.async: (raises: [CancelledError]).} =
     try:
-      let datagram = await connman.outgoing.get()
-      if datagram.len == 0:
-        # In windows, empty datagrams will make the peer stop receiving data
-        return
-      await connman.udp.sendTo(datagram.taddr, datagram.data)
-    except TransportError as e:
+      let datagrams = await connman.outgoing.get()
+
+      # sendTo many datagrams at once
+      #let req = datagrams.mapIt((it.taddr, it.data))
+      #discard connman.udp.sendTo(req)
+
+      for d in datagrams:
+        await connman.udp.sendTo(d.taddr, d.data)
+    except CancelledError as e:
+      raise e
+    except CatchableError as e:
       debug "Failed to send datagram", errorMsg = e.msg
 
   connman.loop = asyncLoop(send)

lsquic/context/client.nim

@@ -4,7 +4,7 @@ import chronos
 import chronos/osdefs
 import ./[context, io, stream]
 import ../[lsquic_ffi, tlsconfig, datagram, timeout, stream]
-import ../helpers/sequninit
+import ../helpers/[sequninit]
 
 proc onNewConn(
     stream_if_ctx: pointer, conn: ptr lsquic_conn_t
@@ -105,7 +105,7 @@ method dial*(
 const BBRv1 = 2
 
 proc new*(
-    T: typedesc[ClientContext], tlsConfig: TLSConfig, outgoing: AsyncQueue[Datagram]
+    T: typedesc[ClientContext], tlsConfig: TLSConfig, outgoing: ManyQueue[Datagram]
 ): Result[T, string] =
   var ctx = ClientContext()
   ctx.tlsConfig = tlsConfig

lsquic/context/context.nim

@@ -18,7 +18,7 @@ type QuicContext* = ref object of RootObj
   engine*: ptr struct_lsquic_engine
   stream_if*: struct_lsquic_stream_if
   tlsConfig*: TLSConfig
-  outgoing*: AsyncQueue[Datagram]
+  outgoing*: ManyQueue[Datagram]
   tickTimeout*: Timeout
   sslCtx*: ptr SSL_CTX
 

lsquic/context/io.nim

@@ -47,6 +47,9 @@ proc sendPacketsOut*(
     for j in 0 ..< curr.iovlen.int:
       totalLen += iovArr[j].iov_len.int
 
+    if totalLen == 0:
+      continue
+
     var data = newSeqUninit[byte](totalLen)
     var currLen: int = 0
     for j in 0 ..< curr.iovlen.int:
@@ -55,13 +58,10 @@ proc sendPacketsOut*(
         continue
       copyMem(addr data[currLen], currIov.iov_base, currIov.iov_len)
       currLen += currIov.iov_len.int
-
+    
     let taddr = toTransportAddress(curr.dest_sa)
     let datagram = Datagram(data: data, ecn: curr.ecn, taddr: taddr)
-    try:
-      quicCtx.outgoing.putNoWait(datagram)
-      sent.inc
-    except AsyncQueueFullError:
-      discard # nothing to do
+    quicCtx.outgoing.put(datagram)
+    sent.inc
 
   sent.cint

lsquic/context/server.nim

@@ -41,7 +41,7 @@ const BBRv1 = 2
 proc new*(
     T: typedesc[ServerContext],
     tlsConfig: TLSConfig,
-    outgoing: AsyncQueue[Datagram],
+    outgoing: ManyQueue[Datagram],
     incoming: AsyncQueue[QuicConnection],
 ): Result[T, string] =
   var ctx = ServerContext()

lsquic/listener.nim

@@ -13,7 +13,7 @@ type Listener* = ref object of ConnectionManager
 proc newListener*(
     tlsConfig: TLSConfig, address: TransportAddress
 ): Result[Listener, string] =
-  let outgoing = newAsyncQueue[Datagram]()
+  let outgoing = ManyQueue[Datagram]()
   let incoming = newAsyncQueue[QuicConnection]()
   let quicContext = ?ServerContext.new(tlsConfig, outgoing, incoming)
   let listener = Listener(incoming: incoming)