nim-dagger/codex/utils/asyncheapqueue.nim

328 lines
9.3 KiB
Nim
Raw Normal View History

2022-05-19 19:56:03 +00:00
## Nim-Codex
Poc 2 (#7) * moving protobuf into bitswap * adding block type * reworking bitswap * adding chunker * adding license header * use 1.2.6 * adding fixed size chunker * add blockstore * add iterator to chunker * more bitswap changes * rename ipfs to dagger * rename to dagger * blockstore inherits from BlockProvider * wip - add core block handling logic * smal changes * use proper block store methods * adding asynq heapqueue * wip prepare for bitswap task runner * adding `$` * adding memory store and tests * fixed chunking * extracted desicion engine from bitswap * added helper random funcs * adding testing helpers * only handle seqs * add peer events * cleanup pending blocks on blockstore event * allow nil handlers * move protobuf type helpers * allow initializing block from Cid * testing and fixes * small fixes * expose `<` * spelling * default value * spelling * pending blocks manager * adding stores manager * more tests a wip around bitswap * small changes * merge bitswap and engine for now * for now run only the new poc's tests * add a more complete ci setup * use template in map * remove p2pd * remove go * dont use asyncCheck * few small changes * adding ability to update items * adding multiple task runners * handle cancelation properly * use Result instead of throwing * wip bitswap tests * moving things around * split out engine again * add request and handlers interface * fix tests * wip - engine tests * remove unused imports * fix tests * cleanup block requesting logic * add block request tests * more block requests * add support for max heap * don't use result * use max heap & send block presence in task handler * add task handler tests * rename store to localStore * cleanup & logging * cancel task on stop * don't depend on local store for events * dont use heap queue for wants * add chronicles * fix issue with peer wants * add test for delayed block sends * remove obsolete tests * wip chunker * run all tests * add todo * misc * remove irrelevant files * removing more files * adding helpers for bitswap tests * moved bitswap file * misc * make blocks timeout longer * adjust block timeout * speedup test * compile with threads * import missing crypto * misc * disable threads for now * fix 32 bit platforms * re-enable threads support in tests
2021-02-26 00:23:22 +00:00
## Copyright (c) 2021 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import std/sequtils
import pkg/chronos
import pkg/stew/results
# Based on chronos AsyncHeapQueue and std/heapqueue
type
QueueType* {.pure.} = enum
Min, Max
AsyncHeapQueue*[T] = ref object of RootRef
## A priority queue
##
## If ``maxsize`` is less than or equal to zero, the queue size is
## infinite. If it is an integer greater than ``0``, then "await put()"
## will block when the queue reaches ``maxsize``, until an item is
## removed by "await get()".
queueType: QueueType
getters: seq[Future[void]]
putters: seq[Future[void]]
queue: seq[T]
maxsize: int
AsyncHQErrors* {.pure.} = enum
Empty, Full
proc newAsyncHeapQueue*[T](
maxsize: int = 0,
queueType: QueueType = QueueType.Min
): AsyncHeapQueue[T] =
Poc 2 (#7) * moving protobuf into bitswap * adding block type * reworking bitswap * adding chunker * adding license header * use 1.2.6 * adding fixed size chunker * add blockstore * add iterator to chunker * more bitswap changes * rename ipfs to dagger * rename to dagger * blockstore inherits from BlockProvider * wip - add core block handling logic * smal changes * use proper block store methods * adding asynq heapqueue * wip prepare for bitswap task runner * adding `$` * adding memory store and tests * fixed chunking * extracted desicion engine from bitswap * added helper random funcs * adding testing helpers * only handle seqs * add peer events * cleanup pending blocks on blockstore event * allow nil handlers * move protobuf type helpers * allow initializing block from Cid * testing and fixes * small fixes * expose `<` * spelling * default value * spelling * pending blocks manager * adding stores manager * more tests a wip around bitswap * small changes * merge bitswap and engine for now * for now run only the new poc's tests * add a more complete ci setup * use template in map * remove p2pd * remove go * dont use asyncCheck * few small changes * adding ability to update items * adding multiple task runners * handle cancelation properly * use Result instead of throwing * wip bitswap tests * moving things around * split out engine again * add request and handlers interface * fix tests * wip - engine tests * remove unused imports * fix tests * cleanup block requesting logic * add block request tests * more block requests * add support for max heap * don't use result * use max heap & send block presence in task handler * add task handler tests * rename store to localStore * cleanup & logging * cancel task on stop * don't depend on local store for events * dont use heap queue for wants * add chronicles * fix issue with peer wants * add test for delayed block sends * remove obsolete tests * wip chunker * run all tests * add todo * misc * remove irrelevant files * removing more files * adding helpers for bitswap tests * moved bitswap file * misc * make blocks timeout longer * adjust block timeout * speedup test * compile with threads * import missing crypto * misc * disable threads for now * fix 32 bit platforms * re-enable threads support in tests
2021-02-26 00:23:22 +00:00
## Creates a new asynchronous queue ``AsyncHeapQueue``.
##
AsyncHeapQueue[T](
getters: newSeq[Future[void]](),
putters: newSeq[Future[void]](),
queue: newSeqOfCap[T](maxsize),
maxsize: maxsize,
queueType: queueType,
)
proc wakeupNext(waiters: var seq[Future[void]]) {.inline.} =
var i = 0
while i < len(waiters):
var waiter = waiters[i]
inc(i)
if not(waiter.finished()):
waiter.complete()
break
if i > 0:
waiters.delete(0..(i-1))
Poc 2 (#7) * moving protobuf into bitswap * adding block type * reworking bitswap * adding chunker * adding license header * use 1.2.6 * adding fixed size chunker * add blockstore * add iterator to chunker * more bitswap changes * rename ipfs to dagger * rename to dagger * blockstore inherits from BlockProvider * wip - add core block handling logic * smal changes * use proper block store methods * adding asynq heapqueue * wip prepare for bitswap task runner * adding `$` * adding memory store and tests * fixed chunking * extracted desicion engine from bitswap * added helper random funcs * adding testing helpers * only handle seqs * add peer events * cleanup pending blocks on blockstore event * allow nil handlers * move protobuf type helpers * allow initializing block from Cid * testing and fixes * small fixes * expose `<` * spelling * default value * spelling * pending blocks manager * adding stores manager * more tests a wip around bitswap * small changes * merge bitswap and engine for now * for now run only the new poc's tests * add a more complete ci setup * use template in map * remove p2pd * remove go * dont use asyncCheck * few small changes * adding ability to update items * adding multiple task runners * handle cancelation properly * use Result instead of throwing * wip bitswap tests * moving things around * split out engine again * add request and handlers interface * fix tests * wip - engine tests * remove unused imports * fix tests * cleanup block requesting logic * add block request tests * more block requests * add support for max heap * don't use result * use max heap & send block presence in task handler * add task handler tests * rename store to localStore * cleanup & logging * cancel task on stop * don't depend on local store for events * dont use heap queue for wants * add chronicles * fix issue with peer wants * add test for delayed block sends * remove obsolete tests * wip chunker * run all tests * add todo * misc * remove irrelevant files * removing more files * adding helpers for bitswap tests * moved bitswap file * misc * make blocks timeout longer * adjust block timeout * speedup test * compile with threads * import missing crypto * misc * disable threads for now * fix 32 bit platforms * re-enable threads support in tests
2021-02-26 00:23:22 +00:00
proc heapCmp[T](x, y: T, max: bool = false): bool {.inline.} =
if max:
return (y < x)
else:
return (x < y)
proc siftdown[T](heap: AsyncHeapQueue[T], startpos, p: int) =
## 'heap' is a heap at all indices >= startpos, except
## possibly for pos. pos is the index of a leaf with a
## possibly out-of-order value. Restore the heap invariant.
##
var pos = p
var newitem = heap[pos]
# Follow the path to the root, moving parents down until
# finding a place newitem fits.
while pos > startpos:
let parentpos = (pos - 1) shr 1
let parent = heap[parentpos]
if heapCmp(newitem, parent, heap.queueType == QueueType.Max):
heap.queue[pos] = parent
pos = parentpos
else:
break
heap.queue[pos] = newitem
proc siftup[T](heap: AsyncHeapQueue[T], p: int) =
let endpos = len(heap)
var pos = p
let startpos = pos
let newitem = heap[pos]
# Bubble up the smaller child until hitting a leaf.
var childpos = 2*pos + 1 # leftmost child position
while childpos < endpos:
# Set childpos to index of smaller child.
let rightpos = childpos + 1
if rightpos < endpos and
not heapCmp(heap[childpos], heap[rightpos], heap.queueType == QueueType.Max):
childpos = rightpos
# Move the smaller child up.
heap.queue[pos] = heap[childpos]
pos = childpos
childpos = 2*pos + 1
# The leaf at pos is empty now. Put newitem there, and bubble it up
# to its final resting place (by sifting its parents down).
heap.queue[pos] = newitem
siftdown(heap, startpos, pos)
proc full*[T](heap: AsyncHeapQueue[T]): bool {.inline.} =
## Return ``true`` if there are ``maxsize`` items in the queue.
##
## Note: If the ``heap`` was initialized with ``maxsize = 0`` (default),
## then ``full()`` is never ``true``.
if heap.maxsize <= 0:
false
else:
(len(heap.queue) >= heap.maxsize)
proc empty*[T](heap: AsyncHeapQueue[T]): bool {.inline.} =
## Return ``true`` if the queue is empty, ``false`` otherwise.
(len(heap.queue) == 0)
proc pushNoWait*[T](heap: AsyncHeapQueue[T], item: T): Result[void, AsyncHQErrors] =
## Push `item` onto heap, maintaining the heap invariant.
##
if heap.full():
return err(AsyncHQErrors.Full)
heap.queue.add(item)
siftdown(heap, 0, len(heap)-1)
heap.getters.wakeupNext()
return ok()
proc push*[T](heap: AsyncHeapQueue[T], item: T) {.async, gcsafe.} =
## Push item into the queue, awaiting for an available slot
## when it's full
##
while heap.full():
var putter = newFuture[void]("AsyncHeapQueue.push")
heap.putters.add(putter)
try:
await putter
except CatchableError as exc:
if not(heap.full()) and not(putter.cancelled()):
heap.putters.wakeupNext()
raise exc
heap.pushNoWait(item).tryGet()
proc popNoWait*[T](heap: AsyncHeapQueue[T]): Result[T, AsyncHQErrors] =
## Pop and return the smallest item from `heap`,
## maintaining the heap invariant.
##
if heap.empty():
return err(AsyncHQErrors.Empty)
let lastelt = heap.queue.pop()
if heap.len > 0:
result = ok(heap[0])
heap.queue[0] = lastelt
siftup(heap, 0)
else:
result = ok(lastelt)
heap.putters.wakeupNext()
proc pop*[T](heap: AsyncHeapQueue[T]): Future[T] {.async.} =
## Remove and return an ``item`` from the beginning of the queue ``heap``.
## If the queue is empty, wait until an item is available.
while heap.empty():
var getter = newFuture[void]("AsyncHeapQueue.pop")
heap.getters.add(getter)
try:
await getter
except CatchableError as exc:
if not(heap.empty()) and not(getter.cancelled()):
heap.getters.wakeupNext()
raise exc
return heap.popNoWait().tryGet()
proc del*[T](heap: AsyncHeapQueue[T], index: Natural) =
## Removes the element at `index` from `heap`,
## maintaining the heap invariant.
##
if heap.empty():
return
swap(heap.queue[^1], heap.queue[index])
let newLen = heap.len - 1
heap.queue.setLen(newLen)
if index < newLen:
heap.siftup(index)
heap.putters.wakeupNext()
proc delete*[T](heap: AsyncHeapQueue[T], item: T) =
## Find and delete an `item` from the `heap`
##
let index = heap.find(item)
if index > -1:
heap.del(index)
proc update*[T](heap: AsyncHeapQueue[T], item: T): bool =
## Update an entry in the heap by reshufling its
## possition, maintaining the heap invariant.
##
let index = heap.find(item)
if index > -1:
# replace item with new one in case it's a copy
heap.queue[index] = item
# re-establish heap order
# TODO: don't start at 0 to avoid reshuffling
# entire heap
heap.siftup(0)
return true
proc pushOrUpdateNoWait*[T](heap: AsyncHeapQueue[T], item: T): Result[void, AsyncHQErrors] =
## Update an item if it exists or push a new one
##
if heap.update(item):
return ok()
return heap.pushNoWait(item)
proc pushOrUpdate*[T](heap: AsyncHeapQueue[T], item: T) {.async.} =
## Update an item if it exists or push a new one
## awaiting until a slot becomes available
##
if not heap.update(item):
await heap.push(item)
proc replace*[T](heap: AsyncHeapQueue[T], item: T): Result[T, AsyncHQErrors] =
## Pop and return the current smallest value, and add the new item.
## This is more efficient than pop() followed by push(), and can be
## more appropriate when using a fixed-size heap. Note that the value
## returned may be larger than item! That constrains reasonable uses of
## this routine unless written as part of a conditional replacement:
##
## .. code-block:: nim
## if item > heap[0]:
## item = replace(heap, item)
##
if heap.empty():
error(AsyncHQErrors.Empty)
result = heap[0]
heap.queue[0] = item
siftup(heap, 0)
proc pushPopNoWait*[T](heap: AsyncHeapQueue[T], item: T): Result[T, AsyncHQErrors] =
## Fast version of a push followed by a pop.
##
if heap.empty():
err(AsyncHQErrors.Empty)
if heap.len > 0 and heapCmp(heap[0], item, heap.queueType == QueueType.Max):
swap(item, heap[0])
siftup(heap, 0)
return item
proc clear*[T](heap: AsyncHeapQueue[T]) {.inline.} =
## Clears all elements of queue ``heap``.
heap.queue.setLen(0)
proc len*[T](heap: AsyncHeapQueue[T]): int {.inline.} =
## Return the number of elements in ``heap``.
len(heap.queue)
proc size*[T](heap: AsyncHeapQueue[T]): int {.inline.} =
## Return the maximum number of elements in ``heap``.
Slot queue (#455) ## Slot queue Adds a slot queue, as per the [slot queue design](https://github.com/codex-storage/codex-research/blob/master/design/sales.md#slot-queue). Any time storage is requested, all slots from that request are immediately added to the queue. Finished, Canclled, Failed requests remove all slots with that request id from the queue. SlotFreed events add a new slot to the queue and SlotFilled events remove the slot from the queue. This allows popping of a slot each time one is processed, making things much simpler. When an entire request of slots is added to the queue, the slot indices are shuffled randomly to hopefully prevent nodes that pick up the same storage requested event from clashing on the first processed slot index. This allowed removal of assigning a random slot index in the SalePreparing state and it also ensured that all SalesAgents will have a slot index assigned to them at the start thus the removal of the optional slotIndex. Remove slotId from SlotFreed event as it was not being used. RequestId and slotIndex were added to the SlotFreed event earlier and those are now being used The slot queue invariant that prioritises queue items added to the queue relies on a scoring mechanism to sort them based on the [sort order in the design document](https://github.com/codex-storage/codex-research/blob/master/design/sales.md#sort-order). When a storage request is handled by the sales module, a slot index was randomly assigned and then the slot was filled. Now, a random slot index is only assigned when adding an entire request to the slot queue. Additionally, the slot is checked that its state is `SlotState.Free` before continuing with the download process. SlotQueue should always ensure the underlying AsyncHeapQueue has one less than the maximum items, ensuring the SlotQueue can always have space to add an additional item regardless if it’s full or not. Constructing `SlotQueue.workers` in `SlotQueue.new` calls `newAsyncQueue` which causes side effects, so the construction call had to be moved to `SlotQueue.start`. Prevent loading request from contract (network request) if there is an existing item in queue for that request. Check availability before adding request to queue. Add ability to query market contract for past events. When new availabilities are added, the `onReservationAdded` callback is triggered in which past `StorageRequested` events are queried, and those slots are added to the queue (filtered by availability on `push` and filtered by state in `SalePreparing`). #### Request Workers Limit the concurrent requests being processed in the queue by using a limited pool of workers (default = 3). Workers are in a data structure of type `AsyncQueue[SlotQueueWorker]`. This allows us to await a `popFirst` for available workers inside of the main SlotQueue event loop Add an `onCleanUp` that stops the agents and removes them from the sales module agent list. `onCleanUp` is called from sales end states (eg ignored, cancelled, finished, failed, errored). Add a `doneProcessing` future to `SlotQueueWorker` to be completed in the `OnProcessSlot` callback. Each `doneProcessing` future created is cancelled and awaited in `SlotQueue.stop` (thanks to `TrackableFuturees`), which forced `stop` to become async. - Cancel dispatched workers and the `onProcessSlot` callbacks, prevents zombie callbacks #### Add TrackableFutures Allow tracking of futures in a module so they can be cancelled at a later time. Useful for asyncSpawned futures, but works for any future. ### Sales module The sales module needed to subscribe to request events to ensure that the request queue was managed correctly on each event. In the process of doing this, the sales agents were updated to avoid subscribing to events in each agent, and instead dispatch received events from the sales module to all created sales agents. This would prevent memory leaks on having too many eventemitters subscribed to. - prevent removal of agents from sales module while stopping, otherwise the agents seq len is modified while iterating An additional sales agent state was added, `SalePreparing`, that handles all state machine setup, such as retrieving the request and subscribing to events that were previously in the `SaleDownloading` state. Once agents have parked in an end state (eg ignored, cancelled, finished, failed, errored), they were not getting cleaned up and the sales module was keeping a handle on their reference. An `onCleanUp` callback was created to be called after the state machine enters an end state, which could prevent a memory leak if the number of requests coming in is high. Move the SalesAgent callback raises pragmas from the Sales module to the proc definition in SalesAgent. This avoids having to catch `Exception`. - remove unneeded error handling as pragmas were moved Move sales.subscriptions from an object containing named subscriptions to a `seq[Subscription]` directly on the sales object. Sales tests: shut down repo after sales stop, to fix SIGABRT in CI ### Add async Promise API - modelled after JavaScript Promise API - alternative to `asyncSpawn` that allows handling of async calls in a synchronous context (including access to the synchronous closure) with less additional procs to be declared - Write less code, catch errors that would otherwise defect in asyncspawn, and execute a callback after completion - Add cancellation callbacks to utils/then, ensuring cancellations are handled properly ## Dependencies - bump codex-contracts-eth to support slot queue (https://github.com/codex-storage/codex-contracts-eth/pull/61) - bump nim-ethers to 0.5.0 - Bump nim-json-rpc submodule to 0bf2bcb --------- Co-authored-by: Jaremy Creechley <creechley@gmail.com>
2023-07-25 02:50:30 +00:00
heap.maxsize
Poc 2 (#7) * moving protobuf into bitswap * adding block type * reworking bitswap * adding chunker * adding license header * use 1.2.6 * adding fixed size chunker * add blockstore * add iterator to chunker * more bitswap changes * rename ipfs to dagger * rename to dagger * blockstore inherits from BlockProvider * wip - add core block handling logic * smal changes * use proper block store methods * adding asynq heapqueue * wip prepare for bitswap task runner * adding `$` * adding memory store and tests * fixed chunking * extracted desicion engine from bitswap * added helper random funcs * adding testing helpers * only handle seqs * add peer events * cleanup pending blocks on blockstore event * allow nil handlers * move protobuf type helpers * allow initializing block from Cid * testing and fixes * small fixes * expose `<` * spelling * default value * spelling * pending blocks manager * adding stores manager * more tests a wip around bitswap * small changes * merge bitswap and engine for now * for now run only the new poc's tests * add a more complete ci setup * use template in map * remove p2pd * remove go * dont use asyncCheck * few small changes * adding ability to update items * adding multiple task runners * handle cancelation properly * use Result instead of throwing * wip bitswap tests * moving things around * split out engine again * add request and handlers interface * fix tests * wip - engine tests * remove unused imports * fix tests * cleanup block requesting logic * add block request tests * more block requests * add support for max heap * don't use result * use max heap & send block presence in task handler * add task handler tests * rename store to localStore * cleanup & logging * cancel task on stop * don't depend on local store for events * dont use heap queue for wants * add chronicles * fix issue with peer wants * add test for delayed block sends * remove obsolete tests * wip chunker * run all tests * add todo * misc * remove irrelevant files * removing more files * adding helpers for bitswap tests * moved bitswap file * misc * make blocks timeout longer * adjust block timeout * speedup test * compile with threads * import missing crypto * misc * disable threads for now * fix 32 bit platforms * re-enable threads support in tests
2021-02-26 00:23:22 +00:00
proc `[]`*[T](heap: AsyncHeapQueue[T], i: Natural) : T {.inline.} =
## Access the i-th element of ``heap`` by order from first to last.
## ``heap[0]`` is the first element, ``heap[^1]`` is the last element.
heap.queue[i]
proc `[]`*[T](heap: AsyncHeapQueue[T], i: BackwardsIndex) : T {.inline.} =
## Access the i-th element of ``heap`` by order from first to last.
## ``heap[0]`` is the first element, ``heap[^1]`` is the last element.
heap.queue[len(heap.queue) - int(i)]
iterator items*[T](heap: AsyncHeapQueue[T]): T {.inline.} =
## Yield every element of ``heap``.
for item in heap.queue.items():
yield item
iterator mitems*[T](heap: AsyncHeapQueue[T]): var T {.inline.} =
## Yield every element of ``heap``.
for mitem in heap.queue.mitems():
yield mitem
iterator pairs*[T](heap: AsyncHeapQueue[T]): tuple[key: int, val: T] {.inline.} =
## Yield every (position, value) of ``heap``.
for pair in heap.queue.pairs():
yield pair
proc contains*[T](heap: AsyncHeapQueue[T], item: T): bool {.inline.} =
## Return true if ``item`` is in ``heap`` or false if not found. Usually used
## via the ``in`` operator.
for e in heap.queue.items():
if e == item: return true
return false
proc `$`*[T](heap: AsyncHeapQueue[T]): string =
## Turn an async queue ``heap`` into its string representation.
var res = "["
for item in heap.queue.items():
if len(res) > 1: res.add(", ")
res.addQuoted(item)
res.add("]")
res