I had to do a lot of IO stuff on my server running Node JS where I did not really care about the result of the operations. The list of incoming tasks could be huge, so I had to limit the number of tasks that could run at the same time. I try to avoid third party libraries wherever I can, so I made my own method.
It seems to work fine, but could I still improve something?
const pending_tasks = [];
const running_tasks = [];
const identifiers = [];
let concurrency = 1;
/*
Set how many tasks will run concurrently.
*/
function set_concurrency (max) {
concurrency = max;
}
/*
Check whether a task with the specified id is already in the queue.
Only tasks that were added to the queue with their id parameter defined can be checked against.
*/
function in_queue (id) {
if (!id) { return false; }
return identifiers.includes(id);
};
/*
Add a task to the list of running tasks.
*/
function fired (id, task) {
if (id) { identifiers.push(id); }
running_tasks.push(task);
}
/*
Delete a task from the list of running tasks.
*/
function finished (id, task) {
if (id) { identifiers.splice(identifiers.indexOf(id), 1); }
running_tasks.splice(running_tasks.indexOf(task), 1);
}
/*
Add a task to the queue.
*/
function pending (id, task, parameters) {
pending_tasks.push({
'id': id,
'task': task,
'parameters': parameters
});
}
/*
Clean up and queue logic after a task has finished.
*/
function next_task (id, task) {
/*
Delete the task from the queue.
*/
finished(id, task);
/*
If there are still pending tasks in the queue ...
*/
if (pending_tasks.length > 0) {
/*
... and there are less tasks running than can run concurrently ...
*/
if (running_tasks.length < concurrency) {
/*
... get another task from the pending list ...
*/
const pending_task = pending_tasks.shift();
/*
... and start it.
*/
fire_and_forget(
pending_task['id'],
pending_task['task'],
pending_task['parameters']
);
}
}
}
/*
Add tasks to the queue and have them execute without caring about the result.
You should set the number of concurrent tasks with the 'set_concurreny (max)' function, if
you wish for more than one task to be executed at a time.
If a specific task should be unique and only have one instance of itself running, give it
an id and check whether it is already in the queue with the 'in_queue (id) function.'
*/
async function fire_and_forget (id, task, parameters) {
/*
If an id is defined, check whether it is already in the queue.
You should check this yourself before executing this function to avoid an exception.
*/
if (in_queue(id)) { throw 'Task with id \'' + id + '\' is already in queue.'; }
/*
If there are less tasks running than can run concurrently, start this task immediately.
*/
if (running_tasks.length < concurrency) {
/*
Create a promise and apply the parameters to the task to start it.
*/
const promised_task = new Promise(function (resolve, reject) {
task.apply(this, parameters)
.then(() => { resolve(); })
.catch(() => { reject(); });
});
/*
The task has finished.
*/
promised_task.then(() => {
next_task(id, promised_task);
}).catch(() => {
next_task(id, promised_task);
});
/*
Add the task to the list of running tasks.
*/
fired(id, promised_task);
}
else {
/*
Add the task to the queue.
*/
pending(id, task, parameters);
}
};
module.exports.set_concurrency = set_concurrency;
module.exports.in_queue = in_queue;
module.exports.fire_and_forget = fire_and_forget;
/*
TEST SCENARIO --- Remove code beneath for production
*/
// Dummy task
async function long_write_task(subject, body) {
await new Promise((resolve) => setTimeout(resolve, 1000));
console.log([subject, body]);
}
// Set concurrency to 3 tasks.
set_concurrency(3);
// Start 10 tasks.
for(let i = 0; i < 10; i++){
// Dummy parameter list.
const parameters = ['subject ' + i, 'body ' + i];
// Check if id already queued.
// Can't be if we just assign the loop iteration number as id, but is best practice.
// If you don't care about running an identical task twice, leave id 'undefined'.
if (!in_queue(i)) {
fire_and_forget(i, long_write_task, parameters);
}
}
EDITED CODE
const enqueued_ids = [];
const enqueued_tasks = [];
let running = 0;
let concurrency = 10;
/*
Set how many tasks will run concurrently.
*/
function set_concurrency (max) {
concurrency = max;
};
/*
Check whether a task with the specified id is already in the queue.
*/
function in_queue (id) {
if ([undefined, null].includes(id)) { return false; }
return enqueued_ids.includes(id);
};
/*
Add the id for a task to the list of enqueued ids.
*/
function enqueue_id (id) {
if ([undefined, null].includes(id)) { return; }
enqueued_ids.push(id);
};
/*
Remove the id after the task has finished.
*/
function dequeue_id (id) {
if ([undefined, null].includes(id)) { return; }
enqueued_ids.splice(enqueued_ids.indexOf(id), 1);
};
/*
Start a new task from the queue.
*/
function fire_task () {
/*
If there are pending tasks in the queue ...
*/
if (enqueued_tasks.length > 0) {
/*
... and there are less tasks running than can run concurrently ...
*/
if (running < concurrency) {
/*
... get a task from the queue ...
*/
const enqueued_task = enqueued_tasks.shift();
const id = enqueued_task['id'];
const task = enqueued_task['task'];
/*
... and start it.
*/
const fired_task = new Promise(function (resolve, reject) {
running++;
task()
.then(() => { resolve(); })
.catch(() => { reject(); });
});
/*
The task has finished.
*/
fired_task.then(() => {
running--;
dequeue_id(id);
fire_task();
}).catch(() => {
running--;
dequeue_id(id);
fire_task();
});
}
}
};
/*
Add tasks to the queue and have them execute without caring about the result.
You should set the number of concurrent tasks with the 'set_concurreny (max)' function, if
you wish for more than one task to be executed at a time.
If a specific task should be unique and only have one instance of itself running, give it
an id and check whether it is already in the queue with the 'in_queue (id)' function.
*/
async function fire_and_forget (id, task) {
/*
If an id is defined, check whether it is already in the queue.
You should check this yourself before executing this function to avoid an exception.
*/
if (in_queue(id)) { throw 'Task with id \'' + id + '\' is already in queue.'; }
/*
If the id is valid, add it to the list of enqueued ids.
*/
enqueue_id(id);
/*
Bind the parameters to the task.
*/
const bound_task = task.bind.apply(task, [null].concat([].slice.call(arguments).slice(2)));
/*
Add the task to the queue.
*/
enqueued_tasks.push({
'id': id,
'task': bound_task
});
/*
Start a new task from the queue.
*/
fire_task();
};
module.exports.set_concurrency = set_concurrency;
module.exports.in_queue = in_queue;
module.exports.fire_and_forget = fire_and_forget;
/*
TEST SCENARIO --- Remove code beneath for production
*/
// Dummy task
async function long_write_task(subject, body) {
await new Promise((resolve) => setTimeout(resolve, 1000));
console.log([subject, body]);
};
// Set concurrency to 3 tasks.
set_concurrency(3);
// Start 10 tasks.
for(let i = 0; i < 10; i++){
// Check if id already queued.
// Can't be if we just assign the loop iteration number as id, but is best practice.
// If you don't care about running an identical task twice, leave id 'undefined'.
if (!in_queue(i)) {
fire_and_forget(i, long_write_task, 'subject ' + i, 'body ' + i);
}
}