Many async optimisations.

* Selectors implementation will now attempt to immediately execute an IO
  operation instead of waiting for a ready notification.
* Removed recursion in asynchttpserver.
* Improved buffered implementation of recvLine in asyncnet.
* Optimised ``respond`` in asynchttpserver removing a possible "Delayed ACK"
  situation.
This commit is contained in:
Dominik Picheta 2014-07-12 22:51:06 +01:00
commit cf5c8a204e
4 changed files with 166 additions and 116 deletions

View file

@ -552,7 +552,18 @@ when defined(windows) or defined(nimdoc):
initAll() initAll()
else: else:
import selectors import selectors
from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK when defined(windows):
import winlean
const
EINTR = WSAEINPROGRESS
EINPROGRESS = WSAEINPROGRESS
EWOULDBLOCK = WSAEWOULDBLOCK
EAGAIN = EINPROGRESS
MSG_NOSIGNAL = 0
else:
from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK,
MSG_NOSIGNAL
type type
TAsyncFD* = distinct cint TAsyncFD* = distinct cint
TCallback = proc (sock: TAsyncFD): bool {.closure,gcsafe.} TCallback = proc (sock: TAsyncFD): bool {.closure,gcsafe.}
@ -693,7 +704,7 @@ else:
proc cb(sock: TAsyncFD): bool = proc cb(sock: TAsyncFD): bool =
result = true result = true
let res = recv(sock.TSocketHandle, addr readBuffer[0], size, let res = recv(sock.TSocketHandle, addr readBuffer[0], size.cint,
flags.cint) flags.cint)
#echo("recv cb res: ", res) #echo("recv cb res: ", res)
if res < 0: if res < 0:
@ -708,7 +719,7 @@ else:
else: else:
readBuffer.setLen(res) readBuffer.setLen(res)
retFuture.complete(readBuffer) retFuture.complete(readBuffer)
if not cb(socket):
addRead(socket, cb) addRead(socket, cb)
return retFuture return retFuture
@ -721,7 +732,8 @@ else:
result = true result = true
let netSize = data.len-written let netSize = data.len-written
var d = data.cstring var d = data.cstring
let res = send(sock.TSocketHandle, addr d[written], netSize, 0.cint) let res = send(sock.TSocketHandle, addr d[written], netSize.cint,
MSG_NOSIGNAL)
if res < 0: if res < 0:
let lastError = osLastError() let lastError = osLastError()
if lastError.int32 notin {EINTR, EWOULDBLOCK, EAGAIN}: if lastError.int32 notin {EINTR, EWOULDBLOCK, EAGAIN}:
@ -734,6 +746,7 @@ else:
result = false # We still have data to send. result = false # We still have data to send.
else: else:
retFuture.complete() retFuture.complete()
if not cb(socket):
addWrite(socket, cb) addWrite(socket, cb)
return retFuture return retFuture
@ -756,6 +769,7 @@ else:
else: else:
register(client.TAsyncFD) register(client.TAsyncFD)
retFuture.complete(($inet_ntoa(sockAddress.sin_addr), client.TAsyncFD)) retFuture.complete(($inet_ntoa(sockAddress.sin_addr), client.TAsyncFD))
if not cb(socket):
addRead(socket, cb) addRead(socket, cb)
return retFuture return retFuture

View file

@ -51,10 +51,15 @@ proc `==`*(protocol: tuple[orig: string, major, minor: int],
proc newAsyncHttpServer*(): PAsyncHttpServer = proc newAsyncHttpServer*(): PAsyncHttpServer =
new result new result
proc sendHeaders*(req: TRequest, headers: PStringTable) {.async.} = proc addHeaders(msg: var string, headers: PStringTable) =
## Sends the specified headers to the requesting client.
for k, v in headers: for k, v in headers:
await req.client.send(k & ": " & v & "\c\L") msg.add(k & ": " & v & "\c\L")
proc sendHeaders*(req: TRequest, headers: PStringTable): PFuture[void] =
## Sends the specified headers to the requesting client.
var msg = ""
addHeaders(msg, headers)
return req.client.send(msg)
proc respond*(req: TRequest, code: THttpCode, proc respond*(req: TRequest, code: THttpCode,
content: string, headers: PStringTable = newStringTable()) {.async.} = content: string, headers: PStringTable = newStringTable()) {.async.} =
@ -64,9 +69,9 @@ proc respond*(req: TRequest, code: THttpCode,
## This procedure will **not** close the client socket. ## This procedure will **not** close the client socket.
var customHeaders = headers var customHeaders = headers
customHeaders["Content-Length"] = $content.len customHeaders["Content-Length"] = $content.len
await req.client.send("HTTP/1.1 " & $code & "\c\L") var msg = "HTTP/1.1 " & $code & "\c\L"
await sendHeaders(req, headers) msg.addHeaders(customHeaders)
await req.client.send("\c\L" & content) await req.client.send(msg & "\c\L" & content)
proc newRequest(): TRequest = proc newRequest(): TRequest =
result.headers = newStringTable(modeCaseInsensitive) result.headers = newStringTable(modeCaseInsensitive)
@ -93,6 +98,7 @@ proc sendStatus(client: PAsyncSocket, status: string): PFuture[void] =
proc processClient(client: PAsyncSocket, address: string, proc processClient(client: PAsyncSocket, address: string,
callback: proc (request: TRequest): PFuture[void]) {.async.} = callback: proc (request: TRequest): PFuture[void]) {.async.} =
while true:
# GET /path HTTP/1.1 # GET /path HTTP/1.1
# Header: val # Header: val
# \n # \n
@ -174,9 +180,9 @@ proc processClient(client: PAsyncSocket, address: string,
# header states otherwise. # header states otherwise.
# In HTTP 1.0 we assume that the connection should not be persistent. # In HTTP 1.0 we assume that the connection should not be persistent.
# Unless the connection header states otherwise. # Unless the connection header states otherwise.
await processClient(client, address, callback)
else: else:
request.client.close() request.client.close()
break
proc serve*(server: PAsyncHttpServer, port: TPort, proc serve*(server: PAsyncHttpServer, port: TPort,
callback: proc (request: TRequest): PFuture[void], callback: proc (request: TRequest): PFuture[void],

View file

@ -110,12 +110,10 @@ proc recv*(socket: PAsyncSocket, size: int,
if socket.currPos >= socket.bufLen: if socket.currPos >= socket.bufLen:
if (flags and MSG_PEEK) == MSG_PEEK: if (flags and MSG_PEEK) == MSG_PEEK:
# We don't want to get another buffer if we're peeking. # We don't want to get another buffer if we're peeking.
result.setLen(read) break
return
let res = await socket.readIntoBuf(flags and (not MSG_PEEK)) let res = await socket.readIntoBuf(flags and (not MSG_PEEK))
if res == 0: if res == 0:
result.setLen(read) break
return
let chunk = min(socket.bufLen-socket.currPos, size-read) let chunk = min(socket.bufLen-socket.currPos, size-read)
copyMem(addr(result[read]), addr(socket.buffer[socket.currPos]), chunk) copyMem(addr(result[read]), addr(socket.buffer[socket.currPos]), chunk)
@ -181,11 +179,43 @@ proc recvLine*(socket: PAsyncSocket): PFuture[string] {.async.} =
## If the socket is disconnected in the middle of a line (before ``\r\L`` ## If the socket is disconnected in the middle of a line (before ``\r\L``
## is read) then line will be set to ``""``. ## is read) then line will be set to ``""``.
## The partial line **will be lost**. ## The partial line **will be lost**.
template addNLIfEmpty(): stmt = template addNLIfEmpty(): stmt =
if result.len == 0: if result.len == 0:
result.add("\c\L") result.add("\c\L")
if socket.isBuffered:
result = ""
if socket.bufLen == 0:
let res = await socket.readIntoBuf(0)
if res == 0:
return
var lastR = false
while true:
if socket.currPos >= socket.bufLen:
let res = await socket.readIntoBuf(0)
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 = "" result = ""
var c = "" var c = ""
while true: while true:

View file

@ -163,7 +163,7 @@ elif defined(linux):
proc newSelector*(): PSelector = proc newSelector*(): PSelector =
new result new result
result.epollFD = epoll_create(64) result.epollFD = epoll_create(64)
result.events = cast[array[64, epoll_event]](alloc0(sizeof(epoll_event)*64)) #result.events = cast[array[64, epoll_event]](alloc0(sizeof(epoll_event)*64))
result.fds = initTable[TSocketHandle, PSelectorKey]() result.fds = initTable[TSocketHandle, PSelectorKey]()
if result.epollFD < 0: if result.epollFD < 0:
OSError(OSLastError()) OSError(OSLastError())