Implementation and analysis of watchablestore in etcd

watchableStore Storage

In the previous class, we have introduced kvstore, Here we introduce watchableStore The implementation of the .Watch The implementation of store A layer called watchableStore, Rewrote store Of Write Method .

//  be located  mvcc/watchable_store_txn.go:22func (tw *watchableStoreTxnWrite) End() {	changes := tw.Changes()	if len(changes) == 0 {		tw.TxnWrite.End()		return	}	rev := tw.Rev() + 1	evs := make([]mvccpb.Event, len(changes))	for i, change := range changes {		evs[i].Kv = &changes[i]		if change.CreateRevision == 0 {			evs[i].Type = mvccpb.DELETE			evs[i].Kv.ModRevision = rev		} else {			evs[i].Type = mvccpb.PUT		}	}	// end write txn under watchable store lock so the updates are visible	// when asynchronous event posting checks the current store revision	tw.s.mu.Lock()	tw.s.notify(rev, evs)	tw.TxnWrite.End()	tw.s.mu.Unlock()}type watchableStoreTxnWrite struct {	TxnWrite	s *watchableStore}func (s *watchableStore) Write(trace *traceutil.Trace) TxnWrite {	return &watchableStoreTxnWrite{s.store.Write(trace), s}}

adopt MVCC Described in the ,store Any write operation , Need to be Write Method TxnWrite. So rewrite here Write Method means that any write operation will go through watchableStore. It's not hard to see from the code above ,watchableStoreTxnWrite At transaction commit , First change this time changes Pack it up Event, And then call notify To notify the change . Finally, the transaction is actually committed TxnWrite.End().

Watch Responsible for registration 、 Manage and trigger Watcher The function of . Let's look at the fields of this structure first ：

//  be located  mvcc/watchable_store.go:47type watchableStore struct {	*store	//  Synchronous read-write lock 	mu sync.RWMutex	//  Blocked in  watch channel  Medium  watcherBatch	victims []watcherBatch	victimc chan struct{}	//  Unsynchronized  watchers	unsynced watcherGroup	//  Synchronized  watchers	synced watcherGroup	stopc chan struct{}	wg    sync.WaitGroup}

every last watchableStore In fact, they all come from store Structure fields and methods , besides , There are two watcherGroup Type field ,watcherGroup Manage multiple watcher, Can be based on key Quickly find and monitor the key One or more of watcher. among unsynced Used to store instances that are not synchronized ,synced Used to store instances that have been synchronized .

according to watchableStore The definition of , We can describe Watch Monitoring process .

watchableStore Received all key After the change of , Will these key hand synced（watchGroup）,synced Can quickly from all key Found listening in key. Will these key Send to the corresponding watcher, these watcher Re pass chan Send out the change information .

synced How to quickly find qualified key Well ？etcd Used in map and adt（ Red and black trees ） To achieve .

Not used alone map Because watch Can listen to a range of key. If you only listen to one key：

watch(start_key: foo, end_key: nil)

Then the corresponding storage is map[key]*watcher. This can be based on key Quickly find the corresponding watcher,etcd That's how it's done . But for a group key Well ？

watch(start_key: foo, end_key: fop)

Here I monitored from foo->fop Between all key, In theory, these key The number of is infinite , So I can't use map. such as ：key=fooac It also belongs to the scope of monitoring .etcd use adt To store this key.

//  be located  mvcc/watcher_group.go:147// watcherGroup  It consists of a series of ranges  watcher  Organized  watcherstype watcherGroup struct {	// keyWatchers has the watchers that watch on a single key	keyWatchers watcherSetByKey	// ranges has the watchers that watch a range; it is sorted by interval	ranges adt.IntervalTree	// watchers is the set of all watchers	watchers watcherSet}

adt The implementation of is not introduced here , Just know adt Can be based on key=fooac Quickly find the scope foo->fop. Find watcher after , call watcher Of send() Method , What will be changed Event Send out .