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
Этот коммит содержится в:
коммит произвёл
GitHub
родитель
bb095c2a1a
Коммит
f964be699f
@@ -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
|
||||
}
|
||||
|
||||
Ссылка в новой задаче
Block a user