concurrency: make the lock table limit configurable

Previously, this was hardcoded to 10K. We now introduce a new
cluster setting to change this, if needed. We go through all
ranges to update the limit on all exisiting lock table's in
response to the cluster setting change.

Release note: None

Epic: none

Author: arulajmani

pkg/kv/kvserver/concurrency/concurrency_control.go

@@ -271,6 +271,11 @@ type LockManager interface {
 	// ExportUnreplicatedLocks runs exporter on each held, unreplicated lock
 	// in the given span until the exporter returns false.
 	ExportUnreplicatedLocks(span roachpb.Span, exporter func(*roachpb.LockAcquisition) bool)
+
+	// SetMaxLockTableSize updates the lock table's maximum size limit. It may
+	// be used to dynamically adjust the lock table's size after it has been
+	// initialized.
+	SetMaxLockTableSize(maxLocks int64)
 }
 
 // TransactionManager is concerned with tracking transactions that have their
@@ -354,10 +359,6 @@ type TestingAccessor interface {
 
 	// TestingTxnWaitQueue returns the concurrency manager's txnWaitQueue.
 	TestingTxnWaitQueue() *txnwait.Queue
-
-	// TestingSetMaxLocks updates the locktable's lock limit. This can be used to
-	// force the locktable to exceed its limit and clear locks.
-	TestingSetMaxLocks(n int64)
 }
 
 ///////////////////////////////////
@@ -789,9 +790,10 @@ type lockTable interface {
 	// String returns a debug string representing the state of the lockTable.
 	String() string
 
-	// TestingSetMaxLocks updates the locktable's lock limit. This can be used to
-	// force the locktable to exceed its limit and clear locks.
-	TestingSetMaxLocks(maxLocks int64)
+	// SetMaxLockTableSize updates the lock table's maximum size limit. It may
+	// be used to dynamically adjust the lock table's size after it has been
+	// initialized.
+	SetMaxLockTableSize(maxLocks int64)
 }
 
 // lockTableGuard is a handle to a request as it waits on conflicting locks in a

pkg/kv/kvserver/concurrency/concurrency_manager.go

@@ -103,6 +103,17 @@ var DiscoveredLocksThresholdToConsultTxnStatusCache = settings.RegisterIntSettin
 	settings.NonNegativeInt,
 )
 
+// DefaultLockTableSize controls the default upper bound on the number of locks
+// in a lock table.
+var DefaultLockTableSize = settings.RegisterIntSetting(
+	settings.SystemOnly,
+	"kv.lock_table.default_size",
+	"the default upper bound on the number of locks in a lock table. This setting "+
+		"controls the maximum number of locks that can be held in memory by the lock table "+
+		"before it starts evicting locks to manage memory pressure.",
+	10000,
+)
+
 // BatchPushedLockResolution controls whether the lock table should allow
 // non-locking readers to defer and batch the resolution of conflicting locks
 // whose holder is known to be pending and have been pushed above the reader's
@@ -199,7 +210,7 @@ type Config struct {
 
 func (c *Config) initDefaults() {
 	if c.MaxLockTableSize == 0 {
-		c.MaxLockTableSize = defaultLockTableSize
+		c.MaxLockTableSize = DefaultLockTableSize.Get(&c.Settings.SV)
 	}
 }
 
@@ -759,9 +770,9 @@ func (m *managerImpl) TestingTxnWaitQueue() *txnwait.Queue {
 	return m.twq.(*txnwait.Queue)
 }
 
-// TestingSetMaxLocks implements the TestingAccessor interface.
-func (m *managerImpl) TestingSetMaxLocks(maxLocks int64) {
-	m.lt.TestingSetMaxLocks(maxLocks)
+// SetMaxLockTableSize implements the LockManager interface.
+func (m *managerImpl) SetMaxLockTableSize(maxLocks int64) {
+	m.lt.SetMaxLockTableSize(maxLocks)
 }
 
 func (r *Request) isSingle(m kvpb.Method) bool {

pkg/kv/kvserver/concurrency/concurrency_manager_test.go

@@ -641,7 +641,7 @@ func TestConcurrencyManagerBasic(t *testing.T) {
 			case "debug-set-max-locks":
 				var n int
 				d.ScanArgs(t, "n", &n)
-				m.TestingSetMaxLocks(int64(n))
+				m.SetMaxLockTableSize(int64(n))
 				return ""
 
 			case "reset":

pkg/kv/kvserver/concurrency/lock_table.go

@@ -30,9 +30,6 @@ import (
 	"github.com/cockroachdb/redact"
 )
 
-// Default upper bound on the number of locks in a lockTable.
-const defaultLockTableSize = 10000
-
 // The kind of waiting that the request is subject to.
 type waitKind int
 
@@ -182,10 +179,6 @@ type treeMu struct {
 	// primarily used for constraining memory consumption. Ideally, we should be
 	// doing better memory accounting than this.
 	numKeysLocked atomic.Int64
-
-	// For dampening the frequency with which we enforce
-	// lockTableImpl.maxKeysLocked.
-	lockAddMaxLocksCheckInterval uint64
 }
 
 // lockTableImpl is an implementation of lockTable.
@@ -266,15 +259,21 @@ type lockTableImpl struct {
 	// table. Locks on both Global and Local keys are stored in the same btree.
 	locks treeMu
 
-	// maxKeysLocked is a soft maximum on amount of per-key lock information
-	// tracking[1]. When it is exceeded, and subject to the dampening in
-	// lockAddMaxLocksCheckInterval, locks will be cleared.
-	//
-	// [1] Simply put, the number of keyLocks objects in the lockTable btree.
-	maxKeysLocked int64
-	// When maxKeysLocked is exceeded, will attempt to clear down to minKeysLocked,
-	// instead of clearing everything.
-	minKeysLocked int64
+	// lockTableLimitsMu contains the lock table limit fields protected by a mutex.
+	lockTableLimitsMu struct {
+		syncutil.Mutex
+		// maxKeysLocked is a soft maximum on amount of per-key lock information
+		// tracking[1]. When it is exceeded, and subject to the dampening in
+		// lockAddMaxLocksCheckInterval, locks will be cleared.
+		//
+		// [1] Simply put, the number of keyLocks objects in the lockTable btree.
+		maxKeysLocked int64
+		// When maxKeysLocked is exceeded, will attempt to clear down to minKeysLocked,
+		// instead of clearing everything.
+		minKeysLocked int64
+		// For dampening the frequency with which we enforce maxKeysLocked.
+		lockAddMaxLocksCheckInterval uint64
+	}
 
 	// txnStatusCache is a small LRU cache that tracks the status of
 	// transactions that have been successfully pushed.
@@ -321,14 +320,27 @@ func newLockTable(
 }
 
 func (t *lockTableImpl) setMaxKeysLocked(maxKeysLocked int64) {
+	t.lockTableLimitsMu.Lock()
+	defer t.lockTableLimitsMu.Unlock()
+
 	// Check at 5% intervals of the max count.
 	lockAddMaxLocksCheckInterval := maxKeysLocked / int64(20)
 	if lockAddMaxLocksCheckInterval == 0 {
 		lockAddMaxLocksCheckInterval = 1
 	}
-	t.maxKeysLocked = maxKeysLocked
-	t.minKeysLocked = maxKeysLocked / 2
-	t.locks.lockAddMaxLocksCheckInterval = uint64(lockAddMaxLocksCheckInterval)
+	t.lockTableLimitsMu.maxKeysLocked = maxKeysLocked
+	t.lockTableLimitsMu.minKeysLocked = maxKeysLocked / 2
+	t.lockTableLimitsMu.lockAddMaxLocksCheckInterval = uint64(lockAddMaxLocksCheckInterval)
+}
+
+// shouldCheckMaxLocks returns true if allocating the supplied sequence number
+// implies that we should check whether the lock table has exceeded its max
+// locks limit or not. We dampen how often the check is performed.
+func (t *lockTableImpl) shouldCheckMaxLocks(seqNum uint64) bool {
+	t.lockTableLimitsMu.Lock()
+	defer t.lockTableLimitsMu.Unlock()
+	interval := t.lockTableLimitsMu.lockAddMaxLocksCheckInterval
+	return seqNum%interval == 0
 }
 
 // lockTableGuardImpl is an implementation of lockTableGuard.
@@ -4161,10 +4173,9 @@ func (t *treeMu) Reset() {
 	t.btree.Reset()
 }
 
-func (t *treeMu) nextLockSeqNum() (seqNum uint64, checkMaxLocks bool) {
+func (t *treeMu) nextLockSeqNum() uint64 {
 	t.lockIDSeqNum++
-	checkMaxLocks = t.lockIDSeqNum%t.lockAddMaxLocksCheckInterval == 0
-	return t.lockIDSeqNum, checkMaxLocks
+	return t.lockIDSeqNum
 }
 
 func (t *lockTableImpl) ScanOptimistic(req Request) lockTableGuard {
@@ -4352,10 +4363,9 @@ func (t *lockTableImpl) AddDiscoveredLock(
 	t.locks.mu.Lock()
 	iter := t.locks.MakeIter()
 	iter.FirstOverlap(&keyLocks{key: key})
-	checkMaxLocks := false
+	var lockSeqNum uint64
 	if !iter.Valid() {
-		var lockSeqNum uint64
-		lockSeqNum, checkMaxLocks = t.locks.nextLockSeqNum()
+		lockSeqNum = t.locks.nextLockSeqNum()
 		l = &keyLocks{id: lockSeqNum, key: key}
 		l.queuedLockingRequests.Init()
 		l.waitingReaders.Init()
@@ -4379,7 +4389,7 @@ func (t *lockTableImpl) AddDiscoveredLock(
 	// Can't release tree.mu until call l.discoveredLock() since someone may
 	// find an empty lock and remove it from the tree.
 	t.locks.mu.Unlock()
-	if checkMaxLocks {
+	if t.shouldCheckMaxLocks(lockSeqNum) {
 		t.checkMaxKeysLockedAndTryClear()
 	}
 	if err != nil {
@@ -4416,7 +4426,7 @@ func (t *lockTableImpl) AcquireLock(acq *roachpb.LockAcquisition) error {
 	// tree.mu.RLock().
 	iter := t.locks.MakeIter()
 	iter.FirstOverlap(&keyLocks{key: acq.Key})
-	checkMaxLocks := false
+	var lockSeqNum uint64
 	if !iter.Valid() {
 		if acq.Durability == lock.Replicated {
 			// Don't remember uncontended replicated locks. The downside is that
@@ -4428,8 +4438,7 @@ func (t *lockTableImpl) AcquireLock(acq *roachpb.LockAcquisition) error {
 			t.locks.mu.Unlock()
 			return nil
 		}
-		var lockSeqNum uint64
-		lockSeqNum, checkMaxLocks = t.locks.nextLockSeqNum()
+		lockSeqNum = t.locks.nextLockSeqNum()
 		kl = &keyLocks{id: lockSeqNum, key: acq.Key}
 		kl.queuedLockingRequests.Init()
 		kl.waitingReaders.Init()
@@ -4462,7 +4471,7 @@ func (t *lockTableImpl) AcquireLock(acq *roachpb.LockAcquisition) error {
 	err := kl.acquireLock(acq, t.clock, t.settings)
 	t.locks.mu.Unlock()
 
-	if checkMaxLocks {
+	if t.shouldCheckMaxLocks(lockSeqNum) {
 		t.checkMaxKeysLockedAndTryClear()
 	}
 	if err != nil {
@@ -4506,9 +4515,12 @@ func (t *lockTableImpl) MarkIneligibleForExport(acq *roachpb.LockAcquisition) er
 // method relieves memory pressure by clearing as much per-key tracking as it
 // can to bring things under budget.
 func (t *lockTableImpl) checkMaxKeysLockedAndTryClear() {
+	t.lockTableLimitsMu.Lock()
+	defer t.lockTableLimitsMu.Unlock()
+
 	totalLocks := t.locks.numKeysLocked.Load()
-	if totalLocks > t.maxKeysLocked {
-		numToClear := totalLocks - t.minKeysLocked
+	if totalLocks > t.lockTableLimitsMu.maxKeysLocked {
+		numToClear := totalLocks - t.lockTableLimitsMu.minKeysLocked
 		numCleared := t.tryClearLocks(false /* force */, int(numToClear))
 		// Update metrics if we successfully cleared any number of locks.
 		if numCleared != 0 {
@@ -4948,8 +4960,8 @@ func (t *lockTableImpl) stringRLocked() string {
 	return sb.String()
 }
 
-// TestingSetMaxLocks implements the lockTable interface.
-func (t *lockTableImpl) TestingSetMaxLocks(maxKeysLocked int64) {
+// SetMaxLockTableSize implements the lockTable interface.
+func (t *lockTableImpl) SetMaxLockTableSize(maxKeysLocked int64) {
 	t.setMaxKeysLocked(maxKeysLocked)
 }
 

pkg/kv/kvserver/concurrency/lock_table_test.go

@@ -203,7 +203,7 @@ func TestLockTableBasic(t *testing.T) {
 				)
 				ltImpl.enabled = true
 				ltImpl.enabledSeq = 1
-				ltImpl.minKeysLocked = 0
+				ltImpl.lockTableLimitsMu.minKeysLocked = 0
 				lt = maybeWrapInVerifyingLockTable(ltImpl)
 				txnsByName = make(map[string]*enginepb.TxnMeta)
 				txnCounter = uint128.FromInts(0, 0)
@@ -895,7 +895,7 @@ func TestLockTableMaxLocks(t *testing.T) {
 		5, roachpb.RangeID(3), hlc.NewClockForTesting(nil), cluster.MakeTestingClusterSettings(),
 		m.LocksShedDueToMemoryLimit, m.NumLockShedDueToMemoryLimitEvents,
 	)
-	lt.minKeysLocked = 0
+	lt.lockTableLimitsMu.minKeysLocked = 0
 	lt.enabled = true
 	var keys []roachpb.Key
 	var guards []lockTableGuard
@@ -990,7 +990,7 @@ func TestLockTableMaxLocks(t *testing.T) {
 	lt.Dequeue(guards[4])
 	lt.Dequeue(guards[5])
 	// Bump up the enforcement interval manually.
-	lt.locks.lockAddMaxLocksCheckInterval = 2
+	lt.lockTableLimitsMu.lockAddMaxLocksCheckInterval = 2
 	// Add another discovered lock.
 	added, err = lt.AddDiscoveredLock(
 		newLock(&txnMeta, keys[9*20+12], lock.Intent),
@@ -1015,8 +1015,8 @@ func TestLockTableMaxLocks(t *testing.T) {
 	require.Equal(t, int64(100), m.LocksShedDueToMemoryLimit.Count())
 	require.Equal(t, int64(69), m.NumLockShedDueToMemoryLimitEvents.Count())
 	// Bump down the enforcement interval manually, and bump up minKeysLocked.
-	lt.locks.lockAddMaxLocksCheckInterval = 1
-	lt.minKeysLocked = 2
+	lt.lockTableLimitsMu.lockAddMaxLocksCheckInterval = 1
+	lt.lockTableLimitsMu.minKeysLocked = 2
 	// Three more guards dequeued. Now only 1 lock is notRemovable.
 	lt.Dequeue(guards[6])
 	lt.Dequeue(guards[7])
@@ -1046,7 +1046,7 @@ func TestLockTableMaxLocks(t *testing.T) {
 	require.Equal(t, int64(104), m.LocksShedDueToMemoryLimit.Count())
 	require.Equal(t, int64(70), m.NumLockShedDueToMemoryLimitEvents.Count())
 	// Restore minKeysLocked to 0.
-	lt.minKeysLocked = 0
+	lt.lockTableLimitsMu.minKeysLocked = 0
 	// Add locks to push us over 5 locks.
 	for i := 16; i < 20; i++ {
 		added, err = lt.AddDiscoveredLock(
@@ -1071,7 +1071,7 @@ func TestLockTableMaxLocksWithMultipleNotRemovableRefs(t *testing.T) {
 		2, roachpb.RangeID(3), hlc.NewClockForTesting(nil), cluster.MakeTestingClusterSettings(),
 		m.LocksShedDueToMemoryLimit, m.NumLockShedDueToMemoryLimitEvents,
 	)
-	lt.minKeysLocked = 0
+	lt.lockTableLimitsMu.minKeysLocked = 0
 	lt.enabled = true
 	var keys []roachpb.Key
 	var guards []lockTableGuard

pkg/kv/kvserver/concurrency/verifiable_lock_table.go

@@ -133,7 +133,7 @@ func (v verifyingLockTable) String() string {
 	return v.lt.String()
 }
 
-// TestingSetMaxLocks implements the lockTable interface.
-func (v verifyingLockTable) TestingSetMaxLocks(maxKeysLocked int64) {
-	v.lt.TestingSetMaxLocks(maxKeysLocked)
+// SetMaxLockTableSize implements the lockTable interface.
+func (v verifyingLockTable) SetMaxLockTableSize(maxKeysLocked int64) {
+	v.lt.SetMaxLockTableSize(maxKeysLocked)
 }

pkg/kv/kvserver/store.go

@@ -37,6 +37,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/ctpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/policyrefresher"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/closedts/sidetransport"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/idalloc"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/intentresolver"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvadmission"
@@ -1743,6 +1744,14 @@ func NewStore(
 	queueAdditionOnSystemConfigUpdateBurst.SetOnChange(&cfg.Settings.SV,
 		updateSystemConfigUpdateQueueLimits)
 
+	concurrency.DefaultLockTableSize.SetOnChange(&cfg.Settings.SV, func(ctx context.Context) {
+		newSize := concurrency.DefaultLockTableSize.Get(&cfg.Settings.SV)
+		s.VisitReplicas(func(repl *Replica) bool {
+			repl.concMgr.SetMaxLockTableSize(newSize)
+			return true // continue
+		})
+	})
+
 	if s.cfg.Gossip != nil {
 		s.storeGossip = NewStoreGossip(
 			cfg.Gossip,