From 10f5a8890cc567d343ea8ee4aee145a6201896ce Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Tue, 7 Jul 2020 11:23:45 +0530 Subject: [PATCH] MM-24972: make GetHubForUserId lock-less and zero-alloc (#14945) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * MM-24972: make GetHubForUserId lock-less and zero-alloc The hubs aren't meant to be modified after the server starts up. But the key problem was the SetHubs method which would zero out the underlying hubs slice. This wasn't ideally necessary because an hub would only be stopped once during server shutdown. However, this was required by a test which would shutdown the hub twice. And since the hub shutdown isn't idempotent, zeroing the slice would just skip over shutting down the hub again. To improve the overall situation, we apply several optimizations. - We use the new hash/maphash package which exposes Go runtime's internal hash algorithms to be used as a package. This is much faster than hash/fnv. - We move around the initialization of the hub to happen before the metrics server starts. This allows us to initialize the hub before any of the hub elements are being accessed. - To make the test run successfully, we do not call th.TearDown. This is fine for a test, because anyways the test process would eventually stop and relinquish the resources to the OS. This allows us to completely remove any mutexes and thereby we can remove all the methods and any edge-case checks related to index being out of bounds. As a result, the fast path becomes very straightforward and zero-alloc. name old time/op new time/op delta GetHubForUserId-8 116ns ± 1% 38ns ± 7% -67.22% (p=0.000 n=10+10) name old alloc/op new alloc/op delta GetHubForUserId-8 36.0B ± 0% 0.0B -100.00% (p=0.000 n=10+10) name old allocs/op new allocs/op delta GetHubForUserId-8 2.00 ± 0% 0.00 -100.00% (p=0.000 n=10+10) Manually tested with some load testing and running Hub tests in -race mode. * remove mutex * incorporate review comments --- app/server.go | 56 ++++++++----------------------------- app/web_hub.go | 50 ++++++++++++++------------------- app/web_hub_test.go | 68 ++++++++++++++++++++++++++------------------- wsapi/api.go | 2 -- 4 files changed, 73 insertions(+), 103 deletions(-) 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() }