feat(uptime): Make consumer able to run in parallel #81409
Conversation
github-actions
bot
added
the
Scope: Backend
Codecov ReportAttention: Patch coverage is ✅ All tests successful. No failed tests found.
Additional details and impacted files@@ Coverage Diff @@
## master #81409 +/- ##
==========================================
+ Coverage 80.36% 80.42% +0.05%
==========================================
Files 7273 7284 +11
Lines 321302 321490 +188
Branches 20948 20948
==========================================
+ Hits 258227 258550 +323
+ Misses 62673 62538 -135
Partials 402 402 |
This adds thread pool parallelization to the uptime consumer. We should potentially consider process parallelizatoin as well, so that we can take advantage of all cores. This is a rough draft of this and isn't tested - based off the issue occurrence consumer and monitor consumer.
wedamija
force-pushed
the
danf/uptime-parallel-consumer
branch
from
1fe893b to
7e170d1
Compare
| return self.create_serial_worker(commit) | ||
|
|
||
| def create_serial_worker(self, commit: Commit) -> ProcessingStrategy[KafkaPayload]: | ||
| return RunTask( |
There was a problem hiding this comment.
you're already using RunTask. wouldn't it be easier to conditionally swap it for RunTaskInThreads instead of maintaining your own threadpool?
There was a problem hiding this comment.
We have special logic here where we want to process a batch of results before processing the next batch. This allows us to make sure that related data is processed serially.
There was a problem hiding this comment.
This is because we need to partition the batches in such a way that uptime results for the same monitor are processed in order.
Using RunTaskInThreads doesn't offer any kind of way to specify ordering
There was a problem hiding this comment.
We're actually doing this in 3 different places now (crons, uptime, issue occrences). I really would love to turn this into some kind of reusable strategy
There was a problem hiding this comment.
Got it. I keep forgetting about this, feels like I asked this before 😅
I think the part where you need to wait for a set of messages before processing the next one can be generalized independently of the inner strategy even.
This adds thread pool parallelization to the uptime consumer. We should potentially consider process parallelisation as well, so that we can take advantage of all cores. Alternatively, it could be worth experimenting with disabling the GIL so we can avoid the complexity of figuring out processes.
This adds thread pool parallelization to the uptime consumer. We should potentially consider process parallelisation as well, so that we can take advantage of all cores. Alternatively, it could be worth experimenting with disabling the GIL so we can avoid the complexity of figuring out processes.