From 9742a0595d3fb0563443b160423f817097c00398 Mon Sep 17 00:00:00 2001 From: Mario de Frutos Dieguez Date: Mon, 2 Mar 2020 19:37:57 +0100 Subject: [PATCH] Include Mutex for Hubs property (#13917) Hubs property has been detected by the race condition detector as a possible one because we could set the Hubs variable and also we're spawning goroutines that could ask for that property This condition is not happening right now because we're setting the Hubs variable in the server start and after that, we're spawning the goroutines but the tests are failing because of this and in the future this could be a problem --- app/server.go | 42 +++++++++++++++++++++++++++++++++++++++++- app/web_hub.go | 36 +++++++++++++++++++++++------------- app/web_hub_test.go | 2 +- 3 files changed, 65 insertions(+), 15 deletions(-) diff --git a/app/server.go b/app/server.go index 353ad2732c..31c7d612e5 100644 --- a/app/server.go +++ b/app/server.go @@ -69,7 +69,8 @@ type Server struct { EmailBatching *EmailBatchingJob EmailRateLimiter *throttled.GCRARateLimiter - Hubs []*Hub + hubsLock sync.RWMutex + hubs []*Hub HubsStopCheckingForDeadlock chan bool PushNotificationsHub PushNotificationsHub @@ -819,3 +820,42 @@ func (s *Server) shutdownDiagnostics() error { return nil } + +// GetHubs returns the list of hubs. This method is safe +// for concurrent use by multiple goroutines. +func (s *Server) GetHubs() []*Hub { + s.hubsLock.RLock() + defer s.hubsLock.RUnlock() + return s.hubs +} + +// getHub gets the element at the given index in the hubs list. This method is safe +// for concurrent use by multiple goroutines. +func (s *Server) GetHub(index int) (*Hub, error) { + s.hubsLock.RLock() + defer s.hubsLock.RUnlock() + if index >= len(s.hubs) { + return nil, errors.New("Hub element doesn't exist") + } + return s.hubs[index], nil +} + +// SetHubs sets a new list of hubs. This method is safe +// for concurrent use by multiple goroutines. +func (s *Server) SetHubs(hubs []*Hub) { + s.hubsLock.Lock() + defer s.hubsLock.Unlock() + s.hubs = hubs +} + +// SetHub sets the element at the given index in the hubs list. This method is safe +// for concurrent use by multiple goroutines. +func (s *Server) SetHub(index int, hub *Hub) error { + s.hubsLock.Lock() + defer s.hubsLock.Unlock() + if index >= len(s.hubs) { + return errors.New("Index is greater than the size of the hubs list") + } + s.hubs[index] = hub + return nil +} diff --git a/app/web_hub.go b/app/web_hub.go index e1fc036449..1ee62dd485 100644 --- a/app/web_hub.go +++ b/app/web_hub.go @@ -62,7 +62,7 @@ func (a *App) NewWebHub() *Hub { func (a *App) TotalWebsocketConnections() int { count := int64(0) - for _, hub := range a.Srv().Hubs { + for _, hub := range a.Srv().GetHubs() { count = count + atomic.LoadInt64(&hub.connectionCount) } @@ -74,13 +74,18 @@ func (a *App) HubStart() { numberOfHubs := runtime.NumCPU() * 2 mlog.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs)) - a.Srv().Hubs = make([]*Hub, numberOfHubs) + a.Srv().SetHubs(make([]*Hub, numberOfHubs)) a.Srv().HubsStopCheckingForDeadlock = make(chan bool, 1) - for i := 0; i < len(a.Srv().Hubs); i++ { - a.Srv().Hubs[i] = a.NewWebHub() - a.Srv().Hubs[i].connectionIndex = i - a.Srv().Hubs[i].Start() + for i := 0; i < len(a.Srv().GetHubs()); i++ { + newHub := a.NewWebHub() + newHub.connectionIndex = i + err := a.Srv().SetHub(i, newHub) + if err != nil { + mlog.Warn("Error starting hub", mlog.Err(err), mlog.Int("index", i)) + continue + } + newHub.Start() } go func() { @@ -93,7 +98,7 @@ func (a *App) HubStart() { for { select { case <-ticker.C: - for _, hub := range a.Srv().Hubs { + for _, hub := range a.Srv().GetHubs() { if len(hub.broadcast) >= DEADLOCK_WARN { mlog.Error( "Hub processing might be deadlock with events in the buffer", @@ -130,22 +135,27 @@ func (a *App) HubStop() { mlog.Warn("We appear to have already sent the stop checking for deadlocks command") } - for _, hub := range a.Srv().Hubs { + for _, hub := range a.Srv().GetHubs() { hub.Stop() } - a.Srv().Hubs = []*Hub{} + a.Srv().SetHubs([]*Hub{}) } func (a *App) GetHubForUserId(userId string) *Hub { - if len(a.Srv().Hubs) == 0 { + if len(a.Srv().GetHubs()) == 0 { return nil } hash := fnv.New32a() hash.Write([]byte(userId)) - index := hash.Sum32() % uint32(len(a.Srv().Hubs)) - return a.Srv().Hubs[index] + index := hash.Sum32() % uint32(len(a.Srv().GetHubs())) + hub, err := a.Srv().GetHub(int(index)) + if err != nil { + mlog.Warn("Requested hub doesn't exist", mlog.Int("hub_index", int(index))) + return nil + } + return hub } func (a *App) HubRegister(webConn *WebConn) { @@ -195,7 +205,7 @@ func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) { hub.Broadcast(message) } } else { - for _, hub := range a.Srv().Hubs { + for _, hub := range a.Srv().GetHubs() { hub.Broadcast(message) } } diff --git a/app/web_hub_test.go b/app/web_hub_test.go index ef0083b00e..9f36c3442c 100644 --- a/app/web_hub_test.go +++ b/app/web_hub_test.go @@ -77,7 +77,7 @@ func TestHubStopRaceCondition(t *testing.T) { wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) defer wc1.Close() - hub := th.App.Srv().Hubs[0] + hub := th.App.Srv().GetHubs()[0] th.App.HubStop() time.Sleep(5 * time.Second)