From 0ff063b970cce61e00fdebf30c3fa1b13c556716 Mon Sep 17 00:00:00 2001 From: Chai Bot Date: Thu, 30 Jul 2026 23:13:10 +0000 Subject: [PATCH] UPSTREAM: 21443: Revert "Reuse events between sync loops" --- server/storage/mvcc/watchable_store.go | 61 +++++++-------------- server/storage/mvcc/watchable_store_test.go | 16 +++--- server/storage/mvcc/watcher_test.go | 4 +- 3 files changed, 31 insertions(+), 50 deletions(-) diff --git a/server/storage/mvcc/watchable_store.go b/server/storage/mvcc/watchable_store.go index 67b2d7f2d38a..1a1d99fed496 100644 --- a/server/storage/mvcc/watchable_store.go +++ b/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,7 +428,7 @@ 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. @@ -465,14 +436,22 @@ func rangeEvents(lg *zap.Logger, b backend.Backend, minRev, maxRev int64) []mvcc 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 diff --git a/server/storage/mvcc/watchable_store_test.go b/server/storage/mvcc/watchable_store_test.go index d7b69adb9976..780d91ce5a66 100644 --- a/server/storage/mvcc/watchable_store_test.go +++ b/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() diff --git a/server/storage/mvcc/watcher_test.go b/server/storage/mvcc/watcher_test.go index 9a9f4f13a75d..d127652e0e16 100644 --- a/server/storage/mvcc/watcher_test.go +++ b/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}