fn maybe_send_append(&mut self, to: u64, send_if_empty: bool) -> bool {
        debug!(
            "execute maybe_send_append, to: {}, send_if_empty: {}",
            to, send_if_empty
        );
        let pr = self.prs.progress.must_get_mut(&to);
        if pr.is_paused() {
            return false;
        }
        let mut m = Message {
            to,
            ..Default::default()
        };

        // get the last log term because `pr.next` is next index, so `lasted = next - 1`
        let term = self.raft_log.term(pr.next - 1);
        let ents = self.raft_log.entries(pr.next, self.max_msg_size);
        if ents.as_ref().map_or_else(|_| true, |ents| ents.is_empty()) && !send_if_empty {
            warn!(
                "ignore send append from, send_if_empty={}, to={:0x}",
                send_if_empty, m.to
            );
            return false;
        }
        if term.is_err() || ents.is_err() {
            info!("term or ents is invalid, ready to sync handle");
            // send snapshot if we failed to get term or entries
            if !pr.recent_active {
                debug!(
                    "ignore sending snapshot to {:#x} since it is not recently active",
                    to
                );
                return false;
            }

            m.field_type = MsgSnap;
            match self.raft_log.snapshot() {
                Ok(snapshot) => {
                    let metadata = snapshot.get_metadata();
                    let (s_index, s_term) = (metadata.get_index(), metadata.get_term());
                    assert!(s_index > 0, "need non-empty snapshot");
                    m.set_snapshot(snapshot);
                    debug!(
                        "{:#x} [first_index: {}, commit: {}] sent snapshot[index: {}, term: {}] to {:#x} [{:?}]",
                        self.id,
                        self.raft_log.first_index(),
                        self.raft_log.committed,
                        s_index,
                        s_term,
                        to,
                        pr
                    );
                    pr.become_snapshot(s_index);
                    debug!(
                        "{:#x} paused sending replication messages to {:#x} [{}]",
                        self.id, to, pr
                    );
                }
                Err(e)
                if e == RaftLogError::FromStorage(
                    StorageError::SnapshotTemporarilyUnavailable,
                ) =>
                    {
                        debug!(
                            "{:#x} failed to send snapshot to {:#x} because snapshot is temporarily unvalidated",
                            self.id, to
                        );
                        return false;
                    }
                Err(e) => panic!("{:?}", e), // TODO(bdarnell)
            }
        } else {
            let term = term.unwrap();
            let ents = ents.unwrap();
            m.field_type = MsgApp;
            // msg.index is the current node lasted index, so `sub - 1`
            m.index = pr.next - 1;
            // msg.logTerm is the current node log lasted term.
            m.logTerm = term;
            m.entries = RepeatedField::from_vec(ents);
            m.commit = self.raft_log.committed;
            let n = m.entries.len();
            if n != 0 {
                match &pr.state {
                    state::StateType::Replicate => {
                        // optimistically increase the next when in StateReplicate
                        let last = m.entries.last().unwrap().get_Index();
                        pr.optimistic_update(last);
                        pr.inflights.add(last);
                        // debug!("#{:0x} progress at replication, append to it for log entry, {:?}, msg: {:?}", to, pr, m);
                    }
                    state::StateType::Probe => {
                        pr.probe_sent = true;
                    }
                    _ => panic!(
                        "{:#x} is sending append in unhandled state {}",
                        self.id, pr.state
                    ),
                }
            }
        }
        // debug!("send append succeeded, msg: {:#?}", m);
        self.send(m);
        true
    }
// sends an append RPC with new entries to the given peer,
// if necessary. Returns true if a message was sent. The sendIfEmpty
// argument controls whether messages with no entries will be sent.
// ("empty" messages are useful to convey updated Commit indexes, but
// are undesirable when we're sending multiple messages in a batch).
sends an append RPC with new entries to the given peer,
if necessary. Returns true if a message was sent. The sendIfEmpty
argument controls whether messages with no entries will be sent.
("empty" messages are useful to convey updated Commit indexes, but
are undesirable when we're sending multiple messages in a batch).
sends an append RPC with new entries to the given peer if necessary . Returns true if a message was sent . The sendIfEmpty argument controls whether messages with no entries will be sent . ( " empty " messages are useful to convey updated Commit indexes but are undesirable when we ' re sending multiple messages in a batch ) .
fn maybe_send_append(&mut self, to: u64, send_if_empty: bool) -> bool {
        debug!(
            "execute maybe_send_append, to: {}, send_if_empty: {}",
            to, send_if_empty
        );
        let pr = self.prs.progress.must_get_mut(&to);
        if pr.is_paused() {
            return false;
        }
        let mut m = Message {
            to,
            ..Default::default()
        };
        // get the last log term because `pr.next` is next index, so `lasted = next - 1`
        let term = self.raft_log.term(pr.next - 1);
        let ents = self.raft_log.entries(pr.next, self.max_msg_size);
        if ents.as_ref().map_or_else(|_| true, |ents| ents.is_empty()) && !send_if_empty {
            warn!(
                "ignore send append from, send_if_empty={}, to={:0x}",
                send_if_empty, m.to
            );
            return false;
        }
        if term.is_err() || ents.is_err() {
            info!("term or ents is invalid, ready to sync handle");
            // send snapshot if we failed to get term or entries
            if !pr.recent_active {
                debug!(
                    "ignore sending snapshot to {:#x} since it is not recently active",
                    to
                );
                return false;
            }
            m.field_type = MsgSnap;
            match self.raft_log.snapshot() {
                Ok(snapshot) => {
                    let metadata = snapshot.get_metadata();
                    let (s_index, s_term) = (metadata.get_index(), metadata.get_term());
                    assert!(s_index > 0, "need non-empty snapshot");
                    m.set_snapshot(snapshot);
                    debug!(
                        "{:#x} [first_index: {}, commit: {}] sent snapshot[index: {}, term: {}] to {:#x} [{:?}]",
                        self.id,
                        self.raft_log.first_index(),
                        self.raft_log.committed,
                        s_index,
                        s_term,
                        to,
                        pr
                    );
                    pr.become_snapshot(s_index);
                    debug!(
                        "{:#x} paused sending replication messages to {:#x} [{}]",
                        self.id, to, pr
                    );
                }
                Err(e)
                if e == RaftLogError::FromStorage(
                    StorageError::SnapshotTemporarilyUnavailable,
                ) =>
                    {
                        debug!(
                            "{:#x} failed to send snapshot to {:#x} because snapshot is temporarily unvalidated",
                            self.id, to
                        );
                        return false;
                    }
                Err(e) => panic!("{:?}", e), // TODO(bdarnell)
            }
        } else {
            let term = term.unwrap();
            let ents = ents.unwrap();
            m.field_type = MsgApp;
            // msg.index is the current node lasted index, so `sub - 1`
            m.index = pr.next - 1;
            // msg.logTerm is the current node log lasted term.
            m.logTerm = term;
            m.entries = RepeatedField::from_vec(ents);
            m.commit = self.raft_log.committed;
            let n = m.entries.len();
            if n != 0 {
                match &pr.state {
                    state::StateType::Replicate => {
                        // optimistically increase the next when in StateReplicate
                        let last = m.entries.last().unwrap().get_Index();
                        pr.optimistic_update(last);
                        pr.inflights.add(last);
                        // debug!("#{:0x} progress at replication, append to it for log entry, {:?}, msg: {:?}", to, pr, m);
                    }
                    state::StateType::Probe => {
                        pr.probe_sent = true;
                    }
                    _ => panic!(
                        "{:#x} is sending append in unhandled state {}",
                        self.id, pr.state
                    ),
                }
            }
        }
        // debug!("send append succeeded, msg: {:#?}", m);
        self.send(m);
        true
    }
fn maybe_send_append ( & mut self , to : u64 , send_if_empty : bool ) -> bool { debug ! ( " " to send_if_empty ) ; let pr = self . prs . progress . must_get_mut ( & to ) ; if pr . is_paused ( ) { return false ; } let mut m = Message { to , .. Default :: default ( ) } ; // get the last log term because `pr.next` is next index, so `lasted = next - 1` let term = self . raft_log . term ( pr . next - 1 ) ; let ents = self . raft_log . entries ( pr . next , self . max_msg_size ) ; if ents . as_ref ( ) . map_or_else ( | _ | true , | ents | ents . is_empty ( ) ) && ! send_if_empty { warn ! ( " " send_if_empty m to ) ; return false ; } if term . is_err ( ) || ents . is_err ( ) { info ! ( " " ) ; // send snapshot if we failed to get term or entries if ! pr . recent_active { debug ! ( " " to ) ; return false ; } m . field_type = MsgSnap ; match self . raft_log . snapshot ( ) { Ok ( snapshot ) => { let metadata = snapshot . get_metadata ( ) ; let ( s_index , s_term ) = ( metadata . get_index ( ) , metadata . get_term ( ) ) ; assert ! ( s_index 0 " " ) ; m . set_snapshot ( snapshot ) ; debug ! ( " " self id self raft_log first_index ( ) self raft_log committed s_index s_term to pr ) ; pr . become_snapshot ( s_index ) ; debug ! ( " " self id to pr ) ; } Err ( e ) if e == RaftLogError :: FromStorage ( StorageError :: SnapshotTemporarilyUnavailable , ) => { debug ! ( " " self id to ) ; return false ; } Err ( e ) => panic ! ( " " e ) , // TODO(bdarnell) } } else { let term = term . unwrap ( ) ; let ents = ents . unwrap ( ) ; m . field_type = MsgApp ; // msg.index is the current node lasted index, so `sub - 1` m . index = pr . next - 1 ; // msg.logTerm is the current node log lasted term. m . logTerm = term ; m . entries = RepeatedField :: from_vec ( ents ) ; m . commit = self . raft_log . committed ; let n = m . entries . len ( ) ; if n != 0 { match & pr . state { state :: StateType :: Replicate => { // optimistically increase the next when in StateReplicate let last = m . entries . last ( ) . unwrap ( ) . get_Index ( ) ; pr . optimistic_update ( last ) ; pr . inflights . add ( last ) ; // debug!("#{:0x} progress at replication, append to it for log entry, {:?}, msg: {:?}", to, pr, m); } state :: StateType :: Probe => { pr . probe_sent = true ; } _ => panic ! ( " " self id pr state ) , } } } // debug!("send append succeeded, msg: {:#?}", m); self . send ( m ) ; true }
sends an append RPC with new entries to the given peer,
if necessary.
sends an append RPC with new entries to the given peer if necessary .
&mut self
to
u64
send_if_empty
bool