raft

Documentation

Index

• Constants
• func Debug(topic logTopic, format string, a ...interface{})
• func DebugAfterReceiveAppendEntries(rf *Raft, args *AppendEntriesArgs, reply *AppendEntriesReply)
• func DebugELT(s, term int)
• func DebugGetInfo(rf *Raft)
• func DebugGetVote(s1, s2, term int)
• func DebugGrantVote(s1, s2, term int)
• func DebugNewCommand(rf *Raft)
• func DebugReceiveAppendEntries(rf *Raft, entry *AppendEntriesArgs)
• func DebugReceiveHB(s1, s2, term int)
• func DebugResetELT(rf *Raft)
• func DebugResetHBT(rf *Raft, idx int)
• func DebugToFollower(rf *Raft, new_term int)
• func DebugToLeader(s, term, num int)
• func GetRandomExpireTime(left, right int32) time.Time
• func SerilizeState(rf *Raft) []byte
• type AppendEntriesArgs
• type AppendEntriesReply
• type ApplyMsg
• type InstallSnapshotArgs
• type InstallSnapshotReply
• type LogEntry
• type Persister
  • func MakePersister() *Persister
  • func (ps *Persister) Copy() *Persister
  • func (ps *Persister) RaftStateSize() int
  • func (ps *Persister) ReadRaftState() []byte
  • func (ps *Persister) ReadSnapshot() []byte
  • func (ps *Persister) SaveRaftState(state []byte)
  • func (ps *Persister) SaveStateAndSnapshot(state []byte, snapshot []byte)
  • func (ps *Persister) SnapshotSize() int
• type Raft
  • func Make(peers []*labrpc.ClientEnd, me int, persister *Persister, applyCh chan ApplyMsg) *Raft
  • func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply)
  • func (rf *Raft) Applier(applyCh chan ApplyMsg)
  • func (rf *Raft) CallAppendEntries(idx int, term int, me int, prevLogIndex int, prevLogTerm int, logs []LogEntry, ...)
  • func (rf *Raft) CallForVote(idx, term, lastLogIndex, lastLogTerm int)
  • func (rf *Raft) CallInstallSnapshot(idx, term, me, lastIncludedIndex, lastIncludedTerm int, data []byte)
  • func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool
  • func (rf *Raft) GetState() (int, bool)
  • func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply)
  • func (rf *Raft) Kill()
  • func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply)
  • func (rf *Raft) ResetAppendTimer(idx int, imme bool)
  • func (rf *Raft) ResetElectionTimer()
  • func (rf *Raft) Snapshot(index int, snapshot []byte)
  • func (rf *Raft) Start(command interface{}) (int, int, bool)
• type RequestVoteArgs
• type RequestVoteReply

Constants

const (
	LEADER    = 0
	CANDIDATE = 1
	FOLLOWER  = 2
)

const (
	ELECTION_TIMER_RESOLUTION = 5 //每5毫秒检查一次计时器是否已过期
	// 投票过期时间范围（毫秒）
	ELECTION_EXPIRE_LEFT  = 200
	ELECTION_EXPIRE_RIGHT = 400
	// heartbeat time (millsecond)
	APPEND_EXPIRE_TIME      = 100
	APPEND_TIMER_RESOLUTION = 2
)

const SnapShotInterval = 10

Functions

func Debug

func Debug(topic logTopic, format string, a ...interface{})

func DebugAfterReceiveAppendEntries

func DebugAfterReceiveAppendEntries(rf *Raft, args *AppendEntriesArgs, reply *AppendEntriesReply)

func DebugELT

func DebugELT(s, term int)

election timeout

func DebugGetInfo

func DebugGetInfo(rf *Raft)

func DebugGetVote

func DebugGetVote(s1, s2, term int)

receive vote

func DebugGrantVote

func DebugGrantVote(s1, s2, term int)

send vote

func DebugNewCommand

func DebugNewCommand(rf *Raft)

func DebugReceiveAppendEntries

func DebugReceiveAppendEntries(rf *Raft, entry *AppendEntriesArgs)

func DebugReceiveHB

func DebugReceiveHB(s1, s2, term int)

func DebugResetELT

func DebugResetELT(rf *Raft)

func DebugResetHBT

func DebugResetHBT(rf *Raft, idx int)

func DebugToFollower

func DebugToFollower(rf *Raft, new_term int)

become to follower

func DebugToLeader

func DebugToLeader(s, term, num int)

become to leader

func GetRandomExpireTime

func GetRandomExpireTime(left, right int32) time.Time

get a random expire time range [cur_time + left, cur_time + right]

func SerilizeState

func SerilizeState(rf *Raft) []byte

Types

type AppendEntriesArgs

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []LogEntry
	LeaderCommit int
}

type AppendEntriesReply

type AppendEntriesReply struct {
	Term          int
	Success       bool
	ConflictTerm  int
	ConflictIndex int
}

type ApplyMsg

type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type InstallSnapshotArgs

type InstallSnapshotArgs struct {
	Term              int
	LeaderId          int
	LastIncludedIndex int
	LastIncludedTerm  int
	Snapshot          []byte
}

type InstallSnapshotReply

type InstallSnapshotReply struct {
	Term int
}

type LogEntry

type LogEntry struct {
	Term    int
	Index   int
	Command interface{}
}

type Persister

type Persister struct {
	// contains filtered or unexported fields
}

func MakePersister

func MakePersister() *Persister

func (*Persister) Copy

func (ps *Persister) Copy() *Persister

func (*Persister) RaftStateSize

func (ps *Persister) RaftStateSize() int

func (*Persister) ReadRaftState

func (ps *Persister) ReadRaftState() []byte

func (*Persister) ReadSnapshot

func (ps *Persister) ReadSnapshot() []byte

func (*Persister) SaveRaftState

func (ps *Persister) SaveRaftState(state []byte)

func (*Persister) SaveStateAndSnapshot

func (ps *Persister) SaveStateAndSnapshot(state []byte, snapshot []byte)

Save both Raft state and K/V snapshot as a single atomic action, to help avoid them getting out of sync.

func (*Persister) SnapshotSize

func (ps *Persister) SnapshotSize() int

type Raft

type Raft struct {
	ElectionExpireTime time.Time   // election 过期 time
	AppendExpireTime   []time.Time // next send append time??????????????????????????????????????语法疑惑
	// contains filtered or unexported fields
}

func Make

func Make(peers []*labrpc.ClientEnd, me int, persister *Persister, applyCh chan ApplyMsg) *Raft

服务或测试程序想要创建一个 Raft 服务器。所有 Raft 服务器（包括此服务器）的端口都在 peers[] 中。此服务器的端口是 peers[me]。 所有服务器的 peers[] 数组顺序相同。 persister 是服务器保存其持久状态的位置，并且如果有的话，最初还保存了最近保存的状态。applyCh 是一个通道， 测试程序或服务期望 Raft 发送 ApplyMsg 消息到其中。Make() 必须快速返回，因此应为任何长时间运行的工作启动 goroutine。

func (*Raft) AppendEntries

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply)

func (*Raft) Applier

func (rf *Raft) Applier(applyCh chan ApplyMsg)

异步将提交发送到 ApplyCh，这是一个生产者和消费者模型。 AppendEntries 的返回结果修改了 commitIdx，表示将元素推送到队列中。 这是一个消费者，它将 lastApplied 索引添加到消费中。

func (*Raft) CallAppendEntries

func (rf *Raft) CallAppendEntries(idx int, term int, me int, prevLogIndex int, prevLogTerm int, logs []LogEntry, leaderCommit int)

func (*Raft) CallForVote

func (rf *Raft) CallForVote(idx, term, lastLogIndex, lastLogTerm int)

func (*Raft) CallInstallSnapshot

func (rf *Raft) CallInstallSnapshot(idx, term, me, lastIncludedIndex, lastIncludedTerm int, data []byte)

func (*Raft) CondInstallSnapshot

func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool

A service wants to switch to snapshot. Only do so if Raft hasn't have more recent info since it communicate the snapshot on applyCh.

func (*Raft) GetState

func (rf *Raft) GetState() (int, bool)

返回currentTerm和此服务器是否认为自己是领导者。

func (*Raft) InstallSnapshot

func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply)

func (*Raft) Kill

func (rf *Raft) Kill()

测试程序在每个测试后不会停止 Raft 创建的 goroutine，但它会调用 Kill() 方法。 您的代码可以使用 killed() 方法来检查是否已调用 Kill()。使用原子变量可以避免使用锁。

问题在于长时间运行的 goroutine 会使用内存并可能占用 CPU 时间，导致后续测试失败并生成混乱的调试输出。 任何具有长时间运行循环的 goroutine 应该调用 killed() 来检查是否应该停止。

func (*Raft) RequestVote

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply)

example RequestVote RPC handler.

func (*Raft) ResetAppendTimer

func (rf *Raft) ResetAppendTimer(idx int, imme bool)

reset heartBeat, imme mean whether send immediately

func (*Raft) ResetElectionTimer

func (rf *Raft) ResetElectionTimer()

reset Election

func (*Raft) Snapshot

func (rf *Raft) Snapshot(index int, snapshot []byte)

the service says it has created a snapshot that has all info up to and including index. this means the service no longer needs the log through (and including) that index. Raft should now trim its log as much as possible.

func (*Raft) Start

func (rf *Raft) Start(command interface{}) (int, int, bool)

这个函数没看懂？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？？

type RequestVoteArgs

type RequestVoteArgs struct {
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

example RequestVote RPC arguments structure. field names must start with capital(大写) letters!

type RequestVoteReply

type RequestVoteReply struct {
	Term        int
	VoteGranted bool
}

example RequestVote RPC reply structure. field names must start with capital letters!