From f964be699fc568ab3396c3bbd85725f4a70b7c99 Mon Sep 17 00:00:00 2001
From: Agniva De Sarker <agnivade@yahoo.co.in>
Date: Wed, 28 Oct 2020 21:42:13 +0530
Subject: [PATCH] MM-29979: make websocket writes zero-alloc (#16098)

* MM-29979: make websocket writes zero-alloc

Instead of allocating a new slice every time we write a message,
we create a json encoder for a byte buffer and then reset the buffer
every time we write a new message.

This allocates a buffer of a constant size per-connection, but gets rid
of a new allocation for every single write, reducing pressure on GC.

After taking a distrbution of message sizes from a load test, it was seen that
2k is a good enough buffer size within which 98.5% of messages remain.

Taking a look at the alloc profiles for (*webconn).writepump:
- master branch: 8.01% of total allocations totalling 64GB
- with this PR: 4.92% and 10GB.

https://mattermost.atlassian.net/browse/MM-29979

* Skip writing in case of an encoding error
---
 app/web_conn.go | 22 +++++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)

diff --git a/app/web_conn.go b/app/web_conn.go
index 6082751a4d..44662ea261 100644
--- a/app/web_conn.go
+++ b/app/web_conn.go
@@ -4,6 +4,8 @@
 package app
 
 import (
+	"bytes"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -169,6 +171,11 @@ func (wc *WebConn) writePump() {
 		wc.WebSocket.Close()
 	}()
 
+	var buf bytes.Buffer
+	// 2k is seen to be a good heuristic under which 98.5% of message sizes remain.
+	buf.Grow(1024 * 2)
+	enc := json.NewEncoder(&buf)
+
 	for {
 		select {
 		case msg, ok := <-wc.send:
@@ -201,20 +208,25 @@ func (wc *WebConn) writePump() {
 				continue
 			}
 
-			var msgBytes []byte
+			buf.Reset()
+			var err error
 			if evtOk {
 				cpyEvt := evt.SetSequence(wc.Sequence)
-				msgBytes = []byte(cpyEvt.ToJson())
+				err = enc.Encode(cpyEvt)
 				wc.Sequence++
 			} else {
-				msgBytes = []byte(msg.ToJson())
+				err = enc.Encode(msg)
+			}
+			if err != nil {
+				mlog.Warn("Error in encoding websocket message", mlog.Err(err))
+				continue
 			}
 
 			if len(wc.send) >= sendFullWarn {
 				logData := []mlog.Field{
 					mlog.String("user_id", wc.UserId),
 					mlog.String("type", msg.EventType()),
-					mlog.Int("size", len(msgBytes)),
+					mlog.Int("size", buf.Len()),
 				}
 				if evtOk {
 					logData = append(logData, mlog.String("channel_id", evt.GetBroadcast().ChannelId))
@@ -224,7 +236,7 @@ func (wc *WebConn) writePump() {
 			}
 
 			wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime))
-			if err := wc.WebSocket.WriteMessage(websocket.TextMessage, msgBytes); err != nil {
+			if err := wc.WebSocket.WriteMessage(websocket.TextMessage, buf.Bytes()); err != nil {
 				wc.logSocketErr("websocket.send", err)
 				return
 			}