avoid chunked encoding / data copies (#267)

* use a body writer to avoid copying data during request sends
* avoid chunked encoding in server - since the full response is being
held in memory, there's no advantage of chunking it on the server side
and we can use the simpler plain response

Author: arnetheduck

json_rpc/clients/httpclient.nim

@@ -57,15 +57,28 @@ method send(
     else:
       @[]
   headers.add(("Content-Type", "application/json"))
+  headers.add(("Content-Length", $reqData.len))
 
   let
     req = HttpClientRequestRef.post(
-      client.httpSession, client.httpAddress, body = reqData, headers = headers
+      client.httpSession,
+      client.httpAddress,
+      headers = headers,
+      body = default(seq[byte]),
     )
 
     res =
       try:
-        await req.send()
+        let wr = await req.open()
+        try:
+          await wr.write(unsafeAddr reqData[0], reqData.len)
+          await wr.finish()
+        finally:
+          await wr.closeWait()
+
+        await req.finish()
+      except AsyncStreamError as exc:
+        raise (ref RpcPostError)(msg: exc.msg, parent: exc)
       except HttpError as exc:
         raise (ref RpcPostError)(msg: exc.msg, parent: exc)
       finally:
@@ -92,15 +105,28 @@ method request(
     else:
       @[]
   headers.add(("Content-Type", "application/json"))
+  headers.add(("Content-Length", $reqData.len))
 
   let
     req = HttpClientRequestRef.post(
-      client.httpSession, client.httpAddress, body = reqData, headers = headers
+      client.httpSession,
+      client.httpAddress,
+      headers = headers,
+      body = default(seq[byte]),
     )
 
     res =
       try:
-        await req.send()
+        let wr = await req.open()
+        try:
+          await wr.write(unsafeAddr reqData[0], reqData.len)
+          await wr.finish()
+        finally:
+          await wr.closeWait()
+
+        await req.finish()
+      except AsyncStreamError as exc:
+        raise (ref RpcPostError)(msg: exc.msg, parent: exc)
       except HttpError as exc:
         raise (ref RpcPostError)(msg: exc.msg, parent: exc)
       finally:

json_rpc/router.nim

@@ -22,6 +22,13 @@ export chronos, jsonmarshal, json, jrpc_sys
 logScope:
   topics = "jsonrpc router"
 
+template errorReturnWorkaround(body) =
+  when NimMajor < 2:
+    body
+    raiseAssert "never hit"
+  else:
+    body
+
 type
   RpcProc* = proc(params: RequestParamsRx): Future[JsonString] {.async.}
     ## Procedure signature accepted as an RPC call by server - if the function
@@ -211,8 +218,8 @@ macro rpc*(server: RpcRouter, formatType, procList: untyped): untyped =
         of nnkIdent, nnkAccQuoted:
           $prc[0]
         else:
-          error "Unsupported rpc proc definition", prc
-          ""  # Nim 1.6
+          errorReturnWorkaround:
+            error "Unsupported rpc proc definition", prc
       result.add rpcImpl(server, path, formatType, prc)
     else:
       error "Only proc definitions are allowed within an rpc context", prc

json_rpc/servers/httpserver.nim

@@ -40,7 +40,7 @@ type
   # This inheritance arrangement is useful for
   # e.g. combo HTTP server
   RpcHttpHandler* = ref object of RpcServer
-    maxChunkSize*: int
+    maxChunkSize* {.deprecated.}: int
 
   RpcHttpServer* = ref object of RpcHttpHandler
     httpServers: seq[HttpServerRef]
@@ -56,26 +56,15 @@ proc serveHTTP*(rpcServer: RpcHttpHandler, request: HttpRequestRef):
 
     let
       data = await rpcServer.route(req)
-      chunkSize = rpcServer.maxChunkSize
-      streamType =
-        if data.len <= chunkSize:
-          HttpResponseStreamType.Plain
-        else:
-          HttpResponseStreamType.Chunked
       response = request.getResponse()
 
     response.addHeader("Content-Type", "application/json")
+    response.addHeader("Content-Length", $data.len)
 
-    await response.prepare(streamType)
-    let maxLen = data.len
+    await response.prepare(HttpResponseStreamType.Plain)
 
-    var len = data.len
-    while len > chunkSize:
-      await response.send(data[maxLen - len].unsafeAddr, chunkSize)
-      len -= chunkSize
-
-    if len > 0:
-      await response.send(data[maxLen - len].unsafeAddr, len)
+    if data.len() > 0:
+      await response.send(unsafeAddr data[0], data.len())
 
     await response.finish()
     response
@@ -221,10 +210,10 @@ proc addSecureHttpServer*(server: RpcHttpServer,
   addSecureHttpServers(server, toSeq(resolveIP(address, port)), tlsPrivateKey, tlsCertificate)
 
 proc new*(T: type RpcHttpServer, authHooks: seq[HttpAuthHook] = @[]): T =
-  T(router: RpcRouter.init(), httpServers: @[], authHooks: authHooks, maxChunkSize: 8192)
+  T(router: RpcRouter.init(), httpServers: @[], authHooks: authHooks)
 
 proc new*(T: type RpcHttpServer, router: RpcRouter, authHooks: seq[HttpAuthHook] = @[]): T =
-  T(router: router, httpServers: @[], authHooks: authHooks, maxChunkSize: 8192)
+  T(router: router, httpServers: @[], authHooks: authHooks)
 
 proc newRpcHttpServer*(authHooks: seq[HttpAuthHook] = @[]): RpcHttpServer =
   RpcHttpServer.new(authHooks)
@@ -273,5 +262,5 @@ proc localAddress*(server: RpcHttpServer): seq[TransportAddress] =
   for item in server.httpServers:
     result.add item.instance.localAddress()
 
-proc setMaxChunkSize*(server: RpcHttpServer, maxChunkSize: int) =
+proc setMaxChunkSize*(server: RpcHttpServer, maxChunkSize: int) {.deprecated: "unused".} =
   server.maxChunkSize = maxChunkSize

tests/testhttp.nim

@@ -8,9 +8,11 @@
 # those terms.
 
 import
-  unittest2, chronos/unittest2/asynctests,
+  unittest2,
+  chronos/unittest2/asynctests,
   ../json_rpc/[rpcserver, rpcclient, jsonmarshal],
-  ./private/helpers
+  ./private/helpers,
+  stew/byteutils
 
 const TestsCount = 100
 const bigChunkSize = 4 * 8192
@@ -19,14 +21,14 @@ suite "JSON-RPC/http":
   setup:
     var httpsrv = newRpcHttpServer(["127.0.0.1:0"])
     # Create RPC on server
-    httpsrv.rpc("myProc") do(input: string, data: array[0..3, int]):
+    httpsrv.rpc("myProc") do(input: string, data: array[0 .. 3, int]):
       result = %("Hello " & input & " data: " & $data)
     httpsrv.rpc("noParamsProc") do():
       result = %("Hello world")
 
     httpsrv.rpc("bigchunkMethod") do() -> seq[byte]:
       result = newSeq[byte](bigChunkSize)
-      for i in 0..<result.len:
+      for i in 0 ..< result.len:
         result[i] = byte(i mod 255)
 
     httpsrv.setMaxChunkSize(8192)
@@ -47,10 +49,11 @@ suite "JSON-RPC/http":
 
   asyncTest "Continuous RPC calls (" & $TestsCount & " messages)":
     var client = newRpcHttpClient()
-    for i in 0..<TestsCount:
+    for i in 0 ..< TestsCount:
       await client.connect("http://" & serverAddress)
       var r = await client.call("myProc", %[%"abc", %[1, 2, 3, i]])
-      check: r.string == "\"Hello abc data: [1, 2, 3, " & $i & "]\""
+      check:
+        r.string == "\"Hello abc data: [1, 2, 3, " & $i & "]\""
       await client.close()
 
   asyncTest "Invalid RPC calls":
@@ -88,3 +91,68 @@ suite "JSON-RPC/http":
 
     check:
       notif
+
+  asyncTest "Chunked encoding":
+    # The server doesn't use chunked encoding but the client might encounter
+    # such servers - test that it works as expected
+    let host = "127.0.0.1"
+    var server = createStreamServer(initTAddress(host & ":0"), {ReuseAddr})
+    defer:
+      await server.closeWait()
+
+    let port = server.localAddress().port
+
+    proc simpleServerResponder() {.async.} =
+      try:
+        let conn = await server.accept()
+
+        # 1. Consume the incoming headers but ignore the body hoping it's small
+        # enough
+        var headerBuf: seq[byte] = newSeq[byte](8192)
+        discard
+          await conn.readUntil(addr headerBuf[0], headerBuf.len, toBytes("\r\n\r\n"))
+
+        # 2. Send HTTP response with Chunked Transfer Encoding
+        var respHeader = (
+          "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n" &
+          "Transfer-Encoding: chunked\r\n\r\n"
+        ).toBytes()
+        discard await conn.write(respHeader)
+
+        # JSON Response broken into two chunks
+        let fullMsg = """{"jsonrpc":"2.0","result":{"s":"part1"},"id":1}"""
+        let mid = fullMsg.len div 2
+        let part1 = fullMsg[0 ..< mid]
+        let part2 = fullMsg[mid ..^ 1]
+
+        # Chunk 1: size in hex + \r\n
+        discard await conn.write(
+          (toHex(part1.len).strip(trailing = false, chars = {'0'}) & "\r\n").toBytes()
+        )
+        discard await conn.write(part1.toBytes())
+
+        # Chunk 2: size in hex + \r\n
+        discard await conn.write(
+          ("\r\n" & toHex(part2.len).strip(trailing = false, chars = {'0'}) & "\r\n").toBytes()
+        )
+        discard await conn.write(part2.toBytes())
+
+        # Final empty chunk to end the stream (0\r\n\r\n)
+        discard await conn.write("\r\n0\r\n\r\n".toBytes())
+        await conn.shutdownWait()
+        await conn.closeWait()
+      except CatchableError as exc:
+        raiseAssert exc.msg
+
+    var st = simpleServerResponder()
+
+    var client = newRpcHttpClient()
+
+    try:
+      await client.connect(host, port, false)
+
+      let res = await client.call("anyMethod", %*{"param": "value"})
+      check res == JsonString """{"s":"part1"}"""
+    finally:
+      await client.close()
+    await st.cancelAndWait()