Best method to manually poll for jobs?

[prescience-data]

I've just tried to implement a manual job getter, but it seems to return void if not jobs are immediately on the queue.

Before going further down this path, I wanted to check if this is a rational method to poll for jobs, or if there is a better way?

Without any real benchmarking my concern is that this approach across hundreds of workers seems like it would be thrashing fairly hard with a sufficiently small delay... 🤔

(Note: The reason for requesting jobs manually is to allow workers to return a completed job result, but not accept a new job until a secondary process is triggered.)

/**
 * Loops on a heartbeat until a job lock is acquired.
 *
 * @param {BullWorker} worker
 * @param {string} uuid
 * @param {number} delay
 * @return {Promise<SocketRequestJob>}
 */
export const acquireJob = async (
  worker: BullWorker,
  { uuid, delay }: JobAcquisitionOptions
): Promise<SocketRequestJob> => {
  // Set up the local variable to receive inbound jobs.
  let job: SocketRequestJob | undefined
  // Loop until a job lock is acquired.
  while (job === undefined) {
    job = (await worker.getNextJob(uuid)) || undefined
    // If no job was acquired, delay for a short delay before trying again.
    if (job === undefined) {
      await waitForTimeout(delay)
    }
  }
  // Return acquired job.
  return job
}

[ccollie]

I'd have to devote a few brain cells, but my approach would be to leverage events:

https://docs.bullmq.io/guide/events

See the following test:

bullmq/src/test/test_events.ts

Line 29 in 297ec30

it('should emit waiting when a job has been added', async function() {

[prescience-data]

Ah interesting, I hadn't considered the on add event ⚡🧠

[manast]

You do not need a delay, since "getNextJob" is already a blocking operation that waits "drainDelay" before it expires. The default value for drainDelay is 5 seconds (note it is expressed in seconds not milliseconds). You can increase this value to reduce the loop interval. "getNextJob" is the most efficient way to get jobs from the queue, I do not recommend using events for this usecase.

[prescience-data]

@manast

I do not recommend using events for this usecase.

Re the events, is that because it would trigger all the workers at the exact ms? or some other reason?

blocking operation that waits "drainDelay" before it expires

So "drainDelay" refers to the worker draining the socket / stream when it finishes checking for any jobs and begins to disconnect? (in this case with an empty result?)

[manast]

So "drainDelay" refers to the worker draining the socket / stream when it finishes checking for any jobs and begins to disconnect? (in this case with an empty result?)

No. When the queue is drained, i.e. does not have any items in the wait list, we issue a blocking command to redis to wait for a job to arrive. This command (brpoplpush) has a "timeout" parameter which specifies for how long the call should be blocking until it just returns, thats the "drainDelay", probably a better name would be more suitable here. The way redis works is that for all the clients that are blocked it will pick one in round robin fashion. If you have many workers you can increase this value considerably and mostly remove any overhead arising from the polling.

[prescience-data]

Ok so to put that in practical terms, this is the ideal implementation?

export const acquireJob = async (
  worker: BullWorker,
  { uuid }: JobAcquisitionOptions
): Promise<SocketRequestJob> => {
  let job: SocketRequestJob | undefined
  while (job === undefined) {
    job = (await worker.getNextJob(uuid)) || undefined
    if (job) {
      return job
    }
  }
}

[manast]

I cannot say if it is ideal for your particular use case, but why would you have a loop inside acquireJob instead of moving the looping higher up?

[prescience-data]

That was just to make it simpler to describe for the example, in the real code that would be inline.
But it sounds like it does what I think it does, which is excellent 🕺✌️

[prescience-data]

@manast
I think I must be doing something wrong here.

Both the server and the worker (appear to be) connected to the "flex" queue, but the worker never picks up a job and Taskforce.sh is showing zero workers...

Logs:

Taskforce:

[manast]

hmm, looks as if the workers are not really connected to the correct redis instance. I recommend you that you write a minimal POC in one file where you both add the jobs and fetch them manually to verify everything works as expected.

[prescience-data]

I just realised I was passing the worker a prefix option and the server (queue dispatcher) no prefix 🙈

[prescience-data]

Just to close this out for anyone stumbling across this in the future, I misunderstood @manast explanation around the drainDelay - I thought a shorter delay was required, but you actually want a longer delay.

The delay isn't related to a "polling" cycle, it refers to how long the worker stays connected to the queue waiting for a job before giving up. So a long number = waiting for a job for a long time (which is what I wanted in this situation).