diff --git a/app/server.go b/app/server.go index 1407b991ac..e40d225ca5 100644 --- a/app/server.go +++ b/app/server.go @@ -8,6 +8,7 @@ import ( "crypto/ecdsa" "crypto/tls" "fmt" + "hash/maphash" "net" "net/http" "net/url" @@ -91,8 +92,8 @@ type Server struct { EmailBatching *EmailBatchingJob EmailRateLimiter *throttled.GCRARateLimiter - hubsLock sync.RWMutex hubs []*Hub + hashSeed maphash.Seed PushNotificationsHub PushNotificationsHub pushNotificationClient *http.Client // TODO: move this to it's own package @@ -177,8 +178,10 @@ func NewServer(options ...Option) (*Server, error) { RootRouter: rootRouter, LocalRouter: localRouter, licenseListeners: map[string]func(*model.License, *model.License){}, + hashSeed: maphash.MakeSeed(), } + mlog.Info("Server is initializing...") for _, option := range options { if err := option(s); err != nil { return nil, errors.Wrap(err, "failed to apply option") @@ -210,6 +213,11 @@ func NewServer(options ...Option) (*Server, error) { // Use this app logger as the global logger (eventually remove all instances of global logging) mlog.InitGlobalLogger(s.Log) + // It is important to initialize the hub only after the global logger is set + // to avoid race conditions while logging from inside the hub. + fakeApp := New(ServerConnector(s)) + fakeApp.HubStart() + if *s.Config().LogSettings.EnableDiagnostics && *s.Config().LogSettings.EnableSentry { if strings.Contains(SENTRY_DSN, "placeholder") { mlog.Warn("Sentry reporting is enabled, but SENTRY_DSN is not set. Disabling reporting.") @@ -308,8 +316,6 @@ func NewServer(options ...Option) (*Server, error) { return nil, err } - mlog.Info("Server is initializing...") - s.initEnterprise() if s.newStore == nil { @@ -393,7 +399,6 @@ func NewServer(options ...Option) (*Server, error) { s.Router = s.RootRouter.PathPrefix(subpath).Subrouter() // FakeApp: remove this when we have the ServePluginRequest and ServePluginPublicRequest migrated in the server - fakeApp := New(ServerConnector(s)) pluginsRoute := s.Router.PathPrefix("/plugins/{plugin_id:[A-Za-z0-9\\_\\-\\.]+}").Subrouter() pluginsRoute.HandleFunc("", fakeApp.ServePluginRequest) pluginsRoute.HandleFunc("/public/{public_file:.*}", fakeApp.ServePluginPublicRequest) @@ -1189,53 +1194,16 @@ 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 -} - func (s *Server) FileBackend() (filesstore.FileBackend, *model.AppError) { license := s.License() return filesstore.NewFileBackend(&s.Config().FileSettings, license != nil && *license.Features.Compliance) } func (s *Server) TotalWebsocketConnections() int { + // This method is only called after the hub is initialized. + // Therefore, no mutex is needed to protect s.hubs. count := int64(0) - for _, hub := range s.GetHubs() { + for _, hub := range s.hubs { count = count + atomic.LoadInt64(&hub.connectionCount) } diff --git a/app/web_hub.go b/app/web_hub.go index 8f6f100e8a..1eec80b38d 100644 --- a/app/web_hub.go +++ b/app/web_hub.go @@ -4,7 +4,7 @@ package app import ( - "hash/fnv" + "hash/maphash" "runtime" "runtime/debug" "strconv" @@ -82,18 +82,16 @@ func (a *App) HubStart() { numberOfHubs := runtime.NumCPU() * 2 mlog.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs)) - a.Srv().SetHubs(make([]*Hub, numberOfHubs)) + hubs := make([]*Hub, numberOfHubs) - 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() + for i := 0; i < numberOfHubs; i++ { + hubs[i] = a.NewWebHub() + hubs[i].connectionIndex = i + hubs[i].Start() } + // Assigning to the hubs slice without any mutex is fine because it is only assigned once + // during the start of the program and always read from after that. + a.srv.hubs = hubs } func (a *App) invalidateCacheForUserSkipClusterSend(userId string) { @@ -116,11 +114,9 @@ func (a *App) InvalidateWebConnSessionCacheForUser(userId string) { func (s *Server) HubStop() { mlog.Info("stopping websocket hub connections") - for _, hub := range s.GetHubs() { + for _, hub := range s.hubs { hub.Stop() } - - s.SetHubs([]*Hub{}) } func (a *App) HubStop() { @@ -129,19 +125,15 @@ func (a *App) HubStop() { // GetHubForUserId returns the hub for a given user id. func (s *Server) GetHubForUserId(userId string) *Hub { - if len(s.GetHubs()) == 0 { - return nil - } - - hash := fnv.New32a() + // TODO: check if caching the userId -> hub mapping + // is worth the memory tradeoff. + // https://mattermost.atlassian.net/browse/MM-26629. + var hash maphash.Hash + hash.SetSeed(s.hashSeed) hash.Write([]byte(userId)) - index := hash.Sum32() % uint32(len(s.GetHubs())) - hub, err := s.GetHub(int(index)) - if err != nil { - mlog.Warn("Requested hub doesn't exist", mlog.Int("hub_index", int(index))) - return nil - } - return hub + index := hash.Sum64() % uint64(len(s.hubs)) + + return s.hubs[int(index)] } func (a *App) GetHubForUserId(userId string) *Hub { @@ -207,7 +199,7 @@ func (s *Server) PublishSkipClusterSend(message *model.WebSocketEvent) { hub.Broadcast(message) } } else { - for _, hub := range s.GetHubs() { + for _, hub := range s.hubs { hub.Broadcast(message) } } @@ -427,8 +419,8 @@ func (h *Hub) Start() { case webSessionMessage := <-h.checkRegistered: conns := connIndex.ForUser(webSessionMessage.userId) var isRegistered bool - for _, item := range conns { - if item.sessionToken.Load().(string) == webSessionMessage.sessionToken { + for _, conn := range conns { + if conn.GetSessionToken() == webSessionMessage.sessionToken { isRegistered = true } } diff --git a/app/web_hub_test.go b/app/web_hub_test.go index 60dbfea599..c1cce92b99 100644 --- a/app/web_hub_test.go +++ b/app/web_hub_test.go @@ -15,13 +15,11 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/mattermost/mattermost-server/v5/mlog" "github.com/mattermost/mattermost-server/v5/model" ) func dummyWebsocketHandler(t *testing.T) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { - mlog.Debug("dummyWebsocketHandler") upgrader := &websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, @@ -72,17 +70,17 @@ func TestHubStopWithMultipleConnections(t *testing.T) { // block the caller indefinitely. func TestHubStopRaceCondition(t *testing.T) { th := Setup(t).InitBasic() - defer th.TearDown() - + // We do not call TearDown because th.TearDown shuts down the hub again. And hub close is not idempotent. + // Making it idempotent is not really important to the server because close only happens once. + // So we just use this quick hack for the test. s := httptest.NewServer(dummyWebsocketHandler(t)) th.App.HubStart() wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) defer wc1.Close() - hub := th.App.Srv().GetHubs()[0] + hub := th.App.Srv().hubs[0] th.App.HubStop() - time.Sleep(5 * time.Second) done := make(chan bool) go func() { @@ -185,6 +183,34 @@ func TestHubConnIndex(t *testing.T) { }) } +func TestHubIsRegistered(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + s := httptest.NewServer(dummyWebsocketHandler(t)) + defer s.Close() + + th.App.HubStart() + wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) + wc2 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) + wc3 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) + defer wc1.Close() + defer wc2.Close() + defer wc3.Close() + + session1 := wc1.session.Load().(*model.Session) + + assert.True(t, th.App.SessionIsRegistered(*session1)) + assert.True(t, th.App.SessionIsRegistered(*wc2.session.Load().(*model.Session))) + assert.True(t, th.App.SessionIsRegistered(*wc3.session.Load().(*model.Session))) + + session4, appErr := th.App.CreateSession(&model.Session{ + UserId: th.BasicUser2.Id, + }) + require.Nil(t, appErr) + assert.False(t, th.App.SessionIsRegistered(*session4)) +} + // Always run this with -benchtime=0.1s // See: https://github.com/golang/go/issues/27217. func BenchmarkHubConnIndex(b *testing.B) { @@ -229,30 +255,16 @@ func BenchmarkHubConnIndex(b *testing.B) { }) } -func TestHubIsRegistered(t *testing.T) { - th := Setup(t).InitBasic() +var hubSink *Hub + +func BenchmarkGetHubForUserId(b *testing.B) { + th := Setup(b).InitBasic() defer th.TearDown() - s := httptest.NewServer(dummyWebsocketHandler(t)) - defer s.Close() - th.App.HubStart() - wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) - wc2 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) - wc3 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) - defer wc1.Close() - defer wc2.Close() - defer wc3.Close() - session1 := wc1.session.Load().(*model.Session) - - assert.True(t, th.App.SessionIsRegistered(*session1)) - assert.True(t, th.App.SessionIsRegistered(*wc2.session.Load().(*model.Session))) - assert.True(t, th.App.SessionIsRegistered(*wc3.session.Load().(*model.Session))) - - session4, appErr := th.App.CreateSession(&model.Session{ - UserId: th.BasicUser2.Id, - }) - require.Nil(t, appErr) - assert.False(t, th.App.SessionIsRegistered(*session4)) + b.ResetTimer() + for i := 0; i < b.N; i++ { + hubSink = th.Server.GetHubForUserId(th.BasicUser.Id) + } } diff --git a/wsapi/api.go b/wsapi/api.go index 1c601c5c70..8b1fdf7adf 100644 --- a/wsapi/api.go +++ b/wsapi/api.go @@ -22,6 +22,4 @@ func Init(s *app.Server) { api.InitUser() api.InitSystem() api.InitStatus() - - a.HubStart() }