From 6075b1cd4ee4a7a20a432d69b917eebbcd9e76c4 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Thu, 24 Oct 2024 22:50:44 +0530 Subject: [PATCH] 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 ``` --- server/channels/app/app_iface.go | 3 +-- server/channels/app/ldap.go | 9 +++---- .../app/opentracing/opentracing_layer.go | 19 ++----------- server/channels/app/platform/service.go | 6 ----- server/channels/app/platform/session.go | 21 +++------------ server/channels/app/platform/web_conn.go | 27 +++++++------------ server/channels/app/platform/web_hub.go | 2 -- server/channels/app/plugin_requests.go | 1 - server/channels/app/session.go | 4 --- server/channels/web/handlers.go | 1 - server/channels/wsapi/websocket_handler.go | 1 - server/public/shared/request/context.go | 13 +-------- 12 files changed, 21 insertions(+), 86 deletions(-) diff --git a/server/channels/app/app_iface.go b/server/channels/app/app_iface.go index 3340adf588..60e7d2b9c0 100644 --- a/server/channels/app/app_iface.go +++ b/server/channels/app/app_iface.go @@ -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 diff --git a/server/channels/app/ldap.go b/server/channels/app/ldap.go index d524a6518e..a3367c42c9 100644 --- a/server/channels/app/ldap.go +++ b/server/channels/app/ldap.go @@ -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) } }) } diff --git a/server/channels/app/opentracing/opentracing_layer.go b/server/channels/app/opentracing/opentracing_layer.go index e3ab71f00a..1a7dec97fb 100644 --- a/server/channels/app/opentracing/opentracing_layer.go +++ b/server/channels/app/opentracing/opentracing_layer.go @@ -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 { diff --git a/server/channels/app/platform/service.go b/server/channels/app/platform/service.go index 4ca223189c..4f33af3739 100644 --- a/server/channels/app/platform/service.go +++ b/server/channels/app/platform/service.go @@ -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{}, } diff --git a/server/channels/app/platform/session.go b/server/channels/app/platform/session.go index e12f22bb96..1847a0d7ba 100644 --- a/server/channels/app/platform/session.go +++ b/server/channels/app/platform/session.go @@ -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) diff --git a/server/channels/app/platform/web_conn.go b/server/channels/app/platform/web_conn.go index af2d41b51b..9c2bd3e149 100644 --- a/server/channels/app/platform/web_conn.go +++ b/server/channels/app/platform/web_conn.go @@ -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() } diff --git a/server/channels/app/platform/web_hub.go b/server/channels/app/platform/web_hub.go index a1db8b290a..7ed5622379 100644 --- a/server/channels/app/platform/web_hub.go +++ b/server/channels/app/platform/web_hub.go @@ -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 diff --git a/server/channels/app/plugin_requests.go b/server/channels/app/plugin_requests.go index 1b2f13bbda..efcdb02c09 100644 --- a/server/channels/app/plugin_requests.go +++ b/server/channels/app/plugin_requests.go @@ -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 diff --git a/server/channels/app/session.go b/server/channels/app/session.go index dcf6d70502..2ccc00dccb 100644 --- a/server/channels/app/session.go +++ b/server/channels/app/session.go @@ -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) } diff --git a/server/channels/web/handlers.go b/server/channels/web/handlers.go index 3294ee7ac0..56c349ba7e 100644 --- a/server/channels/web/handlers.go +++ b/server/channels/web/handlers.go @@ -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)) diff --git a/server/channels/wsapi/websocket_handler.go b/server/channels/wsapi/websocket_handler.go index bbd6f74052..f9ee0b9da2 100644 --- a/server/channels/wsapi/websocket_handler.go +++ b/server/channels/wsapi/websocket_handler.go @@ -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( diff --git a/server/public/shared/request/context.go b/server/public/shared/request/context.go index b7aa525582..6ebd37ceb5 100644 --- a/server/public/shared/request/context.go +++ b/server/public/shared/request/context.go @@ -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 }