Most JavaScript and TypeScript developers are probably familiar with the
Promise API:
async function processData(): Promise<void> {
try {
const data = await fetchData();
console.log(data);
} catch (error) {
console.error(`Error: ${error}`);
}
}
await processData();
Promises can be chained with .then() and .catch(). You can also use
async/await in your JavaScript if your functions return a promise in order
to automatically unwrap your data from your promise.
If you don't absolutely always await your promises, bad things happen, right?
Right?
Well, no. And sometimes you might want to avoid awaiting your Promises on
purpose.
If your code does not care about the success, failure, or side effects of a
Promise, you really should just avoid await-ing it at all.
Promises in Musebot
In my Musebot project, which links Discord with various generative AI solutions, I have to await many different types of requests to different systems.
To keep things simple, I wanted to manage these requests as similarly as possible:
- Requests should be limited to the same system one at a time to limit actively awaited promises and parallel network traffic.
- Requests should be retried on failure.
- Requests should be retried on a delay.
- Requests to different systems should be allowed to process in parallel.
I ended up creating a TaskQueue based on Promises.
export class TaskQueue implements ITaskQueue {
/** Channels for segregating tasks. */
#channels: TaskChannel[] = [];
/**
* Indicates whether there are any active tasks.
* @returns: {boolean}
*/
get isActive(): boolean {
// ...
}
constructor() {
}
/**
* Add a task to the task queue and split it into the corresponding channel.
* @param task The task to add.
*/
add(task: BaseTask<unknown>): void {
// ...
}
/**
* Processes the task queue.
* @private
* @returns {Promise<void>}
*/
async #processQueue(): Promise<void> {
// ...
}
/**
* Retrieves the next tasks to process.
* @private
* @returns {BaseTask<unknown>[]}
*/
#getNextTasks(): BaseTask<unknown>[] {
// ...
}
/**
* Cleans up channels.
* @private
*/
#cleanChannels() {
// ...
}
}
The add(task: BaseTask<unknown>) method is the main entry point for adding
tasks to the appropriate TaskChannel and cannot be await-ed. Once a task is
added to the queue, processQueue() is called to start enumerating eligible
tasks.
Since processQueue() is also not await-ed by add(task: BaseTask<unknown>),
the implementation effectively no-ops on subsequent calls if #getNextTasks()
returns no additional tasks to process from within processQueue(). This allows
tasks to be added as quickly as they are available, concurrently if possible, or
the queue will process them in turn if not.
#getNextTasks() is also called after the previous task list is completed
before exiting.
Due to the potential no-op behavior of calling processQueue() from
add(task: BaseTask<unknown>), this effectively acts as a hint that there
might be new tasks to process in the queue. Only if the queue isn't currently
active will any operation actually occur.
The task queue will not block any code calling into it, and we still get our asynchronous and optionally parallel behavior.
Musebot has been both a really interesting project to work on hilarious to use. If you want to grab your own copy, you can find Musebot for purchase on Discord. Your own SwarmUI/ComfyUI and/or Ollama instances will need to be configured beforehand.
