rpc临时内存池优化

rpc/client.go

@@ -56,7 +56,7 @@ func (slf *Client) Connect(addr string) error {
 }
 
 func (slf *Client) AsycGo(rpcHandler IRpcHandler,serviceMethod string,callback reflect.Value, args interface{},replyParam interface{}) error {
-	call := new(Call)
+	call := MakeCall()
 	call.Reply = replyParam
 	call.callback = &callback
 	call.rpcHandler = rpcHandler
@@ -72,16 +72,13 @@ func (slf *Client) AsycGo(rpcHandler IRpcHandler,serviceMethod string,callback r
 	slf.pendingLock.Lock()
 	slf.startSeq += 1
 	call.Seq = slf.startSeq
-	//request.Seq = slf.startSeq
-	//request.NoReply = false
-	//request.ServiceMethod = serviceMethod
 	request.RpcRequestData = processor.MakeRpcRequest(slf.startSeq,serviceMethod,false,InParam)
-	slf.pending[call.Seq] = call
+	slf.pending[call.Seq] = call//如果下面发送失败，将会一一直存在这里
 	slf.pendingLock.Unlock()
 
-
-
 	bytes,err := processor.Marshal(request.RpcRequestData)
+	processor.ReleaseRpcRequest(request.RpcRequestData)
+
 	if err != nil {
 		return err
 	}
@@ -97,8 +94,7 @@ func (slf *Client) AsycGo(rpcHandler IRpcHandler,serviceMethod string,callback r
 }
 
 func (slf *Client) Go(noReply bool,serviceMethod string, args interface{},reply interface{}) *Call {
-	call := new(Call)
+	call := MakeCall()
-	call.done = make(chan *Call,1)
 	call.Reply = reply
 	call.ServiceMethod = serviceMethod
 
@@ -109,19 +105,15 @@ func (slf *Client) Go(noReply bool,serviceMethod string, args interface{},reply
 	}
 
 	request := &RpcRequest{}
-	//request.NoReply = noReply
 	call.Arg = args
 	slf.pendingLock.Lock()
 	slf.startSeq += 1
 	call.Seq = slf.startSeq
-	//request.Seq = slf.startSeq
-	//	request.ServiceMethod = serviceMethod
 	request.RpcRequestData = processor.MakeRpcRequest(slf.startSeq,serviceMethod,noReply,InParam)
-	slf.pending[call.Seq] = call
+	//slf.pending[call.Seq] = call
 	slf.pendingLock.Unlock()
-
-
 	bytes,err := processor.Marshal(request.RpcRequestData)
+	processor.ReleaseRpcRequest(request.RpcRequestData)
 	if err != nil {
 		call.Err = err
 		return call
@@ -140,37 +132,6 @@ func (slf *Client) Go(noReply bool,serviceMethod string, args interface{},reply
 	return call
 }
 
-type RequestHandler func(Returns interface{},Err *RpcError)
-
-type RpcRequest struct {
-	RpcRequestData IRpcRequestData
-
-	//packhead
-	/*Seq uint64             // sequence number chosen by client
-	ServiceMethod string   // format: "Service.Method"
-	NoReply bool           //是否需要返回
-	//packbody
-	InParam []byte
-*/
-	//other data
-	localReply interface{}
-	localParam interface{} //本地调用的参数列表
-	requestHandle RequestHandler
-	callback *reflect.Value
-}
-
-type RpcResponse struct {
-	RpcResponeData IRpcResponseData
-	/*
-	//head
-	Seq           uint64   // sequence number chosen by client
-	Err *RpcError
-
-	//returns
-	Reply []byte*/
-}
-
-
 func (slf *Client) Run(){
 	defer func() {
 		if r := recover(); r != nil {
@@ -190,6 +151,7 @@ func (slf *Client) Run(){
 		//1.解析head
 		respone := &RpcResponse{}
 		respone.RpcResponeData =processor.MakeRpcResponse(0,nil,nil)
+		defer processor.ReleaseRpcRespose(respone.RpcResponeData)
 		err = processor.Unmarshal(bytes,respone.RpcResponeData)
 		if err != nil {
 			log.Error("rpcClient Unmarshal head error,error:%+v",err)

rpc/jsonprocessor.go

@@ -44,6 +44,13 @@ func (slf *JsonProcessor) MakeRpcResponse(seq uint64,err *RpcError,reply []byte)
 	}
 }
 
+func (slf *JsonProcessor) ReleaseRpcRequest(rpcRequestData IRpcRequestData){
+
+}
+func (slf *JsonProcessor) ReleaseRpcRespose(rpcRequestData IRpcResponseData){
+
+}
+
 func (slf *JsonRpcRequestData) IsReply() bool{
 	return slf.NoReply
 }

rpc/msgpprocessor.go

@@ -1,11 +1,27 @@
 package rpc
 
+import "sync"
+
 type IMsgp interface {
 	UnmarshalMsg(bts []byte) (o []byte, err error)
 	MarshalMsg(b []byte) (o []byte, err error)
 }
 
+var rpcResponeDataPool sync.Pool
+var rpcRequestDataPool sync.Pool
+
 type MsgpProcessor struct {
+
+}
+
+func init(){
+	rpcResponeDataPool.New = func()interface{}{
+		return &MsgpRpcResponseData{}
+	}
+
+	rpcRequestDataPool.New = func()interface{}{
+		return &MsgpRpcRequestData{}
+	}
 }
 
 //go:generate msgp
@@ -31,6 +47,7 @@ type MsgpRpcResponseData struct {
 
 func (slf *MsgpProcessor) Marshal(v interface{}) ([]byte, error){
 	msgp := v.(IMsgp)
+
 	return msgp.MarshalMsg(nil)
 }
 
@@ -41,15 +58,30 @@ func (slf *MsgpProcessor) Unmarshal(data []byte, v interface{}) error{
 }
 
 func (slf *MsgpProcessor) MakeRpcRequest(seq uint64,serviceMethod string,noReply bool,inParam []byte) IRpcRequestData{
-	return &MsgpRpcRequestData{Seq:seq,ServiceMethod:serviceMethod,NoReply:noReply,InParam:inParam}
+	rpcRequestData := rpcRequestDataPool.Get().(*MsgpRpcRequestData)
+	rpcRequestData.Seq = seq
+	rpcRequestData.ServiceMethod = serviceMethod
+	rpcRequestData.NoReply = noReply
+	rpcRequestData.InParam = inParam
+
+	return rpcRequestData//&MsgpRpcRequestData{Seq:seq,ServiceMethod:serviceMethod,NoReply:noReply,InParam:inParam}
 }
 
 func (slf *MsgpProcessor) MakeRpcResponse(seq uint64,err *RpcError,reply []byte) IRpcResponseData {
-	return &MsgpRpcResponseData{
+	rpcRequestData := rpcResponeDataPool.Get().(*MsgpRpcResponseData)
-		Seq:   seq,
+	rpcRequestData.Seq = seq
-		Err:   err.Error(),
+	rpcRequestData.Err = err.Error()
-		Reply: reply,
+	rpcRequestData.Reply = reply
-	}
+
+	return rpcRequestData
+}
+
+func (slf *MsgpProcessor) ReleaseRpcRequest(rpcRequestData IRpcRequestData){
+	rpcRequestDataPool.Put(rpcRequestData)
+}
+
+func (slf *MsgpProcessor) ReleaseRpcRespose(rpcRequestData IRpcResponseData){
+	rpcResponeDataPool.Put(rpcRequestData)
 }
 
 func (slf *MsgpRpcRequestData) IsReply() bool{

rpc/pbprocessor.go

@@ -22,6 +22,16 @@ func (slf *PBRpcResponseData) MakeRespone(seq uint64,err *RpcError,reply []byte)
 
 	return slf
 }
+
+func (slf *PBRpcRequestData) ReleaseRpcRequest(rpcRequestData IRpcRequestData){
+	
+}
+
+func (slf *PBRpcRequestData) ReleaseRpcRespose(rpcRequestData IRpcRequestData){
+
+}
+
+
 func (slf *PBProcessor) Marshal(v interface{}) ([]byte, error){
 	return proto.Marshal(v.(proto.Message))
 }

rpc/processor.go

@@ -1,43 +1,14 @@
 package rpc
 
-/*
-	Seq uint64             // sequence number chosen by client
-	ServiceMethod string   // format: "Service.Method"
-	NoReply bool           //是否需要返回
-	//packbody
-	InParam []byte
-
-
-
-
-type RpcResponse struct {
-	//head
-	Seq           uint64   // sequence number chosen by client
-	Err *RpcError
-
-	//returns
-	Reply []byte
-}
-*/
-type IRpcRequestData interface {
-	GetSeq() uint64
-	GetServiceMethod() string
-	GetInParam() []byte
-	IsReply() bool
-}
-
-type IRpcResponseData interface {
-	GetSeq() uint64
-	GetErr() *RpcError
-	GetReply() []byte
-}
-
 type IRpcProcessor interface {
-	Marshal(v interface{}) ([]byte, error)
+	Marshal(v interface{}) ([]byte, error) //b表示自定义缓冲区，可以填nil，由系统自动分配
 	Unmarshal(data []byte, v interface{}) error
-
 	MakeRpcRequest(seq uint64,serviceMethod string,noReply bool,inParam []byte) IRpcRequestData
 	MakeRpcResponse(seq uint64,err *RpcError,reply []byte) IRpcResponseData
+
+	ReleaseRpcRequest(rpcRequestData IRpcRequestData)
+	ReleaseRpcRespose(rpcRequestData IRpcResponseData)
 }
 
 
+

rpc/rpc.go

@@ -0,0 +1,113 @@
+package rpc
+
+import (
+	"reflect"
+	"sync"
+)
+
+type RpcRequest struct {
+	RpcRequestData IRpcRequestData
+
+	localReply interface{}
+	localParam interface{} //本地调用的参数列表
+	requestHandle RequestHandler
+	callback *reflect.Value
+}
+
+type RpcResponse struct {
+	RpcResponeData IRpcResponseData
+}
+
+type IRpcRequestData interface {
+	GetSeq() uint64
+	GetServiceMethod() string
+	GetInParam() []byte
+	IsReply() bool
+}
+
+type IRpcResponseData interface {
+	GetSeq() uint64
+	GetErr() *RpcError
+	GetReply() []byte
+}
+
+type RequestHandler func(Returns interface{},Err *RpcError)
+
+
+type Call struct {
+	Seq uint64
+	ServiceMethod string
+	Arg interface{}
+	Reply interface{}
+	Respone *RpcResponse
+	Err error
+	done          chan *Call  // Strobes when call is complete.
+	connid int
+	callback *reflect.Value
+	rpcHandler IRpcHandler
+}
+
+func (slf *Call) Clear(){
+	slf.Seq = 0
+	slf.ServiceMethod = ""
+	slf.Arg = nil
+	slf.Reply = nil
+	slf.Respone = nil
+	slf.Err = nil
+	slf.connid = 0
+	slf.callback = nil
+	slf.rpcHandler = nil
+}
+
+func (slf *Call) Done() *Call{
+	return <-slf.done
+}
+
+type RpcHandleFinder interface {
+	FindRpcHandler(serviceMethod string) IRpcHandler
+}
+
+
+var rpcResponePool sync.Pool
+var rpcRequestPool sync.Pool
+var rpcCallPool sync.Pool
+
+func init(){
+	rpcResponePool.New = func()interface{}{
+		return &RpcResponse{}
+	}
+
+	rpcRequestPool.New = func() interface{} {
+		return &RpcRequest{}
+	}
+
+	rpcCallPool.New = func() interface{} {
+		return &Call{done:make(chan *Call,1)}
+	}
+}
+
+func MakeRpcResponse() *RpcResponse{
+	return rpcResponePool.Get().(*RpcResponse)
+}
+
+func MakeRpcRequest() *RpcRequest{
+	return rpcRequestPool.Get().(*RpcRequest)
+}
+
+func MakeCall() *Call {
+	call := rpcCallPool.Get().(*Call)
+
+	return call
+}
+
+func ReleaseRpcResponse(rpcRespone *RpcResponse){
+	rpcResponePool.Put(rpcRespone)
+}
+
+func ReleaseRpcRequest(rpcRequest *RpcRequest){
+	rpcRequestPool.Put(rpcRequest)
+}
+
+func ReleaseCall(call *Call){
+	rpcCallPool.Put(call)
+}

rpc/rpchandler.go

@@ -186,7 +186,7 @@ func (slf *RpcHandler) HandlerRpcResponeCB(call *Call){
 	}else{
 		call.callback.Call([]reflect.Value{reflect.ValueOf(call.Reply),reflect.ValueOf(call.Err)})
 	}
-
+	ReleaseCall(call)
 }
 
 
@@ -203,6 +203,8 @@ func (slf *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
 				}
 		}
 	}()
+	defer processor.ReleaseRpcRequest(request.RpcRequestData)
+	defer ReleaseRpcRequest(request)
 
 	v,ok := slf.mapfunctons[request.RpcRequestData.GetServiceMethod()]
 	if ok == false {
@@ -309,6 +311,7 @@ func (slf *RpcHandler) goRpc(bCast bool,nodeId int,serviceMethod string,args int
 			}
 			//其他的rpcHandler的处理器
 			pCall := pLocalRpcServer.rpcHandlerGo(true,sMethod[0],sMethod[1],args,nil)
+			defer ReleaseCall(pCall)
 			if pCall.Err!=nil {
 				err = pCall.Err
 			}
@@ -316,11 +319,11 @@ func (slf *RpcHandler) goRpc(bCast bool,nodeId int,serviceMethod string,args int
 		}
 
 		//跨node调用
-		pCall := pClient.Go(true,serviceMethod,args,nil)
+		pCall := pClient.Go(false,serviceMethod,args,nil)
 		if pCall.Err!=nil {
 			err = pCall.Err
 		}
-
+		ReleaseCall(pCall)
 	}
 
 	return err
@@ -356,8 +359,9 @@ func (slf *RpcHandler) callRpc(nodeId int,serviceMethod string,args interface{},
 		}
 		//其他的rpcHandler的处理器
 		pCall := pLocalRpcServer.rpcHandlerGo(false,sMethod[0],sMethod[1],args,reply)
-		pResult := pCall.Done()
+		err = pCall.Done().Err
-		return pResult.Err
+		ReleaseCall(pCall)
+		return err
 	}
 
 	//跨node调用
@@ -365,8 +369,9 @@ func (slf *RpcHandler) callRpc(nodeId int,serviceMethod string,args interface{},
 	if pCall.Err != nil {
 		return pCall.Err
 	}
-	pResult := pCall.Done()
+	err = pCall.Done().Err
-	return pResult.Err
+	ReleaseCall(pCall)
+	return err
 }
 
 func (slf *RpcHandler) asyncCallRpc(nodeid int,serviceMethod string,args interface{},callback interface{}) error {
@@ -437,6 +442,8 @@ func (slf *RpcHandler) asyncCallRpc(nodeid int,serviceMethod string,args interfa
 			return nil
 		}
 		pCall := pLocalRpcServer.rpcHandlerGo(false,sMethod[0],sMethod[1],args,reply)
+		defer ReleaseCall(pCall)
+
 		pResult := pCall.Done()
 		return pResult.Err
 	}

rpc/server.go

@@ -13,46 +13,19 @@ import (
 var processor IRpcProcessor = &JsonProcessor{}
 var LittleEndian bool
 
-
-
-type Call struct {
-	Seq uint64
-	ServiceMethod string
-	Arg interface{}
-	Reply interface{}
-	Respone *RpcResponse
-	Err error
-	done          chan *Call  // Strobes when call is complete.
-	connid int
-	callback *reflect.Value
-	rpcHandler IRpcHandler
-}
-
-func (slf *Call) Done() *Call{
-	return <-slf.done
-}
-
-
-
 type Server struct {
 	functions map[interface{}]interface{}
-
 	cmdchannel chan *Call
-
 	rpcHandleFinder RpcHandleFinder
 	rpcserver *network.TCPServer
 }
 
-type RpcHandleFinder interface {
-	FindRpcHandler(serviceMethod string) IRpcHandler
-}
-
 func SetProcessor(proc IRpcProcessor) {
 	processor = proc
 }
 
 func (slf *Server) Init(rpcHandleFinder RpcHandleFinder) {
-	slf.cmdchannel = make(chan *Call,10000)
+	slf.cmdchannel = make(chan *Call,100000)
 	slf.rpcHandleFinder = rpcHandleFinder
 	slf.rpcserver = &network.TCPServer{}
 }
@@ -84,13 +57,10 @@ type RpcAgent struct {
 
 
 func (agent *RpcAgent) WriteRespone(serviceMethod string,seq uint64,reply interface{},err *RpcError) {
-	var rpcRespone RpcResponse
-
-	//rpcRespone.Seq = seq
-	//rpcRespone.Err = err
 	var mReply []byte
 	var rpcError *RpcError
 	var errM error
+
 	if err != nil {
 		rpcError = err
 	} else {
@@ -102,13 +72,15 @@ func (agent *RpcAgent) WriteRespone(serviceMethod string,seq uint64,reply interf
 		}
 	}
 
-	rpcRespone.RpcResponeData = processor.MakeRpcResponse(seq,rpcError,mReply)
+	var rpcResponse RpcResponse
-	bytes,errM :=  processor.Marshal(rpcRespone.RpcResponeData)
+	rpcResponse.RpcResponeData = processor.MakeRpcResponse(seq,rpcError,mReply)
+	bytes,errM :=  processor.Marshal(rpcResponse.RpcResponeData)
 	if errM != nil {
-		log.Error("service method %s %+v Marshal error:%+v!", serviceMethod,rpcRespone,errM)
+		log.Error("service method %s %+v Marshal error:%+v!", serviceMethod,rpcResponse,errM)
 		return
 	}
 
+	processor.ReleaseRpcRespose(rpcResponse.RpcResponeData)
 	errM = agent.conn.WriteMsg(bytes)
 	if errM != nil {
 		log.Error("Rpc %s return is error:%+v",serviceMethod,errM)
@@ -126,17 +98,21 @@ func (agent *RpcAgent) Run() {
 		}
 
 		//解析head
-		var req RpcRequest
+		req := MakeRpcRequest()
 		req.RpcRequestData = processor.MakeRpcRequest(0,"",false,nil)
 		err = processor.Unmarshal(data,req.RpcRequestData)
 		if err != nil {
 			if req.RpcRequestData.GetSeq()>0 {
 				rpcError := RpcError("rpc Unmarshal request is error")
 				agent.WriteRespone(req.RpcRequestData.GetServiceMethod(),req.RpcRequestData.GetSeq(),nil,&rpcError)
+				processor.ReleaseRpcRequest(req.RpcRequestData)
+				ReleaseRpcRequest(req)
 				continue
 			}else{
 				log.Error("rpc Unmarshal request is error: %v", err)
 				//will close tcpconn
+				processor.ReleaseRpcRequest(req.RpcRequestData)
+				ReleaseRpcRequest(req)
 				break
 			}
 		}
@@ -146,6 +122,8 @@ func (agent *RpcAgent) Run() {
 		if len(serviceMethod)!=2 {
 			rpcError := RpcError("rpc request req.ServiceMethod is error")
 			agent.WriteRespone(req.RpcRequestData.GetServiceMethod(),req.RpcRequestData.GetSeq(),nil,&rpcError)
+			processor.ReleaseRpcRequest(req.RpcRequestData)
+			ReleaseRpcRequest(req)
 			log.Debug("rpc request req.ServiceMethod is error")
 			continue
 		}
@@ -154,6 +132,8 @@ func (agent *RpcAgent) Run() {
 		if rpcHandler== nil {
 			rpcError := RpcError(fmt.Sprintf("service method %s not config!", req.RpcRequestData.GetServiceMethod()))
 			agent.WriteRespone(req.RpcRequestData.GetServiceMethod(),req.RpcRequestData.GetSeq(),nil,&rpcError)
+			processor.ReleaseRpcRequest(req.RpcRequestData)
+			ReleaseRpcRequest(req)
 			log.Error("service method %s not config!", req.RpcRequestData.GetServiceMethod())
 			continue
 		}
@@ -164,7 +144,13 @@ func (agent *RpcAgent) Run() {
 			}
 		}
 
-		rpcHandler.PushRequest(&req)
+		err = rpcHandler.PushRequest(req)
+		if err != nil {
+			rpcError := RpcError(err.Error())
+			agent.WriteRespone(req.RpcRequestData.GetServiceMethod(),req.RpcRequestData.GetSeq(),nil,&rpcError)
+			processor.ReleaseRpcRequest(req.RpcRequestData)
+			ReleaseRpcRequest(req)
+		}
 	}
 }
 
@@ -210,8 +196,8 @@ func (slf *Server) myselfRpcHandlerGo(handlerName string,methodName string, args
 
 
 func (slf *Server) rpcHandlerGo(noReply bool,handlerName string,methodName string, args interface{},reply interface{}) *Call {
-	pCall := &Call{}
+	pCall := MakeCall()//&Call{}
-	pCall.done = make( chan *Call,1)
+	//pCall.done = make( chan *Call,1)
 	rpcHandler := slf.rpcHandleFinder.FindRpcHandler(handlerName)
 	if rpcHandler== nil {
 		pCall.Err = fmt.Errorf("service method %s.%s not config!", handlerName,methodName)
@@ -219,12 +205,10 @@ func (slf *Server) rpcHandlerGo(noReply bool,handlerName string,methodName strin
 		pCall.done <- pCall
 		return pCall
 	}
-	var req RpcRequest
+	req :=  MakeRpcRequest()
 
-	//req.ServiceMethod = fmt.Sprintf("%s.%s",handlerName,methodName)
 	req.localParam = args
 	req.localReply = reply
-	//req.NoReply = noReply
 	req.RpcRequestData = processor.MakeRpcRequest(0,fmt.Sprintf("%s.%s",handlerName,methodName),noReply,nil)
 	if noReply == false {
 		req.requestHandle = func(Returns interface{},Err *RpcError){
@@ -238,15 +222,19 @@ func (slf *Server) rpcHandlerGo(noReply bool,handlerName string,methodName strin
 		}
 	}
 
-
+	err := rpcHandler.PushRequest(req)
-	rpcHandler.PushRequest(&req)
+	if err != nil {
+		processor.ReleaseRpcRequest(req.RpcRequestData)
+		ReleaseRpcRequest(req)
+		pCall.Err = err
+		pCall.done <- pCall
+	}
 
 	return pCall
 }
 
 func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,handlerName string,methodName string,args interface{},reply interface{},callback reflect.Value) error {
-	pCall := &Call{}
+	pCall := MakeCall()
-	//pCall.done = make( chan *Call,1)
 	pCall.rpcHandler = callerRpcHandler
 	pCall.callback = &callback
 	pCall.Reply = reply
@@ -257,11 +245,9 @@ func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,h
 		return err
 	}
 
-	var req RpcRequest
+	req := MakeRpcRequest()
-	//req.ServiceMethod = fmt.Sprintf("%s.%s",handlerName,methodName)
 	req.localParam = args
 	req.localReply = reply
-	//req.NoReply = noReply
 	req.RpcRequestData = processor.MakeRpcRequest(0,fmt.Sprintf("%s.%s",handlerName,methodName),noReply,nil)
 	if noReply == false {
 		req.requestHandle = func(Returns interface{},Err *RpcError){
@@ -279,8 +265,11 @@ func (slf *Server) rpcHandlerAsyncGo(callerRpcHandler IRpcHandler,noReply bool,h
 	}
 
 
-	err := rpcHandler.PushRequest(&req)
+	err := rpcHandler.PushRequest(req)
 	if err != nil {
+		processor.ReleaseRpcRequest(req.RpcRequestData)
+		ReleaseRpcRequest(req)
+
 		return err
 	}