1
\$\begingroup\$

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);
    }
}
\$\endgroup\$
  • \$\begingroup\$ Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. \$\endgroup\$ – Heslacher Jun 25 '18 at 8:55
1
\$\begingroup\$

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?

\$\endgroup\$
  • \$\begingroup\$ I have edited the code according to some of your suggestions. Some variables were renamed and I got rid of the running_tasks array and replaced it with a counter. The id is needed to prevent an accidental write to the same file. It is quite specific for my usecase, but the alternative would be to handle this check somewhere else. The next_task function got split up. Then then for the promised_task (now only task) is needed, because a promise needs to be either resolved or rejected, or it will never end. I used bind to get rid of the parameters. Thanks. \$\endgroup\$ – Endauriel Jun 25 '18 at 8:38
  • \$\begingroup\$ @Endauriel Please do not add, remove, or edit code in a question after you've received an answer. The site policy is explained in What to do when someone answers. \$\endgroup\$ – 301_Moved_Permanently Jun 25 '18 at 8:50
  • \$\begingroup\$ @MathiasEttinger I am sorry, I will remove the changed code again. \$\endgroup\$ – Endauriel Jun 25 '18 at 8:54

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.