var chanBufLen … var maxWatchersPerSync … type watchable … type watchableStore … type cancelFunc … func New(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) WatchableKV { … } func newWatchableStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, cfg StoreConfig) *watchableStore { … } func (s *watchableStore) Close() error { … } func (s *watchableStore) NewWatchStream() WatchStream { … } func (s *watchableStore) watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse, fcs ...FilterFunc) (*watcher, cancelFunc) { … } // cancelWatcher removes references of the watcher from the watchableStore func (s *watchableStore) cancelWatcher(wa *watcher) { … } func (s *watchableStore) Restore(b backend.Backend) error { … } // syncWatchersLoop syncs the watcher in the unsynced map every 100ms. func (s *watchableStore) syncWatchersLoop() { … } // syncVictimsLoop tries to write precomputed watcher responses to // watchers that had a blocked watcher channel func (s *watchableStore) syncVictimsLoop() { … } // moveVictims tries to update watches with already pending event data func (s *watchableStore) moveVictims() (moved int) { … } // syncWatchers syncs unsynced watchers by: // 1. choose a set of watchers from the unsynced watcher group // 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() int { … } // kvsToEvents gets all events for the watchers from all key-value pairs func kvsToEvents(lg *zap.Logger, wg *watcherGroup, revs, vals [][]byte) (evs []mvccpb.Event) { … } // notify notifies the fact that given event at the given rev just happened to // watchers that watch on the key of the event. func (s *watchableStore) notify(rev int64, evs []mvccpb.Event) { … } func (s *watchableStore) addVictim(victim watcherBatch) { … } func (s *watchableStore) rev() int64 { … } func (s *watchableStore) progress(w *watcher) { … } func (s *watchableStore) progressAll(watchers map[WatchID]*watcher) bool { … } func (s *watchableStore) progressIfSync(watchers map[WatchID]*watcher, responseWatchID WatchID) bool { … } type watcher … func (w *watcher) send(wr WatchResponse) bool { … }
Codebase Browser
kubernetes