perf(selector): shard simple selector's in-memory locker by owner and jitter retries (#1981)

Signed-off-by: hstarorg <jayhu@sign.global>

Author: hstarorg

token/services/selector/benchmark_test.go

@@ -27,8 +27,8 @@ import (
 type WalletIDByRawIdentityFunc func(rawIdentity []byte) string
 
 type Locker interface {
-	Lock(ctx context.Context, id *token2.ID, txID string, reclaim bool) (string, error)
-	UnlockIDs(ctx context.Context, ids ...*token2.ID) []*token2.ID
+	Lock(ctx context.Context, owner string, id *token2.ID, txID string, reclaim bool) (string, error)
+	UnlockIDs(ctx context.Context, owner string, ids ...*token2.ID) []*token2.ID
 	UnlockByTxID(ctx context.Context, txID string)
 	IsLocked(id *token2.ID) bool
 }
@@ -43,9 +43,9 @@ func (s *extendedSelector) Select(ctx context.Context, ownerFilter token.OwnerFi
 }
 func (s *extendedSelector) Close() error { return s.Selector.Close() }
 
-func (s *extendedSelector) Unselect(id ...*token2.ID) {
+func (s *extendedSelector) Unselect(owner string, id ...*token2.ID) {
 	if s.Lock != nil {
-		s.Lock.UnlockIDs(context.Background(), id...)
+		s.Lock.UnlockIDs(context.Background(), owner, id...)
 	}
 }
 
@@ -72,7 +72,7 @@ func BenchmarkSelectorSingle(b *testing.B) {
 				wg.Add(1)
 				go func(ids []*token2.ID) {
 					defer wg.Done()
-					s.selector.Unselect(ids...)
+					s.selector.Unselect(s.filter.ID(), ids...)
 				}(ids)
 			}
 		})
@@ -110,7 +110,7 @@ func BenchmarkSelectorParallel(b *testing.B) {
 					wg.Add(1)
 					go func(ids []*token2.ID) {
 						defer wg.Done()
-						s.selector.Unselect(ids...)
+						s.selector.Unselect(s.filter.ID(), ids...)
 					}(ids)
 				}
 			})
@@ -199,7 +199,7 @@ type LockerProviderFunction func() selector.Locker
 
 type ExtendedSelector interface {
 	token.Selector
-	Unselect(id ...*token2.ID)
+	Unselect(owner string, id ...*token2.ID)
 }
 
 type MockTokenIterator struct {

token/services/selector/sherdlock/inmemory/locker.go

@@ -15,7 +15,7 @@ import (
 )
 
 type Locker interface {
-	Lock(ctx context.Context, id *token.ID, txID string, reclaim bool) (string, error)
+	Lock(ctx context.Context, owner string, id *token.ID, txID string, reclaim bool) (string, error)
 	UnlockByTxID(ctx context.Context, txID string)
 }
 
@@ -32,7 +32,8 @@ func NewLocker(l Locker) *locker {
 }
 
 func (l *locker) Lock(ctx context.Context, tokenID *token.ID, consumerTxID transaction.ID) error {
-	_, err := l.Locker.Lock(ctx, tokenID, consumerTxID, false)
+	// this adapter has no owner context; the empty owner shares one default shard
+	_, err := l.Locker.Lock(ctx, "", tokenID, consumerTxID, false)
 
 	return err
 }