mvcc/watchable_store.go
func (s *watchableStore) cancelWatcher(wa *watcher) {
	for {
		s.mu.Lock()
		if s.unsynced.delete(wa) {
			slowWatcherGauge.Dec()
			break
		} else if s.synced.delete(wa) {
			break
		} else if wa.compacted {
			break
		} else if wa.ch == nil {
			// already canceled (e.g., cancel/close race)
			break
		}

		if !wa.victim {
			panic("watcher not victim but not in watch groups")
		}

		var victimBatch watcherBatch
		for _, wb := range s.victims {
			if wb[wa] != nil {
				victimBatch = wb
				break
			}
		}
		if victimBatch != nil {
			slowWatcherGauge.Dec()
			delete(victimBatch, wa)
			break
		}

		// victim being processed so not accessible; retry
		s.mu.Unlock()
		time.Sleep(time.Millisecond)
	}

	watcherGauge.Dec()
	wa.ch = nil
	s.mu.Unlock()
}
func (s *watchableStore) cancelWatcher(wa *watcher) {
	for {
		s.mu.Lock()
		if s.unsynced.delete(wa) {
			slowWatcherGauge.Dec()
			break
		} else if s.synced.delete(wa) {
			break
		} else if wa.compacted {
			break
		} else if wa.ch == nil {
			// already canceled (e.g., cancel/close race)
			break
		}

		if !wa.victim {
			panic("watcher not victim but not in watch groups")
		}

		var victimBatch watcherBatch
		for _, wb := range s.victims {
			if wb[wa] != nil {
				victimBatch = wb
				break
			}
		}
		if victimBatch != nil {
			slowWatcherGauge.Dec()
			delete(victimBatch, wa)
			break
		}

		// victim being processed so not accessible; retry
		s.mu.Unlock()
		time.Sleep(time.Millisecond)
	}

	watcherGauge.Dec()
	wa.ch = nil
	s.mu.Unlock()
}
func ( s * watchableStore ) cancelWatcher ( wa * watcher ) { for { s . mu . Lock ( ) if s . unsynced . delete ( wa ) { slowWatcherGauge . Dec ( ) break } else if s . synced . delete ( wa ) { break } else if wa . compacted { break } else if wa . ch == nil { // already canceled (e.g., cancel/close race) break } if ! wa . victim { panic ( " " ) } var victimBatch watcherBatch for _ , wb := range s . victims { if wb [ wa ] != nil { victimBatch = wb break } } if victimBatch != nil { slowWatcherGauge . Dec ( ) delete ( victimBatch , wa ) break } // victim being processed so not accessible; retry s . mu . Unlock ( ) time . Sleep ( time . Millisecond ) } watcherGauge . Dec ( ) wa . ch = nil s . mu . Unlock ( ) }
// cancelWatcher removes references of the watcher from the watchableStore
cancelWatcher removes references of the watcher from the watchableStore