From RQ Workers to an Asyncio Engine: Polling 5,000 Jobs on Three Processes

The previous post in this series, Migrating from ThreadPool to AsyncIO in Python, covered the general recipe. This one is a case study: what happened when we applied it to a real service, where the "threads" were not a ThreadPoolExecutor but fifty RQ worker processes.

The workload looks like a perfect asyncio fit. Callers register jobs. Every second, each job asks a log backend (Graylog) "any new messages for me?", filters the answer, and pushes matches to the caller over a WebSocket. Thousands of jobs, almost all of them waiting on the network, almost all of them finding nothing.

What You'll Learn

✅ Why "one worker per N jobs" stops scaling, with numbers
✅ How to share work between processes with leases instead of a coordinator
✅ Why one coroutine per job was only step one, and batching was the real win
✅ Keeping sync code in threads, with a clear boundary around it
✅ A MySQL deadlock you only get after going concurrent
✅ How to roll out with a switch and a way back

1. The Starting Point

The old design: a worker manager assigns jobs to RQ workers, and each worker loops over its ~10 jobs forever, calling the log backend once per job per poll.

At 500 jobs on 50 workers (10 jobs each), measured against a stub backend with 50 ms latency and a 1 s poll interval:

Two things stand out. The jobs aren't even polled on time, and the CPU is high for a process that mostly waits. About 9 ms of CPU per poll, spent on connection handling, JSON, ORM work, and OS context switches between processes. Adding workers makes it worse, because every new worker adds its own connections and its own overhead.

2. Step One: Coroutines and Leases

The first version of the engine was deliberately boring. One process holds up to N jobs and runs one coroutine per job, each sleeping until its next poll.

The interesting question: how do several engine processes divide the jobs, without a central coordinator that becomes a new single point of failure?

Leases on the database rows. Each job row gets two columns, engine_owner and lease_until. An engine claims jobs, renews the lease while it works, and releases on shutdown. If an engine dies, its leases simply expire and another engine takes over.

-- claim: several engines can run this at once without blocking each other
SELECT id FROM jobs
WHERE status = 1 AND (lease_until IS NULL OR lease_until < NOW())
LIMIT 100
FOR UPDATE SKIP LOCKED;

FOR UPDATE SKIP LOCKED is the quiet hero. Two engines claiming at the same moment never wait on each other and never claim the same row. There's no coordinator: to scale, start another process.

The blocking parts (ORM, the existing polling code) stayed unchanged and ran in a thread pool via asyncio.to_thread. The engine is async at the edges and reuses the proven logic in the middle. That decision kept behavior identical, and the same black-box tests passed on both the old and new job tier.

3. Step Two: The Real Win Was Batching

Coroutines alone only fix the process overhead: the first version still asked the backend one question per job, so the number of backend requests was unchanged.

The bigger shift came from a different observation: a coroutine per job is the wrong unit. Almost every poll finds nothing, so why do 500 separate searches?

Polls that start in the same window now share one search:

message_source_scope:a* OR message_source_scope:b* OR ...

over the union of their time ranges, and each job then gets back exactly what its own query would have returned (its prefix, its time range, the backend's order).

Then we went one level further and replaced "a coroutine per job" with one tick per process per poll interval:

1. Load all of the process's jobs (one query)
2. Run the batched backend searches in parallel
3. Apply the results: one Redis pipeline, one database commit
4. Jobs that found nothing, which is most of them, advance with one UPDATE per 500 rows. Only jobs that actually received messages run the full job code.

This is the pattern worth remembering. asyncio gives you cheap concurrency, but the biggest gains came from using that cheapness to do less work per unit, not to do the same work in more places.

CPU per poll fell from roughly 9 ms to roughly 0.1 ms. Backend requests now grow with the number of engines, not the number of jobs, and ten times as many jobs run in a fraction of the CPU.

4. Gotchas We Hit

The HTTP pool was silently too small

requests defaults to 10 pooled connections per host. With parallel searches in threads, the connection pool was constantly opening and dropping connections, a pattern that looks like network flakiness. Size the pool to the number of threads (we use 64).

Fair share, or one engine hogs everything

If three engines start at once, the first one up can claim every job. Each engine limits its claims to a fair share: active jobs divided by live engines, capped by a per-process maximum. Engines publish a heartbeat to Redis, which also feeds the /status page, so you can see each engine's job count and how long its ticks take.

Make "falling behind" visible

A tick that takes longer than the poll interval means the engine is overloaded. It logs a warning with the load / fetch / apply timings, plus a hint: start another engine. Scaling out is the fix, and the logs say so.

5. The Deadlock You Only Get After Going Concurrent

On a 2,000-job load test, a caller's DELETE failed with a MySQL deadlock, and the job kept running.

Two writers, two lock orders:

• The engine updates many rows at once (lease renewal, moving quiet jobs on). MySQL walked the secondary index (engine_owner, status) first, then locked the rows.
• The REST API updates one job: it locks the row first, then the indexes affected by the status change.

Index-then-row versus row-then-index: a classic lock-order inversion. It never happened with a single writer type.

The fix had two parts:

# 1. The engine reads ids without locks, then updates by primary key only,
#    in id order, which is the same order the REST API locks in.
UPDATE jobs FORCE INDEX (PRIMARY) SET lease_until = ... WHERE id IN (...) ORDER BY id;
 
# 2. The REST handlers retry on 1213 (deadlock) / 1205 (lock wait timeout).

The second is a safety net, not a fix. Whenever two kinds of concurrent writers touch the same table, retry on deadlock, because the transaction is rolled back and you can safely run it again. The unit test reproduces the race: before the change, 2 of 3 runs deadlocked. After, 0 of 8.

6. Rolling Out: A Switch and a Way Back

We didn't replace the old tier. A single setting, JOB_ENGINE=rq|async, picks which one runs, and the deploy script starts only that tier and fails verification if the other one is still running. The async engine ran on the development environment first, the others stayed on RQ unchanged.

Two details made this safe:

• Rollback is automatic. If the compose file on the server has no engine service (an older release), the deploy falls back to RQ.
• Tests run against both tiers. The same 64 black-box tests run once per job tier in CI, so behavior can't drift between them.

7. An Honest Caveat

The engine made polling cheaper and on time. It did not make messages arrive sooner. Each pass queries the backend only up to 10 seconds in the past, so messages reach the caller at least ~10 s late whatever the poll interval. We measured it instead of promising otherwise.

And the biggest single improvement in this whole effort wasn't asyncio at all. Before any of this, MySQL was at 93% CPU because the jobs table had no index for the hot query. Fix the obvious problem first, then decide whether you need a new architecture.

Summary

✅ Don't just convert workers to coroutines. Ask what work you can share (batching) or skip (quiet jobs).
✅ Use leases plus FOR UPDATE SKIP LOCKED for coordinator-free work sharing across processes.
✅ Keep the proven blocking code and run it in threads; make only the edges async.
✅ Size your connection pools to your concurrency.
✅ Concurrent writers need a consistent lock order, plus a retry on deadlock.
✅ Roll out behind a switch, test both implementations with one suite, keep a way back.
✅ Measure the thing you actually care about, and be honest about what didn't improve.

Next in the series: the framework side of the same effort, Migrating from Flask to FastAPI.

Related Posts

• Migrating from Flask to FastAPI
• Upgrading Django 5.2 to 6.0
• Upgrading Docker and Compose: The Forgotten Cron Job