-
Notifications
You must be signed in to change notification settings - Fork 146
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feature: use a version to track metics change
- Add `Versioned<D>` to track changes of an `Arc<D>`.
In openraft, some frequently updated object such metrics are wrapped
in an `Arc`, and some modification is made in place: by storing an
`AtomicU64`.
Thus we can not tell whether an `Arc<D>` is changed by comparing them
with `==` any more.
In order to determine whether to broadcast a metrics instance, we need
an additional `version` to track the changes applied to the `Arc<D>`.
These are all included in the `Versioned<D>`.
- Add trait `Update` to define variant update operation to apply to
`Versioned<D>`.
`Update` has to implement two methods:
- `apply_in_place()` to apply a modification in place if possible
- and `apply_mut()` if it has to apply modification to a cloned
instance.
`Update` will increment the `Versioned.version` by 1 after each
update.
- Reimplement `LeaderMetrics` with `Versioned`.
- Change: type of `LeaderMetrics.replication` from HashMap to BTreeMap
for easing test.
commit-id:2432f7a0- Loading branch information
1 parent
518cc63
commit 966eb28
Showing
12 changed files
with
357 additions
and
63 deletions.
There are no files selected for viewing
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
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| use std::collections::BTreeMap; | ||
| use std::sync::atomic::AtomicU64; | ||
| use std::sync::atomic::Ordering; | ||
| use std::sync::Arc; | ||
|
|
||
| use serde::Deserialize; | ||
| use serde::Serialize; | ||
|
|
||
| use crate::versioned::Update; | ||
| use crate::versioned::UpdateError; | ||
| use crate::LogId; | ||
| use crate::MessageSummary; | ||
| use crate::NodeId; | ||
| use crate::RaftTypeConfig; | ||
| use crate::ReplicationMetrics; | ||
|
|
||
| /// The metrics about the leader. It is Some() only when this node is leader. | ||
| #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] | ||
| pub struct LeaderMetrics<C: RaftTypeConfig> { | ||
| /// Replication metrics of all known replication target: voters and learners | ||
| pub replication: BTreeMap<C::NodeId, ReplicationMetrics<C>>, | ||
| } | ||
|
|
||
| impl<C: RaftTypeConfig> MessageSummary for LeaderMetrics<C> { | ||
| fn summary(&self) -> String { | ||
| let mut res = vec!["LeaderMetrics{".to_string()]; | ||
| for (i, (k, v)) in self.replication.iter().enumerate() { | ||
| if i > 0 { | ||
| res.push(", ".to_string()); | ||
| } | ||
| res.push(format!("{}:{}", k, v.summary())); | ||
| } | ||
|
|
||
| res.push("}".to_string()); | ||
| res.join("") | ||
| } | ||
| } | ||
|
|
||
| /// Update one replication metrics in `LeaderMetrics.replication`. | ||
| pub struct UpdateMatchedLogId<NID: NodeId> { | ||
| pub target: NID, | ||
| pub matched: LogId<NID>, | ||
| } | ||
|
|
||
| impl<C: RaftTypeConfig> Update<LeaderMetrics<C>> for UpdateMatchedLogId<C::NodeId> { | ||
| /// If there is already a record for the target node. Just modify the atomic u64. | ||
| fn apply_in_place(&self, to: &Arc<LeaderMetrics<C>>) -> Result<(), UpdateError> { | ||
| let target_metrics = to.replication.get(&self.target).ok_or(UpdateError::CanNotUpdateInPlace)?; | ||
|
|
||
| if target_metrics.matched_leader_id == self.matched.leader_id { | ||
| target_metrics.matched_index.store(self.matched.index, Ordering::Relaxed); | ||
| return Ok(()); | ||
| } | ||
|
|
||
| Err(UpdateError::CanNotUpdateInPlace) | ||
| } | ||
|
|
||
| /// To insert a new record always work. | ||
| fn apply_mut(&self, to: &mut LeaderMetrics<C>) { | ||
| to.replication.insert(self.target, ReplicationMetrics { | ||
| matched_leader_id: self.matched.leader_id, | ||
| matched_index: AtomicU64::new(self.matched.index), | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| /// Remove one replication metrics in `LeaderMetrics.replication`. | ||
| pub struct RemoveTarget<NID: NodeId> { | ||
| pub target: NID, | ||
| } | ||
|
|
||
| impl<C: RaftTypeConfig> Update<LeaderMetrics<C>> for RemoveTarget<C::NodeId> { | ||
| /// Removing can not be done in place | ||
| fn apply_in_place(&self, _to: &Arc<LeaderMetrics<C>>) -> Result<(), UpdateError> { | ||
| Err(UpdateError::CanNotUpdateInPlace) | ||
| } | ||
|
|
||
| fn apply_mut(&self, to: &mut LeaderMetrics<C>) { | ||
| to.replication.remove(&self.target); | ||
| } | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| use crate::leader_metrics::LeaderMetrics; | ||
| use crate::leader_metrics::UpdateMatchedLogId; | ||
| use crate::testing::DummyConfig; | ||
| use crate::versioned::Updatable; | ||
| use crate::versioned::Versioned; | ||
| use crate::LeaderId; | ||
| use crate::LogId; | ||
| use crate::MessageSummary; | ||
|
|
||
| #[test] | ||
| fn test_versioned() -> anyhow::Result<()> { | ||
| let mut a = Versioned::new(LeaderMetrics::<DummyConfig> { | ||
| replication: Default::default(), | ||
| }); | ||
|
|
||
| assert_eq!("{ver:0, LeaderMetrics{}}", a.summary()); | ||
|
|
||
| // In place update | ||
|
|
||
| a.update(UpdateMatchedLogId { | ||
| target: 1, | ||
| matched: LogId::new(LeaderId::new(1, 2), 3), | ||
| }); | ||
|
|
||
| assert_eq!("{ver:1, LeaderMetrics{1:1-2-3}}", a.summary()); | ||
|
|
||
| let mut b1 = a.clone(); | ||
|
|
||
| // Two instances reference the same data. | ||
| // In place update applies to both instance. | ||
|
|
||
| b1.update(UpdateMatchedLogId { | ||
| target: 1, | ||
| matched: LogId::new(LeaderId::new(1, 2), 5), | ||
| }); | ||
| assert_eq!("{ver:1, LeaderMetrics{1:1-2-5}}", a.summary()); | ||
| assert_eq!("{ver:2, LeaderMetrics{1:1-2-5}}", b1.summary()); | ||
|
|
||
| // In place update is not possible. | ||
| // Fall back to cloned update | ||
|
|
||
| b1.update(UpdateMatchedLogId { | ||
| target: 2, | ||
| matched: LogId::new(LeaderId::new(1, 2), 5), | ||
| }); | ||
| assert_eq!("{ver:1, LeaderMetrics{1:1-2-5}}", a.summary()); | ||
| assert_eq!("{ver:3, LeaderMetrics{1:1-2-5, 2:1-2-5}}", b1.summary()); | ||
|
|
||
| // a and b1 have the same content but not equal, because they reference different data. | ||
|
|
||
| a.update(UpdateMatchedLogId { | ||
| target: 1, | ||
| matched: LogId::new(LeaderId::new(1, 2), 5), | ||
| }); | ||
| a.update(UpdateMatchedLogId { | ||
| target: 2, | ||
| matched: LogId::new(LeaderId::new(1, 2), 5), | ||
| }); | ||
| assert_eq!("{ver:3, LeaderMetrics{1:1-2-5, 2:1-2-5}}", a.summary()); | ||
| assert_eq!("{ver:3, LeaderMetrics{1:1-2-5, 2:1-2-5}}", b1.summary()); | ||
| assert_ne!(a, b1); | ||
|
|
||
| // b2 reference the same data as b1. | ||
|
|
||
| let mut b2 = b1.clone(); | ||
| b2.update(UpdateMatchedLogId { | ||
| target: 2, | ||
| matched: LogId::new(LeaderId::new(1, 2), 9), | ||
| }); | ||
| assert_eq!("{ver:3, LeaderMetrics{1:1-2-5, 2:1-2-9}}", b1.summary()); | ||
| assert_eq!("{ver:4, LeaderMetrics{1:1-2-5, 2:1-2-9}}", b2.summary()); | ||
| assert_ne!(b1, b2); | ||
|
|
||
| // Two Versioned are equal only when they reference the same data and have the same version. | ||
|
|
||
| b1.update(UpdateMatchedLogId { | ||
| target: 2, | ||
| matched: LogId::new(LeaderId::new(1, 2), 9), | ||
| }); | ||
| assert_eq!("{ver:4, LeaderMetrics{1:1-2-5, 2:1-2-9}}", b1.summary()); | ||
| assert_eq!("{ver:4, LeaderMetrics{1:1-2-5, 2:1-2-9}}", b2.summary()); | ||
| assert_eq!(b1, b2); | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| #[test] | ||
| fn test_versioned_methods() -> anyhow::Result<()> { | ||
| let mut a = Versioned::new(LeaderMetrics::<DummyConfig> { | ||
| replication: Default::default(), | ||
| }); | ||
|
|
||
| a.update(UpdateMatchedLogId { | ||
| target: 1, | ||
| matched: LogId::new(LeaderId::new(1, 2), 3), | ||
| }); | ||
|
|
||
| assert_eq!("{ver:1, LeaderMetrics{1:1-2-3}}", a.summary()); | ||
|
|
||
| assert_eq!(1, a.version()); | ||
| assert_eq!("LeaderMetrics{1:1-2-3}", a.data().summary()); | ||
|
|
||
| Ok(()) | ||
| } |
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
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
Oops, something went wrong.