Adjust lock scope for improved concurrency (#1283)

Removed unnecessary context dependency from the Lock interface.
Moved parts of the client patching logic outside the lock scope to
enhance concurrency and reduce contention.

Author: hackerwins

server/backend/database/mongo/client.go

@@ -825,14 +825,15 @@ func (c *Client) FindDocInfoByKeyAndOwner(
 	docKey key.Key,
 	createDocIfNotExist bool,
 ) (*database.DocInfo, error) {
+	now := gotime.Now()
+
 	filter := bson.M{
 		"project_id": clientRefKey.ProjectID,
 		"key":        docKey,
 		"removed_at": bson.M{
 			"$exists": false,
 		},
 	}
-	now := gotime.Now()
 	res, err := c.collection(ColDocuments).UpdateOne(ctx, filter, bson.M{
 		"$set": bson.M{
 			"accessed_at": now,

server/backend/sync/locker.go

@@ -18,7 +18,6 @@
 package sync
 
 import (
-	"context"
 	"errors"
 
 	"github.com/yorkie-team/yorkie/pkg/locker"
@@ -53,32 +52,41 @@ func New() *LockerManager {
 }
 
 // Locker creates locker of the given key.
-func (c *LockerManager) Locker(
-	_ context.Context,
-	key Key,
-) (Locker, error) {
-	return &internalLocker{
+func (c *LockerManager) Locker(key Key) Locker {
+	locker := &internalLocker{
 		key.String(),
 		c.locks,
-	}, nil
+	}
+	locker.lock()
+	return locker
+}
+
+// LockerWithTryLock creates locker of the given key with try lock.
+func (c *LockerManager) LockerWithTryLock(key Key) (Locker, bool) {
+	locker := &internalLocker{
+		key.String(),
+		c.locks,
+	}
+	ok := locker.tryLock()
+	return locker, ok
 }
 
 // A Locker represents an object that can be locked and unlocked.
 type Locker interface {
-	// Lock locks the mutex with a cancelable context
-	Lock(ctx context.Context) error
+	// Lock locks the mutex.
+	lock()
 
 	// TryLock locks the mutex if not already locked by another session.
-	TryLock(ctx context.Context) error
+	tryLock() bool
 
 	// Unlock unlocks the mutex.
-	Unlock(ctx context.Context) error
+	Unlock() error
 
-	// RLock acquires a read lock with a cancelable context.
-	RLock(ctx context.Context) error
+	// RLock acquires a read lock.
+	RLock()
 
 	// RUnlock releases a read lock previously acquired by RLock.
-	RUnlock(ctx context.Context) error
+	RUnlock() error
 }
 
 type internalLocker struct {
@@ -87,42 +95,26 @@ type internalLocker struct {
 }
 
 // Lock locks the mutex.
-func (il *internalLocker) Lock(_ context.Context) error {
+func (il *internalLocker) lock() {
 	il.locks.Lock(il.key)
-
-	return nil
 }
 
 // TryLock locks the mutex if not already locked by another session.
-func (il *internalLocker) TryLock(_ context.Context) error {
-	if !il.locks.TryLock(il.key) {
-		return ErrAlreadyLocked
-	}
-
-	return nil
+func (il *internalLocker) tryLock() bool {
+	return il.locks.TryLock(il.key)
 }
 
 // Unlock unlocks the mutex.
-func (il *internalLocker) Unlock(_ context.Context) error {
-	if err := il.locks.Unlock(il.key); err != nil {
-		return err
-	}
-
-	return nil
+func (il *internalLocker) Unlock() error {
+	return il.locks.Unlock(il.key)
 }
 
 // RLock locks the mutex for reading..
-func (il *internalLocker) RLock(_ context.Context) error {
+func (il *internalLocker) RLock() {
 	il.locks.RLock(il.key)
-
-	return nil
 }
 
 // RUnlock unlocks the read lock.
-func (il *internalLocker) RUnlock(_ context.Context) error {
-	if err := il.locks.RUnlock(il.key); err != nil {
-		return err
-	}
-
-	return nil
+func (il *internalLocker) RUnlock() error {
+	return il.locks.RUnlock(il.key)
 }

server/clients/clients.go

@@ -76,7 +76,14 @@ func Deactivate(
 			return nil, err
 		}
 
-		if err := be.ClusterClient.DetachDocument(ctx, project, actorID, docID, project.PublicKey, docInfo.Key); err != nil {
+		if err := be.ClusterClient.DetachDocument(
+			ctx,
+			project,
+			actorID,
+			docID,
+			project.PublicKey,
+			docInfo.Key,
+		); err != nil {
 			return nil, err
 		}
 	}

server/clients/housekeeping.go

@@ -41,17 +41,9 @@ func DeactivateInactives(
 ) (types.ID, error) {
 	start := time.Now()
 
-	locker, err := be.Lockers.Locker(ctx, deactivateCandidatesKey)
-	if err != nil {
-		return database.DefaultProjectID, err
-	}
-
-	if err := locker.Lock(ctx); err != nil {
-		return database.DefaultProjectID, err
-	}
-
+	locker := be.Lockers.Locker(deactivateCandidatesKey)
 	defer func() {
-		if err := locker.Unlock(ctx); err != nil {
+		if err := locker.Unlock(); err != nil {
 			logging.From(ctx).Error(err)
 		}
 	}()

server/documents/housekeeping.go

@@ -42,17 +42,9 @@ func CompactDocuments(
 ) (types.ID, error) {
 	start := time.Now()
 
-	locker, err := be.Lockers.Locker(ctx, compactionCandidatesKey)
-	if err != nil {
-		return database.DefaultProjectID, err
-	}
-
-	if err := locker.Lock(ctx); err != nil {
-		return database.DefaultProjectID, err
-	}
-
+	locker := be.Lockers.Locker(compactionCandidatesKey)
 	defer func() {
-		if err := locker.Unlock(ctx); err != nil {
+		if err := locker.Unlock(); err != nil {
 			logging.From(ctx).Error(err)
 		}
 	}()

server/packs/packs.go

@@ -76,28 +76,19 @@ func PushPull(
 	opts PushPullOptions,
 ) (*ServerPack, error) {
 	start := gotime.Now()
-	defer func() {
-		be.Metrics.ObservePushPullResponseSeconds(gotime.Since(start).Seconds())
-	}()
 
 	// TODO: Changes may be reordered or missing during communication on the network.
 	// We should check the change.pack with checkpoint to make sure the changes are in the correct order.
 	initialServerSeq := docInfo.ServerSeq
 
 	// 01. push changes: filter out the changes that are already saved in the database.
 	cpAfterPush, pushedChanges := pushChanges(ctx, clientInfo, docInfo, reqPack, initialServerSeq)
-	hostname := be.Config.Hostname
-	be.Metrics.AddPushPullReceivedChanges(hostname, project, reqPack.ChangesLen())
-	be.Metrics.AddPushPullReceivedOperations(hostname, project, reqPack.OperationsLen())
 
 	// 02. pull pack: pull changes or a snapshot from the database and create a response pack.
 	respPack, err := pullPack(ctx, be, clientInfo, docInfo, reqPack, cpAfterPush, initialServerSeq, opts.Mode)
 	if err != nil {
 		return nil, err
 	}
-	be.Metrics.AddPushPullSentChanges(hostname, project, respPack.ChangesLen())
-	be.Metrics.AddPushPullSentOperations(hostname, project, respPack.OperationsLen())
-	be.Metrics.AddPushPullSnapshotBytes(hostname, project, respPack.SnapshotLen())
 
 	// 03. update the client's document and checkpoint.
 	docRefKey := docInfo.RefKey()
@@ -115,7 +106,7 @@ func PushPull(
 		}
 	}
 
-	// 04. store pushed changes, docInfo and checkpoint of the client to DB.
+	// 04. store the pushed changes and update the document info in DB.
 	if len(pushedChanges) > 0 || reqPack.IsRemoved {
 		if err := be.DB.CreateChangeInfos(
 			ctx,
@@ -128,12 +119,7 @@ func PushPull(
 			return nil, err
 		}
 	}
-
-	if !clientInfo.IsServerClient() {
-		if err := be.DB.UpdateClientInfoAfterPushPull(ctx, clientInfo, docInfo); err != nil {
-			return nil, err
-		}
-	}
+	respPack.ApplyDocInfo(docInfo)
 
 	// 05. update min version vector of the document.
 	minVersionVector, err := be.DB.UpdateMinVersionVector(
@@ -149,7 +135,12 @@ func PushPull(
 		respPack.VersionVector = minVersionVector
 	}
 
-	respPack.ApplyDocInfo(docInfo)
+	// 06. update client's checkpoint of the document in DB.
+	if !clientInfo.IsServerClient() {
+		if err := be.DB.UpdateClientInfoAfterPushPull(ctx, clientInfo, docInfo); err != nil {
+			return nil, err
+		}
+	}
 
 	pullLog := strconv.Itoa(respPack.ChangesLen())
 	if respPack.SnapshotLen() > 0 {
@@ -163,8 +154,15 @@ func PushPull(
 		pullLog,
 		gotime.Since(start),
 	)
+	hostname := be.Config.Hostname
+	be.Metrics.AddPushPullReceivedChanges(hostname, project, reqPack.ChangesLen())
+	be.Metrics.AddPushPullReceivedOperations(hostname, project, reqPack.OperationsLen())
+	be.Metrics.AddPushPullSentChanges(hostname, project, respPack.ChangesLen())
+	be.Metrics.AddPushPullSentOperations(hostname, project, respPack.OperationsLen())
+	be.Metrics.AddPushPullSnapshotBytes(hostname, project, respPack.SnapshotLen())
+	be.Metrics.ObservePushPullResponseSeconds(gotime.Since(start).Seconds())
 
-	// 06. publish document change event then store snapshot asynchronously.
+	// 07. publish document change event then store snapshot asynchronously.
 	if len(pushedChanges) > 0 || reqPack.IsRemoved {
 		be.Background.AttachGoroutine(func(ctx context.Context) {
 			publisherID, err := clientInfo.ID.ToActorID()
@@ -200,30 +198,21 @@ func PushPull(
 				}
 			}
 
-			locker, err := be.Lockers.Locker(ctx, SnapshotKey(project.ID, reqPack.DocumentKey))
-			if err != nil {
-				logging.From(ctx).Error(err)
-				return
-			}
-
-			// NOTE: If the snapshot is already being created by another routine, it
-			//       is not necessary to recreate it, so we can skip it.
-			if err := locker.TryLock(ctx); err != nil {
+			// NOTE(hackerwins): If the snapshot is already being created by another routine,
+			// it is not necessary to recreate it, so we can skip it.
+			locker, ok := be.Lockers.LockerWithTryLock(SnapshotKey(project.ID, reqPack.DocumentKey))
+			if !ok {
 				return
 			}
 			defer func() {
-				if err := locker.Unlock(ctx); err != nil {
+				if err := locker.Unlock(); err != nil {
 					logging.From(ctx).Error(err)
 					return
 				}
 			}()
 
 			start := gotime.Now()
-			if err := storeSnapshot(
-				ctx,
-				be,
-				docInfo,
-			); err != nil {
+			if err := storeSnapshot(ctx, be, docInfo); err != nil {
 				logging.From(ctx).Error(err)
 			}
 			be.Metrics.ObservePushPullSnapshotDurationSeconds(
@@ -260,18 +249,18 @@ func BuildInternalDocForServerSeq(
 	docInfo *database.DocInfo,
 	serverSeq int64,
 ) (*document.InternalDocument, error) {
-	docRefKey := docInfo.RefKey()
+	docKey := docInfo.RefKey()
 
 	// NOTE(hackerwins): If the document is already in the cache, we can skip
 	// the database query and use the cached document. If the document's server
 	// sequence in the cache is greater than the given server sequence, we can't
 	// build the document from the document. In this case, we need to
 	// query the database to get the closest snapshot information.
-	doc, ok := be.SnapshotCache.Get(docRefKey)
+	doc, ok := be.SnapshotCache.Get(docKey)
 	if !ok || serverSeq < doc.Checkpoint().ServerSeq {
 		snapshotInfo, err := be.DB.FindClosestSnapshotInfo(
 			ctx,
-			docRefKey,
+			docKey,
 			serverSeq,
 			true,
 		)
@@ -293,7 +282,7 @@ func BuildInternalDocForServerSeq(
 
 	changes, err := be.DB.FindChangesBetweenServerSeqs(
 		ctx,
-		docRefKey,
+		docKey,
 		doc.Checkpoint().ServerSeq+1,
 		serverSeq,
 	)
@@ -314,7 +303,7 @@ func BuildInternalDocForServerSeq(
 	if !be.Config.SnapshotDisableGC {
 		vector, err := be.DB.GetMinVersionVector(
 			ctx,
-			docRefKey,
+			docKey,
 			doc.VersionVector(),
 		)
 		if err != nil {
@@ -330,7 +319,7 @@ func BuildInternalDocForServerSeq(
 	if err != nil {
 		return nil, err
 	}
-	be.SnapshotCache.Add(docRefKey, clone)
+	be.SnapshotCache.Add(docKey, clone)
 
 	if logging.Enabled(zap.DebugLevel) {
 		logging.From(ctx).Debugf(