Batched job processing (opt-in)

[benjie]

Description

Replaces #99 and #470.

If you're at very high scale (e.g. you're running multiple Worker instances, and each instance has high concurrency) then the act of looking for and releasing jobs can start to dominate the load on the database. The PR gives the ability to configure Graphile Worker such that getJob (via localQueueSize), completeJob (via completeJobBatchDelay) and failJob (via failJobBatchDelay) can be batched, thereby reducing this database load (and improving job throughput). This is an opt-in feature, via the following settings:

const preset = {
  worker: {
    localQueueSize: jobConcurrency,
    completeJobBatchDelay: 10, // milliseconds
    failJobBatchDelay: 10, // milliseconds
  }
};

• If localQueueSize >= 1, Pools become responsible for getting jobs and will grab the number of jobs that you specify up front, and distribute these to workers on demand. This is done via a "Local Queue".
• If completeJobBatchDelay >= 0 or failJobBatchDelay >= 0 then pools are also now responsible for completing or failing jobs (respectively); they will wait the specified number of milliseconds after a completeJob or failJob call and batch any other calls made in the interrim; all of these results will be sent to the database at the same time reducing the total number of transactions.

Note that enabling these features changes the behavior of Worker in a few ways:

• Since pools grab a batch of jobs up front they represent a snapshot at that time and newer higher priority jobs will not be evaluated until the batch is done being processed
• Since pools grab a batch of jobs up front, jobs may not be as evenly distributed across workers
• Since pools grab a batch of jobs up front, jobs may not start until a later time than they previously did, potentially increasing latency (but also increasing throughput)

Performance impact

If not enabled, impact is minimal.

If enabled, throughput improvement at the cost of potential latency increases.

The following results were produced with the following setup:

• CPU: i9 14900K
• OS: Ubuntu
• OS tweaks: efficiency cores disabled, and CPU configured to use performance governor
• Job count: 200,000
• Worker instance count: 4 (4 Node.js processes, each running one Worker instance, via the CLI)
• Worker concurrency: 24 (each of the 4 instances can process 24 jobs concurrently, for a total of 96 concurrent jobs)

Base performance:

Jobs per second: 16093.94

With localQueueSize: 500:

Jobs per second: 35177.47

Performance with localQueueSize: 500, completeJobBatchDelay: 0, failJobBatchDelay: 0 (note: even though the numbers are 0 this still enables batching, it is just limited to (roughly) a single JS event loop tick):

Jobs per second: 180684.70

You should note that the workload benchmarked here is a workload designed to put maximal stress on the database (i.e. the tasks are basically no-ops); YMMV with real-world loads.

The CPUs were configured with this script:

#!/usr/bin/env bash

# Enable performance governor
echo performance | sudo tee /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor

# Disable the efficiency cores
for i in {16..31}; do
    echo 0 | sudo tee /sys/devices/system/cpu/cpu$i/online
done

Security impact

Not known.

Checklist

• My code matches the project's code style and yarn lint:fix passes.
• I've added tests for the new feature, and yarn test passes.
• I have detailed the new feature in the relevant documentation.
• I have added this feature to 'Pending' in the RELEASE_NOTES.md file (if one exists).
• If this is a breaking change I've explained why.

Fixes #501

[benjie]

Things to check that I actually implemented (from my notes):

• Nudge on new job notification.
• Must release jobs in the queue if held for greater than a certain period of time (default 5 minutes?)
• Must fetch the next batch even if current batch is ongoing if we know we cannot possibly get enough results
• getJob should return a defer if no jobs currently available
• Must perform polling only when no fetches are in progress and there are no jobs in the queue and there is at least one pending defer
• on pool release, defers must all be resolved to undefined
• When the last job is assigned, automatically fetch the next batch. Then set up polling. Polling should be interrupted if a new job announcement is received.
• Only refetch when the queue is emptied if the previous fetch was at max limit (i.e. 4 jobs for 4 slots).
• Do not fetch on "new job" announcement unless job cache is empty
• Timer is active if workers are waiting (deferreds)
• Reset timer after fetching jobs
• New job announcment triggers job fetch when workers waiting
• Timer is active if and only if queued jobs is zero
• localQueueSize
• localQueueTTL
• Rate limit getJob when the queue is empty to avoid overwhelming the database; make this configurable

Also:

• Make pro compatible
• Revisit ttlExpiredTimer - perhaps this should persist because some of the jobs may have been in the local queue for ages even if the odd worker has completed from time to time.
• Create tests for refetchDelay code

[benjie]

To enable feedback on this PR, I've released it as a canary; install via graphile-worker@canary. Please note that this PR does not have tests or documentation, so I would not recommend using it in a production environment - try it in staging and see if it fixes your performance issues (when configured to do so).

(Also judging by the sporadic CI failures, there's clearly some kind of bug/timing issue somewhere.)

[psteinroe]

hey, after some basic testing i decided to just swing it and deployed graphile worker canary to prod while closely monitoring. we are mainly exporting jobs and they are deduped so not a lot can go wrong.

results are really awesome. improvements pretty much all over the board:

• cpu usage of the db dropped from 40+ to 20
• time consumed by the job query is reduced from 97% to 49%
• job latency (run_at until its done executing) dropped from 3 seconds at peak to around 500ms

current settings, choose randomly really...

    preset: {
      extends: [WorkerProPreset],
      worker: {
        connectionString: DB_CONNECTION_STRING,
        schema: 'graphile_worker',
        sweepThreshold: 600000, // 10 minutes
        concurrentJobs: 24,
        // pollInterval is only relevant for scheduled jobs (after now()) and retries
        pollInterval: 120_000,
        maxPoolSize: 24,
        localQueueSize: 24,
        completeJobBatchDelay: 100, // milliseconds
        failJobBatchDelay: 100, // milliseconds
      },
    },

some visuals:

[psteinroe]

@benjie after two weeks we just had our first crash. here are the logs:

Failed to release job '17207078' after success\; committing seppukuCannot read properties of null (reading 'message')

Worker exited, but pool is in continuous mode, is active, and is not shutting down... Did something go wrong?

Worker exited with error: TypeError: Cannot read properties of null (reading 'message')

Failed to return jobs from local queue to database queue

[benjie]

Thanks for the report of the crash! There's clearly an issue preventing the tests from reliably passing too, I suspect some kind of race condition has slipped in.

Did these errors not have stack traces? If they have stack traces, that would significantly help in tracking down the issue!

[psteinroe]

This is the closest I can get to a stack trace:

[hillac]

Given the error is coming from here:

worker/src/worker.ts

Lines 281 to 282 in 436e299

	.map((e) => (e as Error).message ?? String(e))
	.join("\n")}`,

Could you just coalesce the entryResult.reason a few lines up:
entryResult.reason ?? 'No reason provided'
Or something similar, check if message in the e etc. The promise must have been rejected with null as the reason. Also if the promise is rejected with no reason it will also be undefined, so It's not really safe to assume e is an object.

Although this doesn't actually help find out the real reason the promise rejected. Is it possible your task executor is calling 'reject(null)' on the promise?

[benjie]

@hillac I've fixed that line by adding a ? - you're right that there's nothing forcing the rejection reason to be an object/error. (This may well be an issue that the task itself is rejecting with null.)

How did you determine that was the correct location?

We still seem to have an issue prevent CI passing reliably.

New release contains the ? fix:

[email protected]

[hillac]

@psteinroe's stack trace has the line number, worker.js:222:43. You can look in the node modules path the stack trace shows to see the js code. Or here.

[benjie]

stack trace has the line number, worker.js:222:43

I read worker.js as main.js again 🤦‍♂️ I definitely need more sleep! 😴

[benjie]

Hooray; seems that the instability in this PR was actually in the test framework rather than Worker itself. 4 passes in a row suggests I have stabilized this now, plus the tests are now faster (locally, not in CI) because they run in parallel, and they're independent of each other since they run on clean databases.

[benjie]

Latest canary released: [email protected]

I've extracted the following PRs from this PR so that it's simpler to review:

• New test framework, extracted from #474 #506
• Extract perfTest enhancements from #474 #507

[benjie]

The latest canary is out and it contains a number of fixes to other parts of Worker.