execstats: add latch wait time into cumulative contention time

This patch adds a field `IsLatch` to the `ContentionEvent` protobuf struct that
indicates that a contention event was emitted by the latch table. The event is
added to the trace and used to increment the contention time metrics, but is
only added to the contention registry if
`register_latch_wait_contention_events` is set to true.

This commit also propagates the transaction ID into the latch manager and the
contention events listener. This information is used during a latch wait event
to distinguish between time waiting for a transaction's own requests
(e.g. for QueryIntent) and for real contention between different transactions.

Fixes #103713

Release note (ui change): The contention time metric now includes time spent
waiting to acquire latches in addition to time spent acquiring locks. This
applies to both the cumulative contention time visible in the db console and
the annotations of an `EXPLAIN ANALYZE` output.

Author: DrewKimball

pkg/kv/kvpb/api.go

@@ -2349,7 +2349,13 @@ func (r *IsSpanEmptyResponse) IsEmpty() bool {
 
 // SafeFormat implements redact.SafeFormatter.
 func (c *ContentionEvent) SafeFormat(w redact.SafePrinter, _ rune) {
-	w.Printf("conflicted with %s on %s for %.3fs", c.TxnMeta.ID, c.Key, c.Duration.Seconds())
+	prefix := redact.SafeString("conflicted")
+	if c.IsLatch {
+		prefix = "latch conflict"
+	}
+	w.Printf("%s with %s on %s for %.3fs",
+		prefix, c.TxnMeta.ID, c.Key, c.Duration.Seconds(),
+	)
 }
 
 // String implements fmt.Stringer.

pkg/kv/kvpb/api.proto

@@ -3737,6 +3737,9 @@ message ContentionEvent {
   // Duration spent contending against the other transaction.
   google.protobuf.Duration duration = 3 [(gogoproto.nullable) = false,
                                          (gogoproto.stdduration) = true];
+  // IsLatch, if true, indicates that the contention event was due to waiting to
+  // acquire a latch.
+  bool is_latch = 4;
 }
 
 // ScanStats is a message that tracks miscellaneous statistics of all Gets,

pkg/kv/kvserver/concurrency/concurrency_control.go

@@ -22,7 +22,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
-	"github.com/cockroachdb/redact"
 )
 
 // Manager is a structure that sequences incoming requests and provides
@@ -446,9 +445,8 @@ type Request struct {
 	// not also passed an exiting Guard.
 	LockSpans *lockspanset.LockSpanSet
 
-	// The SafeFormatter capable of formatting the request. This is used to enrich
-	// logging with request level information when latches conflict.
-	BaFmt redact.SafeFormatter
+	// Batch is the batch to which the request belongs.
+	Batch *kvpb.BatchRequest
 
 	// DeadlockTimeout is the amount of time that the request will wait on a lock
 	// before pushing the lock holder's transaction for deadlock detection.
@@ -531,7 +529,12 @@ type latchManager interface {
 	// WaitFor waits for conflicting latches on the specified spans without adding
 	// any latches itself. Fast path for operations that only require flushing out
 	// old operations without blocking any new ones.
-	WaitFor(ctx context.Context, spans *spanset.SpanSet, pp poison.Policy, baFmt redact.SafeFormatter) *Error
+	WaitFor(
+		ctx context.Context,
+		spans *spanset.SpanSet,
+		pp poison.Policy,
+		ba *kvpb.BatchRequest,
+	) *Error
 
 	// Poison a guard's latches, allowing waiters to fail fast.
 	Poison(latchGuard)

pkg/kv/kvserver/concurrency/concurrency_manager.go

@@ -181,6 +181,7 @@ func NewManager(cfg Config) Manager {
 				cfg.SlowLatchGauge,
 				cfg.Settings,
 				cfg.LatchWaitDurations,
+				cfg.Clock,
 			),
 		},
 		lt: lt,
@@ -269,7 +270,7 @@ func (m *managerImpl) sequenceReqWithGuard(
 	// them.
 	if shouldWaitOnLatchesWithoutAcquiring(g.Req) {
 		log.Event(ctx, "waiting on latches without acquiring")
-		return nil, m.lm.WaitFor(ctx, g.Req.LatchSpans, g.Req.PoisonPolicy, g.Req.BaFmt)
+		return nil, m.lm.WaitFor(ctx, g.Req.LatchSpans, g.Req.PoisonPolicy, g.Req.Batch)
 	}
 
 	// Provide the manager with an opportunity to intercept the request. It

pkg/kv/kvserver/concurrency/concurrency_manager_test.go

@@ -220,7 +220,7 @@ func TestConcurrencyManagerBasic(t *testing.T) {
 					LatchSpans:             latchSpans,
 					LockSpans:              lockSpans,
 					PoisonPolicy:           pp,
-					BaFmt:                  ba,
+					Batch:                  ba,
 				}
 				return ""
 

pkg/kv/kvserver/concurrency/latch_manager.go

@@ -12,7 +12,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/poison"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanlatch"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/spanset"
-	"github.com/cockroachdb/redact"
 )
 
 // latchManagerImpl implements the latchManager interface.
@@ -21,15 +20,15 @@ type latchManagerImpl struct {
 }
 
 func (m *latchManagerImpl) Acquire(ctx context.Context, req Request) (latchGuard, *Error) {
-	lg, err := m.m.Acquire(ctx, req.LatchSpans, req.PoisonPolicy, req.BaFmt)
+	lg, err := m.m.Acquire(ctx, req.LatchSpans, req.PoisonPolicy, req.Batch)
 	if err != nil {
 		return nil, kvpb.NewError(err)
 	}
 	return lg, nil
 }
 
 func (m *latchManagerImpl) AcquireOptimistic(req Request) latchGuard {
-	lg := m.m.AcquireOptimistic(req.LatchSpans, req.PoisonPolicy, req.BaFmt)
+	lg := m.m.AcquireOptimistic(req.LatchSpans, req.PoisonPolicy, req.Batch)
 	return lg
 }
 
@@ -48,9 +47,9 @@ func (m *latchManagerImpl) WaitUntilAcquired(
 }
 
 func (m *latchManagerImpl) WaitFor(
-	ctx context.Context, ss *spanset.SpanSet, pp poison.Policy, baFmt redact.SafeFormatter,
+	ctx context.Context, ss *spanset.SpanSet, pp poison.Policy, ba *kvpb.BatchRequest,
 ) *Error {
-	err := m.m.WaitFor(ctx, ss, pp, baFmt)
+	err := m.m.WaitFor(ctx, ss, pp, ba)
 	if err != nil {
 		return kvpb.NewError(err)
 	}

pkg/kv/kvserver/concurrency/lock_table_test.go

@@ -326,7 +326,7 @@ func TestLockTableBasic(t *testing.T) {
 					MaxLockWaitQueueLength: maxLockWaitQueueLength,
 					LatchSpans:             latchSpans,
 					LockSpans:              lockSpans,
-					BaFmt:                  ba,
+					Batch:                  ba,
 				}
 				if txnMeta != nil {
 					// Update the transaction's timestamp, if necessary. The transaction
@@ -890,7 +890,7 @@ func TestLockTableMaxLocks(t *testing.T) {
 			Timestamp:  ba.Timestamp,
 			LatchSpans: latchSpans,
 			LockSpans:  lockSpans,
-			BaFmt:      ba,
+			Batch:      ba,
 		}
 		reqs = append(reqs, req)
 		ltg, err := lt.ScanAndEnqueue(req, nil)
@@ -1030,7 +1030,7 @@ func TestLockTableMaxLocksWithMultipleNotRemovableRefs(t *testing.T) {
 			Timestamp:  ba.Timestamp,
 			LatchSpans: latchSpans,
 			LockSpans:  lockSpans,
-			BaFmt:      ba,
+			Batch:      ba,
 		}
 		ltg, err := lt.ScanAndEnqueue(req, nil)
 		require.Nil(t, err)
@@ -1129,7 +1129,7 @@ func doWork(ctx context.Context, item *workItem, e *workloadExecutor) error {
 			// cancellation, the code makes sure to release latches when returning
 			// early due to error. Otherwise other requests will get stuck and
 			// group.Wait() will not return until the test times out.
-			lg, err = e.lm.Acquire(context.Background(), item.request.LatchSpans, poison.Policy_Error, item.request.BaFmt)
+			lg, err = e.lm.Acquire(context.Background(), item.request.LatchSpans, poison.Policy_Error, item.request.Batch)
 			if err != nil {
 				return err
 			}
@@ -1269,7 +1269,7 @@ func makeWorkItemForRequest(wi workloadItem) workItem {
 }
 
 type workloadExecutor struct {
-	lm spanlatch.Manager
+	lm *spanlatch.Manager
 	lt lockTable
 
 	// Protects the following fields in transactionState: acquiredLocks and
@@ -1285,19 +1285,27 @@ type workloadExecutor struct {
 
 func newWorkLoadExecutor(items []workloadItem, concurrency int) *workloadExecutor {
 	const maxLocks = 100000
-	ltImpl := newLockTable(
-		maxLocks, roachpb.RangeID(3), hlc.NewClockForTesting(nil), cluster.MakeTestingClusterSettings(),
+	clock := hlc.NewClockForTesting(nil)
+	settings := cluster.MakeTestingClusterSettings()
+	lm := spanlatch.Make(
+		nil, /* stopper */
+		nil, /* slowReqs */
+		settings,
+		nil, /* latchWaitDurations */
+		clock,
 	)
+	ltImpl := newLockTable(maxLocks, roachpb.RangeID(3), clock, settings)
 	ltImpl.enabled = true
 	lt := maybeWrapInVerifyingLockTable(ltImpl)
-	return &workloadExecutor{
-		lm:           spanlatch.Manager{},
+	ex := &workloadExecutor{
+		lm:           &lm,
 		lt:           lt,
 		items:        items,
 		transactions: make(map[uuid.UUID]*transactionState),
 		doneWork:     make(chan *workItem),
 		concurrency:  concurrency,
 	}
+	return ex
 }
 
 func (e *workloadExecutor) acquireLock(txn *roachpb.Transaction, toAcq lockToAcquire) error {
@@ -1553,7 +1561,7 @@ func TestLockTableConcurrentSingleRequests(t *testing.T) {
 			Timestamp:  ba.Timestamp,
 			LatchSpans: latchSpans,
 			LockSpans:  lockSpans,
-			BaFmt:      ba,
+			Batch:      ba,
 		}
 		items = append(items, workloadItem{request: request})
 		if txn != nil {
@@ -1693,7 +1701,7 @@ func testLockTableConcurrentRequests(
 			Timestamp:  ba.Timestamp,
 			LatchSpans: latchSpans,
 			LockSpans:  lockSpans,
-			BaFmt:      ba,
+			Batch:      ba,
 		}
 		if txnMeta != nil {
 			request.Txn = &roachpb.Transaction{
@@ -1809,7 +1817,7 @@ func doBenchWork(item *benchWorkItem, env benchEnv, doneCh chan<- error) {
 	firstIter := true
 	ctx := context.Background()
 	for {
-		if lg, err = env.lm.Acquire(context.Background(), item.LatchSpans, poison.Policy_Error, item.BaFmt); err != nil {
+		if lg, err = env.lm.Acquire(context.Background(), item.LatchSpans, poison.Policy_Error, item.Batch); err != nil {
 			doneCh <- err
 			return
 		}
@@ -1856,7 +1864,7 @@ func doBenchWork(item *benchWorkItem, env benchEnv, doneCh chan<- error) {
 		return
 	}
 	// Release locks.
-	if lg, err = env.lm.Acquire(context.Background(), item.LatchSpans, poison.Policy_Error, item.BaFmt); err != nil {
+	if lg, err = env.lm.Acquire(context.Background(), item.LatchSpans, poison.Policy_Error, item.Batch); err != nil {
 		doneCh <- err
 		return
 	}
@@ -1891,7 +1899,7 @@ func createRequests(index int, numOutstanding int, numKeys int, numReadKeys int)
 			Timestamp:  ts,
 			LatchSpans: latchSpans,
 			LockSpans:  lockSpans,
-			BaFmt:      ba,
+			Batch:      ba,
 		},
 	}
 	for i := 0; i < numKeys; i++ {
@@ -1986,15 +1994,15 @@ func BenchmarkLockTable(b *testing.B) {
 						var numRequestsWaited uint64
 						var numScanCalls uint64
 						const maxLocks = 100000
-						lt := newLockTable(
-							maxLocks,
-							roachpb.RangeID(3),
-							hlc.NewClockForTesting(nil),
-							cluster.MakeTestingClusterSettings(),
+						clock := hlc.NewClockForTesting(nil)
+						settings := cluster.MakeTestingClusterSettings()
+						lm := spanlatch.Make(
+							nil /* stopper */, nil /* slowReqs */, settings, nil /* latchWaitDurations */, clock,
 						)
+						lt := newLockTable(maxLocks, roachpb.RangeID(3), clock, settings)
 						lt.enabled = true
 						env := benchEnv{
-							lm:                &spanlatch.Manager{},
+							lm:                &lm,
 							lt:                lt,
 							numRequestsWaited: &numRequestsWaited,
 							numScanCalls:      &numScanCalls,

pkg/kv/kvserver/concurrency/lock_table_waiter_test.go

@@ -147,7 +147,7 @@ func TestLockTableWaiterWithTxn(t *testing.T) {
 		return Request{
 			Txn:       &txn,
 			Timestamp: ba.Timestamp,
-			BaFmt:     ba,
+			Batch:     ba,
 		}
 	}
 
@@ -230,7 +230,7 @@ func TestLockTableWaiterWithNonTxn(t *testing.T) {
 		return Request{
 			Timestamp:      ba.Timestamp,
 			NonTxnPriority: ba.UserPriority,
-			BaFmt:          ba,
+			Batch:          ba,
 		}
 	}
 
@@ -431,7 +431,7 @@ func TestLockTableWaiterWithErrorWaitPolicy(t *testing.T) {
 			Txn:        &txn,
 			Timestamp:  ba.Timestamp,
 			WaitPolicy: ba.WaitPolicy,
-			BaFmt:      ba,
+			Batch:      ba,
 		}
 	}
 	makeHighPriReq := func() Request {
@@ -599,7 +599,7 @@ func TestLockTableWaiterWithLockTimeout(t *testing.T) {
 				Txn:         ba.Txn,
 				Timestamp:   ba.Timestamp,
 				LockTimeout: ba.LockTimeout,
-				BaFmt:       ba,
+				Batch:       ba,
 			}
 		}
 		if !txn {
@@ -610,7 +610,7 @@ func TestLockTableWaiterWithLockTimeout(t *testing.T) {
 				return Request{
 					Timestamp:   ba.Timestamp,
 					LockTimeout: ba.LockTimeout,
-					BaFmt:       ba,
+					Batch:       ba,
 				}
 			}
 		}
@@ -787,7 +787,7 @@ func TestLockTableWaiterIntentResolverError(t *testing.T) {
 	req := Request{
 		Txn:       ba.Txn,
 		Timestamp: ba.Timestamp,
-		BaFmt:     ba,
+		Batch:     ba,
 	}
 
 	// Test with both synchronous and asynchronous pushes.
@@ -847,7 +847,7 @@ func TestLockTableWaiterDeferredIntentResolverError(t *testing.T) {
 	req := Request{
 		Txn:       ba.Txn,
 		Timestamp: ba.Timestamp,
-		BaFmt:     ba,
+		Batch:     ba,
 	}
 	keyA := roachpb.Key("keyA")
 	pusheeTxn := makeTxnProto("pushee")

pkg/kv/kvserver/concurrency/testdata/concurrency_manager/basic

@@ -151,9 +151,13 @@ debug-latch-manager
 write count: 1
  read count: 1
 
+debug-advance-clock ts=123
+----
+
 finish req=req3
 ----
 [-] finish req3: finishing request
+[3] sequence req4: latch conflict with 00000003-0000-0000-0000-000000000000 on ‹"k"› for 123.000s
 [3] sequence req4: scanning lock table for conflicting locks
 [3] sequence req4: sequencing complete, returned guard
 
@@ -226,13 +230,13 @@ debug-advance-clock ts=123
 on-txn-updated txn=txn2 status=pending ts=18,1
 ----
 [-] update txn: increasing timestamp of txn2
-[2] sequence req5: resolving intent ‹"k"› for txn 00000002 with PENDING status and clock observation {1 246.000000000,0}
+[2] sequence req5: resolving intent ‹"k"› for txn 00000002 with PENDING status and clock observation {1 369.000000000,0}
 [2] sequence req5: lock wait-queue event: done waiting
 [2] sequence req5: conflicted with 00000002-0000-0000-0000-000000000000 on ‹"k"› for 123.000s
 [2] sequence req5: acquiring latches
 [2] sequence req5: scanning lock table for conflicting locks
 [2] sequence req5: sequencing complete, returned guard
-[3] sequence req6: resolving intent ‹"k"› for txn 00000002 with PENDING status and clock observation {1 246.000000000,1}
+[3] sequence req6: resolving intent ‹"k"› for txn 00000002 with PENDING status and clock observation {1 369.000000000,1}
 [3] sequence req6: lock wait-queue event: done waiting
 [3] sequence req6: conflicted with 00000002-0000-0000-0000-000000000000 on ‹"k"› for 123.000s
 [3] sequence req6: acquiring latches

pkg/kv/kvserver/concurrency/testdata/concurrency_manager/optimistic

@@ -175,10 +175,14 @@ sequence req=req6 eval-kind=pess-after-opt
 [8] sequence req6: waiting to acquire read latch ‹{a-e}›@12.000000000,1 for request Scan [‹"a"›,‹"e"›), [txn: 00000002], held by write latch ‹d›@10.000000000,1 for request Put [‹"d"›], [txn: 00000003]
 [8] sequence req6: blocked on select in spanlatch.(*Manager).waitForSignal
 
+debug-advance-clock ts=123
+----
+
 # req4 finishing releases the latch and allows req6 to proceed.
 finish req=req4
 ----
 [-] finish req4: finishing request
+[8] sequence req6: latch conflict with 00000003-0000-0000-0000-000000000000 on ‹"d"› for 123.000s
 [8] sequence req6: scanning lock table for conflicting locks
 [8] sequence req6: sequencing complete, returned guard