MM-61225: Revert session pooling (#28901)

This originated from https://github.com/mattermost/mattermost/issues/15249.

However, the original idea was discarded https://github.com/mattermost/mattermost/issues/15249#issuecomment-709713065
as being too complicated to implement. Then I had another
idea to implement it just for session objects.

My thinking was that since every single request allocates a new
session struct, it would be good to use a sync.Pool for that.

However, 4 years later, now we know that the primary bottleneck
in app performance comes from websocket event marshalling.
Therefore, while it would be good to do this, it is difficult
to do it correctly (as shown by the numerous racy tests).

Hence, reverting this.

```release-note
NONE
```
Этот коммит содержится в:
Agniva De Sarker
2024-10-24 22:50:44 +05:30
коммит произвёл GitHub
родитель a8cf496e31
Коммит 6075b1cd4e
12 изменённых файлов: 21 добавлений и 86 удалений

Просмотреть файл

@@ -352,7 +352,7 @@ type AppIface interface {
// SyncLdap starts an LDAP sync job.
// If includeRemovedMembers is true, then members who left or were removed from a team/channel will
// be re-added; otherwise, they will not be re-added.
SyncLdap(rctx request.CTX, includeRemovedMembers bool)
SyncLdap(c request.CTX, includeRemovedMembers bool)
// SyncPlugins synchronizes the plugins installed locally
// with the plugin bundles available in the file store.
SyncPlugins() *model.AppError
@@ -1047,7 +1047,6 @@ type AppIface interface {
RestoreTeam(teamID string) *model.AppError
RestrictUsersGetByPermissions(c request.CTX, userID string, options *model.UserGetOptions) (*model.UserGetOptions, *model.AppError)
RestrictUsersSearchByPermissions(c request.CTX, userID string, options *model.UserSearchOptions) (*model.UserSearchOptions, *model.AppError)
ReturnSessionToPool(session *model.Session)
RevokeAccessToken(c request.CTX, token string) *model.AppError
RevokeAllSessions(c request.CTX, userID string) *model.AppError
RevokeSession(c request.CTX, session *model.Session) *model.AppError

Просмотреть файл

@@ -17,21 +17,20 @@ import (
// SyncLdap starts an LDAP sync job.
// If includeRemovedMembers is true, then members who left or were removed from a team/channel will
// be re-added; otherwise, they will not be re-added.
func (a *App) SyncLdap(rctx request.CTX, includeRemovedMembers bool) {
rctx = rctx.Clone()
func (a *App) SyncLdap(c request.CTX, includeRemovedMembers bool) {
a.Srv().Go(func() {
if license := a.Srv().License(); license != nil && *license.Features.LDAP {
if !*a.Config().LdapSettings.EnableSync {
rctx.Logger().Error("LdapSettings.EnableSync is set to false. Skipping LDAP sync.")
c.Logger().Error("LdapSettings.EnableSync is set to false. Skipping LDAP sync.")
return
}
ldapI := a.Ldap()
if ldapI == nil {
rctx.Logger().Error("Not executing ldap sync because ldap is not available")
c.Logger().Error("Not executing ldap sync because ldap is not available")
return
}
ldapI.StartSynchronizeJob(rctx, false, includeRemovedMembers)
ldapI.StartSynchronizeJob(c, false, includeRemovedMembers)
}
})
}

Просмотреть файл

@@ -15056,21 +15056,6 @@ func (a *OpenTracingAppLayer) RestrictUsersSearchByPermissions(c request.CTX, us
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) ReturnSessionToPool(session *model.Session) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.ReturnSessionToPool")
a.ctx = newCtx
a.app.Srv().Store().SetContext(newCtx)
defer func() {
a.app.Srv().Store().SetContext(origCtx)
a.ctx = origCtx
}()
defer span.Finish()
a.app.ReturnSessionToPool(session)
}
func (a *OpenTracingAppLayer) RevokeAccessToken(c request.CTX, token string) *model.AppError {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.RevokeAccessToken")
@@ -17464,7 +17449,7 @@ func (a *OpenTracingAppLayer) SwitchOAuthToEmail(c request.CTX, email string, pa
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) SyncLdap(rctx request.CTX, includeRemovedMembers bool) {
func (a *OpenTracingAppLayer) SyncLdap(c request.CTX, includeRemovedMembers bool) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SyncLdap")
@@ -17476,7 +17461,7 @@ func (a *OpenTracingAppLayer) SyncLdap(rctx request.CTX, includeRemovedMembers b
}()
defer span.Finish()
a.app.SyncLdap(rctx, includeRemovedMembers)
a.app.SyncLdap(c, includeRemovedMembers)
}
func (a *OpenTracingAppLayer) SyncPlugins() *model.AppError {

Просмотреть файл

@@ -50,7 +50,6 @@ type PlatformService struct {
cacheProvider cache.Provider
statusCache cache.Cache
sessionCache cache.Cache
sessionPool sync.Pool
asymmetricSigningKey atomic.Pointer[ecdsa.PrivateKey]
clientConfig atomic.Value
@@ -124,11 +123,6 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
WebSocketRouter: &WebSocketRouter{
handlers: make(map[string]webSocketHandler),
},
sessionPool: sync.Pool{
New: func() any {
return &model.Session{}
},
},
licenseListeners: map[string]func(*model.License, *model.License){},
additionalClusterHandlers: map[model.ClusterEvent]einterfaces.ClusterMessageHandler{},
}

Просмотреть файл

@@ -14,19 +14,6 @@ import (
"github.com/mattermost/mattermost/server/v8/platform/services/cache"
)
func (ps *PlatformService) ReturnSessionToPool(session *model.Session) {
if session != nil {
session.Id = ""
// All existing prop fields are cleared once the session is retrieved from the pool.
// To speed up that process, clear the props here to avoid doing that in the hot path.
//
// If the request handler spawns a goroutine that uses the session, it might race with this code.
// In that case, the handler should copy the session and use the copy in the goroutine.
clear(session.Props)
ps.sessionPool.Put(session)
}
}
func (ps *PlatformService) CreateSession(c request.CTX, session *model.Session) (*model.Session, error) {
session.Token = ""
@@ -133,8 +120,8 @@ func (ps *PlatformService) ClearAllUsersSessionCache() {
}
func (ps *PlatformService) GetSession(c request.CTX, token string) (*model.Session, error) {
var session = ps.sessionPool.Get().(*model.Session)
if err := ps.sessionCache.Get(token, session); err == nil {
var session model.Session
if err := ps.sessionCache.Get(token, &session); err == nil {
if m := ps.metricsIFace; m != nil {
m.IncrementMemCacheHitCounterSession()
}
@@ -145,7 +132,7 @@ func (ps *PlatformService) GetSession(c request.CTX, token string) (*model.Sessi
}
if session.Id != "" {
return session, nil
return &session, nil
}
return ps.GetSessionContext(c, token)
@@ -206,8 +193,6 @@ func (ps *PlatformService) RevokeSession(c request.CTX, session *model.Session)
func (ps *PlatformService) RevokeAccessToken(c request.CTX, token string) error {
session, _ := ps.GetSession(c, token)
defer ps.ReturnSessionToPool(session)
schan := make(chan error, 1)
go func() {
schan <- ps.Store.Session().Remove(token)

Просмотреть файл

@@ -194,6 +194,15 @@ func (ps *PlatformService) PopulateWebConnConfig(s *model.Session, cfg *WebConnC
// NewWebConn returns a new WebConn instance.
func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runner HookRunner) *WebConn {
userID := cfg.Session.UserId
session := cfg.Session
if cfg.Session.UserId != "" {
ps.Go(func() {
ps.SetStatusOnline(userID, false)
ps.UpdateLastActivityAtIfNeeded(session)
})
}
// Disable TCP_NO_DELAY for higher throughput
var tcpConn *net.TCPConn
switch conn := cfg.WebSocket.UnderlyingConn().(type) {
@@ -245,23 +254,9 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runn
remoteAddress: cfg.RemoteAddress,
xForwardedFor: cfg.XForwardedFor,
}
wc.SetSession(&cfg.Session)
userID := cfg.Session.UserId
if userID != "" {
// UpdateLastActivityAtIfNeeded might block if the Hub is busy.
// Create a goroutine to avoid blocking the creation of the websocket connection.
ps.Go(func() {
ps.SetStatusOnline(userID, false)
session := wc.GetSession()
if session != nil {
ps.UpdateLastActivityAtIfNeeded(*session)
}
})
}
wc.Active.Store(cfg.Active)
wc.SetSession(&cfg.Session)
wc.SetSessionToken(cfg.Session.Token)
wc.SetSessionExpiresAt(cfg.Session.ExpiresAt)
wc.SetConnectionID(cfg.ConnectionID)
@@ -393,8 +388,6 @@ func (wc *WebConn) GetSession() *model.Session {
// SetSession sets the session of the connection.
func (wc *WebConn) SetSession(v *model.Session) {
// Clone the session first as WebConn takes ownership of the object
// and the web.Hub will return it to the [sync.Pool] once the WebConn gets removed.
if v != nil {
v = v.DeepCopy()
}

Просмотреть файл

@@ -649,8 +649,6 @@ func (i *hubConnectionIndex) Add(wc *WebConn) {
}
func (i *hubConnectionIndex) Remove(wc *WebConn) {
wc.Platform.ReturnSessionToPool(wc.GetSession())
userConnIndex, ok := i.byConnection[wc]
if !ok {
return

Просмотреть файл

@@ -148,7 +148,6 @@ func (ch *Channels) servePluginRequest(w http.ResponseWriter, r *http.Request, h
r.Header.Del("Mattermost-User-Id")
if token != "" {
session, err := New(ServerConnector(ch)).GetSession(token)
defer ch.srv.platform.ReturnSessionToPool(session)
csrfCheckPassed := false

Просмотреть файл

@@ -223,10 +223,6 @@ func (a *App) RevokeSessionsFromAllUsers() *model.AppError {
return nil
}
func (a *App) ReturnSessionToPool(session *model.Session) {
a.ch.srv.platform.ReturnSessionToPool(session)
}
func (a *App) ClearSessionCacheForUser(userID string) {
a.ch.srv.platform.ClearUserSessionCache(userID)
}

Просмотреть файл

@@ -292,7 +292,6 @@ func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if token != "" && tokenLocation != app.TokenLocationCloudHeader && tokenLocation != app.TokenLocationRemoteClusterHeader {
session, err := c.App.GetSession(token)
defer c.App.ReturnSessionToPool(session)
if err != nil {
c.Logger.Info("Invalid session", mlog.Err(err))

Просмотреть файл

@@ -30,7 +30,6 @@ func (wh webSocketHandler) ServeWebSocket(conn *platform.WebConn, r *model.WebSo
return
}
session, sessionErr := wh.app.GetSession(conn.GetSessionToken())
defer wh.app.ReturnSessionToPool(session)
if sessionErr != nil {
mlog.Error(

Просмотреть файл

@@ -54,19 +54,9 @@ func TestContext(t testing.TB) *Context {
return EmptyContext(logger)
}
// Clone creates a deep copy of [CTX].
// It should only be used to pass a [CTX] to a separate goroutine that
// has a longer lifespan than the main goroutine handling the request.
// It should be used sparsely as coping [CTX] is often unnecessary.
func (c *Context) Clone() CTX {
return c.clone()
}
// clone creates a deep copy of [Context], allowing clones to apply per-request changes.
// It unexported to prevent leaking the [Context] type from the [CTX] interface.
// clone creates a shallow copy of Context, allowing clones to apply per-request changes.
func (c *Context) clone() *Context {
cCopy := *c
cCopy.session = *c.session.DeepCopy()
return &cCopy
}
@@ -183,5 +173,4 @@ type CTX interface {
WithLogger(mlog.LoggerIFace) CTX
WithContext(ctx context.Context) CTX
With(func(ctx CTX) CTX) CTX
Clone() CTX
}