Agenda: Questions about concurrency and co

[patrickd-]

now(name, [data], [cb])

Schedules a job to run name once immediately.

Reading that I understand that the job called name would be executed asap and only once.

Looking at the tests though

      it('runs a one-time job after its lock expires', function (done) {
        var runCount = 0;

        jobs.define('lock job', {
          lockLifetime: 50
        }, function (job, cb) {
          runCount++;

          if(runCount !== 1) {
            expect(runCount).to.be(2);
            jobs.stop(done);
          }
        });

        jobs.processEvery(50);
        jobs.start();
        jobs.now('lock job', { i: 1 });
      });

the test seems to expect that although .now() was only called once it is executed twice.
What I'd expect is that after calling .now() once is that also job job is executed once because the documentation says "run name once immediately".

Am I getting it wrong?

Also it seems with every now() call an item is created in the database which stays there until you actually call .purge() to remove it. Shouldn't they be automatically removed once they have run? Or is it for logging purposes?

processEvery(interval)

Decreasing the frequency will result in fewer database queries, but more jobs being stored in memory.

Why is that?
In general I have problems to understand how the job system works. Some jobs are in memory some are in the database... Is there a general explanation about how this works?

Concurrency <> Lock limits

concurrency: number maximum number of that job that can be running at once (per instance of agenda)

So concurrency is to ensure how many instances of a job can be running at the same time (which is done by locking?)

lockLimit: number maximum number of that job that can be locked at once (per instance of agenda)

Jobs are locked because they are running right?
What's the difference between lockLimit and concurrency?
Could you give me an example?

schedule(when, name, [data], [cb])

Schedules a job to run name once at a given time. when can be a Date or a String such as tomorrow at 5pm

Looking at the code it is possible that a single schedule call will cause a job to be scheduled multiple times if the lockLifetime is a small value (eg 1) because the default concurrency value is 5.
So just like with .now() it sounds like if you use schedule() you can be sure that a job is only executed once but it's actually not?

Also calling schedule() creates an database entry which is still there after the job has run. But not even calling purge() will clean them up. So with every schedule() call there'll be more and more items in the database that cannot be purge()ed. Shouldn't purge remove them()?

Maybe these are all stupid questions.. well, I'm not sure anymore. The more I worked with agenda the more confused I got :-/

[Hanzofm]

+1 to questions. Please answer it!

[mecp]

Concurrency <> Lock limits

concurrency: number maximum number of that job that can be running at once (per instance of agenda)
So concurrency is to ensure how many instances of a job can be running at the same time (which is done by locking?)

lockLimit: number maximum number of that job that can be locked at once (per instance of agenda)
Jobs are locked because they are running right?
What's the difference between lockLimit and concurrency?
Could you give me an example?

I am not sure if this library is maintained, but it'd be really great if learned users could help here with these question.

[dylanjha]

@joeframbach If you happen to know, any chance you could shed some light on these details?

[lushc]

I haven't been using this library for long but I'll try and answer to the best of my ability. If I've said anything wrong please correct me!

the test seems to expect that although .now() was only called once it is executed twice.

This is because it's testing whether a job is picked back up after its lock expires. Both the lock lifetime and the process tick are set to 50ms, so the job runs once when created with now(), but the job isn't finished by calling cb() so it doesn't automatically unlock. Therefore, on the next tick, the lock has now expired and so the job is picked up and ran again.

Also it seems with every now() call an item is created in the database which stays there ... Or is it for logging purposes?

This is intended to persist context surrounding the job, e.g. how many times it has failed. A now() job won't run again because it hasn't got a repeat interval set, so a nextRunAt time won't be calculated.

Some jobs are in memory some are in the database... Is there a general explanation about how this works?

On each process tick the database is queried to find and lock jobs, filling an in-memory queue of job objects to process (run the job definition, calculate the next run time, save attributes etc.) Say you have a set a process interval of 30 seconds: this means that Agenda will find all unlocked jobs that are due to run before or within the next 30 seconds, locking them and pushing them to the queue. By increasing the interval you're increasing the range of future jobs to be processed, so more jobs will be in-memory. Think of the database as "cold storage" for your jobs.

What's the difference between lockLimit and concurrency?

As you said, concurrency is to do with how many instances of a job can run at once. The lock limit ties into my previous answer of the in-memory queue, basically it ensures that not too many jobs are queued for processing. You'd want to use both of these settings together to help manage the size & performance of the in-memory queue. Say you've got a CPU-intensive job definition and 10,000 jobs in the database, with each individual job containing a lot of data: lockLimit will help control the size of the queue, guarding against using too much memory, while concurrency will ensure you don't bring your server to its knees by running too many of them concurrently.

Looking at the code it is possible that a single schedule call will cause a job to be scheduled multiple times if the lockLifetime is a small value (eg 1) because the default concurrency value is 5.

Yes, if the job had a lock lifetime of 1ms and the lock couldn't be relinquished before the next process tick, then it would indeed run again. That's why there is such a high default of 10 minutes, it's more of an insurance policy that if the job times out or crashes it'll still be picked up in a future tick.

Also calling schedule() creates an database entry which is still there after the job has run. But not even calling purge() will clean them up. So with every schedule() call there'll be more and more items in the database that cannot be purge()ed.

The purpose of purge is to remove jobs in the database that no longer have a matching job definition that's instantiated in your code, e.g. if "myJob" got renamed to "my-job", it'll clear all jobs in the database that are still called "myJob". If you want to cleanup jobs then cancel is the best bet as you can use your own custom query to control what's removed, e.g.

// remove all "myJobs" that aren't scheduled to run again
agenda.cancel({name: 'myJob', nextRunAt: null}, function(err, numRemoved) {
  console.log(numRemoved + ' removed');
});

[joeframbach]

I'll have time tomorrow to take a look

[simison]

@lushc @patrickd- @dylanjha do you feel like docs need improving regarding this (_wanna PR?_) and are questions now answered — can this be closed?

[patrickd-]

It's been a while since I opened the issue so i'm honestly not entirely sure if all the questions I had back then have been answered (was hard to put my confusion into text form). To actually be sure of that i'd have to work on our cron component again.

But lushc's answers looks really good; might be a shame to let them go to waste in here.

That aside this probably can be closed, thanks everyone!

[simison]

@patrickd- thanks! this repo hasn't been very much maintained for a while but we're getting back on track again (#441). Cheers!

I'll leave it for @lushc and @joeframbach to consider if docs need upgrade and let them close this.

[joeframbach]

Sorry, I have not had time to look into these questions. Perhaps other @agenda/maintainers could look at the documentation.

[dylanjha]

I'm running into some of the details here, posted a question on SO if anyone could possibly chime in :). After I get this sorted out I'm happy to put in a PR for the docs to be updated if it can be made more clear:

https://stackoverflow.com/questions/46349947/how-to-set-agenda-job-concurrency-properly

[dylanjha]

I'm seeing some behavior here that seems to be inconsistent with how defaultConcurrency is explained.

Example:

• defaultConcurrency is set to 5
• There are 10 jobs "queued" in the database
• One worker process starts - pulls down 5 jobs and starts running them (expected)
• Second worker process starts - no more jobs are pulled down. (expected 5 more jobs to start running)

It seems like defaultConcurrency is the number of concurrent jobs being run across _all_ agenda processes, when it should be the number of concurrent jobs for _each_ agenda process.

I set up a repo here with very simple steps to demonstrate this (just clone and run a yarn command to see it, exact steps are in the README)

• https://github.com/dylanjha/agenda-example

Anyone @agenda/maintainers might be able to help? I feel like I'm missing a simple configuration to get this to behave has I would expect, and I will submit a PR to update the documentation if I can get it sorted out

Edit: sorry, forgot to add the agenda-example repo link, added it now

[dylanjha]

Can anyone at @agenda/maintainers confirm if defaultConcurrency is expected to work like this:

It seems like defaultConcurrency is the number of concurrent jobs being run across all agenda processes not the number of concurrent jobs for each agenda process.

[princjef]

@dylanjha If I understand the answers to the questions above, it sounds like you would need to set lockLimit to the same value as well to avoid this problem. By default it's set to 0, which would consume as many jobs as were available (in this case, 10), waiting to pass the additional jobs to you for processing until you finish the jobs you're processing. When the second worker comes online, all of the jobs would already be locked, so nothing would be received by it.

Does it work as expected if you set lockLimit to 5?

[dylanjha]

@princjef I'm not sure I understand your question.

it sounds like you would need to set lockLimit to the same value as well to avoid this problem. By default it's set to 0, which would consume as many jobs as were available (in this case, 10)

In the current settings in the example:

const agendaConfig = {
  db: {address: process.env.MONGO_URL},
  defaultConcurrency: process.env.AGENDA_DEFAULT_CONCURRENCY
}

There is no lockLimit set, so the default would be 0, meaning no lock limit and all 10 jobs should be consumed, but that is not the behavior I'm seeing.

If possible, can you elaborate on how lockLimit would affect the issue I am running into? This is the example repo to reproduce the issue: https://github.com/dylanjha/agenda-example

[princjef]

I'm definitely not an expert on agenda, but I was thinking that your problem might have been that the first worker was locking all 10 jobs but only giving you the first five for processing (due to your defaultConcurrency). If all 10 are locked before your second worker comes up, it shouldn't receive any messages until the locks start to expire on the messages being held by worker one. (I'm not positive this is how things work in agenda, but that's my understanding from the docs and descriptions in this issue)

My suggestion would be to try something like this to see if it causes your second worker to receive messages:

const agendaConfig = {
  db: {address: process.env.MONGO_URL},
  defaultConcurrency: process.env.AGENDA_DEFAULT_CONCURRENCY,
  defaultLockLimit: process.env.AGENDA_DEFAULT_CONCURRENCY
}

[dylanjha]

@princjef good thinking! I just tried this and you are 100% right.

To summarize:

If you lock limit is higher than the concurrency then one worker will put a "lock" on more jobs than it can processes. Note that by default lock limit is "no limit" meaning that one worker will lock all the available jobs.

Thank you @princjef, unfortunately I moved away from agenda a couple months ago (#570 & #551) but good to know what the problem was.

Posted this answer on the SO question (https://stackoverflow.com/questions/46349947/how-to-set-agenda-job-concurrency-properly/49389826)