Merge branch 'clean-speedup-2' of https://github.com/def-/Nim into def--clean-speedup-2
This commit is contained in:
Dominik Picheta
commit
03cbf689a1
4 changed files with 268 additions and 99 deletions
|
|
@ -24,7 +24,7 @@
|
|||
##
|
||||
## Chat server
|
||||
## ^^^^^^^^^^^
|
||||
##
|
||||
##
|
||||
## The following example demonstrates a simple chat server.
|
||||
##
|
||||
## .. code-block::nim
|
||||
|
|
@ -69,13 +69,13 @@ type
|
|||
# TODO: I would prefer to just do:
|
||||
# AsyncSocket* {.borrow: `.`.} = distinct Socket. But that doesn't work.
|
||||
AsyncSocketDesc = object
|
||||
fd: SocketHandle
|
||||
fd*: SocketHandle
|
||||
closed: bool ## determines whether this socket has been closed
|
||||
case isBuffered: bool ## determines whether this socket is buffered.
|
||||
of true:
|
||||
buffer: array[0..BufferSize, char]
|
||||
currPos: int # current index in buffer
|
||||
bufLen: int # current length of buffer
|
||||
bufLen*: int # current length of buffer
|
||||
of false: nil
|
||||
case isSsl: bool
|
||||
of true:
|
||||
|
|
@ -182,26 +182,142 @@ 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.} =
|
||||
when defined(windows):
|
||||
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] = '\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
|
||||
else:
|
||||
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
|
||||
|
||||
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 +338,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 +353,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 +368,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 +419,14 @@ 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: ptr string,
|
||||
flags = {SocketFlag.SafeDisconn}) {.async.} =
|
||||
## Reads a line of data from ``socket`` into ``resString``.
|
||||
##
|
||||
## 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 +434,29 @@ 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:
|
||||
result = newFuture[void]("asyncnet.recvLineInto")
|
||||
|
||||
template addNLIfEmpty(): stmt =
|
||||
if resString[].len == 0:
|
||||
resString[].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:
|
||||
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[].setLen(0)
|
||||
return
|
||||
|
||||
case socket.buffer[socket.currPos]
|
||||
of '\r':
|
||||
|
|
@ -353,24 +471,53 @@ proc recvLine*(socket: AsyncSocket,
|
|||
socket.currPos.inc()
|
||||
return
|
||||
else:
|
||||
result.add socket.buffer[socket.currPos]
|
||||
resString[].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[].setLen(0)
|
||||
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()
|
||||
return
|
||||
elif c == "\L":
|
||||
addNLIfEmpty()
|
||||
return
|
||||
add(result.string, c)
|
||||
resString[].add c
|
||||
|
||||
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:
|
||||
|
||||
result = ""
|
||||
await socket.recvLineInto(addr result, flags)
|
||||
|
||||
proc listen*(socket: AsyncSocket, backlog = SOMAXCONN) {.tags: [ReadIOEffect].} =
|
||||
## Marks ``socket`` as accepting connections.
|
||||
|
|
@ -500,11 +647,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()
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue