From e39569b35845d4caa1a95c157b3ea1e90b90c59f Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Thu, 23 Apr 2020 13:16:18 +0530 Subject: [PATCH] MM-23805: Refactor web_hub (#14277) * MM-23800: remove goroutineID and stack printing Each hub has a goroutineID which is calculated with a known hack. The FAQ clearly explains why goroutines don't have an id: https://golang.org/doc/faq#no_goroutine_id. We only added that because sometimes the hub would be deadlocked and having the goroutineID would be useful when getting the stack trace. This is also problematic in stress tests because the hubs would frequently get overloaded and the logs would unnecessarily have stack traces. But that was in the past, and we have done extensive testing with load tests and fuzz testing to smooth any rough edges remaining. Including adding additional metrics for hub buffer size. Monitoring the metrics is a better way to approach this problem. Therefore, we remove these kludges from the code. * Also remove deadlock checking code There is no need for that anymore since we are getting rid of the stack printing anyways. Let's do a wholesale refactor and clean up the codebase. * MM-23805: Refactor web_hub This is a beginning of the refactoring of the websocket code. To start off with, we unexport some methods and constants which did not need to be exported. There are more remaining but some are out of scope for this PR. The main chunk of refactor is to unexport the webconn send channel which was the main cause of panics. Since we were directly sending to the connection from various parts of the codebase, it would be possible that the send channel would be closed and we could still send a message. This would crash the server. To fix this, we refactor the code to centralize all sending from the main hub goroutine. This means we can leverage the connections map to check if the connection exists or not, and only then send the message. We also move the cluster calls to cluster.go. * bring back cluster code inside hub * Incorporate review comments * Address review comments * rename index * MM-23807: Refactor web_conn - Unexport some struct fields and constants which are not necessary to be accessed from outside the package. This will help us moving the entire websocket handling code to a separate package later. - Change some empty string checks to check for empty string rather than doing a len check which is more idiomatic. Both of them compile to the same code. So it doesn't make a difference performance-wise. - Remove redundant ToJson calls to get the length. - Incorporate review comments - Unexport some more methods * Fix field name * Run make app-layers * Add note on hub check --- app/app_iface.go | 24 ++-- app/cluster_handlers.go | 2 +- app/web_conn.go | 196 ++++++++++++++-------------- app/web_conn_test.go | 12 +- app/web_hub.go | 256 ++++++++++++++++++++++--------------- app/web_hub_test.go | 2 +- app/websocket_router.go | 24 ++-- wsapi/websocket_handler.go | 13 +- 8 files changed, 299 insertions(+), 230 deletions(-) diff --git a/app/app_iface.go b/app/app_iface.go index a8874fbe18..8cde6f109c 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -154,6 +154,8 @@ type AppIface interface { GetEmojiStaticUrl(emojiName string) (string, *model.AppError) // GetEnvironmentConfig returns a map of configuration keys whose values have been overridden by an environment variable. GetEnvironmentConfig() map[string]interface{} + // GetHubForUserId returns the hub for a given user id. + GetHubForUserId(userId string) *Hub // GetLdapGroup retrieves a single LDAP group by the given LDAP group id. GetLdapGroup(ldapGroupID string) (*model.Group, *model.AppError) // GetMarketplacePlugins returns a list of plugins from the marketplace-server, @@ -183,6 +185,14 @@ type AppIface interface { GetTeamSchemeChannelRoles(teamId string) (guestRoleName string, userRoleName string, adminRoleName string, err *model.AppError) // GetTotalUsersStats is used for the DM list total GetTotalUsersStats(viewRestrictions *model.ViewUsersRestrictions) (*model.UsersStats, *model.AppError) + // HubRegister registers a connection to a hub. + HubRegister(webConn *WebConn) + // HubStart starts all the hubs. + HubStart() + // HubStop stops all the hubs. + HubStop() + // HubUnregister unregisters a connection from a hub. + HubUnregister(webConn *WebConn) // InstallMarketplacePlugin installs a plugin listed in the marketplace server. It will get the plugin bundle // from the prepackaged folder, if available, or remotely if EnableRemoteMarketplace is true. InstallMarketplacePlugin(request *model.InstallMarketplacePluginRequest) (*model.Manifest, *model.AppError) @@ -204,6 +214,10 @@ type AppIface interface { MakeAuditRecord(event string, initialStatus string) *audit.Record // MarkChanelAsUnreadFromPost will take a post and set the channel as unread from that one. MarkChannelAsUnreadFromPost(postID string, userID string) (*model.ChannelUnreadAt, *model.AppError) + // NewWebConn returns a new WebConn instance. + NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn + // NewWebHub creates a new Hub. + NewWebHub() *Hub // OverrideIconURLIfEmoji changes the post icon override URL prop, if it has an emoji icon, // so that it points to the URL (relative) of the emoji - static if emoji is default, /api if custom. OverrideIconURLIfEmoji(post *model.Post) @@ -281,6 +295,8 @@ type AppIface interface { UpdateChannel(channel *model.Channel) (*model.Channel, *model.AppError) // UpdateChannelScheme saves the new SchemeId of the channel passed. UpdateChannelScheme(channel *model.Channel) (*model.Channel, *model.AppError) + // UpdateWebConnUserActivity sets the LastUserActivityAt of the hub for the given session. + UpdateWebConnUserActivity(session model.Session, activityAt int64) // UploadFile uploads a single file in form of a completely constructed byte array for a channel. UploadFile(data []byte, channelId string, filename string) (*model.FileInfo, *model.AppError) // UploadFileX uploads a single file as specified in t. It applies the upload @@ -523,7 +539,6 @@ type AppIface interface { GetGroupsBySource(groupSource model.GroupSource) ([]*model.Group, *model.AppError) GetGroupsByTeam(teamId string, opts model.GroupSearchOpts) ([]*model.GroupWithSchemeAdmin, int, *model.AppError) GetGroupsByUserId(userId string) ([]*model.Group, *model.AppError) - GetHubForUserId(userId string) *Hub GetIncomingWebhook(hookId string) (*model.IncomingWebhook, *model.AppError) GetIncomingWebhooksForTeamPage(teamId string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError) GetIncomingWebhooksForTeamPageByUser(teamId string, userId string, page, perPage int) ([]*model.IncomingWebhook, *model.AppError) @@ -674,10 +689,6 @@ type AppIface interface { HasPermissionToChannelByPost(askingUserId string, postId string, permission *model.Permission) bool HasPermissionToTeam(askingUserId string, teamId string, permission *model.Permission) bool HasPermissionToUser(askingUserId string, userId string) bool - HubRegister(webConn *WebConn) - HubStart() - HubStop() - HubUnregister(webConn *WebConn) ImageProxy() *imageproxy.ImageProxy ImageProxyAdder() func(string) string ImageProxyRemover() (f func(string) string) @@ -724,8 +735,6 @@ type AppIface interface { MoveFile(oldPath, newPath string) *model.AppError NewClusterDiscoveryService() *ClusterDiscoveryService NewPluginAPI(manifest *model.Manifest) plugin.API - NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn - NewWebHub() *Hub Notification() einterfaces.NotificationInterface NotificationsLog() *mlog.Logger OpenInteractiveDialog(request model.OpenDialogRequest) *model.AppError @@ -949,7 +958,6 @@ type AppIface interface { UpdateUserAuth(userId string, userAuth *model.UserAuth) (*model.UserAuth, *model.AppError) UpdateUserNotifyProps(userId string, props map[string]string) (*model.User, *model.AppError) UpdateUserRoles(userId string, newRoles string, sendWebSocketEvent bool) (*model.User, *model.AppError) - UpdateWebConnUserActivity(session model.Session, activityAt int64) UploadEmojiImage(id string, imageData *multipart.FileHeader) *model.AppError UploadMultipartFiles(teamId string, channelId string, userId string, fileHeaders []*multipart.FileHeader, clientIds []string, now time.Time) (*model.FileUploadResponse, *model.AppError) UpsertGroupMember(groupID string, userID string) (*model.GroupMember, *model.AppError) diff --git a/app/cluster_handlers.go b/app/cluster_handlers.go index f14b51903c..43dba74a86 100644 --- a/app/cluster_handlers.go +++ b/app/cluster_handlers.go @@ -59,7 +59,7 @@ func (a *App) clusterInvalidateCacheForUserHandler(msg *model.ClusterMessage) { } func (a *App) clusterInvalidateCacheForUserTeamsHandler(msg *model.ClusterMessage) { - a.invalidateCacheForUserTeamsSkipClusterSend(msg.Data) + a.InvalidateWebConnSessionCacheForUser(msg.Data) } func (a *App) clusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) { diff --git a/app/web_conn.go b/app/web_conn.go index 99a93267d2..ba0e63772a 100644 --- a/app/web_conn.go +++ b/app/web_conn.go @@ -9,44 +9,49 @@ import ( "sync/atomic" "time" - "github.com/gorilla/websocket" - goi18n "github.com/mattermost/go-i18n/i18n" "github.com/mattermost/mattermost-server/v5/mlog" "github.com/mattermost/mattermost-server/v5/model" + + "github.com/gorilla/websocket" + goi18n "github.com/mattermost/go-i18n/i18n" ) const ( - SEND_QUEUE_SIZE = 256 - SEND_SLOW_WARN = (SEND_QUEUE_SIZE * 50) / 100 - SEND_DEADLOCK_WARN = (SEND_QUEUE_SIZE * 95) / 100 - WRITE_WAIT = 30 * time.Second - PONG_WAIT = 100 * time.Second - PING_PERIOD = (PONG_WAIT * 6) / 10 - AUTH_TIMEOUT = 5 * time.Second - WEBCONN_MEMBER_CACHE_TIME = 1000 * 60 * 30 // 30 minutes + sendQueueSize = 256 + sendSlowWarn = (sendQueueSize * 50) / 100 + sendFullWarn = (sendQueueSize * 95) / 100 + writeWaitTime = 30 * time.Second + pongWaitTime = 100 * time.Second + pingInterval = (pongWaitTime * 6) / 10 + authCheckInterval = 5 * time.Second + webConnMemberCacheTime = 1000 * 60 * 30 // 30 minutes ) +// WebConn represents a single websocket connection to a user. +// It contains all the necesarry state to manage sending/receiving data to/from +// a websocket. type WebConn struct { - sessionExpiresAt int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically - App *App - WebSocket *websocket.Conn - Send chan model.WebSocketMessage + sessionExpiresAt int64 // This should stay at the top for 64-bit alignment of 64-bit words accessed atomically + App *App + WebSocket *websocket.Conn + T goi18n.TranslateFunc + Locale string + Sequence int64 + UserId string + + allChannelMembers map[string]string + lastAllChannelMembersTime int64 + lastUserActivityAt int64 + send chan model.WebSocketMessage sessionToken atomic.Value session atomic.Value - LastUserActivityAt int64 - UserId string - T goi18n.TranslateFunc - Locale string - AllChannelMembers map[string]string - LastAllChannelMembersTime int64 - Sequence int64 - closeOnce sync.Once endWritePump chan struct{} pumpFinished chan struct{} } +// NewWebConn returns a new WebConn instance. func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.TranslateFunc, locale string) *WebConn { - if len(session.UserId) > 0 { + if session.UserId != "" { a.Srv().Go(func() { a.SetStatusOnline(session.UserId, false) a.UpdateLastActivityAtIfNeeded(session) @@ -55,9 +60,9 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra wc := &WebConn{ App: a, - Send: make(chan model.WebSocketMessage, SEND_QUEUE_SIZE), + send: make(chan model.WebSocketMessage, sendQueueSize), WebSocket: ws, - LastUserActivityAt: model.GetMillis(), + lastUserActivityAt: model.GetMillis(), UserId: session.UserId, T: t, Locale: locale, @@ -72,34 +77,38 @@ func (a *App) NewWebConn(ws *websocket.Conn, session model.Session, t goi18n.Tra return wc } +// Close closes the WebConn. func (wc *WebConn) Close() { wc.WebSocket.Close() - wc.closeOnce.Do(func() { - close(wc.endWritePump) - }) <-wc.pumpFinished } +// GetSessionExpiresAt returns the time at which the session expires. func (wc *WebConn) GetSessionExpiresAt() int64 { return atomic.LoadInt64(&wc.sessionExpiresAt) } +// SetSessionExpiresAt sets the time at which the session expires. func (wc *WebConn) SetSessionExpiresAt(v int64) { atomic.StoreInt64(&wc.sessionExpiresAt, v) } +// GetSessionToken returns the session token of the connection. func (wc *WebConn) GetSessionToken() string { return wc.sessionToken.Load().(string) } +// SetSessionToken sets the session token of the connection. func (wc *WebConn) SetSessionToken(v string) { wc.sessionToken.Store(v) } +// GetSession returns the session of the connection. func (wc *WebConn) GetSession() *model.Session { return wc.session.Load().(*model.Session) } +// SetSession sets the session of the connection. func (wc *WebConn) SetSession(v *model.Session) { if v != nil { v = v.DeepCopy() @@ -108,17 +117,18 @@ func (wc *WebConn) SetSession(v *model.Session) { wc.session.Store(v) } +// Pump starts the WebConn instance. After this, the websocket +// is ready to send/receive messages. func (wc *WebConn) Pump() { - ch := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) go func() { + defer wg.Done() wc.writePump() - close(ch) }() wc.readPump() - wc.closeOnce.Do(func() { - close(wc.endWritePump) - }) - <-ch + close(wc.endWritePump) + wg.Wait() wc.App.HubUnregister(wc) close(wc.pumpFinished) } @@ -128,9 +138,9 @@ func (wc *WebConn) readPump() { wc.WebSocket.Close() }() wc.WebSocket.SetReadLimit(model.SOCKET_MAX_MESSAGE_SIZE_KB) - wc.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT)) + wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime)) wc.WebSocket.SetPongHandler(func(string) error { - wc.WebSocket.SetReadDeadline(time.Now().Add(PONG_WAIT)) + wc.WebSocket.SetReadDeadline(time.Now().Add(pongWaitTime)) if wc.IsAuthenticated() { wc.App.Srv().Go(func() { wc.App.SetStatusAwayIfNeeded(wc.UserId, false) @@ -142,12 +152,7 @@ func (wc *WebConn) readPump() { for { var req model.WebSocketRequest if err := wc.WebSocket.ReadJSON(&req); err != nil { - // browsers will appear as CloseNoStatusReceived - if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { - mlog.Debug("websocket.read: client side closed socket", mlog.String("user_id", wc.UserId)) - } else { - mlog.Debug("websocket.read: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err)) - } + wc.logSocketErr("websocket.read", err) return } wc.App.Srv().WebSocketRouter.ServeWebSocket(wc, &req) @@ -155,8 +160,8 @@ func (wc *WebConn) readPump() { } func (wc *WebConn) writePump() { - ticker := time.NewTicker(PING_PERIOD) - authTicker := time.NewTicker(AUTH_TIMEOUT) + ticker := time.NewTicker(pingInterval) + authTicker := time.NewTicker(authCheckInterval) defer func() { ticker.Stop() @@ -166,9 +171,9 @@ func (wc *WebConn) writePump() { for { select { - case msg, ok := <-wc.Send: + case msg, ok := <-wc.send: if !ok { - wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT)) + wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime)) wc.WebSocket.WriteMessage(websocket.CloseMessage, []byte{}) return } @@ -176,11 +181,12 @@ func (wc *WebConn) writePump() { evt, evtOk := msg.(*model.WebSocketEvent) skipSend := false - if len(wc.Send) >= SEND_SLOW_WARN { + if len(wc.send) >= sendSlowWarn { // When the pump starts to get slow we'll drop non-critical messages - if msg.EventType() == model.WEBSOCKET_EVENT_TYPING || - msg.EventType() == model.WEBSOCKET_EVENT_STATUS_CHANGE || - msg.EventType() == model.WEBSOCKET_EVENT_CHANNEL_VIEWED { + switch msg.EventType() { + case model.WEBSOCKET_EVENT_TYPING, + model.WEBSOCKET_EVENT_STATUS_CHANGE, + model.WEBSOCKET_EVENT_CHANNEL_VIEWED: mlog.Info( "websocket.slow: dropping message", mlog.String("user_id", wc.UserId), @@ -201,33 +207,22 @@ func (wc *WebConn) writePump() { msgBytes = []byte(msg.ToJson()) } - if len(wc.Send) >= SEND_DEADLOCK_WARN { - if evtOk { - mlog.Warn( - "websocket.full", - mlog.String("user_id", wc.UserId), - mlog.String("type", msg.EventType()), - mlog.String("channel_id", evt.GetBroadcast().ChannelId), - mlog.Int("size", len(msg.ToJson())), - ) - } else { - mlog.Warn( - "websocket.full", - mlog.String("user_id", wc.UserId), - mlog.String("type", msg.EventType()), - mlog.Int("size", len(msg.ToJson())), - ) + if len(wc.send) >= sendFullWarn { + logData := []mlog.Field{ + mlog.String("user_id", wc.UserId), + mlog.String("type", msg.EventType()), + mlog.Int("size", len(msgBytes)), } + if evtOk { + logData = append(logData, mlog.String("channel_id", evt.GetBroadcast().ChannelId)) + } + + mlog.Warn("websocket.full", logData...) } - wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT)) + wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime)) if err := wc.WebSocket.WriteMessage(websocket.TextMessage, msgBytes); err != nil { - // browsers will appear as CloseNoStatusReceived - if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { - mlog.Debug("websocket.send: client side closed socket", mlog.String("user_id", wc.UserId)) - } else { - mlog.Debug("websocket.send: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err)) - } + wc.logSocketErr("websocket.send", err) return } @@ -237,14 +232,9 @@ func (wc *WebConn) writePump() { } case <-ticker.C: - wc.WebSocket.SetWriteDeadline(time.Now().Add(WRITE_WAIT)) + wc.WebSocket.SetWriteDeadline(time.Now().Add(writeWaitTime)) if err := wc.WebSocket.WriteMessage(websocket.PingMessage, []byte{}); err != nil { - // browsers will appear as CloseNoStatusReceived - if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { - mlog.Debug("websocket.ticker: client side closed socket", mlog.String("user_id", wc.UserId)) - } else { - mlog.Debug("websocket.ticker: closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err)) - } + wc.logSocketErr("websocket.ticker", err) return } @@ -261,13 +251,15 @@ func (wc *WebConn) writePump() { } } +// InvalidateCache resets all internal data of the WebConn. func (wc *WebConn) InvalidateCache() { - wc.AllChannelMembers = nil - wc.LastAllChannelMembersTime = 0 + wc.allChannelMembers = nil + wc.lastAllChannelMembersTime = 0 wc.SetSession(nil) wc.SetSessionExpiresAt(0) } +// IsAuthenticated returns whether the given WebConn is authenticated or not. func (wc *WebConn) IsAuthenticated() bool { // Check the expiry to see if we need to check for a new session if wc.GetSessionExpiresAt() < model.GetMillis() { @@ -324,7 +316,8 @@ func (wc *WebConn) shouldSendEventToGuest(msg *model.WebSocketEvent) bool { return canSee } -func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool { +// shouldSendEvent returns whether the message should be sent or not. +func (wc *WebConn) shouldSendEvent(msg *model.WebSocketEvent) bool { // IMPORTANT: Do not send event if WebConn does not have a session if !wc.IsAuthenticated() { return false @@ -353,7 +346,7 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool { } // If the event is destined to a specific user - if len(msg.GetBroadcast().UserId) > 0 { + if msg.GetBroadcast().UserId != "" { return wc.UserId == msg.GetBroadcast().UserId } @@ -365,31 +358,31 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool { } // Only report events to users who are in the channel for the event - if len(msg.GetBroadcast().ChannelId) > 0 { - if model.GetMillis()-wc.LastAllChannelMembersTime > WEBCONN_MEMBER_CACHE_TIME { - wc.AllChannelMembers = nil - wc.LastAllChannelMembersTime = 0 + if msg.GetBroadcast().ChannelId != "" { + if model.GetMillis()-wc.lastAllChannelMembersTime > webConnMemberCacheTime { + wc.allChannelMembers = nil + wc.lastAllChannelMembersTime = 0 } - if wc.AllChannelMembers == nil { + if wc.allChannelMembers == nil { result, err := wc.App.Srv().Store.Channel().GetAllChannelMembersForUser(wc.UserId, true, false) if err != nil { mlog.Error("webhub.shouldSendEvent.", mlog.Err(err)) return false } - wc.AllChannelMembers = result - wc.LastAllChannelMembersTime = model.GetMillis() + wc.allChannelMembers = result + wc.lastAllChannelMembersTime = model.GetMillis() } - if _, ok := wc.AllChannelMembers[msg.GetBroadcast().ChannelId]; ok { + if _, ok := wc.allChannelMembers[msg.GetBroadcast().ChannelId]; ok { return true } return false } // Only report events to users who are in the team for the event - if len(msg.GetBroadcast().TeamId) > 0 { - return wc.IsMemberOfTeam(msg.GetBroadcast().TeamId) + if msg.GetBroadcast().TeamId != "" { + return wc.isMemberOfTeam(msg.GetBroadcast().TeamId) } if wc.GetSession().Props[model.SESSION_PROP_IS_GUEST] == "true" { @@ -399,10 +392,12 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool { return true } -func (wc *WebConn) IsMemberOfTeam(teamId string) bool { +// IsMemberOfTeam returns whether the user of the WebConn +// is a member of the given teamId or not. +func (wc *WebConn) isMemberOfTeam(teamId string) bool { currentSession := wc.GetSession() - if currentSession == nil || len(currentSession.Token) == 0 { + if currentSession == nil || currentSession.Token == "" { session, err := wc.App.GetSession(wc.GetSessionToken()) if err != nil { mlog.Error("Invalid session.", mlog.Err(err)) @@ -414,3 +409,12 @@ func (wc *WebConn) IsMemberOfTeam(teamId string) bool { return currentSession.GetTeamByTeamId(teamId) != nil } + +func (wc *WebConn) logSocketErr(source string, err error) { + // browsers will appear as CloseNoStatusReceived + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { + mlog.Debug(source+": client side closed socket", mlog.String("user_id", wc.UserId)) + } else { + mlog.Debug(source+": closing websocket", mlog.String("user_id", wc.UserId), mlog.Err(err)) + } +} diff --git a/app/web_conn_test.go b/app/web_conn_test.go index d9a146a8d5..5672bc6a96 100644 --- a/app/web_conn_test.go +++ b/app/web_conn_test.go @@ -86,15 +86,15 @@ func TestWebConnShouldSendEvent(t *testing.T) { event := model.NewWebSocketEvent("some_event", "", "", "", nil) for _, c := range cases { event = event.SetBroadcast(c.Broadcast) - assert.Equal(t, c.User1Expected, basicUserWc.ShouldSendEvent(event), c.Description) - assert.Equal(t, c.User2Expected, basicUser2Wc.ShouldSendEvent(event), c.Description) - assert.Equal(t, c.AdminExpected, adminUserWc.ShouldSendEvent(event), c.Description) + assert.Equal(t, c.User1Expected, basicUserWc.shouldSendEvent(event), c.Description) + assert.Equal(t, c.User2Expected, basicUser2Wc.shouldSendEvent(event), c.Description) + assert.Equal(t, c.AdminExpected, adminUserWc.shouldSendEvent(event), c.Description) } event2 := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_UPDATE_TEAM, th.BasicTeam.Id, "", "", nil) - assert.True(t, basicUserWc.ShouldSendEvent(event2)) - assert.True(t, basicUser2Wc.ShouldSendEvent(event2)) + assert.True(t, basicUserWc.shouldSendEvent(event2)) + assert.True(t, basicUser2Wc.shouldSendEvent(event2)) event3 := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_UPDATE_TEAM, "wrongId", "", "", nil) - assert.False(t, basicUserWc.ShouldSendEvent(event3)) + assert.False(t, basicUserWc.shouldSendEvent(event3)) } diff --git a/app/web_hub.go b/app/web_hub.go index f7623b2ebe..935ab2da5d 100644 --- a/app/web_hub.go +++ b/app/web_hub.go @@ -15,15 +15,23 @@ import ( ) const ( - BROADCAST_QUEUE_SIZE = 4096 + broadcastQueueSize = 4096 ) -type WebConnActivityMessage struct { - UserId string - SessionToken string - ActivityAt int64 +type webConnActivityMessage struct { + userId string + sessionToken string + activityAt int64 } +type webConnDirectMessage struct { + conn *WebConn + msg model.WebSocketMessage +} + +// Hub is the central place to manage all websocket connections in the server. +// It handles different websocket events and sending messages to individual +// user connections. type Hub struct { // connectionCount should be kept first. // See https://github.com/mattermost/mattermost-server/pull/7281 @@ -36,21 +44,23 @@ type Hub struct { stop chan struct{} didStop chan struct{} invalidateUser chan string - activity chan *WebConnActivityMessage - ExplicitStop bool + activity chan *webConnActivityMessage + directMsg chan *webConnDirectMessage + explicitStop bool } +// NewWebHub creates a new Hub. func (a *App) NewWebHub() *Hub { return &Hub{ app: a, register: make(chan *WebConn, 1), unregister: make(chan *WebConn, 1), - broadcast: make(chan *model.WebSocketEvent, BROADCAST_QUEUE_SIZE), + broadcast: make(chan *model.WebSocketEvent, broadcastQueueSize), stop: make(chan struct{}), didStop: make(chan struct{}), invalidateUser: make(chan string), - activity: make(chan *WebConnActivityMessage), - ExplicitStop: false, + activity: make(chan *webConnActivityMessage), + directMsg: make(chan *webConnDirectMessage), } } @@ -58,6 +68,7 @@ func (a *App) TotalWebsocketConnections() int { return a.Srv().TotalWebsocketConnections() } +// HubStart starts all the hubs. func (a *App) HubStart() { // Total number of hubs is twice the number of CPUs. numberOfHubs := runtime.NumCPU() * 2 @@ -77,6 +88,36 @@ func (a *App) HubStart() { } } +func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) { + if message.GetBroadcast().UserId != "" { + hub := a.GetHubForUserId(message.GetBroadcast().UserId) + if hub != nil { + hub.Broadcast(message) + } + return + } + for _, hub := range a.Srv().GetHubs() { + hub.Broadcast(message) + } +} + +func (a *App) invalidateCacheForUserSkipClusterSend(userId string) { + a.Srv().Store.Channel().InvalidateAllChannelMembersForUser(userId) + a.InvalidateWebConnSessionCacheForUser(userId) +} + +func (a *App) invalidateCacheForWebhook(webhookId string) { + a.Srv().Store.Webhook().InvalidateWebhookCache(webhookId) +} + +func (a *App) InvalidateWebConnSessionCacheForUser(userId string) { + hub := a.GetHubForUserId(userId) + if hub != nil { + hub.InvalidateUser(userId) + } +} + +// HubStop stops all the hubs. func (a *App) HubStop() { mlog.Info("stopping websocket hub connections") @@ -87,6 +128,7 @@ func (a *App) HubStop() { a.Srv().SetHubs([]*Hub{}) } +// GetHubForUserId returns the hub for a given user id. func (a *App) GetHubForUserId(userId string) *Hub { if len(a.Srv().GetHubs()) == 0 { return nil @@ -103,6 +145,7 @@ func (a *App) GetHubForUserId(userId string) *Hub { return hub } +// HubRegister registers a connection to a hub. func (a *App) HubRegister(webConn *WebConn) { hub := a.GetHubForUserId(webConn.UserId) if hub != nil { @@ -113,6 +156,7 @@ func (a *App) HubRegister(webConn *WebConn) { } } +// HubUnregister unregisters a connection from a hub. func (a *App) HubUnregister(webConn *WebConn) { hub := a.GetHubForUserId(webConn.UserId) if hub != nil { @@ -149,19 +193,6 @@ func (a *App) Publish(message *model.WebSocketEvent) { } } -func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) { - if message.GetBroadcast().UserId != "" { - hub := a.GetHubForUserId(message.GetBroadcast().UserId) - if hub != nil { - hub.Broadcast(message) - } - } else { - for _, hub := range a.Srv().GetHubs() { - hub.Broadcast(message) - } - } -} - func (a *App) invalidateCacheForChannel(channel *model.Channel) { a.Srv().Store.Channel().InvalidateChannel(channel.Id) a.invalidateCacheForChannelByNameSkipClusterSend(channel.TeamId, channel.Name) @@ -237,7 +268,7 @@ func (a *App) InvalidateCacheForUser(userId string) { } func (a *App) invalidateCacheForUserTeams(userId string) { - a.invalidateCacheForUserTeamsSkipClusterSend(userId) + a.InvalidateWebConnSessionCacheForUser(userId) a.Srv().Store.Team().InvalidateAllTeamIdsForUser(userId) if a.Cluster() != nil { @@ -250,33 +281,7 @@ func (a *App) invalidateCacheForUserTeams(userId string) { } } -func (a *App) invalidateCacheForUserSkipClusterSend(userId string) { - a.Srv().Store.Channel().InvalidateAllChannelMembersForUser(userId) - - hub := a.GetHubForUserId(userId) - if hub != nil { - hub.InvalidateUser(userId) - } -} - -func (a *App) invalidateCacheForUserTeamsSkipClusterSend(userId string) { - hub := a.GetHubForUserId(userId) - if hub != nil { - hub.InvalidateUser(userId) - } -} - -func (a *App) invalidateCacheForWebhook(webhookId string) { - a.Srv().Store.Webhook().InvalidateWebhookCache(webhookId) -} - -func (a *App) InvalidateWebConnSessionCacheForUser(userId string) { - hub := a.GetHubForUserId(userId) - if hub != nil { - hub.InvalidateUser(userId) - } -} - +// UpdateWebConnUserActivity sets the LastUserActivityAt of the hub for the given session. func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64) { hub := a.GetHubForUserId(session.UserId) if hub != nil { @@ -284,13 +289,15 @@ func (a *App) UpdateWebConnUserActivity(session model.Session, activityAt int64) } } +// Register registers a connection to the hub. func (h *Hub) Register(webConn *WebConn) { select { case h.register <- webConn: - case <-h.didStop: + case <-h.stop: } } +// Unregister unregisters a connection from the hub. func (h *Hub) Unregister(webConn *WebConn) { select { case h.unregister <- webConn: @@ -298,37 +305,64 @@ func (h *Hub) Unregister(webConn *WebConn) { } } +// Broadcast broadcasts the message to all connections in the hub. func (h *Hub) Broadcast(message *model.WebSocketEvent) { - if h != nil && h.broadcast != nil && message != nil { + // XXX: The hub nil check is because of the way we setup our tests. We call `app.NewServer()` + // which returns a server, but only after that, we call `wsapi.Init()` through our FakeApp adapter + // to initialize the hub. But in the `NewServer` call itself, we call `RunOldAppInitialization` + // which directly proceeds to broadcast some messages happily. + // This needs to be fixed once the FakeApp adapter goes away. And possibly, we can look into + // doing hub initialization inside NewServer itself. + if h != nil && message != nil { if metrics := h.app.Metrics(); metrics != nil { metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) } select { case h.broadcast <- message: - case <-h.didStop: + case <-h.stop: } } } +// InvalidateUser invalidates the cache for the given user. func (h *Hub) InvalidateUser(userId string) { select { case h.invalidateUser <- userId: - case <-h.didStop: + case <-h.stop: } } +// UpdateActivity sets the LastUserActivityAt field for the connection +// of the user. func (h *Hub) UpdateActivity(userId, sessionToken string, activityAt int64) { select { - case h.activity <- &WebConnActivityMessage{UserId: userId, SessionToken: sessionToken, ActivityAt: activityAt}: - case <-h.didStop: + case h.activity <- &webConnActivityMessage{ + userId: userId, + sessionToken: sessionToken, + activityAt: activityAt, + }: + case <-h.stop: } } +// SendMessage sends the given message to the given connection. +func (h *Hub) SendMessage(conn *WebConn, msg model.WebSocketMessage) { + select { + case h.directMsg <- &webConnDirectMessage{ + conn: conn, + msg: msg, + }: + case <-h.stop: + } +} + +// Stop stops the hub. func (h *Hub) Stop() { close(h.stop) <-h.didStop } +// Start starts the hub. func (h *Hub) Start() { var doStart func() var doRecoverableStart func() @@ -337,85 +371,94 @@ func (h *Hub) Start() { doStart = func() { mlog.Debug("Hub is starting", mlog.Int("index", h.connectionIndex)) - connections := newHubConnectionIndex() + connIndex := newHubConnectionIndex() for { select { - case webCon := <-h.register: - connections.Add(webCon) - atomic.StoreInt64(&h.connectionCount, int64(len(connections.All()))) - if webCon.IsAuthenticated() { - webCon.Send <- webCon.createHelloMessage() + case webConn := <-h.register: + connIndex.Add(webConn) + atomic.StoreInt64(&h.connectionCount, int64(len(connIndex.All()))) + if webConn.IsAuthenticated() { + webConn.send <- webConn.createHelloMessage() } - case webCon := <-h.unregister: - connections.Remove(webCon) - atomic.StoreInt64(&h.connectionCount, int64(len(connections.All()))) + case webConn := <-h.unregister: + connIndex.Remove(webConn) + atomic.StoreInt64(&h.connectionCount, int64(len(connIndex.All()))) - if len(webCon.UserId) == 0 { + if len(webConn.UserId) == 0 { continue } - conns := connections.ForUser(webCon.UserId) + conns := connIndex.ForUser(webConn.UserId) if len(conns) == 0 { h.app.Srv().Go(func() { - h.app.SetStatusOffline(webCon.UserId, false) + h.app.SetStatusOffline(webConn.UserId, false) }) - } else { - var latestActivity int64 = 0 - for _, conn := range conns { - if conn.LastUserActivityAt > latestActivity { - latestActivity = conn.LastUserActivityAt - } - } - if h.app.IsUserAway(latestActivity) { - h.app.Srv().Go(func() { - h.app.SetStatusLastActivityAt(webCon.UserId, latestActivity) - }) + continue + } + var latestActivity int64 = 0 + for _, conn := range conns { + if conn.lastUserActivityAt > latestActivity { + latestActivity = conn.lastUserActivityAt } } + + if h.app.IsUserAway(latestActivity) { + h.app.Srv().Go(func() { + h.app.SetStatusLastActivityAt(webConn.UserId, latestActivity) + }) + } case userId := <-h.invalidateUser: - for _, webCon := range connections.ForUser(userId) { - webCon.InvalidateCache() + for _, webConn := range connIndex.ForUser(userId) { + webConn.InvalidateCache() } case activity := <-h.activity: - for _, webCon := range connections.ForUser(activity.UserId) { - if webCon.GetSessionToken() == activity.SessionToken { - webCon.LastUserActivityAt = activity.ActivityAt + for _, webConn := range connIndex.ForUser(activity.userId) { + if webConn.GetSessionToken() == activity.sessionToken { + webConn.lastUserActivityAt = activity.activityAt } } + case directMsg := <-h.directMsg: + if !connIndex.Has(directMsg.conn) { + continue + } + select { + case directMsg.conn.send <- directMsg.msg: + default: + mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", directMsg.conn.UserId)) + close(directMsg.conn.send) + connIndex.Remove(directMsg.conn) + } case msg := <-h.broadcast: if metrics := h.app.Metrics(); metrics != nil { metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) } - candidates := connections.All() + candidates := connIndex.All() if msg.GetBroadcast().UserId != "" { - candidates = connections.ForUser(msg.GetBroadcast().UserId) + candidates = connIndex.ForUser(msg.GetBroadcast().UserId) } msg = msg.PrecomputeJSON() - for _, webCon := range candidates { - if webCon.ShouldSendEvent(msg) { + for _, webConn := range candidates { + if !connIndex.Has(webConn) { + continue + } + if webConn.shouldSendEvent(msg) { select { - case webCon.Send <- msg: + case webConn.send <- msg: default: - mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", webCon.UserId)) - close(webCon.Send) - connections.Remove(webCon) + mlog.Error("webhub.broadcast: cannot send, closing websocket for user", mlog.String("user_id", webConn.UserId)) + close(webConn.send) + connIndex.Remove(webConn) } } } case <-h.stop: - userIds := make(map[string]bool) - - for _, webCon := range connections.All() { - userIds[webCon.UserId] = true - webCon.Close() + for _, webConn := range connIndex.All() { + webConn.Close() + h.app.SetStatusOffline(webConn.UserId, false) } - for userId := range userIds { - h.app.SetStatusOffline(userId, false) - } - - h.ExplicitStop = true + h.explicitStop = true close(h.didStop) return @@ -429,7 +472,7 @@ func (h *Hub) Start() { } doRecover = func() { - if !h.ExplicitStop { + if !h.explicitStop { if r := recover(); r != nil { mlog.Error("Recovering from Hub panic.", mlog.Any("panic", r)) } else { @@ -494,6 +537,11 @@ func (i *hubConnectionIndex) Remove(wc *WebConn) { delete(i.connectionIndexes, wc) } +func (i *hubConnectionIndex) Has(wc *WebConn) bool { + _, ok := i.connectionIndexes[wc] + return ok +} + func (i *hubConnectionIndex) ForUser(id string) []*WebConn { return i.connectionsByUserId[id] } diff --git a/app/web_hub_test.go b/app/web_hub_test.go index 9f36c3442c..d4a27f8d9d 100644 --- a/app/web_hub_test.go +++ b/app/web_hub_test.go @@ -90,7 +90,7 @@ func TestHubStopRaceCondition(t *testing.T) { hub.UpdateActivity("userId", "sessionToken", 0) - for i := 0; i <= BROADCAST_QUEUE_SIZE; i++ { + for i := 0; i <= broadcastQueueSize; i++ { hub.Broadcast(model.NewWebSocketEvent("", "", "", "", nil)) } diff --git a/app/websocket_router.go b/app/websocket_router.go index 236fca56de..80cc0e7980 100644 --- a/app/websocket_router.go +++ b/app/websocket_router.go @@ -27,13 +27,13 @@ func (wr *WebSocketRouter) Handle(action string, handler webSocketHandler) { func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketRequest) { if r.Action == "" { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.no_action.app_error", nil, "", http.StatusBadRequest) - ReturnWebSocketError(conn, r, err) + returnWebSocketError(wr.app, conn, r, err) return } if r.Seq <= 0 { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_seq.app_error", nil, "", http.StatusBadRequest) - ReturnWebSocketError(conn, r, err) + returnWebSocketError(wr.app, conn, r, err) return } @@ -66,28 +66,32 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque wr.app.HubRegister(conn) resp := model.NewWebSocketResponse(model.STATUS_OK, r.Seq, nil) - conn.Send <- resp + hub := wr.app.GetHubForUserId(conn.UserId) + if hub == nil { + return + } + hub.SendMessage(conn, resp) return } if !conn.IsAuthenticated() { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.not_authenticated.app_error", nil, "", http.StatusUnauthorized) - ReturnWebSocketError(conn, r, err) + returnWebSocketError(wr.app, conn, r, err) return } handler, ok := wr.handlers[r.Action] if !ok { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_action.app_error", nil, "", http.StatusInternalServerError) - ReturnWebSocketError(conn, r, err) + returnWebSocketError(wr.app, conn, r, err) return } handler.ServeWebSocket(conn, r) } -func ReturnWebSocketError(conn *WebConn, r *model.WebSocketRequest, err *model.AppError) { +func returnWebSocketError(app *App, conn *WebConn, r *model.WebSocketRequest, err *model.AppError) { mlog.Error( "websocket routing error.", mlog.Int64("seq", r.Seq), @@ -96,8 +100,12 @@ func ReturnWebSocketError(conn *WebConn, r *model.WebSocketRequest, err *model.A mlog.Err(err), ) + hub := app.GetHubForUserId(conn.UserId) + if hub == nil { + return + } + err.DetailedError = "" errorResp := model.NewWebSocketError(r.Seq, err) - - conn.Send <- errorResp + hub.SendMessage(conn, errorResp) } diff --git a/wsapi/websocket_handler.go b/wsapi/websocket_handler.go index b0da94844a..46ed07f67b 100644 --- a/wsapi/websocket_handler.go +++ b/wsapi/websocket_handler.go @@ -24,6 +24,10 @@ type webSocketHandler struct { func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketRequest) { mlog.Debug("Websocket request", mlog.String("action", r.Action)) + hub := wh.app.GetHubForUserId(conn.UserId) + if hub == nil { + return + } session, sessionErr := wh.app.GetSession(conn.GetSessionToken()) if sessionErr != nil { mlog.Error( @@ -36,8 +40,7 @@ func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketR ) sessionErr.DetailedError = "" errResp := model.NewWebSocketError(r.Seq, sessionErr) - - conn.Send <- errResp + hub.SendMessage(conn, errResp) return } @@ -59,14 +62,12 @@ func (wh webSocketHandler) ServeWebSocket(conn *app.WebConn, r *model.WebSocketR ) err.DetailedError = "" errResp := model.NewWebSocketError(r.Seq, err) - - conn.Send <- errResp + hub.SendMessage(conn, errResp) return } resp := model.NewWebSocketResponse(model.STATUS_OK, r.Seq, data) - - conn.Send <- resp + hub.SendMessage(conn, resp) } func NewInvalidWebSocketParamError(action string, name string) *model.AppError {