Introduce dedicated event publisher per document

server/backend/sync/memory/publisher.go

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2024 The Yorkie Authors. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package memory
+
+import (
+	"context"
+	gosync "sync"
+	time "time"
+
+	"go.uber.org/zap"
+
+	"github.com/yorkie-team/yorkie/server/backend/sync"
+	"github.com/yorkie-team/yorkie/server/logging"
+)
+
+// BatchPublisher is a publisher that publishes events in batch.
+type BatchPublisher struct {
+	events    []sync.DocEvent
+	mutex     gosync.Mutex
+	window    time.Duration
+	maxBatch  int
+	closeChan chan struct{}
+	subs      *Subscriptions
+}
+
+// NewBatchPublisher creates a new BatchPublisher instance.
+func NewBatchPublisher(window time.Duration, maxBatch int, subs *Subscriptions) *BatchPublisher {
+	bp := &BatchPublisher{
+		window:    window,
+		maxBatch:  maxBatch,
+		closeChan: make(chan struct{}),
+		subs:      subs,
+	}
+
+	go bp.processLoop()
+	return bp
+}
+
+// Publish adds the given event to the batch. If the batch is full, it publishes
+// the batch.
+func (bp *BatchPublisher) Publish(ctx context.Context, event sync.DocEvent) {
+	bp.mutex.Lock()
+	defer bp.mutex.Unlock()
+
+	// TODO(hackerwins): If the event is DocumentChangedEvent, we should merge
+	// the events to reduce the number of events to be published.
+	bp.events = append(bp.events, event)
+
+	// TODO(hackerwins): Consider to use processLoop to publish events.
+	if len(bp.events) >= bp.maxBatch {
+		bp.publish(ctx)
+	}
+}
+
+func (bp *BatchPublisher) processLoop() {
+	ticker := time.NewTicker(bp.window)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			bp.mutex.Lock()
+			bp.publish(context.Background())
+			bp.mutex.Unlock()
+		case <-bp.closeChan:
+			return
+		}
+	}
+}
+
+func (bp *BatchPublisher) publish(ctx context.Context) {
+	if len(bp.events) == 0 {
+		return
+	}
+
+	if logging.Enabled(zap.DebugLevel) {
+		logging.From(ctx).Infof(
+			"Publishing batch of %d events for document %s",
+			len(bp.events),
+			bp.subs.docKey,
+		)
+	}
+
+	for _, sub := range bp.subs.Values() {
+		for _, event := range bp.events {
+			if sub.Subscriber().Compare(event.Publisher) == 0 {
+				continue
+			}
+
+			if ok := sub.Publish(event); !ok {
+				logging.From(ctx).Infof(
+					"Publish(%s,%s) to %s timeout or closed",
+					event.Type,
+					event.Publisher,
+					sub.Subscriber(),
+				)
+			}
+		}
+	}
+
+	bp.events = nil
+}
+
+// Close stops the batch publisher
+func (bp *BatchPublisher) Close() {
+	close(bp.closeChan)
+}

server/backend/sync/memory/pubsub.go

@@ -18,6 +18,7 @@ package memory
 
 import (
 	"context"
+	gotime "time"
 
 	"go.uber.org/zap"
 
@@ -32,13 +33,16 @@ import (
 type Subscriptions struct {
 	docKey      types.DocRefKey
 	internalMap *cmap.Map[string, *sync.Subscription]
+	publisher   *BatchPublisher
 }
 
 func newSubscriptions(docKey types.DocRefKey) *Subscriptions {
-	return &Subscriptions{
+	s := &Subscriptions{
 		docKey:      docKey,
 		internalMap: cmap.New[string, *sync.Subscription](),
 	}
+	s.publisher = NewBatchPublisher(100*gotime.Millisecond, 100, s)
+	return s
 }
 
 // Set adds the given subscription.
@@ -53,39 +57,7 @@ func (s *Subscriptions) Values() []*sync.Subscription {
 
 // Publish publishes the given event.
 func (s *Subscriptions) Publish(ctx context.Context, event sync.DocEvent) {
-	// TODO(hackerwins): Introduce batch publish to reduce lock contention.
-	// Problem:
-	//   - High lock contention when publishing events frequently.
-	//   - Redundant events being published in short time windows.
-	// Solution:
-	// 	 - Collect events to publish in configurable time window.
-	// 	 - Keep only the latest event for the same event type.
-	// 	 - Run dedicated publish loop in a single goroutine.
-	// 	 - Batch publish collected events when the time window expires.
-	for _, sub := range s.internalMap.Values() {
-		if sub.Subscriber().Compare(event.Publisher) == 0 {
-			continue
-		}
-
-		if logging.Enabled(zap.DebugLevel) {
-			logging.From(ctx).Debugf(
-				`Publish %s(%s,%s) to %s`,
-				event.Type,
-				s.docKey,
-				event.Publisher,
-				sub.Subscriber(),
-			)
-		}
-
-		if ok := sub.Publish(event); !ok {
-			logging.From(ctx).Warnf(
-				`Publish(%s,%s) to %s timeout or closed`,
-				s.docKey,
-				event.Publisher,
-				sub.Subscriber(),
-			)
-		}
-	}
+	s.publisher.Publish(ctx, event)
 }
 
 // Delete deletes the subscription of the given id.
@@ -103,6 +75,11 @@ func (s *Subscriptions) Len() int {
 	return s.internalMap.Len()
 }
 
+// Close closes the subscriptions.
+func (s *Subscriptions) Close() {
+	s.publisher.Close()
+}
+
 // PubSub is the memory implementation of PubSub, used for single server.
 type PubSub struct {
 	subscriptionsMap *cmap.Map[types.DocRefKey, *Subscriptions]
@@ -170,6 +147,7 @@ func (m *PubSub) Unsubscribe(
 
 		if subs.Len() == 0 {
 			m.subscriptionsMap.Delete(docKey, func(subs *Subscriptions, exists bool) bool {
+				subs.Close()
 				return exists
 			})
 		}