io: add BuferedWriter

Unsurprisingly enough, calling `write` a million times is never going to
be very fast.

BufferedWriter basically does the same thing as serialize.swrite did,
but queues up writes in batches before sending them.

TODO: give sread a similar treatment

2 files changed, 53 insertions, 41 deletions

diff --git a/src/server/buffer.nim b/src/server/buffer.nim
index 96d9c5ff..7bdd46d0 100644
--- a/src/server/buffer.nim
+++ b/src/server/buffer.nim
@@ -27,6 +27,7 @@ import html/env
 import html/event
 import html/formdata as formdata_impl
 import io/bufstream
+import io/bufwriter
 import io/posixstream
 import io/promise
 import io/serialize
@@ -192,8 +193,9 @@ proc buildInterfaceProc(fun: NimNode, funid: string): tuple[fun, name: NimNode]
   let this2 = newIdentDefs(ident("iface"), ident("BufferInterface"))
   let thisval = this2[0]
   body.add(quote do:
-    `thisval`.stream.swrite(BufferCommand.`nup`)
-    `thisval`.stream.swrite(`thisval`.packetid)
+    var writer {.inject.} = `thisval`.stream.initWriter()
+    writer.swrite(BufferCommand.`nup`)
+    writer.swrite(`thisval`.packetid)
   )
   var params2: seq[NimNode]
   var retval2: NimNode
@@ -220,12 +222,14 @@ proc buildInterfaceProc(fun: NimNode, funid: string): tuple[fun, name: NimNode]
     let s = params2[i][0] # sym e.g. url
     body.add(quote do:
       when typeof(`s`) is FileHandle:
-        #TODO flush or something
+        writer.flush()
         SocketStream(`thisval`.stream.source).sendFileHandle(`s`)
       else:
-        `thisval`.stream.swrite(`s`)
+        writer.swrite(`s`)
     )
   body.add(quote do:
+    writer.flush()
+    writer.deinit()
     let promise = `addfun`
     inc `thisval`.packetid
     return promise
@@ -1099,11 +1103,11 @@ proc resolveTask[T](buffer: Buffer, cmd: BufferCommand, res: T) =
   if packetid == 0:
     return # no task to resolve (TODO this is kind of inefficient)
   let len = slen(buffer.tasks[cmd]) + slen(res)
-  buffer.pstream.swrite(len)
-  buffer.pstream.swrite(packetid)
+  buffer.pstream.withWriter w:
+    w.swrite(len)
+    w.swrite(packetid)
+    w.swrite(res)
   buffer.tasks[cmd] = 0
-  buffer.pstream.swrite(res)
-  buffer.pstream.flush()
 
 proc onload(buffer: Buffer) =
   case buffer.state
@@ -1664,7 +1668,8 @@ macro bufferDispatcher(funs: static ProxyMap, buffer: Buffer,
             let `id` = `buffer`.pstream.recvFileHandle()
           else:
             var `id`: `typ`
-            `buffer`.pstream.sread(`id`))
+            `buffer`.pstream.sread(`id`)
+        )
         call.add(id)
     var rval: NimNode
     if v.params[0].kind == nnkEmpty:
@@ -1677,15 +1682,19 @@ macro bufferDispatcher(funs: static ProxyMap, buffer: Buffer,
     if rval == nil:
       resolve.add(quote do:
         let len = slen(`packetid`)
-        buffer.pstream.swrite(len)
-        buffer.pstream.swrite(`packetid`)
+        block:
+          buffer.pstream.withWriter w:
+            w.swrite(len)
+            w.swrite(`packetid`)
       )
     else:
       resolve.add(quote do:
         let len = slen(`packetid`) + slen(`rval`)
-        buffer.pstream.swrite(len)
-        buffer.pstream.swrite(`packetid`)
-        buffer.pstream.swrite(`rval`)
+        block:
+          buffer.pstream.withWriter w:
+            w.swrite(len)
+            w.swrite(`packetid`)
+            w.swrite(`rval`)
       )
     if v.istask:
       let en = v.ename
@@ -1694,7 +1703,8 @@ macro bufferDispatcher(funs: static ProxyMap, buffer: Buffer,
           buffer.savetask = false
           buffer.tasks[BufferCommand.`en`] = `packetid`
         else:
-          `resolve`)
+          `resolve`
+      )
     else:
       stmts.add(resolve)
     ofbranch.add(stmts)
diff --git a/src/server/forkserver.nim b/src/server/forkserver.nim
index 2c00dd4e..12b25dc3 100644
--- a/src/server/forkserver.nim
+++ b/src/server/forkserver.nim
@@ -5,6 +5,7 @@ import std/streams
 import std/tables
 
 import config/config
+import io/bufwriter
 import io/posixstream
 import io/serialize
 import io/serversocket
@@ -23,44 +24,44 @@ type
 
   ForkServer* = ref object
     istream: Stream
-    ostream: Stream
+    ostream: PosixStream
     estream*: PosixStream
 
   ForkServerContext = object
-    istream: Stream
-    ostream: Stream
+    istream: PosixStream
+    ostream: PosixStream
     children: seq[int]
     loaderPid: int
 
 proc newFileLoader*(forkserver: ForkServer; config: LoaderConfig): FileLoader =
-  forkserver.ostream.swrite(fcForkLoader)
-  forkserver.ostream.swrite(config)
-  forkserver.ostream.flush()
+  forkserver.ostream.withWriter w:
+    w.swrite(fcForkLoader)
+    w.swrite(config)
   var process: int
   forkserver.istream.sread(process)
   return FileLoader(process: process, clientPid: getCurrentProcessId())
 
 proc loadForkServerConfig*(forkserver: ForkServer, config: Config) =
-  forkserver.ostream.swrite(fcLoadConfig)
-  forkserver.ostream.swrite(config.getForkServerConfig())
-  forkserver.ostream.flush()
+  forkserver.ostream.withWriter w:
+    w.swrite(fcLoadConfig)
+    w.swrite(config.getForkServerConfig())
 
 proc removeChild*(forkserver: ForkServer, pid: int) =
-  forkserver.ostream.swrite(fcRemoveChild)
-  forkserver.ostream.swrite(pid)
-  forkserver.ostream.flush()
+  forkserver.ostream.withWriter w:
+    w.swrite(fcRemoveChild)
+    w.swrite(pid)
 
 proc forkBuffer*(forkserver: ForkServer; config: BufferConfig; url: URL;
     request: Request; attrs: WindowAttributes; ishtml: bool;
     charsetStack: seq[Charset]): int =
-  forkserver.ostream.swrite(fcForkBuffer)
-  forkserver.ostream.swrite(config)
-  forkserver.ostream.swrite(url)
-  forkserver.ostream.swrite(request)
-  forkserver.ostream.swrite(attrs)
-  forkserver.ostream.swrite(ishtml)
-  forkserver.ostream.swrite(charsetStack)
-  forkserver.ostream.flush()
+  forkserver.ostream.withWriter w:
+    w.swrite(fcForkBuffer)
+    w.swrite(config)
+    w.swrite(url)
+    w.swrite(request)
+    w.swrite(attrs)
+    w.swrite(ishtml)
+    w.swrite(charsetStack)
   var bufferPid: int
   forkserver.istream.sread(bufferPid)
   bufferPid
@@ -186,13 +187,16 @@ proc runForkServer() =
         if i != -1:
           ctx.children.del(i)
       of fcForkBuffer:
-        ctx.ostream.swrite(ctx.forkBuffer())
+        let r = ctx.forkBuffer()
+        ctx.ostream.withWriter w:
+          w.swrite(r)
       of fcForkLoader:
         assert ctx.loaderPid == 0
         var config: LoaderConfig
         ctx.istream.sread(config)
         let pid = ctx.forkLoader(config)
-        ctx.ostream.swrite(pid)
+        ctx.ostream.withWriter w:
+          w.swrite(pid)
         ctx.loaderPid = pid
         ctx.children.add(pid)
       of fcLoadConfig:
@@ -246,15 +250,13 @@ proc newForkServer*(): ForkServer =
     discard close(pipefd_in[0]) # close read
     discard close(pipefd_out[1]) # close write
     discard close(pipefd_err[1]) # close write
-    var writef, readf: File
-    if not open(writef, pipefd_in[1], fmWrite):
-      raise newException(Defect, "Failed to open output handle")
+    var readf: File
     if not open(readf, pipefd_out[0], fmRead):
       raise newException(Defect, "Failed to open input handle")
     let estream = newPosixStream(pipefd_err[0])
     estream.setBlocking(false)
     return ForkServer(
-      ostream: newFileStream(writef),
+      ostream: newPosixStream(pipefd_in[1]),
       istream: newFileStream(readf),
       estream: estream
     )