incredbly slow insert speed for unique jobs. what is the best way forward?

[elee1766]

We noticed incredibly slow performance (sub 100 rps on a 4vcpu/6gb postgres instance. can get exact number if needed) when attempting to insert 1000s of jobs with unique job constraints at the same time, bottlenecked by what i believe is the advisory lock, but maybe i am wrong. What we do know is that when inserting non-unique jobs, either with Insert or InsertMany, we don't have throughput issues.

the problem: we have a task that updates entities in the background when users request such data and the data is stale enough. We would like this this job to be unique by the entity id, but it is often that thousands of these are being inserted at a time. Along with this, a job runs which schedules updates for all ~50,000 entities every single entity every 15 minutes.

with unique insert on, insertion is too slow and we dont schedule all the tasks as fast as we wish. but with unique off, we run into the problem where if network conditions cause us unable to complete jobs in time, we end up scheduling an infinite amount of jobs. however, we don't want to stop scheduling jobs as there may be new entity ids that we need to add jobs for even if all entity updates in the previous batch are not yet complete.

for now we are chunking the batch job into large groups with timeout but that loses us both observability, reduces our parallelism, and heavily complicates the retry logic, since you need to reschedule a new job with your errors. (we simply are not retrying for now)

so far our current ideas are

1. use InsertTx another table that keeps track of the task, and have the task write/delete from table on schedule/ when it succeeds
2. use InsertTx with the jobs table to check the unique constraint ourselves but without the advisory locks, since our task doesn't need as strong guarantees as what the advisory lock provides

but neither of these things felt like very good ideas

[brandur]

@elee1766 This one's going to be tough to fix if you keep going down that path I'm afraid. You are fundamentally executing many, expensive, conflicting operations, and unsurprisingly, that's going to be expensive and without perfectly optimal performance.

I suspect as well that it's more the "conflicting" part that you're having the most trouble with here. We actually did a blog post recently on how River's uniqueness is implemented [1] and it's not all that complex:

1. Open transaction/subtransaction.
2. Take advisory lock.
3. Check if an existing row is there.
4. Perform insert or don't.

None of those pieces take very long, but if you have lots of conflicting unique inserts, then they're all waiting on each other's lock.

The only possible remediation I can think of is if we provided an "optimistic unique check", wherein if it noticed that someone else already had the advisory lock, it'd fall through immediately instead of waiting to acquire it. The downside of this though is that if the other lock taker didn't end up inserting successfully, you could lose a whole row.

In your case, an alternative: drop the uniqueness checks and then implement your job such that it checks on start up the last time its data was updated. If the update was very recent, it falls through with a no op. So you'd still be inserting lots of jobs, but most of them wouldn't be doing any work, and you wouldn't suffer the unique performance penalty.

---

[1] https://riverqueue.com/blog/uniqueness-with-advisory-locks

[brandur]

by conflicting, do you mean that there are actually conflicts in the insert and it's skipping the insert? Or that they are happening at the same time?

Yeah, that is what I meant. I thought you might have gotten yourself in a position where there was a lot of contention on particular locks, so they'd end up waiting on each other and becoming slow.

Thanks for including a benchmark. Taking a glance at that, I suspect there might be an indexing problem that could emerge depending on the unique criteria. The jobs table is indexed by kind, is indexed by state, separately, and is indexed by args, also separately. The predicate in your case will need to query on all of kind, state, and args and Postgres will end up picking one of those indexes, but will have to scan what could be a large data set within it. (In this case, Id is always incremented, so the unique insert will always scan the existing bulk, realize there's no existing equivalent, and do a new insert.)

A real world load might actually perform better than what you have because IRL as you insert jobs the River client will start working them, thereby moving state from available to running and hopefully through to completed, and new inserts could use the state index and not have to scan as many rows since fewer are in (available, running) at any given time.

[elee1766]

A real world load might actually perform better than what you have because IRL as you insert jobs the River client will start working them, thereby moving state from available to running and hopefully through to completed, and new inserts could use the state index and not have to scan as many rows since fewer are in (available, running) at any given time.

during testing of a different app, i remember job insert speed going to a complete crawl after running over night. i had ~500,000 completed jobs, 0 available jobs, and like 10 jobs working at a time. I found out the issue was that jobs were taking forever to schedule, and I was able to get back to scheduling ~100 per second by deleting all the completed jobs, and my short term fix is to reduce the retention time of completed jobs to 1 hour. So unique insert seems to be slow regardless of the state of the jobs. this is anecdotal though, i can make a benchmark later to see.

I was trying to complete a set of ~6,000,000 unique tasks overnight.

[elee1766]

@brandur yeah completing the jobs doesn't seem to change much, at least here.

https://gist.github.com/elee1766/11b2129d07c48e0274ba6a11a99819ec

[brandur]

Okay I got dug into this one a bit, and as is somewhat predictable based on reading the benchmark code, all your time is going to lookups on the unique check. You can see it here it taking about two orders of magnitude longer than the lock or insert:

lock 47.041µs, get: 2.312791ms, insert: 80.5µs

And that's closer to the beginning of the best. Once you let it accumulate rows, it's up to three orders of magnitude longer:

lock 53.709µs, get: 20.829542ms, insert: 235.75µs

But again, this shouldn't be surprising. You've set up your benchmark such that the unique look up will never find a match, and continue to produce an ever larger bulk of existing rows that need to be iterated looking for one:

uniqueOpts := river.UniqueOpts{
    ByArgs: true,

...

_, err := riverClient.Insert(ctx, &job{Id: int(cur.Add(1))}, &river.InsertOpts{

So think about this is going to work: for every unique check, the client has to iterate through ~every row in the set looking for a match of args = @args that will never return true because you modify the args every time so the uniqueness never deduplicates. As the size the data set gets larger, this gets slower.

Postgres starts out using the index on kind, but that won't do a good job because of course you only insert one kind of row ever, so there's a massive number of rows on kind kind = 'job':

=# explain analyze SELECT *
FROM river_job
WHERE kind = 'job'
    AND CASE WHEN true THEN args = '{"Id": 2342352354}' ELSE true END
    AND CASE WHEN false THEN tstzrange(now()::timestamptz, now()::timestamptz, '[)') @> created_at ELSE true END
    AND CASE WHEN false THEN queue = 'default' ELSE true END
    AND CASE WHEN true THEN state::text = any('{"available", "running", "retryable", "scheduled"}'::text[]) ELSE true END;
                                                           QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------
 Bitmap Heap Scan on river_job  (cost=668.44..757.49 rows=1 width=199) (actual time=0.122..0.123 rows=0 loops=1)
   Recheck Cond: (kind = 'job'::text)
   Filter: ((args = '{"Id": 2342352354}'::jsonb) AND ((state)::text = ANY ('{available,running,retryable,scheduled}'::text[])))
   Rows Removed by Filter: 24
   Heap Blocks: exact=8
   ->  Bitmap Index Scan on river_job_kind  (cost=0.00..668.44 rows=24 width=0) (actual time=0.047..0.048 rows=136 loops=1)
         Index Cond: (kind = 'job'::text)
 Planning Time: 0.240 ms
 Execution Time: 0.188 ms

As the bulk of rows grows, Postgres gives up on using any indexes and just does a sequential scan, which given this degenerate scenario, will be no slower than using an index:

=# explain analyze SELECT *
FROM river_job
WHERE kind = 'job'
    AND CASE WHEN true THEN args = '{"Id": 2342352354}' ELSE true END
    AND CASE WHEN false THEN tstzrange(now()::timestamptz, now()::timestamptz, '[)') @> created_at ELSE true END
    AND CASE WHEN false THEN queue = 'default' ELSE true END
    AND CASE WHEN true THEN state::text = any('{"available", "running", "retryable", "scheduled"}'::text[]) ELSE true END;
                                                                       QUERY PLAN                                                            
---------------------------------------------------------------------------------------------------------------------------------------------------------
 Seq Scan on river_job  (cost=0.00..5233.00 rows=1 width=199) (actual time=53.069..53.069 rows=0 loops=1)
   Filter: ((kind = 'job'::text) AND (args = '{"Id": 2342352354}'::jsonb) AND ((state)::text = ANY ('{available,running,retryable,scheduled}'::text[])))
   Rows Removed by Filter: 100000
 Planning Time: 1.408 ms
 Execution Time: 53.113 ms
(5 rows)

Time: 57.943 ms

So in short, a few points:

• Unique job insertion is never going to be anywhere near as fast as non-unique insertion. Fundamentally, more work needs to be carried out to perform an insert.
• If you want to build a benchmark that shows unique insertion under degenerate conditions to make it look even slower than it is, it's totally possible to do so.
• I'm reasonably confident that although uniqueness may not be lightning fast under all conditions, under real world conditions where it's being used reasonably, it's going to be reasonably performant.

Here's two ways of changing your benchmark from slow to lightning fast:

• Don't use unique insertion. Not to beat this drum too hard, but if your unique properties are never going to match an existing job, you don't need unique insertion.
• Change unique properties so that uniqueness matches once in a while, but not so much that you have crazy lock contention.

e.g. Bench goes up to 6k/jobs a sec when using % 100 on ID to keep the set to 100, and adding complete to the set of unique states:

	uniqueOpts := river.UniqueOpts{
		ByArgs: true,
		ByState: []rivertype.JobState{
			rivertype.JobStateAvailable,
			rivertype.JobStateCompleted,
			rivertype.JobStateRunning,
			rivertype.JobStateRetryable,
			rivertype.JobStateScheduled,
		},
	}
	//uniqueOpts = river.UniqueOpts{}
	for i := 0; i < 100000; i++ {
		egg.Go(func() error {
			defer counter.Incr(1)
			_, err := riverClient.Insert(ctx, &job{Id: int(cur.Add(1)) % 100}, &river.InsertOpts{
				MaxAttempts: 0,
				UniqueOpts:  uniqueOpts,
			})
			return err
		})
	}

[elee1766]

hmm yeah. it actually uses the river_job_prioritized_fetching_index

I could schedule the bulk jobs (10,000s at a time) without unique. the issue would be that time period, unique entries from user activity would be slow to queue and create unnecessary lock contention bc it would have to scan the whole bg working queue.

If i put them in a different queue, maybe it would work? but i dont see an index on (kind, queue) (maybe i can just add one).

then i could schedule the on-demand requests as unique to a different queue and it would be fast enough for that, while not interfering with the non-unique background tasks.

i could also get around it by making it two different tasks i believe then, but the thing is that it's the same exact task, so it feels silly to make two different kinds for the exact same thing.

i'll play around and see where this get us.