repl

package
v1.34.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 9, 2024 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

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

	ActionPreparePkt = "ActionPreparePkt"
)
View Source 
const (
	ReplRuning    = 2
	ReplExiting   = 1
	ReplHasExited = -3

	FollowerTransportRuning  = 2
	FollowerTransportExiting = 1
	FollowerTransportExited  = -1
)
View Source 
const (
	ConnIsNullErr = "ConnIsNullErr"
)
View Source 
const (
	ReplProtocolError = 1
)
View Source 
const (
	RequestChanSize = 2048
)

Variables

View Source 
var (
	ErrBadNodes       = errors.New("BadNodesErr")
	ErrArgLenMismatch = errors.New("ArgLenMismatchErr")
)
View Source 
var ErrorUnknownOp = errors.New("unknown opcode")

Functions

This section is empty.

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, c net.Conn) (ft *FollowerTransport, err error)

func (*FollowerTransport) Destory 

func (ft *FollowerTransport) Destory()

func (*FollowerTransport) Write 

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

type MakeExtentRepairReadPacket 

type MakeExtentRepairReadPacket func(partitionID uint64, extentID uint64, offset, size int) (p PacketInterface)

type MakeStreamReadResponsePacket 

type MakeStreamReadResponsePacket func(requestID int64, partitionID uint64, extentID uint64) (p PacketInterface)

type NewPacketFunc 

type NewPacketFunc func() (p PacketInterface)

type Packet 

type Packet struct {
	proto.Packet

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

	AfterPre bool
	// contains filtered or unexported fields
}

func NewNormalExtentWithHoleStreamReadResponsePacket 

func NewNormalExtentWithHoleStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (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 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) GetArg 

func (p *Packet) GetArg() []byte

func (*Packet) GetArgLen 

func (p *Packet) GetArgLen() uint32

func (*Packet) GetCRC 

func (p *Packet) GetCRC() uint32

func (*Packet) GetData 

func (p *Packet) GetData() []byte

func (*Packet) GetExtentID 

func (p *Packet) GetExtentID() uint64

func (*Packet) GetExtentOffset 

func (p *Packet) GetExtentOffset() int64

func (*Packet) GetOpcode 

func (p *Packet) GetOpcode() uint8

func (*Packet) GetPartitionID 

func (p *Packet) GetPartitionID() uint64

func (*Packet) GetResultCode 

func (p *Packet) GetResultCode() uint8

func (*Packet) GetSize 

func (p *Packet) GetSize() uint32

func (*Packet) GetStartT 

func (p *Packet) GetStartT() int64

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) IsMarkSplitExtentOperation 

func (p *Packet) IsMarkSplitExtentOperation() bool

func (*Packet) IsMasterCommand 

func (p *Packet) IsMasterCommand() bool

func (*Packet) IsNormalWriteOperation 

func (p *Packet) IsNormalWriteOperation() bool

func (*Packet) IsRandomWrite 

func (p *Packet) IsRandomWrite() bool

func (*Packet) IsReadOperation 

func (p *Packet) IsReadOperation() bool

func (*Packet) IsSnapshotModWriteAppendOperation 

func (p *Packet) IsSnapshotModWriteAppendOperation() bool

func (*Packet) IsSyncWrite 

func (p *Packet) IsSyncWrite() bool

func (*Packet) IsTinyExtentType 

func (p *Packet) IsTinyExtentType() bool

func (*Packet) IsUrgentLeaderReq 

func (p *Packet) IsUrgentLeaderReq() bool

op need to be processed by dp raft leader.

func (*Packet) PackErrorBody 

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

func (*Packet) ReadFull 

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

func (*Packet) SetArg 

func (p *Packet) SetArg(data []byte)

func (*Packet) SetArglen 

func (p *Packet) SetArglen(len uint32)

func (*Packet) SetCRC 

func (p *Packet) SetCRC(crc uint32)

func (*Packet) SetData 

func (p *Packet) SetData(data []byte)

func (*Packet) SetDegrade 

func (p *Packet) SetDegrade()

func (*Packet) SetExtentOffset 

func (p *Packet) SetExtentOffset(offset int64)

func (*Packet) SetOpCode 

func (p *Packet) SetOpCode(op uint8)

func (*Packet) SetResultCode 

func (p *Packet) SetResultCode(code uint8)

func (*Packet) SetSize 

func (p *Packet) SetSize(size uint32)

func (*Packet) SetStartT 

func (p *Packet) SetStartT(StartT int64)

func (*Packet) ShallDegrade 

func (p *Packet) ShallDegrade() bool

func (*Packet) UnsetDegrade 

func (p *Packet) UnsetDegrade()

type PacketInterface 

type PacketInterface interface {
	IsErrPacket() bool
	WriteToConn(c net.Conn) (err error)
	ReadFromConnWithVer(c net.Conn, timeoutSec int) (err error)
	GetUniqueLogId() (m string)
	GetReqID() int64
	GetPartitionID() uint64
	GetExtentID() uint64
	GetSize() uint32
	GetCRC() uint32
	GetArg() []byte
	GetArgLen() uint32
	GetData() []byte
	GetResultCode() uint8
	GetExtentOffset() int64
	GetStartT() int64
	SetSize(size uint32)
	GetOpcode() uint8
	SetResultCode(uint8)
	SetCRC(crc uint32)
	SetExtentOffset(int64)
	GetOpMsg() (m string)
	ShallDegrade() bool
	SetStartT(StartT int64)
	SetData(data []byte)
	SetOpCode(uint8)
	LogMessage(action, remote string, start int64, err error) (m string)
	PackErrorBody(action, msg string)
	PacketOkReply()
	SetArglen(len uint32)
	SetArg(data []byte)
}

func NewExtentRepairReadPacket 

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

func NewNormalExtentWithHoleRepairReadPacket 

func NewNormalExtentWithHoleRepairReadPacket(partitionID uint64, extentID uint64, offset, size int) (p PacketInterface)

func NewPacketEx 

func NewPacketEx() (p PacketInterface)

func NewStreamReadResponsePacket 

func NewStreamReadResponsePacket(requestID int64, partitionID uint64, extentID uint64) (p PacketInterface)

func NewTinyExtentRepairReadPacket 

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

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) SetSmux 

func (rp *ReplProtocol) SetSmux(f func(addr string) (net.Conn, error), putSmux func(conn net.Conn, force bool))

func (*ReplProtocol) Stop 

func (rp *ReplProtocol) Stop()

Stop stops the replication protocol.

type SmuxConn 

type SmuxConn struct {
	net.Conn
	// contains filtered or unexported fields
}

func (*SmuxConn) Close 

func (d *SmuxConn) Close() error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.