paxos

Documentation

Index

• Constants
• Variables
• func DoAccept(index, proposalNum int, proposalValue string) error
• func DoPrepare(index int, value string, minProposal int) (bool, error)
• func GenerateProposalNum(minProposalNum, ID int) int
• func GetNetwork() string
• func GetPeerNameList() []string
• func GetServerID() int
• func InitAcceptor()
• func InitDB()
• func InitProposerNetwork()
• func ProposerHandleRequst()
• func SendAcceptRequestAndWaitForReply(req *AcceptRequest, done chan struct{}) chan *AcceptResponse
• func SendLearnRequest(req *LearnRequest)
• func SendPrepareRequestAndWaitForReply(req *PrepareRequest, done chan struct{}) chan *PrepareResponse
• func SetBatcher(duration time.Duration, limit int)
• func StartNewInstance(reqs ...Request) error
• type AcceptRequest
• type AcceptResponse
• type Acceptor
  • func GetAcceptorInstance() *Acceptor
  • func (acceptor *Acceptor) OnAccept(req *AcceptRequest, resp *AcceptResponse) error
  • func (acceptor *Acceptor) OnLearn(req *LearnRequest, resp *LearnResponse) error
  • func (acceptor *Acceptor) OnPrepare(req *PrepareRequest, resp *PrepareResponse) error
  • func (acceptor *Acceptor) ReDailMe(serverID *int, resp *struct{}) error
• type BatchRequest
• type Batcher
  • func GetBatcherInstance() *Batcher
  • func (b *Batcher) Run()
• type LearnRequest
• type LearnResponse
• type LevelDB
  • func NewLevelDB(path string) *LevelDB
  • func (level *LevelDB) Close() error
  • func (db *LevelDB) PrintLog(fileName string)
  • func (level *LevelDB) ReadLog(index int) (*LogEntry, error)
  • func (level *LevelDB) Restore(p *Proposer, a *Acceptor, statMachine sm.StatMachine) error
  • func (level *LevelDB) WriteLog(index int, entry *LogEntry) error
• type LogEntry
  • func (le *LogEntry) String() string
• type MyComparator
  • func (cmp MyComparator) Compare(a, b []byte) int
  • func (cmp MyComparator) Name() string
  • func (cmp MyComparator) Separator(dst, a, b []byte) []byte
  • func (cmp MyComparator) Successor(dst, b []byte) []byte
• type PrepareRequest
• type PrepareResponse
• type Proposer
  • func GetProposerInstance() *Proposer
• type Request
• type Store

Constants

const BatcherDefaultDuration = time.Millisecond

const BatcherDefaultLimit = 100

const SHIFT = 3

Variables

var ErrorBadLogFormat = errors.New("bad log format")

var ErrorEmptyKey = errors.New("empty key error")

var ErrorEmptyValue = errors.New("empty value error")

var ErrorNoPeers = errors.New("nil peers present")

var ErrorNotFound = errors.New("not found error")

var ErrorUnkonwCommand = errors.New("unkonw command")

Functions

func DoAccept

func DoAccept(index, proposalNum int, proposalValue string) error

DoAccept starts the accept phase sending an accept request to all the Paxos peers including itself

func DoPrepare

func DoPrepare(index int, value string, minProposal int) (bool, error)

DoPrepare可以确定index位置的值 这里的value格式为 [SET key value]或者[DELETE key] 如果成功提交当前value，返回true，否则返回false

func GenerateProposalNum

func GenerateProposalNum(minProposalNum, ID int) int

GenerateProposalNum Generates a proposal number out of minProposalNum and Server ID

func GetNetwork

func GetNetwork() string

GetNetwork Obtains Network From Environment Variable

func GetPeerNameList

func GetPeerNameList() []string

GetPeerNameList Obtains Peer List From Environment Variable

func GetServerID

func GetServerID() int

func InitAcceptor

func InitAcceptor()

func InitDB

func InitDB()

func InitProposerNetwork

func InitProposerNetwork()

func ProposerHandleRequst

func ProposerHandleRequst()

func SendAcceptRequestAndWaitForReply

func SendAcceptRequestAndWaitForReply(req *AcceptRequest, done chan struct{}) chan *AcceptResponse

func SendLearnRequest

func SendLearnRequest(req *LearnRequest)

func SendPrepareRequestAndWaitForReply

func SendPrepareRequestAndWaitForReply(req *PrepareRequest, done chan struct{}) chan *PrepareResponse

func SetBatcher

func SetBatcher(duration time.Duration, limit int)

func StartNewInstance

func StartNewInstance(reqs ...Request) error

Prepare starts a Paxos round sending a prepare request to all the Paxos peers including itself 这里发起proposal，直到自己提议的value被chosen，具体的数据传输在doPrepare中完成 oper取值范围为SET DELETE NOP

Types

type AcceptRequest

type AcceptRequest struct {
	Index         int
	ProposalNum   int
	ProposalValue string
}

type AcceptResponse

type AcceptResponse struct {
	MinProposal int
}

type Acceptor

type Acceptor struct {
}

func GetAcceptorInstance

func GetAcceptorInstance() *Acceptor

func (*Acceptor) OnAccept

func (acceptor *Acceptor) OnAccept(req *AcceptRequest, resp *AcceptResponse) error

OnAccept return minProposal

func (*Acceptor) OnLearn

func (acceptor *Acceptor) OnLearn(req *LearnRequest, resp *LearnResponse) error

func (*Acceptor) OnPrepare

func (acceptor *Acceptor) OnPrepare(req *PrepareRequest, resp *PrepareResponse) error

OnPrepare return acceptedProposal and acceptedValue

func (*Acceptor) ReDailMe

func (acceptor *Acceptor) ReDailMe(serverID *int, resp *struct{}) error

type BatchRequest

type BatchRequest struct {
	Reqs []Request
	Done chan struct{}
}

type Batcher

type Batcher struct {
	Out chan BatchRequest
	In  chan Request
	// contains filtered or unexported fields
}

func GetBatcherInstance

func GetBatcherInstance() *Batcher

func (*Batcher) Run

func (b *Batcher) Run()

type LearnRequest

type LearnRequest struct {
	Index            int
	AcceptedValue    string
	AcceptedProposal int
}

type LearnResponse

type LearnResponse struct {
}

type LevelDB

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

func NewLevelDB

func NewLevelDB(path string) *LevelDB

func (*LevelDB) Close

func (level *LevelDB) Close() error

func (*LevelDB) PrintLog

func (db *LevelDB) PrintLog(fileName string)

func (*LevelDB) ReadLog

func (level *LevelDB) ReadLog(index int) (*LogEntry, error)

func (*LevelDB) Restore

func (level *LevelDB) Restore(p *Proposer, a *Acceptor, statMachine sm.StatMachine) error

func (*LevelDB) WriteLog

func (level *LevelDB) WriteLog(index int, entry *LogEntry) error

type LogEntry

type LogEntry struct {
	MinProposal      int
	AcceptedProposal int
	AcceptedValue    string
	IsCommited       bool
}

func (*LogEntry) String

func (le *LogEntry) String() string

type MyComparator

type MyComparator struct {
}

func (MyComparator) Compare

func (cmp MyComparator) Compare(a, b []byte) int

func (MyComparator) Name

func (cmp MyComparator) Name() string

func (MyComparator) Separator

func (cmp MyComparator) Separator(dst, a, b []byte) []byte

func (MyComparator) Successor

func (cmp MyComparator) Successor(dst, b []byte) []byte

type PrepareRequest

type PrepareRequest struct {
	Index       int
	ProposalNum int
}

type PrepareResponse

type PrepareResponse struct {
	AcceptedProposal int
	AcceptedValue    string
}

type Proposer

type Proposer struct {
	ServerID int
	LogIndex int // 记录最小的没有Chosen的logIndex
	Peers    []*rpc.Client
	// contains filtered or unexported fields
}

func GetProposerInstance

func GetProposerInstance() *Proposer

type Request

type Request interface {
	GetValue() string
	Done() chan error
}

type Store

type Store interface {
	WriteLog(index int, entry *LogEntry) error
	ReadLog(index int) (*LogEntry, error)
	Restore(p *Proposer, a *Acceptor, sm sm.StatMachine) error // 重启后读取文件恢复状态
	PrintLog(fileName string)
	Close() error
}

var DB Store