Mod: package format form {2 Bytes Header len + Header + 2 Body len + Body} to {Header + Body}

rpc/client.go

@@ -32,7 +32,7 @@ type Client struct {
 	sequence gxatomic.Uint64
 
 	pendingLock      sync.RWMutex
-	pendingResponses map[uint64]*PendingResponse
+	pendingResponses map[SequenceType]*PendingResponse
 }
 
 func NewClient(conf *ClientConfig) (*Client, error) {
@@ -41,7 +41,7 @@ func NewClient(conf *ClientConfig) (*Client, error) {
 	}
 
 	c := &Client{
-		pendingResponses: make(map[uint64]*PendingResponse),
+		pendingResponses: make(map[SequenceType]*PendingResponse),
 		conf:             *conf,
 	}
 	c.pool = newGettyRPCClientConnPool(c, conf.PoolSize, time.Duration(int(time.Second)*conf.PoolTTL))
@@ -85,7 +85,7 @@ func (c *Client) Call(typ CodecType, addr, service, method string, args interfac
 	select {
 	case <-getty.GetTimeWheel().After(c.conf.GettySessionParam.tcpReadTimeout):
 		err = errClientReadTimeout
-		c.RemovePendingResponse(rsp.seq)
+		c.RemovePendingResponse(SequenceType(rsp.seq))
 	case <-rsp.done:
 		err = rsp.err
 	}
@@ -121,9 +121,9 @@ func (c *Client) transfer(session getty.Session, typ CodecType, req *GettyRPCReq
 	)
 
 	sequence = c.sequence.Add(1)
-	pkg.H.Magic = gettyPackageMagic
-	pkg.H.LogID = (uint32)(randomID())
-	pkg.H.Sequence = sequence
+	pkg.H.Magic = MagicType(gettyPackageMagic)
+	pkg.H.LogID = LogIDType(randomID())
+	pkg.H.Sequence = SequenceType(sequence)
 	pkg.H.Command = gettyCmdHbRequest
 	pkg.H.CodecType = typ
 	if req != nil {
@@ -136,7 +136,7 @@ func (c *Client) transfer(session getty.Session, typ CodecType, req *GettyRPCReq
 
 	err = session.WritePkg(pkg, 0)
 	if err != nil {
-		c.RemovePendingResponse(rsp.seq)
+		c.RemovePendingResponse(SequenceType(rsp.seq))
 	}
 
 	return jerrors.Trace(err)
@@ -151,10 +151,10 @@ func (c *Client) PendingResponseCount() int {
 func (c *Client) AddPendingResponse(pr *PendingResponse) {
 	c.pendingLock.Lock()
 	defer c.pendingLock.Unlock()
-	c.pendingResponses[pr.seq] = pr
+	c.pendingResponses[SequenceType(pr.seq)] = pr
 }
 
-func (c *Client) RemovePendingResponse(seq uint64) *PendingResponse {
+func (c *Client) RemovePendingResponse(seq SequenceType) *PendingResponse {
 	c.pendingLock.Lock()
 	defer c.pendingLock.Unlock()
 	if c.pendingResponses == nil {
@@ -167,7 +167,7 @@ func (c *Client) RemovePendingResponse(seq uint64) *PendingResponse {
 	return nil
 }
 
-func (c *Client) ClearPendingResponses() map[uint64]*PendingResponse {
+func (c *Client) ClearPendingResponses() map[SequenceType]*PendingResponse {
 	c.pendingLock.Lock()
 	defer c.pendingLock.Unlock()
 	presps := c.pendingResponses

rpc/codec.go

@@ -19,7 +19,7 @@ import (
 //  getty command
 ////////////////////////////////////////////
 
-type gettyCommand uint32
+type gettyCommand int32
 
 const (
 	gettyDefaultCmd     gettyCommand = 0x00
@@ -45,7 +45,7 @@ func (c gettyCommand) String() string {
 //  getty error code
 ////////////////////////////////////////////
 
-type GettyErrorCode int32
+type GettyErrorCode int16
 
 const (
 	GettyOK   GettyErrorCode = 0x00
@@ -56,7 +56,7 @@ const (
 //  getty codec type
 ////////////////////////////////////////////
 
-type CodecType uint32
+type CodecType int16
 
 const (
 	CodecUnknown  CodecType = 0x00
@@ -104,6 +104,10 @@ func GetCodecType(codecType string) CodecType {
 	return CodecUnknown
 }
 
+////////////////////////////////////////////
+//  getty codec
+////////////////////////////////////////////
+
 type Codec interface {
 	Encode(interface{}) ([]byte, error)
 	Decode([]byte, interface{}) error
@@ -170,8 +174,8 @@ func (c PBCodec) Decode(buf []byte, msg interface{}) error {
 
 const (
 	gettyPackageMagic     = 0x20160905
-	maxPackageLen            = 1024 * 1024
-	rpcPackagePlaceholderLen = 2
+	maxPackageLen         = 4 * 1024 * 1024
+	gettyPackageHeaderLen = (int)((uint)(unsafe.Sizeof(GettyPackageHeader{})))
 )
 
 var (
@@ -181,14 +185,6 @@ var (
 	ErrIllegalMagic    = jerrors.New("package magic is not right.")
 )
 
-var (
-	gettyPackageHeaderLen int
-)
-
-func init() {
-	gettyPackageHeaderLen = (int)((uint)(unsafe.Sizeof(GettyPackageHeader{})))
-}
-
 type RPCPackage interface {
 	Marshal(CodecType, *bytes.Buffer) (int, error)
 	// @buf length should be equal to GettyPkg.GettyPackageHeader.Len
@@ -197,16 +193,25 @@ type RPCPackage interface {
 	GetHeader() interface{}
 }
 
+type (
+	MagicType     int32
+	LogIDType     int32
+	SequenceType  uint64
+	ServiceIDType int32
+	PkgLenType    int32
+)
+
 type GettyPackageHeader struct {
-	Magic    uint32 // magic number
-	LogID    uint32 // log id
-	Sequence uint64 // request/response sequence
+	Magic    MagicType    // magic number
+	LogID    LogIDType    // log id
+	Sequence SequenceType // request/response sequence
 
 	Command   gettyCommand  // operation command code
-	Code    GettyErrorCode // error code
+	ServiceID ServiceIDType // service id
 
-	ServiceID uint32 // service id
+	Code      GettyErrorCode // error code
 	CodecType CodecType
+	PkgLen    PkgLenType
 }
 
 type GettyPackage struct {
@@ -222,69 +227,59 @@ func (p GettyPackage) String() string {
 func (p *GettyPackage) Marshal() (*bytes.Buffer, error) {
 	var (
 		err            error
-		packLen, length int
-		buf             *bytes.Buffer
+		headerBuf, buf *bytes.Buffer
 	)
 
-	packLen = rpcPackagePlaceholderLen + gettyPackageHeaderLen
+	buf = bytes.NewBuffer(make([]byte, gettyPackageHeaderLen))
+
+	// body
 	if p.B != nil {
-		buf = &bytes.Buffer{}
-		length, err = p.B.Marshal(p.H.CodecType, buf)
+		length, err := p.B.Marshal(p.H.CodecType, buf)
 		if err != nil {
 			return nil, jerrors.Trace(err)
 		}
-		packLen = rpcPackagePlaceholderLen + gettyPackageHeaderLen + length
-	}
-	buf0 := &bytes.Buffer{}
-	err = binary.Write(buf0, binary.LittleEndian, uint16(packLen))
-	if err != nil {
-		return nil, jerrors.Trace(err)
+		p.H.PkgLen = PkgLenType(length)
 	}
-	err = binary.Write(buf0, binary.LittleEndian, p.H)
+
+	// header
+	headerBuf = bytes.NewBuffer(nil)
+	err = binary.Write(headerBuf, binary.LittleEndian, p.H)
 	if err != nil {
 		return nil, jerrors.Trace(err)
 	}
-	if p.B != nil {
-		if err = binary.Write(buf0, binary.LittleEndian, buf.Bytes()); err != nil {
-			return nil, jerrors.Trace(err)
-		}
-	}
-	return buf0, nil
+	copy(buf.Bytes(), headerBuf.Bytes()[:gettyPackageHeaderLen])
+
+	return buf, nil
 }
 
 func (p *GettyPackage) Unmarshal(buf *bytes.Buffer) (int, error) {
-	var err error
-	if buf.Len() < rpcPackagePlaceholderLen+gettyPackageHeaderLen {
+	bufLen := buf.Len()
+	if bufLen < gettyPackageHeaderLen {
 		return 0, ErrNotEnoughStream
 	}
-	var packLen uint16
-	err = binary.Read(buf, binary.LittleEndian, &packLen)
-	if err != nil {
-		return 0, jerrors.Trace(err)
-	}
-	if int(packLen) > maxPackageLen {
-		return 0, ErrTooLargePackage
-	}
-	if int(packLen) < gettyPackageHeaderLen {
-		return 0, ErrInvalidPackage
-	}
 
 	// header
 	if err := binary.Read(buf, binary.LittleEndian, &(p.H)); err != nil {
 		return 0, jerrors.Trace(err)
 	}
+
 	if p.H.Magic != gettyPackageMagic {
 		log.Error("@p.H.Magic{%x}, right magic{%x}", p.H.Magic, gettyPackageMagic)
 		return 0, ErrIllegalMagic
 	}
 
-	if int(packLen) > rpcPackagePlaceholderLen+gettyPackageHeaderLen {
-		if err := p.B.Unmarshal(p.H.CodecType, bytes.NewBuffer(buf.Next(int(packLen)-rpcPackagePlaceholderLen-gettyPackageHeaderLen))); err != nil {
+	totalLen := int(PkgLenType(gettyPackageHeaderLen) + p.H.PkgLen)
+	if totalLen > maxPackageLen {
+		return 0, ErrTooLargePackage
+	}
+
+	if bufLen >= totalLen {
+		if err := p.B.Unmarshal(p.H.CodecType, bytes.NewBuffer(buf.Next(int(p.H.PkgLen)))); err != nil {
 			return 0, jerrors.Trace(err)
 		}
 	}
 
-	return int(packLen), nil
+	return totalLen, nil
 }
 
 ////////////////////////////////////////////