logos-storage/logos-storage-nim
1
0
Fork 0
mirror of https://github.com/logos-storage/logos-storage-nim.git synced 2026-01-04 06:23:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cancel and wait for asyncstatemachine futures when stopping (#493)

Browse Source 
* Simplify `.then` (promise api) and tests

* Remove tracked future when cancelled. Add tracked future tests

* Track and cancel statemachine futures

The futures created in each asyncstatemachine instance are tracked, and each future is cancelled and waited in `stop`.

Change `asyncstatemachine.stop` to be async so `machine.trackedFutures.cancelAndWait` could be called.
Add a constructor for `asyncstatemachine` that initialises the `trackedFutures` instance, and call the constructor from derived class constructors.
This commit is contained in:
Eric 2023-07-31 15:09:34 +10:00 committed by GitHub
parent 1d161d383e
commit 3e80de3454
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 536 additions and 403 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
codex/purchasing/purchase.nim
View File

@ -31,13 +31,14 @@ func new*(
clock: Clock clock: Clock
): Purchase = ): Purchase =
## create a new instance of a Purchase ## create a new instance of a Purchase
## ##
Purchase( var purchase = Purchase.new()
future: Future[void].new(), purchase.future = Future[void].new()
requestId: requestId, purchase.requestId = requestId
market: market, purchase.market = market
clock: clock purchase.clock = clock
)
return purchase
func new*( func new*(
_: type Purchase, _: type Purchase,

2
codex/sales.nim
View File

@ -15,10 +15,10 @@ import ./sales/salescontext
import ./sales/salesagent import ./sales/salesagent
import ./sales/statemachine import ./sales/statemachine
import ./sales/slotqueue import ./sales/slotqueue
import ./sales/trackedfutures
import ./sales/states/preparing import ./sales/states/preparing
import ./sales/states/unknown import ./sales/states/unknown
import ./utils/then import ./utils/then
import ./utils/trackedfutures
## Sales holds a list of available storage that it may sell. ## Sales holds a list of available storage that it may sell.
## ##

15
codex/sales/salesagent.nim
View File

@ -32,12 +32,13 @@ proc newSalesAgent*(context: SalesContext,
requestId: RequestId, requestId: RequestId,
slotIndex: UInt256, slotIndex: UInt256,
request: ?StorageRequest): SalesAgent = request: ?StorageRequest): SalesAgent =
SalesAgent( var agent = SalesAgent.new()
context: context, agent.context = context
data: SalesData( agent.data = SalesData(
requestId: requestId, requestId: requestId,
slotIndex: slotIndex, slotIndex: slotIndex,
request: request)) request: request)
return agent
proc retrieveRequest*(agent: SalesAgent) {.async.} = proc retrieveRequest*(agent: SalesAgent) {.async.} =
let data = agent.data let data = agent.data
@ -96,5 +97,5 @@ proc unsubscribe*(agent: SalesAgent) {.async.} =
agent.subscribed = false agent.subscribed = false
proc stop*(agent: SalesAgent) {.async.} = proc stop*(agent: SalesAgent) {.async.} =
procCall Machine(agent).stop() await Machine(agent).stop()
await agent.unsubscribe() await agent.unsubscribe()

3
codex/sales/slotqueue.nim
View File

@ -1,5 +1,4 @@
import std/sequtils import std/sequtils
import std/sugar
import std/tables import std/tables
import pkg/chronicles import pkg/chronicles
import pkg/chronos import pkg/chronos
@ -7,13 +6,13 @@ import pkg/questionable
import pkg/questionable/results import pkg/questionable/results
import pkg/upraises import pkg/upraises
import ./reservations import ./reservations
import ./trackedfutures
import ../errors import ../errors
import ../rng import ../rng
import ../utils import ../utils
import ../contracts/requests import ../contracts/requests
import ../utils/asyncheapqueue import ../utils/asyncheapqueue
import ../utils/then import ../utils/then
import ../utils/trackedfutures
logScope: logScope:
topics = "marketplace slotqueue" topics = "marketplace slotqueue"

46
codex/utils/asyncstatemachine.nim
View File

@ -1,7 +1,10 @@
import std/sugar
import pkg/questionable import pkg/questionable
import pkg/chronos import pkg/chronos
import pkg/chronicles import pkg/chronicles
import pkg/upraises import pkg/upraises
import ./trackedfutures
import ./then
push: {.upraises:[].} push: {.upraises:[].}
@ -10,8 +13,8 @@ type
state: State state: State
running: Future[void] running: Future[void]
scheduled: AsyncQueue[Event] scheduled: AsyncQueue[Event]
scheduling: Future[void]
started: bool started: bool
trackedFutures: TrackedFutures
State* = ref object of RootObj State* = ref object of RootObj
Query*[T] = proc(state: State): T Query*[T] = proc(state: State): T
Event* = proc(state: State): ?State {.gcsafe, upraises:[].} Event* = proc(state: State): ?State {.gcsafe, upraises:[].}
@ -19,6 +22,9 @@ type
logScope: logScope:
topics = "statemachine" topics = "statemachine"
proc new*[T: Machine](_: type T): T =
T(trackedFutures: TrackedFutures.new())
method `$`*(state: State): string {.base.} = method `$`*(state: State): string {.base.} =
raiseAssert "not implemented" raiseAssert "not implemented"
@ -60,21 +66,21 @@ proc run(machine: Machine, state: State) {.async.} =
discard discard
proc scheduler(machine: Machine) {.async.} = proc scheduler(machine: Machine) {.async.} =
proc onRunComplete(udata: pointer) {.gcsafe.} = var running: Future[void]
var fut = cast[FutureBase](udata)
if fut.failed():
machine.schedule(machine.onError(fut.error))
try: try:
while true: while machine.started:
let event = await machine.scheduled.get() let event = await machine.scheduled.get().track(machine)
if next =? event(machine.state): if next =? event(machine.state):
if not machine.running.isNil: if not running.isNil and not running.finished:
await machine.running.cancelAndWait() await running.cancelAndWait()
machine.state = next machine.state = next
debug "enter state", state = machine.state debug "enter state", state = machine.state
machine.running = machine.run(machine.state) running = machine.run(machine.state)
machine.running.addCallback(onRunComplete) running
.track(machine)
.catch((err: ref CatchableError) =>
machine.schedule(machine.onError(err))
)
except CancelledError: except CancelledError:
discard discard
@ -84,18 +90,20 @@ proc start*(machine: Machine, initialState: State) =
if machine.scheduled.isNil: if machine.scheduled.isNil:
machine.scheduled = newAsyncQueue[Event]() machine.scheduled = newAsyncQueue[Event]()
machine.scheduling = machine.scheduler()
machine.started = true machine.started = true
machine.scheduler()
.track(machine)
.catch((err: ref CatchableError) =>
error("Error in scheduler", error = err.msg)
)
machine.schedule(Event.transition(machine.state, initialState)) machine.schedule(Event.transition(machine.state, initialState))
proc stop*(machine: Machine) = proc stop*(machine: Machine) {.async.} =
if not machine.started: if not machine.started:
return return
if not machine.scheduling.isNil: machine.started = false
machine.scheduling.cancel() await machine.trackedFutures.cancelTracked()
if not machine.running.isNil:
machine.running.cancel()
machine.state = nil machine.state = nil
machine.started = false

137
codex/utils/then.nim
View File

@ -22,12 +22,11 @@ import pkg/upraises
# `.catch` is called when the `Future` fails. In the case when the `Future` # `.catch` is called when the `Future` fails. In the case when the `Future`
# returns a `Result[T, ref CatchableError` (or `?!T`), `.catch` will be called # returns a `Result[T, ref CatchableError` (or `?!T`), `.catch` will be called
# if the `Result` contains an error. If the `Future` is already failed (or # if the `Result` contains an error. If the `Future` is already failed (or
# `Future[?!T]` contains an error), the `.catch` callback will be excuted # `Future[?!T]` contains an error), the `.catch` callback will be executed
# immediately. # immediately.
# NOTE: Cancelled `Futures` are discarded as bubbling the `CancelledError` to # `.cancelled` is called when the `Future` is cancelled. If the `Future` is
# the synchronous closure will likely cause an unintended and unhandled # already cancelled, the `.cancelled` callback will be executed immediately.
# exception.
# More info on JavaScript's Promise API can be found at: # More info on JavaScript's Promise API can be found at:
# https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise # https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise
@ -56,44 +55,30 @@ runnableExamples:
type type
OnSuccess*[T] = proc(val: T) {.gcsafe, upraises: [].} OnSuccess*[T] = proc(val: T) {.gcsafe, upraises: [].}
OnError* = proc(err: ref CatchableError) {.gcsafe, upraises: [].} OnError* = proc(err: ref CatchableError) {.gcsafe, upraises: [].}
OnCancelled* = proc() {.gcsafe, upraises: [].}
proc ignoreError(err: ref CatchableError) = discard proc ignoreError(err: ref CatchableError) = discard
proc ignoreCancelled() = discard
template handleFinished(future: FutureBase,
onError: OnError,
onCancelled: OnCancelled) =
template returnOrError(future: FutureBase, onError: OnError) =
if not future.finished: if not future.finished:
return return
if future.cancelled: if future.cancelled:
# do not bubble as closure is synchronous onCancelled()
return return
if future.failed: if future.failed:
onError(future.error) onError(future.error)
return return
proc then*(future: Future[void], onSuccess: OnSuccess[void]): Future[void] =
proc then*(future: Future[void],
onError: OnError):
Future[void] =
proc cb(udata: pointer) = proc cb(udata: pointer) =
future.returnOrError(onError) future.handleFinished(ignoreError, ignoreCancelled)
proc cancellation(udata: pointer) =
if not future.finished():
future.removeCallback(cb)
future.addCallback(cb)
future.cancelCallback = cancellation
return future
proc then*(future: Future[void],
onSuccess: OnSuccess[void],
onError: OnError = ignoreError):
Future[void] =
proc cb(udata: pointer) =
future.returnOrError(onError)
onSuccess() onSuccess()
proc cancellation(udata: pointer) = proc cancellation(udata: pointer) =
@ -104,42 +89,13 @@ proc then*(future: Future[void],
future.cancelCallback = cancellation future.cancelCallback = cancellation
return future return future
proc then*[T](future: Future[T], proc then*[T](future: Future[T], onSuccess: OnSuccess[T]): Future[T] =
onSuccess: OnSuccess[T],
onError: OnError = ignoreError):
Future[T] =
proc cb(udata: pointer) = proc cb(udata: pointer) =
future.returnOrError(onError) future.handleFinished(ignoreError, ignoreCancelled)
without val =? future.read.catch, err: if val =? future.read.catch:
onError(err)
return
onSuccess(val)
proc cancellation(udata: pointer) =
if not future.finished():
future.removeCallback(cb)
future.addCallback(cb)
future.cancelCallback = cancellation
return future
proc then*[T](future: Future[?!T],
onSuccess: OnSuccess[T],
onError: OnError = ignoreError):
Future[?!T] =
proc cb(udata: pointer) =
future.returnOrError(onError)
try:
without val =? future.read, err:
onError(err)
return
onSuccess(val) onSuccess(val)
except CatchableError as e:
onError(e)
proc cancellation(udata: pointer) = proc cancellation(udata: pointer) =
if not future.finished(): if not future.finished():
@ -149,18 +105,16 @@ proc then*[T](future: Future[?!T],
future.cancelCallback = cancellation future.cancelCallback = cancellation
return future return future
proc then*(future: Future[?!void], proc then*[T](future: Future[?!T], onSuccess: OnSuccess[T]): Future[?!T] =
onError: OnError = ignoreError):
Future[?!void] =
proc cb(udata: pointer) = proc cb(udata: pointer) =
future.returnOrError(onError) future.handleFinished(ignoreError, ignoreCancelled)
try: try:
if err =? future.read.errorOption: if val =? future.read:
onError(err) onSuccess(val)
except CatchableError as e: except CatchableError as e:
onError(e) ignoreError(e)
proc cancellation(udata: pointer) = proc cancellation(udata: pointer) =
if not future.finished(): if not future.finished():
@ -170,22 +124,17 @@ proc then*(future: Future[?!void],
future.cancelCallback = cancellation future.cancelCallback = cancellation
return future return future
proc then*(future: Future[?!void], proc then*(future: Future[?!void], onSuccess: OnSuccess[void]): Future[?!void] =
onSuccess: OnSuccess[void],
onError: OnError = ignoreError):
Future[?!void] =
proc cb(udata: pointer) = proc cb(udata: pointer) =
future.returnOrError(onError) future.handleFinished(ignoreError, ignoreCancelled)
try: try:
if err =? future.read.errorOption: if future.read.isOk:
onError(err) onSuccess()
return
except CatchableError as e: except CatchableError as e:
onError(e) ignoreError(e)
return return
onSuccess()
proc cancellation(udata: pointer) = proc cancellation(udata: pointer) =
if not future.finished(): if not future.finished():
@ -197,8 +146,10 @@ proc then*(future: Future[?!void],
proc catch*[T](future: Future[T], onError: OnError) = proc catch*[T](future: Future[T], onError: OnError) =
if future.isNil: return
proc cb(udata: pointer) = proc cb(udata: pointer) =
future.returnOrError(onError) future.handleFinished(onError, ignoreCancelled)
proc cancellation(udata: pointer) = proc cancellation(udata: pointer) =
if not future.finished(): if not future.finished():
@ -209,8 +160,10 @@ proc catch*[T](future: Future[T], onError: OnError) =
proc catch*[T](future: Future[?!T], onError: OnError) = proc catch*[T](future: Future[?!T], onError: OnError) =
if future.isNil: return
proc cb(udata: pointer) = proc cb(udata: pointer) =
future.returnOrError(onError) future.handleFinished(onError, ignoreCancelled)
try: try:
if err =? future.read.errorOption: if err =? future.read.errorOption:
@ -224,3 +177,31 @@ proc catch*[T](future: Future[?!T], onError: OnError) =
future.addCallback(cb) future.addCallback(cb)
future.cancelCallback = cancellation future.cancelCallback = cancellation
proc cancelled*[T](future: Future[T], onCancelled: OnCancelled): Future[T] =
proc cb(udata: pointer) =
future.handleFinished(ignoreError, onCancelled)
proc cancellation(udata: pointer) =
if not future.finished():
future.removeCallback(cb)
onCancelled()
future.addCallback(cb)
future.cancelCallback = cancellation
return future
proc cancelled*[T](future: Future[?!T], onCancelled: OnCancelled): Future[?!T] =
proc cb(udata: pointer) =
future.handleFinished(ignoreError, onCancelled)
proc cancellation(udata: pointer) =
if not future.finished():
future.removeCallback(cb)
onCancelled()
future.addCallback(cb)
future.cancelCallback = cancellation
return future

27
codex/sales/trackedfutures.nim → codex/utils/trackedfutures.nim
View File

@ -12,21 +12,25 @@ type
logScope: logScope:
topics = "trackable futures" topics = "trackable futures"
proc track*[T](self: TrackedFutures, fut: Future[T]): Future[T] = proc len*(self: TrackedFutures): int = self.futures.len
logScope:
id = fut.id
proc removeFuture() = proc removeFuture(self: TrackedFutures, future: FutureBase) =
if not self.cancelling and not fut.isNil: if not self.cancelling and not future.isNil:
trace "removing tracked future" trace "removing tracked future"
self.futures.del(fut.id) self.futures.del(future.id)
proc track*[T](self: TrackedFutures, fut: Future[T]): Future[T] =
if self.cancelling:
return fut
trace "tracking future", id = fut.id
self.futures[fut.id] = FutureBase(fut)
fut fut
.then((val: T) => removeFuture()) .then((val: T) => self.removeFuture(fut))
.catch((e: ref CatchableError) => removeFuture()) .cancelled(() => self.removeFuture(fut))
.catch((e: ref CatchableError) => self.removeFuture(fut))
trace "tracking future"
self.futures[fut.id] = FutureBase(fut)
return fut return fut
proc track*[T, U](future: Future[T], self: U): Future[T] = proc track*[T, U](future: Future[T], self: U): Future[T] =
@ -43,4 +47,5 @@ proc cancelTracked*(self: TrackedFutures) {.async.} =
trace "cancelling tracked future", id = future.id trace "cancelling tracked future", id = future.id
await future.cancelAndWait() await future.cancelAndWait()
self.futures.clear()
self.cancelling = false self.cancelling = false

1
tests/codex/sales/testsalesagent.nim
View File

@ -10,6 +10,7 @@ import pkg/codex/proving
import ../helpers/mockmarket import ../helpers/mockmarket
import ../helpers/mockclock import ../helpers/mockclock
import ../helpers/eventually import ../helpers/eventually
import ../helpers
import ../examples import ../examples
var onCancelCalled = false var onCancelCalled = false

1
tests/codex/testutils.nim
View File

@ -3,5 +3,6 @@ import ./utils/testkeyutils
import ./utils/testasyncstatemachine import ./utils/testasyncstatemachine
import ./utils/testtimer import ./utils/testtimer
import ./utils/testthen import ./utils/testthen
import ./utils/testtrackedfutures
{.warning[UnusedImport]: off.} {.warning[UnusedImport]: off.}

4
tests/codex/utils/testasyncstatemachine.nim
View File

@ -99,7 +99,7 @@ asyncchecksuite "async state machines":
test "stops scheduling and current state": test "stops scheduling and current state":
machine.start(State2.new()) machine.start(State2.new())
await sleepAsync(1.millis) await sleepAsync(1.millis)
machine.stop() await machine.stop()
machine.schedule(moveToNextStateEvent) machine.schedule(moveToNextStateEvent)
await sleepAsync(1.millis) await sleepAsync(1.millis)
check runs == [0, 1, 0, 0] check runs == [0, 1, 0, 0]
@ -130,5 +130,5 @@ asyncchecksuite "async state machines":
machine.start(State2.new()) machine.start(State2.new())
check eventually machine.query(description).isSome check eventually machine.query(description).isSome
machine.stop() await machine.stop()
check machine.query(description).isNone check machine.query(description).isNone

621
tests/codex/utils/testthen.nim
View File

@ -5,340 +5,409 @@ import pkg/questionable/results
import codex/utils/then import codex/utils/then
import ../helpers import ../helpers
proc newError(): ref CatchableError =
(ref CatchableError)(msg: "some error")
asyncchecksuite "then - Future[void]": asyncchecksuite "then - Future[void]":
var returnsVoidWasRun: bool var error = newError()
var error = (ref CatchableError)(msg: "some error") var future: Future[void]
setup: setup:
returnsVoidWasRun = false future = newFuture[void]("test void")
proc returnsVoid() {.async.} = teardown:
await sleepAsync 1.millis if not future.finished:
returnsVoidWasRun = true raiseAssert "test should finish future"
proc returnsVoidError() {.async.} = test "then callback is fired when future is already finished":
raise error var firedImmediately = false
future.complete()
discard future.then(proc() = firedImmediately = true)
check eventually firedImmediately
proc returnsVoidCancelled() {.async.} = test "then callback is fired after future is finished":
await sleepAsync(1.seconds) var fired = false
discard future.then(proc() = fired = true)
future.complete()
check eventually fired
proc wasCancelled(error: ref CancelledError): bool = test "catch callback is fired when future is already failed":
not error.isNil and error.msg == "Future operation cancelled!" var actual: ref CatchableError
future.fail(error)
future.catch(proc(err: ref CatchableError) = actual = err)
check eventually actual == error
test "calls async proc when returns Future[void]": test "catch callback is fired after future is failed":
discard returnsVoid().then( var actual: ref CatchableError
proc(err: ref CatchableError) = discard future.catch(proc(err: ref CatchableError) = actual = err)
) future.fail(error)
check eventually returnsVoidWasRun check eventually actual == error
test "calls onSuccess when Future[void] complete": test "cancelled callback is fired when future is already cancelled":
var fired = false
await future.cancelAndWait()
discard future.cancelled(proc() = fired = true)
check eventually fired
test "cancelled callback is fired after future is cancelled":
var fired = false
discard future.cancelled(proc() = fired = true)
await future.cancelAndWait()
check eventually fired
test "does not fire other callbacks when successful":
var onSuccessCalled = false var onSuccessCalled = false
discard returnsVoid().then( var onCancelledCalled = false
proc() = onSuccessCalled = true, var onCatchCalled = false
proc(err: ref CatchableError) = discard
)
check eventually returnsVoidWasRun
check eventually onSuccessCalled
test "can pass only onSuccess for Future[void]": future
.then(proc() = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
future.complete()
check eventually onSuccessCalled
check always (not onCancelledCalled and not onCatchCalled)
test "does not fire other callbacks when fails":
var onSuccessCalled = false var onSuccessCalled = false
discard returnsVoid().then( var onCancelledCalled = false
proc() = onSuccessCalled = true var onCatchCalled = false
)
check eventually returnsVoidWasRun
check eventually onSuccessCalled
test "can chain onSuccess when Future[void] complete": future
.then(proc() = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
future.fail(error)
check eventually onCatchCalled
check always (not onCancelledCalled and not onSuccessCalled)
test "does not fire other callbacks when cancelled":
var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
future
.then(proc() = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
await future.cancelAndWait()
check eventually onCancelledCalled
check always (not onSuccessCalled and not onCatchCalled)
test "can chain onSuccess when future completes":
var onSuccessCalledTimes = 0 var onSuccessCalledTimes = 0
discard returnsVoid() discard future
.then(proc() = inc onSuccessCalledTimes) .then(proc() = inc onSuccessCalledTimes)
.then(proc() = inc onSuccessCalledTimes) .then(proc() = inc onSuccessCalledTimes)
.then(proc() = inc onSuccessCalledTimes) .then(proc() = inc onSuccessCalledTimes)
future.complete()
check eventually onSuccessCalledTimes == 3 check eventually onSuccessCalledTimes == 3
test "calls onError when Future[void] fails":
var errorActual: ref CatchableError
discard returnsVoidError().then(
proc() = discard,
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "calls onError when Future[void] fails":
var errorActual: ref CatchableError
discard returnsVoidError().then(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "catch callback fired when Future[void] fails":
var errorActual: ref CatchableError
returnsVoidError().catch(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "does not fire onSuccess callback when Future[void] fails":
var onSuccessCalled = false
returnsVoidError()
.then(proc() = onSuccessCalled = true)
.then(proc() = onSuccessCalled = true)
.catch(proc(e: ref CatchableError) = discard)
check always (not onSuccessCalled)
asyncchecksuite "then - Future[T]": asyncchecksuite "then - Future[T]":
var returnsValWasRun: bool var error = newError()
var error = (ref CatchableError)(msg: "some error") var future: Future[int]
setup: setup:
returnsValWasRun = false future = newFuture[int]("test void")
proc returnsVal(): Future[int] {.async.} = teardown:
await sleepAsync 1.millis if not future.finished:
returnsValWasRun = true raiseAssert "test should finish future"
return 1
proc returnsValError(): Future[int] {.async.} = test "then callback is fired when future is already finished":
raise error var cbVal = 0
future.complete(1)
discard future.then(proc(val: int) = cbVal = val)
check eventually cbVal == 1
proc returnsValCancelled(): Future[int] {.async.} = test "then callback is fired after future is finished":
await sleepAsync(1.seconds) var cbVal = 0
discard future.then(proc(val: int) = cbVal = val)
future.complete(1)
check eventually cbVal == 1
proc wasCancelled(error: ref CancelledError): bool = test "catch callback is fired when future is already failed":
not error.isNil and error.msg == "Future operation cancelled!" var actual: ref CatchableError
future.fail(error)
future.catch(proc(err: ref CatchableError) = actual = err)
check eventually actual == error
test "calls onSuccess when Future[T] complete": test "catch callback is fired after future is failed":
var returnedVal = 0 var actual: ref CatchableError
discard returnsVal().then( future.catch(proc(err: ref CatchableError) = actual = err)
proc(val: int) = returnedVal = val, future.fail(error)
proc(err: ref CatchableError) = discard check eventually actual == error
)
check eventually returnsValWasRun
check eventually returnedVal == 1
test "can pass only onSuccess for Future[T]": test "cancelled callback is fired when future is already cancelled":
var returnedVal = 0 var fired = false
discard returnsVal().then( await future.cancelAndWait()
proc(val: int) = returnedVal = val discard future.cancelled(proc() = fired = true)
) check eventually fired
check eventually returnsValWasRun
check eventually returnedVal == 1
test "can chain onSuccess when Future[T] complete": test "cancelled callback is fired after future is cancelled":
var onSuccessCalledWith: seq[int] = @[] var fired = false
discard returnsVal() discard future.cancelled(proc() = fired = true)
.then(proc(val: int) = onSuccessCalledWith.add(val)) await future.cancelAndWait()
.then(proc(val: int) = onSuccessCalledWith.add(val)) check eventually fired
.then(proc(val: int) = onSuccessCalledWith.add(val))
check eventually onSuccessCalledWith == @[1, 1, 1]
test "calls onError when Future[T] fails": test "does not fire other callbacks when successful":
var errorActual: ref CatchableError
discard returnsValError().then(
proc(val: int) = discard,
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "catch callback fired when Future[T] fails":
var errorActual: ref CatchableError
returnsValError().catch(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "does not fire onSuccess callback when Future[T] fails":
var onSuccessCalled = false var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
returnsValError() future
.then(proc(val: int) = onSuccessCalled = true) .then(proc(val: int) = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
future.complete(1)
check eventually onSuccessCalled
check always (not onCancelledCalled and not onCatchCalled)
test "does not fire other callbacks when fails":
var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
future
.then(proc(val: int) = onSuccessCalled = true) .then(proc(val: int) = onSuccessCalled = true)
.catch(proc(e: ref CatchableError) = discard) .cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
check always (not onSuccessCalled) future.fail(error)
asyncchecksuite "then - Future[?!void]": check eventually onCatchCalled
var returnsResultVoidWasRun: bool check always (not onCancelledCalled and not onSuccessCalled)
var error = (ref CatchableError)(msg: "some error")
setup: test "does not fire other callbacks when cancelled":
returnsResultVoidWasRun = false
proc returnsResultVoid(): Future[?!void] {.async.} =
await sleepAsync 1.millis
returnsResultVoidWasRun = true
return success()
proc returnsResultVoidError(): Future[?!void] {.async.} =
return failure(error)
proc returnsResultVoidErrorUncaught(): Future[?!void] {.async.} =
raise error
proc returnsResultVoidCancelled(): Future[?!void] {.async.} =
await sleepAsync(1.seconds)
return success()
proc wasCancelled(error: ref CancelledError): bool =
not error.isNil and error.msg == "Future operation cancelled!"
test "calls onSuccess when Future[?!void] complete":
var onSuccessCalled = false var onSuccessCalled = false
discard returnsResultVoid().then( var onCancelledCalled = false
proc() = onSuccessCalled = true, var onCatchCalled = false
proc(err: ref CatchableError) = discard
)
check eventually returnsResultVoidWasRun
check eventually onSuccessCalled
test "can pass only onSuccess for Future[?!void]": future
var onSuccessCalled = false .then(proc(val: int) = onSuccessCalled = true)
discard returnsResultVoid().then( .cancelled(proc() = onCancelledCalled = true)
proc() = onSuccessCalled = true .catch(proc(e: ref CatchableError) = onCatchCalled = true)
)
check eventually returnsResultVoidWasRun
check eventually onSuccessCalled
test "can chain onSuccess when Future[?!void] complete": await future.cancelAndWait()
check eventually onCancelledCalled
check always (not onSuccessCalled and not onCatchCalled)
test "can chain onSuccess when future completes":
var onSuccessCalledTimes = 0 var onSuccessCalledTimes = 0
discard returnsResultVoid() discard future
.then(proc() = inc onSuccessCalledTimes) .then(proc(val: int) = inc onSuccessCalledTimes)
.then(proc() = inc onSuccessCalledTimes) .then(proc(val: int) = inc onSuccessCalledTimes)
.then(proc() = inc onSuccessCalledTimes) .then(proc(val: int) = inc onSuccessCalledTimes)
future.complete(1)
check eventually onSuccessCalledTimes == 3 check eventually onSuccessCalledTimes == 3
test "calls onError when Future[?!void] fails": asyncchecksuite "then - Future[?!void]":
var errorActual: ref CatchableError var error = newError()
discard returnsResultVoidError().then( var future: Future[?!void]
proc() = discard,
proc(e: ref CatchableError) = errorActual = e
)
await sleepAsync(10.millis)
check eventually error == errorActual
test "calls onError when Future[?!void] fails":
var errorActual: ref CatchableError
discard returnsResultVoidError().then(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "catch callback fired when Future[?!void] fails":
var errorActual: ref CatchableError
returnsResultVoidError().catch(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "does not fire onSuccess callback when Future[?!void] fails":
var onSuccessCalled = false
returnsResultVoidError()
.then(proc() = onSuccessCalled = true)
.then(proc() = onSuccessCalled = true)
.catch(proc(e: ref CatchableError) = discard)
check always (not onSuccessCalled)
test "catch callback fired when Future[?!void] fails with uncaught error":
var errorActual: ref CatchableError
returnsResultVoidErrorUncaught().catch(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
asyncchecksuite "then - Future[?!T]":
var returnsResultValWasRun: bool
var error = (ref CatchableError)(msg: "some error")
setup: setup:
returnsResultValWasRun = false future = newFuture[?!void]("test void")
proc returnsResultVal(): Future[?!int] {.async.} = teardown:
await sleepAsync 1.millis if not future.finished:
returnsResultValWasRun = true raiseAssert "test should finish future"
return success(2)
proc returnsResultValError(): Future[?!int] {.async.} = test "then callback is fired when future is already finished":
return failure(error) var firedImmediately = false
future.complete(success())
discard future.then(proc() = firedImmediately = true)
check eventually firedImmediately
proc returnsResultValErrorUncaught(): Future[?!int] {.async.} = test "then callback is fired after future is finished":
raise error var fired = false
discard future.then(proc() = fired = true)
future.complete(success())
check eventually fired
proc returnsResultValCancelled(): Future[?!int] {.async.} = test "catch callback is fired when future is already failed":
await sleepAsync(1.seconds) var actual: ref CatchableError
return success(3) future.fail(error)
future.catch(proc(err: ref CatchableError) = actual = err)
check eventually actual == error
proc wasCancelled(error: ref CancelledError): bool = test "catch callback is fired after future is failed":
not error.isNil and error.msg == "Future operation cancelled!" var actual: ref CatchableError
future.catch(proc(err: ref CatchableError) = actual = err)
future.fail(error)
check eventually actual == error
test "calls onSuccess when Future[?!T] completes": test "cancelled callback is fired when future is already cancelled":
var actualVal = 0 var fired = false
discard returnsResultVal().then( await future.cancelAndWait()
proc(val: int) = actualVal = val, discard future.cancelled(proc() = fired = true)
proc(err: ref CatchableError) = discard check eventually fired
)
check eventually returnsResultValWasRun
check eventually actualVal == 2
test "can pass only onSuccess for Future[?!T]": test "cancelled callback is fired after future is cancelled":
var actualVal = 0 var fired = false
discard returnsResultVal().then( discard future.cancelled(proc() = fired = true)
proc(val: int) = actualVal = val await future.cancelAndWait()
) check eventually fired
check eventually returnsResultValWasRun
check eventually actualVal == 2
test "can chain onSuccess when Future[?!T] complete": test "does not fire other callbacks when successful":
var onSuccessCalledWith: seq[int] = @[]
discard returnsResultVal()
.then(proc(val: int) = onSuccessCalledWith.add val)
.then(proc(val: int) = onSuccessCalledWith.add val)
.then(proc(val: int) = onSuccessCalledWith.add val)
check eventually onSuccessCalledWith == @[2, 2, 2]
test "calls onError when Future[?!T] fails":
var errorActual: ref CatchableError
discard returnsResultValError().then(
proc(val: int) = discard,
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "calls onError when Future[?!T] fails":
var errorActual: ref CatchableError
discard returnsResultValError().then(
proc(val: int) = discard,
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "catch callback fired when Future[?!T] fails":
var errorActual: ref CatchableError
returnsResultValError().catch(
proc(e: ref CatchableError) = errorActual = e
)
check eventually error == errorActual
test "does not fire onSuccess callback when Future[?!T] fails":
var onSuccessCalled = false var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
returnsResultValError() future
.then(proc() = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
future.complete(success())
check eventually onSuccessCalled
check always (not onCancelledCalled and not onCatchCalled)
test "does not fire other callbacks when fails":
var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
future
.then(proc() = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
future.fail(error)
check eventually onCatchCalled
check always (not onCancelledCalled and not onSuccessCalled)
test "does not fire other callbacks when cancelled":
var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
future
.then(proc() = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
await future.cancelAndWait()
check eventually onCancelledCalled
check always (not onSuccessCalled and not onCatchCalled)
test "can chain onSuccess when future completes":
var onSuccessCalledTimes = 0
discard future
.then(proc() = inc onSuccessCalledTimes)
.then(proc() = inc onSuccessCalledTimes)
.then(proc() = inc onSuccessCalledTimes)
future.complete(success())
check eventually onSuccessCalledTimes == 3
asyncchecksuite "then - Future[?!T]":
var error = newError()
var future: Future[?!int]
setup:
future = newFuture[?!int]("test void")
teardown:
if not future.finished:
raiseAssert "test should finish future"
test "then callback is fired when future is already finished":
var cbVal = 0
future.complete(success(1))
discard future.then(proc(val: int) = cbVal = val)
check eventually cbVal == 1
test "then callback is fired after future is finished":
var cbVal = 0
discard future.then(proc(val: int) = cbVal = val)
future.complete(success(1))
check eventually cbVal == 1
test "catch callback is fired when future is already failed":
var actual: ref CatchableError
future.fail(error)
future.catch(proc(err: ref CatchableError) = actual = err)
check eventually actual == error
test "catch callback is fired after future is failed":
var actual: ref CatchableError
future.catch(proc(err: ref CatchableError) = actual = err)
future.fail(error)
check eventually actual == error
test "cancelled callback is fired when future is already cancelled":
var fired = false
await future.cancelAndWait()
discard future.cancelled(proc() = fired = true)
check eventually fired
test "cancelled callback is fired after future is cancelled":
var fired = false
discard future.cancelled(proc() = fired = true)
await future.cancelAndWait()
check eventually fired
test "does not fire other callbacks when successful":
var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
future
.then(proc(val: int) = onSuccessCalled = true) .then(proc(val: int) = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
future.complete(success(1))
check eventually onSuccessCalled
check always (not onCancelledCalled and not onCatchCalled)
test "does not fire other callbacks when fails":
var onSuccessCalled = false
var onCancelledCalled = false
var onCatchCalled = false
future
.then(proc(val: int) = onSuccessCalled = true) .then(proc(val: int) = onSuccessCalled = true)
.catch(proc(e: ref CatchableError) = discard) .cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
check always (not onSuccessCalled) future.fail(error)
test "catch callback fired when Future[?!T] fails with uncaught error": check eventually onCatchCalled
var errorActual: ref CatchableError check always (not onCancelledCalled and not onSuccessCalled)
returnsResultValErrorUncaught() test "does not fire other callbacks when cancelled":
.then(proc(val: int) = discard) var onSuccessCalled = false
.then(proc(val: int) = discard) var onCancelledCalled = false
.catch(proc(e: ref CatchableError) = errorActual = e) var onCatchCalled = false
check eventually error == errorActual future
.then(proc(val: int) = onSuccessCalled = true)
.cancelled(proc() = onCancelledCalled = true)
.catch(proc(e: ref CatchableError) = onCatchCalled = true)
await future.cancelAndWait()
check eventually onCancelledCalled
check always (not onSuccessCalled and not onCatchCalled)
test "can chain onSuccess when future completes":
var onSuccessCalledTimes = 0
discard future
.then(proc(val: int) = inc onSuccessCalledTimes)
.then(proc(val: int) = inc onSuccessCalledTimes)
.then(proc(val: int) = inc onSuccessCalledTimes)
future.complete(success(1))
check eventually onSuccessCalledTimes == 3

67
tests/codex/utils/testtrackedfutures.nim Normal file
View File

@ -0,0 +1,67 @@
import pkg/asynctest
import pkg/chronos
import codex/utils/trackedfutures
import ../helpers/eventually
import ../helpers
type Module = object
trackedFutures: TrackedFutures
asyncchecksuite "tracked futures":
var module: Module
setup:
module = Module(trackedFutures: TrackedFutures.new())
test "starts with zero tracked futures":
check module.trackedFutures.len == 0
test "tracks unfinished futures":
let fut = newFuture[void]("test")
discard fut.track(module)
check module.trackedFutures.len == 1
test "does not track completed futures":
let fut = newFuture[void]("test")
fut.complete()
discard fut.track(module)
check eventually module.trackedFutures.len == 0
test "does not track failed futures":
let fut = newFuture[void]("test")
fut.fail((ref CatchableError)(msg: "some error"))
discard fut.track(module)
check eventually module.trackedFutures.len == 0
test "does not track cancelled futures":
let fut = newFuture[void]("test")
await fut.cancelAndWait()
discard fut.track(module)
check eventually module.trackedFutures.len == 0
test "removes tracked future when finished":
let fut = newFuture[void]("test")
discard fut.track(module)
fut.complete()
check eventually module.trackedFutures.len == 0
test "removes tracked future when cancelled":
let fut = newFuture[void]("test")
discard fut.track(module)
await fut.cancelAndWait()
check eventually module.trackedFutures.len == 0
test "cancels and removes all tracked futures":
let fut1 = newFuture[void]("test1")
let fut2 = newFuture[void]("test2")
let fut3 = newFuture[void]("test3")
discard fut1.track(module)
discard fut2.track(module)
discard fut3.track(module)
await module.trackedFutures.cancelTracked()
check eventually fut1.cancelled
check eventually fut2.cancelled
check eventually fut3.cancelled
check eventually module.trackedFutures.len == 0