Feature: variable worker concurrency - #1557
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughAdds optional buffer_concurrency to worker configuration and runtime behavior (semaphore pre-acquire, held buffer slots, fetch-time buffering, and runtime scaling). Adds App.run_worker_async_background() to start a Worker as an asyncio background task and return it immediately. ChangesBuffer Concurrency and Background Worker Support
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
procrastinate/worker.py (1)
39-58:⚠️ Potential issue | 🟠 Major | ⚡ Quick winReject invalid capacity combinations in
Worker.__init__
concurrency/buffer_concurrencyare not validated, sototal_capacity = concurrency + buffer_concurrencycan be0(or negative). In that case_run_loop()createsasyncio.Semaphore(self.total_capacity), and_fetch_and_process_jobs()blocks forever on the firstacquire()whentotal_capacity == 0;set_concurrency()can’t recover because it’s capped by the sametotal_capacity. Validateconcurrency >= 0,buffer_concurrency >= 0, andconcurrency + buffer_concurrency > 0.Possible fix
self.worker_name = name + if concurrency < 0 or buffer_concurrency < 0: + raise ValueError( + "concurrency and buffer_concurrency must be non-negative" + ) + total_capacity = concurrency + buffer_concurrency + if total_capacity <= 0: + raise ValueError( + "concurrency + buffer_concurrency must be greater than 0" + ) self.concurrency = concurrency self.buffer_concurrency = buffer_concurrency - self.total_capacity = concurrency + buffer_concurrency + self.total_capacity = total_capacity self._held_buffer_slots = 0🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@procrastinate/worker.py` around lines 39 - 58, Validate the concurrency inputs in Worker.__init__: ensure concurrency and buffer_concurrency are integers >= 0 and that total_capacity = concurrency + buffer_concurrency is > 0; if any check fails, raise a ValueError with a clear message. Update the constructor (Worker.__init__) to perform these checks before computing total_capacity so that downstream methods like _run_loop (which creates an asyncio.Semaphore(self.total_capacity)), _fetch_and_process_jobs (which acquires that semaphore), and set_concurrency won’t deadlock on a zero or negative capacity.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@procrastinate/worker.py`:
- Around line 39-58: Validate the concurrency inputs in Worker.__init__: ensure
concurrency and buffer_concurrency are integers >= 0 and that total_capacity =
concurrency + buffer_concurrency is > 0; if any check fails, raise a ValueError
with a clear message. Update the constructor (Worker.__init__) to perform these
checks before computing total_capacity so that downstream methods like _run_loop
(which creates an asyncio.Semaphore(self.total_capacity)),
_fetch_and_process_jobs (which acquires that semaphore), and set_concurrency
won’t deadlock on a zero or negative capacity.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 722d5793-51b6-4fbd-92e7-f7c3f74b2e00
📒 Files selected for processing (2)
procrastinate/app.pyprocrastinate/worker.py
…nges. Add run_worker_async_background() to return worker for control
0d84aec to
01f9296
Compare
for more information, see https://pre-commit.ci
Done (01f9296) |
|
Hi @thigger , What is your motivation behind this change? |
My tasks are fairly long-running (LLM-based) and if too many pile up in the LLM server queue then the kv cache ends up being evicted so things slow down a fair bit; I'm running dynamic concurrency with a watcher that keeps an eye on the LLM server so that the requests don't end up queueing. |
2 participants
For initial discussion; happy to sort documentation etc if you're happy with this route. I'm developing on Windows so a lot of tests fail on main at the moment; I'll have a look at spinning up WSL to run tests if this approach is OK. It seems to be working fine for me.
Works by allocating a larger semaphore (=concurrency plus buffer); you can then vary concurrency from zero to initial concurrency+buffer.
On buffer=0 should behave the same as the original worker.
Added app.run_worker_async_background() which returns the worker so its concurrency can be adjusted.
Related: ticket #1552
Successful PR Checklist:
PR label(s):
Summary by CodeRabbit