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

PFuture[void] now works.

Browse source 
Return types can also be completely ommitted. PFuture[void] will then be
implicitly deduced.
This commit is contained in:
Dominik Picheta 2014-03-15 00:09:33 +00:00
commit de64e8ec22
2 changed files with 65 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

82
lib/pure/asyncio2.nim
View file

@ -43,6 +43,14 @@ proc complete*[T](future: PFuture[T], val: T) =
if future.cb != nil: if future.cb != nil:
future.cb() future.cb()
proc complete*(future: PFuture[void]) =
## Completes a void ``future``.
assert(not future.finished, "Future already finished, cannot finish twice.")
assert(future.error == nil)
future.finished = true
if future.cb != nil:
future.cb()
proc fail*[T](future: PFuture[T], error: ref EBase) = proc fail*[T](future: PFuture[T], error: ref EBase) =
## Completes ``future`` with ``error``. ## Completes ``future`` with ``error``.
assert(not future.finished, "Future already finished, cannot finish twice.") assert(not future.finished, "Future already finished, cannot finish twice.")
@ -76,7 +84,8 @@ proc read*[T](future: PFuture[T]): T =
## If the result of the future is an error then that error will be raised. ## If the result of the future is an error then that error will be raised.
if future.finished: if future.finished:
if future.error != nil: raise future.error if future.error != nil: raise future.error
return future.value when T isnot void:
return future.value
else: else:
# TODO: Make a custom exception type for this? # TODO: Make a custom exception type for this?
raise newException(EInvalidValue, "Future still in progress.") raise newException(EInvalidValue, "Future still in progress.")
@ -243,13 +252,13 @@ when defined(windows) or defined(nimdoc):
RemoteSockaddr, RemoteSockaddrLength) RemoteSockaddr, RemoteSockaddrLength)
proc connect*(p: PDispatcher, socket: TSocketHandle, address: string, port: TPort, proc connect*(p: PDispatcher, socket: TSocketHandle, address: string, port: TPort,
af = AF_INET): PFuture[int] = af = AF_INET): PFuture[void] =
## Connects ``socket`` to server at ``address:port``. ## Connects ``socket`` to server at ``address:port``.
## ##
## Returns a ``PFuture`` which will complete when the connection succeeds ## Returns a ``PFuture`` which will complete when the connection succeeds
## or an error occurs. ## or an error occurs.
verifyPresence(p, socket) verifyPresence(p, socket)
var retFuture = newFuture[int]()# TODO: Change to void when that regression is fixed. var retFuture = newFuture[void]()
# Apparently ``ConnectEx`` expects the socket to be initially bound: # Apparently ``ConnectEx`` expects the socket to be initially bound:
var saddr: Tsockaddr_in var saddr: Tsockaddr_in
saddr.sin_family = int16(toInt(af)) saddr.sin_family = int16(toInt(af))
@ -271,7 +280,7 @@ when defined(windows) or defined(nimdoc):
proc (sock: TSocketHandle, bytesCount: DWord, errcode: TOSErrorCode) = proc (sock: TSocketHandle, bytesCount: DWord, errcode: TOSErrorCode) =
if not retFuture.finished: if not retFuture.finished:
if errcode == TOSErrorCode(-1): if errcode == TOSErrorCode(-1):
retFuture.complete(0) retFuture.complete()
else: else:
retFuture.fail(newException(EOS, osErrorMsg(errcode))) retFuture.fail(newException(EOS, osErrorMsg(errcode)))
) )
@ -281,7 +290,7 @@ when defined(windows) or defined(nimdoc):
if ret: if ret:
# Request to connect completed immediately. # Request to connect completed immediately.
success = true success = true
retFuture.complete(0) retFuture.complete()
# We don't deallocate ``ol`` here because even though this completed # We don't deallocate ``ol`` here because even though this completed
# immediately poll will still be notified about its completion and it will # immediately poll will still be notified about its completion and it will
# free ``ol``. # free ``ol``.
@ -362,11 +371,11 @@ when defined(windows) or defined(nimdoc):
# free ``ol``. # free ``ol``.
return retFuture return retFuture
proc send*(p: PDispatcher, socket: TSocketHandle, data: string): PFuture[int] = proc send*(p: PDispatcher, socket: TSocketHandle, data: string): 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(p, socket) verifyPresence(p, socket)
var retFuture = newFuture[int]() var retFuture = newFuture[void]()
var dataBuf: TWSABuf var dataBuf: TWSABuf
dataBuf.buf = data dataBuf.buf = data
@ -378,7 +387,7 @@ when defined(windows) or defined(nimdoc):
proc (sock: TSocketHandle, bytesCount: DWord, errcode: TOSErrorCode) = proc (sock: TSocketHandle, bytesCount: DWord, errcode: TOSErrorCode) =
if not retFuture.finished: if not retFuture.finished:
if errcode == TOSErrorCode(-1): if errcode == TOSErrorCode(-1):
retFuture.complete(0) retFuture.complete()
else: else:
retFuture.fail(newException(EOS, osErrorMsg(errcode))) retFuture.fail(newException(EOS, osErrorMsg(errcode)))
) )
@ -391,7 +400,7 @@ when defined(windows) or defined(nimdoc):
retFuture.fail(newException(EOS, osErrorMsg(err))) retFuture.fail(newException(EOS, osErrorMsg(err)))
dealloc(ol) dealloc(ol)
else: else:
retFuture.complete(0) retFuture.complete()
# We don't deallocate ``ol`` here because even though this completed # We don't deallocate ``ol`` here because even though this completed
# immediately poll will still be notified about its completion and it will # immediately poll will still be notified about its completion and it will
# free ``ol``. # free ``ol``.
@ -564,12 +573,12 @@ else:
# (e.g. socket disconnected). # (e.g. socket disconnected).
proc connect*(p: PDispatcher, socket: TSocketHandle, address: string, port: TPort, proc connect*(p: PDispatcher, socket: TSocketHandle, address: string, port: TPort,
af = AF_INET): PFuture[int] = af = AF_INET): PFuture[void] =
var retFuture = newFuture[int]() var retFuture = newFuture[void]()
proc cb(sock: TSocketHandle): bool = proc cb(sock: TSocketHandle): bool =
# We have connected. # We have connected.
retFuture.complete(0) retFuture.complete()
return true return true
var aiList = getAddrInfo(address, port, af) var aiList = getAddrInfo(address, port, af)
@ -581,7 +590,7 @@ else:
if ret == 0: if ret == 0:
# Request to connect completed immediately. # Request to connect completed immediately.
success = true success = true
retFuture.complete(0) retFuture.complete()
break break
else: else:
lastError = osLastError() lastError = osLastError()
@ -635,8 +644,8 @@ else:
addRead(p, socket, cb) addRead(p, socket, cb)
return retFuture return retFuture
proc send*(p: PDispatcher, socket: TSocketHandle, data: string): PFuture[int] = proc send*(p: PDispatcher, socket: TSocketHandle, data: string): PFuture[void] =
var retFuture = newFuture[int]() var retFuture = newFuture[void]()
var written = 0 var written = 0
@ -656,7 +665,7 @@ else:
if res != netSize: if res != netSize:
result = false # We still have data to send. result = false # We still have data to send.
else: else:
retFuture.complete(0) retFuture.complete()
addWrite(p, socket, cb) addWrite(p, socket, cb)
return retFuture return retFuture
@ -789,12 +798,17 @@ macro async*(prc: stmt): stmt {.immediate.} =
hint("Processing " & prc[0].getName & " as an async proc.") hint("Processing " & prc[0].getName & " as an async proc.")
let returnType = prc[3][0]
var subtypeName = ""
# Verify that the return type is a PFuture[T] # Verify that the return type is a PFuture[T]
if prc[3][0].kind == nnkIdent: if returnType.kind == nnkIdent:
error("Expected return type of 'PFuture' got '" & $prc[3][0] & "'") error("Expected return type of 'PFuture' got '" & $returnType & "'")
elif prc[3][0].kind == nnkBracketExpr: elif returnType.kind == nnkBracketExpr:
if $prc[3][0][0] != "PFuture": if $returnType[0] != "PFuture":
error("Expected return type of 'PFuture' got '" & $prc[3][0][0] & "'") error("Expected return type of 'PFuture' got '" & $returnType[0] & "'")
subtypeName = $returnType[1].ident
elif returnType.kind == nnkEmpty:
subtypeName = "void"
# TODO: Why can't I use genSym? I get illegal capture errors for Syms. # TODO: Why can't I use genSym? I get illegal capture errors for Syms.
# TODO: It seems genSym is broken. Change all usages back to genSym when fixed # TODO: It seems genSym is broken. Change all usages back to genSym when fixed
@ -807,20 +821,24 @@ macro async*(prc: stmt): stmt {.immediate.} =
newVarStmt(retFutureSym, newVarStmt(retFutureSym,
newCall( newCall(
newNimNode(nnkBracketExpr).add( newNimNode(nnkBracketExpr).add(
newIdentNode("newFuture"), newIdentNode(!"newFuture"), # TODO: Strange bug here? Remove the `!`.
prc[3][0][1])))) # Get type from return type of this proc. newIdentNode(subtypeName))))) # Get type from return type of this proc
echo(treeRepr(outerProcBody))
# -> iterator nameIter(): PFutureBase {.closure.} = # -> iterator nameIter(): PFutureBase {.closure.} =
# -> var result: T # -> var result: T
# -> <proc_body> # -> <proc_body>
# -> complete(retFuture, result) # -> complete(retFuture, result)
var iteratorNameSym = newIdentNode($prc[0].getName & "Iter") #genSym(nskIterator, $prc[0].ident & "Iter") var iteratorNameSym = newIdentNode($prc[0].getName & "Iter") #genSym(nskIterator, $prc[0].ident & "Iter")
var procBody = prc[6].processBody(retFutureSym) var procBody = prc[6].processBody(retFutureSym)
procBody.insert(0, newNimNode(nnkVarSection).add( if subtypeName != "void":
newIdentDefs(newIdentNode("result"), prc[3][0][1]))) # -> var result: T procBody.insert(0, newNimNode(nnkVarSection).add(
procBody.add( newIdentDefs(newIdentNode("result"), returnType[1]))) # -> var result: T
newCall(newIdentNode("complete"), procBody.add(
retFutureSym, newIdentNode("result"))) # -> complete(retFuture, result) newCall(newIdentNode("complete"),
retFutureSym, newIdentNode("result"))) # -> complete(retFuture, result)
else:
# -> complete(retFuture)
procBody.add(newCall(newIdentNode("complete"), retFutureSym))
var closureIterator = newProc(iteratorNameSym, [newIdentNode("PFutureBase")], var closureIterator = newProc(iteratorNameSym, [newIdentNode("PFutureBase")],
procBody, nnkIteratorDef) procBody, nnkIteratorDef)
@ -855,6 +873,12 @@ macro async*(prc: stmt): stmt {.immediate.} =
for i in 0 .. <result[4].len: for i in 0 .. <result[4].len:
if result[4][i].ident == !"async": if result[4][i].ident == !"async":
result[4].del(i) result[4].del(i)
if subtypeName == "void":
# Add discardable pragma.
result[4].add(newIdentNode("discardable"))
if returnType.kind == nnkEmpty:
# Add PFuture[void]
result[3][0] = parseExpr("PFuture[void]")
result[6] = outerProcBody result[6] = outerProcBody

22
tests/async/tasyncawait.nim
View file

@ -14,18 +14,18 @@ const
var clientCount = 0 var clientCount = 0
proc sendMessages(disp: PDispatcher, client: TSocketHandle): PFuture[int] {.async.} = proc sendMessages(disp: PDispatcher, client: TSocketHandle) {.async.} =
for i in 0 .. <messagesToSend: for i in 0 .. <messagesToSend:
discard await disp.send(client, "Message " & $i & "\c\L") await disp.send(client, "Message " & $i & "\c\L")
proc launchSwarm(disp: PDispatcher, port: TPort): PFuture[int] {.async.} = proc launchSwarm(disp: PDispatcher, port: TPort) {.async.} =
for i in 0 .. <swarmSize: for i in 0 .. <swarmSize:
var sock = disp.socket() var sock = disp.socket()
#disp.register(sock) #disp.register(sock)
discard await disp.connect(sock, "localhost", port) await disp.connect(sock, "localhost", port)
when true: when true:
discard await sendMessages(disp, sock) await sendMessages(disp, sock)
disp.close(sock) disp.close(sock)
else: else:
# Issue #932: https://github.com/Araq/Nimrod/issues/932 # Issue #932: https://github.com/Araq/Nimrod/issues/932
@ -34,7 +34,7 @@ proc launchSwarm(disp: PDispatcher, port: TPort): PFuture[int] {.async.} =
proc () = proc () =
disp.close(sock) disp.close(sock)
proc readMessages(disp: PDispatcher, client: TSocketHandle): PFuture[int] {.async.} = proc readMessages(disp: PDispatcher, client: TSocketHandle) {.async.} =
while true: while true:
var line = await disp.recvLine(client) var line = await disp.recvLine(client)
if line == "": if line == "":
@ -47,16 +47,18 @@ proc readMessages(disp: PDispatcher, client: TSocketHandle): PFuture[int] {.asyn
else: else:
doAssert false doAssert false
proc createServer(disp: PDispatcher, port: TPort): PFuture[int] {.async.} = proc createServer(disp: PDispatcher, port: TPort) {.async.} =
var server = disp.socket() var server = disp.socket()
#disp.register(server) #disp.register(server)
server.bindAddr(port) server.bindAddr(port)
server.listen() server.listen()
while true: while true:
discard readMessages(disp, await disp.accept(server)) var client = await disp.accept(server)
readMessages(disp, client)
# TODO: Test: readMessages(disp, await disp.accept(server))
discard disp.createServer(TPort(10335)) disp.createServer(TPort(10335))
discard disp.launchSwarm(TPort(10335)) disp.launchSwarm(TPort(10335))
while true: while true:
disp.poll() disp.poll()
if clientCount == swarmSize: break if clientCount == swarmSize: break