________
           --|Snapshot|                 timeout
           |  --------                  ______
recover    |       ^                   |      |
snapshot / |       |snapshot           |      |
higher     |       |                   v      |     recv majority votes
term       |    --------    timeout    -----------                        -----------
           |-> |Follower| ----------> | Candidate |--------------------> |  Leader   |
                --------               -----------                        -----------
                   ^          higher term/ |                         higher term |
                   |            new leader |                                     |
                   |_______________________|____________________________________ |

var commandTypes map[string]Command // 存储 command

func newCommand(name string, data []byte) (Command, error)  // 新建 command
func RegisterCommand(command Command)                       // 注册 command 到 commandTypes

// log_entry是客户端发起的command存储在日志文件中的内容
type LogEntry struct {
	pb       *protobuf.LogEntry // proto 编码的 log_entry
	Position int64              // position in the log file
	log      *Log               // 预写日志
	event    *ev                // 日志事件
}

func (*LogEntry).Command() []byte
func (*LogEntry).CommandName() string
func (*LogEntry).Decode(r io.Reader) (int, error) // Decode from log file
func (*LogEntry).Encode(w io.Writer) (int, error) // Encode into log file
func (*LogEntry).Index() uint64
func (*LogEntry).Term() uint64

// log 持久存储的日志项的集合。
type Log struct {
	ApplyFunc   func(*LogEntry, Command) (interface{}, error) // 日志被应用至状态机的方法
	file        *os.File
	path        string
	entries     []*LogEntry // 内存日志项缓存
	commitIndex uint64      // 日志提交点，小于该提交点的日志均已经被应用至状态机
	mutex       sync.RWMutex
	startIndex  uint64  // 缓存中第一个日志项索引
	startTerm   uint64
	initialized bool    // Log 是否初始化
}

func (*Log).CommitIndex() uint64 // 返回 log_entry 提交点
func (*Log).appendEntries(entries []*protobuf.LogEntry) error // 添加log_entries
func (*Log).appendEntry(entry *LogEntry) error
func (*Log).close()
func (*Log).commitInfo() (index uint64, term uint64) // 返回提交点的 index/term
func (*Log).compact(index uint64, term uint64) error // 将 log 写入文件
func (*Log).containsEntry(index uint64, term uint64) bool // log_entry 是否存在于内存
func (*Log).createEntry(term uint64, command Command, e *ev) (*LogEntry, error) // 新建 log_entry
func (*Log).currentIndex() uint64
func (*Log).currentTerm() uint64
func (*Log).flushCommitIndex()
func (*Log).getEntriesAfter(index uint64, maxLogEntriesPerRequest uint64) ([]*LogEntry, uint64) // 返回给定 index 后的 log_entries，没有返回 nil
func (*Log).getEntry(index uint64) *LogEntry
func (*Log).internalCurrentIndex() uint64 // 获取当前 index (无锁)
func (*Log).isEmpty() bool
func (*Log).lastCommandName() string // 返回最后 log_entry 的 command name
func (*Log).lastInfo() (index uint64, term uint64) // 返回最后 log_entry 的 index/term
func (*Log).nextIndex() uint64 // return next index
func (*Log).open(path string) error // 打开日志文件并读取现有条目，日志可以保持打开状态，并继续向日志末尾追加条目
func (*Log).setCommitIndex(index uint64) error // 设置提交点
func (*Log).sync() error // sync to disk
func (*Log).truncate(index uint64, term uint64) error // 将日志截断为给定的索引和项，这仅在索引处的日志未提交时才有效
func (*Log).updateCommitIndex(index uint64)
func (*Log).writeEntry(entry *LogEntry, w io.Writer) (int64, error) // append log_entry

// The request sent to a server to append entries to the log.
type AppendEntriesRequest struct {
    Term         uint64
    PrevLogIndex uint64
    PrevLogTerm  uint64
    CommitIndex  uint64
    LeaderName   string
    Entries      []*protobuf.LogEntry
}

func (*AppendEntriesRequest).Decode(r io.Reader) (int, error) 
func (*AppendEntriesRequest).Encode(w io.Writer) (int, error) 

type AppendEntriesResponse struct {
    pb     *protobuf.AppendEntriesResponse // response
    peer   string   // 目标节点
    append bool     // 是否追加成功
}

func (*AppendEntriesResponse).CommitIndex() uint64 // 对应节点 commit index
func (*AppendEntriesResponse).Decode(r io.Reader) (int, error) 
func (*AppendEntriesResponse).Encode(w io.Writer) (int, error) 
func (*AppendEntriesResponse).Index() uint64    // 对应节点最后的 index
func (*AppendEntriesResponse).Success() bool    // 是否追加成功
func (*AppendEntriesResponse).Term() uint64     // 对应节点最后的 term

// The request sent to a server to vote for a candidate to become a leader. 
// 发送到服务器的请求，以投票给候选人成为领导者。
type RequestVoteRequest struct {
    peer          *Peer
    Term          uint64
    LastLogIndex  uint64
    LastLogTerm   uint64
    CandidateName string
}

func (*RequestVoteRequest).Decode(r io.Reader) (int, error) 
func (*RequestVoteRequest).Encode(w io.Writer) (int, error) 

type RequestVoteResponse struct {
    peer        *Peer
    Term        uint64
    VoteGranted bool // 是否投票
}

func (*RequestVoteResponse).Decode(r io.Reader) (int, error)
func (*RequestVoteResponse).Encode(w io.Writer) (int, error)

// Snapshot 表示系统当前状态的内存表示
type Snapshot struct {
    LastIndex uint64 `json:"lastIndex"`
    LastTerm  uint64 `json:"lastTerm"`

    // Cluster configuration.
    Peers []*Peer `json:"peers"`
    State []byte  `json:"state"` // state machine
    Path  string  `json:"path"`
}

func (*Snapshot).remove() error // deletes the snapshot file
func (*Snapshot).save() error   // writes the snapshot to file

// 设置接收快照状态请求
type SnapshotRequest struct {
    LeaderName string
    LastIndex  uint64
    LastTerm   uint64
}

func (*SnapshotRequest).Decode(r io.Reader) (int, error)
func (*SnapshotRequest).Encode(w io.Writer) (int, error)

type SnapshotResponse struct {
    Success bool `json:"success"` // 成功后设置 server 状态为 snapshoting，后续继续接收 snapshot
}

func (*SnapshotResponse).Decode(r io.Reader) (int, error)
func (*SnapshotResponse).Encode(w io.Writer) (int, error)

// state machine request
type SnapshotRecoveryRequest struct {
    LeaderName string
    LastIndex  uint64
    LastTerm   uint64
    Peers      []*Peer
    State      []byte // state machine
}

func (*SnapshotRecoveryRequest).Decode(r io.Reader) (int, error)
func (*SnapshotRecoveryRequest).Encode(w io.Writer) (int, error)

// Recover state sent from request. then send response
type SnapshotRecoveryResponse struct {
    Term        uint64
    Success     bool
    CommitIndex uint64
}

func (*SnapshotRecoveryResponse).Decode(r io.Reader) (int, error)
func (*SnapshotRecoveryResponse).Encode(w io.Writer) (int, error)

// StateMachine 是允许 server 保存和恢复状态机的接口
type StateMachine interface {
	Save() ([]byte, error)
	Recovery([]byte) error
}

// Transporter 是允许 server 将请求传输到其他节点的接口。
type Transporter interface {
	SendVoteRequest(server Server, peer *Peer, req *RequestVoteRequest) *RequestVoteResponse
	SendAppendEntriesRequest(server Server, peer *Peer, req *AppendEntriesRequest) *AppendEntriesResponse
	SendSnapshotRequest(server Server, peer *Peer, req *SnapshotRequest) *SnapshotResponse
	SendSnapshotRecoveryRequest(server Server, peer *Peer, req *SnapshotRecoveryRequest) *SnapshotRecoveryResponse
}

// raft 中的事件接口
type Event interface {
	Type() string
	Source() interface{}
	Value() interface{}
	PrevValue() interface{}
}

// 用于定义事件实例
type event struct {
    typ       string        // 事件类型
    source    interface{}   // 分派事件的对象
    value     interface{}   // 事件关联的当前值 
    prevValue interface{}   // 事件相关的前一个值
}

func (*event).PrevValue() interface{}
func (*event).Source() interface{}
func (*event).Type() string
func (*event).Value() interface{}

// 目前定义的事件种类
StateChangeEventType                = "stateChange"
LeaderChangeEventType               = "leaderChange"
TermChangeEventType                 = "termChange"
CommitEventType                     = "commit"
AddPeerEventType                    = "addPeer"
RemovePeerEventType                 = "removePeer"
HeartbeatIntervalEventType          = "heartbeatInterval"
ElectionTimeoutThresholdEventType   = "electionTimeoutThreshold"
HeartbeatEventType                  = "heartbeat"

// EventListener is a function that can receive event notifications.
type EventListener func(Event)

// EventListeners represents a collection of individual listeners.
type eventListeners []EventListener

// eventDispatcher 负责管理指定事件的侦听器，并将事件通知分派给这些侦听器
type eventDispatcher struct {
    sync.RWMutex
    source    interface{}
    listeners map[string]eventListeners
}

func (*eventDispatcher).AddEventListener(typ string, listener EventListener)
func (*eventDispatcher).DispatchEvent(e Event) // 当事件触发时，遍历执行这些侦听器
func (*eventDispatcher).RemoveEventListener(typ string, listener EventListener)

type Peer struct {
    server            *server                           // peer中的某些方法会依赖server的状态，如peer内的appendEntries方法需要获取server的currentTerm
    Name              string  `json:"name"`             // 主机名
    ConnectionString  string  `json:"connectionString"` // peer的ip地址，形式为”ip:port”
    prevLogIndex      uint64                            // 该节点上一个 log_entry index
    stopChan          chan bool                         // stop channel
    heartbeatInterval time.Duration                     // ticker   
    lastActivity      time.Time                         // 记录peer的上次活跃时间
    sync.RWMutex            
}

func (*Peer).LastActivity() time.Time
func (*Peer).clone() *Peer  // Clones the state of the peer. The clone is not attached to a server and the heartbeat timer will not exist.
func (*Peer).flush()    // flushes an AppendEntries RPC or Snapshot RPC
func (*Peer).getPrevLogIndex() uint64
func (*Peer).heartbeat(c chan bool) // Listens to the heartbeat timeout and flushes an AppendEntries RPC.
func (*Peer).sendAppendEntriesRequest(req *AppendEntriesRequest)
func (*Peer).sendSnapshotRecoveryRequest()
func (*Peer).sendSnapshotRequest(req *SnapshotRequest)
func (*Peer).sendVoteRequest(req *RequestVoteRequest, c chan *RequestVoteResponse)
func (*Peer).setHeartbeatInterval(duration time.Duration)
func (*Peer).setLastActivity(now time.Time)
func (*Peer).setPrevLogIndex(value uint64)
func (*Peer).startHeartbeat()   // Starts the peer heartbeat.
func (*Peer).stopHeartbeat(flush bool)  // Stops the peer heartbeat.

// 服务器参与共识协议，可以作为追随者、候选人或领导者。
type Server interface {
	Name() string   // server name          
	Context() interface{}   
	StateMachine() StateMachine // 获取 state machine
	Leader() string // 获取当前共识协议的 leader
	State() string  // 获取当前节点的角色状态
	Path() string   // 存储路径
	LogPath() string    // log Path
	SnapshotPath(lastIndex uint64, lastTerm uint64) string  
	Term() uint64
	CommitIndex() uint64
	VotedFor() string   // 投票给了谁
	MemberCount() int   // 当前共识协议成员数
	QuorumSize() int    // 检索仲裁所需的服务器数量 
	IsLogEmpty() bool
	LogEntries() []*LogEntry
	LastCommandName() string
	GetState() string
	ElectionTimeout() time.Duration
	SetElectionTimeout(duration time.Duration)
	HeartbeatInterval() time.Duration
	SetHeartbeatInterval(duration time.Duration)
	Transporter() Transporter   // 获取 server 直接通信对象
	SetTransporter(t Transporter)
	AppendEntries(req *AppendEntriesRequest) *AppendEntriesResponse
	RequestVote(req *RequestVoteRequest) *RequestVoteResponse
	RequestSnapshot(req *SnapshotRequest) *SnapshotResponse
	SnapshotRecoveryRequest(req *SnapshotRecoveryRequest) *SnapshotRecoveryResponse
	AddPeer(name string, connectiongString string) error
	RemovePeer(name string) error
	Peers() map[string]*Peer
	Init() error
	Start() error
	Stop()
	Running() bool
	Do(command Command) (interface{}, error)    // 客户命令执行入口
	TakeSnapshot() error    // 创建 snapshot
	LoadSnapshot() error    // 加载 snapshot
	AddEventListener(string, EventListener)
	FlushCommitIndex()
} 

type server struct {
    *eventDispatcher

    name        string
    path        string
    state       string // 每个节点总是处于以下状态的一种：follower、candidate、leader
    transporter Transporter
    context     interface{}
    currentTerm uint64

    votedFor   string
    log        *Log
    leader     string
    peers      map[string]*Peer // raft中每个节点需要了解其他节点信息，尤其是leader节点
    mutex      sync.RWMutex
    syncedPeer map[string]bool // 对于leader来说，该成员记录了日志已经被sync到了哪些follower

    stopped           chan bool
    c                 chan *ev // 当前节点的命令通道，所有的命令都通过该channel来传递
    electionTimeout   time.Duration
    heartbeatInterval time.Duration

    snapshot *Snapshot

    // PendingSnapshot is an unfinished snapshot.
    // After the pendingSnapshot is saved to disk,
    // it will be set to snapshot and also will be
    // set to nil.
    //
    // PendingSnapshot 是未完成的快照。
    // pendingSnapshot 保存到磁盘后，
    // 它将被设置为快照，并设置为nil。
    pendingSnapshot *Snapshot

    stateMachine            StateMachine // 压缩、解压 log
    maxLogEntriesPerRequest uint64

    connectionString string // ip:port

    routineGroup sync.WaitGroup
}

func (s *server) Do(command Command) (interface{}, error) {
	return s.send(command)
}

// Sends an event to the event loop to be processed. The function will wait
// until the event is actually processed before returning.
func (s *server) send(value interface{}) (interface{}, error) {
	if !s.Running() {
		return nil, StopError
	}

	event := &ev{target: value, c: make(chan error, 1)}
	select {
	case s.c <- event:
	case <-s.stopped:
		return nil, StopError
	}
	select {
	case <-s.stopped:
		return nil, StopError
	case err := <-event.c:
		return event.returnValue, err
	}
}

func (s *server) leaderLoop() {
    logIndex, _ := s.log.lastInfo()
    ......
    // Begin to collect response from followers
    for s.State() == Leader {
        select {
        case <-s.stopped:
            ......
        case e := <-s.c:
            switch req := e.target.(type) {
            // 代表客户端命令
            case Command:
                s.processCommand(req, e)
                continue
            ......
            }
        }
    }
}

// Processes a command.
func (s *server) processCommand(command Command, e *ev) {
	s.debugln("server.command.process")

	// Create an entry for the command in the log.
	entry, err := s.log.createEntry(s.currentTerm, command, e)

	if err != nil {
		s.debugln("server.command.log.entry.error:", err)
		e.c <- err
		return
	}

	if err := s.log.appendEntry(entry); err != nil {
		s.debugln("server.command.log.error:", err)
		e.c <- err
		return
	}

	s.syncedPeer[s.Name()] = true
	if len(s.peers) == 0 {
		commitIndex := s.log.currentIndex()
		s.log.setCommitIndex(commitIndex)
		s.debugln("commit index ", commitIndex)
	}
}

// Listens to the heartbeat timeout and flushes an AppendEntries RPC.
func (p *Peer) heartbeat(c chan bool) {
    stopChan := p.stopChan
    c <- true
    ticker := time.Tick(p.heartbeatInterval)
 
    for {
        select {
        case flush := <-stopChan:
            if flush {
                // before we can safely remove a node
                // we must flush the remove command to the node first
                p.flush()
                return
            } else {
                return
            }
 
        case <-ticker:
            start := time.Now()
            p.flush()
            duration := time.Now().Sub(start)
            p.server.DispatchEvent(newEvent(HeartbeatEventType, duration, nil))
        }
    }
}
 
func (p *Peer) flush() {
    debugln("peer.heartbeat.flush: ", p.Name)
    prevLogIndex := p.getPrevLogIndex()
    term := p.server.currentTerm
 
    entries, prevLogTerm := p.server.log.getEntriesAfter(prevLogIndex, p.server.maxLogEntriesPerRequest)
 
    if entries != nil {
        p.sendAppendEntriesRequest(newAppendEntriesRequest(term, prevLogIndex, prevLogTerm, p.server.log.CommitIndex(), p.server.name, entries))
    } else {
        p.sendSnapshotRequest(newSnapshotRequest(p.server.name, p.server.snapshot))
    }
}

// Sends an AppendEntries request to the peer through the transport.
func (p *Peer) sendAppendEntriesRequest(req *AppendEntriesRequest) {
    resp := p.server.Transporter().SendAppendEntriesRequest(p.server, p, req)
    if resp == nil {
        p.server.DispatchEvent(newEvent(HeartbeatIntervalEventType, p, nil))
        return
    }
    p.setLastActivity(time.Now())
    // If successful then update the previous log index.
    p.Lock()
    if resp.Success() {
        ......
    }
    ......
    resp.peer = p.Name
    // Send response to server for processing.
    p.server.sendAsync(resp)
}

func (s *server) leaderLoop() {
    logIndex, _ := s.log.lastInfo()
    ......
    // Begin to collect response from followers
    for s.State() == Leader {
        select {
        case e := <-s.c:
            switch req := e.target.(type) {
            case Command:
                s.processCommand(req, e)
                continue
            case *AppendEntriesRequest:
                e.returnValue, _ = s.processAppendEntriesRequest(req)
            case *AppendEntriesResponse:
                s.processAppendEntriesResponse(req)
            case *RequestVoteRequest:
                e.returnValue, _ = s.processRequestVoteRequest(req)
            }
 
            // Callback to event.
            e.c <- err
        }
    }
    s.syncedPeer = nil
}

func (s *server) processAppendEntriesResponse(resp *AppendEntriesResponse) {
    // If we find a higher term then change to a follower and exit.
    if resp.Term() > s.Term() {
        s.updateCurrentTerm(resp.Term(), "")
        return
    }
 
    // panic response if it's not successful.
    if !resp.Success() {
        return
    }
 
    // if one peer successfully append a log from the leader term,
    // we add it to the synced list
    if resp.append == true {
        s.syncedPeer[resp.peer] = true
    }
 
    if len(s.syncedPeer) < s.QuorumSize() {
        return
    }
    // Determine the committed index that a majority has.
    var indices []uint64
    indices = append(indices, s.log.currentIndex())
    for _, peer := range s.peers {
        indices = append(indices, peer.getPrevLogIndex())
    }
    sort.Sort(sort.Reverse(uint64Slice(indices)))
 
    commitIndex := indices[s.QuorumSize()-1]
    committedIndex := s.log.commitIndex
 
    if commitIndex > committedIndex {
        s.log.sync()
        s.log.setCommitIndex(commitIndex)
    }
}

// Updates the commit index and writes entries after that index to the stable storage.
func (l *Log) setCommitIndex(index uint64) error {
    l.mutex.Lock()
    defer l.mutex.Unlock()
 
    // this is not error any more after limited the number of sending entries
    // commit up to what we already have
    if index > l.startIndex+uint64(len(l.entries)) {
        index = l.startIndex + uint64(len(l.entries))
    }
    if index < l.commitIndex {
        return nil
    }
 
    for i := l.commitIndex + 1; i <= index; i++ {
        entryIndex := i - 1 - l.startIndex
        entry := l.entries[entryIndex]
 
        l.commitIndex = entry.Index()
 
        // Decode the command.
        command, err := newCommand(entry.CommandName(), entry.Command())
        if err != nil {
            return err
        }
        returnValue, err := l.ApplyFunc(entry, command)
        if entry.event != nil {
            entry.event.returnValue = returnValue
            entry.event.c <- err
        }
        _, isJoinCommand := command.(JoinCommand)
        if isJoinCommand {
            return nil
        }
    }
    return nil
}

func (s *server) followerLoop() {
    since := time.Now()
    electionTimeout := s.ElectionTimeout()
    timeoutChan := afterBetween(s.ElectionTimeout(), s.ElectionTimeout()*2)
 
    for s.State() == Follower {
        var err error
        update := false
        select {
        ......
        // 超过一定时间未收到请求
        case <-timeoutChan:
            if s.promotable() {
                // 状态变为Candidate
                s.setState(Candidate)
            } else {
                update = true
            }
        }
    }
    ......
}
 
// The main event loop for the server
func (s *server) loop() {
    defer s.debugln("server.loop.end")
 
    state := s.State()
 
    for state != Stopped {
        switch state {
        case Follower:
            s.followerLoop()
        // 状态变为Candidate后，进入candidateLoop
        case Candidate:
            s.candidateLoop()
        case Leader:
            s.leaderLoop()
        case Snapshotting:
            s.snapshotLoop()
        }
        state = s.State()
    }
}

func (s *server) candidateLoop() {
    for s.State() == Candidate {
        if doVote {
            s.currentTerm++
            s.votedFor = s.name
            // 向所有其他节点发起Vote请求
            respChan = make(chan *RequestVoteResponse, len(s.peers))
            for _, peer := range s.peers {
                s.routineGroup.Add(1)
                go func(peer *Peer) {
                    defer s.routineGroup.Done()
                            
                    peer.sendVoteRequest(newRequestVoteRequest(s.currentTerm, s.name, lastLogIndex, lastLogTerm), respChan)
                    }(peer)
                }
                // 自己给自己投一票
                votesGranted = 1
                timeoutChan = afterBetween(s.ElectionTimeout(), s.ElectionTimeout()*2)
                doVote = false
            }
            // 如果多数节点同意我作为Leader，设置新状态
            if votesGranted == s.QuorumSize() {
                s.setState(Leader)
                return
        }
    
        // 等待其他节点的选主请求的响应
        select {
        case <-s.stopped:
            s.setState(Stopped)
            return
    
        case resp := <-respChan:
            if success := s.processVoteResponse(resp); success {
                votesGranted++
            }
        ......
        case <-timeoutChan:
            // 如果再一次超时了，重新发起选主请求
            doVote = true
        }
    }
}

// Processes a "request vote" request.
func (s *server) processRequestVoteRequest(req *RequestVoteRequest) (*RequestVoteResponse, bool) {
    if req.Term < s.Term() {
        return newRequestVoteResponse(s.currentTerm, false), false
    }
    if req.Term > s.Term() {
        s.updateCurrentTerm(req.Term, "")
    } else if s.votedFor != "" && s.votedFor != req.CandidateName {
        return newRequestVoteResponse(s.currentTerm, false), false
    }
 
    lastIndex, lastTerm := s.log.lastInfo()
    if lastIndex > req.LastLogIndex || lastTerm > req.LastLogTerm {
        return newRequestVoteResponse(s.currentTerm, false), false
    }
    s.votedFor = req.CandidateName
    return newRequestVoteResponse(s.currentTerm, true), true
}

func (c *DefaultJoinCommand) Apply(server Server) (interface{}, error) {
    err := server.AddPeer(c.Name, c.ConnectionString)
    return []byte("join"), err
}
 
func (c *DefaultLeaveCommand) Apply(server Server) (interface{}, error) {
    err := server.RemovePeer(c.Name)
    return []byte("leave"), err
}

func (s *server) AddPeer(name string, connectiongString string) error {
    if s.peers[name] != nil {
        return nil
    }
 
    if s.name != name {
        peer := newPeer(s, name, connectiongString, s.heartbeatInterval)
        // 如果是主上新增一个peer，那还需要启动后台协程发送
        if s.State() == Leader {
            peer.startHeartbeat()
        }
        s.peers[peer.Name] = peer
        s.DispatchEvent(newEvent(AddPeerEventType, name, nil))
    }
    // Write the configuration to file.
    s.writeConf()
    return nil
}
 
// Removes a peer from the server.
func (s *server) RemovePeer(name string) error {
    // Skip the Peer if it has the same name as the Server
    if name != s.Name() {
        // Return error if peer doesn't exist.
        peer := s.peers[name]
        if peer == nil {
            return fmt.Errorf("raft: Peer not found: %s", name)
        }
        // 如果是Leader，停止给移除节点的心跳协程
        if s.State() == Leader {
            s.routineGroup.Add(1)
            go func() {
                defer s.routineGroup.Done()
                peer.stopHeartbeat(true)
            }()
        }
        delete(s.peers, name)
        s.DispatchEvent(newEvent(RemovePeerEventType, name, nil))
    }
    // Write the configuration to file.
    s.writeConf()
    return nil
}

func (s *server) TakeSnapshot() error {
    ......
    lastIndex, lastTerm := s.log.commitInfo()
    ......
    path := s.SnapshotPath(lastIndex, lastTerm)
    s.pendingSnapshot = &Snapshot{lastIndex, lastTerm, nil, nil, path}
    // 首先应用保存状态机当前状态
    state, err := s.stateMachine.Save()
    if err != nil {
        return err
    }
 
    // 准备Snapshot状态：包括当前日志的index，当前peer等
    peers := make([]*Peer, 0, len(s.peers)+1)
    for _, peer := range s.peers {
        peers = append(peers, peer.clone())
    }
    s.pendingSnapshot.Peers = peers
    s.pendingSnapshot.State = state
    s.saveSnapshot()
 
    // 最后，回收日志项:s.log.compact()
    // 我们在快照之后保留一些日志条目。  
    // 我们不希望将整个快照发送到稍慢的机器 
    if lastIndex-s.log.startIndex > NumberOfLogEntriesAfterSnapshot {
        compactIndex := lastIndex - NumberOfLogEntriesAfterSnapshot
        compactTerm := s.log.getEntry(compactIndex).Term()
        s.log.compact(compactIndex, compactTerm)
    }
    return nil
}

// compact the log before index (including index)
func (l *Log) compact(index uint64, term uint64) error {
	var entries []*LogEntry

	l.mutex.Lock()
	defer l.mutex.Unlock()

	if index == 0 {
		return nil
	}
	// nothing to compaction
	// the index may be greater than the current index if
	// we just recovery from on snapshot
	if index >= l.internalCurrentIndex() {
		entries = make([]*LogEntry, 0)
	} else {
		// get all log entries after index
		entries = l.entries[index-l.startIndex:]
	}

	// create a new log file and add all the entries
	new_file_path := l.path + ".new"
	file, err := os.OpenFile(new_file_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
	if err != nil {
		return err
	}
	for _, entry := range entries {
		position, _ := l.file.Seek(0, os.SEEK_CUR)
		entry.Position = position

		if _, err = entry.Encode(file); err != nil {
			file.Close()
			os.Remove(new_file_path)
			return err
		}
	}
	file.Sync()

	old_file := l.file

	// rename the new log file
	err = os.Rename(new_file_path, l.path)
	if err != nil {
		file.Close()
		os.Remove(new_file_path)
		return err
	}
	l.file = file

	// close the old log file
	old_file.Close()

	// compaction the in memory log
	l.entries = entries
	l.startIndex = index
	l.startTerm = term
	return nil
}