mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

MM-24972: make GetHubForUserId lock-less and zero-alloc (#14945)

Просмотр исходного кода 
* 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
Этот коммит содержится в:
Agniva De Sarker
2020-07-07 11:23:45 +05:30
коммит произвёл GitHub
родитель de6a57cdc3
Коммит 10f5a8890c
4 изменённых файлов: 73 добавлений и 103 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

56
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)
}

50
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
}
}

68
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)
}
}

2
wsapi/api.go
Просмотреть файл

@@ -22,6 +22,4 @@ func Init(s *app.Server) {
api.InitUser()
api.InitSystem()
api.InitStatus()
a.HubStart()
}