Merge pull request #5813 from nekto0n/encoder-pointer

rafthttp: use pointers to avoid extra copies upon message encoding

rafthttp/coder.go

@@ -18,7 +18,7 @@ import "github.com/coreos/etcd/raft/raftpb"
 
 type encoder interface {
 	// encode encodes the given message to an output stream.
-	encode(m raftpb.Message) error
+	encode(m *raftpb.Message) error
 }
 
 type decoder interface {

rafthttp/msg_codec.go

@@ -28,11 +28,11 @@ type messageEncoder struct {
 	w io.Writer
 }
 
-func (enc *messageEncoder) encode(m raftpb.Message) error {
+func (enc *messageEncoder) encode(m *raftpb.Message) error {
 	if err := binary.Write(enc.w, binary.BigEndian, uint64(m.Size())); err != nil {
 		return err
 	}
-	_, err := enc.w.Write(pbutil.MustMarshal(&m))
+	_, err := enc.w.Write(pbutil.MustMarshal(m))
 	return err
 }
 

rafthttp/msg_codec_test.go

@@ -48,7 +48,7 @@ func TestMessage(t *testing.T) {
 	for i, tt := range tests {
 		b := &bytes.Buffer{}
 		enc := &messageEncoder{w: b}
-		if err := enc.encode(tt); err != nil {
+		if err := enc.encode(&tt); err != nil {
 			t.Errorf("#%d: unexpected encode message error: %v", i, err)
 			continue
 		}

rafthttp/msgappv2_codec.go

@@ -82,7 +82,7 @@ func newMsgAppV2Encoder(w io.Writer, fs *stats.FollowerStats) *msgAppV2Encoder {
 	}
 }
 
-func (enc *msgAppV2Encoder) encode(m raftpb.Message) error {
+func (enc *msgAppV2Encoder) encode(m *raftpb.Message) error {
 	start := time.Now()
 	switch {
 	case isLinkHeartbeatMessage(m):
@@ -135,7 +135,7 @@ func (enc *msgAppV2Encoder) encode(m raftpb.Message) error {
 			return err
 		}
 		// write message
-		if _, err := enc.w.Write(pbutil.MustMarshal(&m)); err != nil {
+		if _, err := enc.w.Write(pbutil.MustMarshal(m)); err != nil {
 			return err
 		}
 

rafthttp/msgappv2_codec_test.go

@@ -107,7 +107,7 @@ func TestMsgAppV2(t *testing.T) {
 	dec := newMsgAppV2Decoder(b, types.ID(2), types.ID(1))
 
 	for i, tt := range tests {
-		if err := enc.encode(tt); err != nil {
+		if err := enc.encode(&tt); err != nil {
 			t.Errorf("#%d: unexpected encode message error: %v", i, err)
 			continue
 		}

rafthttp/snapshot_sender.go

@@ -143,7 +143,7 @@ func createSnapBody(merged snap.Message) io.ReadCloser {
 	buf := new(bytes.Buffer)
 	enc := &messageEncoder{w: buf}
 	// encode raft message
-	if err := enc.encode(merged.Message); err != nil {
+	if err := enc.encode(&merged.Message); err != nil {
 		plog.Panicf("encode message error (%v)", err)
 	}
 

rafthttp/stream.go

@@ -84,7 +84,7 @@ var (
 	linkHeartbeatMessage = raftpb.Message{Type: raftpb.MsgHeartbeat}
 )
 
-func isLinkHeartbeatMessage(m raftpb.Message) bool {
+func isLinkHeartbeatMessage(m *raftpb.Message) bool {
 	return m.Type == raftpb.MsgHeartbeat && m.From == 0 && m.To == 0
 }
 
@@ -146,7 +146,7 @@ func (cw *streamWriter) run() {
 	for {
 		select {
 		case <-heartbeatc:
-			err := enc.encode(linkHeartbeatMessage)
+			err := enc.encode(&linkHeartbeatMessage)
 			unflushed += linkHeartbeatMessage.Size()
 			if err == nil {
 				flusher.Flush()
@@ -163,7 +163,7 @@ func (cw *streamWriter) run() {
 			heartbeatc, msgc = nil, nil
 
 		case m := <-msgc:
-			err := enc.encode(m)
+			err := enc.encode(&m)
 			if err == nil {
 				unflushed += m.Size()
 
@@ -354,7 +354,7 @@ func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
 			continue
 		}
 
-		if isLinkHeartbeatMessage(m) {
+		if isLinkHeartbeatMessage(&m) {
 			// raft is not interested in link layer
 			// heartbeat message, so we should ignore
 			// it.