jyapayne/Nim
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement safe flags for socket operations.

Browse source
This commit is contained in:
Dominik Picheta 2014-07-13 22:32:39 +01:00
commit ac8ddb0720
4 changed files with 108 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

43
lib/pure/asyncdispatch.nim
View file

@ -11,8 +11,9 @@ include "system/inclrtl"
import os, oids, tables, strutils, macros import os, oids, tables, strutils, macros
import rawsockets import rawsockets, net
export TPort
export TPort, TSocketFlags
#{.injectStmt: newGcInvariant().} #{.injectStmt: newGcInvariant().}
@ -353,7 +354,7 @@ when defined(windows) or defined(nimdoc):
return retFuture return retFuture
proc recv*(socket: TAsyncFD, size: int, proc recv*(socket: TAsyncFD, size: int,
flags: int = 0): PFuture[string] = flags = {TSocketFlags.SafeDisconn}): PFuture[string] =
## Reads **up to** ``size`` bytes from ``socket``. Returned future will ## Reads **up to** ``size`` bytes from ``socket``. Returned future will
## complete once all the data requested is read, a part of the data has been ## 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 ## read, or the socket has disconnected in which case the future will
@ -373,7 +374,7 @@ when defined(windows) or defined(nimdoc):
dataBuf.len = size dataBuf.len = size
var bytesReceived: DWord var bytesReceived: DWord
var flagsio = flags.DWord var flagsio = flags.toOSFlags().DWord
var ol = PCustomOverlapped() var ol = PCustomOverlapped()
GC_ref(ol) GC_ref(ol)
ol.data = TCompletionData(sock: socket, cb: ol.data = TCompletionData(sock: socket, cb:
@ -403,7 +404,10 @@ when defined(windows) or defined(nimdoc):
dealloc dataBuf.buf dealloc dataBuf.buf
dataBuf.buf = nil dataBuf.buf = nil
GC_unref(ol) GC_unref(ol)
retFuture.fail(newException(EOS, osErrorMsg(err))) if flags.isDisconnectionError(err):
retFuture.complete("")
else:
retFuture.fail(newException(EOS, osErrorMsg(err)))
elif ret == 0 and bytesReceived == 0 and dataBuf.buf[0] == '\0': elif ret == 0 and bytesReceived == 0 and dataBuf.buf[0] == '\0':
# We have to ensure that the buffer is empty because WSARecv will tell # We have to ensure that the buffer is empty because WSARecv will tell
# us immediatelly when it was disconnected, even when there is still # us immediatelly when it was disconnected, even when there is still
@ -434,7 +438,8 @@ when defined(windows) or defined(nimdoc):
# free ``ol``. # free ``ol``.
return retFuture return retFuture
proc send*(socket: TAsyncFD, data: string): PFuture[void] = proc send*(socket: TAsyncFD, data: string,
flags = {TSocketFlags.SafeDisconn}): PFuture[void] =
## Sends ``data`` to ``socket``. The returned future will complete once all ## Sends ``data`` to ``socket``. The returned future will complete once all
## data has been sent. ## data has been sent.
verifyPresence(socket) verifyPresence(socket)
@ -444,7 +449,7 @@ when defined(windows) or defined(nimdoc):
dataBuf.buf = data # since this is not used in a callback, this is fine dataBuf.buf = data # since this is not used in a callback, this is fine
dataBuf.len = data.len dataBuf.len = data.len
var bytesReceived, flags: DWord var bytesReceived, lowFlags: DWord
var ol = PCustomOverlapped() var ol = PCustomOverlapped()
GC_ref(ol) GC_ref(ol)
ol.data = TCompletionData(sock: socket, cb: ol.data = TCompletionData(sock: socket, cb:
@ -457,12 +462,15 @@ when defined(windows) or defined(nimdoc):
) )
let ret = WSASend(socket.TSocketHandle, addr dataBuf, 1, addr bytesReceived, let ret = WSASend(socket.TSocketHandle, addr dataBuf, 1, addr bytesReceived,
flags, cast[POverlapped](ol), nil) lowFlags, cast[POverlapped](ol), nil)
if ret == -1: if ret == -1:
let err = osLastError() let err = osLastError()
if err.int32 != ERROR_IO_PENDING: if err.int32 != ERROR_IO_PENDING:
retFuture.fail(newException(EOS, osErrorMsg(err)))
GC_unref(ol) GC_unref(ol)
if flags.isDisconnectionError(err):
retFuture.complete()
else:
retFuture.fail(newException(EOS, osErrorMsg(err)))
else: else:
retFuture.complete() retFuture.complete()
# We don't deallocate ``ol`` here because even though this completed # We don't deallocate ``ol`` here because even though this completed
@ -706,7 +714,7 @@ else:
return retFuture return retFuture
proc recv*(socket: TAsyncFD, size: int, proc recv*(socket: TAsyncFD, size: int,
flags: int = 0): PFuture[string] = flags = {TSocketFlags.SafeDisconn}): PFuture[string] =
var retFuture = newFuture[string]() var retFuture = newFuture[string]()
var readBuffer = newString(size) var readBuffer = newString(size)
@ -719,7 +727,10 @@ else:
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}:
retFuture.fail(newException(EOS, osErrorMsg(lastError))) if flags.isDisconnectionError(lastError):
retFuture.complete("")
else:
retFuture.fail(newException(EOS, osErrorMsg(lastError)))
else: else:
result = false # We still want this callback to be called. result = false # We still want this callback to be called.
elif res == 0: elif res == 0:
@ -733,7 +744,8 @@ else:
addRead(socket, cb) addRead(socket, cb)
return retFuture return retFuture
proc send*(socket: TAsyncFD, data: string): PFuture[void] = proc send*(socket: TAsyncFD, data: string,
flags = {TSocketFlags.SafeDisconn}): PFuture[void] =
var retFuture = newFuture[void]() var retFuture = newFuture[void]()
var written = 0 var written = 0
@ -747,7 +759,10 @@ else:
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}:
retFuture.fail(newException(EOS, osErrorMsg(lastError))) if flags.isDisconnectionError(lastError):
retFuture.complete("")
else:
retFuture.fail(newException(EOS, osErrorMsg(lastError)))
else: else:
result = false # We still want this callback to be called. result = false # We still want this callback to be called.
else: else:
@ -1065,7 +1080,7 @@ proc recvLine*(socket: TAsyncFD): PFuture[string] {.async.} =
if c.len == 0: if c.len == 0:
return "" return ""
if c == "\r": if c == "\r":
c = await recv(socket, 1, MSG_PEEK) c = await recv(socket, 1, {TSocketFlags.SafeDisconn, TSocketFlags.Peek})
if c.len > 0 and c == "\L": if c.len > 0 and c == "\L":
discard await recv(socket, 1) discard await recv(socket, 1)
addNLIfEmpty() addNLIfEmpty()

37
lib/pure/asyncnet.nim
View file

@ -80,7 +80,8 @@ proc connect*(socket: PAsyncSocket, address: string, port: TPort,
## or an error occurs. ## or an error occurs.
result = connect(socket.fd.TAsyncFD, address, port, af) result = connect(socket.fd.TAsyncFD, address, port, af)
proc readIntoBuf(socket: PAsyncSocket, flags: int): PFuture[int] {.async.} = proc readIntoBuf(socket: PAsyncSocket,
flags: set[TSocketFlags]): PFuture[int] {.async.} =
var data = await recv(socket.fd.TAsyncFD, BufferSize, flags) var data = await recv(socket.fd.TAsyncFD, BufferSize, flags)
if data.len != 0: if data.len != 0:
copyMem(addr socket.buffer[0], addr data[0], data.len) copyMem(addr socket.buffer[0], addr data[0], data.len)
@ -89,7 +90,7 @@ proc readIntoBuf(socket: PAsyncSocket, flags: int): PFuture[int] {.async.} =
result = data.len result = data.len
proc recv*(socket: PAsyncSocket, size: int, proc recv*(socket: PAsyncSocket, size: int,
flags: int = 0): PFuture[string] {.async.} = flags = {TSocketFlags.SafeDisconn}): PFuture[string] {.async.} =
## Reads ``size`` bytes from ``socket``. Returned future will complete once ## Reads ``size`` bytes from ``socket``. Returned future will complete once
## all of the requested data is read. If socket is disconnected during the ## all of the requested data is read. If socket is disconnected during the
## recv operation then the future may complete with only a part of the ## recv operation then the future may complete with only a part of the
@ -100,7 +101,7 @@ proc recv*(socket: PAsyncSocket, size: int,
let originalBufPos = socket.currPos let originalBufPos = socket.currPos
if socket.bufLen == 0: if socket.bufLen == 0:
let res = await socket.readIntoBuf(flags and (not MSG_PEEK)) let res = await socket.readIntoBuf(flags - {TSocketFlags.Peek})
if res == 0: if res == 0:
result.setLen(0) result.setLen(0)
return return
@ -108,10 +109,10 @@ proc recv*(socket: PAsyncSocket, size: int,
var read = 0 var read = 0
while read < size: while read < size:
if socket.currPos >= socket.bufLen: if socket.currPos >= socket.bufLen:
if (flags and MSG_PEEK) == MSG_PEEK: if TSocketFlags.Peek in flags:
# We don't want to get another buffer if we're peeking. # We don't want to get another buffer if we're peeking.
break break
let res = await socket.readIntoBuf(flags and (not MSG_PEEK)) let res = await socket.readIntoBuf(flags - {TSocketFlags.Peek})
if res == 0: if res == 0:
break break
@ -120,18 +121,19 @@ proc recv*(socket: PAsyncSocket, size: int,
read.inc(chunk) read.inc(chunk)
socket.currPos.inc(chunk) socket.currPos.inc(chunk)
if (flags and MSG_PEEK) == MSG_PEEK: if TSocketFlags.Peek in flags:
# Restore old buffer cursor position. # Restore old buffer cursor position.
socket.currPos = originalBufPos socket.currPos = originalBufPos
result.setLen(read) result.setLen(read)
else: else:
result = await recv(socket.fd.TAsyncFD, size, flags) result = await recv(socket.fd.TAsyncFD, size, flags)
proc send*(socket: PAsyncSocket, data: string): PFuture[void] = proc send*(socket: PAsyncSocket, data: string,
flags = {TSocketFlags.SafeDisconn}): PFuture[void] =
## Sends ``data`` to ``socket``. The returned future will complete once all ## Sends ``data`` to ``socket``. The returned future will complete once all
## data has been sent. ## data has been sent.
assert socket != nil assert socket != nil
result = send(socket.fd.TAsyncFD, data) result = send(socket.fd.TAsyncFD, data, flags)
proc acceptAddr*(socket: PAsyncSocket): proc acceptAddr*(socket: PAsyncSocket):
PFuture[tuple[address: string, client: PAsyncSocket]] = PFuture[tuple[address: string, client: PAsyncSocket]] =
@ -166,7 +168,8 @@ proc accept*(socket: PAsyncSocket): PFuture[PAsyncSocket] =
retFut.complete(future.read.client) retFut.complete(future.read.client)
return retFut return retFut
proc recvLine*(socket: PAsyncSocket): PFuture[string] {.async.} = proc recvLine*(socket: PAsyncSocket,
flags = {TSocketFlags.SafeDisconn}): PFuture[string] {.async.} =
## Reads a line of data from ``socket``. Returned future will complete once ## Reads a line of data from ``socket``. Returned future will complete once
## a full line is read or an error occurs. ## a full line is read or an error occurs.
## ##
@ -179,21 +182,23 @@ 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**.
##
## **Warning**: The ``Peek`` flag is not yet implemented.
template addNLIfEmpty(): stmt = template addNLIfEmpty(): stmt =
if result.len == 0: if result.len == 0:
result.add("\c\L") result.add("\c\L")
assert TSocketFlags.Peek notin flags ## TODO:
if socket.isBuffered: if socket.isBuffered:
result = "" result = ""
if socket.bufLen == 0: if socket.bufLen == 0:
let res = await socket.readIntoBuf(0) let res = await socket.readIntoBuf(flags)
if res == 0: if res == 0:
return return
var lastR = false var lastR = false
while true: while true:
if socket.currPos >= socket.bufLen: if socket.currPos >= socket.bufLen:
let res = await socket.readIntoBuf(0) let res = await socket.readIntoBuf(flags)
if res == 0: if res == 0:
result = "" result = ""
break break
@ -214,18 +219,16 @@ proc recvLine*(socket: PAsyncSocket): PFuture[string] {.async.} =
result.add socket.buffer[socket.currPos] result.add socket.buffer[socket.currPos]
socket.currPos.inc() socket.currPos.inc()
else: else:
result = "" result = ""
var c = "" var c = ""
while true: while true:
c = await recv(socket, 1) c = await recv(socket, 1, flags)
if c.len == 0: if c.len == 0:
return "" return ""
if c == "\r": if c == "\r":
c = await recv(socket, 1, MSG_PEEK) c = await recv(socket, 1, flags + {TSocketFlags.Peek})
if c.len > 0 and c == "\L": if c.len > 0 and c == "\L":
let dummy = await recv(socket, 1) let dummy = await recv(socket, 1, flags)
assert dummy == "\L" assert dummy == "\L"
addNLIfEmpty() addNLIfEmpty()
return return

71
lib/pure/net.nim
View file

@ -350,6 +350,30 @@ type
ETimeout* = object of ESynch ETimeout* = object of ESynch
TSocketFlags* {.pure.} = enum
Peek,
SafeDisconn ## Ensures disconnection exceptions (ECONNRESET, EPIPE etc) are not thrown.
proc isDisconnectionError*(flags: set[TSocketFlags],
lastError: TOSErrorCode): bool =
## Determines whether ``lastError`` is a disconnection error. Only does this
## if flags contains ``SafeDisconn``.
when useWinVersion:
TSocketFlags.SafeDisconn in flags and
lastError.int32 in {WSAECONNRESET, WSAECONNABORTED, WSAENETRESET,
WSAEDISCON}
else:
TSocketFlags.SafeDisconn in flags and
lastError.int32 in {ECONNRESET, EPIPE, ENETRESET}
proc toOSFlags*(socketFlags: set[TSocketFlags]): cint =
## Converts the flags into the underlying OS representation.
for f in socketFlags:
case f
of TSocketFlags.Peek:
result = result or MSG_PEEK
of TSocketFlags.SafeDisconn: continue
proc createSocket(fd: TSocketHandle, isBuff: bool): PSocket = proc createSocket(fd: TSocketHandle, isBuff: bool): PSocket =
assert fd != osInvalidSocket assert fd != osInvalidSocket
new(result) new(result)
@ -470,7 +494,8 @@ when defined(ssl):
if SSLSetFd(socket.sslHandle, socket.fd) != 1: if SSLSetFd(socket.sslHandle, socket.fd) != 1:
SSLError() SSLError()
proc socketError*(socket: PSocket, err: int = -1, async = false) = proc socketError*(socket: PSocket, err: int = -1, async = false,
lastError = (-1).TOSErrorCode) =
## Raises an EOS error based on the error code returned by ``SSLGetError`` ## Raises an EOS error based on the error code returned by ``SSLGetError``
## (for SSL sockets) and ``osLastError`` otherwise. ## (for SSL sockets) and ``osLastError`` otherwise.
## ##
@ -500,17 +525,17 @@ proc socketError*(socket: PSocket, err: int = -1, async = false) =
else: SSLError("Unknown Error") else: SSLError("Unknown Error")
if err == -1 and not (when defined(ssl): socket.isSSL else: false): if err == -1 and not (when defined(ssl): socket.isSSL else: false):
let lastError = osLastError() let lastE = if lastError.int == -1: osLastError() else: lastError
if async: if async:
when useWinVersion: when useWinVersion:
if lastError.int32 == WSAEWOULDBLOCK: if lastE.int32 == WSAEWOULDBLOCK:
return return
else: osError(lastError) else: osError(lastE)
else: else:
if lastError.int32 == EAGAIN or lastError.int32 == EWOULDBLOCK: if lastE.int32 == EAGAIN or lastE.int32 == EWOULDBLOCK:
return return
else: osError(lastError) else: osError(lastE)
else: osError(lastError) else: osError(lastE)
proc listen*(socket: PSocket, backlog = SOMAXCONN) {.tags: [FReadIO].} = proc listen*(socket: PSocket, backlog = SOMAXCONN) {.tags: [FReadIO].} =
## Marks ``socket`` as accepting connections. ## Marks ``socket`` as accepting connections.
@ -881,7 +906,8 @@ proc recv*(socket: PSocket, data: pointer, size: int, timeout: int): int {.
result = read result = read
proc recv*(socket: PSocket, data: var string, size: int, timeout = -1): int = proc recv*(socket: PSocket, data: var string, size: int, timeout = -1,
flags = {TSocketFlags.SafeDisconn}): int =
## Higher-level version of ``recv``. ## Higher-level version of ``recv``.
## ##
## When 0 is returned the socket's connection has been closed. ## When 0 is returned the socket's connection has been closed.
@ -893,11 +919,15 @@ proc recv*(socket: PSocket, data: var string, size: int, timeout = -1): int =
## within the time specified an ETimeout exception will be raised. ## within the time specified an ETimeout exception will be raised.
## ##
## **Note**: ``data`` must be initialised. ## **Note**: ``data`` must be initialised.
##
## **Warning**: Only the ``SafeDisconn`` flag is currently supported.
data.setLen(size) data.setLen(size)
result = recv(socket, cstring(data), size, timeout) result = recv(socket, cstring(data), size, timeout)
if result < 0: if result < 0:
data.setLen(0) data.setLen(0)
socket.socketError(result) let lastError = osLastError()
if flags.isDisconnectionError(lastError): return
socket.socketError(result, lastError = lastError)
data.setLen(result) data.setLen(result)
proc peekChar(socket: PSocket, c: var char): int {.tags: [FReadIO].} = proc peekChar(socket: PSocket, c: var char): int {.tags: [FReadIO].} =
@ -920,7 +950,8 @@ proc peekChar(socket: PSocket, c: var char): int {.tags: [FReadIO].} =
return return
result = recv(socket.fd, addr(c), 1, MSG_PEEK) result = recv(socket.fd, addr(c), 1, MSG_PEEK)
proc readLine*(socket: PSocket, line: var TaintedString, timeout = -1) {. proc readLine*(socket: PSocket, line: var TaintedString, timeout = -1,
flags = {TSocketFlags.SafeDisconn}) {.
tags: [FReadIO, FTime].} = tags: [FReadIO, FTime].} =
## Reads a line of data from ``socket``. ## Reads a line of data from ``socket``.
## ##
@ -934,11 +965,18 @@ proc readLine*(socket: PSocket, line: var TaintedString, timeout = -1) {.
## ##
## A timeout can be specified in miliseconds, if data is not received within ## A timeout can be specified in miliseconds, if data is not received within
## the specified time an ETimeout exception will be raised. ## the specified time an ETimeout exception will be raised.
##
## **Warning**: Only the ``SafeDisconn`` flag is currently supported.
template addNLIfEmpty(): stmt = template addNLIfEmpty(): stmt =
if line.len == 0: if line.len == 0:
line.add("\c\L") line.add("\c\L")
template raiseSockError(): stmt {.dirty, immediate.} =
let lastError = osLastError()
if flags.isDisconnectionError(lastError): setLen(line.string, 0); return
socket.socketError(n, lastError = lastError)
var waited = 0.0 var waited = 0.0
setLen(line.string, 0) setLen(line.string, 0)
@ -946,14 +984,14 @@ proc readLine*(socket: PSocket, line: var TaintedString, timeout = -1) {.
var c: char var c: char
discard waitFor(socket, waited, timeout, 1, "readLine") discard waitFor(socket, waited, timeout, 1, "readLine")
var n = recv(socket, addr(c), 1) var n = recv(socket, addr(c), 1)
if n < 0: socket.socketError() if n < 0: raiseSockError()
elif n == 0: return elif n == 0: setLen(line.string, 0); return
if c == '\r': if c == '\r':
discard waitFor(socket, waited, timeout, 1, "readLine") discard waitFor(socket, waited, timeout, 1, "readLine")
n = peekChar(socket, c) n = peekChar(socket, c)
if n > 0 and c == '\L': if n > 0 and c == '\L':
discard recv(socket, addr(c), 1) discard recv(socket, addr(c), 1)
elif n <= 0: socket.socketError() elif n <= 0: raiseSockError()
addNLIfEmpty() addNLIfEmpty()
return return
elif c == '\L': elif c == '\L':
@ -1021,11 +1059,14 @@ proc send*(socket: PSocket, data: pointer, size: int): int {.
const MSG_NOSIGNAL = 0 const MSG_NOSIGNAL = 0
result = send(socket.fd, data, size, int32(MSG_NOSIGNAL)) result = send(socket.fd, data, size, int32(MSG_NOSIGNAL))
proc send*(socket: PSocket, data: string) {.tags: [FWriteIO].} = proc send*(socket: PSocket, data: string,
flags = {TSocketFlags.SafeDisconn}) {.tags: [FWriteIO].} =
## sends data to a socket. ## sends data to a socket.
let sent = send(socket, cstring(data), data.len) let sent = send(socket, cstring(data), data.len)
if sent < 0: if sent < 0:
socketError(socket) let lastError = osLastError()
if flags.isDisconnectionError(lastError): return
socketError(socket, lastError = lastError)
if sent != data.len: if sent != data.len:
raise newException(EOS, "Could not send all data.") raise newException(EOS, "Could not send all data.")

5
lib/pure/rawsockets.nim
View file

@ -21,11 +21,12 @@ const useWinVersion = defined(Windows) or defined(nimdoc)
when useWinVersion: when useWinVersion:
import winlean import winlean
export WSAEWOULDBLOCK export WSAEWOULDBLOCK, WSAECONNRESET, WSAECONNABORTED, WSAENETRESET,
WSAEDISCON
else: else:
import posix import posix
export fcntl, F_GETFL, O_NONBLOCK, F_SETFL, EAGAIN, EWOULDBLOCK, MSG_NOSIGNAL, export fcntl, F_GETFL, O_NONBLOCK, F_SETFL, EAGAIN, EWOULDBLOCK, MSG_NOSIGNAL,
EINTR, EINPROGRESS EINTR, EINPROGRESS, ECONNRESET, EPIPE, ENETRESET
export TSocketHandle, TSockaddr_in, TAddrinfo, INADDR_ANY, TSockAddr, TSockLen, export TSocketHandle, TSockaddr_in, TAddrinfo, INADDR_ANY, TSockAddr, TSockLen,
inet_ntoa, recv, `==`, connect, send, accept, recvfrom, sendto inet_ntoa, recv, `==`, connect, send, accept, recvfrom, sendto