codex-storage
/
nim-chronos
mirror of https://github.com/codex-storage/nim-chronos.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix cast[pointer] issues on nim-devel. (#381) 

* Fix cast[pointer] issues on nim-devel.

* More cast[T] fixes.

* Fix compilation error.

* Add oserrno.nim
Further fixes for Windows and Linux.

* MacOS fixes.

* More Windows fixes and attempt to fix 1.2, 1.4 branches.

* Implicitly import/export oserrno.

* Replace oserrno with osdefs.

* Return back oserrno.

* epoll to oserrno.

* datagram/stream to oserrno.

* common to oserrno.

* test to oserrno.
This commit is contained in:
Eugene Kabanov 2023-04-30 09:20:08 +03:00 committed by GitHub
parent e05d2f8e96
commit 3118f8c1b2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 1690 additions and 294 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
chronos/asyncloop.nim
View File

@ -152,14 +152,15 @@ elif defined(macosx) or defined(freebsd) or defined(netbsd) or
defined(openbsd) or defined(dragonfly) or defined(macos) or
defined(linux) or defined(android) or defined(solaris):
import "."/selectors2
from posix import EINTR, EAGAIN, EINPROGRESS, EWOULDBLOCK, MSG_PEEK,
MSG_NOSIGNAL,
import "."/oserrno
from posix import MSG_PEEK, MSG_NOSIGNAL,
SIGHUP, SIGINT, SIGQUIT, SIGILL, SIGTRAP, SIGABRT,
SIGBUS, SIGFPE, SIGKILL, SIGUSR1, SIGSEGV, SIGUSR2,
SIGPIPE, SIGALRM, SIGTERM, SIGPIPE
export SIGHUP, SIGINT, SIGQUIT, SIGILL, SIGTRAP, SIGABRT,
SIGBUS, SIGFPE, SIGKILL, SIGUSR1, SIGSEGV, SIGUSR2,
SIGPIPE, SIGALRM, SIGTERM, SIGPIPE
export oserrno
type
CallbackFunc* = proc (arg: pointer) {.gcsafe, raises: [Defect].}
@ -282,6 +283,12 @@ proc raiseOsDefect*(error: OSErrorCode, msg = "") {.noreturn, noinline.} =
raise (ref Defect)(msg: msg & "\n[" & $int(error) & "] " & osErrorMsg(error) &
"\n" & getStackTrace())
func toPointer(error: OSErrorCode): pointer =
when sizeof(int) == 8:
cast[pointer](uint64(uint32(error)))
else:
cast[pointer](uint32(error))
func toException*(v: OSErrorCode): ref OSError = newOSError(v)
# This helper will allow to use `tryGet()` and raise OSError for
# Result[T, OSErrorCode] values.
@ -518,27 +525,30 @@ when defined(windows):
## Closes a socket and ensures that it is unregistered.
let loop = getThreadDispatcher()
loop.handles.excl(fd)
let param =
if closeFd(SocketHandle(fd)) == 0:
OSErrorCode(0)
else:
osLastError()
if not isNil(aftercb):
var acb = AsyncCallback(function: aftercb, udata: cast[pointer](param))
loop.callbacks.addLast(acb)
let
param = toPointer(
if closeFd(SocketHandle(fd)) == 0:
OSErrorCode(0)
else:
osLastError()
)
if not(isNil(aftercb)):
loop.callbacks.addLast(AsyncCallback(function: aftercb, udata: param))
proc closeHandle*(fd: AsyncFD, aftercb: CallbackFunc = nil) =
## Closes a (pipe/file) handle and ensures that it is unregistered.
let loop = getThreadDispatcher()
loop.handles.excl(fd)
let param =
if closeFd(HANDLE(fd)) == 0:
OSErrorCode(0)
else:
osLastError()
if not isNil(aftercb):
var acb = AsyncCallback(function: aftercb, udata: cast[pointer](param))
loop.callbacks.addLast(acb)
let
param = toPointer(
if closeFd(HANDLE(fd)) == 0:
OSErrorCode(0)
else:
osLastError()
)
if not(isNil(aftercb)):
loop.callbacks.addLast(AsyncCallback(function: aftercb, udata: param))
proc contains*(disp: PDispatcher, fd: AsyncFD): bool =
## Returns ``true`` if ``fd`` is registered in thread's dispatcher.
@ -720,21 +730,22 @@ elif defined(macosx) or defined(freebsd) or defined(netbsd) or
let loop = getThreadDispatcher()
proc continuation(udata: pointer) =
let param =
if SocketHandle(fd) in loop.selector:
let ures = unregister2(fd)
if ures.isErr():
discard closeFd(cint(fd))
ures.error()
else:
if closeFd(cint(fd)) != 0:
osLastError()
let
param = toPointer(
if SocketHandle(fd) in loop.selector:
let ures = unregister2(fd)
if ures.isErr():
discard closeFd(cint(fd))
ures.error()
else:
OSErrorCode(0)
else:
OSErrorCode(osdefs.EBADF)
if not isNil(aftercb):
aftercb(cast[pointer](param))
if closeFd(cint(fd)) != 0:
osLastError()
else:
OSErrorCode(0)
else:
OSErrorCode(osdefs.EBADF)
)
if not(isNil(aftercb)): aftercb(param)
withData(loop.selector, cint(fd), adata) do:
# We are scheduling reader and writer callbacks to be called

11
chronos/ioselects/ioselectors_epoll.nim
View File

@ -42,7 +42,7 @@ proc getVirtualId[T](s: Selector[T]): SelectResult[int32] =
ok(s.virtualHoles.popLast())
else:
if s.virtualId == low(int32):
err(OSErrorCode(EMFILE))
err(oserrno.EMFILE)
else:
dec(s.virtualId)
ok(s.virtualId)
@ -139,7 +139,7 @@ proc trigger2*(event: SelectEvent): SelectResult[void] =
if res == -1:
err(osLastError())
elif res != sizeof(uint64):
err(OSErrorCode(osdefs.EINVAL))
err(oserrno.EINVAL)
else:
ok()
@ -521,11 +521,11 @@ proc prepareKey[T](s: Selector[T], event: EpollEvent): Opt[ReadyKey] =
if (event.events and EPOLLERR) != 0:
rkey.events.incl(Event.Error)
rkey.errorCode = OSErrorCode(ECONNRESET)
rkey.errorCode = oserrno.ECONNRESET
if (event.events and EPOLLHUP) != 0 or (event.events and EPOLLRDHUP) != 0:
rkey.events.incl(Event.Error)
rkey.errorCode = OSErrorCode(ECONNRESET)
rkey.errorCode = oserrno.ECONNRESET
if (event.events and EPOLLOUT) != 0:
rkey.events.incl(Event.Write)
@ -580,7 +580,8 @@ proc prepareKey[T](s: Selector[T], event: EpollEvent): Opt[ReadyKey] =
let res = handleEintr(osdefs.read(fdi32, addr data, sizeof(uint64)))
if res != sizeof(uint64):
let errorCode = osLastError()
if errorCode == EAGAIN:
case errorCode
of oserrno.EAGAIN:
return Opt.none(ReadyKey)
else:
rkey.events.incl({Event.User, Event.Error})

20
chronos/ioselects/ioselectors_kqueue.nim
View File

@ -58,6 +58,12 @@ proc toString(key: int32|cint|SocketHandle|int): string =
else:
Base10.toString(uint32(fdi32))
proc toPointer(data: int32): pointer =
when sizeof(int) == 8:
cast[pointer](uint64(uint32(data)))
else:
cast[pointer](uint32(data))
template addKey[T](s: Selector[T], key: int32, skey: SelectorKey[T]) =
if s.fds.hasKeyOrPut(key, skey):
raiseAssert "Descriptor [" & key.toString() &
@ -154,7 +160,7 @@ proc trigger2*(event: SelectEvent): SelectResult[void] =
if res == -1:
err(osLastError())
elif res != sizeof(uint64):
err(OSErrorCode(osdefs.EINVAL))
err(oserrno.EINVAL)
else:
ok()
@ -310,7 +316,7 @@ proc registerSignal*[T](s: Selector[T], signal: int,
# To be compatible with linux semantic we need to "eat" signals
signal(cint(signal), SIG_IGN)
changes.modifyKQueue(0, uint(signal), EVFILT_SIGNAL, EV_ADD, 0, 0,
cast[pointer](uint32(fdi32)))
fdi32.toPointer())
if handleEintr(kevent(s.kqFd, addr(changes[0]), cint(1), nil, 0, nil)) == -1:
let errorCode = osLastError()
s.freeKey(fdi32)
@ -341,7 +347,7 @@ proc registerProcess*[T](s: Selector[T], pid: int,
s.addKey(fdi32, selectorKey)
changes.modifyKQueue(0, uint(uint32(pid)), EVFILT_PROC, flags, NOTE_EXIT,
0, cast[pointer](uint32(fdi32)))
0, fdi32.toPointer())
if handleEintr(kevent(s.kqFd, addr(changes[0]), cint(1), nil, 0, nil)) == -1:
s.freeKey(fdi32)
return err(osLastError())
@ -490,14 +496,14 @@ proc prepareKey[T](s: Selector[T], event: KEvent): Opt[ReadyKey] =
of EVFILT_READ:
if (event.flags and EV_EOF) != 0:
rkey.events.incl(Event.Error)
rkey.errorCode = OSErrorCode(ECONNRESET)
rkey.errorCode = oserrno.ECONNRESET
if Event.User in pkey.events:
var data: uint64 = 0
if handleEintr(osdefs.read(cint(event.ident), addr data,
sizeof(uint64))) != sizeof(uint64):
let errorCode = osLastError()
if errorCode == EAGAIN:
if errorCode == oserrno.EAGAIN:
# Someone already consumed event data
return Opt.none(ReadyKey)
else:
@ -510,7 +516,7 @@ proc prepareKey[T](s: Selector[T], event: KEvent): Opt[ReadyKey] =
of EVFILT_WRITE:
if (event.flags and EV_EOF) != 0:
rkey.events.incl(Event.Error)
rkey.errorCode = OSErrorCode(ECONNRESET)
rkey.errorCode = oserrno.ECONNRESET
rkey.events.incl(Event.Write)
@ -577,7 +583,7 @@ proc selectInto2*[T](s: Selector[T], timeout: int,
maxEventsCount, ptrTimeout)
if res < 0:
let errorCode = osLastError()
if errorCode == EINTR:
if errorCode == oserrno.EINTR:
continue
return err(errorCode)
else:

87
chronos/osdefs.nim
View File

@ -6,18 +6,10 @@
# Licensed under either of
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
from std/os import osLastError, osErrorMsg, OSErrorCode, raiseOSError,
newOSError
export osLastError, osErrorMsg, OSError, OSErrorCode, raiseOSError, newOSError
import oserrno
export oserrno
when defined(windows):
from std/winlean import SocketHandle, SockLen, SockAddr, InAddr,
In6_addr, Sockaddr_in, Sockaddr_in6, Sockaddr_storage,
AddrInfo
export SocketHandle, SockLen, SockAddr, InAddr,
In6_addr, Sockaddr_in, Sockaddr_in6, Sockaddr_storage, AddrInfo
# Prerequisites for constants
template WSAIORW*(x, y): untyped = (IOC_INOUT or x or y)
template WSAIOW*(x, y): untyped =
@ -25,6 +17,49 @@ when defined(windows):
((clong(sizeof(int32)) and clong(IOCPARM_MASK)) shl 16) or (x shl 8) or y
type
Sockaddr_storage* {.final, pure.} = object
ss_family*: uint16
ss_pad1: array[6, byte]
ss_align: int64
ss_pad2: array[112, byte]
InAddr* {.final, pure, union.} = object
s_addr*: uint32
In6Addr* {.final, pure, union.} = object
s_addr*: array[16, byte]
Sockaddr_in* {.final, pure.} = object
sin_family*: uint16
sin_port*: uint16
sin_addr*: InAddr
sin_zero*: array[0..7, char]
Sockaddr_in6* {.final, pure.} = object
sin6_family*: uint16
sin6_port*: uint16
sin6_flowinfo*: uint32
sin6_addr*: In6Addr
sin6_scope_id*: uint32
SockLen* = cuint
SockAddr* {.final, pure.} = object
sa_family*: uint16
sa_data*: array[14, char]
AddrInfo* {.final, pure.} = object
ai_flags*: cint ## Input flags.
ai_family*: cint ## Address family of socket.
ai_socktype*: cint ## Socket type.
ai_protocol*: cint ## Protocol of socket.
ai_addrlen*: csize_t ## Length of socket address.
ai_canonname*: pointer ## Canonical name of service location.
ai_addr*: ptr SockAddr ## Socket address of socket.
ai_next*: ptr AddrInfo ## Pointer to next in list.
SocketHandle* = distinct int
HANDLE* = distinct uint
GUID* {.final, pure.} = object
D1*: uint32
@ -104,36 +139,6 @@ when defined(windows):
PIPE_UNLIMITED_INSTANCES* = 255'u32
DEFAULT_PIPE_SIZE* = 65536'u32
ERROR_SUCCESS* = 0
ERROR_FILE_NOT_FOUND* = 2
ERROR_TOO_MANY_OPEN_FILES* = 4
ERROR_ACCESS_DENIED* = 5
ERROR_BROKEN_PIPE* = 109
ERROR_BUFFER_OVERFLOW* = 111
ERROR_PIPE_BUSY* = 231
ERROR_NO_DATA* = 232
ERROR_PIPE_NOT_CONNECTED* = 233
ERROR_PIPE_CONNECTED* = 535
ERROR_OPERATION_ABORTED* = 995
ERROR_IO_PENDING* = 997
ERROR_CONNECTION_REFUSED* = 1225
ERROR_CONNECTION_ABORTED* = 1236
WSAEMFILE* = 10024
WSAENETDOWN* = 10050
WSAENETRESET* = 10052
WSAECONNABORTED* = 10053
WSAECONNRESET* = 10054
WSAENOBUFS* = 10055
WSAETIMEDOUT* = 10060
WSAEADDRINUSE* = 10048
WSAEDISCON* = 10101
WSANOTINITIALISED* = 10093
WSAENOTSOCK* = 10038
WSAEINPROGRESS* = 10036
WSAEINTR* = 10004
WSAEWOULDBLOCK* = 10035
ERROR_NETNAME_DELETED* = 64
STATUS_PENDING* = 0x103
IOCPARM_MASK* = 0x7f'u32
@ -1283,8 +1288,6 @@ when defined(posix):
INVALID_SOCKET* = SocketHandle(-1)
INVALID_HANDLE_VALUE* = cint(-1)
proc `==`*(x: OSErrorCode, y: int): bool =
int(x) == y
proc `==`*(x: SocketHandle, y: int): bool =
x == SocketHandle(y)

1346
chronos/oserrno.nim Normal file
View File

File diff suppressed because it is too large Load Diff

12
chronos/osutils.nim
View File

@ -7,9 +7,9 @@
# Apache License, version 2.0, (LICENSE-APACHEv2)
# MIT license (LICENSE-MIT)
import stew/results
import osdefs
import osdefs, oserrno
export results
export results, osdefs, oserrno
when (NimMajor, NimMinor) < (1, 4):
{.push raises: [Defect].}
@ -187,10 +187,10 @@ when defined(windows):
let cleanupFlag =
block:
let errorCode = osLastError()
case int(errorCode)
of osdefs.ERROR_PIPE_CONNECTED:
case errorCode
of ERROR_PIPE_CONNECTED:
false
of osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
if DescriptorFlag.NonBlock in writeset:
var bytesRead = 0.DWORD
if getOverlappedResult(pipeIn, addr ovl, bytesRead, 1) == FALSE:
@ -215,7 +215,7 @@ else:
var res = 0
while true:
res = body
if not((res == -1) and (osLastError() == EINTR)):
if not((res == -1) and (osLastError() == oserrno.EINTR)):
break
res

4
chronos/selectors2.nim
View File

@ -32,8 +32,8 @@
# backwards-compatible.
import stew/results
import osdefs, osutils
export results
import osdefs, osutils, oserrno
export results, oserrno
const
asyncEventsCount* {.intdefine.} = 64

67
chronos/transports/common.nim
View File

@ -12,10 +12,17 @@ when (NimMajor, NimMinor) < (1, 4):
else:
{.push raises: [].}
import std/[strutils, nativesockets, net]
import std/[strutils]
import stew/[base10, byteutils]
import ".."/[asyncloop, osdefs]
export net
import ".."/[asyncloop, osdefs, oserrno]
from std/net import Domain, `==`, IpAddress, IpAddressFamily, parseIpAddress,
SockType, Protocol, Port, `$`
from std/nativesockets import toInt, `$`
export Domain, `==`, IpAddress, IpAddressFamily, parseIpAddress, SockType,
Protocol, Port, toInt, `$`
const
DefaultStreamBufferSize* = 4096 ## Default buffer size for stream
@ -600,35 +607,37 @@ proc isLiteral*[T](s: seq[T]): bool {.inline.} =
else:
(cast[ptr SeqHeader](s).reserved and (1 shl (sizeof(int) * 8 - 2))) != 0
template getTransportTooManyError*(code: int = 0): ref TransportTooManyError =
template getTransportTooManyError*(
code = OSErrorCode(0)
): ref TransportTooManyError =
let msg =
when defined(posix):
if code == 0:
if code == OSErrorCode(0):
"Too many open transports"
elif code == EMFILE:
elif code == oserrno.EMFILE:
"[EMFILE] Too many open files in the process"
elif code == ENFILE:
elif code == oserrno.ENFILE:
"[ENFILE] Too many open files in system"
elif code == ENOBUFS:
elif code == oserrno.ENOBUFS:
"[ENOBUFS] No buffer space available"
elif code == ENOMEM:
elif code == oserrno.ENOMEM:
"[ENOMEM] Not enough memory availble"
else:
"[" & $code & "] Too many open transports"
"[" & $int(code) & "] Too many open transports"
elif defined(windows):
case code
of 0:
of OSErrorCode(0):
"Too many open transports"
of osdefs.ERROR_TOO_MANY_OPEN_FILES:
of ERROR_TOO_MANY_OPEN_FILES:
"[ERROR_TOO_MANY_OPEN_FILES] Too many open files"
of osdefs.WSAENOBUFS:
of WSAENOBUFS:
"[WSAENOBUFS] No buffer space available"
of osdefs.WSAEMFILE:
of WSAEMFILE:
"[WSAEMFILE] Too many open sockets"
else:
"[" & $code & "] Too many open transports"
"[" & $int(code) & "] Too many open transports"
else:
"[" & $code & "] Too many open transports"
"[" & $int(code) & "] Too many open transports"
newException(TransportTooManyError, msg)
template getConnectionAbortedError*(m: string = ""): ref TransportAbortedError =
@ -639,32 +648,34 @@ template getConnectionAbortedError*(m: string = ""): ref TransportAbortedError =
"[ECONNABORTED] " & m
newException(TransportAbortedError, msg)
template getConnectionAbortedError*(code: int): ref TransportAbortedError =
template getConnectionAbortedError*(
code: OSErrorCode
): ref TransportAbortedError =
let msg =
when defined(posix):
if code == 0:
if code == OSErrorCode(0):
"[ECONNABORTED] Connection has been aborted before being accepted"
elif code == EPERM:
elif code == oserrno.EPERM:
"[EPERM] Firewall rules forbid connection"
elif code == ETIMEDOUT:
elif code == oserrno.ETIMEDOUT:
"[ETIMEDOUT] Operation has been timed out"
else:
"[" & $code & "] Connection has been aborted"
"[" & $int(code) & "] Connection has been aborted"
elif defined(windows):
case code
of 0, osdefs.WSAECONNABORTED:
of OSErrorCode(0), oserrno.WSAECONNABORTED:
"[ECONNABORTED] Connection has been aborted before being accepted"
of osdefs.WSAENETDOWN:
of WSAENETDOWN:
"[ENETDOWN] Network is down"
of osdefs.WSAENETRESET:
of oserrno.WSAENETRESET:
"[ENETRESET] Network dropped connection on reset"
of osdefs.WSAECONNRESET:
of oserrno.WSAECONNRESET:
"[ECONNRESET] Connection reset by peer"
of osdefs.WSAETIMEDOUT:
of WSAETIMEDOUT:
"[ETIMEDOUT] Connection timed out"
else:
"[" & $code & "] Connection has been aborted"
"[" & $int(code) & "] Connection has been aborted"
else:
"[" & $code & "] Connection has been aborted"
"[" & $int(code) & "] Connection has been aborted"
newException(TransportAbortedError, msg)

30
chronos/transports/datagram.nim
View File

@ -14,7 +14,7 @@ else:
import std/deques
when not(defined(windows)): import ".."/selectors2
import ".."/[asyncloop, osdefs, handles]
import ".."/[asyncloop, osdefs, oserrno, handles]
import "."/common
type
@ -139,10 +139,11 @@ when defined(windows):
transp.state.excl(WritePending)
let err = transp.wovl.data.errCode
let vector = transp.queue.popFirst()
if err == OSErrorCode(-1):
case err
of OSErrorCode(-1):
if not(vector.writer.finished()):
vector.writer.complete()
elif int(err) == osdefs.ERROR_OPERATION_ABORTED:
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.incl(WritePaused)
if not(vector.writer.finished()):
@ -170,13 +171,14 @@ when defined(windows):
DWORD(0), cast[POVERLAPPED](addr transp.wovl), nil)
if ret != 0:
let err = osLastError()
if int(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.excl(WritePending)
transp.state.incl(WritePaused)
if not(vector.writer.finished()):
vector.writer.complete()
elif int(err) == osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
transp.queue.addFirst(vector)
else:
transp.state.excl(WritePending)
@ -201,14 +203,15 @@ when defined(windows):
## Continuation
transp.state.excl(ReadPending)
let err = transp.rovl.data.errCode
if err == OSErrorCode(-1):
case err
of OSErrorCode(-1):
let bytesCount = transp.rovl.data.bytesCount
if bytesCount == 0:
transp.state.incl({ReadEof, ReadPaused})
fromSAddr(addr transp.raddr, transp.ralen, raddr)
transp.buflen = int(bytesCount)
asyncSpawn transp.function(transp, raddr)
elif int(err) == osdefs.ERROR_OPERATION_ABORTED:
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt or closeSocket() call.
transp.state.incl(ReadPaused)
if ReadClosed in transp.state and not(transp.future.finished()):
@ -237,15 +240,16 @@ when defined(windows):
cast[POVERLAPPED](addr transp.rovl), nil)
if ret != 0:
let err = osLastError()
if int(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.excl(ReadPending)
transp.state.incl(ReadPaused)
elif int(err) == osdefs.WSAECONNRESET:
of WSAECONNRESET:
transp.state.excl(ReadPending)
transp.state.incl({ReadPaused, ReadEof})
break
elif int(err) == osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
discard
else:
transp.state.excl(ReadPending)
@ -419,7 +423,8 @@ else:
asyncSpawn transp.function(transp, raddr)
else:
let err = osLastError()
if int(err) == EINTR:
case err
of oserrno.EINTR:
continue
else:
transp.buflen = 0
@ -454,7 +459,8 @@ else:
vector.writer.complete()
else:
let err = osLastError()
if int(err) == EINTR:
case err
of oserrno.EINTR:
continue
else:
if not(vector.writer.finished()):

13
chronos/transports/osnet.nim
View File

@ -838,17 +838,15 @@ elif defined(macosx) or defined(macos) or defined(bsd):
if family == osdefs.AF_INET:
fromSAddr(cast[ptr Sockaddr_storage](ifap.ifa_netmask),
SockLen(sizeof(Sockaddr_in)), na)
if cint(ifaddress.host.family) == osdefs.AF_INET:
if ifaddress.host.family == AddressFamily.IPv4:
ifaddress.net = IpNet.init(ifaddress.host, na)
elif family == osdefs.AF_INET6:
fromSAddr(cast[ptr Sockaddr_storage](ifap.ifa_netmask),
SockLen(sizeof(Sockaddr_in6)), na)
if cint(ifaddress.host.family) == osdefs.AF_INET6:
if ifaddress.host.family == AddressFamily.IPv6:
ifaddress.net = IpNet.init(ifaddress.host, na)
if ifaddress.host.family != AddressFamily.None:
if len(res[i].addresses) == 0:
res[i].addresses = newSeq[InterfaceAddress]()
res[i].addresses.add(ifaddress)
ifap = ifap.ifa_next
@ -1047,10 +1045,11 @@ elif defined(windows):
var addresses = cast[ptr IpAdapterAddressesXp](addr buffer[0])
gres = getAdaptersAddresses(osdefs.AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX,
nil, addresses, addr size)
if gres == ERROR_SUCCESS:
case OSErrorCode(gres)
of ERROR_SUCCESS:
buffer.setLen(size)
break
elif gres == ERROR_BUFFER_OVERFLOW:
of ERROR_BUFFER_OVERFLOW:
discard
else:
break
@ -1058,7 +1057,7 @@ elif defined(windows):
if tries >= MaxTries:
break
if gres == ERROR_SUCCESS:
if OSErrorCode(gres) == ERROR_SUCCESS:
var slider = cast[ptr IpAdapterAddressesXp](addr buffer[0])
while not isNil(slider):
var iface = NetworkInterface(

310
chronos/transports/stream.nim
View File

@ -13,7 +13,7 @@ else:
{.push raises: [].}
import std/deques
import ".."/[asyncloop, handles, osdefs, osutils]
import ".."/[asyncloop, handles, osdefs, osutils, oserrno]
import common
type
@ -322,10 +322,12 @@ when defined(windows):
(t).wwsabuf.buf = cast[cstring](v.buf)
(t).wwsabuf.len = cast[ULONG](v.buflen)
proc isConnResetError(err: OSErrorCode): bool {.inline.} =
result = (err == OSErrorCode(osdefs.WSAECONNRESET)) or
(err == OSErrorCode(osdefs.WSAECONNABORTED)) or
(err == OSErrorCode(osdefs.ERROR_PIPE_NOT_CONNECTED))
func isConnResetError(err: OSErrorCode): bool {.inline.} =
case err
of WSAECONNRESET, WSAECONNABORTED, ERROR_PIPE_NOT_CONNECTED:
true
else:
false
proc writeStreamLoop(udata: pointer) {.gcsafe, nimcall.} =
var bytesCount: uint32
@ -343,7 +345,8 @@ when defined(windows):
## Continuation
transp.state.excl(WritePending)
let err = transp.wovl.data.errCode
if err == OSErrorCode(-1):
case err
of OSErrorCode(-1):
bytesCount = transp.wovl.data.bytesCount
var vector = transp.queue.popFirst()
if bytesCount == 0:
@ -377,7 +380,7 @@ when defined(windows):
# This conversion to `int` safe, because its impossible
# to call write() with size bigger than `int`.
vector.writer.complete(int(transp.wwsabuf.len))
elif int(err) == osdefs.ERROR_OPERATION_ABORTED:
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.incl({WritePaused, WriteEof})
let vector = transp.queue.popFirst()
@ -416,7 +419,8 @@ when defined(windows):
cast[POVERLAPPED](addr transp.wovl), nil)
if ret != 0:
let err = osLastError()
if int(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.excl(WritePending)
transp.state.incl({WritePaused, WriteEof})
@ -424,7 +428,7 @@ when defined(windows):
vector.writer.complete(0)
completePendingWriteQueue(transp.queue, 0)
break
elif int(err) == osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
transp.queue.addFirst(vector)
else:
transp.state.excl(WritePending)
@ -456,7 +460,8 @@ when defined(windows):
nil, 0'u32)
if ret == 0:
let err = osLastError()
if int(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.excl(WritePending)
transp.state.incl({WritePaused, WriteEof})
@ -464,7 +469,7 @@ when defined(windows):
vector.writer.complete(0)
completePendingWriteQueue(transp.queue, 0)
break
elif int(err) == osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
transp.queue.addFirst(vector)
else:
transp.state.excl(WritePending)
@ -496,8 +501,8 @@ when defined(windows):
cast[POVERLAPPED](addr transp.wovl))
if ret == 0:
let err = osLastError()
if int(err) in {osdefs.ERROR_OPERATION_ABORTED,
osdefs.ERROR_NO_DATA}:
case err
of ERROR_OPERATION_ABORTED, ERROR_NO_DATA:
# CancelIO() interrupt
transp.state.excl(WritePending)
transp.state.incl({WritePaused, WriteEof})
@ -505,7 +510,7 @@ when defined(windows):
vector.writer.complete(0)
completePendingWriteQueue(transp.queue, 0)
break
elif int(err) == osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
transp.queue.addFirst(vector)
else:
transp.state.excl(WritePending)
@ -539,7 +544,8 @@ when defined(windows):
## Continuation
transp.state.excl(ReadPending)
let err = transp.rovl.data.errCode
if err == OSErrorCode(-1):
case err
of OSErrorCode(-1):
let bytesCount = transp.rovl.data.bytesCount
if bytesCount == 0:
transp.state.incl({ReadEof, ReadPaused})
@ -552,19 +558,20 @@ when defined(windows):
transp.roffset = transp.offset
if transp.offset == len(transp.buffer):
transp.state.incl(ReadPaused)
elif int(err) in {osdefs.ERROR_OPERATION_ABORTED,
osdefs.ERROR_CONNECTION_ABORTED,
osdefs.ERROR_BROKEN_PIPE,
osdefs.ERROR_NETNAME_DELETED}:
of ERROR_OPERATION_ABORTED, ERROR_CONNECTION_ABORTED,
ERROR_BROKEN_PIPE:
# CancelIO() interrupt or closeSocket() call.
transp.state.incl(ReadPaused)
elif transp.kind == TransportKind.Socket and
(int(err) in {osdefs.ERROR_NETNAME_DELETED,
osdefs.WSAECONNABORTED}):
transp.state.incl({ReadEof, ReadPaused})
elif transp.kind == TransportKind.Pipe and
(int(err) in {osdefs.ERROR_PIPE_NOT_CONNECTED}):
transp.state.incl({ReadEof, ReadPaused})
of ERROR_NETNAME_DELETED, WSAECONNABORTED:
if transp.kind == TransportKind.Socket:
transp.state.incl({ReadEof, ReadPaused})
else:
transp.setReadError(err)
of ERROR_PIPE_NOT_CONNECTED:
if transp.kind == TransportKind.Pipe:
transp.state.incl({ReadEof, ReadPaused})
else:
transp.setReadError(err)
else:
transp.setReadError(err)
@ -593,16 +600,18 @@ when defined(windows):
cast[POVERLAPPED](addr transp.rovl), nil)
if ret != 0:
let err = osLastError()
if int32(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.excl(ReadPending)
transp.state.incl(ReadPaused)
elif int32(err) in {osdefs.WSAECONNRESET, osdefs.WSAENETRESET,
osdefs.WSAECONNABORTED}:
of WSAECONNRESET, WSAENETRESET, WSAECONNABORTED:
transp.state.excl(ReadPending)
transp.state.incl({ReadEof, ReadPaused})
transp.completeReader()
elif int32(err) != osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
discard
else:
transp.state.excl(ReadPending)
transp.state.incl(ReadPaused)
transp.setReadError(err)
@ -616,16 +625,18 @@ when defined(windows):
cast[POVERLAPPED](addr transp.rovl))
if ret == 0:
let err = osLastError()
if int32(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt
transp.state.excl(ReadPending)
transp.state.incl(ReadPaused)
elif int32(err) in {osdefs.ERROR_BROKEN_PIPE,
osdefs.ERROR_PIPE_NOT_CONNECTED}:
of ERROR_BROKEN_PIPE, ERROR_PIPE_NOT_CONNECTED:
transp.state.excl(ReadPending)
transp.state.incl({ReadEof, ReadPaused})
transp.completeReader()
elif int32(err) != osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
discard
else:
transp.state.excl(ReadPending)
transp.state.incl(ReadPaused)
transp.setReadError(err)
@ -769,8 +780,8 @@ when defined(windows):
var ovl = cast[RefCustomOverlapped](udata)
if not(retFuture.finished()):
if ovl.data.errCode == OSErrorCode(-1):
if setsockopt(SocketHandle(sock), cint(osdefs.SOL_SOCKET),
cint(osdefs.SO_UPDATE_CONNECT_CONTEXT), nil,
if setsockopt(SocketHandle(sock), cint(SOL_SOCKET),
cint(SO_UPDATE_CONNECT_CONTEXT), nil,
SockLen(0)) != 0'i32:
let err = wsaGetLastError()
sock.closeSocket()
@ -796,9 +807,12 @@ when defined(windows):
cint(slen), nil, 0, nil,
cast[POVERLAPPED](povl))
# We will not process immediate completion, to avoid undefined behavior.
if res == osdefs.FALSE:
if res == FALSE:
let err = osLastError()
if int32(err) != osdefs.ERROR_IO_PENDING:
case err
of ERROR_IO_PENDING:
discard
else:
GC_unref(povl)
sock.closeSocket()
retFuture.fail(getTransportOsError(err))
@ -809,7 +823,7 @@ when defined(windows):
## Unix domain socket emulation with Windows Named Pipes.
# For some reason Nim compiler does not detect `pipeHandle` usage in
# pipeContinuation() procedure, so we marking it as {.used.} here.
var pipeHandle {.used.} = osdefs.INVALID_HANDLE_VALUE
var pipeHandle {.used.} = INVALID_HANDLE_VALUE
var pipeContinuation: proc (udata: pointer) {.gcsafe, raises: [Defect].}
pipeContinuation = proc (udata: pointer) {.gcsafe, raises: [Defect].} =
@ -821,15 +835,16 @@ when defined(windows):
pipeName = toWideString(pipeAsciiName).valueOr:
retFuture.fail(getTransportOsError(error))
return
genericFlags = osdefs.GENERIC_READ or osdefs.GENERIC_WRITE
shareFlags = osdefs.FILE_SHARE_READ or osdefs.FILE_SHARE_WRITE
genericFlags = GENERIC_READ or GENERIC_WRITE
shareFlags = FILE_SHARE_READ or FILE_SHARE_WRITE
pipeHandle = createFile(pipeName, genericFlags, shareFlags,
nil, osdefs.OPEN_EXISTING,
osdefs.FILE_FLAG_OVERLAPPED, HANDLE(0))
nil, OPEN_EXISTING,
FILE_FLAG_OVERLAPPED, HANDLE(0))
free(pipeName)
if pipeHandle == osdefs.INVALID_HANDLE_VALUE:
if pipeHandle == INVALID_HANDLE_VALUE:
let err = osLastError()
if int32(err) == osdefs.ERROR_PIPE_BUSY:
case err
of ERROR_PIPE_BUSY:
discard setTimer(Moment.fromNow(50.milliseconds),
pipeContinuation, nil)
else:
@ -856,23 +871,22 @@ when defined(windows):
openMode =
if FirstPipe notin server.flags:
server.flags.incl(FirstPipe)
osdefs.PIPE_ACCESS_DUPLEX or osdefs.FILE_FLAG_OVERLAPPED or
osdefs.FILE_FLAG_FIRST_PIPE_INSTANCE
PIPE_ACCESS_DUPLEX or FILE_FLAG_OVERLAPPED or
FILE_FLAG_FIRST_PIPE_INSTANCE
else:
osdefs.PIPE_ACCESS_DUPLEX or osdefs.FILE_FLAG_OVERLAPPED
pipeMode = osdefs.PIPE_TYPE_BYTE or osdefs.PIPE_READMODE_BYTE or
osdefs.PIPE_WAIT
PIPE_ACCESS_DUPLEX or FILE_FLAG_OVERLAPPED
pipeMode = PIPE_TYPE_BYTE or PIPE_READMODE_BYTE or PIPE_WAIT
pipeHandle = createNamedPipe(pipeName, openMode, pipeMode,
osdefs.PIPE_UNLIMITED_INSTANCES,
PIPE_UNLIMITED_INSTANCES,
DWORD(server.bufferSize),
DWORD(server.bufferSize),
DWORD(0), nil)
free(pipeName)
if pipeHandle == osdefs.INVALID_HANDLE_VALUE:
if pipeHandle == INVALID_HANDLE_VALUE:
return err(osLastError())
let res = register2(AsyncFD(pipeHandle))
if res.isErr():
discard osdefs.closeHandle(pipeHandle)
discard closeHandle(pipeHandle)
return err(res.error())
ok(AsyncFD(pipeHandle))
@ -886,7 +900,8 @@ when defined(windows):
## Continuation
server.apending = false
if server.status notin {ServerStatus.Stopped, ServerStatus.Closed}:
if ovl.data.errCode == OSErrorCode(-1):
case ovl.data.errCode
of OSErrorCode(-1):
var ntransp: StreamTransport
var flags = {WinServerPipe}
if NoPipeFlash in server.flags:
@ -901,7 +916,7 @@ when defined(windows):
# Start tracking transport
trackStream(ntransp)
asyncSpawn server.function(server, ntransp)
elif int32(ovl.data.errCode) == osdefs.ERROR_OPERATION_ABORTED:
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt or close call.
if server.status in {ServerStatus.Closed, ServerStatus.Stopped}:
server.clean()
@ -931,19 +946,18 @@ when defined(windows):
openMode =
if FirstPipe notin server.flags:
server.flags.incl(FirstPipe)
osdefs.PIPE_ACCESS_DUPLEX or osdefs.FILE_FLAG_OVERLAPPED or
osdefs.FILE_FLAG_FIRST_PIPE_INSTANCE
PIPE_ACCESS_DUPLEX or FILE_FLAG_OVERLAPPED or
FILE_FLAG_FIRST_PIPE_INSTANCE
else:
osdefs.PIPE_ACCESS_DUPLEX or osdefs.FILE_FLAG_OVERLAPPED
pipeMode = osdefs.PIPE_TYPE_BYTE or osdefs.PIPE_READMODE_BYTE or
osdefs.PIPE_WAIT
PIPE_ACCESS_DUPLEX or FILE_FLAG_OVERLAPPED
pipeMode = PIPE_TYPE_BYTE or PIPE_READMODE_BYTE or PIPE_WAIT
pipeHandle = createNamedPipe(pipeName, openMode, pipeMode,
osdefs.PIPE_UNLIMITED_INSTANCES,
PIPE_UNLIMITED_INSTANCES,
DWORD(server.bufferSize),
DWORD(server.bufferSize),
DWORD(0), nil)
free(pipeName)
if pipeHandle == osdefs.INVALID_HANDLE_VALUE:
if pipeHandle == INVALID_HANDLE_VALUE:
raiseOsDefect(osLastError(), "acceptPipeLoop(): Unable to create " &
"new pipe")
server.sock = AsyncFD(pipeHandle)
@ -955,12 +969,12 @@ when defined(windows):
cast[POVERLAPPED](addr server.aovl))
if res == 0:
let errCode = osLastError()
if errCode == osdefs.ERROR_OPERATION_ABORTED:
if errCode == ERROR_OPERATION_ABORTED:
server.apending = false
break
elif errCode == osdefs.ERROR_IO_PENDING:
elif errCode == ERROR_IO_PENDING:
discard
elif errCode == osdefs.ERROR_PIPE_CONNECTED:
elif errCode == ERROR_PIPE_CONNECTED:
discard
else:
raiseOsDefect(errCode, "acceptPipeLoop(): Unable to establish " &
@ -983,9 +997,10 @@ when defined(windows):
## Continuation
server.apending = false
if server.status notin {ServerStatus.Stopped, ServerStatus.Closed}:
if ovl.data.errCode == OSErrorCode(-1):
if setsockopt(SocketHandle(server.asock), cint(osdefs.SOL_SOCKET),
cint(osdefs.SO_UPDATE_ACCEPT_CONTEXT),
case ovl.data.errCode
of OSErrorCode(-1):
if setsockopt(SocketHandle(server.asock), cint(SOL_SOCKET),
cint(SO_UPDATE_ACCEPT_CONTEXT),
addr server.sock,
SockLen(sizeof(SocketHandle))) != 0'i32:
let errCode = OSErrorCode(wsaGetLastError())
@ -1006,7 +1021,7 @@ when defined(windows):
trackStream(ntransp)
asyncSpawn server.function(server, ntransp)
elif int32(ovl.data.errCode) == osdefs.ERROR_OPERATION_ABORTED:
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt or close.
server.asock.closeSocket()
if server.status in {ServerStatus.Closed, ServerStatus.Stopped}:
@ -1049,12 +1064,12 @@ when defined(windows):
dwReceiveDataLength, dwLocalAddressLength,
dwRemoteAddressLength, addr dwBytesReceived,
cast[POVERLAPPED](addr server.aovl))
if res == osdefs.FALSE:
if res == FALSE:
let errCode = osLastError()
if errCode == osdefs.ERROR_OPERATION_ABORTED:
if errCode == ERROR_OPERATION_ABORTED:
server.apending = false
break
elif errCode == osdefs.ERROR_IO_PENDING:
elif errCode == ERROR_IO_PENDING:
discard
else:
raiseOsDefect(errCode, "acceptLoop(): Unable to accept " &
@ -1121,13 +1136,14 @@ when defined(windows):
else:
case ovl.data.errCode
of OSErrorCode(-1):
if setsockopt(SocketHandle(server.asock), cint(osdefs.SOL_SOCKET),
cint(osdefs.SO_UPDATE_ACCEPT_CONTEXT),
if setsockopt(SocketHandle(server.asock), cint(SOL_SOCKET),
cint(SO_UPDATE_ACCEPT_CONTEXT),
addr server.sock,
SockLen(sizeof(SocketHandle))) != 0'i32:
let err = osLastError()
server.asock.closeSocket()
if err == osdefs.WSAENOTSOCK:
case err
of WSAENOTSOCK:
# This can be happened when server get closed, but continuation
# was already scheduled, so we failing it not with OS error.
retFuture.fail(getServerUseClosedError())
@ -1147,18 +1163,15 @@ when defined(windows):
# Start tracking transport
trackStream(ntransp)
retFuture.complete(ntransp)
of OSErrorCode(osdefs.ERROR_OPERATION_ABORTED):
of ERROR_OPERATION_ABORTED:
# CancelIO() interrupt or close.
server.asock.closeSocket()
retFuture.fail(getServerUseClosedError())
server.clean()
of OSErrorCode(osdefs.WSAENETDOWN),
OSErrorCode(osdefs.WSAENETRESET),
OSErrorCode(osdefs.WSAECONNABORTED),
OSErrorCode(osdefs.WSAECONNRESET),
OSErrorCode(osdefs.WSAETIMEDOUT):
of WSAENETDOWN, WSAENETRESET, WSAECONNABORTED, WSAECONNRESET,
WSAETIMEDOUT:
server.asock.closeSocket()
retFuture.fail(getConnectionAbortedError(int(ovl.data.errCode)))
retFuture.fail(getConnectionAbortedError(ovl.data.errCode))
server.clean()
else:
server.asock.closeSocket()
@ -1189,7 +1202,8 @@ when defined(windows):
server.sock.closeHandle()
server.clean()
else:
if ovl.data.errCode == OSErrorCode(-1):
case ovl.data.errCode
of OSErrorCode(-1):
var ntransp: StreamTransport
var flags = {WinServerPipe}
if NoPipeFlash in server.flags:
@ -1210,8 +1224,7 @@ when defined(windows):
trackStream(ntransp)
retFuture.complete(ntransp)
elif int32(ovl.data.errCode) in {osdefs.ERROR_OPERATION_ABORTED,
osdefs.ERROR_PIPE_NOT_CONNECTED}:
of ERROR_OPERATION_ABORTED, ERROR_PIPE_NOT_CONNECTED:
# CancelIO() interrupt or close call.
retFuture.fail(getServerUseClosedError())
server.clean()
@ -1236,10 +1249,9 @@ when defined(windows):
Protocol.IPPROTO_TCP)
if server.asock == asyncInvalidSocket:
let err = osLastError()
case int(err)
of osdefs.ERROR_TOO_MANY_OPEN_FILES,
osdefs.WSAENOBUFS, osdefs.WSAEMFILE:
retFuture.fail(getTransportTooManyError(int(err)))
case err
of ERROR_TOO_MANY_OPEN_FILES, WSAENOBUFS, WSAEMFILE:
retFuture.fail(getTransportTooManyError(err))
else:
retFuture.fail(getTransportOsError(err))
return retFuture
@ -1258,19 +1270,19 @@ when defined(windows):
dwReceiveDataLength, dwLocalAddressLength,
dwRemoteAddressLength, addr dwBytesReceived,
cast[POVERLAPPED](addr server.aovl))
if res == osdefs.FALSE:
if res == FALSE:
let err = osLastError()
case int(err)
of osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
server.apending = false
retFuture.fail(getServerUseClosedError())
return retFuture
of osdefs.ERROR_IO_PENDING:
of ERROR_IO_PENDING:
discard
of osdefs.WSAECONNRESET, osdefs.WSAECONNABORTED, osdefs.WSAENETDOWN,
osdefs.WSAENETRESET, osdefs.WSAETIMEDOUT:
of WSAECONNRESET, WSAECONNABORTED, WSAENETDOWN,
WSAENETRESET, WSAETIMEDOUT:
server.apending = false
retFuture.fail(getConnectionAbortedError(int(err)))
retFuture.fail(getConnectionAbortedError(err))
return retFuture
else:
server.apending = false
@ -1284,7 +1296,8 @@ when defined(windows):
server.apending = true
if server.sock == asyncInvalidPipe:
let err = server.errorCode
if int32(err) == osdefs.ERROR_TOO_MANY_OPEN_FILES:
case err
of ERROR_TOO_MANY_OPEN_FILES:
retFuture.fail(getTransportTooManyError())
else:
retFuture.fail(getTransportOsError(err))
@ -1297,12 +1310,12 @@ when defined(windows):
cast[POVERLAPPED](addr server.aovl))
if res == 0:
let err = osLastError()
if int32(err) == osdefs.ERROR_OPERATION_ABORTED:
case err
of ERROR_OPERATION_ABORTED:
server.apending = false
retFuture.fail(getServerUseClosedError())
return retFuture
elif int32(err) in {osdefs.ERROR_IO_PENDING,
osdefs.ERROR_PIPE_CONNECTED}:
of ERROR_IO_PENDING, ERROR_PIPE_CONNECTED:
discard
else:
server.apending = false
@ -1317,7 +1330,7 @@ else:
import ../sendfile
proc isConnResetError(err: OSErrorCode): bool {.inline.} =
(err == OSErrorCode(ECONNRESET)) or (err == OSErrorCode(EPIPE))
(err == oserrno.ECONNRESET) or (err == oserrno.EPIPE)
proc writeStreamLoop(udata: pointer) =
if isNil(udata):
@ -1345,33 +1358,32 @@ else:
while len(transp.queue) > 0:
template handleError() =
let err = osLastError()
if cint(err) == EINTR:
case err
of oserrno.EINTR:
# Signal happened while writing - try again with all data
transp.queue.addFirst(vector)
continue
if cint(err) in [EWOULDBLOCK, EAGAIN]:
of oserrno.EWOULDBLOCK:
# Socket buffer is full - wait until next write notification - in
# particular, ensure removeWriter is not called
transp.queue.addFirst(vector)
return
# The errors below will clear the write queue, meaning we'll exit the
# loop
if isConnResetError(err):
# Soft error happens which indicates that remote peer got
# disconnected, complete all pending writes in queue with 0.
transp.state.incl({WriteEof})
if not(vector.writer.finished()):
vector.writer.complete(0)
completePendingWriteQueue(transp.queue, 0)
else:
transp.state.incl({WriteError})
let error = getTransportOsError(err)
if not(vector.writer.finished()):
vector.writer.fail(error)
failPendingWriteQueue(transp.queue, error)
# The errors below will clear the write queue, meaning we'll exit the
# loop
if isConnResetError(err):
# Soft error happens which indicates that remote peer got
# disconnected, complete all pending writes in queue with 0.
transp.state.incl({WriteEof})
if not(vector.writer.finished()):
vector.writer.complete(0)
completePendingWriteQueue(transp.queue, 0)
else:
transp.state.incl({WriteError})
let error = getTransportOsError(err)
if not(vector.writer.finished()):
vector.writer.fail(error)
failPendingWriteQueue(transp.queue, error)
var vector = transp.queue.popFirst()
case vector.kind
@ -1443,7 +1455,8 @@ else:
len(transp.buffer) - transp.offset, cint(0)))
if res < 0:
let err = osLastError()
if int(err) == ECONNRESET:
case err
of oserrno.ECONNRESET:
transp.state.incl({ReadEof, ReadPaused})
let rres = removeReader2(transp.fd)
if rres.isErr():
@ -1553,7 +1566,8 @@ else:
proto)
if sock == asyncInvalidSocket:
let err = osLastError()
if int(err) == EMFILE:
case err
of oserrno.EMFILE:
retFuture.fail(getTransportTooManyError())
else:
retFuture.fail(getTransportOsError(err))
@ -1641,7 +1655,8 @@ else:
# and the connection shall be established asynchronously.
#
# http://www.madore.org/~david/computers/connect-intr.html
if (errorCode == EINPROGRESS) or (errorCode == EINTR):
case errorCode
of oserrno.EINPROGRESS, oserrno.EINTR:
let res = addWriter2(sock, continuation)
if res.isErr():
discard unregisterAndCloseFd(sock)
@ -1692,7 +1707,7 @@ else:
discard closeFd(cint(sock))
else:
let errorCode = sres.error()
if errorCode != EAGAIN:
if errorCode != oserrno.EAGAIN:
# This EAGAIN error appears only when server get closed, while
# acceptLoop() reader callback is already scheduled.
raiseOsDefect(errorCode, "acceptLoop(): Unable to accept connection")
@ -1745,14 +1760,15 @@ else:
addr slen, flags)
if sres.isErr():
let errorCode = sres.error()
if errorCode == EAGAIN:
case errorCode
of oserrno.EAGAIN:
# This error appears only when server get closed, while accept()
# continuation is already scheduled.
retFuture.fail(getServerUseClosedError())
elif cint(errorCode) in {EMFILE, ENFILE, ENOBUFS, ENOMEM}:
retFuture.fail(getTransportTooManyError(cint(errorCode)))
elif cint(errorCode) in {ECONNABORTED, EPERM, ETIMEDOUT}:
retFuture.fail(getConnectionAbortedError(cint(errorCode)))
of oserrno.EMFILE, oserrno.ENFILE, oserrno.ENOBUFS, oserrno.ENOMEM:
retFuture.fail(getTransportTooManyError(errorCode))
of oserrno.ECONNABORTED, oserrno.EPERM, oserrno.ETIMEDOUT:
retFuture.fail(getConnectionAbortedError(errorCode))
else:
retFuture.fail(getTransportOsError(errorCode))
# Error is already happened so we ignore removeReader2() errors.
@ -1933,15 +1949,13 @@ proc createStreamServer*(host: TransportAddress,
serverSocket = sock
# SO_REUSEADDR is not useful for Unix domain sockets.
if ServerFlags.ReuseAddr in flags:
if not(setSockOpt(serverSocket, osdefs.SOL_SOCKET,
osdefs.SO_REUSEADDR, 1)):
if not(setSockOpt(serverSocket, SOL_SOCKET, SO_REUSEADDR, 1)):
let err = osLastError()
if sock == asyncInvalidSocket:
discard closeFd(SocketHandle(serverSocket))
raiseTransportOsError(err)
if ServerFlags.ReusePort in flags:
if not(setSockOpt(serverSocket, osdefs.SOL_SOCKET,
osdefs.SO_REUSEPORT, 1)):
if not(setSockOpt(serverSocket, SOL_SOCKET, SO_REUSEPORT, 1)):
let err = osLastError()
if sock == asyncInvalidSocket:
discard closeFd(SocketHandle(serverSocket))
@ -1955,8 +1969,8 @@ proc createStreamServer*(host: TransportAddress,
discard closeFd(SocketHandle(serverSocket))
raiseTransportOsError(err)
host.toSAddr(saddr, slen)
if osdefs.bindSocket(SocketHandle(serverSocket),
cast[ptr SockAddr](addr saddr), slen) != 0:
if bindSocket(SocketHandle(serverSocket),
cast[ptr SockAddr](addr saddr), slen) != 0:
let err = osLastError()
if sock == asyncInvalidSocket:
discard closeFd(SocketHandle(serverSocket))
@ -2165,21 +2179,21 @@ template fastWrite(transp: auto, pbytes: var ptr byte, rbytes: var int,
# Not all bytes written - keep going
else:
let err = osLastError()
if cint(err) in [EAGAIN, EWOULDBLOCK]:
case err
of oserrno.EWOULDBLOCK:
break # No bytes written, add to queue
if cint(err) == EINTR:
of oserrno.EINTR:
continue
if isConnResetError(err):
transp.state.incl({WriteEof})
retFuture.complete(0)
return retFuture
else:
transp.state.incl({WriteError})
let error = getTransportOsError(err)
retFuture.fail(error)
return retFuture
if isConnResetError(err):
transp.state.incl({WriteEof})
retFuture.complete(0)
return retFuture
else:
transp.state.incl({WriteError})
let error = getTransportOsError(err)
retFuture.fail(error)
return retFuture
proc write*(transp: StreamTransport, pbytes: pointer,
nbytes: int): Future[int] =
@ -2340,7 +2354,7 @@ template readLoop(name, body: untyped): untyped =
raiseOsDefect(errorCode, "readLoop(): Unable to resume reading")
else:
transp.reader.complete()
if errorCode == ESRCH:
if errorCode == oserrno.ESRCH:
# ESRCH 3 "No such process"
# This error could be happened on pipes only, when process which
# owns and communicates through this pipe (stdin, stdout, stderr) is

9
tests/teststream.nim
View File

@ -7,7 +7,7 @@
# MIT license (LICENSE-MIT)
import std/[strutils, os]
import unittest2
import ".."/chronos, ".."/chronos/osdefs
import ".."/chronos, ".."/chronos/[osdefs, oserrno]
{.used.}
@ -639,12 +639,11 @@ suite "Stream Transport test suite":
var transp = await connect(address)
doAssert(isNil(transp))
except TransportOsError as e:
let ecode = int(e.code)
when defined(windows):
result = (ecode == ERROR_FILE_NOT_FOUND) or
(ecode == ERROR_CONNECTION_REFUSED)
return (e.code == ERROR_FILE_NOT_FOUND) or
(e.code == ERROR_CONNECTION_REFUSED)
else:
result = (ecode == ECONNREFUSED) or (ecode == ENOENT)
return (e.code == oserrno.ECONNREFUSED) or (e.code == oserrno.ENOENT)
proc serveClient16(server: StreamServer, transp: StreamTransport) {.async.} =
var res = await transp.write(BigMessagePattern)