Improve performance of queue locking

[ben-pr-p]

Feature description

The current query, with 161 job queues and several thousand jobs on one queue:

# explain analyze 
select jobs.queue_name, jobs.id
from "graphile_worker".jobs
where (jobs.locked_at is null or jobs.locked_at < (now() - '4 hours'::interval))
and (
  jobs.queue_name is null
or
  exists (
    select 1
    from "graphile_worker".job_queues
    where job_queues.queue_name = jobs.queue_name
    and (job_queues.locked_at is null or job_queues.locked_at < (now() - '4 hours'::interval))
    for update
    skip locked
  )
)
and run_at <= now()
and attempts < max_attempts
and (null is null or jobs.task_identifier = any(null))
and (null is null or (flags ?| null) is not true)
order by priority asc, run_at asc, id asc
limit 1
for update
skip locked;
                                                                                       QUERY PLAN                                                                                       
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Limit  (cost=0.41..227.88 rows=1 width=38) (actual time=242.627..242.629 rows=0 loops=1)
   ->  LockRows  (cost=0.41..2317910.12 rows=10190 width=38) (actual time=242.625..242.626 rows=0 loops=1)
         ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=0.41..2317808.22 rows=10190 width=38) (actual time=242.624..242.626 rows=0 loops=1)
               Index Cond: (run_at <= now())
               Filter: ((attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))) AND ((queue_name IS NULL) OR (SubPlan 1)))
               Rows Removed by Filter: 8859
               SubPlan 1
                 ->  LockRows  (cost=0.52..194.36 rows=1 width=10) (actual time=0.026..0.026 rows=0 loops=8857)
                       ->  Index Scan using job_queues_pkey on job_queues  (cost=0.52..194.35 rows=1 width=10) (actual time=0.026..0.026 rows=0 loops=8857)
                             Index Cond: (queue_name = jobs.queue_name)
                             Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                             Rows Removed by Filter: 1
 Planning Time: 0.250 ms
 Execution Time: 242.679 ms
(14 rows)

An alternative query, on the same table:

# explain analyze 
select jobs.queue_name, jobs.id
from "graphile_worker".jobs
where (jobs.locked_at is null or jobs.locked_at < (now() - '4 hours'::interval))
and (
  jobs.queue_name is null
or
  jobs.queue_name in (
    select queue_name
    from "graphile_worker".job_queues
    where true                                   
    and (job_queues.locked_at is null or job_queues.locked_at < (now() - '4 hours'::interval))
    for update
    skip locked
  )
)
and run_at <= now()
and attempts < max_attempts
and (null is null or jobs.task_identifier = any(null))
and (null is null or (flags ?| null) is not true)
order by priority asc, run_at asc, id asc
limit 1
for update
skip locked;
                                                                                      QUERY PLAN                                                                                       
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Limit  (cost=3554.48..3558.87 rows=1 width=38) (actual time=13.327..13.330 rows=0 loops=1)
   ->  LockRows  (cost=3554.48..48298.47 rows=10190 width=38) (actual time=13.326..13.328 rows=0 loops=1)
         ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=3554.48..48196.57 rows=10190 width=38) (actual time=13.326..13.327 rows=0 loops=1)
               Index Cond: (run_at <= now())
               Filter: (((queue_name IS NULL) OR (hashed SubPlan 1)) AND (attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))))
               Rows Removed by Filter: 8859
               SubPlan 1
                 ->  LockRows  (cost=0.00..3553.65 rows=166 width=42) (actual time=0.068..6.381 rows=158 loops=1)
                       ->  Seq Scan on job_queues  (cost=0.00..3551.99 rows=166 width=42) (actual time=0.012..5.852 rows=158 loops=1)
                             Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                             Rows Removed by Filter: 2
 Planning Time: 0.486 ms
 Execution Time: 13.396 ms
(13 rows)

Motivating example

We're seeing 70-100% CPU utilization from get_job on pretty big machines when there are more than a few thousand jobs that need to be skipped because their queue is locked.

Breaking changes

Shouldn't be any, although I'm not 100% sure what the behavior of for update skip locked in the subquery is

Supporting development

I [tick all that apply]:

• am interested in building this feature myself
• am interested in collaborating on building this feature
• am willing to help testing this feature before it's released
• am willing to write a test-driven test suite for this feature (before it exists)
• am a Graphile sponsor ❤️
• have an active support or consultancy contract with Graphile

[ben-pr-p]

It looks like for the exists route, the exists subquery is running N times. If I remove the for update skip locked, I get the same runtimes:

# explain analyze 
select jobs.queue_name, jobs.id
from "graphile_worker".jobs
where (jobs.locked_at is null or jobs.locked_at < (now() - '4 hours'::interval))
and (
  jobs.queue_name is null
or
  exists (
    select 1
    from "graphile_worker".job_queues
    where job_queues.queue_name = jobs.queue_name
    and (job_queues.locked_at is null or job_queues.locked_at < (now() - '4 hours'::interval))
  )
)
and run_at <= now()
and attempts < max_attempts
and (null is null or jobs.task_identifier = any(null))
and (null is null or (flags ?| null) is not true)
order by priority asc, run_at asc, id asc
limit 1
for update
skip locked;
                                                                                               QUERY PLAN                                                                                                
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Limit  (cost=0.41..227.85 rows=1 width=38) (actual time=12.680..12.683 rows=0 loops=1)
   ->  LockRows  (cost=0.41..2342140.38 rows=10298 width=38) (actual time=12.679..12.681 rows=0 loops=1)
         ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=0.41..2342037.40 rows=10298 width=38) (actual time=12.678..12.680 rows=0 loops=1)
               Index Cond: (run_at <= now())
               Filter: ((attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))) AND ((queue_name IS NULL) OR (alternatives: SubPlan 1 or hashed SubPlan 2)))
               Rows Removed by Filter: 8859
               SubPlan 1
                 ->  Index Scan using job_queues_pkey on job_queues  (cost=0.52..194.35 rows=1 width=0) (never executed)
                       Index Cond: (queue_name = jobs.queue_name)
                       Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
               SubPlan 2
                 ->  Seq Scan on job_queues job_queues_1  (cost=0.00..3551.99 rows=166 width=36) (actual time=0.013..5.584 rows=158 loops=1)
                       Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                       Rows Removed by Filter: 2
 Planning Time: 0.392 ms
 Execution Time: 12.759 ms
(16 rows)

Looks like the for update skip locked prevents PostgreSQL from freely reoptimizing it

[benjie]

Looks like the for update skip locked prevents PostgreSQL from freely reoptimizing it

Right; but does removing it impact parallel performance? Individual query performance may be improved, but this query and the resulting updates should be able to run with lots of concurrency without causing locks/deadlocks.

[ben-pr-p]

I believe that my second query in my original comment is good! I'm not suggesting we remove for update skip locked, just suggesting that the combination of it + exists is the reason the worse query plan is selected

[williamoliveira]

Current query LockRows on query plan
LockRows (cost=0.52..194.36 rows=1 width=10) (actual time=0.026..0.026 rows=0 loops=8857)

Proposed query LockRows on query plan
LockRows (cost=0.00..3553.65 rows=166 width=42) (actual time=0.068..6.381 rows=158 loops=1)

Wouldn't this mean its locking more rows than necessary?

[benjie]

Sounds like it’s locking 158 rows, yes. Definitely not ideal.

[williamoliveira]

Couldn't we do something like this just to lock job_queues? (removed for update skip locked from the sub query and put it in the select session to only run for the selected job row)

explain analyze
select jobs.queue_name, jobs.id, (select queue_name from job_queues where jobs.queue_name = job_queues.queue_name limit 1 for update skip locked)
from jobs
where (jobs.locked_at is null or jobs.locked_at < (now() - '4 hours'::interval))
and (
  jobs.queue_name is null
or
  jobs.queue_name in (
    select queue_name
    from job_queues
    where true
    and (job_queues.locked_at is null or job_queues.locked_at < (now() - '4 hours'::interval))
  )
)
and run_at <= now()
and attempts < max_attempts
and (null is null or jobs.task_identifier = any(null))
and (null is null or (flags ?| null) is not true)
order by priority asc, run_at asc, id asc
limit 1
for update
skip locked;

It has LockRows with rows=1 on the SubPlan 1

Limit  (cost=29.36..38.21 rows=1 width=61) (actual time=102.575..102.577 rows=1 loops=1)
  ->  LockRows  (cost=29.36..7431851.05 rows=840060 width=61) (actual time=102.575..102.576 rows=1 loops=1)
        ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=29.36..7423450.45 rows=840060 width=61) (actual time=102.573..102.574 rows=1 loops=1)
              Index Cond: (run_at <= now())
              Filter: (((queue_name IS NULL) OR (hashed SubPlan 2)) AND (attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))))
              Rows Removed by Filter: 499999
              SubPlan 1
                ->  Limit  (cost=0.28..8.30 rows=1 width=9) (actual time=0.033..0.034 rows=1 loops=1)
                      ->  LockRows  (cost=0.28..8.30 rows=1 width=9) (actual time=0.032..0.032 rows=1 loops=1)
                            ->  Index Scan using job_queues_pkey on job_queues  (cost=0.28..8.29 rows=1 width=9) (actual time=0.023..0.023 rows=1 loops=1)
                                  Index Cond: (queue_name = jobs.queue_name)
              SubPlan 2
                ->  Seq Scan on job_queues job_queues_1  (cost=0.00..26.50 rows=920 width=3) (actual time=0.015..0.089 rows=900 loops=1)
                      Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                      Rows Removed by Filter: 100
Planning Time: 0.104 ms
Execution Time: 102.641 ms

In comparison the current query plan on my test setup:

Limit  (cost=0.56..17.72 rows=1 width=29) (actual time=809.187..809.188 rows=1 loops=1)
  ->  LockRows  (cost=0.56..14414829.31 rows=840060 width=29) (actual time=809.186..809.186 rows=1 loops=1)
        ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=0.56..14406428.71 rows=840060 width=29) (actual time=809.184..809.184 rows=1 loops=1)
              Index Cond: (run_at <= now())
              Filter: ((attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))) AND ((queue_name IS NULL) OR (SubPlan 1)))
              Rows Removed by Filter: 499999
              SubPlan 1
                ->  LockRows  (cost=0.28..8.31 rows=1 width=10) (actual time=0.001..0.001 rows=0 loops=499738)
                      ->  Index Scan using job_queues_pkey on job_queues  (cost=0.28..8.30 rows=1 width=10) (actual time=0.001..0.001 rows=0 loops=499738)
                            Index Cond: (queue_name = jobs.queue_name)
                            Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                            Rows Removed by Filter: 1
Planning Time: 0.151 ms
Execution Time: 809.247 ms

And Ben's query as it is:

Limit  (cost=38.56..39.10 rows=1 width=29) (actual time=109.615..109.616 rows=1 loops=1)
  ->  LockRows  (cost=38.56..457262.10 rows=840060 width=29) (actual time=109.613..109.614 rows=1 loops=1)
        ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=38.56..448861.50 rows=840060 width=29) (actual time=109.606..109.607 rows=1 loops=1)
              Index Cond: (run_at <= now())
              Filter: (((queue_name IS NULL) OR (hashed SubPlan 1)) AND (attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))))
              Rows Removed by Filter: 499999
              SubPlan 1
                ->  LockRows  (cost=0.00..35.70 rows=920 width=9) (actual time=0.081..1.829 rows=900 loops=1)
                      ->  Seq Scan on job_queues  (cost=0.00..26.50 rows=920 width=9) (actual time=0.059..0.419 rows=900 loops=1)
                            Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                            Rows Removed by Filter: 100
Planning Time: 0.341 ms
Execution Time: 109.731 ms

[benjie]

Interesting idea, but I don't think this will work as you intend:

select jobs.queue_name, jobs.id, (
  select queue_name from job_queues where jobs.queue_name = job_queues.queue_name limit 1 for update skip locked
)

Specifically, if that job_queue is locked then it'll be skipped but that won't stop the job from being returned. We'd then have to have local logic to check that the two queue_names match up and if not re-attempt job fetching. This seems less than ideal.

Further, if you don't skip locked in the other subquery:

or
  jobs.queue_name in (
    select queue_name
    from job_queues
    where true
    and (job_queues.locked_at is null or job_queues.locked_at < (now() - '4 hours'::interval))
  )

then to resolve the subquery the query will need to hang until locks are released, which again is not ideal.

[williamoliveira]

Oh I see, did not think about that

I also tried with a simple inner join (would only work for named queues) and I get the same results as the original query. And with that I am officially out of ideas :/

explain analyze
select jobs.queue_name, jobs.id
from jobs
inner join job_queues on jobs.queue_name = job_queues.queue_name
                             and (job_queues.locked_at is null or job_queues.locked_at < (now() - '4 hours'::interval))
where (jobs.locked_at is null or jobs.locked_at < (now() - '4 hours'::interval))
and run_at <= now()
and attempts < max_attempts
and (null is null or jobs.task_identifier = any(null))
and (null is null or (flags ?| null) is not true)
order by priority asc, run_at asc, id asc
limit 1
for update
skip locked;

  ->  LockRows  (cost=0.84..963806.65 rows=1512109 width=35) (actual time=826.904..826.905 rows=1 loops=1)
        ->  Nested Loop  (cost=0.84..948685.56 rows=1512109 width=35) (actual time=826.884..826.884 rows=1 loops=1)
              ->  Index Scan using jobs_priority_run_at_id_locked_at_without_failures_idx on jobs  (cost=0.56..444623.20 rows=1680121 width=29) (actual time=0.014..118.007 rows=499900 loops=1)
                    Index Cond: (run_at <= now())
                    Filter: ((attempts < max_attempts) AND ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval))))
                    Rows Removed by Filter: 100
              ->  Index Scan using job_queues_pkey on job_queues  (cost=0.28..0.30 rows=1 width=9) (actual time=0.001..0.001 rows=0 loops=499900)
                    Index Cond: (queue_name = jobs.queue_name)
                    Filter: ((locked_at IS NULL) OR (locked_at < (now() - '04:00:00'::interval)))
                    Rows Removed by Filter: 1
Planning Time: 0.201 ms
Execution Time: 826.955 ms

[benjie]

I'm going to leave this open, but I don't think there's an obvious win right now.

[williamoliveira]

A hint of how bad this problem gets: yesterday on production we got 100% CPU on a 192 ACU AWS Aurora Serverless DB with just ~10k jobs

[ben-pr-p]

I think the solution is a separate pending jobs table which are blocked because of the queue.

Before insert, if a job has a queue_name that's locked, put it in the pending jobs table.

After delete, if the completed job has a queue_name, bring a new pending job from the pending jobs table into the jobs table.

This will have 0 meaningful overhead if you don't use queues because Postgres triggers WHEN (NEW.queue_name is not null) is performant, and will keep the table small and fast for the other jobs that need to be run during this time that are not part of the queue.

We could even implement this as an extension to graphile-worker, since it doesn't change how the user or SQL code interacts with the table at all

[benjie]

@williamoliveira I'd be interested to hear how you get on with a regular (rather than Aurora) database.

[benjie]

@ben-pr-p That's a really really interesting solution. 🤔

[benjie]

Enforcement: create a unique (queue_name) index on jobs table to forbid two jobs from same queue existing.

Situations:

Queue doesn't exist

Insert into queue, insert into jobs table.

If there's a concurrent transaction doing the same thing then one will have to roll back due to two writes to the same unique key in queues table 👍

Queue exists but is empty

Insert into the jobs table.

If there's a concurrent transaction doing the same thing then one will have to roll back due to two writes to the same unique key in the jobs table (due to the "enforcement" we added above).

Queue exists and is not empty

Write to separate table.

Job succeeds

If it was from a queue:

• Delete job
• If there's a job queued in that queue:
  • delete it from the queue and add it to the jobs table
• Otherwise:
  • delete the queue

Job fails

If it was from a queue:

• Insert into the queued jobs table in a failed state
• Delete from jobs table
• If there's a job queued in that queue:
  • delete it from the queue and add it to the jobs table
• Otherwise:
  • delete the queue

Job times out

• As current behaviour.

---

Note: checking emptiness of the queue is not free. In particular, we cannot skip locked. However, it is a unique single row lookup (thanks to the "enforcement" above) so it's not too expensive. Needs more thought as to what it means for contention.

We might be able to solve this by actually putting a flag on the queue itself to indicate emptiness so that we don't need to touch the jobs table straight away.

---

I'm not sure if deleting the queue may cause issues in some isolation levels - I need to think about that when I'm more awake. Maybe we don't really need to delete the queue... but I don't want the number of queues growing unbounded.

[ben-pr-p]

I don't love the behavior around the unique (queue_name) constraint, and I think that might even be considered a breaking change. Now, all users of queues, regardless of performance problems, would have to incorporate retry behavior when before they could get away with not having it.

I also think that since we'll be inserting into the jobs table both from add_job and complete_job, these conflicts could be pretty common.

To get around that, I think you could try to incorporate select for update on the job_queues table, or even pg_advisory_xact_lock (to handle the case where the queue doesn't exist), but this might cause a lot of contention.

In my mind, the best solution doesn't impose a unique constraint, and maybe doesn't even need a job_queues table.

If we add a teensy index:

    "jobs_queue_name_idx"  (queue_name) where (queue_name is not null)

Or maybe even:

    "jobs_queue_name_idx"  (queue_name) where (queue_name is not null) include (locked_at)

And then change get_job to:

  select jobs.queue_name, jobs.id into v_queue_name, v_job_id
    from :GRAPHILE_WORKER_SCHEMA.jobs
    where (jobs.locked_at is null or jobs.locked_at < (now - job_expiry))
    and (
      jobs.queue_name is null
    or
      not exists (
        select 1
        from :GRAPHILE_WORKER_SCHEMA.jobs other_job_in_queue
        where other_job_in_queue.queue_name = jobs.queue_name
        and (other_job_in_queue.locked_at is null or other_job_in_queue.locked_at < (now - job_expiry))
        for update
        -- skip locked - I think we don't want skip locked, so we wait if another transaction is updating this row
      )
    )
    and run_at <= now
    and attempts < max_attempts
    and (task_identifiers is null or task_identifier = any(task_identifiers))
    and (forbidden_flags is null or (flags ?| forbidden_flags) is not true)
    order by priority asc, run_at asc, id asc
    limit 1
    for update
    skip locked;

From there, the behavior should be pretty similar to what you spec'ed out, but with a tolerance for "a few" (as many as transactions run simultaneously) many jobs in the table.

The only difference I see is that before bring a job from the queued jobs table to the jobs table, you'd want to check if there was already a job in that queue in the table that should be worked on before bringing one in from the queued jobs table. That check itself should be fast, since the jobs_queue_name_idx is teensy.

My gut is that's the most performant, but I suppose testing would really reveal that!