Merge pull request #35382 from jbeda/nits

pkg/client/cache/controller.go

@@ -29,7 +29,7 @@ import (
 type Config struct {
 	// The queue for your objects; either a FIFO or
 	// a DeltaFIFO. Your Process() function should accept
-	// the output of this Oueue's Pop() method.
+	// the output of this Queue's Pop() method.
 	Queue
 
 	// Something that can list and watch your objects.
@@ -121,6 +121,11 @@ func (c *Controller) Requeue(obj interface{}) error {
 // TODO: Consider doing the processing in parallel. This will require a little thought
 // to make sure that we don't end up processing the same object multiple times
 // concurrently.
+//
+// TODO: Plumb through the stopCh here (and down to the queue) so that this can
+// actually exit when the controller is stopped. Or just give up on this stuff
+// ever being stoppable. Converting this whole package to use Context would
+// also be helpful.
 func (c *Controller) processLoop() {
 	for {
 		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
@@ -134,7 +139,7 @@ func (c *Controller) processLoop() {
 }
 
 // ResourceEventHandler can handle notifications for events that happen to a
-// resource.  The events are informational only, so you can't return an
+// resource. The events are informational only, so you can't return an
 // error.
 //  * OnAdd is called when an object is added.
 //  * OnUpdate is called when an object is modified. Note that oldObj is the

pkg/client/cache/controller_test.go

@@ -283,6 +283,9 @@ func TestHammerController(t *testing.T) {
 	time.Sleep(100 * time.Millisecond)
 	close(stop)
 
+	// TODO: Verify that no goroutines were leaked here and that everything shut
+	// down cleanly.
+
 	outputSetLock.Lock()
 	t.Logf("got: %#v", outputSet)
 }

pkg/client/cache/reflector.go

@@ -45,7 +45,7 @@ import (
 
 // Reflector watches a specified resource and causes all changes to be reflected in the given store.
 type Reflector struct {
-	// name identifies this reflector.  By default it will be a file:line if possible.
+	// name identifies this reflector. By default it will be a file:line if possible.
 	name string
 
 	// The type of object we expect to place in the store.
@@ -74,12 +74,6 @@ var (
 	// However, it can be modified to avoid periodic resync to break the
 	// TCP connection.
 	minWatchTimeout = 5 * time.Minute
-	// If we are within 'forceResyncThreshold' from the next planned resync
-	// and are just before issuing Watch(), resync will be forced now.
-	forceResyncThreshold = 3 * time.Second
-	// We try to set timeouts for Watch() so that we will finish about
-	// than 'timeoutThreshold' from next planned periodic resync.
-	timeoutThreshold = 1 * time.Second
 )
 
 // NewNamespaceKeyedIndexerAndReflector creates an Indexer and a Reflector
@@ -114,7 +108,7 @@ func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{},
 	return r
 }
 
-// internalPackages are packages that ignored when creating a default reflector name.  These packages are in the common
+// internalPackages are packages that ignored when creating a default reflector name. These packages are in the common
 // call chains to NewReflector, so they'd be low entropy names for reflectors
 var internalPackages = []string{"kubernetes/pkg/client/cache/", "/runtime/asm_"}
 

pkg/util/workqueue/delaying_queue.go

@@ -24,7 +24,7 @@ import (
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 )
 
-// DelayingInterface is an Interface that can Add an item at a later time.  This makes it easier to
+// DelayingInterface is an Interface that can Add an item at a later time. This makes it easier to
 // requeue items after failures without ending up in a hot-loop.
 type DelayingInterface interface {
 	Interface
@@ -68,6 +68,9 @@ type delayingType struct {
 	stopCh chan struct{}
 
 	// heartbeat ensures we wait no more than maxWait before firing
+	//
+	// TODO: replace with Ticker (and add to clock) so this can be cleaned up.
+	// clock.Tick will leak.
 	heartbeat <-chan time.Time
 
 	// waitingForAdd is an ordered slice of items to be added to the contained work queue
@@ -115,7 +118,7 @@ func (q *delayingType) AddAfter(item interface{}, duration time.Duration) {
 	}
 }
 
-// maxWait keeps a max bound on the wait time.  It's just insurance against weird things happening.
+// maxWait keeps a max bound on the wait time. It's just insurance against weird things happening.
 // Checking the queue every 10 seconds isn't expensive and we know that we'll never end up with an
 // expired item sitting for more than 10 seconds.
 const maxWait = 10 * time.Second
@@ -192,6 +195,9 @@ func (q *delayingType) waitingLoop() {
 // inserts the given entry into the sorted entries list
 // same semantics as append()... the given slice may be modified,
 // and the returned value should be used
+//
+// TODO: This should probably be converted to use container/heap to improve
+// running time for a large number of items.
 func insert(entries []waitFor, knownEntries map[t]time.Time, entry waitFor) []waitFor {
 	// if the entry is already in our retry list and the existing time is before the new one, just skip it
 	existingTime, exists := knownEntries[entry.data]

pkg/util/workqueue/parallelizer.go

@@ -33,6 +33,10 @@ func Parallelize(workers, pieces int, doWorkPiece DoWorkPieceFunc) {
 	}
 	close(toProcess)
 
+	if pieces < workers {
+		workers = pieces
+	}
+
 	wg := sync.WaitGroup{}
 	wg.Add(workers)
 	for i := 0; i < workers; i++ {

pkg/util/workqueue/queue.go

@@ -154,7 +154,7 @@ func (q *Type) Done(item interface{}) {
 	}
 }
 
-// Shutdown will cause q to ignore all new items added to it. As soon as the
+// ShutDown will cause q to ignore all new items added to it. As soon as the
 // worker goroutines have drained the existing items in the queue, they will be
 // instructed to exit.
 func (q *Type) ShutDown() {

pkg/util/workqueue/queue_test.go

@@ -129,3 +129,33 @@ func TestLen(t *testing.T) {
 		t.Errorf("Expected %v, got %v", e, a)
 	}
 }
+
+func TestReinsert(t *testing.T) {
+	q := workqueue.New()
+	q.Add("foo")
+
+	// Start processing
+	i, _ := q.Get()
+	if i != "foo" {
+		t.Errorf("Expected %v, got %v", "foo", i)
+	}
+
+	// Add it back while processing
+	q.Add(i)
+
+	// Finish it up
+	q.Done(i)
+
+	// It should be back on the queue
+	i, _ = q.Get()
+	if i != "foo" {
+		t.Errorf("Expected %v, got %v", "foo", i)
+	}
+
+	// Finish that one up
+	q.Done(i)
+
+	if a := q.Len(); a != 0 {
+		t.Errorf("Expected queue to be empty. Has %v items", a)
+	}
+}

pkg/util/workqueue/rate_limitting_queue.go

@@ -16,10 +16,10 @@ limitations under the License.
 
 package workqueue
 
-// RateLimitingInterface is an Interface that can Add an item at a later time.  This makes it easier to
-// requeue items after failures without ending up in a hot-loop.
+// RateLimitingInterface is an interface that rate limits items being added to the queue.
 type RateLimitingInterface interface {
 	DelayingInterface
+
 	// AddRateLimited adds an item to the workqueue after the rate limiter says its ok
 	AddRateLimited(item interface{})
 
@@ -27,6 +27,7 @@ type RateLimitingInterface interface {
 	// or for success, we'll stop the rate limiter from tracking it.  This only clears the `rateLimiter`, you
 	// still have to call `Done` on the queue.
 	Forget(item interface{})
+
 	// NumRequeues returns back how many times the item was requeued
 	NumRequeues(item interface{}) int
 }