Batch subscriber #148
Open
Batch subscriber #148
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
fulmicoton
force-pushed
the
batch-subscriber
branch
11 times, most recently
from
f09f842
to
6fdf427
Compare
fulmicoton
commented
May 2, 2024
fulmicoton
force-pushed
the
batch-subscriber
branch
from
6fdf427
to
4335922
Compare
fulmicoton
force-pushed
the
batch-subscriber
branch
4 times, most recently
from
2505843
to
0046e5a
Compare
Upon reset_node and apply_update, we want to be able to react to a batch of changes rather than individual changes. In quickwit for instance, we want to be able to react to the reception of a batch of deleted shard and group this reaction into a single metastore call.
fulmicoton
force-pushed
the
batch-subscriber
branch
from
0046e5a
to
58cecb6
Compare
| let node_state = self.cluster_state.node_state_mut(chitchat_id); | ||
| ) -> anyhow::Result<()> { | ||
| // We validate the version is compatible with what is in key_values. | ||
| if !key_values.is_empty() { |
There was a problem hiding this comment.
this check is redundant (if key_values is empty, the for will not iterate)
| listener_idx: usize, | ||
| // Callbacks is a hashmap because as we delete listeners, we create "holes" in the | ||
| // callback_id -> callback mapping | ||
| callbacks: HashMap<usize, CallbackEntry>, |
There was a problem hiding this comment.
can you make this a HashMap<CallbackId, CallbackEntry> to make clear it goes hand in hand with listeners
| NodeState { | ||
| chitchat_id, | ||
| heartbeat: Heartbeat(0), | ||
| key_values: Default::default(), | ||
| max_version: 0u64, | ||
| listeners, | ||
| // listeners, |
There was a problem hiding this comment.
Suggested change
| // listeners, |
Comment on lines
+783
to
+796
| impl<'a> Deref for NodeStateMut<'a> { | ||
| type Target = NodeState; | ||
|
|
||
| fn deref(&self) -> &NodeState { | ||
| self.node_state_mut | ||
| } | ||
| } | ||
|
|
||
| impl<'a> DerefMut for NodeStateMut<'a> { | ||
| fn deref_mut(&mut self) -> &mut NodeState { | ||
| self.node_state_mut | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
I think deref makes sense, but deref_mut is bound to cause mutations which don't notify listeners
Comment on lines
+902
to
+904
| key.to_string(), | ||
| VersionedValue { | ||
| value: value.to_string(), |
There was a problem hiding this comment.
these to_string() look redundant
trinity-1686a
requested changes
May 16, 2024
Comment on lines
+1184
to
1189
| // 0 stale values | ||
| { | ||
| cluster_state.node_state_mut(&node5); | ||
| } | ||
|
|
||
| // 3 stale values |
There was a problem hiding this comment.
I don't understand where the stale value count come from. Before node5 and node6 had respectively 3 and 0 (instead of 0 and 3) stale values. I don't see an obvious reason this changed
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Upon reset_node and apply_update, we want to be
able to react to a batch of changes rather than individual changes.
In quickwit for instance, we want to be able to react to the reception of a batch of deleted shard and group this reaction into a single metastore call.