repl

Documentation

Index

• Constants
• Variables
• type FollowerPacket
  • func NewFollowerPacket() (fp *FollowerPacket)
  • func (p *FollowerPacket) IsErrPacket() bool
  • func (p *FollowerPacket) PackErrorBody(action, msg string)
• type FollowerTransport
  • func NewFollowersTransport(addr string) (ft *FollowerTransport, err error)
  • func (ft *FollowerTransport) Destory()
  • func (ft *FollowerTransport) Write(p *FollowerPacket)
• type Packet
  • func NewExtentRepairReadPacket(partitionID uint64, extentID uint64, offset, size int) (p *Packet)
  • func NewPacket() (p *Packet)
  • func NewPacketToGetAllWatermarks(partitionID uint64, extentType uint8) (p *Packet)
  • func NewPacketToNotifyExtentRepair(partitionID uint64) (p *Packet)
  • func NewPacketToReadTinyDeleteRecord(partitionID uint64, offset int64) (p *Packet)
  • func NewReadTinyDeleteRecordResponsePacket(requestID int64, partitionID uint64) (p *Packet)
  • func NewStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p *Packet)
  • func NewTinyExtentRepairReadPacket(partitionID uint64, extentID uint64, offset, size int) (p *Packet)
  • func NewTinyExtentStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p *Packet)
  • func (p *Packet) AfterTp() (ok bool)
  • func (p *Packet) BeforeTp(clusterID string) (ok bool)
  • func (p *Packet) IsBatchDeleteExtents() bool
  • func (p *Packet) IsBroadcastMinAppliedID() bool
  • func (p *Packet) IsCreateExtentOperation() bool
  • func (p *Packet) IsErrPacket() bool
  • func (p *Packet) IsForwardPacket() bool
  • func (p *Packet) IsLeaderPacket() (ok bool)
  • func (p *Packet) IsMarkDeleteExtentOperation() bool
  • func (p *Packet) IsMasterCommand() bool
  • func (p *Packet) IsRandomWrite() bool
  • func (p *Packet) IsReadOperation() bool
  • func (p *Packet) IsSingleReplicatePacket() bool
  • func (p *Packet) IsSyncWrite() bool
  • func (p *Packet) IsTinyExtentType() bool
  • func (p *Packet) IsWriteOperation() bool
  • func (p *Packet) PackErrorBody(action, msg string)
  • func (p *Packet) ReadFromConnFromCli(c net.Conn, deadlineTime time.Duration) (err error)
  • func (p *Packet) ReadFull(c net.Conn, opcode uint8, readSize int) (err error)
• type ReplProtocol
  • func NewReplProtocol(inConn net.Conn, prepareFunc func(p *Packet) error, ...) *ReplProtocol
  • func (rp *ReplProtocol) OperatorAndForwardPktGoRoutine()
  • func (rp *ReplProtocol) ReceiveResponseFromFollowersGoRoutine()
  • func (rp *ReplProtocol) ServerConn()
  • func (rp *ReplProtocol) Stop()

Constants

const (
	ActionSendToFollowers     = "ActionSendToFollowers"
	ActionReceiveFromFollower = "ActionReceiveFromFollower"
	ActionWriteToClient       = "ActionWriteToClient"
	ActionCheckReply          = "ActionCheckReply"

	ActionPreparePkt = "ActionPreparePkt"
)

const (
	ReplRuning    = 2
	ReplExiting   = 1
	ReplHasExited = -3

	FollowerTransportRuning  = 2
	FollowerTransportExiting = 1
	FollowerTransportExited  = -1
)

const (
	ConnIsNullErr = "ConnIsNullErr"
)

const (
	ReplProtocolError = 1
)

const (
	RequestChanSize = 10240
)

Variables

var (
	ErrBadNodes       = errors.New("BadNodesErr")
	ErrArgLenMismatch = errors.New("ArgLenMismatchErr")
)

var (
	ErrorUnknownOp = errors.New("unknown opcode")
)

Types

type FollowerPacket

type FollowerPacket struct {
	proto.Packet
	// contains filtered or unexported fields
}

func NewFollowerPacket

func NewFollowerPacket() (fp *FollowerPacket)

func (*FollowerPacket) IsErrPacket

func (p *FollowerPacket) IsErrPacket() bool

func (*FollowerPacket) PackErrorBody

func (p *FollowerPacket) PackErrorBody(action, msg string)

type FollowerTransport

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

func NewFollowersTransport

func NewFollowersTransport(addr string) (ft *FollowerTransport, err error)

func (*FollowerTransport) Destory

func (ft *FollowerTransport) Destory()

func (*FollowerTransport) Write

func (ft *FollowerTransport) Write(p *FollowerPacket)

type Packet

type Packet struct {
	proto.Packet

	IsReleased int32 // TODO what is released?
	Object     interface{}
	TpObject   *exporter.TimePointCount
	NeedReply  bool
	OrgBuffer  []byte
	// contains filtered or unexported fields
}

func NewExtentRepairReadPacket

func NewExtentRepairReadPacket(partitionID uint64, extentID uint64, offset, size int) (p *Packet)

func NewPacket

func NewPacket() (p *Packet)

func NewPacketToGetAllWatermarks

func NewPacketToGetAllWatermarks(partitionID uint64, extentType uint8) (p *Packet)

func NewPacketToNotifyExtentRepair

func NewPacketToNotifyExtentRepair(partitionID uint64) (p *Packet)

func NewPacketToReadTinyDeleteRecord

func NewPacketToReadTinyDeleteRecord(partitionID uint64, offset int64) (p *Packet)

func NewReadTinyDeleteRecordResponsePacket

func NewReadTinyDeleteRecordResponsePacket(requestID int64, partitionID uint64) (p *Packet)

func NewStreamReadResponsePacket

func NewStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p *Packet)

func NewTinyExtentRepairReadPacket

func NewTinyExtentRepairReadPacket(partitionID uint64, extentID uint64, offset, size int) (p *Packet)

func NewTinyExtentStreamReadResponsePacket

func NewTinyExtentStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p *Packet)

func (*Packet) AfterTp

func (p *Packet) AfterTp() (ok bool)

func (*Packet) BeforeTp

func (p *Packet) BeforeTp(clusterID string) (ok bool)

func (*Packet) IsBatchDeleteExtents

func (p *Packet) IsBatchDeleteExtents() bool

func (*Packet) IsBroadcastMinAppliedID

func (p *Packet) IsBroadcastMinAppliedID() bool

func (*Packet) IsCreateExtentOperation

func (p *Packet) IsCreateExtentOperation() bool

func (*Packet) IsErrPacket

func (p *Packet) IsErrPacket() bool

func (*Packet) IsForwardPacket

func (p *Packet) IsForwardPacket() bool

func (*Packet) IsLeaderPacket

func (p *Packet) IsLeaderPacket() (ok bool)

A leader packet is the packet send to the leader and does not require packet forwarding.

func (*Packet) IsMarkDeleteExtentOperation

func (p *Packet) IsMarkDeleteExtentOperation() bool

func (*Packet) IsMasterCommand

func (p *Packet) IsMasterCommand() bool

func (*Packet) IsRandomWrite

func (p *Packet) IsRandomWrite() bool

func (*Packet) IsReadOperation

func (p *Packet) IsReadOperation() bool

func (*Packet) IsSingleReplicatePacket

func (p *Packet) IsSingleReplicatePacket() bool

func (*Packet) IsSyncWrite

func (p *Packet) IsSyncWrite() bool

func (*Packet) IsTinyExtentType

func (p *Packet) IsTinyExtentType() bool

func (*Packet) IsWriteOperation

func (p *Packet) IsWriteOperation() bool

func (*Packet) PackErrorBody

func (p *Packet) PackErrorBody(action, msg string)

func (*Packet) ReadFromConnFromCli

func (p *Packet) ReadFromConnFromCli(c net.Conn, deadlineTime time.Duration) (err error)

func (*Packet) ReadFull

func (p *Packet) ReadFull(c net.Conn, opcode uint8, readSize int) (err error)

type ReplProtocol

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

ReplProtocol defines the struct of the replication protocol. 1. ServerConn reads a packet from the client socket, and analyzes the addresses of the followers. 2. After the preparation, the packet is send to toBeProcessedCh. If failure happens, send it to the response channel. 3. OperatorAndForwardPktGoRoutine fetches a packet from toBeProcessedCh, and determine if it needs to be forwarded to the followers. 4. receiveResponse fetches a reply from responseCh, executes postFunc, and writes a response to the client if necessary.

func NewReplProtocol

func NewReplProtocol(inConn net.Conn, prepareFunc func(p *Packet) error,
	operatorFunc func(p *Packet, c net.Conn) error, postFunc func(p *Packet) error) *ReplProtocol

func (*ReplProtocol) OperatorAndForwardPktGoRoutine

func (rp *ReplProtocol) OperatorAndForwardPktGoRoutine()

OperatorAndForwardPktGoRoutine reads packets from the to-be-processed channel and writes responses to the client.

1. Read a packet from toBeProcessCh, and determine if it needs to be forwarded or not. If the answer is no, then process the packet locally and put it into responseCh.
2. If the packet needs to be forwarded, the first send it to the followers, and execute the operator function. Then notify receiveResponse to read the followers' responses.
3. Read a reply from responseCh, and write to the client.

func (*ReplProtocol) ReceiveResponseFromFollowersGoRoutine

func (rp *ReplProtocol) ReceiveResponseFromFollowersGoRoutine()

Receive response from all followers.

func (*ReplProtocol) ServerConn

func (rp *ReplProtocol) ServerConn()

ServerConn keeps reading data from the socket to analyze the follower address, execute the prepare function, and throw the packets to the to-be-processed channel.

func (*ReplProtocol) Stop

func (rp *ReplProtocol) Stop()

Stop stops the replication protocol.