async fn leader_append_deamon(&self) {
        let temp_term;
        {
            let raft = self.inner.lock().unwrap();
            assert!(
                raft.state.is_leader(),
                "Non-leader S{} create a leader append deamon",
                raft.me
            );
            temp_term = raft.state.term;
        }
        loop {
            let (mut rpcs, next_index) = {
                let raft = self.inner.lock().unwrap();
                (raft.send_append_rpc(), raft.next_index.clone())
            };
            let inner = Arc::clone(&self.inner);
            task::spawn(async move {
                // next_index means the first index of log in AppendRPC which send to server peer_id
                while let Some(Ok((peer_id, log_len, res))) = rpcs.next().await {
                    let next_index = next_index[peer_id];
                    let mut raft = inner.lock().unwrap();
                    info!("[Append] S{}(term = {}) get append reply from S{}: {:?}", raft.me, raft.state.term, peer_id, res);
                    if !raft.check_incoming_term(res.term) || raft.state.term != temp_term {
                        break;
                    }
                    if res.success {
                        let last_send_index = next_index + log_len as u64 - 1;
                        raft.update_next_index(peer_id, last_send_index + 1);
                        raft.update_match_index(peer_id, last_send_index);
                    } else {
                        let conflict_info = res.conflict.unwrap();
                        raft.next_index[peer_id] = match conflict_info.reason {
                           ConflictReason::NoEntry => {
                               conflict_info.first_index + 1
                           }
                           ConflictReason::AlreadySnapshot => {
                               conflict_info.first_index + 1
                           }
                           ConflictReason::TermMismatch => {
                               let start = conflict_info.first_index.max(raft.state.last_included_index);
                               let mut new_next_index = conflict_info.first_index;
                               for index in (start..next_index-1).rev() {
                                   if raft.term_at(index) == conflict_info.term {
                                       new_next_index = index + 1;
                                       break;
                                   }
                               }
                               new_next_index
                           }
                        };
                        assert!(raft.next_index[peer_id] > 0);
                        info!("[Append] Leader {} append to {} failed due to log inconsistency, next_index = {}",
                            raft.me, peer_id, raft.next_index[peer_id]);
                    }
                }
            })
            .detach();
            sleep(APPEND_PERIOD).await;
            {
                let raft = self.inner.lock().unwrap();
                if raft.state.term != temp_term {
                    break;
                }
            }
        }
    }
/// used for send heartbeat peridodically
/// Note: supposed to call this every time becoming leader
/// and this coroutine will terminate automatically after downgrading from leader
used for send heartbeat peridodically
Note: supposed to call this every time becoming leader
and this coroutine will terminate automatically after downgrading from leader
used for send heartbeat peridodically Note : supposed to call this every time becoming leader and this coroutine will terminate automatically after downgrading from leader
async fn leader_append_deamon(&self) {
        let temp_term;
        {
            let raft = self.inner.lock().unwrap();
            assert!(
                raft.state.is_leader(),
                "Non-leader S{} create a leader append deamon",
                raft.me
            );
            temp_term = raft.state.term;
        }
        loop {
            let (mut rpcs, next_index) = {
                let raft = self.inner.lock().unwrap();
                (raft.send_append_rpc(), raft.next_index.clone())
            };
            let inner = Arc::clone(&self.inner);
            task::spawn(async move {
                // next_index means the first index of log in AppendRPC which send to server peer_id
                while let Some(Ok((peer_id, log_len, res))) = rpcs.next().await {
                    let next_index = next_index[peer_id];
                    let mut raft = inner.lock().unwrap();
                    info!("[Append] S{}(term = {}) get append reply from S{}: {:?}", raft.me, raft.state.term, peer_id, res);
                    if !raft.check_incoming_term(res.term) || raft.state.term != temp_term {
                        break;
                    }
                    if res.success {
                        let last_send_index = next_index + log_len as u64 - 1;
                        raft.update_next_index(peer_id, last_send_index + 1);
                        raft.update_match_index(peer_id, last_send_index);
                    } else {
                        let conflict_info = res.conflict.unwrap();
                        raft.next_index[peer_id] = match conflict_info.reason {
                           ConflictReason::NoEntry => {
                               conflict_info.first_index + 1
                           }
                           ConflictReason::AlreadySnapshot => {
                               conflict_info.first_index + 1
                           }
                           ConflictReason::TermMismatch => {
                               let start = conflict_info.first_index.max(raft.state.last_included_index);
                               let mut new_next_index = conflict_info.first_index;
                               for index in (start..next_index-1).rev() {
                                   if raft.term_at(index) == conflict_info.term {
                                       new_next_index = index + 1;
                                       break;
                                   }
                               }
                               new_next_index
                           }
                        };
                        assert!(raft.next_index[peer_id] > 0);
                        info!("[Append] Leader {} append to {} failed due to log inconsistency, next_index = {}",
                            raft.me, peer_id, raft.next_index[peer_id]);
                    }
                }
            })
            .detach();
            sleep(APPEND_PERIOD).await;
            {
                let raft = self.inner.lock().unwrap();
                if raft.state.term != temp_term {
                    break;
                }
            }
        }
    }
async fn leader_append_deamon ( & self ) { let temp_term ; { let raft = self . inner . lock ( ) . unwrap ( ) ; assert ! ( raft state is_leader ( ) " " raft me ) ; temp_term = raft . state . term ; } loop { let ( mut rpcs , next_index ) = { let raft = self . inner . lock ( ) . unwrap ( ) ; ( raft . send_append_rpc ( ) , raft . next_index . clone ( ) ) } ; let inner = Arc :: clone ( & self . inner ) ; task :: spawn ( async move { // next_index means the first index of log in AppendRPC which send to server peer_id while let Some ( Ok ( ( peer_id , log_len , res ) ) ) = rpcs . next ( ) . await { let next_index = next_index [ peer_id ] ; let mut raft = inner . lock ( ) . unwrap ( ) ; info ! ( " " raft me raft state term peer_id res ) ; if ! raft . check_incoming_term ( res . term ) || raft . state . term != temp_term { break ; } if res . success { let last_send_index = next_index + log_len as u64 - 1 ; raft . update_next_index ( peer_id , last_send_index + 1 ) ; raft . update_match_index ( peer_id , last_send_index ) ; } else { let conflict_info = res . conflict . unwrap ( ) ; raft . next_index [ peer_id ] = match conflict_info . reason { ConflictReason :: NoEntry => { conflict_info . first_index + 1 } ConflictReason :: AlreadySnapshot => { conflict_info . first_index + 1 } ConflictReason :: TermMismatch => { let start = conflict_info . first_index . max ( raft . state . last_included_index ) ; let mut new_next_index = conflict_info . first_index ; for index in ( start .. next_index - 1 ) . rev ( ) { if raft . term_at ( index ) == conflict_info . term { new_next_index = index + 1 ; break ; } } new_next_index } } ; assert ! ( raft next_index [ peer_id ] 0 ) ; info ! ( " " raft me peer_id raft next_index [ peer_id ] ) ; } } } ) . detach ( ) ; sleep ( APPEND_PERIOD ) . await ; { let raft = self . inner . lock ( ) . unwrap ( ) ; if raft . state . term != temp_term { break ; } } } }
used for send heartbeat peridodically
Note: supposed to call this every time becoming leader
and this coroutine will terminate automatically after downgrading from leader
used for send heartbeat peridodically Note : supposed to call this every time becoming leader and this coroutine will terminate automatically after downgrading from leader
&self
&self