raft

Documentation

Overview

raft consensus module

Index

• func CheckLeader(svcs []*Server) int
• func GetClusterIdAddr(cluster string) (map[int64]string, error)
• func GetLeader(svcs []*Server) int
• func WaitSettingAddr(svcs []*Server) map[int64]string
• type CMState
  • func (s CMState) String() string
• type ConsensusModule
  • func NewConsensusModule(id int64, peerIds []int64, server *Server, ready <-chan struct{}) *ConsensusModule
  • func (cm *ConsensusModule) AppendEntries(ctx context.Context, request *raftv1.AppendEntriesRequest) (*raftv1.AppendEntriesResponse, error)
  • func (cm *ConsensusModule) Report() (id int64, term int64, isLeader bool)
  • func (cm *ConsensusModule) RequestVote(ctx context.Context, request *raftv1.RequestVoteRequest) (*raftv1.RequestVoteResponse, error)
  • func (cm *ConsensusModule) Stop()
  • func (cm *ConsensusModule) Submit(command []byte) bool
• type LogEntry
• type RPCProxy
  • func (rpp *RPCProxy) AppendEntries(ctx context.Context, request *raftv1.AppendEntriesRequest) (*raftv1.AppendEntriesResponse, error)
  • func (rpp *RPCProxy) RequestVote(ctx context.Context, request *raftv1.RequestVoteRequest) (*raftv1.RequestVoteResponse, error)
• type RaftPersistent
  • func (rp *RaftPersistent) AppendLog(rollback, entries []LogEntry) error
  • func (rp *RaftPersistent) GetTerm() int64
  • func (rp *RaftPersistent) GetVotedFor() int64
  • func (rp *RaftPersistent) InitAndLoadLog(dest *[]LogEntry) error
  • func (rp *RaftPersistent) LogLen() int64
  • func (rp *RaftPersistent) SetTerm(term int64) error
  • func (rp *RaftPersistent) SetVotedFor(votefor int64) error
• type Server
  • func MakeCluster(n int) []*Server
  • func NewServer(sid int64, cluster string, ready <-chan struct{}) *Server
  • func (s *Server) ConnectToPeer(peerId int64, addr string) error
  • func (s *Server) DisconnectAll()
  • func (s *Server) DisconnectPeer(peerId int64) error
  • func (s *Server) GetListenAddr() string
  • func (s *Server) Serve()
  • func (s *Server) SetServerAddr(sidAddr map[int64]string)
  • func (s *Server) Shutdown()
  • func (s *Server) WaitConnectToPeers()

Functions

func CheckLeader

func CheckLeader(svcs []*Server) int

count leader

func GetClusterIdAddr

func GetClusterIdAddr(cluster string) (map[int64]string, error)

func GetLeader

func GetLeader(svcs []*Server) int

get leader id

func WaitSettingAddr

func WaitSettingAddr(svcs []*Server) map[int64]string

Types

type CMState

type CMState int

const (
	Follower CMState = iota
	Candidate
	Leader
	Dead
)

func (CMState) String

func (s CMState) String() string

type ConsensusModule

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

func NewConsensusModule

func NewConsensusModule(id int64, peerIds []int64, server *Server, ready <-chan struct{}) *ConsensusModule

func (*ConsensusModule) AppendEntries

func (cm *ConsensusModule) AppendEntries(ctx context.Context, request *raftv1.AppendEntriesRequest) (*raftv1.AppendEntriesResponse, error)

func (*ConsensusModule) Report

func (cm *ConsensusModule) Report() (id int64, term int64, isLeader bool)

func (*ConsensusModule) RequestVote

func (cm *ConsensusModule) RequestVote(ctx context.Context, request *raftv1.RequestVoteRequest) (*raftv1.RequestVoteResponse, error)

func (*ConsensusModule) Stop

func (cm *ConsensusModule) Stop()

func (*ConsensusModule) Submit

func (cm *ConsensusModule) Submit(command []byte) bool

type LogEntry

type LogEntry struct {
	Command []byte
	Term    int64
}

type RPCProxy

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

func (*RPCProxy) AppendEntries

func (rpp *RPCProxy) AppendEntries(ctx context.Context, request *raftv1.AppendEntriesRequest) (*raftv1.AppendEntriesResponse, error)

func (*RPCProxy) RequestVote

func (rpp *RPCProxy) RequestVote(ctx context.Context, request *raftv1.RequestVoteRequest) (*raftv1.RequestVoteResponse, error)

type RaftPersistent

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

func (*RaftPersistent) AppendLog

func (rp *RaftPersistent) AppendLog(rollback, entries []LogEntry) error

func (*RaftPersistent) GetTerm

func (rp *RaftPersistent) GetTerm() int64

func (*RaftPersistent) GetVotedFor

func (rp *RaftPersistent) GetVotedFor() int64

func (*RaftPersistent) InitAndLoadLog

func (rp *RaftPersistent) InitAndLoadLog(dest *[]LogEntry) error

func (*RaftPersistent) LogLen

func (rp *RaftPersistent) LogLen() int64

func (*RaftPersistent) SetTerm

func (rp *RaftPersistent) SetTerm(term int64) error

func (*RaftPersistent) SetVotedFor

func (rp *RaftPersistent) SetVotedFor(votefor int64) error

type Server

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

func MakeCluster

func MakeCluster(n int) []*Server

func NewServer

func NewServer(sid int64, cluster string, ready <-chan struct{}) *Server

func (*Server) ConnectToPeer

func (s *Server) ConnectToPeer(peerId int64, addr string) error

func (*Server) DisconnectAll

func (s *Server) DisconnectAll()

func (*Server) DisconnectPeer

func (s *Server) DisconnectPeer(peerId int64) error

func (*Server) GetListenAddr

func (s *Server) GetListenAddr() string

func (*Server) Serve

func (s *Server) Serve()

func (*Server) SetServerAddr

func (s *Server) SetServerAddr(sidAddr map[int64]string)

func (*Server) Shutdown

func (s *Server) Shutdown()

func (*Server) WaitConnectToPeers

func (s *Server) WaitConnectToPeers()