status-im
/
nim-chronos
mirror of https://github.com/status-im/nim-chronos.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fix datagram bugs. 

Make changes to asyncloop according to zahary review.
This commit is contained in:
cheatfate 2018-06-04 03:54:17 +03:00
parent 0b7115eba9
commit 2758abd7a6
2 changed files with 53 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
asyncdispatch2/asyncloop.nim
View File

@ -292,8 +292,6 @@ when defined(windows) or defined(nimdoc):
## (Unix) for the specified dispatcher. ## (Unix) for the specified dispatcher.
return disp.ioPort return disp.ioPort
# ZAH: Shouldn't all of these procs be defined over the Dispatcher type?
# The "global" variants can be defined as templates passing the global dispatcher
proc register*(fd: AsyncFD) = proc register*(fd: AsyncFD) =
## Registers ``fd`` with the dispatcher. ## Registers ``fd`` with the dispatcher.
let p = getGlobalDispatcher() let p = getGlobalDispatcher()

92
asyncdispatch2/transports/datagram.nim
View File

@ -11,31 +11,22 @@ import net, nativesockets, os, deques, strutils
import ../asyncloop, ../handles import ../asyncloop, ../handles
import common import common
when defined(windows):
import winlean
else:
import posix
type type
VectorKind = enum VectorKind = enum
WithoutAddress, WithAddress WithoutAddress, WithAddress
when defined(windows):
import winlean
type
GramVector = object GramVector = object
kind: VectorKind # Vector kind (with address/without address) kind: VectorKind # Vector kind (with address/without address)
buf: ptr TWSABuf # Writer vector buffer
address: TransportAddress # Destination address address: TransportAddress # Destination address
writer: Future[void] # Writer vector completion Future
else:
import posix
type
GramVector = object
kind: VectorKind # Vector kind (with address/without address)
buf: pointer # Writer buffer pointer buf: pointer # Writer buffer pointer
buflen: int # Writer buffer size buflen: int # Writer buffer size
address: TransportAddress # Destination address
writer: Future[void] # Writer vector completion Future writer: Future[void] # Writer vector completion Future
type
DatagramServer* = ref object of RootRef DatagramServer* = ref object of RootRef
## Datagram server object ## Datagram server object
transport*: DatagramTransport ## Datagram transport transport*: DatagramTransport ## Datagram transport
@ -76,15 +67,22 @@ template setWriteError(t, e: untyped) =
(t).state.incl(WriteError) (t).state.incl(WriteError)
(t).error = newException(TransportOsError, osErrorMsg((e))) (t).error = newException(TransportOsError, osErrorMsg((e)))
template setWriterWSABuffer(t, v: untyped) =
(t).wwsabuf.buf = cast[cstring](v.buf)
(t).wwsabuf.len = cast[int32](v.buflen)
when defined(windows): when defined(windows):
type type
WindowsDatagramTransport* = ref object of DatagramTransport WindowsDatagramTransport* = ref object of DatagramTransport
rovl: CustomOverlapped rovl: CustomOverlapped # Reader OVERLAPPED structure
wovl: CustomOverlapped wovl: CustomOverlapped # Writer OVERLAPPED structure
raddr: Sockaddr_storage raddr: Sockaddr_storage # Reader address storage
ralen: SockLen ralen: SockLen # Reader address length
rflag: int32 rflag: int32 # Reader flags storage
wsabuf: TWSABuf rwsabuf: TWSABuf # Reader WSABUF structure
waddr: Sockaddr_storage # Writer address storage
wlen: SockLen # Writer address length
wwsabuf: TWSABuf # Writer WSABUF structure
template finishWriter(t: untyped) = template finishWriter(t: untyped) =
var vv = (t).queue.popFirst() var vv = (t).queue.popFirst()
@ -111,19 +109,21 @@ when defined(windows):
transp.finishWriter() transp.finishWriter()
else: else:
## Initiation ## Initiation
var saddr: Sockaddr_storage
var slen: SockLen
transp.state.incl(WritePending) transp.state.incl(WritePending)
let fd = SocketHandle(ovl.data.fd) let fd = SocketHandle(ovl.data.fd)
var vector = transp.queue.popFirst() var vector = transp.queue.popFirst()
transp.setWriterWSABuffer(vector)
var ret: cint
if vector.kind == WithAddress: if vector.kind == WithAddress:
toSockAddr(vector.address.address, vector.address.port, saddr, slen) toSockAddr(vector.address.address, vector.address.port,
else: transp.waddr, transp.wlen)
toSockAddr(transp.remote.address, transp.remote.port, saddr, slen) ret = WSASendTo(fd, addr transp.wwsabuf, DWORD(1), addr bytesCount,
let ret = WSASendTo(fd, vector.buf, DWORD(1), addr bytesCount, DWORD(0), cast[ptr SockAddr](addr transp.waddr),
DWORD(0), cast[ptr SockAddr](addr saddr), cint(transp.wlen),
cint(slen),
cast[POVERLAPPED](addr transp.wovl), nil) cast[POVERLAPPED](addr transp.wovl), nil)
else:
ret = WSASend(fd, addr transp.wwsabuf, DWORD(1), addr bytesCount,
DWORD(0), cast[POVERLAPPED](addr transp.wovl), nil)
if ret != 0: if ret != 0:
let err = osLastError() let err = osLastError()
if int(err) == ERROR_OPERATION_ABORTED: if int(err) == ERROR_OPERATION_ABORTED:
@ -178,11 +178,8 @@ when defined(windows):
let fd = SocketHandle(ovl.data.fd) let fd = SocketHandle(ovl.data.fd)
transp.rflag = 0 transp.rflag = 0
transp.ralen = SockLen(sizeof(Sockaddr_storage)) transp.ralen = SockLen(sizeof(Sockaddr_storage))
let ret = WSARecvFrom(fd, let ret = WSARecvFrom(fd, addr transp.rwsabuf, DWORD(1),
addr transp.wsabuf, addr bytesCount, addr transp.rflag,
DWORD(1),
addr bytesCount,
addr transp.rflag,
cast[ptr SockAddr](addr transp.raddr), cast[ptr SockAddr](addr transp.raddr),
cast[ptr cint](addr transp.ralen), cast[ptr cint](addr transp.ralen),
cast[POVERLAPPED](addr transp.rovl), nil) cast[POVERLAPPED](addr transp.rovl), nil)
@ -275,7 +272,17 @@ when defined(windows):
if sock == asyncInvalidSocket: if sock == asyncInvalidSocket:
closeAsyncSocket(localSock) closeAsyncSocket(localSock)
raiseOsError(err) raiseOsError(err)
if remote.port != Port(0): if remote.port != Port(0):
var saddr: Sockaddr_storage
var slen: SockLen
toSockAddr(remote.address, remote.port, saddr, slen)
if connect(SocketHandle(localSock), cast[ptr SockAddr](addr saddr),
slen) != 0:
let err = osLastError()
if sock == asyncInvalidSocket:
closeAsyncSocket(localSock)
raiseOsError(err)
wresult.remote = remote wresult.remote = remote
wresult.fd = localSock wresult.fd = localSock
@ -289,7 +296,7 @@ when defined(windows):
udata: cast[pointer](wresult)) udata: cast[pointer](wresult))
wresult.wovl.data = CompletionData(fd: localSock, cb: writeDatagramLoop, wresult.wovl.data = CompletionData(fd: localSock, cb: writeDatagramLoop,
udata: cast[pointer](wresult)) udata: cast[pointer](wresult))
wresult.wsabuf = TWSABuf(buf: cast[cstring](addr wresult.buffer[0]), wresult.rwsabuf = TWSABuf(buf: cast[cstring](addr wresult.buffer[0]),
len: int32(len(wresult.buffer))) len: int32(len(wresult.buffer)))
GC_ref(wresult) GC_ref(wresult)
result = cast[DatagramTransport](wresult) result = cast[DatagramTransport](wresult)
@ -301,7 +308,7 @@ when defined(windows):
proc close*(transp: DatagramTransport) = proc close*(transp: DatagramTransport) =
## Closes and frees resources of transport ``transp``. ## Closes and frees resources of transport ``transp``.
if ReadClosed notin transp.state and WriteClosed notin transp.state: if ReadClosed notin transp.state and WriteClosed notin transp.state:
discard cancelIo(Handle(transp.fd)) # discard cancelIo(Handle(transp.fd))
closeAsyncSocket(transp.fd) closeAsyncSocket(transp.fd)
transp.state.incl(WriteClosed) transp.state.incl(WriteClosed)
transp.state.incl(ReadClosed) transp.state.incl(ReadClosed)
@ -525,11 +532,6 @@ proc send*(transp: DatagramTransport, pbytes: pointer,
if transp.remote.port == Port(0): if transp.remote.port == Port(0):
raise newException(TransportError, "Remote peer is not set!") raise newException(TransportError, "Remote peer is not set!")
var waitFuture = newFuture[void]("datagram.transport.send") var waitFuture = newFuture[void]("datagram.transport.send")
when defined(windows):
var wsabuf = TWSABuf(buf: cast[cstring](pbytes), len: int32(nbytes))
var vector = GramVector(kind: WithoutAddress, buf: addr wsabuf,
writer: waitFuture)
else:
var vector = GramVector(kind: WithoutAddress, buf: pbytes, buflen: nbytes, var vector = GramVector(kind: WithoutAddress, buf: pbytes, buflen: nbytes,
writer: waitFuture) writer: waitFuture)
transp.queue.addLast(vector) transp.queue.addLast(vector)
@ -546,16 +548,10 @@ proc sendTo*(transp: DatagramTransport, pbytes: pointer, nbytes: int,
checkClosed(transp) checkClosed(transp)
var saddr: Sockaddr_storage var saddr: Sockaddr_storage
var slen: SockLen var slen: SockLen
var vector: GramVector
toSockAddr(remote.address, remote.port, saddr, slen) toSockAddr(remote.address, remote.port, saddr, slen)
var waitFuture = newFuture[void]("datagram.transport.sendto") var waitFuture = newFuture[void]("datagram.transport.sendto")
when defined(windows): var vector = GramVector(kind: WithAddress, buf: pbytes, buflen: nbytes,
var wsabuf = TWSABuf(buf: cast[cstring](pbytes), len: int32(nbytes)) writer: waitFuture, address: remote)
vector = GramVector(kind: WithAddress, buf: addr wsabuf,
address: remote, writer: waitFuture)
else:
vector = GramVector(kind: WithAddress, buf: pbytes, buflen: nbytes,
address: remote, writer: waitFuture)
transp.queue.addLast(vector) transp.queue.addLast(vector)
if WritePaused in transp.state: if WritePaused in transp.state:
transp.resumeWrite() transp.resumeWrite()