raft

README

raft

Documentation

Index

• Variables
• type AppendEntriesArg
• type AppendEntriesReply
• type ApplyMsg
  • func NewApplyMsg(index LogEntryIndex, command interface{}) ApplyMsg
• type ApplyReceipt
• type Archive
  • func (a *Archive) LoadSnapshot() ([]byte, error)
  • func (a *Archive) LoadStatus() ([]byte, error)
  • func (a *Archive) StateSize() int
  • func (a *Archive) StoreSnapshot(data []byte) error
  • func (a *Archive) StoreStatus(data []byte) error
• type Config
• type ConnectInfo
• type Connectivity
  • func (c *Connectivity) ListenAndServe(serviceName string, v interface{}, addr string) (err error)
  • func (c *Connectivity) Port() int
• type InstallSnapshotArg
• type InstallSnapshotReply
• type KV
  • func NewKV(peers []Peer, currentIndex NodeIndex, store StableStore, snapshotThreshold int) *KV
  • func (kv *KV) Get(k string) (v string, err error)
  • func (kv *KV) GetDirty(k string) (v string, ok bool)
  • func (kv *KV) Init()
  • func (kv *KV) Load()
  • func (kv *KV) LoadFromData(data []byte)
  • func (kv *KV) Run()
  • func (kv *KV) Set(k, v string) (err error)
  • func (kv *KV) Shutdown()
  • func (kv *KV) Store(index int)
• type Log
• type LogEntry
• type LogEntryIndex
• type LogSnapshot
• type NodeIndex
• type Peer
• type RPCDelegate
  • func (d *RPCDelegate) AppendEntries(arg AppendEntriesArg, reply *AppendEntriesReply) error
  • func (d *RPCDelegate) RequestVote(arg RequestVoteArg, reply *RequestVoteReply) error
• type Raft
  • func NewRaft(peers []Peer, me NodeIndex, store StableStore, applyCh chan ApplyMsg) *Raft
  • func (r *Raft) AppendEntries(arg *AppendEntriesArg, reply *AppendEntriesReply)
  • func (r *Raft) Apply(command interface{}) (receipt ApplyReceipt)
  • func (r *Raft) CheckLeadership(term Term) bool
  • func (r *Raft) DidSnapshot(index int)
  • func (r *Raft) InstallSnapshot(arg *InstallSnapshotArg, reply *InstallSnapshotReply)
  • func (r *Raft) IsLeader() bool
  • func (r *Raft) RequestVote(arg *RequestVoteArg, reply *RequestVoteReply)
  • func (r *Raft) Restore()
  • func (r *Raft) Run()
  • func (r *Raft) Shutdown()
  • func (r *Raft) Store()
• type RequestVoteArg
• type RequestVoteReply
• type Role
• type StableStore
• type Term

Variables

var (
	ErrKVNotLeader = errors.New("kv: current node not a leader")
	ErrKeyNotExist = errors.New("kv: key not exist")
	ErrShutdown    = errors.New("kv: shutdown")
)

var DEBUG = true

var (
	ErrNotLeader = errors.New("kv: current node not a leader")
)

Types

type AppendEntriesArg

type AppendEntriesArg struct {
	Term         Term
	LeaderID     NodeIndex
	PrevLogIndex LogEntryIndex
	PrevLogTerm  LogEntryIndex
	Entries      []LogEntry
	LeaderCommit LogEntryIndex
}

type AppendEntriesReply

type AppendEntriesReply struct {
	CurrentTerm Term
	Success     bool
	// return conflict info to leader
	ConflictTerm  Term
	ConflictIndex LogEntryIndex
}

type ApplyMsg

type ApplyMsg struct {
	Index   LogEntryIndex
	Command interface{}

	UseSnapshot bool   // ignore for lab2; only used in lab3
	Snapshot    []byte // ignore for lab2; only used in lab3
}

func NewApplyMsg

func NewApplyMsg(index LogEntryIndex, command interface{}) ApplyMsg

type ApplyReceipt

type ApplyReceipt struct {
	Term  Term
	Index LogEntryIndex
	Err   error
}

type Archive

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

simulated StableStore todo: replace with a local db

func (*Archive) LoadSnapshot

func (a *Archive) LoadSnapshot() ([]byte, error)

func (*Archive) LoadStatus

func (a *Archive) LoadStatus() ([]byte, error)

func (*Archive) StateSize

func (a *Archive) StateSize() int

func (*Archive) StoreSnapshot

func (a *Archive) StoreSnapshot(data []byte) error

func (*Archive) StoreStatus

func (a *Archive) StoreStatus(data []byte) error

type Config

type Config struct {
	ElectionTimeout  time.Duration
	HeartbeatTimeout time.Duration
}

Config stores config for raft

type ConnectInfo

type ConnectInfo struct {
	Addrs []string
	Index int
}

type Connectivity

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

func (*Connectivity) ListenAndServe

func (c *Connectivity) ListenAndServe(serviceName string, v interface{}, addr string) (err error)

func (*Connectivity) Port

func (c *Connectivity) Port() int

type InstallSnapshotArg

type InstallSnapshotArg struct {
	Term              Term
	LeaderID          NodeIndex
	LastIncludedIndex LogEntryIndex
	LastIncludedTerm  Term
	Snapshot          []byte
}

type InstallSnapshotReply

type InstallSnapshotReply struct {
	CurrentTerm Term
}

type KV

type KV struct {
	Raft *Raft
	// contains filtered or unexported fields
}

func NewKV

func NewKV(peers []Peer, currentIndex NodeIndex, store StableStore, snapshotThreshold int) *KV

func (*KV) Get

func (kv *KV) Get(k string) (v string, err error)

func (*KV) GetDirty

func (kv *KV) GetDirty(k string) (v string, ok bool)

func (*KV) Init

func (kv *KV) Init()

func (*KV) Load

func (kv *KV) Load()

func (*KV) LoadFromData

func (kv *KV) LoadFromData(data []byte)

func (*KV) Run

func (kv *KV) Run()

func (*KV) Set

func (kv *KV) Set(k, v string) (err error)

func (*KV) Shutdown

func (kv *KV) Shutdown()

func (*KV) Store

func (kv *KV) Store(index int)

type Log

type Log struct {
	Logs []LogEntry

	CommitIndex LogEntryIndex
	LastApplied LogEntryIndex
	Snapshot    LogSnapshot
	// contains filtered or unexported fields
}

type LogEntry

type LogEntry struct {
	Term    int
	Command interface{}
}

type LogEntryIndex

type LogEntryIndex = int

type LogSnapshot

type LogSnapshot struct {
	LastIncludedIndex LogEntryIndex
	LastIncludedTerm  Term
}

type NodeIndex

type NodeIndex = int

type aliases help understand variables

type Peer

type Peer interface {
	Call(method string, arg interface{}, reply interface{}) error
}

Peer is a general interface for RPC client endpoints It can be used to replace the original RPC with a testable one

type RPCDelegate

type RPCDelegate struct {
	Raft *Raft
}

RPCDelegate delegates Raft objects for RPC service

func (*RPCDelegate) AppendEntries

func (d *RPCDelegate) AppendEntries(arg AppendEntriesArg, reply *AppendEntriesReply) error

func (*RPCDelegate) RequestVote

func (d *RPCDelegate) RequestVote(arg RequestVoteArg, reply *RequestVoteReply) error

type Raft

type Raft struct {
	Config

	StableStore

	Id          NodeIndex
	Role        Role
	CurrentTerm Term

	Log
	// contains filtered or unexported fields
}

Raft

func NewRaft

func NewRaft(peers []Peer, me NodeIndex, store StableStore, applyCh chan ApplyMsg) *Raft

func (*Raft) AppendEntries

func (r *Raft) AppendEntries(arg *AppendEntriesArg, reply *AppendEntriesReply)

func (*Raft) Apply

func (r *Raft) Apply(command interface{}) (receipt ApplyReceipt)

func (*Raft) CheckLeadership

func (r *Raft) CheckLeadership(term Term) bool

CheckLeadership help outer service's async operations check leadership.

func (*Raft) DidSnapshot

func (r *Raft) DidSnapshot(index int)

after outer service perform a snapshot

func (*Raft) InstallSnapshot

func (r *Raft) InstallSnapshot(arg *InstallSnapshotArg, reply *InstallSnapshotReply)

func (*Raft) IsLeader

func (r *Raft) IsLeader() bool

IsLeader is concurrent safe to user by outer service.

func (*Raft) RequestVote

func (r *Raft) RequestVote(arg *RequestVoteArg, reply *RequestVoteReply)

func (*Raft) Restore

func (r *Raft) Restore()

func (*Raft) Run

func (r *Raft) Run()

func (*Raft) Shutdown

func (r *Raft) Shutdown()

func (*Raft) Store

func (r *Raft) Store()

type RequestVoteArg

type RequestVoteArg struct {
	Term         Term
	CandidateID  NodeIndex
	LastLogIndex LogEntryIndex
	LastLogTerm  Term
}

type RequestVoteReply

type RequestVoteReply struct {
	CurrentTerm Term
	VoteGranted bool
}

type Role

type Role = int

const (
	Follower Role
	Candidate
	Leader
)

type StableStore

type StableStore interface {
	LoadStatus() ([]byte, error)
	LoadSnapshot() ([]byte, error)
	StoreStatus([]byte) error
	StoreSnapshot([]byte) error
	StateSize() int
}

type Term

type Term = int