raft

Documentation

Index

• func EncodeRPCMsg(msg RPCMsg) ([]byte, error)
• func Panicf(format string, args ...interface{})
• func RecoverValueString(value interface{}) (msg string)
• func StackTrace(depth int) string
• type IncomingRPCMsg
• type LogEntry
• type LogIndex
• type LogReplayFunc
• type LogStore
  • func NewLogStore(filePath string) *LogStore
  • func (s *LogStore) AppendEntries(entries []LogEntry) error
  • func (s *LogStore) Close()
  • func (s *LogStore) LastIndex() LogIndex
  • func (s *LogStore) LastTerm() Term
  • func (s *LogStore) Open(replayFn LogReplayFunc) error
• type Logger
• type PersistentState
• type PersistentStore
  • func NewPersistentStore(filePath string) *PersistentStore
  • func (s *PersistentStore) Close()
  • func (s *PersistentStore) Open() error
  • func (s *PersistentStore) Read(state *PersistentState) error
  • func (s *PersistentStore) Write(state PersistentState) error
• type RPCAppendEntriesRequest
  • func (msg *RPCAppendEntriesRequest) GetTerm() Term
  • func (msg *RPCAppendEntriesRequest) GetType() string
  • func (msg *RPCAppendEntriesRequest) String() string
• type RPCAppendEntriesResponse
  • func (msg *RPCAppendEntriesResponse) GetTerm() Term
  • func (msg *RPCAppendEntriesResponse) GetType() string
  • func (msg *RPCAppendEntriesResponse) String() string
• type RPCMsg
  • func DecodeRPCMsg(data []byte) (RPCMsg, error)
• type RPCRequestVoteRequest
  • func (msg *RPCRequestVoteRequest) GetTerm() Term
  • func (msg *RPCRequestVoteRequest) GetType() string
  • func (msg *RPCRequestVoteRequest) String() string
• type RPCRequestVoteResponse
  • func (msg *RPCRequestVoteResponse) GetTerm() Term
  • func (msg *RPCRequestVoteResponse) GetType() string
  • func (msg *RPCRequestVoteResponse) String() string
• type Server
  • func NewServer(cfg ServerCfg) (*Server, error)
  • func (s *Server) DataDirectoryPath() string
  • func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request)
  • func (s *Server) Start(errorChan chan<- error) error
  • func (s *Server) Stop()
• type ServerAddress
• type ServerCfg
• type ServerData
• type ServerId
• type ServerSet
• type ServerState
• type Term

Functions

func EncodeRPCMsg

func EncodeRPCMsg(msg RPCMsg) ([]byte, error)

func Panicf

func Panicf(format string, args ...interface{})

func RecoverValueString

func RecoverValueString(value interface{}) (msg string)

func StackTrace

func StackTrace(depth int) string

Types

type IncomingRPCMsg

type IncomingRPCMsg struct {
	SourceId ServerId
	Msg      RPCMsg
}

type LogEntry

type LogEntry struct {
	Term Term
	Data []byte
}

type LogIndex

type LogIndex int64

type LogReplayFunc

type LogReplayFunc func(*LogEntry) error

type LogStore

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

func NewLogStore

func NewLogStore(filePath string) *LogStore

func (*LogStore) AppendEntries

func (s *LogStore) AppendEntries(entries []LogEntry) error

func (*LogStore) Close

func (s *LogStore) Close()

func (*LogStore) LastIndex

func (s *LogStore) LastIndex() LogIndex

func (*LogStore) LastTerm

func (s *LogStore) LastTerm() Term

func (*LogStore) Open

func (s *LogStore) Open(replayFn LogReplayFunc) error

type Logger

type Logger interface {
	Debug(int, string, ...interface{})
	Info(string, ...interface{})
	Error(string, ...interface{})
}

type PersistentState

type PersistentState struct {
	CurrentTerm Term
	VotedFor    ServerId
}

type PersistentStore

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

func NewPersistentStore

func NewPersistentStore(filePath string) *PersistentStore

func (*PersistentStore) Close

func (s *PersistentStore) Close()

func (*PersistentStore) Open

func (s *PersistentStore) Open() error

func (*PersistentStore) Read

func (s *PersistentStore) Read(state *PersistentState) error

func (*PersistentStore) Write

func (s *PersistentStore) Write(state PersistentState) error

type RPCAppendEntriesRequest

type RPCAppendEntriesRequest struct {
	Term         Term
	LeaderId     ServerId
	PrevLogIndex LogIndex
	PrevLogTerm  Term
	Entries      []LogEntry
	LeaderCommit LogIndex
}

func (*RPCAppendEntriesRequest) GetTerm

func (msg *RPCAppendEntriesRequest) GetTerm() Term

func (*RPCAppendEntriesRequest) GetType

func (msg *RPCAppendEntriesRequest) GetType() string

func (*RPCAppendEntriesRequest) String

func (msg *RPCAppendEntriesRequest) String() string

type RPCAppendEntriesResponse

type RPCAppendEntriesResponse struct {
	Term    Term
	Success bool
}

func (*RPCAppendEntriesResponse) GetTerm

func (msg *RPCAppendEntriesResponse) GetTerm() Term

func (*RPCAppendEntriesResponse) GetType

func (msg *RPCAppendEntriesResponse) GetType() string

func (*RPCAppendEntriesResponse) String

func (msg *RPCAppendEntriesResponse) String() string

type RPCMsg

type RPCMsg interface {
	GetType() string
	GetTerm() Term

	fmt.Stringer
}

func DecodeRPCMsg

func DecodeRPCMsg(data []byte) (RPCMsg, error)

type RPCRequestVoteRequest

type RPCRequestVoteRequest struct {
	Term         Term
	CandidateId  ServerId
	LastLogIndex LogIndex
	LastLogTerm  Term
}

func (*RPCRequestVoteRequest) GetTerm

func (msg *RPCRequestVoteRequest) GetTerm() Term

func (*RPCRequestVoteRequest) GetType

func (msg *RPCRequestVoteRequest) GetType() string

func (*RPCRequestVoteRequest) String

func (msg *RPCRequestVoteRequest) String() string

type RPCRequestVoteResponse

type RPCRequestVoteResponse struct {
	Term        Term
	VoteGranted bool
}

func (*RPCRequestVoteResponse) GetTerm

func (msg *RPCRequestVoteResponse) GetTerm() Term

func (*RPCRequestVoteResponse) GetType

func (msg *RPCRequestVoteResponse) GetType() string

func (*RPCRequestVoteResponse) String

func (msg *RPCRequestVoteResponse) String() string

type Server

type Server struct {
	Cfg ServerCfg
	Log Logger

	Id            ServerId
	LocalAddress  ServerAddress
	PublicAddress ServerAddress
	// contains filtered or unexported fields
}

func NewServer

func NewServer(cfg ServerCfg) (*Server, error)

func (*Server) DataDirectoryPath

func (s *Server) DataDirectoryPath() string

func (*Server) ServeHTTP

func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request)

func (*Server) Start

func (s *Server) Start(errorChan chan<- error) error

func (*Server) Stop

func (s *Server) Stop()

type ServerAddress

type ServerAddress = string

type ServerCfg

type ServerCfg struct {
	Id      ServerId
	Servers ServerSet

	DataDirectory string

	Logger Logger

	MinElectionTimeout time.Duration
	MaxElectionTimeout time.Duration

	HeartbeatInterval time.Duration

	LogReplayFunc LogReplayFunc
}

type ServerData

type ServerData struct {
	LocalAddress  ServerAddress `json:"local_address"`
	PublicAddress ServerAddress `json:"public_address"`
}

type ServerId

type ServerId = string

type ServerSet

type ServerSet map[ServerId]ServerData

type ServerState

type ServerState string

const (
	ServerStateFollower  ServerState = "follower"
	ServerStateCandidate ServerState = "candidate"
	ServerStateLeader    ServerState = "leader"
)

type Term

type Term int64