From c26559b1242e0805edda75dbaf4996f3ce3f616f Mon Sep 17 00:00:00 2001 From: Mike Spreitzer Date: Thu, 2 Jan 2020 01:53:17 -0800 Subject: [PATCH] Updated comments on internal abstractions in client-go/tools/cache The comments on Store and Queue and the FIFOs ceased being accurate long ago. Kubernetes-commit: 7a7ccb797e0c977c8049c1490f5a1f92f0bfbbb2 --- tools/cache/delta_fifo.go | 28 ++++++++++------ tools/cache/fifo.go | 56 +++++++++++++++++++++----------- tools/cache/mutation_detector.go | 5 ++- tools/cache/shared_informer.go | 11 +++---- tools/cache/store.go | 39 ++++++++++++++++++---- tools/cache/thread_safe_store.go | 38 +++++++++++----------- 6 files changed, 115 insertions(+), 62 deletions(-) diff --git a/tools/cache/delta_fifo.go b/tools/cache/delta_fifo.go index 55ecdcdf..8246794a 100644 --- a/tools/cache/delta_fifo.go +++ b/tools/cache/delta_fifo.go @@ -26,7 +26,7 @@ import ( "k8s.io/klog" ) -// NewDeltaFIFO returns a Store which can be used process changes to items. +// NewDeltaFIFO returns a Queue which can be used to process changes to items. // // keyFunc is used to figure out what key an object should have. (It's // exposed in the returned DeltaFIFO's KeyOf() method, with bonus features.) @@ -67,7 +67,9 @@ func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO { return f } -// DeltaFIFO is like FIFO, but allows you to process deletes. +// DeltaFIFO is like FIFO, but allows the PopProcessFunc to process +// deletes. The accumulator associated with a given object's key is a +// slice of Delta values for that object. // // DeltaFIFO is a producer-consumer queue, where a Reflector is // intended to be the producer, and the consumer is whatever calls @@ -77,22 +79,25 @@ func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO { // * You want to process every object change (delta) at most once. // * When you process an object, you want to see everything // that's happened to it since you last processed it. -// * You want to process the deletion of objects. +// * You want to process the deletion of some of the objects. // * You might want to periodically reprocess objects. // // DeltaFIFO's Pop(), Get(), and GetByKey() methods return // interface{} to satisfy the Store/Queue interfaces, but it // will always return an object of type Deltas. // +// A DeltaFIFO's knownObjects KeyListerGetter provides get/list access +// to a set of "known objects" that is used for two purposes. One is +// to conditionalize delete operations: it is only for a known object +// that a Delete Delta is recorded (this applies to both Delete and +// Replace). The deleted object will be included in the +// DeleteFinalStateUnknown markers, and those objects could be stale. +// The other purpose is in the Resync operation, which adds a Sync +// Delta for every known object. +// // A note on threading: If you call Pop() in parallel from multiple // threads, you could end up with multiple threads processing slightly // different versions of the same object. -// -// A note on the KeyLister used by the DeltaFIFO: It's main purpose is -// to list keys that are "known", for the purpose of figuring out which -// items have been deleted when Replace() or Delete() are called. The deleted -// object will be included in the DeleteFinalStateUnknown markers. These objects -// could be stale. type DeltaFIFO struct { // lock/cond protects access to 'items' and 'queue'. lock sync.RWMutex @@ -187,7 +192,7 @@ func (f *DeltaFIFO) Update(obj interface{}) error { // Delete is just like Add, but makes an Deleted Delta. If the item does not // already exist, it will be ignored. (It may have already been deleted by a -// Replace (re-list), for example. +// Replace (re-list), for example.) func (f *DeltaFIFO) Delete(obj interface{}) error { id, err := f.KeyOf(obj) if err != nil { @@ -313,6 +318,9 @@ func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) err f.items[id] = newDeltas f.cond.Broadcast() } else { + // This never happens, because dedupDeltas never returns an empty list + // when given a non-empty list (as it is here). + // But if somehow it ever does return an empty list, then // We need to remove this from our map (extra items in the queue are // ignored if they are not in the map). delete(f.items, id) diff --git a/tools/cache/fifo.go b/tools/cache/fifo.go index 7a3bc3d3..0172ae72 100644 --- a/tools/cache/fifo.go +++ b/tools/cache/fifo.go @@ -24,7 +24,7 @@ import ( ) // PopProcessFunc is passed to Pop() method of Queue interface. -// It is supposed to process the element popped from the queue. +// It is supposed to process the accumulator popped from the queue. type PopProcessFunc func(interface{}) error // ErrRequeue may be returned by a PopProcessFunc to safely requeue @@ -44,26 +44,38 @@ func (e ErrRequeue) Error() string { return e.Err.Error() } -// Queue is exactly like a Store, but has a Pop() method too. +// Queue extends Store with a collection of keys to "process". +// Every Add, Update, or Delete may put the object's key in that collection. +// A Queue has a way to derive the corresponding key given an accumulator. +// A Queue can be accessed concurrently from multiple goroutines. +// A Queue can be "closed", after which Pop operations return an error. type Queue interface { Store - // Pop blocks until it has something to process. - // It returns the object that was process and the result of processing. - // The PopProcessFunc may return an ErrRequeue{...} to indicate the item - // should be requeued before releasing the lock on the queue. + // Pop blocks until there is at least one key to process or the + // Queue is closed. In the latter case Pop returns with an error. + // In the former case Pop atomically picks one key to process, + // removes that (key, accumulator) association from the Store, and + // processes the accumulator. Pop returns the accumulator that + // was processed and the result of processing. The PopProcessFunc + // may return an ErrRequeue{inner} and in this case Pop will (a) + // return that (key, accumulator) association to the Queue as part + // of the atomic processing and (b) return the inner error from + // Pop. Pop(PopProcessFunc) (interface{}, error) - // AddIfNotPresent adds a value previously - // returned by Pop back into the queue as long - // as nothing else (presumably more recent) - // has since been added. + // AddIfNotPresent puts the given accumulator into the Queue (in + // association with the accumulator's key) if and only if that key + // is not already associated with a non-empty accumulator. AddIfNotPresent(interface{}) error - // HasSynced returns true if the first batch of items has been popped + // HasSynced returns true if the first batch of keys have all been + // popped. The first batch of keys are those of the first Replace + // operation if that happened before any Add, Update, or Delete; + // otherwise the first batch is empty. HasSynced() bool - // Close queue + // Close the queue Close() } @@ -79,11 +91,16 @@ func Pop(queue Queue) interface{} { return result } -// FIFO receives adds and updates from a Reflector, and puts them in a queue for -// FIFO order processing. If multiple adds/updates of a single item happen while -// an item is in the queue before it has been processed, it will only be -// processed once, and when it is processed, the most recent version will be -// processed. This can't be done with a channel. +// FIFO is a Queue in which (a) each accumulator is simply the most +// recently provided object and (b) the collection of keys to process +// is a FIFO. The accumulators all start out empty, and deleting an +// object from its accumulator empties the accumulator. The Resync +// operation is a no-op. +// +// Thus: if multiple adds/updates of a single object happen while that +// object's key is in the queue before it has been processed then it +// will only be processed once, and when it is processed the most +// recent version will be processed. This can't be done with a channel // // FIFO solves this use case: // * You want to process every object (exactly) once. @@ -94,7 +111,7 @@ func Pop(queue Queue) interface{} { type FIFO struct { lock sync.RWMutex cond sync.Cond - // We depend on the property that items in the set are in the queue and vice versa. + // We depend on the property that every key in `items` is also in `queue` items map[string]interface{} queue []string @@ -326,7 +343,8 @@ func (f *FIFO) Replace(list []interface{}, resourceVersion string) error { return nil } -// Resync will touch all objects to put them into the processing queue +// Resync will ensure that every object in the Store has its key in the queue. +// This should be a no-op, because that property is maintained by all operations. func (f *FIFO) Resync() error { f.lock.Lock() defer f.lock.Unlock() diff --git a/tools/cache/mutation_detector.go b/tools/cache/mutation_detector.go index fa6acab3..885c1ebc 100644 --- a/tools/cache/mutation_detector.go +++ b/tools/cache/mutation_detector.go @@ -36,9 +36,12 @@ func init() { mutationDetectionEnabled, _ = strconv.ParseBool(os.Getenv("KUBE_CACHE_MUTATION_DETECTOR")) } -// MutationDetector is able to monitor if the object be modified outside. +// MutationDetector is able to monitor objects for mutation within a limited window of time type MutationDetector interface { + // AddObject adds the given object to the set being monitored for a while from now AddObject(obj interface{}) + + // Run starts the monitoring and does not return until the monitoring is stopped. Run(stopCh <-chan struct{}) } diff --git a/tools/cache/shared_informer.go b/tools/cache/shared_informer.go index 44ddfdff..8f6b6347 100644 --- a/tools/cache/shared_informer.go +++ b/tools/cache/shared_informer.go @@ -168,21 +168,21 @@ type SharedIndexInformer interface { } // NewSharedInformer creates a new instance for the listwatcher. -func NewSharedInformer(lw ListerWatcher, objType runtime.Object, resyncPeriod time.Duration) SharedInformer { - return NewSharedIndexInformer(lw, objType, resyncPeriod, Indexers{}) +func NewSharedInformer(lw ListerWatcher, objOfWatchedType runtime.Object, resyncPeriod time.Duration) SharedInformer { + return NewSharedIndexInformer(lw, objOfWatchedType, resyncPeriod, Indexers{}) } // NewSharedIndexInformer creates a new instance for the listwatcher. -func NewSharedIndexInformer(lw ListerWatcher, objType runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer { +func NewSharedIndexInformer(lw ListerWatcher, objOfWatchedType runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer { realClock := &clock.RealClock{} sharedIndexInformer := &sharedIndexInformer{ processor: &sharedProcessor{clock: realClock}, indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers), listerWatcher: lw, - objectType: objType, + objectType: objOfWatchedType, resyncCheckPeriod: defaultEventHandlerResyncPeriod, defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod, - cacheMutationDetector: NewCacheMutationDetector(fmt.Sprintf("%T", objType)), + cacheMutationDetector: NewCacheMutationDetector(fmt.Sprintf("%T", objOfWatchedType)), clock: realClock, } return sharedIndexInformer @@ -244,7 +244,6 @@ type sharedIndexInformer struct { processor *sharedProcessor cacheMutationDetector MutationDetector - // This block is tracked to handle late initialization of the controller listerWatcher ListerWatcher objectType runtime.Object diff --git a/tools/cache/store.go b/tools/cache/store.go index fc844efe..eb71a94d 100644 --- a/tools/cache/store.go +++ b/tools/cache/store.go @@ -23,27 +23,52 @@ import ( "k8s.io/apimachinery/pkg/api/meta" ) -// Store is a generic object storage interface. Reflector knows how to watch a server -// and update a store. A generic store is provided, which allows Reflector to be used -// as a local caching system, and an LRU store, which allows Reflector to work like a -// queue of items yet to be processed. +// Store is a generic object storage and processing interface. A +// Store holds a map from string keys to accumulators, and has +// operations to add, update, and delete a given object to/from the +// accumulator currently associated with a given key. A Store also +// knows how to extract the key from a given object, so many operations +// are given only the object. // -// Store makes no assumptions about stored object identity; it is the responsibility -// of a Store implementation to provide a mechanism to correctly key objects and to -// define the contract for obtaining objects by some arbitrary key type. +// In the simplest Store implementations each accumulator is simply +// the last given object and thus the Store's behavior is simple +// storage. +// +// Reflector knows how to watch a server and update a Store. This +// package provides a variety of implementations of Store. type Store interface { + + // Add adds the given object to the accumulator associated with the given object's key Add(obj interface{}) error + + // Update updates the given object in the accumulator associated with the given object's key Update(obj interface{}) error + + // Delete deletes the given object from the accumulator associated with the given object's key Delete(obj interface{}) error + + // List returns a list of all the currently non-empty accumulators List() []interface{} + + // ListKeys returns a list of all the keys currently associated with non-empty accumulators ListKeys() []string + + // Get returns the accumulator associated with the given object's key Get(obj interface{}) (item interface{}, exists bool, err error) + + // GetByKey returns the accumulator associated with the given key GetByKey(key string) (item interface{}, exists bool, err error) // Replace will delete the contents of the store, using instead the // given list. Store takes ownership of the list, you should not reference // it after calling this function. Replace([]interface{}, string) error + + // Resync is meaningless in the terms appearing here but has + // meaning in some implementations that have non-trivial + // additional behavior. In general the idea is to tee up the + // current non-empty accumulators or their keys for + // reconsideration (whatever that means). Resync() error } diff --git a/tools/cache/thread_safe_store.go b/tools/cache/thread_safe_store.go index e7232514..775fe9a1 100644 --- a/tools/cache/thread_safe_store.go +++ b/tools/cache/thread_safe_store.go @@ -131,8 +131,8 @@ func (c *threadSafeMap) Replace(items map[string]interface{}, resourceVersion st } } -// Index returns a list of items that match on the index function -// Index is thread-safe so long as you treat all items as immutable +// Index returns a list of items that match the given object on the index function. +// Index is thread-safe so long as you treat all items as immutable. func (c *threadSafeMap) Index(indexName string, obj interface{}) ([]interface{}, error) { c.lock.RLock() defer c.lock.RUnlock() @@ -142,37 +142,37 @@ func (c *threadSafeMap) Index(indexName string, obj interface{}) ([]interface{}, return nil, fmt.Errorf("Index with name %s does not exist", indexName) } - indexKeys, err := indexFunc(obj) + indexedValues, err := indexFunc(obj) if err != nil { return nil, err } index := c.indices[indexName] - var returnKeySet sets.String - if len(indexKeys) == 1 { + var storeKeySet sets.String + if len(indexedValues) == 1 { // In majority of cases, there is exactly one value matching. // Optimize the most common path - deduping is not needed here. - returnKeySet = index[indexKeys[0]] + storeKeySet = index[indexedValues[0]] } else { // Need to de-dupe the return list. // Since multiple keys are allowed, this can happen. - returnKeySet = sets.String{} - for _, indexKey := range indexKeys { - for key := range index[indexKey] { - returnKeySet.Insert(key) + storeKeySet = sets.String{} + for _, indexedValue := range indexedValues { + for key := range index[indexedValue] { + storeKeySet.Insert(key) } } } - list := make([]interface{}, 0, returnKeySet.Len()) - for absoluteKey := range returnKeySet { - list = append(list, c.items[absoluteKey]) + list := make([]interface{}, 0, storeKeySet.Len()) + for storeKey := range storeKeySet { + list = append(list, c.items[storeKey]) } return list, nil } -// ByIndex returns a list of items that match an exact value on the index function -func (c *threadSafeMap) ByIndex(indexName, indexKey string) ([]interface{}, error) { +// ByIndex returns a list of the items whose indexed values in the given index include the given indexed value +func (c *threadSafeMap) ByIndex(indexName, indexedValue string) ([]interface{}, error) { c.lock.RLock() defer c.lock.RUnlock() @@ -183,7 +183,7 @@ func (c *threadSafeMap) ByIndex(indexName, indexKey string) ([]interface{}, erro index := c.indices[indexName] - set := index[indexKey] + set := index[indexedValue] list := make([]interface{}, 0, set.Len()) for key := range set { list = append(list, c.items[key]) @@ -192,9 +192,9 @@ func (c *threadSafeMap) ByIndex(indexName, indexKey string) ([]interface{}, erro return list, nil } -// IndexKeys returns a list of keys that match on the index function. +// IndexKeys returns a list of the Store keys of the objects whose indexed values in the given index include the given indexed value. // IndexKeys is thread-safe so long as you treat all items as immutable. -func (c *threadSafeMap) IndexKeys(indexName, indexKey string) ([]string, error) { +func (c *threadSafeMap) IndexKeys(indexName, indexedValue string) ([]string, error) { c.lock.RLock() defer c.lock.RUnlock() @@ -205,7 +205,7 @@ func (c *threadSafeMap) IndexKeys(indexName, indexKey string) ([]string, error) index := c.indices[indexName] - set := index[indexKey] + set := index[indexedValue] return set.List(), nil }