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

3 files changed, 178 insertions, 106 deletions

diff --git a/src/io/bufwriter.nim b/src/io/bufwriter.nim
new file mode 100644
index 00000000..99c7ed94
--- /dev/null
+++ b/src/io/bufwriter.nim
@@ -0,0 +1,175 @@
+# Write data to streams.
+
+import std/options
+import std/sets
+import std/tables
+
+import io/dynstream
+
+import types/blob
+import types/formdata
+import types/url
+import types/opt
+
+type BufferedWriter* = object
+  stream: DynStream
+  buffer: ptr UncheckedArray[uint8]
+  bufSize: int
+  bufLen: int
+
+{.warning[Deprecated]: off.}:
+  proc `=destroy`(writer: var BufferedWriter) =
+    if writer.buffer != nil:
+      dealloc(writer.buffer)
+      writer.buffer = nil
+
+proc initWriter*(stream: DynStream; sizeInit = 64): BufferedWriter =
+  return BufferedWriter(
+    stream: stream,
+    buffer: cast[ptr UncheckedArray[uint8]](alloc(sizeInit)),
+    bufSize: sizeInit,
+    bufLen: 0
+  )
+
+proc flush*(writer: var BufferedWriter) =
+  let stream = writer.stream
+  var n = 0
+  while true:
+    n += stream.sendData(addr writer.buffer[n], writer.bufLen - n)
+    if n == writer.bufLen:
+      break
+  writer.bufLen = 0
+  stream.sflush()
+
+proc deinit*(writer: var BufferedWriter) =
+  dealloc(writer.buffer)
+  writer.buffer = nil
+  writer.bufSize = 0
+  writer.bufLen = 0
+
+template withWriter*(stream: DynStream; w, body: untyped) =
+  var w {.inject.} = stream.initWriter()
+  body
+  w.flush()
+  w.deinit()
+
+proc swrite*(writer: var BufferedWriter; n: SomeNumber)
+proc swrite*[T](writer: var BufferedWriter; s: set[T])
+proc swrite*[T: enum](writer: var BufferedWriter; x: T)
+proc swrite*(writer: var BufferedWriter; s: string)
+proc swrite*(writer: var BufferedWriter; b: bool)
+proc swrite*(writer: var BufferedWriter; url: URL)
+proc swrite*(writer: var BufferedWriter; tup: tuple)
+proc swrite*[I, T](writer: var BufferedWriter; a: array[I, T])
+proc swrite*(writer: var BufferedWriter; s: seq)
+proc swrite*[U, V](writer: var BufferedWriter; t: Table[U, V])
+proc swrite*(writer: var BufferedWriter; obj: object)
+proc swrite*(writer: var BufferedWriter; obj: ref object)
+proc swrite*(writer: var BufferedWriter; part: FormDataEntry)
+proc swrite*(writer: var BufferedWriter; blob: Blob)
+proc swrite*[T](writer: var BufferedWriter; o: Option[T])
+proc swrite*[T, E](writer: var BufferedWriter; o: Result[T, E])
+
+proc writeData(writer: var BufferedWriter; buffer: pointer; len: int) =
+  let targetLen = writer.bufLen + len
+  let missing = targetLen - writer.bufSize
+  if missing > 0:
+    let target = writer.bufSize + missing
+    writer.bufSize *= 2
+    if writer.bufSize < target:
+      writer.bufSize = target
+    let p = realloc(writer.buffer, writer.bufSize)
+    writer.buffer = cast[ptr UncheckedArray[uint8]](p)
+  copyMem(addr writer.buffer[writer.bufLen], buffer, len)
+  writer.bufLen = targetLen
+
+proc swrite*(writer: var BufferedWriter; n: SomeNumber) =
+  writer.writeData(unsafeAddr n, sizeof(n))
+
+proc swrite*[T: enum](writer: var BufferedWriter; x: T) =
+  static:
+    doAssert sizeof(int) >= sizeof(T)
+  writer.swrite(int(x))
+
+proc swrite*[T](writer: var BufferedWriter; s: set[T]) =
+  writer.swrite(s.card)
+  for e in s:
+    writer.swrite(e)
+
+proc swrite*(writer: var BufferedWriter; s: string) =
+  writer.swrite(s.len)
+  if s.len > 0:
+    writer.writeData(unsafeAddr s[0], s.len)
+
+proc swrite*(writer: var BufferedWriter; b: bool) =
+  if b:
+    writer.swrite(1u8)
+  else:
+    writer.swrite(0u8)
+
+proc swrite*(writer: var BufferedWriter; url: URL) =
+  if url != nil:
+    writer.swrite(url.serialize())
+  else:
+    writer.swrite("")
+
+proc swrite*(writer: var BufferedWriter; tup: tuple) =
+  for f in tup.fields:
+    writer.swrite(f)
+
+proc swrite*[I, T](writer: var BufferedWriter; a: array[I, T]) =
+  for x in a:
+    writer.swrite(x)
+
+proc swrite*(writer: var BufferedWriter; s: seq) =
+  writer.swrite(s.len)
+  for x in s:
+    writer.swrite(x)
+
+proc swrite*[U, V](writer: var BufferedWriter; t: Table[U, V]) =
+  writer.swrite(t.len)
+  for k, v in t:
+    writer.swrite(k)
+    writer.swrite(v)
+
+proc swrite*(writer: var BufferedWriter; obj: object) =
+  for f in obj.fields:
+    writer.swrite(f)
+
+proc swrite*(writer: var BufferedWriter; obj: ref object) =
+  writer.swrite(obj != nil)
+  if obj != nil:
+    writer.swrite(obj[])
+
+proc swrite*(writer: var BufferedWriter; part: FormDataEntry) =
+  writer.swrite(part.isstr)
+  writer.swrite(part.name)
+  writer.swrite(part.filename)
+  if part.isstr:
+    writer.swrite(part.svalue)
+  else:
+    writer.swrite(part.value)
+
+#TODO clean up this mess
+proc swrite*(writer: var BufferedWriter; blob: Blob) =
+  writer.swrite(blob.isfile)
+  if blob.isfile:
+    writer.swrite(WebFile(blob).path)
+  else:
+    writer.swrite(blob.ctype)
+    writer.swrite(blob.size)
+    writer.writeData(blob.buffer, int(blob.size))
+
+proc swrite*[T](writer: var BufferedWriter; o: Option[T]) =
+  writer.swrite(o.isSome)
+  if o.isSome:
+    writer.swrite(o.get)
+
+proc swrite*[T, E](writer: var BufferedWriter; o: Result[T, E]) =
+  writer.swrite(o.isOk)
+  if o.isOk:
+    when not (T is void):
+      writer.swrite(o.get)
+  else:
+    when not (E is void):
+      writer.swrite(o.error)
diff --git a/src/io/dynstream.nim b/src/io/dynstream.nim
index d4c7760f..ec38c595 100644
--- a/src/io/dynstream.nim
+++ b/src/io/dynstream.nim
@@ -24,6 +24,9 @@ method seek*(s: DynStream; off: int) {.base.} =
 method sclose*(s: DynStream) {.base.} =
   assert false
 
+method sflush*(s: DynStream) {.base.} =
+  discard
+
 proc recvData*(s: DynStream; buffer: var openArray[uint8]): int {.inline.} =
   return s.recvData(addr buffer[0], buffer.len)
 
diff --git a/src/io/serialize.nim b/src/io/serialize.nim
index 4dcb79f0..3ffa40ac 100644
--- a/src/io/serialize.nim
+++ b/src/io/serialize.nim
@@ -10,73 +10,54 @@ import types/formdata
 import types/url
 import types/opt
 
-proc swrite*(stream: Stream, n: SomeNumber)
 proc sread*(stream: Stream, n: var SomeNumber)
 func slen*(n: SomeNumber): int
 
-proc swrite*[T](stream: Stream, s: set[T])
 proc sread*[T](stream: Stream, s: var set[T])
 func slen*[T](s: set[T]): int
 
-proc swrite*[T: enum](stream: Stream, x: T)
 proc sread*[T: enum](stream: Stream, x: var T)
 func slen*[T: enum](x: T): int
 
-proc swrite*(stream: Stream, s: string)
 proc sread*(stream: Stream, s: var string)
 func slen*(s: string): int
 
-proc swrite*(stream: Stream, b: bool)
 proc sread*(stream: Stream, b: var bool)
 func slen*(b: bool): int
 
-proc swrite*(stream: Stream, url: URL)
 proc sread*(stream: Stream, url: var URL)
 func slen*(url: URL): int
 
-proc swrite*(stream: Stream, tup: tuple)
 proc sread*(stream: Stream, tup: var tuple)
 func slen*(tup: tuple): int
 
-proc swrite*[I, T](stream: Stream, a: array[I, T])
 proc sread*[I, T](stream: Stream, a: var array[I, T])
 func slen*[I, T](a: array[I, T]): int
 
-proc swrite*(stream: Stream, s: seq)
 proc sread*(stream: Stream, s: var seq)
 func slen*(s: seq): int
 
-proc swrite*[U, V](stream: Stream, t: Table[U, V])
 proc sread*[U, V](stream: Stream, t: var Table[U, V])
 func slen*[U, V](t: Table[U, V]): int
 
-proc swrite*(stream: Stream, obj: object)
 proc sread*(stream: Stream, obj: var object)
 func slen*(obj: object): int
 
-proc swrite*(stream: Stream, obj: ref object)
 proc sread*(stream: Stream, obj: var ref object)
 func slen*(obj: ref object): int
 
-proc swrite*(stream: Stream, part: FormDataEntry)
 proc sread*(stream: Stream, part: var FormDataEntry)
 func slen*(part: FormDataEntry): int
 
-proc swrite*(stream: Stream, blob: Blob)
 proc sread*(stream: Stream, blob: var Blob)
 func slen*(blob: Blob): int
 
-proc swrite*[T](stream: Stream, o: Option[T])
 proc sread*[T](stream: Stream, o: var Option[T])
 func slen*[T](o: Option[T]): int
 
-proc swrite*[T, E](stream: Stream, o: Result[T, E])
 proc sread*[T, E](stream: Stream, o: var Result[T, E])
 func slen*[T, E](o: Result[T, E]): int
 
-proc swrite*(stream: Stream, n: SomeNumber) =
-  stream.write(n)
-
 proc sread*(stream: Stream, n: var SomeNumber) =
   if stream.readData(addr n, sizeof(n)) < sizeof(n):
     raise newException(EOFError, "eof")
@@ -84,11 +65,6 @@ proc sread*(stream: Stream, n: var SomeNumber) =
 func slen*(n: SomeNumber): int =
   return sizeof(n)
 
-proc swrite*[T: enum](stream: Stream, x: T) =
-  static:
-    doAssert sizeof(int) >= sizeof(T)
-  stream.swrite(int(x))
-
 proc sread*[T: enum](stream: Stream, x: var T) =
   var i: int
   stream.sread(i)
@@ -97,11 +73,6 @@ proc sread*[T: enum](stream: Stream, x: var T) =
 func slen*[T: enum](x: T): int =
   return sizeof(int)
 
-proc swrite*[T](stream: Stream, s: set[T]) =
-  stream.swrite(s.card)
-  for e in s:
-    stream.swrite(e)
-
 proc sread*[T](stream: Stream, s: var set[T]) =
   var len: int
   stream.sread(len)
@@ -115,10 +86,6 @@ func slen*[T](s: set[T]): int =
   for x in s:
     result += slen(x)
 
-proc swrite*(stream: Stream, s: string) =
-  stream.swrite(s.len)
-  stream.write(s)
-
 proc sread*(stream: Stream, s: var string) =
   var len: int
   stream.sread(len)
@@ -133,12 +100,6 @@ proc sread*(stream: Stream, s: var string) =
 func slen*(s: string): int =
   slen(s.len) + s.len
 
-proc swrite*(stream: Stream, b: bool) =
-  if b:
-    stream.swrite(1u8)
-  else:
-    stream.swrite(0u8)
-
 proc sread*(stream: Stream, b: var bool) =
   var n: uint8
   stream.sread(n)
@@ -151,12 +112,6 @@ proc sread*(stream: Stream, b: var bool) =
 func slen*(b: bool): int =
   return sizeof(uint8)
 
-proc swrite*(stream: Stream, url: URL) =
-  if url != nil:
-    stream.swrite(url.serialize())
-  else:
-    stream.swrite("")
-
 proc sread*(stream: Stream, url: var URL) =
   var s: string
   stream.sread(s)
@@ -174,10 +129,6 @@ func slen*(url: URL): int =
     return slen("")
   return slen(url.serialize())
 
-proc swrite*(stream: Stream, tup: tuple) =
-  for f in tup.fields:
-    stream.swrite(f)
-
 proc sread*(stream: Stream, tup: var tuple) =
   for f in tup.fields:
     stream.sread(f)
@@ -186,10 +137,6 @@ func slen*(tup: tuple): int =
   for f in tup.fields:
     result += slen(f)
 
-proc swrite*[I, T](stream: Stream; a: array[I, T]) =
-  for x in a:
-    stream.swrite(x)
-
 proc sread*[I, T](stream: Stream; a: var array[I, T]) =
   for x in a.mitems:
     stream.sread(x)
@@ -198,11 +145,6 @@ func slen*[I, T](a: array[I, T]): int =
   for x in a:
     result += slen(x)
 
-proc swrite*(stream: Stream, s: seq) =
-  stream.swrite(s.len)
-  for x in s:
-    stream.swrite(x)
-
 proc sread*(stream: Stream, s: var seq) =
   var len: int
   stream.sread(len)
@@ -215,12 +157,6 @@ func slen*(s: seq): int =
   for x in s:
     result += slen(x)
 
-proc swrite*[U, V](stream: Stream, t: Table[U, V]) =
-  stream.swrite(t.len)
-  for k, v in t:
-    stream.swrite(k)
-    stream.swrite(v)
-
 proc sread*[U, V](stream: Stream, t: var Table[U, V]) =
   var len: int
   stream.sread(len)
@@ -237,10 +173,6 @@ func slen*[U, V](t: Table[U, V]): int =
     result += slen(k)
     result += slen(v)
 
-proc swrite*(stream: Stream, obj: object) =
-  for f in obj.fields:
-    stream.swrite(f)
-
 proc sread*(stream: Stream, obj: var object) =
   for f in obj.fields:
     stream.sread(f)
@@ -249,11 +181,6 @@ func slen*(obj: object): int =
   for f in obj.fields:
     result += slen(f)
 
-proc swrite*(stream: Stream, obj: ref object) =
-  stream.swrite(obj != nil)
-  if obj != nil:
-    stream.swrite(obj[])
-
 proc sread*(stream: Stream, obj: var ref object) =
   var n: bool
   stream.sread(n)
@@ -266,15 +193,6 @@ func slen*(obj: ref object): int =
   if obj != nil:
     result += slen(obj[])
 
-proc swrite*(stream: Stream, part: FormDataEntry) =
-  stream.swrite(part.isstr)
-  stream.swrite(part.name)
-  stream.swrite(part.filename)
-  if part.isstr:
-    stream.swrite(part.svalue)
-  else:
-    stream.swrite(part.value)
-
 proc sread*(stream: Stream, part: var FormDataEntry) =
   var isstr: bool
   stream.sread(isstr)
@@ -298,16 +216,6 @@ func slen*(part: FormDataEntry): int =
   else:
     result += slen(part.value)
 
-#TODO clean up this mess
-proc swrite*(stream: Stream, blob: Blob) =
-  stream.swrite(blob.isfile)
-  if blob.isfile:
-    stream.swrite(WebFile(blob).path)
-  else:
-    stream.swrite(blob.ctype)
-    stream.swrite(blob.size)
-    stream.writeData(blob.buffer, int(blob.size))
-
 proc sread*(stream: Stream, blob: var Blob) =
   var isfile: bool
   stream.sread(isfile)
@@ -336,11 +244,6 @@ func slen*(blob: Blob): int =
     result += slen(blob.size)
     result += int(blob.size) #TODO ??
 
-proc swrite*[T](stream: Stream, o: Option[T]) =
-  stream.swrite(o.isSome)
-  if o.isSome:
-    stream.swrite(o.get)
-
 proc sread*[T](stream: Stream, o: var Option[T]) =
   var x: bool
   stream.sread(x)
@@ -356,15 +259,6 @@ func slen*[T](o: Option[T]): int =
   if o.isSome:
     result += slen(o.get)
 
-proc swrite*[T, E](stream: Stream, o: Result[T, E]) =
-  stream.swrite(o.isOk)
-  if o.isOk:
-    when not (T is void):
-      stream.swrite(o.get)
-  else:
-    when not (E is void):
-      stream.swrite(o.error)
-
 proc sread*[T, E](stream: Stream, o: var Result[T, E]) =
   var x: bool
   stream.sread(x)