repl

Documentation

Index

• Constants
• Variables
• 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 NewPacketToTinyDeleteRecord(partitionID uint64, offset int64) (p *Packet)
  • func NewStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p *Packet)
  • func NewTinyDeleteRecordResponsePacket(requestID int64, partitionID uint64) (p *Packet)
  • func (p *Packet) AfterTp() (ok bool)
  • func (p *Packet) BeforeTp(clusterID string) (ok bool)
  • func (p *Packet) IsErrPacket() bool
  • func (p *Packet) IsMasterCommand() 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, readSize int) (err error)
• type ReplProtocol
  • func NewReplProtocol(inConn *net.TCPConn, 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
)

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 Packet

type Packet struct {
	proto.Packet

	IsReleased int32 // TODO what is released?
	Object     interface{}
	TpObject   *exporter.TimePointCount
	NeedReply  bool
	// 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 NewPacketToTinyDeleteRecord

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

func NewStreamReadResponsePacket

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

func NewTinyDeleteRecordResponsePacket

func NewTinyDeleteRecordResponsePacket(requestID int64, partitionID 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) IsErrPacket

func (p *Packet) IsErrPacket() bool

func (*Packet) IsMasterCommand

func (p *Packet) IsMasterCommand() 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, 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.TCPConn, prepareFunc func(p *Packet) error,
	operatorFunc func(p *Packet, c *net.TCPConn) 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.