Generic Non-Blocking Task Management ("Queue") for discovery and nodes domains

[emmacasolin]

Specification

Unattended discovery was added in #320, however, there is no concept of priority within the queue. There are three ways that a vertex (a node or identity) can be added to the discovery queue, and they should follow this order of priority:

1. Manually, via the discovery methods queueDiscoveryByNode() and queueDiscoveryByIdentity() (these are called in the commands identities discover (explicit discovery) and identities trust (explicitly setting a permission, so we want the Gestalt to be updated via discovery).
2. As a step in the discovery process whereby child vertices are added into the discovery queue in order to discover the entire connected gestalt.
3. Automatically by a process of rediscovery when we want to update existing Gestalts in the Gestalt Graph (to be addressed in Discovery - revisiting Gestalt Vertices and error handling #328).

Vertexes with a higher priority should be discovered first, either by being placed at the front of the queue or by modifying the traversal method of the queue. The priority queue could also be further optimised by grouping vertices from the same gestalt together when this is known (for example when adding child vertices).

Additional context

• Original PR for implementing Unattended Discovery: Growing the Gestalt Graph and Implementing Social Discovery #320
• Issue for re-adding discovered vertices into the queue to update the gestalt graph Discovery - revisiting Gestalt Vertices and error handling #328
• Indexing Domains and DB Levels for Faster & Flexible Lookup #188 - Indexing the LevelDB may be required to be able to implement a persistent priority queue
• Integrate Automatic Indexing into DB js-db#1 - general indexing into js-db
• https://nicolas-van.github.io/modern-async/modern-async/1.1.0/Queue.html

Tasks

1. Modify the existing Discovery Queue to be a Priority Queue
2. Ensure that when a user interactively wants to discover a gestalt vertex that it becomes the highest priority and gets executed first
3. Look into the potential for further optimising the priority queue, for example by having multiple points of comparison with varying levels of importance that can influence the priority of a particular vertex in the queue

[CMCDragonkai]

There's a go implementation of persistent priority queue backed by leveldb here: https://github.com/beeker1121/goque. It can be used as a reference for this. Also a JS implementation here: https://github.com/eugeneware/level-q (the goque is probably more comprehensive).

Our priority queue needs to by default maintain order, because we do want to know the sorted list of jobs. But also allow us to add a special priority number on top.

From my imagination:

This reminds me of the indexing problem, where you can ask for a list of rows sorted by several columns. The first column would dictate the base sort, then subsequent columns would sort any ambiguous sub-orders.

Imagine we had 2 indexes. The first being your priority index using an arbitrary number, the second being the monotonic time index using IdSortable. You could sort on the priority index first, then sort on IdSortable second.

Maybe this then has a relationship to #188.

However this would only be for if you are looking up items. If we are streaming data from the level db that may be more complicated.

[CMCDragonkai]

This can be done with a compound index. Prefix can be the priority number (lexinted), suffix can be IdSortable.

This means you can then stream results that are always ordered in terms of priority first then by time second.

Priority can start at 0 by default, and one can increment priorities depending on the origin of the task. Like tasks emitted by user wanting to lookup something can be set to a higher priority number.

We could do this directly by changing the queue domain key. But I'd suggest first solving the indexing issue in general first then building a compound index on top.

[CMCDragonkai]

We discovered that the priority queue can also benefit from a uniqueness index creating a uniqueness constraint: #311 (comment)

This means that duplicate tasks cannot go into the priority queue. Not entirely sure if this is required because a queue can still say they should process the same task over and over.

[CMCDragonkai]

We should have a concurrency bound in the queue. This means how many tasks should be executed at the same time. By default unbounded meaning all tasks gets executed immediately without waiting to be done.

For IO bound tasks, you might as well have unbounded concurrency. For CPU-bound it can be sent to the web worker pool which is bounded by core count. Battery usage optimisation may also affect our limit too.

[emmacasolin]

A generic Queue class has been implemented here: 91287ab

This queue is not persistent, or a priority queue, however, it is designed to be a generic queue that can eventually be used in all places that require this functionality (including the Discovery Queue). The generic Queue can be refactored to meet this issue and #328 at some point in the future, potentially incorporating the DB.

[CMCDragonkai]

Renamed this issue to the general idea of non-blocking task management. It now has to solve for discovery, nodes management in terms of setting nodes, pinging nodes and garbage collection, as well as in relation to:

• Refactor error handling of failed Node Connections created from the Discovery domain #354
• Write tests to cover all possible states of the Discovery, NodeConnection, and discovered Node during the discovery process #349
• Supporting the "refresh" operation for NodeGraph buckets #345
• Discovery - revisiting Gestalt Vertices and error handling #328
• Asynchronous Promise Cancellation with Cancellable Promises, AbortController and Generic Timer #297

There's a relationship between the queue design and the EventBus system, as well as our WorkerManager.

Most important is for us to develop a Task abstraction. It can be a class Task, that represents a "lazy promise". Promises in JS are strictly evaluated, while these tasks will need to be lazily evaluated. Then our task management system can convert our lazy tasks to strict promises (which represents futures). More background info here: https://en.wikipedia.org/wiki/Futures_and_promises

Our task manager will need to have configurable:

• Concurrency limit - indicates the bound on the pool of currently executing tasks (can be 1 to unbounded/infinite)
• Executor - choosing to execute by Node's event-loop, or by passing it into the WorkerManager to be executed in a separate thread or core, the former should be used for IO-bound tasks, the latter should be used for CPU-bound tasks. This could be specified by the task creator, rather than the task manager itself.

Stretch goal is to also incorporate "time/calendar-scheduling" so that tasks can be executed at a point in time like cron.

Interaction between EventBus and task manager may be considered. The event bus is about communicating changes between domains, but the task manager is the one actually executing the tasks.

Tasks can be:

• Re-ordered or reprioritised or given priorities
• Can be observed for success or failure
• Can have errors handled
• Can be cancelled using our design for abort signal, and have their real side-effects cancelled
• Can be monitored for progress
• Can be persistent - backed by leveldb

This rabbit hole for this goes deep. So we should make sure not to feature creep our non-blocking task queuing needs.

[CMCDragonkai]

Example of prior work: https://github.com/fluture-js/Fluture

[CMCDragonkai]

Also to clarify, we are not creating a "generic distributed job queue", that's the realm of things like redis queue and https://en.wikipedia.org/wiki/List_of_job_scheduler_software. There's so much of this already. We just need something in-process relative to Polykey.

[tegefaulkes]

Along with the configurable concurrency limit and executor, I think we should have an interface for the queue as well. Depending on the situation we may need just a simple queue, a priority queue, a persistent database queue like discovery uses, etc etc...
So far as the Queue cares it only needs to support push and shift. So we can make the Queue a generic class and pass it any implementation we want for storing the queue so long as it extends the interface.

It shouldn't be too hard to make the change. We just need to decide if this degree of control is desired. I can see a need for it though.

[tegefaulkes]

Is this a part of #326 ?

[CMCDragonkai]

Is this a part of #326 ?

Nope, this can be done later.

[CMCDragonkai]

More prior work:

• https://nicolas-van.github.io/modern-async/modern-async/1.1.0/Queue.html

Actually the entire modern-async library is quite interesting, as it has implemented some of the things that we've been working on as well. But I think we won't just bring in that library, but instead extract parts out of it for our own implementation.

It has interesting ideas for promise cancellation as well, Delayer, Scheduler and Queue can all help. I believe that our usage of async generators and decorators might be more advanced though.

The library also exports a bunch of collection combinators that can work with asynchronous execution. So instead of say Array.map mapping a synchronous function, it could map an asynchronous function, and then wait for all of them to finish. Basically we do this with Promise.all atm. It also works with asynchronous iterables like async generators. This I feel is a different kind of thing, and I'd only want to bring in these utility functions where it's relevant. Maybe if JS had better treeshaking and package specific imports (MatrixAI/TypeScript-Demo-Lib#32), it would work well.

[CMCDragonkai]

When scheduling a new task. The task object is created.

This task object needs to represent a lazy promise.

A lazy promise in this case means that the promise doesn't mean that the execution has started. It's simply queued.

Because tasks are persisted into the DB, it's possible for the PK agent to be restarted, and one may wish to await for a given task ID. That means acquiring a promise for that task ID.

I'm thinking that we can lazily create a promise, that is one to one for each task. Then you can await this promise's result.

If multiple calls to acquiring this promise is made, the same promise is shared among all callers. This means a single resolve or reject call will send all awaiters.

Here's an example:

async function main () {

  const p = new Promise((resolve, reject) => {
    setTimeout(() => {
      reject(new Error('oh no'))
    }, 500);
    setTimeout(resolve, 1000);
  });

  const f = async () => {
    await p;
    return 'f';
  };

  const g = async () => {
    await p;
    return 'g';
  };

  const r = await Promise.allSettled([
    f(),
    g()
  ]);

  console.log(r);

  // @ts-ignore
  console.log(r[0].reason === r[1].reason); // This is `true`

  // The same exception object is thrown to all awaiters

}

void main();

There are some constraints:

1. You cannot acquire a promise for a task that does not exist in the queue.
2. Creation of the promise may involve hitting the disk if the promise doesn't already exist relative to a task ID.
3. Creation of the promises has to be protected against race conditions with the object locking map pattern.
4. If you get a promise, it is guaranteed that within some finite amount of time that this promise will eventually resolve.

How is point 4 guaranteed? It is only possible to get a task promise in 2 ways:

1. During creation of a new task, the promise can be created afterwards.
2. One can ask queue.getTask().

Now because it's a lazy promise, this could mean that the task is already executed by the time you ask for a promise for the task. This is only possible if the task is no longer in the queue (or is in some invalid state).

In this situation, when asking for the promise, the promise should be immediately rejected. Alternatively since the acquisition of this promise is lazy, one may throw an exception at this point. The point is, if you do get a promise, the promise must be settled eventually.

One of the initial ways to do this is to add an event listener for every task as soon as it is put into the queue. The problem with this is that now you get in-memory space complexity of O(n), where you have 1 listener for every task.

Listeners aren't always necessary, and maybe lots of tasks are put into the queue. In such a case, we can make the promise/listener itself lazy.

getTask(): Promise<Task> {
  // if we give you back a Task
  // you can be guaranteed to have it resolved or rejected
  // if we cannot give you a Task, because it's already executed, then we throw an exception at this point
}

Alternatively we do something like:

getTask(): Promise<Task | undefined>;

And undefined means it's not possible to give you a promise to the task.

[CMCDragonkai]

Another problem is that exactly is a Task? Is a data structure, or is it a promise? Maybe it's both?

The Task could be a class instance with enumerable properties, along with a method that allows one to acquire the promise?

I'm considering an API like this:

// false means that you are tracking the task immediately, (this is the default)
const task = await queue.pushTask(lazy: false);

// Waits for the promise
await task;

// true means you are not tracking the task
const task = await queue.pushTask(lazy: true);

// Now it may result in an exception if the task is already executed
// This has to distinguish from the task itself being rejected
// ErrorTaskReference
// ErrorTaskRejected
await task;

Something like this means Task is in fact an extension of the Promise. Or at least a class that has the then method. It doesn't actually have to satisfy the entire promise interface. Which would include .catch and .finally.

Either way, the lazy boolean allows one to switch from a lazy promise to an eager promise.

In this sense, lazy simply means whether the task itself is being tracked or not. If it is being tracked, then await task is always guaranteed to either result in ErrorTaskRejected or the task's result. If it is not being tracked, then it only starts tracking when await task is called. Which calls the then method. At this point it may throw ErrorTaskMissing or ErrorTaskRejected... or anything else.

I may have something like:

ErrorTaskMissing - task itself is no longer around, it may already been fulfilled
ErrorTaskReference - task handler was not found
ErrorTaskRejected - task was rejected, see the cause chain

[CMCDragonkai]

Is there a utility in having ErrorTaskRejected? Maybe only to create a set of possible exceptions, as the cause chain can have anything that the task handler itself throws. Otherwise we are just rethrowing the exceptions. https://gist.github.com/CMCDragonkai/08266b1463158f4156f66d4bf077add6

[CMCDragonkai]

So within Queue, we will have 2 methods that that are used as part of start and stop.

protected async startProcessing(): Promise<void>;
protected async stopProcessing(): Promise<void>;

Their job is to peek into the job schedule. (I've started to realise that this is more a "schedule" not a queue, since the priority doesn't apply until the tasks are due for execution).

In the job schedule, they find:

• Tasks that are due for execution are dispatched the execution queue
• The next task not due for execution will have its scheduled time set as a setTimeout

The startProcessing will also be called by the scheduleTask method. This is because there be no tasks in the schedule, and upon scheduling a task, we trigger the start processing again.

Calling startProcess should be idempotent, as in, if the processing is already started, then nothing happens. It would only matter if the setTimeout delay should be made smaller because a more recent task has be scheduled.

[CMCDragonkai]

The queue now would have 2 "queues".

1. Schedule - this is purely time based, by using IdSortable as the TaskId, this means all tasks have unique scheduling time (up to the maximum amount of ticks the IdSortable is capable of). Which means there will always be 1 task that is in front to be executed. Here priority does not matter, it is simply a matter of scheduling time
2. Execution Queue - this is the queue of jobs that are actually pending execution right now, here is where priority actually matters. This can make use of dynamic priority assuming the number of tasks sitting here is not too much. One can just round robin here, or select the task that has sat here the longest. Perhaps a double index sort between initial priority and time of insertion.

If the task is never fulfilled (resolve/rejected), it should stay in the execution queue (which should still be persistent).

[CMCDragonkai]

Originally in order to "connect" a lazy promise to a task execution, this was done with a callback that I called a "listener".

Now I realised that the deconstructed promise is itself already a set of callbacks to be executed on the task execution.

This means during actual task dispatch, we could just do something like:

taskHandlerExecution(...taskParameters).then(
  resolveTaskP,
  rejectTaskP
).finally(() => {
  this.promises.delete(taskId.toString() as TaskIdString);
});

That is, the promise that comes from executing the task handler gets connected to the deconstructed promise of the task abstraction.

And at the end, the task promise is deleted once the task is done.

This only occurs IF the task promise was first created. If it was lazy promise, it may never get created in which, and if so, nothing is there to observe/await the task execution. That's fine as the task's side effects continues to be done.

[CMCDragonkai]

So now I have:

• Queue - this is an encapsulated dependency of Scheduler
• Scheduler - this is the main access to the tasks domain

During the PolykeyAgent start process, we expect that the Scheduler is going to be a required dependency of other relevant domains.

const scheduler = new Scheduler();
await Discovery.createDiscovery({ scheduler });
await NodeGraph.createNodeGraph({ scheduler });
await scheduler.start();

Why not use await Scheduler.createScheduler();? This is because, this would require us to inject handlers from the very beginning, and these handlers are only known by the other domains. Which results in a circular dependency.

Here we are directly constructing the Scheduler and using it like a StartStop system.

However it is actually CDSS, as there is a destroy method too that removes all the persisted state.

[CMCDragonkai]

One of the issues with this is that certain methods must be possible by the time it is constructed, but not necessarily asynchronously started...

But at the same time asynchronous start is necessary to do any async setup such as creating the database levels... etc.

So now I'm thinking that start and stop is still used, and thus createScheduler is used, but when Scheduler.start just doesn't actually start the processing of tasks, necessitating one to call scheduler.startProcessing().

But then it's not going to be symmetric if the Scheduler.stop does call stopProcessing but start doesn't call startProcessing.

[CMCDragonkai]

The alternative is that we use a callback/hook abstraction similar to the EventBus. So now instead function hooks is registered in the other domains first. These end up calling the scheduler system. However I'm not sure if these work if during their start, it will end up calling these callbacks which won't even have the handlers registered.

Another alternative is that Scheduler is only StartStop instead, but again this isn't nice, if there needs to be asynchronous creation routines.

[CMCDragonkai]

I've added a delay boolean to Scheduler.start in order to not start the processing. That way users can start scheduling with Scheduler.startProcessing() manually afterwards.

By default the delay is false, so that by default the processing does already start.

This means in the PolykeyAgent.createPolykeyAgent, we should instead see something like:

const scheduler = await Scheduler.createScheduler({ delay: true });
await Discovery.createDiscovery({ scheduler });
await NodeGraph.createNodeGraph({ scheduler });
await scheduler.startProcessing();

When stopping the processing this doesn't actually stop the execution of any tasks, it just stops the processing of the scheduler.

[CMCDragonkai]

The scheduler doesn't execute the tasks. It dispatches to the queue. The queue assigns tasks to workers, workers is what executes the task. At the same time, the workers may also pull tasks from the queue when they are idle.

[CMCDragonkai]

The Queue will need to work with threadsjs queue too: https://threads.js.org/usage-pool. I'm not sure yet if this means our WorkerManager will need to be changed to work with the Queue, since I don't really want there to be 2 queues. Maybe Queue is for managing the queue persistence, while embedding the WorkerManager in-memory queue that is one to one for each task that is persisted.

Alternatively we actually don't use WorkerManager pool, and instead manage our own "pool" directly. This means either extending Pool from threadsjs if possible. However ideally the Queue can also work without the WorkerManager being available, but I'm not sure if this is possible without very different behaviour. Without there being worker threads, you don't really have the task pulling behaviour, instead one just assigns tasks to a concurrency limit.

Note that threadsjs has 2 concurrency limits:

1. The parallel number of workers to launch
2. The number of concurrent async tasks to run

The only real reason to use workers is to run CPU-intensive tasks, not IO-intensive tasks. (Hard to know precisely until we do benchmarking). So the number of concurrent async tasks should be 1. Therefore we ignore 2. in our design of the Queue.

If the worker manager was not available, the parallel number of workers to launch should be the same concurrency limit of the number of tasks to run concurrently in our Queue. In the former, we would use the os.cpu() count, the latter, this can be specified with some number, with it defaulting to 1...

Actually we can always default it to 1, and then override it with the os.cpu() count if we expect to supply it with workers.

---

One issue with this is that the WorkerManager is also used for other things where extra CPU intensive tasks is just offloaded. Perhaps instead WorkerManager stays the same, and the injection of worker manager, means a concurrent number of tasks are dispatched to the worker manager but awaited for normally. We would need some way of checking the capacity of the workers before pushing a task into it.

[CMCDragonkai]

I haven't completed the full design of Task class. But I suspect it needs to be similar to lazy promise here: https://github.com/sindresorhus/p-lazy/blob/main/index.js, and even threadsjs representation uses a then method to allow await to work on their objects. Their type is:

/**
 * Task that has been `pool.queued()`-ed.
 */
export interface QueuedTask<ThreadType extends Thread, Return> {
    /** @private */
    id: number;
    /** @private */
    run: TaskRunFunction<ThreadType, Return>;
    /**
     * Queued tasks can be cancelled until the pool starts running them on a worker thread.
     */
    cancel(): void;
    /**
     * `QueuedTask` is thenable, so you can `await` it.
     * Resolves when the task has successfully been executed. Rejects if the task fails.
     */
    then: Promise<Return>["then"];
}

[CMCDragonkai]

If Task is in fact a class Task extends Promise, it would have properties that would be enumerable, and properties that are not. We may need to specify this explicitly: https://debugmode.net/2020/06/18/how-to-make-a-property-non-enumerable-in-javascript/

Alternative is to form an a plain object like threadsjs does instead of using classes.

[CMCDragonkai]

There are some interesting timer APIs: https://nodejs.org/api/timers.html#timeoutrefresh

[CMCDragonkai]

Note that since TaskId is a IdSortable, it's strictly monotonic due to our storing of the last task ID...

But this assumes the last Task ID is always stored, and we are intending on deleting tasks off the schedule once completed. I wasn't thinking keeping historical tasks are useful (except for maybe debugging? Although it seems like it would be dropped in production, and logging/tracing systems should be maintaining the audit log).

This means the last task ID may be undefined. So we would store the last Task ID regardless of whether there are any tasks left in the scheduler.

Furthermore, when the clock is shifted backwards, the time will be incremented by 1 until it is greater than the last time. The 1 is the smallest unit of precision, in which case this would be 1 millisecond.

Afterwards, it will be strictly monotonic ID but have a weakly monotonic timestamp up to 4096 IDs per millisecond. After 4096 it would roll over.

The expectation is that it's not possible to generate more than 4096 IDs in a millisecond, so by that time, the time must have increased by at least 1 millisecond.

Anyway this means we need to store Scheduler/lastTaskId separate from the Scheduler/tasks level.

[CMCDragonkai]

Benchmark in js-workers shows that the overhead to call the workers takes about 1.16 to 1.5ms.

https://github.com/MatrixAI/js-workers/blob/27fb8c3051f0880b81a14ea7daee47b15a94dd89/benches/results/worker_manager_metrics.txt#L2

Worker Threads Intersection.xlsx

A CPU intensive task should be greater than that time to be worth sending to the worker.

However most scheduling work seems it might not actually be CPU intensive. Like NodeGraph and Discovery is mostly IO. I suppose discovery may have have CPU work to pattern match the data to find the right data on the pages it loads, but this should be dominated by the time spent on IO.

Furthermore sending it to a worker can introduce locking problems. The async locks do not work across the worker threads, they only work within the same event loop context. They are not thread-safe nor process-safe.

This should mean that we should not directly integrate WorkerManager into the Queue, instead individual domains may have their handlers directly pass work to the WorkerManager. The Queue does not decide this since it does not know the nature of the task. The domain that registers the handlers can decide the nature of the task. So they can execute within the main thread, or send it off to a web worker and await for it.

[CMCDragonkai]

This means naturally the Queue can have either 1 as a concurrency limit or 0 to indicate unbound concurrent limit. With an unbound concurrency limit, it just immediately proceeds to execute everything that is due for execution.

Priority only comes into play with a concurrency limit so that things get put into priority order. Otherwise all tasks will be asynchronous and immediately executed.

The worker's concurrency/parallel limit is not a concern of the Queue then.

[CMCDragonkai]

We decided not to bother with preventing resource starvation, however an idea is like this.

1. Take advantage of DB's natural key ordering.
2. Create a bimap index of Priority/Timestamp -> Task Id AND Timestamp/Priority -> Task Id
3. Now we can iterate task ids based on 2 compound indexes: highest priority + earliest timestamp AND earliest timestamp + highest priority
4. Use dynamic programming/kinetic priority function that iterates through both sublevels (indexes) simultaneously to fill up a fixed concurrency pool (if unlimited, this policy is unnecessary, just iterate through as fast as posssible)

Simultaneous iteration that uses the timestamp to weight the priority, where the timestamp delta starts from 0 and goes towards infinity. Once could say that this multiples the priority based on a "rate". A delta of 0 multiplies by 1. A delta of infinity multiplies by infinity. Therefore the rate produces a multiplier between 1 to infinity.

Here is an example of the 3 policies:

• pM = tD + 1 - linear
• pM = tD^e - exponential (increasing at an increasing rate)
• pM = ln(tD + 1) + 1 - logarithmic (increasing at an decreasing rate)

https://www.desmos.com/calculator/wnezpfgxqc

[CMCDragonkai]

Once we have the tasks system, all other domains should not have any kind of background processing implemented, they should delegate ALL of that functionality into the tasks system.

[CMCDragonkai]

The task management is ready. However integration into discovery and nodes domains is being done in #445.

Priority management is static, we won't bother with dynamic priority in #329 (comment) before we see it be a problem.

Issue description here is still relevant to #445, since it contains notes on how best to refactor the discovery system.