diff --git a/app/app.go b/app/app.go index ef1471f5e8..45da3ce875 100644 --- a/app/app.go +++ b/app/app.go @@ -70,7 +70,6 @@ func (a *App) InitServer() { a.notification = a.srv.Notification a.saml = a.srv.Saml - a.StartPushNotificationsHubWorkers() a.AddConfigListener(func(oldConfig *model.Config, newConfig *model.Config) { if *oldConfig.GuestAccountsSettings.Enable && !*newConfig.GuestAccountsSettings.Enable { if appErr := a.DeactivateGuests(); appErr != nil { diff --git a/app/app_iface.go b/app/app_iface.go index 56bfd19c23..34283821ee 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -902,7 +902,6 @@ type AppIface interface { SlackUploadFile(slackPostFile *SlackFile, uploads map[string]*zip.File, teamId string, channelId string, userId string, slackTimestamp string) (*model.FileInfo, bool) SoftDeleteTeam(teamId string) *model.AppError Srv() *Server - StartPushNotificationsHubWorkers() SubmitInteractiveDialog(request model.SubmitDialogRequest) (*model.SubmitDialogResponse, *model.AppError) SwitchEmailToLdap(email, password, code, ldapLoginId, ldapPassword string) (string, *model.AppError) SwitchEmailToOAuth(w http.ResponseWriter, r *http.Request, email, password, code, service string) (string, *model.AppError) diff --git a/app/notification_push.go b/app/notification_push.go index fc4c2b607e..71ba57593f 100644 --- a/app/notification_push.go +++ b/app/notification_push.go @@ -4,9 +4,10 @@ package app import ( - "hash/fnv" "net/http" + "runtime" "strings" + "sync" "github.com/pkg/errors" @@ -24,11 +25,11 @@ const ( notificationTypeUpdateBadge notificationType = "update_badge" ) -const PUSH_NOTIFICATION_HUB_WORKERS = 1000 -const PUSH_NOTIFICATIONS_HUB_BUFFER_PER_WORKER = 50 - type PushNotificationsHub struct { - Channels []chan PushNotification + notificationsChan chan PushNotification + app *App // XXX: This will go away once push notifications move to their own package. + sema chan struct{} + wg *sync.WaitGroup } type PushNotification struct { @@ -46,13 +47,6 @@ type PushNotification struct { replyToThreadType string } -func (hub *PushNotificationsHub) GetGoChannelFromUserId(userId string) chan PushNotification { - h := fnv.New32a() - h.Write([]byte(userId)) - chanIdx := h.Sum32() % PUSH_NOTIFICATION_HUB_WORKERS - return hub.Channels[chanIdx] -} - func (a *App) sendPushNotificationSync(post *model.Post, user *model.User, channel *model.Channel, channelName string, senderName string, explicitMention bool, channelWideMention bool, replyToThreadType string) *model.AppError { cfg := a.Config() @@ -143,8 +137,7 @@ func (a *App) sendPushNotification(notification *PostNotification, user *model.U channelName := notification.GetChannelName(nameFormat, user.Id) senderName := notification.GetSenderName(nameFormat, *cfg.ServiceSettings.EnablePostUsernameOverride) - c := a.Srv().PushNotificationsHub.GetGoChannelFromUserId(user.Id) - c <- PushNotification{ + a.Srv().PushNotificationsHub.notificationsChan <- PushNotification{ notificationType: notificationTypeMessage, post: post, user: user, @@ -217,8 +210,7 @@ func (a *App) clearPushNotificationSync(currentSessionId, userId, channelId stri } func (a *App) clearPushNotification(currentSessionId, userId, channelId string) { - channel := a.Srv().PushNotificationsHub.GetGoChannelFromUserId(userId) - channel <- PushNotification{ + a.Srv().PushNotificationsHub.notificationsChan <- PushNotification{ notificationType: notificationTypeClear, currentSessionId: currentSessionId, userId: userId, @@ -245,63 +237,78 @@ func (a *App) updateMobileAppBadgeSync(userId string) *model.AppError { } func (a *App) UpdateMobileAppBadge(userId string) { - channel := a.Srv().PushNotificationsHub.GetGoChannelFromUserId(userId) - channel <- PushNotification{ + a.Srv().PushNotificationsHub.notificationsChan <- PushNotification{ notificationType: notificationTypeUpdateBadge, userId: userId, } } func (s *Server) createPushNotificationsHub() { + buffer := *s.Config().EmailSettings.PushNotificationBuffer + // XXX: This can be _almost_ removed except that there is a dependency with + // a.ClearSessionCacheForUser(session.UserId) which invalidates caches, + // which then takes to web_hub code. It's a bit complicated, so leaving as is for now. + fakeApp := New(ServerConnector(s)) hub := PushNotificationsHub{ - Channels: []chan PushNotification{}, - } - for x := 0; x < PUSH_NOTIFICATION_HUB_WORKERS; x++ { - hub.Channels = append(hub.Channels, make(chan PushNotification, PUSH_NOTIFICATIONS_HUB_BUFFER_PER_WORKER)) + notificationsChan: make(chan PushNotification, buffer), + app: fakeApp, + wg: new(sync.WaitGroup), } + go hub.start() s.PushNotificationsHub = hub } -func (a *App) pushNotificationWorker(notifications chan PushNotification) { - for notification := range notifications { - var err *model.AppError - switch notification.notificationType { - case notificationTypeClear: - err = a.clearPushNotificationSync(notification.currentSessionId, notification.userId, notification.channelId) - case notificationTypeMessage: - err = a.sendPushNotificationSync( - notification.post, - notification.user, - notification.channel, - notification.channelName, - notification.senderName, - notification.explicitMention, - notification.channelWideMention, - notification.replyToThreadType, - ) - case notificationTypeUpdateBadge: - err = a.updateMobileAppBadgeSync(notification.userId) - default: - mlog.Error("Invalid notification type", mlog.String("notification_type", string(notification.notificationType))) - } +func (hub *PushNotificationsHub) start() { + hub.sema = make(chan struct{}, runtime.NumCPU()*8) // numCPU * 8 is a good amount of concurrency. - if err != nil { - mlog.Error("Unable to send push notification", mlog.String("notification_type", string(notification.notificationType)), mlog.Err(err)) - } + for notification := range hub.notificationsChan { + // Adding to the waitgroup first. + hub.wg.Add(1) + // Get token. + hub.sema <- struct{}{} + go func(notification PushNotification) { + defer func() { + // Release token. + <-hub.sema + // Now marking waitgroup as done. + hub.wg.Done() + }() + + var err *model.AppError + switch notification.notificationType { + case notificationTypeClear: + err = hub.app.clearPushNotificationSync(notification.currentSessionId, notification.userId, notification.channelId) + case notificationTypeMessage: + err = hub.app.sendPushNotificationSync( + notification.post, + notification.user, + notification.channel, + notification.channelName, + notification.senderName, + notification.explicitMention, + notification.channelWideMention, + notification.replyToThreadType, + ) + case notificationTypeUpdateBadge: + err = hub.app.updateMobileAppBadgeSync(notification.userId) + default: + mlog.Error("Invalid notification type", mlog.String("notification_type", string(notification.notificationType))) + } + + if err != nil { + mlog.Error("Unable to send push notification", mlog.String("notification_type", string(notification.notificationType)), mlog.Err(err)) + } + }(notification) } } -func (a *App) StartPushNotificationsHubWorkers() { - for x := 0; x < PUSH_NOTIFICATION_HUB_WORKERS; x++ { - channel := a.Srv().PushNotificationsHub.Channels[x] - a.Srv().Go(func() { a.pushNotificationWorker(channel) }) - } +func (hub *PushNotificationsHub) stop() { + close(hub.notificationsChan) + hub.wg.Wait() } func (s *Server) StopPushNotificationsHubWorkers() { - for _, channel := range s.PushNotificationsHub.Channels { - close(channel) - } + s.PushNotificationsHub.stop() } func (a *App) sendToPushProxy(msg *model.PushNotification, session *model.Session) error { diff --git a/app/opentracing_layer.go b/app/opentracing_layer.go index 89b2f67327..0e0c282105 100644 --- a/app/opentracing_layer.go +++ b/app/opentracing_layer.go @@ -13315,21 +13315,6 @@ func (a *OpenTracingAppLayer) SoftDeleteTeam(teamId string) *model.AppError { return resultVar0 } -func (a *OpenTracingAppLayer) StartPushNotificationsHubWorkers() { - origCtx := a.ctx - span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.StartPushNotificationsHubWorkers") - - 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.StartPushNotificationsHubWorkers() -} - func (a *OpenTracingAppLayer) SubmitInteractiveDialog(request model.SubmitDialogRequest) (*model.SubmitDialogResponse, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.SubmitInteractiveDialog") diff --git a/app/server.go b/app/server.go index cb6adec74a..664357f6aa 100644 --- a/app/server.go +++ b/app/server.go @@ -95,7 +95,8 @@ type Server struct { hubsLock sync.RWMutex hubs []*Hub - PushNotificationsHub PushNotificationsHub + PushNotificationsHub PushNotificationsHub + pushNotificationClient *http.Client // TODO: move this to it's own package runjobs bool Jobs *jobs.JobServer @@ -137,8 +138,7 @@ type Server struct { phase2PermissionsMigrationComplete bool - HTTPService httpservice.HTTPService - pushNotificationClient *http.Client // TODO: move this to it's own package + HTTPService httpservice.HTTPService ImageProxy *imageproxy.ImageProxy diff --git a/model/config.go b/model/config.go index 9cc8cdf065..f57cb8c0a7 100644 --- a/model/config.go +++ b/model/config.go @@ -1377,6 +1377,7 @@ type EmailSettings struct { SendPushNotifications *bool PushNotificationServer *string PushNotificationContents *string + PushNotificationBuffer *int EnableEmailBatching *bool EmailBatchingBufferSize *int EmailBatchingInterval *int @@ -1477,6 +1478,10 @@ func (s *EmailSettings) SetDefaults(isUpdate bool) { s.PushNotificationContents = NewString(FULL_NOTIFICATION) } + if s.PushNotificationBuffer == nil { + s.PushNotificationBuffer = NewInt(1000) + } + if s.EnableEmailBatching == nil { s.EnableEmailBatching = NewBool(false) }