block,cache: trace and stats for wait duration (for concurrent reads)

cache.Handle.GetWithReadHandle returns a non-zero wait duration on
a cache miss, which resulted in the request waiting for a turn
to do the read, or to wait for the block to be read by another
reader.

Tracing now happens on high wait duration. Wait duration is included
in iterator stats, since this is a block miss. And wait duration is
included in the read duration when the block was read by the caller
after waiting.

Author: sumeerbhola

db_test.go

@@ -30,7 +30,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
-	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/cockroachdb/pebble/wal"
@@ -1493,7 +1492,7 @@ func (t *testTracer) IsTracingEnabled(ctx context.Context) bool {
 }
 
 func TestTracing(t *testing.T) {
-	defer block.DeterministicReadBlockDurationForTesting()()
+	defer base.DeterministicReadDurationForTesting()()
 
 	var tracer testTracer
 	buf := &tracer.buf

internal/base/stopwatch.go

@@ -0,0 +1,43 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package base
+
+import (
+	"time"
+
+	"github.com/cockroachdb/crlib/crtime"
+)
+
+// DeterministicReadDurationForTesting is for tests that want a deterministic
+// value of the time to read (that is not in the cache). The return value is a
+// function that must be called before the test exits.
+func DeterministicReadDurationForTesting() func() {
+	drbdForTesting := deterministicReadDurationForTesting
+	deterministicReadDurationForTesting = true
+	return func() {
+		deterministicReadDurationForTesting = drbdForTesting
+	}
+}
+
+var deterministicReadDurationForTesting = false
+
+type deterministicStopwatchForTesting struct {
+	startTime crtime.Mono
+}
+
+func MakeStopwatch() deterministicStopwatchForTesting {
+	return deterministicStopwatchForTesting{startTime: crtime.NowMono()}
+}
+
+func (w deterministicStopwatchForTesting) Stop() time.Duration {
+	dur := w.startTime.Elapsed()
+	if deterministicReadDurationForTesting {
+		dur = SlowReadTracingThreshold
+	}
+	return dur
+}
+
+// TODO(sumeer): should the threshold be configurable.
+const SlowReadTracingThreshold = 5 * time.Millisecond

internal/cache/cache.go

@@ -280,7 +280,8 @@ func (c *Handle) Get(fileNum base.DiskFileNum, offset uint64) *Value {
 // This method can block before returning since multiple concurrent gets for
 // the same cache value will take turns getting a ReadHandle, which represents
 // permission to do the read. This blocking respects context cancellation, in
-// which case an error is returned (and not a valid ReadHandle).
+// which case an error is returned (and not a valid ReadHandle). The
+// waitDuration returns the duration of the blocking.
 //
 // When blocking, the errorDuration return value can be non-zero and is
 // populated with the total duration that other readers that observed an error
@@ -294,18 +295,27 @@ func (c *Handle) Get(fileNum base.DiskFileNum, offset uint64) *Value {
 // in a valid Handle being returned. This is a case where cacheHit=false.
 func (c *Handle) GetWithReadHandle(
 	ctx context.Context, fileNum base.DiskFileNum, offset uint64,
-) (cv *Value, rh ReadHandle, errorDuration time.Duration, cacheHit bool, err error) {
+) (
+	cv *Value,
+	rh ReadHandle,
+	errorDuration time.Duration,
+	waitDuration time.Duration,
+	cacheHit bool,
+	err error,
+) {
 	k := makeKey(c.id, fileNum, offset)
 	cv, re := c.cache.getShard(k).getWithMaybeReadEntry(k, true /* desireReadEntry */)
 	if cv != nil {
-		return cv, ReadHandle{}, 0, true, nil
+		return cv, ReadHandle{}, 0, 0, true, nil
 	}
+	waitStopwatch := base.MakeStopwatch()
 	cv, errorDuration, err = re.waitForReadPermissionOrHandle(ctx)
+	waitDuration = waitStopwatch.Stop()
 	if err != nil || cv != nil {
 		re.unrefAndTryRemoveFromMap()
-		return cv, ReadHandle{}, errorDuration, false, err
+		return cv, ReadHandle{}, errorDuration, waitDuration, false, err
 	}
-	return nil, ReadHandle{entry: re}, errorDuration, false, nil
+	return nil, ReadHandle{entry: re}, errorDuration, waitDuration, false, nil
 }
 
 // Set sets the cache value for the specified file and offset, overwriting an

internal/cache/read_shard_test.go

@@ -254,7 +254,7 @@ func TestReadShardConcurrent(t *testing.T) {
 	for _, r := range differentReaders {
 		for j := 0; j < r.numReaders; j++ {
 			go func(r *testSyncReaders, index int) {
-				v, rh, _, _, err := r.handle.GetWithReadHandle(context.Background(), r.fileNum, r.offset)
+				v, rh, _, _, _, err := r.handle.GetWithReadHandle(context.Background(), r.fileNum, r.offset)
 				require.NoError(t, err)
 				if v != nil {
 					require.Equal(t, r.val, v.RawBuffer())

metrics_test.go

@@ -195,7 +195,7 @@ func TestMetrics(t *testing.T) {
 	if runtime.GOARCH == "386" {
 		t.Skip("skipped on 32-bit due to slightly varied output")
 	}
-	defer block.DeterministicReadBlockDurationForTesting()()
+	defer base.DeterministicReadDurationForTesting()()
 
 	var d *DB
 	var iters map[string]*Iterator

sstable/block/block.go

@@ -15,7 +15,6 @@ import (
 	"unsafe"
 
 	"github.com/cespare/xxhash/v2"
-	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/crlib/fifo"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -28,6 +27,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 	"github.com/cockroachdb/pebble/sstable/block/blockkind"
+	"github.com/cockroachdb/redact"
 )
 
 // Kind is a convenience alias.
@@ -371,18 +371,23 @@ func (r *Reader) Read(
 				return CacheBufferHandle(cv), nil
 			}
 		}
-		value, err := r.doRead(ctx, env, readHandle, bh, kind, initBlockMetadataFn)
+		value, err := r.doRead(ctx, env, readHandle, bh, kind, 0, initBlockMetadataFn)
 		if err != nil {
 			return BufferHandle{}, env.maybeReportCorruption(err)
 		}
 		return value.MakeHandle(), nil
 	}
 
-	cv, crh, errorDuration, hit, err := r.opts.CacheOpts.CacheHandle.GetWithReadHandle(
+	cv, crh, errorDuration, waitDuration, hit, err := r.opts.CacheOpts.CacheHandle.GetWithReadHandle(
 		ctx, r.opts.CacheOpts.FileNum, bh.Offset)
-	if errorDuration > 5*time.Millisecond && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
+	const slowDur = 5 * time.Millisecond
+	if waitDuration > slowDur && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
 		r.opts.LoggerAndTracer.Eventf(
-			ctx, "waited for turn when %s time wasted by failed reads", errorDuration.String())
+			ctx, "waited for reading turn for %v", waitDuration)
+	}
+	if errorDuration > slowDur && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
+		r.opts.LoggerAndTracer.Eventf(
+			ctx, "failed reads by others wasted %v", errorDuration)
 	}
 	// TODO(sumeer): consider tracing when waited longer than some duration
 	// for turn to do the read.
@@ -400,11 +405,15 @@ func (r *Reader) Read(
 		}
 		if hit {
 			recordCacheHit(ctx, env, readHandle, bh, kind)
+		} else {
+			// The block was not in the cache, and someone else read it, but this
+			// caller had to wait for that read. So account for it in the stats.
+			env.BlockRead(kind, bh.Length, waitDuration)
 		}
 		return CacheBufferHandle(cv), nil
 	}
 
-	value, err := r.doRead(ctx, env, readHandle, bh, kind, initBlockMetadataFn)
+	value, err := r.doRead(ctx, env, readHandle, bh, kind, waitDuration, initBlockMetadataFn)
 	if err != nil {
 		crh.SetReadError(err)
 		return BufferHandle{}, env.maybeReportCorruption(err)
@@ -423,9 +432,6 @@ func recordCacheHit(
 	env.BlockServedFromCache(kind, bh.Length)
 }
 
-// TODO(sumeer): should the threshold be configurable.
-const slowReadTracingThreshold = 5 * time.Millisecond
-
 // doRead is a helper for Read that does the read, checksum check,
 // decompression, and returns either a Value or an error.
 func (r *Reader) doRead(
@@ -434,6 +440,7 @@ func (r *Reader) doRead(
 	readHandle objstorage.ReadHandle,
 	bh Handle,
 	kind Kind,
+	waitBeforeReadDuration time.Duration,
 	initBlockMetadataFn func(*Metadata, []byte) error,
 ) (Value, error) {
 	ctx = objiotracing.WithBlockKind(ctx, kind)
@@ -447,21 +454,28 @@ func (r *Reader) doRead(
 	}
 
 	compressed := Alloc(int(bh.Length+TrailerLen), env.BufferPool)
-	readStopwatch := makeStopwatch()
+	readStopwatch := base.MakeStopwatch()
 	var err error
 	if readHandle != nil {
 		err = readHandle.ReadAt(ctx, compressed.BlockData(), int64(bh.Offset))
 	} else {
 		err = r.readable.ReadAt(ctx, compressed.BlockData(), int64(bh.Offset))
 	}
-	readDuration := readStopwatch.stop()
+	readDuration := readStopwatch.Stop()
 	// Call IsTracingEnabled to avoid the allocations of boxing integers into an
 	// interface{}, unless necessary.
-	if readDuration >= slowReadTracingThreshold && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
+	if (readDuration+waitBeforeReadDuration) >= base.SlowReadTracingThreshold &&
+		r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
 		_, file1, line1, _ := runtime.Caller(1)
 		_, file2, line2, _ := runtime.Caller(2)
-		r.opts.LoggerAndTracer.Eventf(ctx, "reading block of %d bytes took %s (fileNum=%s; %s/%s:%d -> %s/%s:%d)",
-			int(bh.Length+TrailerLen), readDuration.String(),
+		var waitDurStr string
+		if waitBeforeReadDuration > 0 {
+			waitDurStr = fmt.Sprintf("+ %v wait ", waitBeforeReadDuration)
+		}
+		r.opts.LoggerAndTracer.Eventf(
+			ctx, "reading block kind %s of %d bytes took %v %s(fileNum=%s; %s/%s:%d -> %s/%s:%d)",
+			redact.SafeString(kind.String()),
+			int(bh.Length+TrailerLen), readDuration, redact.SafeString(waitDurStr),
 			r.opts.CacheOpts.FileNum,
 			filepath.Base(filepath.Dir(file2)), filepath.Base(file2), line2,
 			filepath.Base(filepath.Dir(file1)), filepath.Base(file1), line1)
@@ -470,7 +484,7 @@ func (r *Reader) doRead(
 		compressed.Release()
 		return Value{}, err
 	}
-	env.BlockRead(kind, bh.Length, readDuration)
+	env.BlockRead(kind, bh.Length, readDuration+waitBeforeReadDuration)
 	if err = ValidateChecksum(r.checksumType, compressed.BlockData(), bh); err != nil {
 		compressed.Release()
 		err = errors.Wrapf(err, "pebble: file %s", r.opts.CacheOpts.FileNum)
@@ -557,51 +571,22 @@ func ReadRaw(
 		return nil, base.CorruptionErrorf("pebble: invalid file %s (file size is too small)", errors.Safe(fileNum))
 	}
 
-	readStopwatch := makeStopwatch()
+	readStopwatch := base.MakeStopwatch()
 	var err error
 	if readHandle != nil {
 		err = readHandle.ReadAt(ctx, buf, off)
 	} else {
 		err = f.ReadAt(ctx, buf, off)
 	}
-	readDuration := readStopwatch.stop()
+	readDuration := readStopwatch.Stop()
 	// Call IsTracingEnabled to avoid the allocations of boxing integers into an
 	// interface{}, unless necessary.
-	if readDuration >= slowReadTracingThreshold && logger.IsTracingEnabled(ctx) {
-		logger.Eventf(ctx, "reading footer of %d bytes took %s",
-			len(buf), readDuration.String())
+	if readDuration >= base.SlowReadTracingThreshold && logger.IsTracingEnabled(ctx) {
+		logger.Eventf(ctx, "reading footer of %d bytes took %v",
+			len(buf), readDuration)
 	}
 	if err != nil {
 		return nil, errors.Wrap(err, "pebble: invalid file (could not read footer)")
 	}
 	return buf, nil
 }
-
-// DeterministicReadBlockDurationForTesting is for tests that want a
-// deterministic value of the time to read a block (that is not in the cache).
-// The return value is a function that must be called before the test exits.
-func DeterministicReadBlockDurationForTesting() func() {
-	drbdForTesting := deterministicReadBlockDurationForTesting
-	deterministicReadBlockDurationForTesting = true
-	return func() {
-		deterministicReadBlockDurationForTesting = drbdForTesting
-	}
-}
-
-var deterministicReadBlockDurationForTesting = false
-
-type deterministicStopwatchForTesting struct {
-	startTime crtime.Mono
-}
-
-func makeStopwatch() deterministicStopwatchForTesting {
-	return deterministicStopwatchForTesting{startTime: crtime.NowMono()}
-}
-
-func (w deterministicStopwatchForTesting) stop() time.Duration {
-	dur := w.startTime.Elapsed()
-	if deterministicReadBlockDurationForTesting {
-		dur = slowReadTracingThreshold
-	}
-	return dur
-}

sstable/block/category_stats.go

@@ -161,7 +161,10 @@ type CategoryStats struct {
 	// cache.
 	BlockBytesInCache uint64
 	// BlockReadDuration is the total duration to read the bytes not in the
-	// cache, i.e., BlockBytes-BlockBytesInCache.
+	// cache, i.e., BlockBytes-BlockBytesInCache. When multiple concurrent
+	// readers wait for each other, and only one does the read, this will
+	// account for the total time spent waiting plus potentially reading for all
+	// those readers, so it can over count. Such over counting should be rare.
 	BlockReadDuration time.Duration
 }