How to Centralize Concurrency Management

[izhouteng]

Description

1 node instance, this code works fine.

But I start 8 instances with PM2 and actually the concurrency will become 40.

How can I control concurrency under multiple instances of PM2?

const bullQueue = require('bull');
const queue = new bullQueue('taskqueue', {...})
const concurrency = 5;
queue.process('jobTypeA', concurrency, job => {...do something...});

Minimal, Working Test code to reproduce the issue.

(An easy to reproduce test case will dramatically decrease the resolution time.)

Bull version

Additional information

[manast]

I am not sure I understand the question, because just stating the obvious and what you said yourself, you control the number of workers with pm2 and the concurrency per worker with the concurrency option...

[manast]

@the-spyke Ok, and you have a rate-limit enabled in this case?

[manast]

Sounds strange that just 1k would create such effect on the CPU, something else must be at play here.

[the-spyke]

@manast I don't have a rate limiter. That's why I started to read the docs about it and concurrency option. After dispatching the batch I see a huge spike in SetTypeCmds (others too, but much less). I'd like to understand how it should work before I will change any code.

For example, concurrency is 1 by default, which means you need rate limiter only if you want to "add pauses" between tasks. And in case of a k8s concurrency will be equal to the scaling factor of the service. Also, how rate limiter would work, if I have 2 workers with

new Queue('name', { limiter: { max 5, duration: 1000 } });

and another service adding tasks into

new Queue('name');

Another one: how concurrency works if the queue has only one handler registered as *.

[manast]

Adding 1k jobs (if done in parallel) would take less than one second, so no chance that the Redis CPU would create such a big spike. The concurrency setting just describes how many jobs a given worker (a Queue instance with a process function defined) will process in parallel. For IO based jobs a concurrency of something between 100 and 300 works normally quite well.
For the rest of the questions you should be able to find everything you ask in the documentation.

[the-spyke]

My guess was that the spike appears because of too many concurrent workers. But docs don't say anything special about *, so in my case it should be default concurrency 1 multiplied by 3 (scaling). So, I decided to look into rate limiting. I suppose every queue instance manages rate limit for itself, but seeing limiter code in the moveToActive.lua made me wonder if it is baked into job data by the producing queue instead.