Merge #147913

147913: kvserver: don't subsume self-blocking latency in scheduler latency r=tbg a=tbg

We noticed elevated p99s during scale testing that disappeared when disabling
internal timeseries storage. The theory is that the timeseries replicas are
quite busy, and the scheduler latency was (unintentionally, I argue) picking up
raft **handling** latencies because it wasn't accounting for the case in which
handling a replica was delayed by an inflight handling of that same replica.

This commit addresses this issue, so that the scheduling latency metrics truly
reflect only the scheduling latency.

Fixes #147911
Informs #147758

Release note: None


Co-authored-by: Tobias Grieger <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/scheduler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvflowcontrol/rac2"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util"
+	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -22,6 +23,10 @@ import (
 
 const rangeIDChunkSize = 1000
 
+type testProcessorI interface {
+	processTestEvent(roachpb.RangeID, *raftSchedulerShard, raftScheduleState)
+}
+
 type rangeIDChunk struct {
 	// Valid contents are buf[rd:wr], read at buf[rd], write at buf[wr].
 	buf    [rangeIDChunkSize]roachpb.RangeID
@@ -134,11 +139,14 @@ const (
 	stateRaftTick
 	stateRACv2PiggybackedAdmitted
 	stateRACv2RangeController
+	stateTestIntercept // used for testing, CrdbTestBuild only
 )
 
 type raftScheduleState struct {
 	flags raftScheduleFlags
-	begin crtime.Mono
+	// When this event was queued. This is set if and only if the item is present
+	// in the raft scheduler shard's queue.
+	queued crtime.Mono
 
 	// The number of ticks queued. Usually it's 0 or 1, but may go above if the
 	// scheduling or processing is slow. It is limited by raftScheduler.maxTicks,
@@ -375,8 +383,12 @@ func (ss *raftSchedulerShard) worker(
 		ss.state[id] = raftScheduleState{flags: stateQueued}
 		ss.Unlock()
 
+		if util.RaceEnabled && state.queued == 0 {
+			// See state.queued for the invariant being checked here.
+			log.Fatalf(ctx, "raftSchedulerShard.worker called with zero queued: %+v", state)
+		}
 		// Record the scheduling latency for the range.
-		lat := state.begin.Elapsed()
+		lat := state.queued.Elapsed()
 		metrics.RaftSchedulerLatency.RecordValue(int64(lat))
 
 		// Process requests first. This avoids a scenario where a tick and a
@@ -413,6 +425,9 @@ func (ss *raftSchedulerShard) worker(
 		if state.flags&stateRACv2RangeController != 0 {
 			processor.processRACv2RangeController(ctx, id)
 		}
+		if buildutil.CrdbTestBuild && state.flags&stateTestIntercept != 0 {
+			processor.(testProcessorI).processTestEvent(id, ss, state)
+		}
 
 		ss.Lock()
 		state = ss.state[id]
@@ -442,6 +457,13 @@ func (ss *raftSchedulerShard) worker(
 			//   and the worker does not go back to sleep between the current
 			//   iteration and the next iteration, so no change to num_signals
 			//   is needed.
+			//
+			// NB: we overwrite state.begin unconditionally since the next processing
+			// can not possibly happen before the current processing is done (i.e.
+			// now). We do not want the scheduler latency to pick up the time spent
+			// handling this replica.
+			state.queued = crtime.NowMono()
+			ss.state[id] = state
 			ss.queue.Push(id)
 		}
 	}
@@ -473,11 +495,13 @@ func (ss *raftSchedulerShard) enqueue1Locked(
 	if newState.flags&stateQueued == 0 {
 		newState.flags |= stateQueued
 		queued++
+		if util.RaceEnabled && newState.queued != 0 {
+			// See newState.queued for the invariant being checked here.
+			log.Fatalf(context.Background(), "raftSchedulerShard.enqueue1Locked called with non-zero queued: %+v", newState)
+		}
+		newState.queued = now
 		ss.queue.Push(id)
 	}
-	if newState.begin == 0 {
-		newState.begin = now
-	}
 	ss.state[id] = newState
 	return queued
 }

pkg/kv/kvserver/scheduler_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -135,15 +136,19 @@ type testProcessor struct {
 		rac2RangeController     map[roachpb.RangeID]int
 		ready                   func(roachpb.RangeID)
 	}
+	testEventCh chan func(roachpb.RangeID, *raftSchedulerShard, raftScheduleState)
 }
 
+var _ testProcessorI = (*testProcessor)(nil)
+
 func newTestProcessor() *testProcessor {
 	p := &testProcessor{}
 	p.mu.raftReady = make(map[roachpb.RangeID]int)
 	p.mu.raftRequest = make(map[roachpb.RangeID]int)
 	p.mu.raftTick = make(map[roachpb.RangeID]int)
 	p.mu.rac2PiggybackedAdmitted = make(map[roachpb.RangeID]int)
 	p.mu.rac2RangeController = make(map[roachpb.RangeID]int)
+	p.testEventCh = make(chan func(roachpb.RangeID, *raftSchedulerShard, raftScheduleState), 10)
 	return p
 }
 
@@ -192,6 +197,16 @@ func (p *testProcessor) processRACv2RangeController(_ context.Context, rangeID r
 	p.mu.Unlock()
 }
 
+func (p *testProcessor) processTestEvent(
+	id roachpb.RangeID, ss *raftSchedulerShard, ev raftScheduleState,
+) {
+	select {
+	case fn := <-p.testEventCh:
+		fn(id, ss, ev)
+	default:
+	}
+}
+
 func (p *testProcessor) readyCount(rangeID roachpb.RangeID) int {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -352,6 +367,74 @@ func TestSchedulerBuffering(t *testing.T) {
 	}
 }
 
+func TestSchedulerEnqueueWhileProcessing(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	skip.UnderNonTestBuild(t) // stateTestIntercept needs CrdbTestBuild
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+
+	m := newStoreMetrics(metric.TestSampleInterval)
+	p := newTestProcessor()
+	s := newRaftScheduler(log.MakeTestingAmbientContext(stopper.Tracer()), m, p, 1, 1, 1, 5)
+	s.Start(stopper)
+
+	done := make(chan struct{})
+
+	// Inject code into the "middle" of event processing - after having consumed
+	// from the queue, but before re-checking of overlapping enqueue calls.
+	p.testEventCh <- func(id roachpb.RangeID, ss *raftSchedulerShard, ev raftScheduleState) {
+		// First call into this method.
+		//
+		// The event calling into us must have `ev.queued` set; it was set when
+		// enqueuing.
+		assert.NotZero(t, ev.queued)
+
+		// Even though our event is currently being processed, there is a queued
+		// and otherwise blank event in the scheduler state (which is how we have
+		// concurrent enqueue calls coalesce onto the still pending processing of
+		// the current event).
+		ss.Lock()
+		statePre := ss.state[id]
+		ss.Unlock()
+
+		assert.Zero(t, statePre.queued)
+		assert.Equal(t, stateQueued, statePre.flags)
+
+		// Simulate a concurrent actor that enqueues the same range again.
+		// This will not trigger the interceptor again, since the done channel
+		// is closed by that time.
+		s.enqueue1(stateTestIntercept, 1)
+
+		// Seeing that there is an existing "queued" event, the enqueue call below
+		// should not populate `queued`.  Instead, this will be the job of our
+		// caller when it *actually* pushes into the queue again after fully
+		// having handled `ev`.
+		ss.Lock()
+		statePost := ss.state[id]
+		ss.Unlock()
+
+		assert.Zero(t, statePost.queued)
+		assert.Equal(t, stateQueued|stateTestIntercept, statePost.flags)
+		close(done)
+	}
+	p.testEventCh <- func(id roachpb.RangeID, shard *raftSchedulerShard, ev raftScheduleState) {
+		// Second call into this method, i.e. the overlappingly-enqeued event is
+		// being processed. Check that `queued` is now set.
+		assert.NotZero(t, ev.queued)
+	}
+	s.enqueue1(stateTestIntercept, 1) // will become 'ev' in the intercept
+	select {
+	case <-done:
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+}
+
 func TestNewSchedulerShards(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)