ethstats: avoid concurrent write on websocket, fixes #21403

ethstats/ethstats.go

@@ -28,6 +28,7 @@ import (
  "runtime"
  "strconv"
  "strings"
+ "sync"
  "time"
 
  "github.com/ethereum/go-ethereum/common"
@@ -80,6 +81,18 @@ type Service struct {
 
  pongCh chan struct{} // Pong notifications are fed into this channel
  histCh chan []uint64 // History request block numbers are fed into this channel
+
+ // Gorilla websocket docs:
+ // Connections support one concurrent reader and one concurrent writer.
+ // Applications are responsible for ensuring that no more than one goroutine calls the write methods
+ // - NextWriter, SetWriteDeadline, WriteMessage, WriteJSON, EnableWriteCompression, SetCompressionLevel
+ // concurrently and that no more than one goroutine calls the read methods
+ // - NextReader, SetReadDeadline, ReadMessage, ReadJSON, SetPongHandler, SetPingHandler
+ // concurrently.
+ // The Close and WriteControl methods can be called concurrently with all other methods.
+ //
+ // In our case, we use a single mutex for both reading and writing.
+ connMu sync.Mutex // Mutex to prevent concurrent write/read on the websocket connection
 }
 
 // New returns a monitoring service ready for stats reporting.
@@ -299,19 +312,24 @@ func (s *Service) readLoop(conn *websocket.Conn) {
  for {
  // Retrieve the next generic network packet and bail out on error
  var blob json.RawMessage
+ s.connMu.Lock()
  if err := conn.ReadJSON(&blob); err != nil {
  log.Warn("Failed to retrieve stats server message", "err", err)
+ s.connMu.Unlock()
  return
  }
  // If the network packet is a system ping, respond to it directly
  var ping string
  if err := json.Unmarshal(blob, &ping); err == nil && strings.HasPrefix(ping, "primus::ping::") {
  if err := conn.WriteJSON(strings.Replace(ping, "ping", "pong", -1)); err != nil {
  log.Warn("Failed to respond to system ping message", "err", err)
+ s.connMu.Unlock()
  return
  }
+ s.connMu.Unlock()
  continue
  }
+ s.connMu.Unlock()
  // Not a system ping, try to decode an actual state message
  var msg map[string][]interface{}
  if err := json.Unmarshal(blob, &msg); err != nil {
@@ -432,6 +450,8 @@ func (s *Service) login(conn *websocket.Conn) error {
  login := map[string][]interface{}{
  "emit": {"hello", auth},
  }
+ s.connMu.Lock()
+ defer s.connMu.Unlock()
  if err := conn.WriteJSON(login); err != nil {
  return err
  }
@@ -474,6 +494,8 @@ func (s *Service) reportLatency(conn *websocket.Conn) error {
  "clientTime": start.String(),
  }},
  }
+ s.connMu.Lock()
+ defer s.connMu.Unlock()
  if err := conn.WriteJSON(ping); err != nil {
  return err
  }
@@ -547,6 +569,8 @@ func (s *Service) reportBlock(conn *websocket.Conn, block *types.Block) error {
  report := map[string][]interface{}{
  "emit": {"block", stats},
  }
+ s.connMu.Lock()
+ defer s.connMu.Unlock()
  return conn.WriteJSON(report)
 }
 
@@ -661,6 +685,8 @@ func (s *Service) reportHistory(conn *websocket.Conn, list []uint64) error {
  report := map[string][]interface{}{
  "emit": {"history", stats},
  }
+ s.connMu.Unlock()
+ defer s.connMu.Unlock()
  return conn.WriteJSON(report)
 }
 
@@ -691,6 +717,8 @@ func (s *Service) reportPending(conn *websocket.Conn) error {
  report := map[string][]interface{}{
  "emit": {"pending", stats},
  }
+ s.connMu.Lock()
+ defer s.connMu.Unlock()
  return conn.WriteJSON(report)
 }
 
@@ -746,5 +774,7 @@ func (s *Service) reportStats(conn *websocket.Conn) error {
  report := map[string][]interface{}{
  "emit": {"stats", stats},
  }
+ s.connMu.Lock()
+ defer s.connMu.Unlock()
  return conn.WriteJSON(report)
 }