ethstats: split read and write lock, otherwise they'll lock up

This commit is contained in:
Péter Szilágyi 2020-08-10 14:33:22 +03:00
parent cbbc54c495
commit 9f45d6efae
No known key found for this signature in database
GPG Key ID: E9AE538CEDF8293D

@ -99,19 +99,20 @@ type Service struct {
// connWrapper is a wrapper to prevent concurrent-write or concurrent-read on the // connWrapper is a wrapper to prevent concurrent-write or concurrent-read on the
// websocket. // websocket.
// From 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.
// //
// The connWrapper uses a single mutex for both reading and writing. // From 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.
type connWrapper struct { type connWrapper struct {
conn *websocket.Conn conn *websocket.Conn
mu sync.Mutex
rlock sync.Mutex
wlock sync.Mutex
} }
func newConnectionWrapper(conn *websocket.Conn) *connWrapper { func newConnectionWrapper(conn *websocket.Conn) *connWrapper {
@ -120,15 +121,17 @@ func newConnectionWrapper(conn *websocket.Conn) *connWrapper {
// WriteJSON wraps corresponding method on the websocket but is safe for concurrent calling // WriteJSON wraps corresponding method on the websocket but is safe for concurrent calling
func (w *connWrapper) WriteJSON(v interface{}) error { func (w *connWrapper) WriteJSON(v interface{}) error {
w.mu.Lock() w.wlock.Lock()
defer w.mu.Unlock() defer w.wlock.Unlock()
return w.conn.WriteJSON(v) return w.conn.WriteJSON(v)
} }
// ReadJSON wraps corresponding method on the websocket but is safe for concurrent calling // ReadJSON wraps corresponding method on the websocket but is safe for concurrent calling
func (w *connWrapper) ReadJSON(v interface{}) error { func (w *connWrapper) ReadJSON(v interface{}) error {
w.mu.Lock() w.rlock.Lock()
defer w.mu.Unlock() defer w.rlock.Unlock()
return w.conn.ReadJSON(v) return w.conn.ReadJSON(v)
} }
@ -275,6 +278,7 @@ func (s *Service) loop() {
continue continue
} }
go s.readLoop(conn) go s.readLoop(conn)
// Send the initial stats so our node looks decent from the get go // Send the initial stats so our node looks decent from the get go
if err = s.report(conn); err != nil { if err = s.report(conn); err != nil {
log.Warn("Initial stats report failed", "err", err) log.Warn("Initial stats report failed", "err", err)