Given the frustration that I tend to have whenever I look at this part of code, I thought I’d take a few notes as I reminded myself how things worked.
Entry point
There are two state methods that create all watchers:
Watch and WatchAllModels (state.go)
Each of these return a *Multiwatcher (multiwatcher.go).
Multiwatcher
A multiwatcher is created with a *storeManger. The store managers are worksers running in the State instance. This is not good, and we should move them, but history.
The state workers object has two methods:
allManager (for all events across a single model )
allModelManager (for all events across all the models)
The Multiwatcher isn’t a worker, but instead still has the old Stop method.
What’s more, the Stop method doesn’t ensure that the watcher stops, only that it has processed the stop request.
It looks like the store manager reaches into the multiwatcher structure to store state, in particular the revno and stopped. (Note the Stop method doesn’t set stopped.) The stopped value is set in the storeManager handle method, as the storeManager holds a reference to the multiwatcher in an internal map.
The Next call grabs all changes since the last Next call. If there are no changes since the last call, the Next call blocks until there is a change or the watcher is stopped.
storeManager
Both of the store managers are created on demand for the first time they are requested for an all watcher.
If the request for the worker from the state workers is an error that isn’t a NotFound, a dead store manager is used.
The store managers are created with a Backing. The backing differs depending on whether you are watching a single model or all the models.
The backing implementations are found in allwatcher.go.
type Backing interface {
// GetAll retrieves information about all information
// known to the Backing and stashes it in the Store.
GetAll(all *multiwatcherStore) error
// Changed informs the backing about a change received
// from a watcher channel. The backing is responsible for
// updating the Store to reflect the change.
Changed(all *multiwatcherStore, change watcher.Change) error
// Watch watches for any changes and sends them
// on the given channel.
Watch(in chan<- watcher.Change)
// Unwatch stops watching for changes on the
// given channel.
Unwatch(in chan<- watcher.Change)
// Release cleans up resources opened by the Backing.
Release() error
// NOTE: this is only called in tests.
}
The storeManager type is a Worker. It has a loop function, and primarly calls Watch on the Backing, and defers the Unwatch.
A storeManager has a multiwatcherStore, whose job it is to remember everything that the Backing is interested in.
Future work
We want to get the allWatchers out of the state objects. Key here is to move them to a worker that depends on the state worker. The API server then also depends on this worker so it can provide the multiwatchers to the clients. The model cache worker should also depend on this new worker rather than state itself.
We should be able to consolidate the allWatcher and allModelWatcher into one worker and have the allWatcher just be a filter of the allModelWatcher.
The basis of the store manager should be used for the worker main loop.
