unexport scheduler and add health warning

Author: Jim Minter

pkg/image/controller/factory.go

@@ -92,7 +92,7 @@ func NewScheduledImageStreamController(client imageclient.Interface, informer im
 		listerSynced: informer.Informer().HasSynced,
 	}
 
-	controller.scheduler = NewScheduler(opts.Buckets(), bucketLimiter, controller.syncTimed)
+	controller.scheduler = newScheduler(opts.Buckets(), bucketLimiter, controller.syncTimed)
 
 	informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    controller.addImageStream,

pkg/image/controller/scheduled_image_controller.go

@@ -36,7 +36,7 @@ type ScheduledImageStreamController struct {
 	rateLimiter flowcontrol.RateLimiter
 
 	// scheduler for timely image re-imports
-	scheduler *Scheduler
+	scheduler *scheduler
 }
 
 // Importing is invoked when the controller decides to import a stream in order to push back

pkg/image/controller/scheduler.go

@@ -7,12 +7,19 @@ import (
 	"k8s.io/client-go/util/flowcontrol"
 )
 
-// Scheduler is a self-balancing, rate-limited, bucketed queue that can periodically invoke
+// NOTE: scheduler's semantics do not lend it for reuse elsewhere and its use in
+// this package quite probably has some odd corner cases/race conditions.  If
+// these cause problems in the future, this implementation should be replaced
+// with a new and simpler one based on container/heap.  End users looking for a
+// component like this: see if k8s.io/client-go/util/workqueue.NewDelayingQueue
+// suits your needs.
+
+// scheduler is a self-balancing, rate-limited, bucketed queue that can periodically invoke
 // an action on all items in a bucket before moving to the next bucket. A ratelimiter sets
 // an upper bound on the number of buckets processed per unit time. The queue has a key and a
 // value, so both uniqueness and equality can be tested (key must be unique, value can carry
 // info for the next processing). Items remain in the queue until removed by a call to Remove().
-type Scheduler struct {
+type scheduler struct {
 	handle   func(key, value interface{})
 	position int
 	limiter  flowcontrol.RateLimiter
@@ -23,32 +30,32 @@ type Scheduler struct {
 
 type bucket map[interface{}]interface{}
 
-// NewScheduler creates a scheduler with bucketCount buckets, a rate limiter for restricting
+// newScheduler creates a scheduler with bucketCount buckets, a rate limiter for restricting
 // the rate at which buckets are processed, and a function to invoke when items are scanned in
 // a bucket.
 // TODO: remove DEBUG statements from this file once this logic has been adequately validated.
-func NewScheduler(bucketCount int, bucketLimiter flowcontrol.RateLimiter, fn func(key, value interface{})) *Scheduler {
+func newScheduler(bucketCount int, bucketLimiter flowcontrol.RateLimiter, fn func(key, value interface{})) *scheduler {
 	// Add one more bucket to serve as the "current" bucket
 	bucketCount++
 	buckets := make([]bucket, bucketCount)
 	for i := range buckets {
 		buckets[i] = make(bucket)
 	}
-	return &Scheduler{
+	return &scheduler{
 		handle:  fn,
 		buckets: buckets,
 		limiter: bucketLimiter,
 	}
 }
 
 // RunUntil launches the scheduler until ch is closed.
-func (s *Scheduler) RunUntil(ch <-chan struct{}) {
+func (s *scheduler) RunUntil(ch <-chan struct{}) {
 	go utilwait.Until(s.RunOnce, 0, ch)
 }
 
 // RunOnce takes a single item out of the current bucket and processes it. If
 // the bucket is empty, we wait for the rate limiter before returning.
-func (s *Scheduler) RunOnce() {
+func (s *scheduler) RunOnce() {
 	key, value, last := s.next()
 	if last {
 		s.limiter.Accept()
@@ -58,13 +65,13 @@ func (s *Scheduler) RunOnce() {
 }
 
 // at returns the bucket index relative to the current bucket.
-func (s *Scheduler) at(inc int) int {
+func (s *scheduler) at(inc int) int {
 	return (s.position + inc + len(s.buckets)) % len(s.buckets)
 }
 
 // next takes a key from the current bucket and places it in the last bucket, returns the
 // removed key. Returns true if the current bucket is empty and no key and value were returned.
-func (s *Scheduler) next() (interface{}, interface{}, bool) {
+func (s *scheduler) next() (interface{}, interface{}, bool) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -85,7 +92,7 @@ func (s *Scheduler) next() (interface{}, interface{}, bool) {
 // removes the previous key and value and will place the item in a new bucket. This allows callers to ensure
 // that Add'ing a new item to the queue purges old versions of the item, while Remove can be conditional on
 // removing only the known old version.
-func (s *Scheduler) Add(key, value interface{}) {
+func (s *scheduler) Add(key, value interface{}) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -114,7 +121,7 @@ func (s *Scheduler) Add(key, value interface{}) {
 
 // Remove takes the key out of all buckets. If value is non-nil, the key will only be removed if it has
 // the same value. Returns true if the key was removed.
-func (s *Scheduler) Remove(key, value interface{}) bool {
+func (s *scheduler) Remove(key, value interface{}) bool {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -132,7 +139,7 @@ func (s *Scheduler) Remove(key, value interface{}) bool {
 }
 
 // Delay moves the key to the end of the chain if it exists.
-func (s *Scheduler) Delay(key interface{}) {
+func (s *scheduler) Delay(key interface{}) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -149,7 +156,7 @@ func (s *Scheduler) Delay(key interface{}) {
 }
 
 // Len returns the number of scheduled items.
-func (s *Scheduler) Len() int {
+func (s *scheduler) Len() int {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -162,7 +169,7 @@ func (s *Scheduler) Len() int {
 
 // Map returns a copy of the scheduler contents, but does not copy the keys or values themselves.
 // If values and keys are not immutable, changing the value will affect the value in the queue.
-func (s *Scheduler) Map() map[interface{}]interface{} {
+func (s *scheduler) Map() map[interface{}]interface{} {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 

pkg/image/controller/scheduler_test.go

@@ -14,7 +14,7 @@ import (
 
 func TestScheduler(t *testing.T) {
 	keys := []string{}
-	s := NewScheduler(2, flowcontrol.NewFakeAlwaysRateLimiter(), func(key, value interface{}) {
+	s := newScheduler(2, flowcontrol.NewFakeAlwaysRateLimiter(), func(key, value interface{}) {
 		keys = append(keys, key.(string))
 	})
 
@@ -59,7 +59,7 @@ func TestScheduler(t *testing.T) {
 }
 
 func TestSchedulerAddAndDelay(t *testing.T) {
-	s := NewScheduler(3, flowcontrol.NewFakeAlwaysRateLimiter(), func(key, value interface{}) {})
+	s := newScheduler(3, flowcontrol.NewFakeAlwaysRateLimiter(), func(key, value interface{}) {})
 	// 3 is the last bucket, 0 is the current bucket
 	s.Add("first", "other")
 	if s.buckets[3]["first"] != "other" {
@@ -105,7 +105,7 @@ func TestSchedulerAddAndDelay(t *testing.T) {
 }
 
 func TestSchedulerRemove(t *testing.T) {
-	s := NewScheduler(2, flowcontrol.NewFakeAlwaysRateLimiter(), func(key, value interface{}) {})
+	s := newScheduler(2, flowcontrol.NewFakeAlwaysRateLimiter(), func(key, value interface{}) {})
 	s.Add("test", "other")
 	if s.Remove("test", "value") {
 		t.Fatal(s)
@@ -204,7 +204,7 @@ func TestSchedulerSanity(t *testing.T) {
 	limiter := flowcontrol.NewTokenBucketRateLimiterWithClock(1, 1, clock)
 
 	m := map[int]int{}
-	s := NewScheduler(buckets, limiter, func(key, value interface{}) {
+	s := newScheduler(buckets, limiter, func(key, value interface{}) {
 		fmt.Printf("%v: %v\n", clock.Now().UTC(), key)
 		m[key.(int)]++
 	})