MM-25700 : Use a counting semaphore for push notifications hub (#14758)

Automatic Merge

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 {

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)

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 {

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")

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
 

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