use std::cmp;
use eraftpb::{
ConfChange, ConfChangeType, Entry, EntryType, HardState, Message, MessageType, Snapshot,
};
use hashbrown::{HashMap, HashSet};
use protobuf;
use protobuf::RepeatedField;
use rand::{self, Rng};
use super::errors::{Error, Result, StorageError};
use super::progress::{CandidacyStatus, Configuration, Progress, ProgressSet, ProgressState};
use super::raft_log::{self, RaftLog};
use super::read_only::{ReadOnly, ReadOnlyOption, ReadState};
use super::storage::Storage;
use super::Config;
const CAMPAIGN_PRE_ELECTION: &[u8] = b"CampaignPreElection";
const CAMPAIGN_ELECTION: &[u8] = b"CampaignElection";
const CAMPAIGN_TRANSFER: &[u8] = b"CampaignTransfer";
#[derive(Debug, PartialEq, Clone, Copy)]
pub enum StateRole {
Follower,
Candidate,
Leader,
PreCandidate,
}
impl Default for StateRole {
fn default() -> StateRole {
StateRole::Follower
}
}
pub const INVALID_ID: u64 = 0;
pub const INVALID_INDEX: u64 = 0;
#[derive(Default, PartialEq, Debug)]
pub struct SoftState {
pub leader_id: u64,
pub raft_state: StateRole,
}
#[derive(Default, Getters)]
pub struct Raft<T: Storage> {
pub term: u64,
pub vote: u64,
pub id: u64,
pub read_states: Vec<ReadState>,
pub raft_log: RaftLog<T>,
pub max_inflight: usize,
pub max_msg_size: u64,
prs: Option<ProgressSet>,
pub state: StateRole,
pub is_learner: bool,
pub votes: HashMap<u64, bool>,
pub msgs: Vec<Message>,
pub leader_id: u64,
pub lead_transferee: Option<u64>,
pub pending_conf_index: u64,
#[get = "pub"]
pending_membership_change: Option<ConfChange>,
pub read_only: ReadOnly,
pub election_elapsed: usize,
heartbeat_elapsed: usize,
pub check_quorum: bool,
pub pre_vote: bool,
skip_bcast_commit: bool,
heartbeat_timeout: usize,
election_timeout: usize,
randomized_election_timeout: usize,
min_election_timeout: usize,
max_election_timeout: usize,
tag: String,
}
trait AssertSend: Send {}
impl<T: Storage + Send> AssertSend for Raft<T> {}
fn new_message(to: u64, field_type: MessageType, from: Option<u64>) -> Message {
let mut m = Message::new();
m.set_to(to);
if let Some(id) = from {
m.set_from(id);
}
m.set_msg_type(field_type);
m
}
pub fn vote_resp_msg_type(t: MessageType) -> MessageType {
match t {
MessageType::MsgRequestVote => MessageType::MsgRequestVoteResponse,
MessageType::MsgRequestPreVote => MessageType::MsgRequestPreVoteResponse,
_ => panic!("Not a vote message: {:?}", t),
}
}
impl<T: Storage> Raft<T> {
#[allow(clippy::new_ret_no_self)]
pub fn new(c: &Config, store: T) -> Result<Raft<T>> {
c.validate()?;
let raft_state = store.initial_state()?;
let conf_state = &raft_state.conf_state;
let raft_log = RaftLog::new(store, c.tag.clone());
let mut peers: &[u64] = &c.peers;
let mut learners: &[u64] = &c.learners;
if !conf_state.get_nodes().is_empty() || !conf_state.get_learners().is_empty() {
if !peers.is_empty() || !learners.is_empty() {
panic!(
"{} cannot specify both new(peers/learners) and ConfState.(Nodes/Learners)",
c.tag
)
}
peers = conf_state.get_nodes();
learners = conf_state.get_learners();
}
let mut r = Raft {
id: c.id,
read_states: Default::default(),
raft_log,
max_inflight: c.max_inflight_msgs,
max_msg_size: c.max_size_per_msg,
prs: Some(ProgressSet::with_capacity(peers.len(), learners.len())),
state: StateRole::Follower,
is_learner: false,
check_quorum: c.check_quorum,
pre_vote: c.pre_vote,
read_only: ReadOnly::new(c.read_only_option),
heartbeat_timeout: c.heartbeat_tick,
election_timeout: c.election_tick,
votes: Default::default(),
msgs: Default::default(),
leader_id: Default::default(),
lead_transferee: None,
term: Default::default(),
election_elapsed: Default::default(),
pending_conf_index: Default::default(),
pending_membership_change: Default::default(),
vote: Default::default(),
heartbeat_elapsed: Default::default(),
randomized_election_timeout: 0,
min_election_timeout: c.min_election_tick(),
max_election_timeout: c.max_election_tick(),
skip_bcast_commit: c.skip_bcast_commit,
tag: c.tag.to_owned(),
};
for p in peers {
let pr = Progress::new(1, r.max_inflight);
if let Err(e) = r.mut_prs().insert_voter(*p, pr) {
panic!("{}", e);
}
}
for p in learners {
let pr = Progress::new(1, r.max_inflight);
if let Err(e) = r.mut_prs().insert_learner(*p, pr) {
panic!("{}", e);
};
if *p == r.id {
r.is_learner = true;
}
}
if raft_state.hard_state != HardState::new() {
r.load_state(&raft_state.hard_state);
}
if c.applied > 0 {
r.commit_apply(c.applied);
}
let term = r.term;
r.become_follower(term, INVALID_ID);
let pending_conf_state = raft_state.pending_conf_state();
let pending_conf_state_start_index = raft_state.pending_conf_state_start_index();
match (pending_conf_state, pending_conf_state_start_index) {
(Some(state), Some(idx)) => {
r.begin_membership_change(&ConfChange::from((*idx, state.clone())))?;
}
(None, None) => (),
_ => unreachable!("Should never find pending_conf_change without an index."),
};
info!(
"{} newRaft [peers: {:?}, term: {:?}, commit: {}, applied: {}, last_index: {}, \
last_term: {}, pending_membership_change: {:?}]",
r.tag,
r.prs().voters().collect::<Vec<_>>(),
r.term,
r.raft_log.committed,
r.raft_log.get_applied(),
r.raft_log.last_index(),
r.raft_log.last_term(),
r.pending_membership_change(),
);
Ok(r)
}
#[inline]
pub fn get_store(&self) -> &T {
self.raft_log.get_store()
}
#[inline]
pub fn mut_store(&mut self) -> &mut T {
self.raft_log.mut_store()
}
#[inline]
pub fn get_snap(&self) -> Option<&Snapshot> {
self.raft_log.get_unstable().snapshot.as_ref()
}
#[inline]
pub fn pending_read_count(&self) -> usize {
self.read_only.pending_read_count()
}
#[inline]
pub fn ready_read_count(&self) -> usize {
self.read_states.len()
}
pub fn soft_state(&self) -> SoftState {
SoftState {
leader_id: self.leader_id,
raft_state: self.state,
}
}
pub fn hard_state(&self) -> HardState {
let mut hs = HardState::new();
hs.set_term(self.term);
hs.set_vote(self.vote);
hs.set_commit(self.raft_log.committed);
hs
}
pub fn in_lease(&self) -> bool {
self.state == StateRole::Leader && self.check_quorum
}
#[doc(hidden)]
pub fn set_randomized_election_timeout(&mut self, t: usize) {
assert!(self.min_election_timeout <= t && t < self.max_election_timeout);
self.randomized_election_timeout = t;
}
pub fn get_election_timeout(&self) -> usize {
self.election_timeout
}
pub fn get_heartbeat_timeout(&self) -> usize {
self.heartbeat_timeout
}
pub fn get_heartbeat_elapsed(&self) -> usize {
self.heartbeat_elapsed
}
pub fn get_randomized_election_timeout(&self) -> usize {
self.randomized_election_timeout
}
#[inline]
pub fn skip_bcast_commit(&mut self, skip: bool) {
self.skip_bcast_commit = skip;
}
#[inline]
fn set_pending_membership_change(&mut self, maybe_change: impl Into<Option<ConfChange>>) {
let maybe_change = maybe_change.into();
if let Some(ref change) = maybe_change {
let index = change.get_start_index();
assert!(self.pending_membership_change.is_none() || index == self.pending_conf_index);
if index > self.pending_conf_index {
self.pending_conf_index = index;
}
}
self.pending_membership_change = maybe_change.clone();
}
#[inline]
pub fn began_membership_change_at(&self) -> Option<u64> {
self.pending_membership_change
.as_ref()
.map(|v| v.get_start_index())
}
fn send(&mut self, mut m: Message) {
debug!("Sending from {} to {}: {:?}", self.id, m.get_to(), m);
m.set_from(self.id);
if m.get_msg_type() == MessageType::MsgRequestVote
|| m.get_msg_type() == MessageType::MsgRequestPreVote
|| m.get_msg_type() == MessageType::MsgRequestVoteResponse
|| m.get_msg_type() == MessageType::MsgRequestPreVoteResponse
{
if m.get_term() == 0 {
panic!(
"{} term should be set when sending {:?}",
self.tag,
m.get_msg_type()
);
}
} else {
if m.get_term() != 0 {
panic!(
"{} term should not be set when sending {:?} (was {})",
self.tag,
m.get_msg_type(),
m.get_term()
);
}
if m.get_msg_type() != MessageType::MsgPropose
&& m.get_msg_type() != MessageType::MsgReadIndex
{
m.set_term(self.term);
}
}
self.msgs.push(m);
}
fn prepare_send_snapshot(&mut self, m: &mut Message, pr: &mut Progress, to: u64) -> bool {
if !pr.recent_active {
debug!(
"{} ignore sending snapshot to {} since it is not recently active",
self.tag, to
);
return false;
}
m.set_msg_type(MessageType::MsgSnapshot);
let snapshot_r = self.raft_log.snapshot();
if let Err(e) = snapshot_r {
if e == Error::Store(StorageError::SnapshotTemporarilyUnavailable) {
debug!(
"{} failed to send snapshot to {} because snapshot is temporarily \
unavailable",
self.tag, to
);
return false;
}
panic!("{} unexpected error: {:?}", self.tag, e);
}
let snapshot = snapshot_r.unwrap();
if snapshot.get_metadata().get_index() == 0 {
panic!("{} need non-empty snapshot", self.tag);
}
let (sindex, sterm) = (
snapshot.get_metadata().get_index(),
snapshot.get_metadata().get_term(),
);
m.set_snapshot(snapshot);
debug!(
"{} [firstindex: {}, commit: {}] sent snapshot[index: {}, term: {}] to {} \
[{:?}]",
self.tag,
self.raft_log.first_index(),
self.raft_log.committed,
sindex,
sterm,
to,
pr
);
pr.become_snapshot(sindex);
debug!(
"{} paused sending replication messages to {} [{:?}]",
self.tag, to, pr
);
true
}
fn prepare_send_entries(
&mut self,
m: &mut Message,
pr: &mut Progress,
term: u64,
ents: Vec<Entry>,
) {
m.set_msg_type(MessageType::MsgAppend);
m.set_index(pr.next_idx - 1);
m.set_log_term(term);
m.set_entries(RepeatedField::from_vec(ents));
m.set_commit(self.raft_log.committed);
if !m.get_entries().is_empty() {
match pr.state {
ProgressState::Replicate => {
let last = m.get_entries().last().unwrap().get_index();
pr.optimistic_update(last);
pr.ins.add(last);
}
ProgressState::Probe => pr.pause(),
_ => panic!(
"{} is sending append in unhandled state {:?}",
self.tag, pr.state
),
}
}
}
pub fn send_append(&mut self, to: u64, pr: &mut Progress) {
if pr.is_paused() {
trace!(
"Skipping sending to {}, it's paused. Progress: {:?}",
to,
pr
);
return;
}
let term = self.raft_log.term(pr.next_idx - 1);
let ents = self.raft_log.entries(pr.next_idx, self.max_msg_size);
let mut m = Message::new();
m.set_to(to);
if term.is_err() || ents.is_err() {
trace!(
"{} Skipping sending to {}, term: {:?}, ents: {:?}",
self.tag,
to,
term,
ents,
);
if !self.prepare_send_snapshot(&mut m, pr, to) {
return;
}
} else {
self.prepare_send_entries(&mut m, pr, term.unwrap(), ents.unwrap());
}
self.send(m);
}
fn send_heartbeat(&mut self, to: u64, pr: &Progress, ctx: Option<Vec<u8>>) {
let mut m = Message::new();
m.set_to(to);
m.set_msg_type(MessageType::MsgHeartbeat);
let commit = cmp::min(pr.matched, self.raft_log.committed);
m.set_commit(commit);
if let Some(context) = ctx {
m.set_context(context);
}
self.send(m);
}
pub fn bcast_append(&mut self) {
let self_id = self.id;
let mut prs = self.take_prs();
prs.iter_mut()
.filter(|&(id, _)| *id != self_id)
.for_each(|(id, pr)| self.send_append(*id, pr));
self.set_prs(prs);
}
pub fn bcast_heartbeat(&mut self) {
let ctx = self.read_only.last_pending_request_ctx();
self.bcast_heartbeat_with_ctx(ctx)
}
#[cfg_attr(feature = "cargo-clippy", allow(clippy::needless_pass_by_value))]
fn bcast_heartbeat_with_ctx(&mut self, ctx: Option<Vec<u8>>) {
let self_id = self.id;
let mut prs = self.take_prs();
prs.iter_mut()
.filter(|&(id, _)| *id != self_id)
.for_each(|(id, pr)| self.send_heartbeat(*id, pr, ctx.clone()));
self.set_prs(prs);
}
pub fn maybe_commit(&mut self) -> bool {
let mci = self.prs().maximal_committed_index();
self.raft_log.maybe_commit(mci, self.term)
}
pub fn commit_apply(&mut self, applied: u64) {
#[allow(deprecated)]
self.raft_log.applied_to(applied);
let start_index = self
.pending_membership_change
.as_ref()
.map(|v| Some(v.get_start_index()))
.unwrap_or(None);
if let Some(index) = start_index {
if applied >= index && self.state == StateRole::Leader {
self.append_finalize_conf_change_entry();
}
}
}
fn append_finalize_conf_change_entry(&mut self) {
let mut conf_change = ConfChange::new();
conf_change.set_change_type(ConfChangeType::FinalizeMembershipChange);
let data = protobuf::Message::write_to_bytes(&conf_change).unwrap();
let mut entry = Entry::new();
entry.set_entry_type(EntryType::EntryConfChange);
entry.set_data(data);
self.append_entry(&mut [entry]);
self.bcast_append();
}
pub fn reset(&mut self, term: u64) {
if self.term != term {
self.term = term;
self.vote = INVALID_ID;
}
self.leader_id = INVALID_ID;
self.reset_randomized_election_timeout();
self.election_elapsed = 0;
self.heartbeat_elapsed = 0;
self.abort_leader_transfer();
self.votes.clear();
self.pending_conf_index = 0;
self.read_only = ReadOnly::new(self.read_only.option);
let last_index = self.raft_log.last_index();
let self_id = self.id;
for (&id, pr) in self.mut_prs().iter_mut() {
pr.reset(last_index + 1);
if id == self_id {
pr.matched = last_index;
}
}
}
pub fn append_entry(&mut self, es: &mut [Entry]) {
let mut li = self.raft_log.last_index();
for (i, e) in es.iter_mut().enumerate() {
e.set_term(self.term);
e.set_index(li + 1 + i as u64);
}
li = self.raft_log.append(es);
let self_id = self.id;
self.mut_prs().get_mut(self_id).unwrap().maybe_update(li);
self.maybe_commit();
}
pub fn tick(&mut self) -> bool {
match self.state {
StateRole::Follower | StateRole::PreCandidate | StateRole::Candidate => {
self.tick_election()
}
StateRole::Leader => self.tick_heartbeat(),
}
}
pub fn tick_election(&mut self) -> bool {
self.election_elapsed += 1;
if !self.pass_election_timeout() || !self.promotable() {
return false;
}
self.election_elapsed = 0;
let m = new_message(INVALID_ID, MessageType::MsgHup, Some(self.id));
self.step(m).is_ok();
true
}
fn tick_heartbeat(&mut self) -> bool {
self.heartbeat_elapsed += 1;
self.election_elapsed += 1;
let mut has_ready = false;
if self.election_elapsed >= self.election_timeout {
self.election_elapsed = 0;
if self.check_quorum {
let m = new_message(INVALID_ID, MessageType::MsgCheckQuorum, Some(self.id));
has_ready = true;
self.step(m).is_ok();
}
if self.state == StateRole::Leader && self.lead_transferee.is_some() {
self.abort_leader_transfer()
}
}
if self.state != StateRole::Leader {
return has_ready;
}
if self.heartbeat_elapsed >= self.heartbeat_timeout {
self.heartbeat_elapsed = 0;
has_ready = true;
let m = new_message(INVALID_ID, MessageType::MsgBeat, Some(self.id));
self.step(m).is_ok();
}
has_ready
}
pub fn become_follower(&mut self, term: u64, leader_id: u64) {
self.reset(term);
self.leader_id = leader_id;
self.state = StateRole::Follower;
info!("{} became follower at term {}", self.tag, self.term);
}
pub fn become_candidate(&mut self) {
assert_ne!(
self.state,
StateRole::Leader,
"invalid transition [leader -> candidate]"
);
let term = self.term + 1;
self.reset(term);
let id = self.id;
self.vote = id;
self.state = StateRole::Candidate;
info!("{} became candidate at term {}", self.tag, self.term);
}
pub fn become_pre_candidate(&mut self) {
assert_ne!(
self.state,
StateRole::Leader,
"invalid transition [leader -> pre-candidate]"
);
self.state = StateRole::PreCandidate;
self.votes = HashMap::default();
self.leader_id = INVALID_ID;
info!("{} became pre-candidate at term {}", self.tag, self.term);
}
pub fn become_leader(&mut self) {
trace!("ENTER become_leader");
assert_ne!(
self.state,
StateRole::Follower,
"invalid transition [follower -> leader]"
);
let term = self.term;
self.reset(term);
self.leader_id = self.id;
self.state = StateRole::Leader;
let id = self.id;
self.mut_prs().get_mut(id).unwrap().become_replicate();
self.pending_conf_index = self.raft_log.last_index();
self.append_entry(&mut [Entry::new()]);
let change_start_index = self
.pending_membership_change
.as_ref()
.map(|v| Some(v.get_start_index()))
.unwrap_or(None);
if let Some(index) = change_start_index {
trace!(
"Checking if we need to finalize again..., began: {}, applied: {}, committed: {}",
index,
self.raft_log.applied,
self.raft_log.committed
);
if index <= self.raft_log.committed {
self.append_finalize_conf_change_entry();
}
}
info!("{} became leader at term {}", self.tag, self.term);
trace!("EXIT become_leader");
}
fn num_pending_conf(&self, ents: &[Entry]) -> usize {
ents.iter()
.filter(|e| e.get_entry_type() == EntryType::EntryConfChange)
.count()
}
pub fn campaign(&mut self, campaign_type: &[u8]) {
let (vote_msg, term) = if campaign_type == CAMPAIGN_PRE_ELECTION {
self.become_pre_candidate();
(MessageType::MsgRequestPreVote, self.term + 1)
} else {
self.become_candidate();
(MessageType::MsgRequestVote, self.term)
};
let self_id = self.id;
let acceptance = true;
info!(
"{} received {:?} from {} at term {}",
self.id, vote_msg, self_id, self.term
);
self.register_vote(self_id, acceptance);
if let CandidacyStatus::Elected = self.prs().candidacy_status(&self.votes) {
if campaign_type == CAMPAIGN_PRE_ELECTION {
self.campaign(CAMPAIGN_ELECTION);
} else {
self.become_leader();
}
return;
}
let prs = self.take_prs();
prs.voter_ids()
.iter()
.filter(|&id| *id != self_id)
.for_each(|&id| {
info!(
"{} [logterm: {}, index: {}] sent {:?} request to {} at term {}",
self.tag,
self.raft_log.last_term(),
self.raft_log.last_index(),
vote_msg,
id,
self.term
);
let mut m = new_message(id, vote_msg, None);
m.set_term(term);
m.set_index(self.raft_log.last_index());
m.set_log_term(self.raft_log.last_term());
if campaign_type == CAMPAIGN_TRANSFER {
m.set_context(campaign_type.to_vec());
}
self.send(m);
});
self.set_prs(prs);
}
fn register_vote(&mut self, id: u64, vote: bool) {
self.votes.entry(id).or_insert(vote);
}
pub fn step(&mut self, m: Message) -> Result<()> {
if m.get_term() == 0 {
} else if m.get_term() > self.term {
if m.get_msg_type() == MessageType::MsgRequestVote
|| m.get_msg_type() == MessageType::MsgRequestPreVote
{
let force = m.get_context() == CAMPAIGN_TRANSFER;
let in_lease = self.check_quorum
&& self.leader_id != INVALID_ID
&& self.election_elapsed < self.election_timeout;
if !force && in_lease {
info!(
"{} [logterm: {}, index: {}, vote: {}] ignored {:?} vote from \
{} [logterm: {}, index: {}] at term {}: lease is not expired \
(remaining ticks: {})",
self.tag,
self.raft_log.last_term(),
self.raft_log.last_index(),
self.vote,
m.get_msg_type(),
m.get_from(),
m.get_log_term(),
m.get_index(),
self.term,
self.election_timeout - self.election_elapsed
);
return Ok(());
}
}
if m.get_msg_type() == MessageType::MsgRequestPreVote
|| (m.get_msg_type() == MessageType::MsgRequestPreVoteResponse && !m.get_reject())
{
} else {
info!(
"{} [term: {}] received a {:?} message with higher term from {} [term: {}]",
self.tag,
self.term,
m.get_msg_type(),
m.get_from(),
m.get_term()
);
if m.get_msg_type() == MessageType::MsgAppend
|| m.get_msg_type() == MessageType::MsgHeartbeat
|| m.get_msg_type() == MessageType::MsgSnapshot
{
self.become_follower(m.get_term(), m.get_from());
} else {
self.become_follower(m.get_term(), INVALID_ID);
}
}
} else if m.get_term() < self.term {
if (self.check_quorum || self.pre_vote)
&& (m.get_msg_type() == MessageType::MsgHeartbeat
|| m.get_msg_type() == MessageType::MsgAppend)
{
let to_send = new_message(m.get_from(), MessageType::MsgAppendResponse, None);
self.send(to_send);
} else if m.get_msg_type() == MessageType::MsgRequestPreVote {
info!(
"{} [log_term: {}, index: {}, vote: {}] rejected {:?} from {} [log_term: {}, index: {}] at term {}",
self.id,
self.raft_log.last_term(),
self.raft_log.last_index(),
self.vote,
m.get_msg_type(),
m.get_from(),
m.get_log_term(),
m.get_index(),
self.term,
);
let mut to_send =
new_message(m.get_from(), MessageType::MsgRequestPreVoteResponse, None);
to_send.set_term(self.term);
to_send.set_reject(true);
self.send(to_send);
} else {
info!(
"{} [term: {}] ignored a {:?} message with lower term from {} [term: {}]",
self.tag,
self.term,
m.get_msg_type(),
m.get_from(),
m.get_term()
);
}
return Ok(());
}
#[cfg(feature = "failpoint")]
fail_point!("before_step");
match m.get_msg_type() {
MessageType::MsgHup => {
if self.state != StateRole::Leader {
let ents = self
.raft_log
.slice(
self.raft_log.applied + 1,
self.raft_log.committed + 1,
raft_log::NO_LIMIT,
)
.expect("unexpected error getting unapplied entries");
let n = self.num_pending_conf(&ents);
if n != 0 && self.raft_log.committed > self.raft_log.applied {
warn!(
"{} cannot campaign at term {} since there are still {} pending \
configuration changes to apply",
self.tag, self.term, n
);
return Ok(());
}
info!(
"{} is starting a new election at term {}",
self.tag, self.term
);
if self.pre_vote {
self.campaign(CAMPAIGN_PRE_ELECTION);
} else {
self.campaign(CAMPAIGN_ELECTION);
}
} else {
debug!("{} ignoring MsgHup because already leader", self.tag);
}
}
MessageType::MsgRequestVote | MessageType::MsgRequestPreVote => {
let can_vote = (self.vote == m.get_from()) ||
(self.vote == INVALID_ID && self.leader_id == INVALID_ID) ||
(m.msg_type == MessageType::MsgRequestPreVote && m.get_term() > self.term);
if can_vote && self.raft_log.is_up_to_date(m.get_index(), m.get_log_term()) {
self.log_vote_approve(&m);
let mut to_send =
new_message(m.get_from(), vote_resp_msg_type(m.get_msg_type()), None);
to_send.set_reject(false);
to_send.set_term(m.get_term());
self.send(to_send);
if m.get_msg_type() == MessageType::MsgRequestVote {
self.election_elapsed = 0;
self.vote = m.get_from();
}
} else {
self.log_vote_reject(&m);
let mut to_send =
new_message(m.get_from(), vote_resp_msg_type(m.get_msg_type()), None);
to_send.set_reject(true);
to_send.set_term(self.term);
self.send(to_send);
}
}
_ => match self.state {
StateRole::PreCandidate | StateRole::Candidate => self.step_candidate(m)?,
StateRole::Follower => self.step_follower(m)?,
StateRole::Leader => self.step_leader(m)?,
},
}
Ok(())
}
#[inline(always)]
pub fn begin_membership_change(&mut self, conf_change: &ConfChange) -> Result<()> {
if conf_change.get_change_type() != ConfChangeType::BeginMembershipChange {
return Err(Error::ViolatesContract(format!(
"{:?} != BeginMembershipChange",
conf_change.get_change_type()
)));
}
let configuration = if conf_change.has_configuration() {
conf_change.get_configuration().clone()
} else {
return Err(Error::ViolatesContract(
"!ConfChange::has_configuration()".into(),
));
};
if conf_change.get_start_index() == 0 {
return Err(Error::ViolatesContract(
"!ConfChange::has_start_index()".into(),
));
};
self.set_pending_membership_change(conf_change.clone());
let max_inflights = self.max_inflight;
self.mut_prs()
.begin_membership_change(configuration, Progress::new(1, max_inflights))?;
Ok(())
}
#[inline(always)]
pub fn finalize_membership_change(&mut self, conf_change: &ConfChange) -> Result<()> {
if conf_change.get_change_type() != ConfChangeType::FinalizeMembershipChange {
return Err(Error::ViolatesContract(format!(
"{:?} != BeginMembershipChange",
conf_change.get_change_type()
)));
}
if conf_change.has_configuration() {
return Err(Error::ViolatesContract(
"ConfChange::has_configuration()".into(),
));
};
let leader_in_new_set = self
.prs()
.next_configuration()
.as_ref()
.map(|config| config.contains(self.leader_id))
.ok_or_else(|| Error::NoPendingMembershipChange)?;
if !leader_in_new_set {
let last_term = self.raft_log.last_term();
if self.state == StateRole::Leader {
self.become_follower(last_term, INVALID_ID);
} else {
self.leader_id = INVALID_ID;
}
}
self.mut_prs().finalize_membership_change()?;
self.set_pending_membership_change(None);
Ok(())
}
fn log_vote_approve(&self, m: &Message) {
info!(
"{} [logterm: {}, index: {}, vote: {}] cast {:?} for {} [logterm: {}, index: {}] \
at term {}",
self.tag,
self.raft_log.last_term(),
self.raft_log.last_index(),
self.vote,
m.get_msg_type(),
m.get_from(),
m.get_log_term(),
m.get_index(),
self.term
);
}
fn log_vote_reject(&self, m: &Message) {
info!(
"{} [logterm: {}, index: {}, vote: {}] rejected {:?} from {} [logterm: {}, index: \
{}] at term {}",
self.tag,
self.raft_log.last_term(),
self.raft_log.last_index(),
self.vote,
m.get_msg_type(),
m.get_from(),
m.get_log_term(),
m.get_index(),
self.term
);
}
fn handle_append_response(
&mut self,
m: &Message,
prs: &mut ProgressSet,
old_paused: &mut bool,
send_append: &mut bool,
maybe_commit: &mut bool,
) {
let pr = prs.get_mut(m.get_from()).unwrap();
pr.recent_active = true;
if m.get_reject() {
debug!(
"{} received msgAppend rejection(lastindex: {}) from {} for index {}",
self.tag,
m.get_reject_hint(),
m.get_from(),
m.get_index()
);
if pr.maybe_decr_to(m.get_index(), m.get_reject_hint()) {
debug!(
"{} decreased progress of {} to [{:?}]",
self.tag,
m.get_from(),
pr
);
if pr.state == ProgressState::Replicate {
pr.become_probe();
}
*send_append = true;
}
return;
}
*old_paused = pr.is_paused();
if !pr.maybe_update(m.get_index()) {
return;
}
if let Some(lead_transferee) = self.lead_transferee {
let last_index = self.raft_log.last_index();
if m.get_from() == lead_transferee && pr.matched == last_index {
info!(
"{} sent MsgTimeoutNow to {} after received MsgAppResp",
self.tag,
m.get_from()
);
self.send_timeout_now(m.get_from());
}
}
match pr.state {
ProgressState::Probe => pr.become_replicate(),
ProgressState::Snapshot => {
if !pr.maybe_snapshot_abort() {
return;
}
debug!(
"{} snapshot aborted, resumed sending replication messages to {} \
[{:?}]",
self.tag,
m.get_from(),
pr
);
pr.become_probe();
}
ProgressState::Replicate => pr.ins.free_to(m.get_index()),
}
*maybe_commit = true;
}
fn handle_heartbeat_response(
&mut self,
m: &Message,
prs: &mut ProgressSet,
send_append: &mut bool,
more_to_send: &mut Option<Message>,
) {
{
let pr = prs.get_mut(m.get_from()).unwrap();
pr.recent_active = true;
pr.resume();
if pr.state == ProgressState::Replicate && pr.ins.full() {
pr.ins.free_first_one();
}
if pr.matched < self.raft_log.last_index() {
*send_append = true;
}
if self.read_only.option != ReadOnlyOption::Safe || m.get_context().is_empty() {
return;
}
}
if !prs.has_quorum(&self.read_only.recv_ack(m)) {
return;
}
let rss = self.read_only.advance(m);
for rs in rss {
let mut req = rs.req;
if req.get_from() == INVALID_ID || req.get_from() == self.id {
let rs = ReadState {
index: rs.index,
request_ctx: req.take_entries()[0].take_data(),
};
self.read_states.push(rs);
} else {
let mut to_send = Message::new();
to_send.set_to(req.get_from());
to_send.set_msg_type(MessageType::MsgReadIndexResp);
to_send.set_index(rs.index);
to_send.set_entries(req.take_entries());
*more_to_send = Some(to_send);
}
}
}
fn handle_transfer_leader(&mut self, m: &Message, prs: &mut ProgressSet) {
let from = m.get_from();
if prs.learner_ids().contains(&from) {
debug!("{} is learner. Ignored transferring leadership", from);
return;
}
let lead_transferee = from;
let last_lead_transferee = self.lead_transferee;
if last_lead_transferee.is_some() {
if last_lead_transferee.unwrap() == lead_transferee {
info!(
"{} [term {}] transfer leadership to {} is in progress, ignores request \
to same node {}",
self.tag, self.term, lead_transferee, lead_transferee
);
return;
}
self.abort_leader_transfer();
info!(
"{} [term {}] abort previous transferring leadership to {}",
self.tag,
self.term,
last_lead_transferee.unwrap()
);
}
if lead_transferee == self.id {
debug!(
"{} is already leader. Ignored transferring leadership to self",
self.tag
);
return;
}
info!(
"{} [term {}] starts to transfer leadership to {}",
self.tag, self.term, lead_transferee
);
self.election_elapsed = 0;
self.lead_transferee = Some(lead_transferee);
let pr = prs.get_mut(from).unwrap();
if pr.matched == self.raft_log.last_index() {
self.send_timeout_now(lead_transferee);
info!(
"{} sends MsgTimeoutNow to {} immediately as {} already has up-to-date log",
self.tag, lead_transferee, lead_transferee
);
} else {
self.send_append(lead_transferee, pr);
}
}
fn handle_snapshot_status(&mut self, m: &Message, pr: &mut Progress) {
if m.get_reject() {
pr.snapshot_failure();
pr.become_probe();
debug!(
"{} snapshot failed, resumed sending replication messages to {} [{:?}]",
self.tag,
m.get_from(),
pr
);
} else {
pr.become_probe();
debug!(
"{} snapshot succeeded, resumed sending replication messages to {} [{:?}]",
self.tag,
m.get_from(),
pr
);
}
pr.pause();
}
fn check_message_with_progress(
&mut self,
m: &mut Message,
send_append: &mut bool,
old_paused: &mut bool,
maybe_commit: &mut bool,
more_to_send: &mut Option<Message>,
) {
if self.prs().get(m.get_from()).is_none() {
debug!("{} no progress available for {}", self.tag, m.get_from());
return;
}
let mut prs = self.take_prs();
match m.get_msg_type() {
MessageType::MsgAppendResponse => {
self.handle_append_response(m, &mut prs, old_paused, send_append, maybe_commit);
}
MessageType::MsgHeartbeatResponse => {
self.handle_heartbeat_response(m, &mut prs, send_append, more_to_send);
}
MessageType::MsgSnapStatus => {
let pr = prs.get_mut(m.get_from()).unwrap();
if pr.state == ProgressState::Snapshot {
self.handle_snapshot_status(m, pr);
}
}
MessageType::MsgUnreachable => {
let pr = prs.get_mut(m.get_from()).unwrap();
if pr.state == ProgressState::Replicate {
pr.become_probe();
}
debug!(
"{} failed to send message to {} because it is unreachable [{:?}]",
self.tag,
m.get_from(),
pr
);
}
MessageType::MsgTransferLeader => {
self.handle_transfer_leader(m, &mut prs);
}
_ => {}
}
self.set_prs(prs);
}
fn step_leader(&mut self, mut m: Message) -> Result<()> {
match m.get_msg_type() {
MessageType::MsgBeat => {
self.bcast_heartbeat();
return Ok(());
}
MessageType::MsgCheckQuorum => {
if !self.check_quorum_active() {
warn!(
"{} stepped down to follower since quorum is not active",
self.tag
);
let term = self.term;
self.become_follower(term, INVALID_ID);
}
return Ok(());
}
MessageType::MsgPropose => {
if m.get_entries().is_empty() {
panic!("{} stepped empty MsgProp", self.tag);
}
if !self.prs().voter_ids().contains(&self.id) {
return Err(Error::ProposalDropped);
}
if self.lead_transferee.is_some() {
debug!(
"{} [term {}] transfer leadership to {} is in progress; dropping \
proposal",
self.tag,
self.term,
self.lead_transferee.unwrap()
);
return Err(Error::ProposalDropped);
}
for (i, e) in m.mut_entries().iter_mut().enumerate() {
if e.get_entry_type() == EntryType::EntryConfChange {
if self.has_pending_conf() {
info!(
"propose conf {:?} ignored since pending unapplied \
configuration [index {}, applied {}]",
e, self.pending_conf_index, self.raft_log.applied
);
*e = Entry::new();
e.set_entry_type(EntryType::EntryNormal);
} else {
self.pending_conf_index = self.raft_log.last_index() + i as u64 + 1;
}
}
}
self.append_entry(&mut m.mut_entries());
self.bcast_append();
return Ok(());
}
MessageType::MsgReadIndex => {
if self.raft_log.term(self.raft_log.committed).unwrap_or(0) != self.term {
return Ok(());
}
let mut self_set = HashSet::default();
self_set.insert(self.id);
if !self.prs().has_quorum(&self_set) {
match self.read_only.option {
ReadOnlyOption::Safe => {
let ctx = m.get_entries()[0].get_data().to_vec();
self.read_only.add_request(self.raft_log.committed, m);
self.bcast_heartbeat_with_ctx(Some(ctx));
}
ReadOnlyOption::LeaseBased => {
let mut read_index = self.raft_log.committed;
if m.get_from() == INVALID_ID || m.get_from() == self.id {
let rs = ReadState {
index: read_index,
request_ctx: m.take_entries()[0].take_data(),
};
self.read_states.push(rs);
} else {
let mut to_send = Message::new();
to_send.set_to(m.get_from());
to_send.set_msg_type(MessageType::MsgReadIndexResp);
to_send.set_index(read_index);
to_send.set_entries(m.take_entries());
self.send(to_send);
}
}
}
} else {
let rs = ReadState {
index: self.raft_log.committed,
request_ctx: m.take_entries()[0].take_data(),
};
self.read_states.push(rs);
}
return Ok(());
}
_ => {}
}
let mut send_append = false;
let mut maybe_commit = false;
let mut old_paused = false;
let mut more_to_send = None;
self.check_message_with_progress(
&mut m,
&mut send_append,
&mut old_paused,
&mut maybe_commit,
&mut more_to_send,
);
if maybe_commit {
if self.maybe_commit() {
if self.should_bcast_commit() {
self.bcast_append();
}
} else if old_paused {
send_append = true;
}
}
if send_append {
let from = m.get_from();
let mut prs = self.take_prs();
self.send_append(from, prs.get_mut(from).unwrap());
self.set_prs(prs);
}
if let Some(to_send) = more_to_send {
self.send(to_send)
}
Ok(())
}
fn step_candidate(&mut self, m: Message) -> Result<()> {
match m.get_msg_type() {
MessageType::MsgPropose => {
info!(
"{} no leader at term {}; dropping proposal",
self.tag, self.term
);
return Err(Error::ProposalDropped);
}
MessageType::MsgAppend => {
debug_assert_eq!(self.term, m.get_term());
self.become_follower(m.get_term(), m.get_from());
self.handle_append_entries(&m);
}
MessageType::MsgHeartbeat => {
debug_assert_eq!(self.term, m.get_term());
self.become_follower(m.get_term(), m.get_from());
self.handle_heartbeat(m);
}
MessageType::MsgSnapshot => {
debug_assert_eq!(self.term, m.get_term());
self.become_follower(m.get_term(), m.get_from());
self.handle_snapshot(m);
}
MessageType::MsgRequestPreVoteResponse | MessageType::MsgRequestVoteResponse => {
if (self.state == StateRole::PreCandidate
&& m.get_msg_type() != MessageType::MsgRequestPreVoteResponse)
|| (self.state == StateRole::Candidate
&& m.get_msg_type() != MessageType::MsgRequestVoteResponse)
{
return Ok(());
}
let acceptance = !m.get_reject();
let msg_type = m.get_msg_type();
let from_id = m.get_from();
info!(
"{} received {:?}{} from {} at term {}",
self.id,
msg_type,
if !acceptance { " rejection" } else { "" },
from_id,
self.term
);
self.register_vote(from_id, acceptance);
match self.prs().candidacy_status(&self.votes) {
CandidacyStatus::Elected => {
if self.state == StateRole::PreCandidate {
self.campaign(CAMPAIGN_ELECTION);
} else {
self.become_leader();
self.bcast_append();
}
}
CandidacyStatus::Ineligible => {
let term = self.term;
self.become_follower(term, INVALID_ID);
}
CandidacyStatus::Eligible => (),
};
}
MessageType::MsgTimeoutNow => debug!(
"{} [term {} state {:?}] ignored MsgTimeoutNow from {}",
self.tag,
self.term,
self.state,
m.get_from()
),
_ => {}
}
Ok(())
}
fn step_follower(&mut self, mut m: Message) -> Result<()> {
match m.get_msg_type() {
MessageType::MsgPropose => {
if self.leader_id == INVALID_ID {
info!(
"{} no leader at term {}; dropping proposal",
self.tag, self.term
);
return Err(Error::ProposalDropped);
}
m.set_to(self.leader_id);
self.send(m);
}
MessageType::MsgAppend => {
self.election_elapsed = 0;
self.leader_id = m.get_from();
self.handle_append_entries(&m);
}
MessageType::MsgHeartbeat => {
self.election_elapsed = 0;
self.leader_id = m.get_from();
self.handle_heartbeat(m);
}
MessageType::MsgSnapshot => {
self.election_elapsed = 0;
self.leader_id = m.get_from();
self.handle_snapshot(m);
}
MessageType::MsgTransferLeader => {
if self.leader_id == INVALID_ID {
info!(
"{} no leader at term {}; dropping leader transfer msg",
self.tag, self.term
);
return Ok(());
}
m.set_to(self.leader_id);
self.send(m);
}
MessageType::MsgTimeoutNow => {
if self.promotable() {
info!(
"{} [term {}] received MsgTimeoutNow from {} and starts an election to \
get leadership.",
self.tag,
self.term,
m.get_from()
);
self.campaign(CAMPAIGN_TRANSFER);
} else {
info!(
"{} received MsgTimeoutNow from {} but is not promotable",
self.tag,
m.get_from()
);
}
}
MessageType::MsgReadIndex => {
if self.leader_id == INVALID_ID {
info!(
"{} no leader at term {}; dropping index reading msg",
self.tag, self.term
);
return Ok(());
}
m.set_to(self.leader_id);
self.send(m);
}
MessageType::MsgReadIndexResp => {
if m.get_entries().len() != 1 {
error!(
"{} invalid format of MsgReadIndexResp from {}, entries count: {}",
self.tag,
m.get_from(),
m.get_entries().len()
);
return Ok(());
}
let rs = ReadState {
index: m.get_index(),
request_ctx: m.take_entries()[0].take_data(),
};
self.read_states.push(rs);
}
_ => {}
}
Ok(())
}
pub fn handle_append_entries(&mut self, m: &Message) {
if m.get_index() < self.raft_log.committed {
debug!("{} Got message with lower index than committed.", self.tag);
let mut to_send = Message::new();
to_send.set_to(m.get_from());
to_send.set_msg_type(MessageType::MsgAppendResponse);
to_send.set_index(self.raft_log.committed);
self.send(to_send);
return;
}
let mut to_send = Message::new();
to_send.set_to(m.get_from());
to_send.set_msg_type(MessageType::MsgAppendResponse);
match self.raft_log.maybe_append(
m.get_index(),
m.get_log_term(),
m.get_commit(),
m.get_entries(),
) {
Some(mlast_index) => {
to_send.set_index(mlast_index);
self.send(to_send);
}
None => {
debug!(
"{} [logterm: {:?}, index: {}] rejected msgApp [logterm: {}, index: {}] \
from {}",
self.tag,
self.raft_log.term(m.get_index()),
m.get_index(),
m.get_log_term(),
m.get_index(),
m.get_from()
);
to_send.set_index(m.get_index());
to_send.set_reject(true);
to_send.set_reject_hint(self.raft_log.last_index());
self.send(to_send);
}
}
}
pub fn handle_heartbeat(&mut self, mut m: Message) {
self.raft_log.commit_to(m.get_commit());
let mut to_send = Message::new();
to_send.set_to(m.get_from());
to_send.set_msg_type(MessageType::MsgHeartbeatResponse);
to_send.set_context(m.take_context());
self.send(to_send);
}
fn handle_snapshot(&mut self, mut m: Message) {
let (sindex, sterm) = (
m.get_snapshot().get_metadata().get_index(),
m.get_snapshot().get_metadata().get_term(),
);
if self.restore(m.take_snapshot()) {
info!(
"{} [commit: {}, term: {}] restored snapshot [index: {}, term: {}]",
self.tag, self.term, self.raft_log.committed, sindex, sterm
);
let mut to_send = Message::new();
to_send.set_to(m.get_from());
to_send.set_msg_type(MessageType::MsgAppendResponse);
to_send.set_index(self.raft_log.last_index());
self.send(to_send);
} else {
info!(
"{} [commit: {}] ignored snapshot [index: {}, term: {}]",
self.tag, self.raft_log.committed, sindex, sterm
);
let mut to_send = Message::new();
to_send.set_to(m.get_from());
to_send.set_msg_type(MessageType::MsgAppendResponse);
to_send.set_index(self.raft_log.committed);
self.send(to_send);
}
}
fn restore_raft(&mut self, snap: &Snapshot) -> Option<bool> {
let meta = snap.get_metadata();
if self.raft_log.match_term(meta.get_index(), meta.get_term()) {
info!(
"{} [commit: {}, lastindex: {}, lastterm: {}] fast-forwarded commit to \
snapshot [index: {}, term: {}]",
self.tag,
self.raft_log.committed,
self.raft_log.last_index(),
self.raft_log.last_term(),
meta.get_index(),
meta.get_term()
);
self.raft_log.commit_to(meta.get_index());
return Some(false);
}
if self.prs().iter().len() != 0 && !self.is_learner {
for &id in meta.get_conf_state().get_learners() {
if id == self.id {
error!(
"{} can't become learner when restores snapshot [index: {}, term: {}]",
self.tag,
meta.get_index(),
meta.get_term(),
);
return Some(false);
}
}
}
info!(
"{} [commit: {}, lastindex: {}, lastterm: {}] starts to restore snapshot \
[index: {}, term: {}]",
self.tag,
self.raft_log.committed,
self.raft_log.last_index(),
self.raft_log.last_term(),
meta.get_index(),
meta.get_term()
);
let next_idx = self.raft_log.last_index() + 1;
let mut prs = ProgressSet::restore_snapmeta(meta, next_idx, self.max_inflight);
prs.get_mut(self.id).unwrap().matched = next_idx - 1;
if self.is_learner && prs.configuration().voters().contains(&self.id) {
self.is_learner = false;
}
self.prs = Some(prs);
if meta.get_pending_membership_change_index() > 0 {
let cs = meta.get_pending_membership_change().clone();
let mut conf_change = ConfChange::new();
conf_change.set_change_type(ConfChangeType::BeginMembershipChange);
conf_change.set_configuration(cs);
conf_change.set_start_index(meta.get_pending_membership_change_index());
self.pending_membership_change = Some(conf_change);
}
None
}
pub fn restore(&mut self, snap: Snapshot) -> bool {
if snap.get_metadata().get_index() < self.raft_log.committed {
return false;
}
if let Some(b) = self.restore_raft(&snap) {
return b;
}
self.raft_log.restore(snap);
true
}
#[inline]
pub fn has_pending_conf(&self) -> bool {
self.pending_conf_index > self.raft_log.applied || self.pending_membership_change.is_some()
}
pub fn should_bcast_commit(&self) -> bool {
!self.skip_bcast_commit || self.has_pending_conf()
}
pub fn promotable(&self) -> bool {
self.prs().voter_ids().contains(&self.id)
}
pub fn propose_membership_change(&mut self, config: impl Into<Configuration>) -> Result<()> {
if self.state != StateRole::Leader {
return Err(Error::InvalidState(self.state));
}
let config = config.into();
config.valid()?;
debug!(
"Replicating SetNodes with voters ({:?}), learners ({:?}).",
config.voters(),
config.learners()
);
let destination_index = self.raft_log.last_index() + 1;
let mut conf_change = ConfChange::new();
conf_change.set_change_type(ConfChangeType::BeginMembershipChange);
conf_change.set_configuration(config.into());
conf_change.set_start_index(destination_index);
let data = protobuf::Message::write_to_bytes(&conf_change)?;
let mut entry = Entry::new();
entry.set_entry_type(EntryType::EntryConfChange);
entry.set_data(data);
let mut message = Message::new();
message.set_msg_type(MessageType::MsgPropose);
message.set_from(self.id);
message.set_index(destination_index);
message.set_entries(RepeatedField::from_vec(vec![entry]));
self.step(message)?;
Ok(())
}
fn add_voter_or_learner(&mut self, id: u64, learner: bool) -> Result<()> {
debug!(
"Adding node (learner: {}) with ID {} to peers.",
learner, id
);
let result = if learner {
let progress = Progress::new(self.raft_log.last_index() + 1, self.max_inflight);
self.mut_prs().insert_learner(id, progress)
} else if self.prs().learner_ids().contains(&id) {
self.mut_prs().promote_learner(id)
} else {
let progress = Progress::new(self.raft_log.last_index() + 1, self.max_inflight);
self.mut_prs().insert_voter(id, progress)
};
if let Err(e) = result {
error!("{}", e);
return Err(e);
}
if self.id == id {
self.is_learner = learner
};
self.mut_prs().get_mut(id).unwrap().recent_active = true;
result
}
pub fn add_node(&mut self, id: u64) -> Result<()> {
self.add_voter_or_learner(id, false)
}
pub fn add_learner(&mut self, id: u64) -> Result<()> {
self.add_voter_or_learner(id, true)
}
pub fn remove_node(&mut self, id: u64) -> Result<()> {
self.mut_prs().remove(id)?;
if self.prs().voter_ids().is_empty() && self.prs().learner_ids().is_empty() {
return Ok(());
}
if self.maybe_commit() {
self.bcast_append();
}
if self.state == StateRole::Leader && self.lead_transferee == Some(id) {
self.abort_leader_transfer();
}
Ok(())
}
pub fn set_progress(&mut self, id: u64, matched: u64, next_idx: u64, is_learner: bool) {
let mut p = Progress::new(next_idx, self.max_inflight);
p.matched = matched;
if is_learner {
if let Err(e) = self.mut_prs().insert_learner(id, p) {
panic!("{}", e);
}
} else if let Err(e) = self.mut_prs().insert_voter(id, p) {
panic!("{}", e);
}
}
pub fn take_prs(&mut self) -> ProgressSet {
self.prs.take().unwrap()
}
pub fn set_prs(&mut self, prs: ProgressSet) {
self.prs = Some(prs);
}
pub fn prs(&self) -> &ProgressSet {
self.prs.as_ref().unwrap()
}
pub fn mut_prs(&mut self) -> &mut ProgressSet {
self.prs.as_mut().unwrap()
}
pub fn load_state(&mut self, hs: &HardState) {
if hs.get_commit() < self.raft_log.committed || hs.get_commit() > self.raft_log.last_index()
{
panic!(
"{} hs.commit {} is out of range [{}, {}]",
self.tag,
hs.get_commit(),
self.raft_log.committed,
self.raft_log.last_index()
)
}
self.raft_log.committed = hs.get_commit();
self.term = hs.get_term();
self.vote = hs.get_vote();
}
pub fn pass_election_timeout(&self) -> bool {
self.election_elapsed >= self.randomized_election_timeout
}
pub fn reset_randomized_election_timeout(&mut self) {
let prev_timeout = self.randomized_election_timeout;
let timeout =
rand::thread_rng().gen_range(self.min_election_timeout, self.max_election_timeout);
debug!(
"{} reset election timeout {} -> {} at {}",
self.tag, prev_timeout, timeout, self.election_elapsed
);
self.randomized_election_timeout = timeout;
}
fn check_quorum_active(&mut self) -> bool {
let self_id = self.id;
self.mut_prs().quorum_recently_active(self_id)
}
pub fn send_timeout_now(&mut self, to: u64) {
let msg = new_message(to, MessageType::MsgTimeoutNow, None);
self.send(msg);
}
pub fn abort_leader_transfer(&mut self) {
self.lead_transferee = None;
}
pub fn is_in_membership_change(&self) -> bool {
self.prs().is_in_membership_change()
}
}