raft.go
func (r *Raft) startStopReplication() {
	inConfig := make(map[ServerID]bool, len(r.configurations.latest.Servers))
	lastIdx := r.getLastIndex()

	// Start replication goroutines that need starting
	for _, server := range r.configurations.latest.Servers {
		if server.ID == r.localID {
			continue
		}
		inConfig[server.ID] = true
		if _, ok := r.leaderState.replState[server.ID]; !ok {
			r.logger.Info(fmt.Sprintf("Added peer %v, starting replication", server.ID))
			s := &followerReplication{
				peer:        server,
				commitment:  r.leaderState.commitment,
				stopCh:      make(chan uint64, 1),
				triggerCh:   make(chan struct{}, 1),
				currentTerm: r.getCurrentTerm(),
				nextIndex:   lastIdx + 1,
				lastContact: time.Now(),
				notify:      make(map[*verifyFuture]struct{}),
				notifyCh:    make(chan struct{}, 1),
				stepDown:    r.leaderState.stepDown,
			}
			r.leaderState.replState[server.ID] = s
			r.goFunc(func() { r.replicate(s) })
			asyncNotifyCh(s.triggerCh)
		}
	}

	// Stop replication goroutines that need stopping
	for serverID, repl := range r.leaderState.replState {
		if inConfig[serverID] {
			continue
		}
		// Replicate up to lastIdx and stop
		r.logger.Info(fmt.Sprintf("Removed peer %v, stopping replication after %v", serverID, lastIdx))
		repl.stopCh <- lastIdx
		close(repl.stopCh)
		delete(r.leaderState.replState, serverID)
	}
}
func (r *Raft) startStopReplication() {
	inConfig := make(map[ServerID]bool, len(r.configurations.latest.Servers))
	lastIdx := r.getLastIndex()

	// Start replication goroutines that need starting
	for _, server := range r.configurations.latest.Servers {
		if server.ID == r.localID {
			continue
		}
		inConfig[server.ID] = true
		if _, ok := r.leaderState.replState[server.ID]; !ok {
			r.logger.Info(fmt.Sprintf("Added peer %v, starting replication", server.ID))
			s := &followerReplication{
				peer:        server,
				commitment:  r.leaderState.commitment,
				stopCh:      make(chan uint64, 1),
				triggerCh:   make(chan struct{}, 1),
				currentTerm: r.getCurrentTerm(),
				nextIndex:   lastIdx + 1,
				lastContact: time.Now(),
				notify:      make(map[*verifyFuture]struct{}),
				notifyCh:    make(chan struct{}, 1),
				stepDown:    r.leaderState.stepDown,
			}
			r.leaderState.replState[server.ID] = s
			r.goFunc(func() { r.replicate(s) })
			asyncNotifyCh(s.triggerCh)
		}
	}

	// Stop replication goroutines that need stopping
	for serverID, repl := range r.leaderState.replState {
		if inConfig[serverID] {
			continue
		}
		// Replicate up to lastIdx and stop
		r.logger.Info(fmt.Sprintf("Removed peer %v, stopping replication after %v", serverID, lastIdx))
		repl.stopCh <- lastIdx
		close(repl.stopCh)
		delete(r.leaderState.replState, serverID)
	}
}
func ( r * Raft ) startStopReplication ( ) { inConfig := make ( map [ ServerID ] bool , len ( r . configurations . latest . Servers ) ) lastIdx := r . getLastIndex ( ) // Start replication goroutines that need starting for _ , server := range r . configurations . latest . Servers { if server . ID == r . localID { continue } inConfig [ server . ID ] = true if _ , ok := r . leaderState . replState [ server . ID ] ; ! ok { r . logger . Info ( fmt . Sprintf ( " " , server . ID ) ) s := & followerReplication { peer : server , commitment : r . leaderState . commitment , stopCh : make ( chan uint64 , 1 ) , triggerCh : make ( chan struct { } , 1 ) , currentTerm : r . getCurrentTerm ( ) , nextIndex : lastIdx + 1 , lastContact : time . Now ( ) , notify : make ( map [ * verifyFuture ] struct { } ) , notifyCh : make ( chan struct { } , 1 ) , stepDown : r . leaderState . stepDown , } r . leaderState . replState [ server . ID ] = s r . goFunc ( func ( ) { r . replicate ( s ) } ) asyncNotifyCh ( s . triggerCh ) } } // Stop replication goroutines that need stopping for serverID , repl := range r . leaderState . replState { if inConfig [ serverID ] { continue } // Replicate up to lastIdx and stop r . logger . Info ( fmt . Sprintf ( " " , serverID , lastIdx ) ) repl . stopCh <- lastIdx close ( repl . stopCh ) delete ( r . leaderState . replState , serverID ) } }
// startStopReplication will set up state and start asynchronous replication to
// new peers, and stop replication to removed peers. Before removing a peer,
// it'll instruct the replication routines to try to replicate to the current
// index. This must only be called from the main thread.
startStopReplication will set up state and start asynchronous replication to new peers and stop replication to removed peers . Before removing a peer it ll instruct the replication routines to try to replicate to the current index . This must only be called from the main thread .