Skip to content

mvcc: refactor watch store by consolidating watcherGroup.choose/chooseAll #21108

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zhijun42 wants to merge 2 commits into from
Closed

mvcc: refactor watch store by consolidating watcherGroup.choose/chooseAll #21108

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
30d43ed
Consolidate choose and chooseAll
zhijun42 Jan 12, 2026
95db1a0
Address feedback
zhijun42 Jan 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 43 additions & 1 deletion server/storage/mvcc/watchable_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
package mvcc

import (
"fmt"
"math"
"sync"
"time"

Expand Down Expand Up @@ -360,7 +362,7 @@ func (s *watchableStore) syncWatchers(evs []mvccpb.Event) (int, []mvccpb.Event)
curRev := s.store.currentRev
compactionRev := s.store.compactMainRev

wg, minRev := s.unsynced.choose(maxWatchersPerSync, curRev, compactionRev)
wg, minRev := s.unsafeSelectForSync(maxWatchersPerSync, curRev, compactionRev)
evs = rangeEventsWithReuse(s.store.lg, s.store.b, evs, minRev, curRev+1)

victims := make(watcherBatch)
Expand Down Expand Up @@ -413,6 +415,46 @@ func (s *watchableStore) syncWatchers(evs []mvccpb.Event) (int, []mvccpb.Event)
return s.unsynced.size(), evs
}

// unsafeSelectForSync selects up to maxWatchers from the unsynced group for syncing.
// It handles compacted watchers by sending compaction responses and removing them.
// Returns the selected watchers and the minimum revision needed for DB fetch.
// Unsafe because it assumes s.mu is held by the caller.
func (s *watchableStore) unsafeSelectForSync(maxWatchers int, curRev, compactRev int64) (*watcherGroup, int64) {
minRev := int64(math.MaxInt64)
ret := newWatcherGroup()
if s.unsynced.size() == 0 {
return &ret, minRev
}
for w := range s.unsynced.watchers {

@fuweid fuweid Jan 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Behavior changes slightly when maxWatchersPerSync < s.unsynced.size(): because s.unsynced is a map, iteration order is random, and the new loop can traverse many entries while still filling the batch. This means more compacted watchers may be handled (and compact responses sent) earlier than before, but in a nondeterministic order and if w.ch is not too busy.

@serathius serathius Jan 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean that new implementation will return maxWatchers, no matter number of compaction attempts it did, while the previous implementation would count compactions into maxWatchers. Have I understood it correctly?

This means that now resync work mount is not bounded by maxWatchers, but by maxWatchers + numCompacted. So we could be slowing down resync.

Good catch, this is a good suggestion as the presented PR was meant as refactor.

This could be imply changed by changing ret.size() >= maxWatchers to watchersConsidered >= maxWatchers.

@ahrtr ahrtr Jan 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good catch.

@fuweid fuweid Jan 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean that new implementation will return maxWatchers, no matter number of compaction attempts it did, while the previous implementation would count compactions into maxWatchers. Have I understood it correctly?

Yes. map iteration order is nodeterministic. So it may go through all the watchers for that.

if w.minRev > curRev {
// Watchers moved from synced to unsynced during Restore() may have future revisions.
// They will be included but won't match any events, then moved back to synced.
if !w.restore {
panic(fmt.Errorf("watcher minimum revision %d should not exceed current revision %d", w.minRev, curRev))
}
w.restore = false
}
if w.minRev < compactRev {
select {
case w.ch <- WatchResponse{WatchID: w.id, CompactRevision: compactRev}:
w.compacted = true
s.unsynced.delete(w)
default:
// retry next time
}
continue
}
if minRev > w.minRev {
minRev = w.minRev
}
ret.add(w)
if ret.size() >= maxWatchers {
break
}
Comment on lines +450 to +453

@ahrtr ahrtr Jan 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This isn't consistent with the current behaviour. You may return an incorrect minRev if you break the loop due to reaching the maxWatchers.

@serathius serathius Jan 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think returning returning minRev for the watchers is consistent with previous behavior.

@fuweid fuweid Jan 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

minRev part is correct. However, it may scan more than before.
We can go like this

for w := range s.unsynced.watchers {
     if ret.size() >= maxWatchers {
			break
		}
     ret.add(w)
    ...
}

@serathius serathius Jan 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, not sure. I have another suggestion, there is no sense trying to compile code in our heads, maybe we should start from ensuring the functions are properly tested before changing anything :P

@ahrtr ahrtr Jan 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

re-read the function, and confirmed that the minRev is correct.

@ahrtr ahrtr Jan 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ensuring the functions are properly tested before changing anything

agreed on this.

It'd better to add an unit test separately to cover different scenarios, and ensure all cases pass on main. Afterwards, rebase this PR, ensure this PR doesn't break any case.

}
return &ret, minRev
}

// 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 {
Expand Down
50 changes: 0 additions & 50 deletions server/storage/mvcc/watcher_group.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,6 @@
package mvcc

import (
"fmt"
"math"

"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/pkg/v3/adt"
)
Expand Down Expand Up @@ -218,53 +215,6 @@ func (wg *watcherGroup) delete(wa *watcher) bool {
return true
}

// choose selects watchers from the watcher group to update
func (wg *watcherGroup) choose(maxWatchers int, curRev, compactRev int64) (*watcherGroup, int64) {
if len(wg.watchers) < maxWatchers {
return wg, wg.chooseAll(curRev, compactRev)
}
ret := newWatcherGroup()
for w := range wg.watchers {
if maxWatchers <= 0 {
break
}
maxWatchers--
ret.add(w)
}
return &ret, ret.chooseAll(curRev, compactRev)
}

func (wg *watcherGroup) chooseAll(curRev, compactRev int64) int64 {
minRev := int64(math.MaxInt64)
for w := range wg.watchers {
if w.minRev > curRev {
// after network partition, possibly choosing future revision watcher from restore operation
// with watch key "proxy-namespace__lostleader" and revision "math.MaxInt64 - 2"
// do not panic when such watcher had been moved from "synced" watcher during restore operation
if !w.restore {
panic(fmt.Errorf("watcher minimum revision %d should not exceed current revision %d", w.minRev, curRev))
}

// mark 'restore' done, since it's chosen
w.restore = false
}
if w.minRev < compactRev {
select {
case w.ch <- WatchResponse{WatchID: w.id, CompactRevision: compactRev}:
w.compacted = true
wg.delete(w)
default:
// retry next time
}
continue
}
if minRev > w.minRev {
minRev = w.minRev
}
}
return minRev
}

// watcherSetByKey gets the set of watchers that receive events on the given key.
func (wg *watcherGroup) watcherSetByKey(key string) watcherSet {
wkeys := wg.keyWatchers[key]
Expand Down