impl(pubsub): message ordering in Publisher #4124
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #4124 +/- ##
==========================================
- Coverage 95.25% 95.24% -0.02%
==========================================
Files 178 178
Lines 6811 6807 -4
==========================================
- Hits 6488 6483 -5
- Misses 323 324 +1 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
suzmue
left a comment
suzmue
left a comment
There was a problem hiding this comment.
As far as ordering, this looks pretty good. However, I think this removed support for non-ordered messages where we can have multiple inflight batches.
Please let me know if I'm mistaken.
| for (_, outstanding) in pending_batches.iter_mut() { | ||
| outstanding.flush(&mut inflight); | ||
| for (_, batch_worker) in batch_workers.iter_mut() { | ||
| let (tx, _) = oneshot::channel(); |
There was a problem hiding this comment.
I'm confused, why do we need this channel?
There was a problem hiding this comment.
Oh I see its because you don't want to wait on the result.
if this doesn't use a lot of resources seems fine.
| let (tx, rx) = oneshot::channel(); | ||
| batch_worker | ||
| .send(ToBatchWorker::Flush(tx)) | ||
| .expect("Batch worker should not close the channel"); |
There was a problem hiding this comment.
nit: Consider using a helper message or helper function to avoid the duplication of this error message all over.
| } | ||
| } | ||
|
|
||
| /// A background worker that continuously handle Publisher commands for a specific ordering key. |
There was a problem hiding this comment.
| /// A background worker that continuously handle Publisher commands for a specific ordering key. | |
| /// A background worker that continuously handles Publisher commands for a specific ordering key. |
| pending_batch: Batch, | ||
| // TODO(#4012): Track pending messages as within key message ordering is | ||
| // not currently respected during a failure. | ||
| pending_msges: VecDeque<BundledMessage>, |
There was a problem hiding this comment.
| pending_msges: VecDeque<BundledMessage>, | |
| pending_msgs: VecDeque<BundledMessage>, |
| pending_batch: Batch, | ||
| // TODO(#4012): Track pending messages as within key message ordering is | ||
| // not currently respected during a failure. | ||
| pending_msges: VecDeque<BundledMessage>, |
| /// `Worker` drops the Sender. | ||
| pub(crate) async fn run(mut self) { | ||
| // While it is possible to use Some(JoinHandle) here as there is at max | ||
| // a single inflight task at any given time, the use of JoinSet |
There was a problem hiding this comment.
This isn't true when ordering key is "".
| loop { | ||
| if self.pending_batch.is_empty() && self.pending_msges.is_empty() { | ||
| break; | ||
| } |
There was a problem hiding this comment.
This can be a while loop.
| loop { | |
| if self.pending_batch.is_empty() && self.pending_msges.is_empty() { | |
| break; | |
| } | |
| while !self.pending_batch.is_empty() || !self.pending_msges.is_empty() { |
| || self.pending_batch.size() >= self.batching_options.byte_threshold | ||
| } | ||
|
|
||
| // Move pending publishes to the pending batch respecting batch thresholds. |
There was a problem hiding this comment.
| // Move pending publishes to the pending batch respecting batch thresholds. | |
| // Move pending messages to the pending batch respecting batch thresholds. |
| for (_, outstanding) in pending_batches.iter_mut() { | ||
| outstanding.flush(&mut inflight); | ||
| for (_, batch_worker) in batch_workers.iter_mut() { | ||
| let (tx, _) = oneshot::channel(); |
There was a problem hiding this comment.
Oh I see its because you don't want to wait on the result.
if this doesn't use a lot of resources seems fine.
| break; | ||
| } | ||
| self.move_to_batch(); | ||
| self.pending_batch.flush(self.client.clone(), self.topic.clone(), &mut inflight); |
There was a problem hiding this comment.
using flush for the timer seems slightly inefficient since if we have a max batch size of 1000 and we have a batch out and 1001 messages in the queue, then we we will send that single message as a batch of 1 even if more come in. I guess that is a possibility with having the timeout anyway, so we maybe its not a real problem and we don't need it to be changed necessarily.
This also makes we wonder actually if we have an outstanding batch, would it be better to just leave all messages in the channel? to avoid having to copy things around as much? I'm just curious about this. Something to maybe try out later.
2 participants
Support messaging ordering in Publisher by creating a new BatchWorker which handles that batching and message ordering logic.
When the Worker receives an Publish operation, it demultiplex it into the corresponding BatchWorker which ensures that there is at most one inflight batch at any given time.
When the Worker receives a Flush operation, it flushes all BatchWorker pending message and waits until the operation is complete.
As of now, the set of BatchWorkers grows indefinitely. This is be address in the next PR.