tasyncawait now works on Linux.
Reworked detection of a file descriptor being closed with epoll (in the case of sockets it is when the remote host disconnects). Ensured that events are only updated when they change.
This commit is contained in:
parent
9b5357da5a
commit
2ce07042fd
3 changed files with 69 additions and 45 deletions
|
|
@ -479,9 +479,11 @@ else:
|
|||
discard p.selector.update(sock, events)
|
||||
|
||||
proc addRead(p: PDispatcher, sock: TSocketHandle, cb: TCallback) =
|
||||
#echo("addRead")
|
||||
if sock notin p.selector:
|
||||
var data = PData(sock: sock, readCBs: @[cb], writeCBs: @[])
|
||||
p.selector.register(sock, {EvRead}, data.PObject)
|
||||
#echo("registered")
|
||||
else:
|
||||
p.selector[sock].data.PData.readCBs.add(cb)
|
||||
p.update(sock, p.selector[sock].events + {EvRead})
|
||||
|
|
@ -498,7 +500,7 @@ else:
|
|||
for info in p.selector.select(timeout):
|
||||
let data = PData(info.key.data)
|
||||
assert data.sock == info.key.fd
|
||||
|
||||
#echo("In poll ", data.sock.cint)
|
||||
if EvRead in info.events:
|
||||
# Callback may add items to ``data.readCBs`` which causes issues if
|
||||
# we are iterating over ``data.readCBs`` at the same time. We therefore
|
||||
|
|
@ -517,11 +519,17 @@ else:
|
|||
if not cb(data.sock):
|
||||
# Callback wants to be called again.
|
||||
data.writeCBs.add(cb)
|
||||
|
||||
var newEvents: set[TEvent]
|
||||
if data.readCBs.len != 0: newEvents = {EvRead}
|
||||
if data.writeCBs.len != 0: newEvents = newEvents + {EvWrite}
|
||||
p.update(data.sock, newEvents)
|
||||
|
||||
if info.key in p.selector:
|
||||
var newEvents: set[TEvent]
|
||||
if data.readCBs.len != 0: newEvents = {EvRead}
|
||||
if data.writeCBs.len != 0: newEvents = newEvents + {EvWrite}
|
||||
if newEvents != info.key.events:
|
||||
echo(info.key.events, " -> ", newEvents)
|
||||
p.update(data.sock, newEvents)
|
||||
else:
|
||||
# FD no longer a part of the selector. Likely been closed
|
||||
# (e.g. socket disconnected).
|
||||
|
||||
proc connect*(p: PDispatcher, socket: TSocketHandle, address: string, port: TPort,
|
||||
af = AF_INET): PFuture[int] =
|
||||
|
|
@ -569,6 +577,7 @@ else:
|
|||
result = true
|
||||
let netSize = size - sizeRead
|
||||
let res = recv(sock, addr readBuffer[sizeRead], netSize, flags.cint)
|
||||
#echo("recv cb res: ", res)
|
||||
if res < 0:
|
||||
let lastError = osLastError()
|
||||
if lastError.int32 notin {EINTR, EWOULDBLOCK, EAGAIN}:
|
||||
|
|
@ -576,6 +585,7 @@ else:
|
|||
else:
|
||||
result = false # We still want this callback to be called.
|
||||
elif res == 0:
|
||||
#echo("Disconnected recv: ", sizeRead)
|
||||
# Disconnected
|
||||
if sizeRead == 0:
|
||||
retFuture.complete("")
|
||||
|
|
@ -588,6 +598,7 @@ else:
|
|||
result = false # We want to read all the data requested.
|
||||
else:
|
||||
retFuture.complete(readBuffer)
|
||||
#echo("Recv cb result: ", result)
|
||||
|
||||
addRead(p, socket, cb)
|
||||
return retFuture
|
||||
|
|
@ -833,9 +844,13 @@ proc recvLine*(p: PDispatcher, socket: TSocketHandle): PFuture[string] {.async.}
|
|||
result = ""
|
||||
var c = ""
|
||||
while true:
|
||||
#echo("1")
|
||||
c = await p.recv(socket, 1)
|
||||
#echo("Received ", c.len)
|
||||
if c.len == 0:
|
||||
#echo("returning")
|
||||
return
|
||||
#echo("2")
|
||||
if c == "\r":
|
||||
c = await p.recv(socket, 1, MSG_PEEK)
|
||||
if c.len > 0 and c == "\L":
|
||||
|
|
@ -845,7 +860,9 @@ proc recvLine*(p: PDispatcher, socket: TSocketHandle): PFuture[string] {.async.}
|
|||
elif c == "\L":
|
||||
addNLIfEmpty()
|
||||
return
|
||||
#echo("3")
|
||||
add(result.string, c)
|
||||
#echo("4")
|
||||
|
||||
when isMainModule:
|
||||
|
||||
|
|
@ -859,6 +876,7 @@ when isMainModule:
|
|||
proc main(p: PDispatcher): PFuture[int] {.async.} =
|
||||
discard await p.connect(sock, "irc.freenode.net", TPort(6667))
|
||||
while true:
|
||||
echo("recvLine")
|
||||
var line = await p.recvLine(sock)
|
||||
echo("Line is: ", line.repr)
|
||||
if line == "":
|
||||
|
|
@ -882,7 +900,7 @@ when isMainModule:
|
|||
else:
|
||||
when true:
|
||||
|
||||
var f = p.connect(sock, "irc.freenode.org", TPort(6667))
|
||||
var f = p.connect(sock, "irc.poop.nl", TPort(6667))
|
||||
f.callback =
|
||||
proc (future: PFuture[int]) =
|
||||
echo("Connected in future!")
|
||||
|
|
@ -898,11 +916,13 @@ when isMainModule:
|
|||
sock.bindAddr(TPort(6667))
|
||||
sock.listen()
|
||||
proc onAccept(future: PFuture[TSocketHandle]) =
|
||||
echo "Accepted"
|
||||
var t = p.send(future.read, "test\c\L")
|
||||
let client = future.read
|
||||
echo "Accepted ", client.cint
|
||||
var t = p.send(client, "test\c\L")
|
||||
t.callback =
|
||||
proc (future: PFuture[int]) =
|
||||
echo(future.read)
|
||||
echo("Send: ", future.read)
|
||||
client.close()
|
||||
|
||||
var f = p.accept(sock)
|
||||
f.callback = onAccept
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue