From 0c53e7b85a3a81f445c4c64ffbef860d0f315445 Mon Sep 17 00:00:00 2001 From: Zhijie Zhan Date: Thu, 17 Sep 2020 14:27:44 +0800 Subject: [PATCH] quorum: port single group joint commit (#394) Signed-off-by: accelsao --- Cargo.toml | 2 + src/quorum.rs | 16 +- src/quorum/datadriven_test.rs | 141 ++++++++ src/quorum/joint.rs | 15 + src/quorum/majority.rs | 85 +++++ src/quorum/testdata/joint_commit.txt | 481 +++++++++++++++++++++++++++ 6 files changed, 735 insertions(+), 5 deletions(-) create mode 100644 src/quorum/datadriven_test.rs create mode 100644 src/quorum/testdata/joint_commit.txt diff --git a/Cargo.toml b/Cargo.toml index d58edaca6..4d36bdf65 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,8 @@ slog-async = "2.3.0" slog-envlogger = "2.1.0" slog-stdlog = "4" slog-term = "2.4.0" +anyhow = "1.0.32" +datadriven = { path = "datadriven", version = "0.1.0" } [[bench]] name = "benches" diff --git a/src/quorum.rs b/src/quorum.rs index 545f9d80f..f23e5d0a4 100644 --- a/src/quorum.rs +++ b/src/quorum.rs @@ -1,5 +1,6 @@ // Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0. - +#[cfg(test)] +pub mod datadriven_test; pub mod joint; pub mod majority; @@ -28,10 +29,15 @@ pub struct Index { impl Display for Index { #[inline] fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - if self.index != u64::MAX { - write!(f, "[{}]{}", self.group_id, self.index) - } else { - write!(f, "[{}]∞", self.group_id) + match self.group_id { + 0 => match self.index { + u64::MAX => write!(f, "∞"), + index => write!(f, "{}", index), + }, + group_id => match self.index { + u64::MAX => write!(f, "[{}]∞", group_id), + index => write!(f, "[{}]{}", group_id, index), + }, } } } diff --git a/src/quorum/datadriven_test.rs b/src/quorum/datadriven_test.rs new file mode 100644 index 000000000..f37130e34 --- /dev/null +++ b/src/quorum/datadriven_test.rs @@ -0,0 +1,141 @@ +use crate::quorum::{AckIndexer, Index}; +use crate::{default_logger, HashSet, JointConfig, MajorityConfig}; +use datadriven::{run_test, TestData}; + +fn test_quorum(data: &TestData) -> String { + // Two majority configs. The first one is always used (though it may + // be empty) and the second one is used iff joint is true. + let mut joint = false; + let mut ids: Vec = Vec::new(); + let mut idsj: Vec = Vec::new(); + + // The committed indexes for the nodes in the config in the order in + // which they appear in (ids,idsj), without repetition. An underscore + // denotes an omission (i.e. no information for this voter); this is + // different from 0. + // + // For example, + // cfg=(1,2) cfgj=(2,3,4) idxs=(_,5,_,7) initializes the idx for voter 2 + // to 5 and that for voter 4 to 7 (and no others). + // + // cfgj=zero is specified to instruct the test harness to treat cfgj + // as zero instead of not specified (i.e. it will trigger a joint + // quorum test instead of a majority quorum test for cfg only). + let mut idxs: Vec = Vec::new(); + + for arg in &data.cmd_args { + for val in &arg.vals { + match arg.key.as_str() { + "cfg" => { + let n: u64 = val.parse().expect("type of n should be u64"); + ids.push(n); + } + "cfgj" => { + joint = true; + + if val == "zero" { + assert_eq!(arg.vals.len(), 1, "cannot mix 'zero' into configuration") + } else { + let n: u64 = val.parse().expect("type of n should be u64"); + idsj.push(n); + } + } + "idx" => { + let mut n: u64 = 0; + if val != "_" { + n = val.parse().expect("type of n should be u64"); + if n == 0 { + panic!("use '_' as 0, check {}", data.pos) + } + } + idxs.push(Index { + index: n, + group_id: 0, + }); + } + _ => { + panic!("unknown arg: {}", arg.key); + } + } + } + } + + let ids_set: HashSet = ids.iter().cloned().collect(); + let idsj_set: HashSet = idsj.iter().cloned().collect(); + + // Build the two majority configs. + let c = MajorityConfig::new(ids_set); + let cj = MajorityConfig::new(idsj_set); + + let make_lookuper = |idxs: &[Index], ids: &[u64], idsj: &[u64]| -> AckIndexer { + let mut l = AckIndexer::default(); + // next to consume from idxs + let mut p: usize = 0; + for id in ids.iter().chain(idsj) { + if !l.contains_key(id) && p < idxs.len() { + l.insert(*id, idxs[p]); + p += 1; + } + } + + // Zero entries are created by _ placeholders; we don't want + // them in the lookuper because "no entry" is different from + // "zero entry". Note that we prevent tests from specifying + // zero commit indexes, so that there's no confusion between + // the two concepts. + l.retain(|_, index| index.index > 0); + l + }; + + // verify length of voters + let input = idxs.len(); + let voters = JointConfig::new_joint_from_majorities(c.clone(), cj.clone()) + .ids() + .len(); + + if voters != input { + return format!( + "error: mismatched input (explicit or _) for voters {:?}: {:?}", + voters, input + ); + } + + // buffer for expected value + let mut buf = String::new(); + + match data.cmd.as_str() { + "committed" => { + let use_group_commit = false; + + let l = make_lookuper(&idxs, &ids, &idsj); + + // Branch based on whether this is a majority or joint quorum + // test case. + if joint { + let cc = JointConfig::new_joint_from_majorities(c.clone(), cj.clone()); + buf.push_str(&cc.describe(&l)); + let idx = cc.committed_index(use_group_commit, &l); + // Interchanging the majorities shouldn't make a difference. If it does, print. + let a_idx = JointConfig::new_joint_from_majorities(cj, c) + .committed_index(use_group_commit, &l); + if a_idx != idx { + buf.push_str(&format!("{} <-- via symmetry\n", a_idx.0)); + } + buf.push_str(&format!("{}\n", idx.0)); + } else { + // TODO(accelsao): port majority commit + } + } + _ => { + panic!("unknown command: {}", data.cmd); + } + } + buf +} + +#[test] +fn test_data_driven_quorum() -> anyhow::Result<()> { + let logger = default_logger(); + run_test("src/quorum/testdata", test_quorum, false, &logger)?; + Ok(()) +} diff --git a/src/quorum/joint.rs b/src/quorum/joint.rs index 164602037..9294b2ab8 100644 --- a/src/quorum/joint.rs +++ b/src/quorum/joint.rs @@ -23,6 +23,14 @@ impl Configuration { } } + #[cfg(test)] + pub(crate) fn new_joint_from_majorities( + incoming: MajorityConfig, + outgoing: MajorityConfig, + ) -> Self { + Self { incoming, outgoing } + } + /// Creates an empty configuration with given capacity. pub fn with_capacity(cap: usize) -> Configuration { Configuration { @@ -80,4 +88,11 @@ impl Configuration { pub fn contains(&self, id: u64) -> bool { self.incoming.contains(&id) || self.outgoing.contains(&id) } + + /// Describe returns a (multi-line) representation of the commit indexes for the + /// given lookuper. + #[cfg(test)] + pub(crate) fn describe(&self, l: &impl AckedIndexer) -> String { + MajorityConfig::new(self.ids().iter().collect()).describe(l) + } } diff --git a/src/quorum/majority.rs b/src/quorum/majority.rs index 6e9dd8fad..57b4f45dc 100644 --- a/src/quorum/majority.rs +++ b/src/quorum/majority.rs @@ -2,6 +2,7 @@ use super::{AckedIndexer, Index, VoteResult}; use crate::{DefaultHashBuilder, HashSet}; + use std::mem::MaybeUninit; use std::ops::{Deref, DerefMut}; use std::{cmp, slice, u64}; @@ -130,6 +131,90 @@ impl Configuration { VoteResult::Lost } } + + /// Describe returns a (multi-line) representation of the commit indexes for the + /// given lookuper. + /// Including `Index`,`Id` and the number of smaller index (represented as the bar) + /// + /// Print `?` if `Index` is not exist. + /// + /// e.g. + /// ```txt + /// idx + /// x> 100 (id=1) + /// xx> 101 (id=2) + /// > 99 (id=3) + /// 100 + /// ``` + #[cfg(test)] + pub(crate) fn describe(&self, l: &impl AckedIndexer) -> String { + use std::fmt::Write; + + let n = self.voters.len(); + if n == 0 { + return "".to_string(); + } + + struct Tup { + id: u64, + idx: Option, + // length of bar displayed for this Tup + bar: usize, + } + + // Below, populate .bar so that the i-th largest commit index has bar i (we + // plot this as sort of a progress bar). The actual code is a bit more + // complicated and also makes sure that equal index => equal bar. + + let mut info = Vec::with_capacity(n); + + for &id in &self.voters { + let idx = l.acked_index(id); + info.push(Tup { id, idx, bar: 0 }) + } + + info.sort_by(|a, b| { + (a.idx.unwrap_or_default().index, a.id).cmp(&(b.idx.unwrap_or_default().index, b.id)) + }); + + for i in 0..n { + if i > 0 + && info[i - 1].idx.unwrap_or_default().index < info[i].idx.unwrap_or_default().index + { + info[i].bar = i; + } + } + + info.sort_by(|a, b| a.id.cmp(&b.id)); + + let mut buf = String::new(); + buf.push_str(" ".repeat(n).as_str()); + buf.push_str(" idx\n"); + + for tup in info { + match tup.idx { + Some(idx) => { + buf.push_str("x".repeat(tup.bar).as_str()); + buf.push('>'); + buf.push_str(" ".repeat(n - tup.bar).as_str()); + writeln!(buf, " {:>5} (id={})", format!("{}", idx), tup.id) + .expect("Error occurred while trying to write in String"); + } + None => { + buf.push('?'); + buf.push_str(" ".repeat(n).as_str()); + writeln!( + buf, + " {:>5} (id={})", + format!("{}", Index::default()), + tup.id + ) + .expect("Error occurred while trying to write in String"); + } + } + } + buf + } } impl Deref for Configuration { diff --git a/src/quorum/testdata/joint_commit.txt b/src/quorum/testdata/joint_commit.txt new file mode 100644 index 000000000..3d2096484 --- /dev/null +++ b/src/quorum/testdata/joint_commit.txt @@ -0,0 +1,481 @@ +# No difference between a simple majority quorum and a simple majority quorum +# joint with an empty majority quorum. (This is asserted for all datadriven tests +# by the framework, so we don't dwell on it more). +# +# Note that by specifying cfgj explicitly we tell the test harness to treat the +# input as a joint quorum and not a majority quorum. If we didn't specify +# cfgj=zero the test would pass just the same, but it wouldn't be exercising the +# joint quorum path. +committed cfg=(1,2,3) cfgj=zero idx=(100,101,99) +---- + idx +x> 100 (id=1) +xx> 101 (id=2) +> 99 (id=3) +100 + +# Joint nonoverlapping singleton quorums. + +committed cfg=(1) cfgj=(2) idx=(_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +0 + +# Voter 1 has 100 committed, 2 nothing. This means we definitely won't commit +# past 100. +committed cfg=(1) cfgj=(2) idx=(100,_) +---- + idx +x> 100 (id=1) +? 0 (id=2) +0 + +# Committed index collapses once both majorities do, to the lower index. +committed cfg=(1) cfgj=(2) idx=(13, 100) +---- + idx +> 13 (id=1) +x> 100 (id=2) +13 + +# Joint overlapping (i.e. identical) singleton quorum. + +committed cfg=(1) cfgj=(1) idx=(_) +---- + idx +? 0 (id=1) +0 + +committed cfg=(1) cfgj=(1) idx=(100) +---- + idx +> 100 (id=1) +100 + + + +# Two-node config joint with non-overlapping single node config +committed cfg=(1,3) cfgj=(2) idx=(_,_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +? 0 (id=3) +0 + +committed cfg=(1,3) cfgj=(2) idx=(100,_,_) +---- + idx +xx> 100 (id=1) +? 0 (id=2) +? 0 (id=3) +0 + +# 1 has 100 committed, 2 has 50 (collapsing half of the joint quorum to 50). +committed cfg=(1,3) cfgj=(2) idx=(100,_,50) +---- + idx +xx> 100 (id=1) +x> 50 (id=2) +? 0 (id=3) +0 + +# 2 reports 45, collapsing the other half (to 45). +committed cfg=(1,3) cfgj=(2) idx=(100,45,50) +---- + idx +xx> 100 (id=1) +x> 50 (id=2) +> 45 (id=3) +45 + +# Two-node config with overlapping single-node config. + +committed cfg=(1,2) cfgj=(2) idx=(_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +0 + +# 1 reports 100. +committed cfg=(1,2) cfgj=(2) idx=(100,_) +---- + idx +x> 100 (id=1) +? 0 (id=2) +0 + +# 2 reports 100. +committed cfg=(1,2) cfgj=(2) idx=(_,100) +---- + idx +? 0 (id=1) +x> 100 (id=2) +0 + +committed cfg=(1,2) cfgj=(2) idx=(50,100) +---- + idx +> 50 (id=1) +x> 100 (id=2) +50 + +committed cfg=(1,2) cfgj=(2) idx=(100,50) +---- + idx +x> 100 (id=1) +> 50 (id=2) +50 + + + +# Joint non-overlapping two-node configs. + +committed cfg=(1,2) cfgj=(3,4) idx=(50,_,_,_) +---- + idx +xxx> 50 (id=1) +? 0 (id=2) +? 0 (id=3) +? 0 (id=4) +0 + +committed cfg=(1,2) cfgj=(3,4) idx=(50,_,49,_) +---- + idx +xxx> 50 (id=1) +? 0 (id=2) +xx> 49 (id=3) +? 0 (id=4) +0 + +committed cfg=(1,2) cfgj=(3,4) idx=(50,48,49,_) +---- + idx +xxx> 50 (id=1) +x> 48 (id=2) +xx> 49 (id=3) +? 0 (id=4) +0 + +committed cfg=(1,2) cfgj=(3,4) idx=(50,48,49,47) +---- + idx +xxx> 50 (id=1) +x> 48 (id=2) +xx> 49 (id=3) +> 47 (id=4) +47 + +# Joint overlapping two-node configs. +committed cfg=(1,2) cfgj=(2,3) idx=(_,_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +? 0 (id=3) +0 + +committed cfg=(1,2) cfgj=(2,3) idx=(100,_,_) +---- + idx +xx> 100 (id=1) +? 0 (id=2) +? 0 (id=3) +0 + +committed cfg=(1,2) cfgj=(2,3) idx=(_,100,_) +---- + idx +? 0 (id=1) +xx> 100 (id=2) +? 0 (id=3) +0 + +committed cfg=(1,2) cfgj=(2,3) idx=(_,100,99) +---- + idx +? 0 (id=1) +xx> 100 (id=2) +x> 99 (id=3) +0 + +committed cfg=(1,2) cfgj=(2,3) idx=(101,100,99) +---- + idx +xx> 101 (id=1) +x> 100 (id=2) +> 99 (id=3) +99 + +# Joint identical two-node configs. +committed cfg=(1,2) cfgj=(1,2) idx=(_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +0 + +committed cfg=(1,2) cfgj=(1,2) idx=(_,40) +---- + idx +? 0 (id=1) +x> 40 (id=2) +0 + +committed cfg=(1,2) cfgj=(1,2) idx=(41,40) +---- + idx +x> 41 (id=1) +> 40 (id=2) +40 + + + +# Joint disjoint three-node configs. + +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(_,_,_,_,_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +? 0 (id=3) +? 0 (id=4) +? 0 (id=5) +? 0 (id=6) +0 + +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(100,_,_,_,_,_) +---- + idx +xxxxx> 100 (id=1) +? 0 (id=2) +? 0 (id=3) +? 0 (id=4) +? 0 (id=5) +? 0 (id=6) +0 + +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(100,_,_,90,_,_) +---- + idx +xxxxx> 100 (id=1) +? 0 (id=2) +? 0 (id=3) +xxxx> 90 (id=4) +? 0 (id=5) +? 0 (id=6) +0 + +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(100,99,_,_,_,_) +---- + idx +xxxxx> 100 (id=1) +xxxx> 99 (id=2) +? 0 (id=3) +? 0 (id=4) +? 0 (id=5) +? 0 (id=6) +0 + +# First quorum <= 99, second one <= 97. Both quorums guarantee that 90 is +# committed. +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(_,99,90,97,95,_) +---- + idx +? 0 (id=1) +xxxxx> 99 (id=2) +xx> 90 (id=3) +xxxx> 97 (id=4) +xxx> 95 (id=5) +? 0 (id=6) +90 + +# First quorum collapsed to 92. Second one already had at least 95 committed, +# so the result also collapses. +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(92,99,90,97,95,_) +---- + idx +xx> 92 (id=1) +xxxxx> 99 (id=2) +x> 90 (id=3) +xxxx> 97 (id=4) +xxx> 95 (id=5) +? 0 (id=6) +92 + +# Second quorum collapses, but nothing changes in the output. +committed cfg=(1,2,3) cfgj=(4,5,6) idx=(92,99,90,97,95,77) +---- + idx +xx> 92 (id=1) +xxxxx> 99 (id=2) +x> 90 (id=3) +xxxx> 97 (id=4) +xxx> 95 (id=5) +> 77 (id=6) +92 + + +# Joint overlapping three-node configs. + +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(_,_,_,_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +? 0 (id=3) +? 0 (id=4) +? 0 (id=5) +0 + +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(100,_,_,_,_) +---- + idx +xxxx> 100 (id=1) +? 0 (id=2) +? 0 (id=3) +? 0 (id=4) +? 0 (id=5) +0 + +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(100,101,_,_,_) +---- + idx +xxx> 100 (id=1) +xxxx> 101 (id=2) +? 0 (id=3) +? 0 (id=4) +? 0 (id=5) +0 + +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(100,101,100,_,_) +---- + idx +xx> 100 (id=1) +xxxx> 101 (id=2) +> 100 (id=3) +? 0 (id=4) +? 0 (id=5) +0 + +# Second quorum could commit either 98 or 99, but first quorum is open. +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(_,100,_,99,98) +---- + idx +? 0 (id=1) +xxxx> 100 (id=2) +? 0 (id=3) +xxx> 99 (id=4) +xx> 98 (id=5) +0 + +# Additionally, first quorum can commit either 100 or 99 +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(_,100,99,99,98) +---- + idx +? 0 (id=1) +xxxx> 100 (id=2) +xx> 99 (id=3) +> 99 (id=4) +x> 98 (id=5) +98 + +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(1,100,99,99,98) +---- + idx +> 1 (id=1) +xxxx> 100 (id=2) +xx> 99 (id=3) +> 99 (id=4) +x> 98 (id=5) +98 + +committed cfg=(1,2,3) cfgj=(1,4,5) idx=(100,100,99,99,98) +---- + idx +xxx> 100 (id=1) +> 100 (id=2) +x> 99 (id=3) +> 99 (id=4) +> 98 (id=5) +99 + + +# More overlap. + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(_,_,_,_) +---- + idx +? 0 (id=1) +? 0 (id=2) +? 0 (id=3) +? 0 (id=4) +0 + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(_,100,99,_) +---- + idx +? 0 (id=1) +xxx> 100 (id=2) +xx> 99 (id=3) +? 0 (id=4) +99 + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(98,100,99,_) +---- + idx +x> 98 (id=1) +xxx> 100 (id=2) +xx> 99 (id=3) +? 0 (id=4) +99 + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(100,100,99,_) +---- + idx +xx> 100 (id=1) +> 100 (id=2) +x> 99 (id=3) +? 0 (id=4) +99 + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(100,100,99,98) +---- + idx +xx> 100 (id=1) +> 100 (id=2) +x> 99 (id=3) +> 98 (id=4) +99 + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(100,_,_,101) +---- + idx +xx> 100 (id=1) +? 0 (id=2) +? 0 (id=3) +xxx> 101 (id=4) +0 + +committed cfg=(1,2,3) cfgj=(2,3,4) idx=(100,99,_,101) +---- + idx +xx> 100 (id=1) +x> 99 (id=2) +? 0 (id=3) +xxx> 101 (id=4) +99 + +# Identical. This is also exercised in the test harness, so it's listed here +# only briefly. +committed cfg=(1,2,3) cfgj=(1,2,3) idx=(50,45,_) +---- + idx +xx> 50 (id=1) +x> 45 (id=2) +? 0 (id=3) +45 \ No newline at end of file