diff options
| -rw-r--r-- | lib/pure/asyncdispatch.nim | 142 | ||||
| -rw-r--r-- | lib/pure/asynchttpserver.nim | 95 | ||||
| -rw-r--r-- | lib/pure/asyncnet.nim | 135 | ||||
| -rw-r--r-- | lib/pure/strtabs.nim | 8 | ||||
| -rw-r--r-- | lib/pure/uri.nim | 33 |
5 files changed, 317 insertions, 96 deletions
diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim index 27f77cef2..bec2632d5 100644 --- a/lib/pure/asyncdispatch.nim +++ b/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") diff --git a/lib/pure/asynchttpserver.nim b/lib/pure/asynchttpserver.nim index dc5a55dcc..74e9e9f36 100644 --- a/lib/pure/asynchttpserver.nim +++ b/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) -## runForever() +## waitFor server.serve(Port(8080), cb) 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, - content: string, headers = newStringTable()) {.async.} = +proc respond*(req: Request, code: HttpCode, content: string, + 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 = - result.headers = newStringTable(modeCaseInsensitive) - result.hostname = "" - result.body = "" + if headers != nil: + msg.addHeaders(headers) + msg.add("Content-Length: " & $content.len & "\c\L\c\L") + msg.add(content) + result = req.client.send(msg) proc parseHeader(line: string): tuple[key, value: string] = var i = 0 @@ -149,59 +145,68 @@ 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.hostname = address + request.headers.clear(modeCaseInsensitive) + request.hostname.shallowCopy(address) assert client != nil request.client = client # First line - GET /path HTTP/1.1 - let line = await client.recvLine() # TODO: Timeouts. - if line == "": + lineFut.mget().setLen(0) + 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] - let path = lineParts[1] - let protocol = lineParts[2] + var i = 0 + for linePart in lineFut.mget.split(' '): + 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() - if headerLine == "": - client.close(); return - if headerLine == "\c\L": break - # TODO: Compiler crash - #let (key, value) = parseHeader(headerLine) - let kv = parseHeader(headerLine) - request.headers[kv.key] = kv.value + lineFut.mget.setLen(0) + lineFut.clean() + await client.recvLineInto(lineFut) - request.reqMethod = reqMethod - request.url = parseUri(path) - try: - request.protocol = protocol.parseProtocol() - except ValueError: - asyncCheck request.respond(Http400, "Invalid request protocol. Got: " & - protocol) - continue + if lineFut.mget == "": + client.close(); return + if lineFut.mget == "\c\L": break + let (key, value) = parseHeader(lineFut.mget) + request.headers[key] = value - if reqMethod.normalize == "post": + if request.reqMethod == "post": # Check for Expect header if request.headers.hasKey("Expect"): if request.headers["Expect"].toLower == "100-continue": await client.sendStatus("100 Continue") else: await client.sendStatus("417 Expectation Failed") - + # Read the body # - Check for Content-length header if request.headers.hasKey("Content-Length"): @@ -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 @@ -247,7 +252,7 @@ proc serve*(server: AsyncHttpServer, port: Port, server.socket.setSockOpt(OptReuseAddr, true) server.socket.bindAddr(port, address) server.socket.listen() - + while true: # TODO: Causes compiler crash. #var (address, client) = await server.socket.acceptAddr() diff --git a/lib/pure/asyncnet.nim b/lib/pure/asyncnet.nim index 39d05d36b..62e85042f 100644 --- a/lib/pure/asyncnet.nim +++ b/lib/pure/asyncnet.nim @@ -24,7 +24,7 @@ ## ## Chat server ## ^^^^^^^^^^^ -## +## ## The following example demonstrates a simple chat server. ## ## .. code-block::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, - flags: set[SocketFlag]): Future[int] {.async.} = +template readInto(buf: cstring, size: int, socket: AsyncSocket, + 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) - if data.len != 0: - copyMem(buf, addr data[0], data.len) + var recvIntoFut = recvInto(socket.fd.TAsyncFD, buf, size, flags) + yield recvIntoFut # Not in SSL mode. - result = data.len + res = recvIntoFut.read() + res -proc readIntoBuf(socket: AsyncSocket, - flags: set[SocketFlag]): Future[int] {.async.} = - result = await readInto(addr socket.buffer[0], BufferSize, socket, flags) +template readIntoBuf(socket: AsyncSocket, + flags: set[SocketFlag]): int = + 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,15 +307,17 @@ proc accept*(socket: AsyncSocket, retFut.complete(future.read.client) return retFut -proc recvLine*(socket: AsyncSocket, - flags = {SocketFlag.SafeDisconn}): Future[string] {.async.} = - ## Reads a line of data from ``socket``. Returned future will complete once - ## a full line is read or an error occurs. +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`` @@ -318,27 +325,37 @@ proc recvLine*(socket: AsyncSocket, ## The partial line **will be lost**. ## ## **Warning**: The ``Peek`` flag is not yet implemented. - ## - ## **Warning**: ``recvLine`` on unbuffered sockets assumes that the protocol - ## uses ``\r\L`` to delimit a new line. - template addNLIfEmpty(): stmt = - if result.len == 0: - result.add("\c\L") + ## + ## **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: - result = "" if socket.bufLen == 0: - let res = await socket.readIntoBuf(flags) + let res = socket.readIntoBuf(flags) if res == 0: + resString.complete() return var lastR = false while true: if socket.currPos >= socket.bufLen: - let res = await socket.readIntoBuf(flags) + let res = socket.readIntoBuf(flags) if res == 0: - result = "" - break + resString.mget().setLen(0) + resString.complete() + return case socket.buffer[socket.currPos] of '\r': @@ -347,30 +364,68 @@ proc recvLine*(socket: AsyncSocket, of '\L': addNLIfEmpty() socket.currPos.inc() + resString.complete() return else: if lastR: socket.currPos.inc() + resString.complete() return else: - result.add socket.buffer[socket.currPos] + resString.mget.add socket.buffer[socket.currPos] socket.currPos.inc() else: - result = "" var c = "" while true: - c = await recv(socket, 1, flags) + let recvFut = recv(socket, 1, flags) + c = recvFut.read() if c.len == 0: - return "" + resString.mget.setLen(0) + resString.complete() + return if c == "\r": - c = await recv(socket, 1, flags) # Skip \L + 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 - add(result.string, c) + 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 + ## a full line is read or an error occurs. + ## + ## 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**: ``recvLine`` on unbuffered sockets assumes that the protocol + ## uses ``\r\L`` to delimit a new line. + template addNLIfEmpty(): stmt = + if result.len == 0: + result.add("\c\L") + assert SocketFlag.Peek notin flags ## TODO: + + # TODO: Optimise this. + var resString = newFutureVar[string]("asyncnet.recvLine") + await socket.recvLineInto(resString, flags) + result = resString.mget() proc listen*(socket: AsyncSocket, backlog = SOMAXCONN) {.tags: [ReadIOEffect].} = ## Marks ``socket`` as accepting connections. @@ -500,11 +555,11 @@ when not defined(testing) and isMainModule: proc (future: Future[void]) = echo("Send") client.close() - + var f = accept(sock) f.callback = onAccept - + var f = accept(sock) f.callback = onAccept runForever() - + diff --git a/lib/pure/strtabs.nim b/lib/pure/strtabs.nim index 727d5a386..7fdd994f2 100644 --- a/lib/pure/strtabs.nim +++ b/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".} = @@ -227,7 +233,7 @@ proc `$`*(t: StringTableRef): string {.rtl, extern: "nstDollar".} = result = "{:}" else: result = "{" - for key, val in pairs(t): + for key, val in pairs(t): if result.len > 1: result.add(", ") result.add(key) result.add(": ") diff --git a/lib/pure/uri.nim b/lib/pure/uri.nim index b0afb75f9..1890a9bf4 100644 --- a/lib/pure/uri.nim +++ b/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 - result.port = "" - result.username = result.hostname - result.hostname = "" + swap result.password, result.port + result.port.setLen(0) + swap result.username, 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 = - ## Parses a URI. - result = initUri() +proc resetUri(uri: var Uri) = + for f in uri.fields: + 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] == '/' |