Merge branch 'def--clean-speedup-2' into devel

lib/pure/asyncdispatch.nim

@@ -145,6 +145,8 @@ type
   Future*[T] = ref object of FutureBase ## Typed future.
     value: T ## Stored value
 
+  FutureVar*[T] = distinct Future[T]
+
 {.deprecated: [PFutureBase: FutureBase, PFuture: Future].}
 
 
@@ -162,6 +164,19 @@ proc newFuture*[T](fromProc: string = "unspecified"): Future[T] =
     result.fromProc = fromProc
     currentID.inc()
 
+proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] =
+  ## Create a new ``FutureVar``. This Future type is ideally suited for
+  ## situations where you want to avoid unnecessary allocations of Futures.
+  ##
+  ## Specifying ``fromProc``, which is a string specifying the name of the proc
+  ## that this future belongs to, is a good habit as it helps with debugging.
+  result = FutureVar[T](newFuture[T](fromProc))
+
+proc clean*[T](future: FutureVar[T]) =
+  ## Resets the ``finished`` status of ``future``.
+  Future[T](future).finished = false
+  Future[T](future).error = nil
+
 proc checkFinished[T](future: Future[T]) =
   when not defined(release):
     if future.finished:
@@ -194,6 +209,15 @@ proc complete*(future: Future[void]) =
   if future.cb != nil:
     future.cb()
 
+proc complete*[T](future: FutureVar[T]) =
+  ## Completes a ``FutureVar``.
+  template fut: expr = Future[T](future)
+  checkFinished(fut)
+  assert(fut.error == nil)
+  fut.finished = true
+  if fut.cb != nil:
+    fut.cb()
+
 proc fail*[T](future: Future[T], error: ref Exception) =
   ## Completes ``future`` with ``error``.
   #assert(not future.finished, "Future already finished, cannot finish twice.")
@@ -264,6 +288,13 @@ proc readError*[T](future: Future[T]): ref Exception =
   else:
     raise newException(ValueError, "No error in future.")
 
+proc mget*[T](future: FutureVar[T]): var T =
+  ## Returns a mutable value stored in ``future``.
+  ##
+  ## Unlike ``read``, this function will not raise an exception if the
+  ## Future has not been finished.
+  result = Future[T](future).value
+
 proc finished*[T](future: Future[T]): bool =
   ## Determines whether ``future`` has completed.
   ##
@@ -634,6 +665,93 @@ when defined(windows) or defined(nimdoc):
       # free ``ol``.
     return retFuture
 
+  proc recvInto*(socket: TAsyncFD, buf: cstring, size: int,
+                flags = {SocketFlag.SafeDisconn}): Future[int] =
+    ## Reads **up to** ``size`` bytes from ``socket`` into ``buf``, which must
+    ## at least be of that size. Returned future will complete once all the
+    ## data requested is read, a part of the data has been read, or the socket
+    ## has disconnected in which case the future will complete with a value of
+    ## ``0``.
+    ##
+    ## **Warning**: The ``Peek`` socket flag is not supported on Windows.
+
+
+    # Things to note:
+    #   * When WSARecv completes immediately then ``bytesReceived`` is very
+    #     unreliable.
+    #   * Still need to implement message-oriented socket disconnection,
+    #     '\0' in the message currently signifies a socket disconnect. Who
+    #     knows what will happen when someone sends that to our socket.
+    verifyPresence(socket)
+    assert SocketFlag.Peek notin flags, "Peek not supported on Windows."
+
+    var retFuture = newFuture[int]("recvInto")
+
+    #buf[] = '\0'
+    var dataBuf: TWSABuf
+    dataBuf.buf = buf
+    dataBuf.len = size
+
+    var bytesReceived: Dword
+    var flagsio = flags.toOSFlags().Dword
+    var ol = PCustomOverlapped()
+    GC_ref(ol)
+    ol.data = TCompletionData(fd: socket, cb:
+      proc (fd: TAsyncFD, bytesCount: Dword, errcode: OSErrorCode) =
+        if not retFuture.finished:
+          if errcode == OSErrorCode(-1):
+            if bytesCount == 0 and dataBuf.buf[0] == '\0':
+              retFuture.complete(0)
+            else:
+              retFuture.complete(bytesCount)
+          else:
+            if flags.isDisconnectionError(errcode):
+              retFuture.complete(0)
+            else:
+              retFuture.fail(newException(OSError, osErrorMsg(errcode)))
+        if dataBuf.buf != nil:
+          dataBuf.buf = nil
+    )
+
+    let ret = WSARecv(socket.SocketHandle, addr dataBuf, 1, addr bytesReceived,
+                      addr flagsio, cast[POVERLAPPED](ol), nil)
+    if ret == -1:
+      let err = osLastError()
+      if err.int32 != ERROR_IO_PENDING:
+        if dataBuf.buf != nil:
+          dataBuf.buf = nil
+        GC_unref(ol)
+        if flags.isDisconnectionError(err):
+          retFuture.complete(0)
+        else:
+          retFuture.fail(newException(OSError, osErrorMsg(err)))
+    elif ret == 0 and bytesReceived == 0 and dataBuf.buf[0] == '\0':
+      # We have to ensure that the buffer is empty because WSARecv will tell
+      # us immediately when it was disconnected, even when there is still
+      # data in the buffer.
+      # We want to give the user as much data as we can. So we only return
+      # the empty string (which signals a disconnection) when there is
+      # nothing left to read.
+      retFuture.complete(0)
+      # TODO: "For message-oriented sockets, where a zero byte message is often
+      # allowable, a failure with an error code of WSAEDISCON is used to
+      # indicate graceful closure."
+      # ~ http://msdn.microsoft.com/en-us/library/ms741688%28v=vs.85%29.aspx
+    else:
+      # Request to read completed immediately.
+      # From my tests bytesReceived isn't reliable.
+      let realSize =
+        if bytesReceived == 0:
+          size
+        else:
+          bytesReceived
+      assert realSize <= size
+      retFuture.complete(realSize)
+      # We don't deallocate ``ol`` here because even though this completed
+      # immediately poll will still be notified about its completion and it will
+      # free ``ol``.
+    return retFuture
+
   proc send*(socket: TAsyncFD, data: string,
              flags = {SocketFlag.SafeDisconn}): Future[void] =
     ## Sends ``data`` to ``socket``. The returned future will complete once all
@@ -983,6 +1101,30 @@ else:
     addRead(socket, cb)
     return retFuture
 
+  proc recvInto*(socket: TAsyncFD, buf: cstring, size: int,
+                  flags = {SocketFlag.SafeDisconn}): Future[int] =
+    var retFuture = newFuture[int]("recvInto")
+
+    proc cb(sock: TAsyncFD): bool =
+      result = true
+      let res = recv(sock.SocketHandle, buf, size.cint,
+                     flags.toOSFlags())
+      if res < 0:
+        let lastError = osLastError()
+        if lastError.int32 notin {EINTR, EWOULDBLOCK, EAGAIN}:
+          if flags.isDisconnectionError(lastError):
+            retFuture.complete(0)
+          else:
+            retFuture.fail(newException(OSError, osErrorMsg(lastError)))
+        else:
+          result = false # We still want this callback to be called.
+      else:
+        retFuture.complete(res)
+    # TODO: The following causes a massive slowdown.
+    #if not cb(socket):
+    addRead(socket, cb)
+    return retFuture
+
   proc send*(socket: TAsyncFD, data: string,
              flags = {SocketFlag.SafeDisconn}): Future[void] =
     var retFuture = newFuture[void]("send")

lib/pure/asynchttpserver.nim

@@ -23,8 +23,7 @@
 ##    proc cb(req: Request) {.async.} =
 ##      await req.respond(Http200, "Hello World")
 ##
-##    asyncCheck server.serve(Port(8080), cb)
+##    waitFor server.serve(Port(8080), cb)
-##    runForever()
 
 import strtabs, asyncnet, asyncdispatch, parseutils, uri, strutils
 type
@@ -109,22 +108,19 @@ proc sendHeaders*(req: Request, headers: StringTableRef): Future[void] =
   addHeaders(msg, headers)
   return req.client.send(msg)
 
-proc respond*(req: Request, code: HttpCode,
+proc respond*(req: Request, code: HttpCode, content: string,
-        content: string, headers = newStringTable()) {.async.} =
+              headers: StringTableRef = nil): Future[void] =
   ## Responds to the request with the specified ``HttpCode``, headers and
   ## content.
   ##
   ## This procedure will **not** close the client socket.
-  var customHeaders = headers
-  customHeaders["Content-Length"] = $content.len
   var msg = "HTTP/1.1 " & $code & "\c\L"
-  msg.addHeaders(customHeaders)
-  await req.client.send(msg & "\c\L" & content)
 
-proc newRequest(): Request =
+  if headers != nil:
-  result.headers = newStringTable(modeCaseInsensitive)
+    msg.addHeaders(headers)
-  result.hostname = ""
+  msg.add("Content-Length: " & $content.len & "\c\L\c\L")
-  result.body = ""
+  msg.add(content)
+  result = req.client.send(msg)
 
 proc parseHeader(line: string): tuple[key, value: string] =
   var i = 0
@@ -149,52 +145,61 @@ proc sendStatus(client: AsyncSocket, status: string): Future[void] =
 proc processClient(client: AsyncSocket, address: string,
                    callback: proc (request: Request):
                       Future[void] {.closure, gcsafe.}) {.async.} =
+  var request: Request
+  request.url = initUri()
+  request.headers = newStringTable(modeCaseInsensitive)
+  var lineFut = newFutureVar[string]("asynchttpserver.processClient")
+  lineFut.mget() = newStringOfCap(80)
+  var key, value = ""
+
   while not client.isClosed:
     # GET /path HTTP/1.1
     # Header: val
     # \n
-    var request = newRequest()
+    request.headers.clear(modeCaseInsensitive)
-    request.hostname = address
+    request.hostname.shallowCopy(address)
     assert client != nil
     request.client = client
 
     # First line - GET /path HTTP/1.1
-    let line = await client.recvLine() # TODO: Timeouts.
+    lineFut.mget().setLen(0)
-    if line == "":
+    lineFut.clean()
+    await client.recvLineInto(lineFut) # TODO: Timeouts.
+    if lineFut.mget == "":
       client.close()
       return
-    let lineParts = line.split(' ')
-    if lineParts.len != 3:
-      await request.respond(Http400, "Invalid request. Got: " & line)
-      continue
 
-    let reqMethod = lineParts[0]
+    var i = 0
-    let path = lineParts[1]
+    for linePart in lineFut.mget.split(' '):
-    let protocol = lineParts[2]
+      case i
+      of 0: request.reqMethod.shallowCopy(linePart.normalize)
+      of 1: parseUri(linePart, request.url)
+      of 2:
+        try:
+          request.protocol = parseProtocol(linePart)
+        except ValueError:
+          asyncCheck request.respond(Http400,
+            "Invalid request protocol. Got: " & linePart)
+          continue
+      else:
+        await request.respond(Http400, "Invalid request. Got: " & lineFut.mget)
+        continue
+      inc i
 
     # Headers
-    var i = 0
     while true:
       i = 0
-      let headerLine = await client.recvLine()
+      lineFut.mget.setLen(0)
-      if headerLine == "":
+      lineFut.clean()
+      await client.recvLineInto(lineFut)
+
+      if lineFut.mget == "":
         client.close(); return
-      if headerLine == "\c\L": break
+      if lineFut.mget == "\c\L": break
-      # TODO: Compiler crash
+      let (key, value) = parseHeader(lineFut.mget)
-      #let (key, value) = parseHeader(headerLine)
+      request.headers[key] = value
-      let kv = parseHeader(headerLine)
-      request.headers[kv.key] = kv.value
 
-    request.reqMethod = reqMethod
+    if request.reqMethod == "post":
-    request.url = parseUri(path)
-    try:
-      request.protocol = protocol.parseProtocol()
-    except ValueError:
-      asyncCheck request.respond(Http400, "Invalid request protocol. Got: " &
-          protocol)
-      continue
-
-    if reqMethod.normalize == "post":
       # Check for Expect header
       if request.headers.hasKey("Expect"):
         if request.headers["Expect"].toLower == "100-continue":
@@ -215,11 +220,11 @@ proc processClient(client: AsyncSocket, address: string,
         await request.respond(Http400, "Bad Request. No Content-Length.")
         continue
 
-    case reqMethod.normalize
+    case request.reqMethod
     of "get", "post", "head", "put", "delete", "trace", "options", "connect", "patch":
       await callback(request)
     else:
-      await request.respond(Http400, "Invalid request method. Got: " & reqMethod)
+      await request.respond(Http400, "Invalid request method. Got: " & request.reqMethod)
 
     # Persistent connections
     if (request.protocol == HttpVer11 and

lib/pure/asyncnet.nim

@@ -182,26 +182,30 @@ proc connect*(socket: AsyncSocket, address: string, port: Port,
       sslSetConnectState(socket.sslHandle)
       sslLoop(socket, flags, sslDoHandshake(socket.sslHandle))
 
-proc readInto(buf: cstring, size: int, socket: AsyncSocket,
+template readInto(buf: cstring, size: int, socket: AsyncSocket,
-              flags: set[SocketFlag]): Future[int] {.async.} =
+                  flags: set[SocketFlag]): int =
+  ## Reads **up to** ``size`` bytes from ``socket`` into ``buf``. Note that
+  ## this is a template and not a proc.
+  var res = 0
   if socket.isSsl:
     when defined(ssl):
       # SSL mode.
       sslLoop(socket, flags,
         sslRead(socket.sslHandle, buf, size.cint))
-      result = opResult
+      res = opResult
   else:
-    var data = await recv(socket.fd.TAsyncFD, size, flags)
+    var recvIntoFut = recvInto(socket.fd.TAsyncFD, buf, size, flags)
-    if data.len != 0:
+    yield recvIntoFut
-      copyMem(buf, addr data[0], data.len)
     # Not in SSL mode.
-    result = data.len
+    res = recvIntoFut.read()
+  res
 
-proc readIntoBuf(socket: AsyncSocket,
+template readIntoBuf(socket: AsyncSocket,
-    flags: set[SocketFlag]): Future[int] {.async.} =
+    flags: set[SocketFlag]): int =
-  result = await readInto(addr socket.buffer[0], BufferSize, socket, flags)
+  var size = readInto(addr socket.buffer[0], BufferSize, socket, flags)
   socket.currPos = 0
-  socket.bufLen = result
+  socket.bufLen = size
+  size
 
 proc recv*(socket: AsyncSocket, size: int,
            flags = {SocketFlag.SafeDisconn}): Future[string] {.async.} =
@@ -222,10 +226,11 @@ proc recv*(socket: AsyncSocket, size: int,
   ## to be read then the future will complete with a value of ``""``.
   if socket.isBuffered:
     result = newString(size)
+    shallow(result)
     let originalBufPos = socket.currPos
 
     if socket.bufLen == 0:
-      let res = await socket.readIntoBuf(flags - {SocketFlag.Peek})
+      let res = socket.readIntoBuf(flags - {SocketFlag.Peek})
       if res == 0:
         result.setLen(0)
         return
@@ -236,7 +241,7 @@ proc recv*(socket: AsyncSocket, size: int,
         if SocketFlag.Peek in flags:
           # We don't want to get another buffer if we're peeking.
           break
-        let res = await socket.readIntoBuf(flags - {SocketFlag.Peek})
+        let res = socket.readIntoBuf(flags - {SocketFlag.Peek})
         if res == 0:
           break
 
@@ -251,7 +256,7 @@ proc recv*(socket: AsyncSocket, size: int,
     result.setLen(read)
   else:
     result = newString(size)
-    let read = await readInto(addr result[0], size, socket, flags)
+    let read = readInto(addr result[0], size, socket, flags)
     result.setLen(read)
 
 proc send*(socket: AsyncSocket, data: string,
@@ -302,6 +307,97 @@ proc accept*(socket: AsyncSocket,
         retFut.complete(future.read.client)
   return retFut
 
+proc recvLineInto*(socket: AsyncSocket, resString: FutureVar[string],
+    flags = {SocketFlag.SafeDisconn}) {.async.} =
+  ## Reads a line of data from ``socket`` into ``resString``.
+  ##
+  ## The ``resString`` future and the string value contained within must both
+  ## be initialised.
+  ##
+  ## If a full line is read ``\r\L`` is not
+  ## added to ``line``, however if solely ``\r\L`` is read then ``line``
+  ## will be set to it.
+  ##
+  ## If the socket is disconnected, ``line`` will be set to ``""``.
+  ##
+  ## If the socket is disconnected in the middle of a line (before ``\r\L``
+  ## is read) then line will be set to ``""``.
+  ## The partial line **will be lost**.
+  ##
+  ## **Warning**: The ``Peek`` flag is not yet implemented.
+  ##
+  ## **Warning**: ``recvLineInto`` on unbuffered sockets assumes that the
+  ## protocol uses ``\r\L`` to delimit a new line.
+  assert SocketFlag.Peek notin flags ## TODO:
+  assert(not resString.mget.isNil(),
+      "String inside resString future needs to be initialised")
+  result = newFuture[void]("asyncnet.recvLineInto")
+
+  # TODO: Make the async transformation check for FutureVar params and complete
+  # them when the result future is completed.
+  # Can we replace the result future with the FutureVar?
+
+  template addNLIfEmpty(): stmt =
+    if resString.mget.len == 0:
+      resString.mget.add("\c\L")
+
+  if socket.isBuffered:
+    if socket.bufLen == 0:
+      let res = socket.readIntoBuf(flags)
+      if res == 0:
+        resString.complete()
+        return
+
+    var lastR = false
+    while true:
+      if socket.currPos >= socket.bufLen:
+        let res = socket.readIntoBuf(flags)
+        if res == 0:
+          resString.mget().setLen(0)
+          resString.complete()
+          return
+
+      case socket.buffer[socket.currPos]
+      of '\r':
+        lastR = true
+        addNLIfEmpty()
+      of '\L':
+        addNLIfEmpty()
+        socket.currPos.inc()
+        resString.complete()
+        return
+      else:
+        if lastR:
+          socket.currPos.inc()
+          resString.complete()
+          return
+        else:
+          resString.mget.add socket.buffer[socket.currPos]
+      socket.currPos.inc()
+  else:
+    var c = ""
+    while true:
+      let recvFut = recv(socket, 1, flags)
+      c = recvFut.read()
+      if c.len == 0:
+        resString.mget.setLen(0)
+        resString.complete()
+        return
+      if c == "\r":
+        let recvFut = recv(socket, 1, flags) # Skip \L
+        c = recvFut.read()
+        assert c == "\L"
+        addNLIfEmpty()
+        resString.complete()
+        return
+      elif c == "\L":
+        addNLIfEmpty()
+        resString.complete()
+        return
+      resString.mget.add c
+
+  resString.complete()
+
 proc recvLine*(socket: AsyncSocket,
     flags = {SocketFlag.SafeDisconn}): Future[string] {.async.} =
   ## Reads a line of data from ``socket``. Returned future will complete once
@@ -325,52 +421,11 @@ proc recvLine*(socket: AsyncSocket,
     if result.len == 0:
       result.add("\c\L")
   assert SocketFlag.Peek notin flags ## TODO:
-  if socket.isBuffered:
-    result = ""
-    if socket.bufLen == 0:
-      let res = await socket.readIntoBuf(flags)
-      if res == 0:
-        return
 
-    var lastR = false
+  # TODO: Optimise this.
-    while true:
+  var resString = newFutureVar[string]("asyncnet.recvLine")
-      if socket.currPos >= socket.bufLen:
+  await socket.recvLineInto(resString, flags)
-        let res = await socket.readIntoBuf(flags)
+  result = resString.mget()
-        if res == 0:
-          result = ""
-          break
-
-      case socket.buffer[socket.currPos]
-      of '\r':
-        lastR = true
-        addNLIfEmpty()
-      of '\L':
-        addNLIfEmpty()
-        socket.currPos.inc()
-        return
-      else:
-        if lastR:
-          socket.currPos.inc()
-          return
-        else:
-          result.add socket.buffer[socket.currPos]
-      socket.currPos.inc()
-  else:
-    result = ""
-    var c = ""
-    while true:
-      c = await recv(socket, 1, flags)
-      if c.len == 0:
-        return ""
-      if c == "\r":
-        c = await recv(socket, 1, flags) # Skip \L
-        assert c == "\L"
-        addNLIfEmpty()
-        return
-      elif c == "\L":
-        addNLIfEmpty()
-        return
-      add(result.string, c)
 
 proc listen*(socket: AsyncSocket, backlog = SOMAXCONN) {.tags: [ReadIOEffect].} =
   ## Marks ``socket`` as accepting connections.

lib/pure/strtabs.nim

@@ -168,6 +168,12 @@ proc newStringTable*(mode: StringTableMode): StringTableRef {.
   result.counter = 0
   newSeq(result.data, startSize)
 
+proc clear*(s: StringTableRef, mode: StringTableMode) =
+  ## resets a string table to be empty again.
+  s.mode = mode
+  s.counter = 0
+  s.data.setLen(startSize)
+
 proc newStringTable*(keyValuePairs: varargs[string],
                      mode: StringTableMode): StringTableRef {.
   rtl, extern: "nst$1WithPairs".} =

lib/pure/uri.nim

@@ -53,10 +53,10 @@ proc parseAuthority(authority: string, result: var Uri) =
   while true:
     case authority[i]
     of '@':
-      result.password = result.port
+      swap result.password, result.port
-      result.port = ""
+      result.port.setLen(0)
-      result.username = result.hostname
+      swap result.username, result.hostname
-      result.hostname = ""
+      result.hostname.setLen(0)
       inPort = false
     of ':':
       inPort = true
@@ -75,7 +75,7 @@ proc parsePath(uri: string, i: var int, result: var Uri) =
   # The 'mailto' scheme's PATH actually contains the hostname/username
   if result.scheme.toLower == "mailto":
     parseAuthority(result.path, result)
-    result.path = ""
+    result.path.setLen(0)
 
   if uri[i] == '?':
     i.inc # Skip '?'
@@ -85,13 +85,21 @@ proc parsePath(uri: string, i: var int, result: var Uri) =
     i.inc # Skip '#'
     i.inc parseUntil(uri, result.anchor, {}, i)
 
-proc initUri(): Uri =
+proc initUri*(): Uri =
+  ## Initializes a URI.
   result = Uri(scheme: "", username: "", password: "", hostname: "", port: "",
                 path: "", query: "", anchor: "")
 
-proc parseUri*(uri: string): Uri =
+proc resetUri(uri: var Uri) =
-  ## Parses a URI.
+  for f in uri.fields:
-  result = initUri()
+    when f is string:
+      f.setLen(0)
+    else:
+      f = false
+
+proc parseUri*(uri: string, result: var Uri) =
+  ## Parses a URI. The `result` variable will be cleared before.
+  resetUri(result)
 
   var i = 0
 
@@ -105,7 +113,7 @@ proc parseUri*(uri: string): Uri =
   if uri[i] != ':':
     # Assume this is a reference URI (relative URI)
     i = 0
-    result.scheme = ""
+    result.scheme.setLen(0)
     parsePath(uri, i, result)
     return
   i.inc # Skip ':'
@@ -124,6 +132,11 @@ proc parseUri*(uri: string): Uri =
   # Path
   parsePath(uri, i, result)
 
+proc parseUri*(uri: string): Uri =
+  ## Parses a URI and returns it.
+  result = initUri()
+  parseUri(uri, result)
+
 proc removeDotSegments(path: string): string =
   var collection: seq[string] = @[]
   let endsWithSlash = path[path.len-1] == '/'