diff options
Diffstat (limited to 'lib/pure')
| -rw-r--r-- | lib/pure/asyncdispatch.nim | 282 | ||||
| -rw-r--r-- | lib/pure/asyncmacro.nim | 51 | ||||
| -rw-r--r-- | lib/pure/asyncnet.nim | 21 | ||||
| -rw-r--r-- | lib/pure/basic2d.nim | 4 | ||||
| -rw-r--r-- | lib/pure/basic3d.nim | 6 | ||||
| -rw-r--r-- | lib/pure/includes/asyncfutures.nim | 293 | ||||
| -rw-r--r-- | lib/pure/selectors.nim | 3 |
7 files changed, 362 insertions, 298 deletions
diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim index b4e28d9bc..8336da1fb 100644 --- a/lib/pure/asyncdispatch.nim +++ b/lib/pure/asyncdispatch.nim @@ -156,290 +156,10 @@ export Port, SocketFlag ## * Can't await in a ``except`` body ## * Forward declarations for async procs are broken, ## link includes workaround: https://github.com/nim-lang/Nim/issues/3182. -## * FutureVar[T] needs to be completed manually. # TODO: Check if yielded future is nil and throw a more meaningful exception -# -- Futures - -type - FutureBase* = ref object of RootObj ## Untyped future. - cb: proc () {.closure,gcsafe.} - finished: bool - error*: ref Exception ## Stored exception - errorStackTrace*: string - when not defined(release): - stackTrace: string ## For debugging purposes only. - id: int - fromProc: string - - Future*[T] = ref object of FutureBase ## Typed future. - value: T ## Stored value - - FutureVar*[T] = distinct Future[T] - - FutureError* = object of Exception - cause*: FutureBase - -{.deprecated: [PFutureBase: FutureBase, PFuture: Future].} - -when not defined(release): - var currentID = 0 - -proc callSoon*(cbproc: proc ()) {.gcsafe.} - -proc newFuture*[T](fromProc: string = "unspecified"): Future[T] = - ## Creates a new future. - ## - ## Specifying ``fromProc``, which is a string specifying the name of the proc - ## that this future belongs to, is a good habit as it helps with debugging. - new(result) - result.finished = false - when not defined(release): - result.stackTrace = getStackTrace() - result.id = currentID - result.fromProc = fromProc - currentID.inc() - -proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] = - ## Create a new ``FutureVar``. This Future type is ideally suited for - ## situations where you want to avoid unnecessary allocations of Futures. - ## - ## Specifying ``fromProc``, which is a string specifying the name of the proc - ## that this future belongs to, is a good habit as it helps with debugging. - result = FutureVar[T](newFuture[T](fromProc)) - -proc clean*[T](future: FutureVar[T]) = - ## Resets the ``finished`` status of ``future``. - Future[T](future).finished = false - Future[T](future).error = nil - -proc checkFinished[T](future: Future[T]) = - ## Checks whether `future` is finished. If it is then raises a - ## ``FutureError``. - when not defined(release): - if future.finished: - var msg = "" - msg.add("An attempt was made to complete a Future more than once. ") - msg.add("Details:") - msg.add("\n Future ID: " & $future.id) - msg.add("\n Created in proc: " & future.fromProc) - msg.add("\n Stack trace to moment of creation:") - msg.add("\n" & indent(future.stackTrace.strip(), 4)) - when T is string: - msg.add("\n Contents (string): ") - msg.add("\n" & indent(future.value.repr, 4)) - msg.add("\n Stack trace to moment of secondary completion:") - msg.add("\n" & indent(getStackTrace().strip(), 4)) - var err = newException(FutureError, msg) - err.cause = future - raise err - -proc complete*[T](future: Future[T], val: T) = - ## Completes ``future`` with value ``val``. - #assert(not future.finished, "Future already finished, cannot finish twice.") - checkFinished(future) - assert(future.error == nil) - future.value = val - future.finished = true - if future.cb != nil: - future.cb() - -proc complete*(future: Future[void]) = - ## Completes a void ``future``. - #assert(not future.finished, "Future already finished, cannot finish twice.") - checkFinished(future) - assert(future.error == nil) - future.finished = true - if future.cb != nil: - future.cb() - -proc complete*[T](future: FutureVar[T]) = - ## Completes a ``FutureVar``. - template fut: expr = Future[T](future) - checkFinished(fut) - assert(fut.error == nil) - fut.finished = true - if fut.cb != nil: - fut.cb() - -proc fail*[T](future: Future[T], error: ref Exception) = - ## Completes ``future`` with ``error``. - #assert(not future.finished, "Future already finished, cannot finish twice.") - checkFinished(future) - future.finished = true - future.error = error - future.errorStackTrace = - if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error) - if future.cb != nil: - future.cb() - else: - # This is to prevent exceptions from being silently ignored when a future - # is discarded. - # TODO: This may turn out to be a bad idea. - # Turns out this is a bad idea. - #raise error - discard - -proc `callback=`*(future: FutureBase, cb: proc () {.closure,gcsafe.}) = - ## Sets the callback proc to be called when the future completes. - ## - ## If future has already completed then ``cb`` will be called immediately. - ## - ## **Note**: You most likely want the other ``callback`` setter which - ## passes ``future`` as a param to the callback. - future.cb = cb - if future.finished: - callSoon(future.cb) - -proc `callback=`*[T](future: Future[T], - cb: proc (future: Future[T]) {.closure,gcsafe.}) = - ## Sets the callback proc to be called when the future completes. - ## - ## If future has already completed then ``cb`` will be called immediately. - future.callback = proc () = cb(future) - -proc injectStacktrace[T](future: Future[T]) = - # TODO: Come up with something better. - when not defined(release): - var msg = "" - msg.add("\n " & future.fromProc & "'s lead up to read of failed Future:") - - if not future.errorStackTrace.isNil and future.errorStackTrace != "": - msg.add("\n" & indent(future.errorStackTrace.strip(), 4)) - else: - msg.add("\n Empty or nil stack trace.") - future.error.msg.add(msg) - -proc read*[T](future: Future[T]): T = - ## Retrieves the value of ``future``. Future must be finished otherwise - ## this function will fail with a ``ValueError`` exception. - ## - ## If the result of the future is an error then that error will be raised. - if future.finished: - if future.error != nil: - injectStacktrace(future) - raise future.error - when T isnot void: - return future.value - else: - # TODO: Make a custom exception type for this? - raise newException(ValueError, "Future still in progress.") - -proc readError*[T](future: Future[T]): ref Exception = - ## Retrieves the exception stored in ``future``. - ## - ## An ``ValueError`` exception will be thrown if no exception exists - ## in the specified Future. - if future.error != nil: return future.error - else: - raise newException(ValueError, "No error in future.") - -proc mget*[T](future: FutureVar[T]): var T = - ## Returns a mutable value stored in ``future``. - ## - ## Unlike ``read``, this function will not raise an exception if the - ## Future has not been finished. - result = Future[T](future).value - -proc finished*[T](future: Future[T]): bool = - ## Determines whether ``future`` has completed. - ## - ## ``True`` may indicate an error or a value. Use ``failed`` to distinguish. - future.finished - -proc failed*(future: FutureBase): bool = - ## Determines whether ``future`` completed with an error. - return future.error != nil - -proc asyncCheck*[T](future: Future[T]) = - ## Sets a callback on ``future`` which raises an exception if the future - ## finished with an error. - ## - ## This should be used instead of ``discard`` to discard void futures. - future.callback = - proc () = - if future.failed: - injectStacktrace(future) - raise future.error - -proc `and`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] = - ## Returns a future which will complete once both ``fut1`` and ``fut2`` - ## complete. - var retFuture = newFuture[void]("asyncdispatch.`and`") - fut1.callback = - proc () = - if not retFuture.finished: - if fut1.failed: retFuture.fail(fut1.error) - elif fut2.finished: retFuture.complete() - fut2.callback = - proc () = - if not retFuture.finished: - if fut2.failed: retFuture.fail(fut2.error) - elif fut1.finished: retFuture.complete() - return retFuture - -proc `or`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] = - ## Returns a future which will complete once either ``fut1`` or ``fut2`` - ## complete. - var retFuture = newFuture[void]("asyncdispatch.`or`") - proc cb[X](fut: Future[X]) = - if fut.failed: retFuture.fail(fut.error) - if not retFuture.finished: retFuture.complete() - fut1.callback = cb[T] - fut2.callback = cb[Y] - return retFuture - -proc all*[T](futs: varargs[Future[T]]): auto = - ## Returns a future which will complete once - ## all futures in ``futs`` complete. - ## - ## If the awaited futures are not ``Future[void]``, the returned future - ## will hold the values of all awaited futures in a sequence. - ## - ## If the awaited futures *are* ``Future[void]``, - ## this proc returns ``Future[void]``. - - when T is void: - var - retFuture = newFuture[void]("asyncdispatch.all") - completedFutures = 0 - - let totalFutures = len(futs) - - for fut in futs: - fut.callback = proc(f: Future[T]) = - if f.failed: - retFuture.fail(f.error) - elif not retFuture.finished: - inc(completedFutures) - - if completedFutures == totalFutures: - retFuture.complete() - - return retFuture - - else: - var - retFuture = newFuture[seq[T]]("asyncdispatch.all") - retValues = newSeq[T](len(futs)) - completedFutures = 0 - - for i, fut in futs: - proc setCallback(i: int) = - fut.callback = proc(f: Future[T]) = - if f.failed: - retFuture.fail(f.error) - elif not retFuture.finished: - retValues[i] = f.read() - inc(completedFutures) - - if completedFutures == len(retValues): - retFuture.complete(retValues) - - setCallback(i) - - return retFuture +include includes/asyncfutures type PDispatcherBase = ref object of RootRef diff --git a/lib/pure/asyncmacro.nim b/lib/pure/asyncmacro.nim index f70afaafa..3d004e84c 100644 --- a/lib/pure/asyncmacro.nim +++ b/lib/pure/asyncmacro.nim @@ -25,7 +25,7 @@ proc skipStmtList(node: NimNode): NimNode {.compileTime.} = result = node[0] template createCb(retFutureSym, iteratorNameSym, - name: untyped) = + name, futureVarCompletions: untyped) = var nameIterVar = iteratorNameSym #{.push stackTrace: off.} proc cb {.closure,gcsafe.} = @@ -44,6 +44,8 @@ template createCb(retFutureSym, iteratorNameSym, raise else: retFutureSym.fail(getCurrentException()) + + futureVarCompletions cb() #{.pop.} proc generateExceptionCheck(futSym, @@ -119,8 +121,22 @@ template createVar(result: var NimNode, futSymName: string, result.add newVarStmt(futSym, asyncProc) # -> var future<x> = y useVar(result, futSym, valueReceiver, rootReceiver, fromNode) +proc createFutureVarCompletions(futureVarIdents: seq[NimNode]): NimNode + {.compileTime.} = + result = newStmtList() + # Add calls to complete each FutureVar parameter. + for ident in futureVarIdents: + # Only complete them if they have not been completed already by the user. + result.add newIfStmt( + ( + newCall(newIdentNode("not"), + newDotExpr(ident, newIdentNode("finished"))), + newCall(newIdentNode("complete"), ident) + ) + ) + proc processBody(node, retFutureSym: NimNode, - subTypeIsVoid: bool, + subTypeIsVoid: bool, futureVarIdents: seq[NimNode], tryStmt: NimNode): NimNode {.compileTime.} = #echo(node.treeRepr) result = node @@ -134,11 +150,14 @@ proc processBody(node, retFutureSym: NimNode, else: result.add newCall(newIdentNode("complete"), retFutureSym) else: - let x = node[0].processBody(retFutureSym, subTypeIsVoid, tryStmt) + let x = node[0].processBody(retFutureSym, subTypeIsVoid, + futureVarIdents, tryStmt) if x.kind == nnkYieldStmt: result.add x else: result.add newCall(newIdentNode("complete"), retFutureSym, x) + result.add createFutureVarCompletions(futureVarIdents) + result.add newNimNode(nnkReturnStmt, node).add(newNilLit()) return # Don't process the children of this return stmt of nnkCommand, nnkCall: @@ -196,7 +215,8 @@ proc processBody(node, retFutureSym: NimNode, # Transform ``except`` body. # TODO: Could we perform some ``await`` transformation here to get it # working in ``except``? - tryBody[1] = processBody(n[1], retFutureSym, subTypeIsVoid, nil) + tryBody[1] = processBody(n[1], retFutureSym, subTypeIsVoid, + futureVarIdents, nil) proc processForTry(n: NimNode, i: var int, res: NimNode): bool {.compileTime.} = @@ -207,7 +227,7 @@ proc processBody(node, retFutureSym: NimNode, var skipped = n.skipStmtList() while i < skipped.len: var processed = processBody(skipped[i], retFutureSym, - subTypeIsVoid, n) + subTypeIsVoid, futureVarIdents, n) # Check if we transformed the node into an exception check. # This suggests skipped[i] contains ``await``. @@ -239,7 +259,8 @@ proc processBody(node, retFutureSym: NimNode, else: discard for i in 0 .. <result.len: - result[i] = processBody(result[i], retFutureSym, subTypeIsVoid, nil) + result[i] = processBody(result[i], retFutureSym, subTypeIsVoid, + futureVarIdents, nil) proc getName(node: NimNode): string {.compileTime.} = case node.kind @@ -252,6 +273,14 @@ proc getName(node: NimNode): string {.compileTime.} = else: error("Unknown name.") +proc getFutureVarIdents(params: NimNode): seq[NimNode] {.compileTime.} = + result = @[] + for i in 1 .. <len(params): + expectKind(params[i], nnkIdentDefs) + if params[i][1].kind == nnkBracketExpr and + ($params[i][1][0].ident).normalize == "futurevar": + result.add(params[i][0]) + proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} = ## This macro transforms a single procedure into a closure iterator. ## The ``async`` macro supports a stmtList holding multiple async procedures. @@ -282,6 +311,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} = let subtypeIsVoid = returnType.kind == nnkEmpty or (baseType.kind == nnkIdent and returnType[1].ident == !"void") + let futureVarIdents = getFutureVarIdents(prc[3]) + var outerProcBody = newNimNode(nnkStmtList, prc[6]) # -> var retFuture = newFuture[T]() @@ -304,7 +335,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} = # -> <proc_body> # -> complete(retFuture, result) var iteratorNameSym = genSym(nskIterator, $prc[0].getName & "Iter") - var procBody = prc[6].processBody(retFutureSym, subtypeIsVoid, nil) + var procBody = prc[6].processBody(retFutureSym, subtypeIsVoid, + futureVarIdents, nil) # don't do anything with forward bodies (empty) if procBody.kind != nnkEmpty: if not subtypeIsVoid: @@ -326,6 +358,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} = # -> complete(retFuture) procBody.add(newCall(newIdentNode("complete"), retFutureSym)) + procBody.add(createFutureVarCompletions(futureVarIdents)) + var closureIterator = newProc(iteratorNameSym, [newIdentNode("FutureBase")], procBody, nnkIteratorDef) closureIterator[4] = newNimNode(nnkPragma, prc[6]).add(newIdentNode("closure")) @@ -334,7 +368,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} = # -> createCb(retFuture) #var cbName = newIdentNode("cb") var procCb = getAst createCb(retFutureSym, iteratorNameSym, - newStrLitNode(prc[0].getName)) + newStrLitNode(prc[0].getName), + createFutureVarCompletions(futureVarIdents)) outerProcBody.add procCb # -> return retFuture diff --git a/lib/pure/asyncnet.nim b/lib/pure/asyncnet.nim index 14ebde4a2..3b64c278f 100644 --- a/lib/pure/asyncnet.nim +++ b/lib/pure/asyncnet.nim @@ -159,7 +159,9 @@ when defineSsl: await socket.fd.AsyncFd.send(data, flags) proc appeaseSsl(socket: AsyncSocket, flags: set[SocketFlag], - sslError: cint) {.async.} = + sslError: cint): Future[bool] {.async.} = + ## Returns ``true`` if ``socket`` is still connected, otherwise ``false``. + result = true case sslError of SSL_ERROR_WANT_WRITE: await sendPendingSslData(socket, flags) @@ -173,6 +175,7 @@ when defineSsl: elif length == 0: # connection not properly closed by remote side or connection dropped SSL_set_shutdown(socket.sslHandle, SSL_RECEIVED_SHUTDOWN) + result = false else: raiseSSLError("Cannot appease SSL.") @@ -180,13 +183,27 @@ when defineSsl: op: expr) = var opResult {.inject.} = -1.cint while opResult < 0: + # Call the desired operation. opResult = op # Bit hackish here. # TODO: Introduce an async template transformation pragma? + + # Send any remaining pending SSL data. yield sendPendingSslData(socket, flags) + + # If the operation failed, try to see if SSL has some data to read + # or write. if opResult < 0: let err = getSslError(socket.sslHandle, opResult.cint) - yield appeaseSsl(socket, flags, err.cint) + let fut = appeaseSsl(socket, flags, err.cint) + yield fut + if not fut.read(): + # Socket disconnected. + if SocketFlag.SafeDisconn in flags: + break + else: + raiseSSLError("Socket has been disconnected") + proc connect*(socket: AsyncSocket, address: string, port: Port) {.async.} = ## Connects ``socket`` to server at ``address:port``. diff --git a/lib/pure/basic2d.nim b/lib/pure/basic2d.nim index 7d74424fa..e4696c6a8 100644 --- a/lib/pure/basic2d.nim +++ b/lib/pure/basic2d.nim @@ -117,13 +117,13 @@ proc safeArccos(v:float):float= template makeBinOpVector(s:expr)= - ## implements binary operators + , - , * and / for vectors + ## implements binary operators ``+``, ``-``, ``*`` and ``/`` for vectors proc s*(a,b:Vector2d):Vector2d {.inline,noInit.} = vector2d(s(a.x,b.x),s(a.y,b.y)) proc s*(a:Vector2d,b:float):Vector2d {.inline,noInit.} = vector2d(s(a.x,b),s(a.y,b)) proc s*(a:float,b:Vector2d):Vector2d {.inline,noInit.} = vector2d(s(a,b.x),s(a,b.y)) template makeBinOpAssignVector(s:expr)= - ## implements inplace binary operators += , -= , /= and *= for vectors + ## implements inplace binary operators ``+=``, ``-=``, ``/=`` and ``*=`` for vectors proc s*(a:var Vector2d,b:Vector2d) {.inline.} = s(a.x,b.x) ; s(a.y,b.y) proc s*(a:var Vector2d,b:float) {.inline.} = s(a.x,b) ; s(a.y,b) diff --git a/lib/pure/basic3d.nim b/lib/pure/basic3d.nim index 424c191f8..f7a9c237c 100644 --- a/lib/pure/basic3d.nim +++ b/lib/pure/basic3d.nim @@ -117,7 +117,6 @@ proc safeArccos(v:float):float= return arccos(clamp(v,-1.0,1.0)) template makeBinOpVector(s:expr)= - ## implements binary operators + , - , * and / for vectors proc s*(a,b:Vector3d):Vector3d {.inline,noInit.} = vector3d(s(a.x,b.x),s(a.y,b.y),s(a.z,b.z)) proc s*(a:Vector3d,b:float):Vector3d {.inline,noInit.} = @@ -126,11 +125,10 @@ template makeBinOpVector(s:expr)= vector3d(s(a,b.x),s(a,b.y),s(a,b.z)) template makeBinOpAssignVector(s:expr)= - ## implements inplace binary operators += , -= , /= and *= for vectors proc s*(a:var Vector3d,b:Vector3d) {.inline.} = - s(a.x,b.x) ; s(a.y,b.y) ; s(a.z,b.z) + s(a.x,b.x); s(a.y,b.y); s(a.z,b.z) proc s*(a:var Vector3d,b:float) {.inline.} = - s(a.x,b) ; s(a.y,b) ; s(a.z,b) + s(a.x,b); s(a.y,b); s(a.z,b) diff --git a/lib/pure/includes/asyncfutures.nim b/lib/pure/includes/asyncfutures.nim new file mode 100644 index 000000000..fda78c1a1 --- /dev/null +++ b/lib/pure/includes/asyncfutures.nim @@ -0,0 +1,293 @@ + +# TODO: This shouldn't need to be included, but should ideally be exported. +type + FutureBase* = ref object of RootObj ## Untyped future. + cb: proc () {.closure,gcsafe.} + finished: bool + error*: ref Exception ## Stored exception + errorStackTrace*: string + when not defined(release): + stackTrace: string ## For debugging purposes only. + id: int + fromProc: string + + Future*[T] = ref object of FutureBase ## Typed future. + value: T ## Stored value + + FutureVar*[T] = distinct Future[T] + + FutureError* = object of Exception + cause*: FutureBase + +{.deprecated: [PFutureBase: FutureBase, PFuture: Future].} + +when not defined(release): + var currentID = 0 + +proc callSoon*(cbproc: proc ()) {.gcsafe.} + +proc newFuture*[T](fromProc: string = "unspecified"): Future[T] = + ## Creates a new future. + ## + ## Specifying ``fromProc``, which is a string specifying the name of the proc + ## that this future belongs to, is a good habit as it helps with debugging. + new(result) + result.finished = false + when not defined(release): + result.stackTrace = getStackTrace() + result.id = currentID + result.fromProc = fromProc + currentID.inc() + +proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] = + ## Create a new ``FutureVar``. This Future type is ideally suited for + ## situations where you want to avoid unnecessary allocations of Futures. + ## + ## Specifying ``fromProc``, which is a string specifying the name of the proc + ## that this future belongs to, is a good habit as it helps with debugging. + result = FutureVar[T](newFuture[T](fromProc)) + +proc clean*[T](future: FutureVar[T]) = + ## Resets the ``finished`` status of ``future``. + Future[T](future).finished = false + Future[T](future).error = nil + +proc checkFinished[T](future: Future[T]) = + ## Checks whether `future` is finished. If it is then raises a + ## ``FutureError``. + when not defined(release): + if future.finished: + var msg = "" + msg.add("An attempt was made to complete a Future more than once. ") + msg.add("Details:") + msg.add("\n Future ID: " & $future.id) + msg.add("\n Created in proc: " & future.fromProc) + msg.add("\n Stack trace to moment of creation:") + msg.add("\n" & indent(future.stackTrace.strip(), 4)) + when T is string: + msg.add("\n Contents (string): ") + msg.add("\n" & indent(future.value.repr, 4)) + msg.add("\n Stack trace to moment of secondary completion:") + msg.add("\n" & indent(getStackTrace().strip(), 4)) + var err = newException(FutureError, msg) + err.cause = future + raise err + +proc complete*[T](future: Future[T], val: T) = + ## Completes ``future`` with value ``val``. + #assert(not future.finished, "Future already finished, cannot finish twice.") + checkFinished(future) + assert(future.error == nil) + future.value = val + future.finished = true + if future.cb != nil: + future.cb() + +proc complete*(future: Future[void]) = + ## Completes a void ``future``. + #assert(not future.finished, "Future already finished, cannot finish twice.") + checkFinished(future) + assert(future.error == nil) + future.finished = true + if future.cb != nil: + future.cb() + +proc complete*[T](future: FutureVar[T]) = + ## Completes a ``FutureVar``. + template fut: expr = Future[T](future) + checkFinished(fut) + assert(fut.error == nil) + fut.finished = true + if fut.cb != nil: + fut.cb() + +proc complete*[T](future: FutureVar[T], val: T) = + ## Completes a ``FutureVar`` with value ``val``. + ## + ## Any previously stored value will be overwritten. + template fut: expr = Future[T](future) + checkFinished(fut) + assert(fut.error == nil) + fut.finished = true + fut.value = val + if fut.cb != nil: + fut.cb() + +proc fail*[T](future: Future[T], error: ref Exception) = + ## Completes ``future`` with ``error``. + #assert(not future.finished, "Future already finished, cannot finish twice.") + checkFinished(future) + future.finished = true + future.error = error + future.errorStackTrace = + if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error) + if future.cb != nil: + future.cb() + else: + # This is to prevent exceptions from being silently ignored when a future + # is discarded. + # TODO: This may turn out to be a bad idea. + # Turns out this is a bad idea. + #raise error + discard + +proc `callback=`*(future: FutureBase, cb: proc () {.closure,gcsafe.}) = + ## Sets the callback proc to be called when the future completes. + ## + ## If future has already completed then ``cb`` will be called immediately. + ## + ## **Note**: You most likely want the other ``callback`` setter which + ## passes ``future`` as a param to the callback. + future.cb = cb + if future.finished: + callSoon(future.cb) + +proc `callback=`*[T](future: Future[T], + cb: proc (future: Future[T]) {.closure,gcsafe.}) = + ## Sets the callback proc to be called when the future completes. + ## + ## If future has already completed then ``cb`` will be called immediately. + future.callback = proc () = cb(future) + +proc injectStacktrace[T](future: Future[T]) = + # TODO: Come up with something better. + when not defined(release): + var msg = "" + msg.add("\n " & future.fromProc & "'s lead up to read of failed Future:") + + if not future.errorStackTrace.isNil and future.errorStackTrace != "": + msg.add("\n" & indent(future.errorStackTrace.strip(), 4)) + else: + msg.add("\n Empty or nil stack trace.") + future.error.msg.add(msg) + +proc read*[T](future: Future[T] | FutureVar[T]): T = + ## Retrieves the value of ``future``. Future must be finished otherwise + ## this function will fail with a ``ValueError`` exception. + ## + ## If the result of the future is an error then that error will be raised. + let fut = Future[T](future) + if fut.finished: + if fut.error != nil: + injectStacktrace(fut) + raise fut.error + when T isnot void: + return fut.value + else: + # TODO: Make a custom exception type for this? + raise newException(ValueError, "Future still in progress.") + +proc readError*[T](future: Future[T]): ref Exception = + ## Retrieves the exception stored in ``future``. + ## + ## An ``ValueError`` exception will be thrown if no exception exists + ## in the specified Future. + if future.error != nil: return future.error + else: + raise newException(ValueError, "No error in future.") + +proc mget*[T](future: FutureVar[T]): var T = + ## Returns a mutable value stored in ``future``. + ## + ## Unlike ``read``, this function will not raise an exception if the + ## Future has not been finished. + result = Future[T](future).value + +proc finished*[T](future: Future[T] | FutureVar[T]): bool = + ## Determines whether ``future`` has completed. + ## + ## ``True`` may indicate an error or a value. Use ``failed`` to distinguish. + (Future[T](future)).finished + +proc failed*(future: FutureBase): bool = + ## Determines whether ``future`` completed with an error. + return future.error != nil + +proc asyncCheck*[T](future: Future[T]) = + ## Sets a callback on ``future`` which raises an exception if the future + ## finished with an error. + ## + ## This should be used instead of ``discard`` to discard void futures. + future.callback = + proc () = + if future.failed: + injectStacktrace(future) + raise future.error + +proc `and`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] = + ## Returns a future which will complete once both ``fut1`` and ``fut2`` + ## complete. + var retFuture = newFuture[void]("asyncdispatch.`and`") + fut1.callback = + proc () = + if not retFuture.finished: + if fut1.failed: retFuture.fail(fut1.error) + elif fut2.finished: retFuture.complete() + fut2.callback = + proc () = + if not retFuture.finished: + if fut2.failed: retFuture.fail(fut2.error) + elif fut1.finished: retFuture.complete() + return retFuture + +proc `or`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] = + ## Returns a future which will complete once either ``fut1`` or ``fut2`` + ## complete. + var retFuture = newFuture[void]("asyncdispatch.`or`") + proc cb[X](fut: Future[X]) = + if fut.failed: retFuture.fail(fut.error) + if not retFuture.finished: retFuture.complete() + fut1.callback = cb[T] + fut2.callback = cb[Y] + return retFuture + +proc all*[T](futs: varargs[Future[T]]): auto = + ## Returns a future which will complete once + ## all futures in ``futs`` complete. + ## + ## If the awaited futures are not ``Future[void]``, the returned future + ## will hold the values of all awaited futures in a sequence. + ## + ## If the awaited futures *are* ``Future[void]``, + ## this proc returns ``Future[void]``. + + when T is void: + var + retFuture = newFuture[void]("asyncdispatch.all") + completedFutures = 0 + + let totalFutures = len(futs) + + for fut in futs: + fut.callback = proc(f: Future[T]) = + if f.failed: + retFuture.fail(f.error) + elif not retFuture.finished: + inc(completedFutures) + + if completedFutures == totalFutures: + retFuture.complete() + + return retFuture + + else: + var + retFuture = newFuture[seq[T]]("asyncdispatch.all") + retValues = newSeq[T](len(futs)) + completedFutures = 0 + + for i, fut in futs: + proc setCallback(i: int) = + fut.callback = proc(f: Future[T]) = + if f.failed: + retFuture.fail(f.error) + elif not retFuture.finished: + retValues[i] = f.read() + inc(completedFutures) + + if completedFutures == len(retValues): + retFuture.complete(retValues) + + setCallback(i) + + return retFuture diff --git a/lib/pure/selectors.nim b/lib/pure/selectors.nim index cba101fff..506b2cec0 100644 --- a/lib/pure/selectors.nim +++ b/lib/pure/selectors.nim @@ -377,7 +377,8 @@ proc contains*(s: Selector, key: SelectorKey): bool = proc len*(s: Selector): int = ## Retrieves the number of registered file descriptors in this Selector. - return s.fds.len + when not defined(nimdoc): + return s.fds.len {.deprecated: [TEvent: Event, PSelectorKey: SelectorKey, TReadyInfo: ReadyInfo, PSelector: Selector].} |