Skip to content

Commit

Permalink
fix: ensure jobs are filled up till concurrency reached
Browse files Browse the repository at this point in the history
  • Loading branch information
simllll committed Oct 15, 2020
1 parent 39950f3 commit 1a8bb31
Show file tree
Hide file tree
Showing 2 changed files with 51 additions and 9 deletions.
21 changes: 12 additions & 9 deletions src/JobProcessor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -154,10 +154,8 @@ export class JobProcessor {
'job [%s] lock status: shouldLock = %s',
name,
shouldLock,
status?.locked,
this.jobQueue.length,
this.lockedJobs.length,
this.totalLockLimit
`${status?.locked} >= ${jobDefinition?.lockLimit}`,
`${this.lockedJobs.length} >= ${this.totalLockLimit}`
);
return shouldLock;
}
Expand Down Expand Up @@ -240,6 +238,8 @@ export class JobProcessor {
this.lockedJobs.push(jobToEnqueue);
this.enqueueJob(jobToEnqueue);
this.jobProcessing();
} else {
log.extend('lockOnTheFly')('cannot lock job [%s] on the fly', job.attrs.name);
}
}
} finally {
Expand Down Expand Up @@ -378,6 +378,9 @@ export class JobProcessor {
this.jobProcessing();
}, runIn);
}

// additionally run again and check if there are more jobs that we can process right now (as long concurrency not reached)
setImmediate(() => this.jobProcessing());
}

/**
Expand Down Expand Up @@ -437,8 +440,8 @@ export class JobProcessor {
log.extend('runOrRetry')('[%s:%s] processing job', job.attrs.name, job.attrs._id);

// check if the job is still alive
const checkIfJobIsDead = () => {
// check every processInterval
const checkIfJobIsStillAlive = () => {
// check every "this.agenda.definitions[job.attrs.name].lockLifetime"" (or at mininum every processEvery)
return new Promise((resolve, reject) =>
setTimeout(() => {
if (job.isDead()) {
Expand All @@ -456,13 +459,13 @@ export class JobProcessor {
resolve();
return;
}
resolve(checkIfJobIsDead());
}, this.processEvery)
resolve(checkIfJobIsStillAlive());
}, Math.max(this.processEvery, this.agenda.definitions[job.attrs.name].lockLifetime))
);
};

// CALL THE ACTUAL METHOD TO PROCESS THE JOB!!!
await Promise.race([job.run(), checkIfJobIsDead()]);
await Promise.race([job.run(), checkIfJobIsStillAlive()]);

log.extend('runOrRetry')(
'[%s:%s] processing job successfull',
Expand Down
39 changes: 39 additions & 0 deletions test/jobprocessor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -134,4 +134,43 @@ describe('JobProcessor', function () {

expect(promiseResult).to.not.be.an('error');
});

it('ensure concurrency is filled up', async () => {
agenda.maxConcurrency(300);
agenda.lockLimit(150);
agenda.defaultLockLimit(20);
agenda.defaultConcurrency(10);

for (let jobI = 0; jobI < 10; jobI++) {
agenda.define(
`test job ${jobI}`,
async job => {
await new Promise(resolve => setTimeout(resolve, 5000));
},
{ lockLifetime: 10000 }
);
}

// queue up jobs
for (let jobI = 0; jobI < 10; jobI++) {
for (let jobJ = 0; jobJ < 25; jobJ++) {
agenda.now(`test job ${jobI}`);
}
}

await agenda.start();

const allJobsStarted = new Promise(async resolve => {
let runningJobs = 0;
do {
runningJobs = (await agenda.getRunningStats()).runningJobs as number;
await new Promise(wait => setTimeout(wait, 50));
} while (runningJobs < 100);
resolve('all started');
});

expect(
await Promise.race([allJobsStarted, new Promise(resolve => setTimeout(resolve, 1500))])
).to.equal('all started');
});
});

0 comments on commit 1a8bb31

Please sign in to comment.