trjhtr

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);
 

It definitely makes sense. I can provide a few nit-picks since you are asking:

I always start with the names:

• 
  pending_tasks and running_tasks look like parallel structures, but their elements are different-- this is a little surprising to a new reader. I'd make them consistent or give them more distinct names.


• the nomenclature floats a little bit. Are you running or firing or enqueuing tasks?


• 
  id or identifier? Pick one.

Questions I had while reading the code:

It seems like if you enqueue the same task function with different parameters you can get into trouble. You remove it from the running list based only on the function, but if you have two running with different parameters, it may remove the wrong one. It may not matter, but...

I'm not sure about the use of id. I'm not sure how useful it is for the amount of complexity it adds.

The next_task function does several different things. It is cleaning up the previous task, checking the concurrency, and possibly launching a new task. I'd recommend you break these up into the two or three different actions.

Something is off in the definition of promised_task. The then doesn't seem right-- looks like a copy/paste error.

Consider using Function bind to solve several of the problems. You could bind the parameters to the function when you first see it, and then not have to carry around parameters at all. It would then make removing the function when it's done less ambiguous.

Do you really need to keep all the running tasks around? Could you just increment a count representing the number of running tasks?