OCPBUGS-100390: UPSTREAM: 21443: Revert "Reuse events between sync loops"

server/storage/mvcc/watchable_store.go

@@ -226,7 +226,6 @@ func (s *watchableStore) syncWatchersLoop() {
 
 	delayTicker := time.NewTicker(watchResyncPeriod)
 	defer delayTicker.Stop()
-	var evs []mvccpb.Event
 
 	for {
 		s.mu.RLock()
@@ -236,7 +235,7 @@ func (s *watchableStore) syncWatchersLoop() {
 
 		unsyncedWatchers := 0
 		if lastUnsyncedWatchers > 0 {
-			unsyncedWatchers, evs = s.syncWatchers(evs)
+			unsyncedWatchers = s.syncWatchers()
 		}
 		syncDuration := time.Since(st)
 
@@ -344,12 +343,12 @@ func (s *watchableStore) moveVictims() (moved int) {
 //  2. iterate over the set to get the minimum revision and remove compacted watchers
 //  3. use minimum revision to get all key-value pairs and send those events to watchers
 //  4. remove synced watchers in set from unsynced group and move to synced group
-func (s *watchableStore) syncWatchers(evs []mvccpb.Event) (int, []mvccpb.Event) {
+func (s *watchableStore) syncWatchers() int {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	if s.unsynced.size() == 0 {
-		return 0, []mvccpb.Event{}
+		return 0
 	}
 
 	s.store.revMu.RLock()
@@ -362,7 +361,7 @@ func (s *watchableStore) syncWatchers(evs []mvccpb.Event) (int, []mvccpb.Event)
 	compactionRev := s.store.compactMainRev
 
 	wg, minRev := s.unsynced.choose(maxWatchersPerSync, curRev, compactionRev)
-	evs = rangeEventsWithReuse(s.store.lg, s.store.b, evs, minRev, curRev+1)
+	evs := rangeEvents(s.store.lg, s.store.b, minRev, curRev+1, wg)
 
 	victims := make(watcherBatch)
 	wb := newWatcherBatch(wg, evs)
@@ -411,43 +410,15 @@ func (s *watchableStore) syncWatchers(evs []mvccpb.Event) (int, []mvccpb.Event)
 	}
 	slowWatcherGauge.Set(float64(s.unsynced.size() + vsz))
 
-	return s.unsynced.size(), evs
-}
-
-// rangeEventsWithReuse returns events in range [minRev, maxRev), while reusing already provided events.
-func rangeEventsWithReuse(lg *zap.Logger, b backend.Backend, evs []mvccpb.Event, minRev, maxRev int64) []mvccpb.Event {
-	if len(evs) == 0 {
-		return rangeEvents(lg, b, minRev, maxRev)
-	}
-	// append from left
-	if evs[0].Kv.ModRevision > minRev {
-		evs = append(rangeEvents(lg, b, minRev, evs[0].Kv.ModRevision), evs...)
-	}
-	// cut from left
-	prefixIndex := 0
-	for prefixIndex < len(evs) && evs[prefixIndex].Kv.ModRevision < minRev {
-		prefixIndex++
-	}
-	evs = evs[prefixIndex:]
-
-	if len(evs) == 0 {
-		return rangeEvents(lg, b, minRev, maxRev)
-	}
-	// append from right
-	if evs[len(evs)-1].Kv.ModRevision+1 < maxRev {
-		evs = append(evs, rangeEvents(lg, b, evs[len(evs)-1].Kv.ModRevision+1, maxRev)...)
-	}
-	// cut from right
-	suffixIndex := len(evs) - 1
-	for suffixIndex >= 0 && evs[suffixIndex].Kv.ModRevision >= maxRev {
-		suffixIndex--
-	}
-	evs = evs[:suffixIndex+1]
-	return evs
+	return s.unsynced.size()
 }
 
 // rangeEvents returns events in range [minRev, maxRev).
-func rangeEvents(lg *zap.Logger, b backend.Backend, minRev, maxRev int64) []mvccpb.Event {
+func rangeEvents(lg *zap.Logger, b backend.Backend, minRev, maxRev int64, c contains) []mvccpb.Event {
+	if minRev < 0 {
+		lg.Warn("Unexpected negative revision range start", zap.Int64("minRev", minRev))
+		minRev = 0
+	}
 	minBytes, maxBytes := NewRevBytes(), NewRevBytes()
 	minBytes = RevToBytes(Revision{Main: minRev}, minBytes)
 	maxBytes = RevToBytes(Revision{Main: maxRev}, maxBytes)
@@ -457,22 +428,30 @@ func rangeEvents(lg *zap.Logger, b backend.Backend, minRev, maxRev int64) []mvcc
 	tx := b.ReadTx()
 	tx.RLock()
 	revs, vs := tx.UnsafeRange(schema.Key, minBytes, maxBytes, 0)
-	evs := kvsToEvents(lg, revs, vs)
+	evs := kvsToEvents(lg, c, revs, vs)
 	// Must unlock after kvsToEvents, because vs (come from boltdb memory) is not deep copy.
 	// We can only unlock after Unmarshal, which will do deep copy.
 	// Otherwise we will trigger SIGSEGV during boltdb re-mmap.
 	tx.RUnlock()
 	return evs
 }
 
+type contains interface {
+	contains(string) bool
+}
+
 // kvsToEvents gets all events for the watchers from all key-value pairs
-func kvsToEvents(lg *zap.Logger, revs, vals [][]byte) (evs []mvccpb.Event) {
+func kvsToEvents(lg *zap.Logger, c contains, revs, vals [][]byte) (evs []mvccpb.Event) {
 	for i, v := range vals {
 		var kv mvccpb.KeyValue
 		if err := kv.Unmarshal(v); err != nil {
 			lg.Panic("failed to unmarshal mvccpb.KeyValue", zap.Error(err))
 		}
 
+		if !c.contains(string(kv.Key)) {
+			continue
+		}
+
 		ty := mvccpb.PUT
 		if isTombstone(revs[i]) {
 			ty = mvccpb.DELETE

server/storage/mvcc/watchable_store_test.go

@@ -313,7 +313,7 @@ func TestSyncWatchers(t *testing.T) {
 
 	assert.Empty(t, s.synced.watcherSetByKey(string(testKey)))
 	assert.Len(t, s.unsynced.watcherSetByKey(string(testKey)), watcherN)
-	s.syncWatchers([]mvccpb.Event{})
+	s.syncWatchers()
 	assert.Len(t, s.synced.watcherSetByKey(string(testKey)), watcherN)
 	assert.Empty(t, s.unsynced.watcherSetByKey(string(testKey)))
 
@@ -429,17 +429,19 @@ func TestRangeEvents(t *testing.T) {
 		{minRev: 5, maxRev: 6, expectEvents: expectEvents[3:5]},
 		{minRev: 6, maxRev: 6, expectEvents: expectEvents[5:5]},
 	}
-	// reuse the evs to test rangeEventsWithReuse
-	var evs []mvccpb.Event
 	for i, tc := range tcs {
 		t.Run(fmt.Sprintf("%d rangeEvents(%d, %d)", i, tc.minRev, tc.maxRev), func(t *testing.T) {
-			assert.ElementsMatch(t, tc.expectEvents, rangeEvents(lg, b, tc.minRev, tc.maxRev))
-			evs = rangeEventsWithReuse(lg, b, evs, tc.minRev, tc.maxRev)
-			assert.ElementsMatch(t, tc.expectEvents, evs)
+			assert.ElementsMatch(t, tc.expectEvents, rangeEvents(lg, b, tc.minRev, tc.maxRev, fakeContains{}))
 		})
 	}
 }
 
+type fakeContains struct{}
+
+func (f fakeContains) contains(string) bool {
+	return true
+}
+
 // TestWatchCompacted tests a watcher that watches on a compacted revision.
 func TestWatchCompacted(t *testing.T) {
 	b, _ := betesting.NewDefaultTmpBackend(t)
@@ -514,7 +516,7 @@ func TestWatchNoEventLossOnCompact(t *testing.T) {
 	// fill up w.Chan() with 1 buf via 2 compacted watch response
 	sImpl, ok := s.(*watchableStore)
 	require.Truef(t, ok, "TestWatchNoEventLossOnCompact: needs a WatchableKV implementation")
-	sImpl.syncWatchers([]mvccpb.Event{})
+	sImpl.syncWatchers()
 
 	for len(watchers) > 0 {
 		resp := <-w.Chan()

server/storage/mvcc/watcher_test.go

@@ -353,7 +353,7 @@ func TestWatcherRequestProgress(t *testing.T) {
 			id, _ := w.Watch(0, notTestKey, nil, tc.startRev)
 			w.RequestProgress(id)
 			asssertProgressSent(t, w, id, tc.expectProgressBeforeSync)
-			s.syncWatchers([]mvccpb.Event{})
+			s.syncWatchers()
 			w.RequestProgress(id)
 			asssertProgressSent(t, w, id, tc.expectProgressAfterSync)
 		})
@@ -403,7 +403,7 @@ func TestWatcherRequestProgressAll(t *testing.T) {
 	default:
 	}
 
-	s.syncWatchers([]mvccpb.Event{})
+	s.syncWatchers()
 
 	w.RequestProgressAll()
 	wrs := WatchResponse{WatchID: clientv3.InvalidWatchID, Revision: 2}