published by bot

(https://github.com/kubernetes/contrib/tree/master/mungegithub)

copied from https://github.com/kubernetes/kubernetes.git, branch master,
last commit is a2b65f03bf83013d1af05c99b7aa22049ca63de3

util/workqueue/delaying_queue.go

@@ -0,0 +1,246 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package workqueue
+
+import (
+	"sort"
+	"time"
+
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/util/clock"
+)
+
+// 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
+	// AddAfter adds an item to the workqueue after the indicated duration has passed
+	AddAfter(item interface{}, duration time.Duration)
+}
+
+// NewDelayingQueue constructs a new workqueue with delayed queuing ability
+func NewDelayingQueue() DelayingInterface {
+	return newDelayingQueue(clock.RealClock{}, "")
+}
+
+func NewNamedDelayingQueue(name string) DelayingInterface {
+	return newDelayingQueue(clock.RealClock{}, name)
+}
+
+func newDelayingQueue(clock clock.Clock, name string) DelayingInterface {
+	ret := &delayingType{
+		Interface:          NewNamed(name),
+		clock:              clock,
+		heartbeat:          clock.Tick(maxWait),
+		stopCh:             make(chan struct{}),
+		waitingTimeByEntry: map[t]time.Time{},
+		waitingForAddCh:    make(chan waitFor, 1000),
+		metrics:            newRetryMetrics(name),
+	}
+
+	go ret.waitingLoop()
+
+	return ret
+}
+
+// delayingType wraps an Interface and provides delayed re-enquing
+type delayingType struct {
+	Interface
+
+	// clock tracks time for delayed firing
+	clock clock.Clock
+
+	// stopCh lets us signal a shutdown to the waiting loop
+	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
+	waitingForAdd []waitFor
+	// waitingTimeByEntry holds wait time by entry, so we can lookup pre-existing indexes
+	waitingTimeByEntry map[t]time.Time
+	// waitingForAddCh is a buffered channel that feeds waitingForAdd
+	waitingForAddCh chan waitFor
+
+	// metrics counts the number of retries
+	metrics retryMetrics
+}
+
+// waitFor holds the data to add and the time it should be added
+type waitFor struct {
+	data    t
+	readyAt time.Time
+}
+
+// ShutDown gives a way to shut off this queue
+func (q *delayingType) ShutDown() {
+	q.Interface.ShutDown()
+	close(q.stopCh)
+}
+
+// AddAfter adds the given item to the work queue after the given delay
+func (q *delayingType) AddAfter(item interface{}, duration time.Duration) {
+	// don't add if we're already shutting down
+	if q.ShuttingDown() {
+		return
+	}
+
+	q.metrics.retry()
+
+	// immediately add things with no delay
+	if duration <= 0 {
+		q.Add(item)
+		return
+	}
+
+	select {
+	case <-q.stopCh:
+		// unblock if ShutDown() is called
+	case q.waitingForAddCh <- waitFor{data: item, readyAt: q.clock.Now().Add(duration)}:
+	}
+}
+
+// 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
+
+// waitingLoop runs until the workqueue is shutdown and keeps a check on the list of items to be added.
+func (q *delayingType) waitingLoop() {
+	defer utilruntime.HandleCrash()
+
+	// Make a placeholder channel to use when there are no items in our list
+	never := make(<-chan time.Time)
+
+	for {
+		if q.Interface.ShuttingDown() {
+			// discard waiting entries
+			q.waitingForAdd = nil
+			q.waitingTimeByEntry = nil
+			return
+		}
+
+		now := q.clock.Now()
+
+		// Add ready entries
+		readyEntries := 0
+		for _, entry := range q.waitingForAdd {
+			if entry.readyAt.After(now) {
+				break
+			}
+			q.Add(entry.data)
+			delete(q.waitingTimeByEntry, entry.data)
+			readyEntries++
+		}
+		q.waitingForAdd = q.waitingForAdd[readyEntries:]
+
+		// Set up a wait for the first item's readyAt (if one exists)
+		nextReadyAt := never
+		if len(q.waitingForAdd) > 0 {
+			nextReadyAt = q.clock.After(q.waitingForAdd[0].readyAt.Sub(now))
+		}
+
+		select {
+		case <-q.stopCh:
+			return
+
+		case <-q.heartbeat:
+			// continue the loop, which will add ready items
+
+		case <-nextReadyAt:
+			// continue the loop, which will add ready items
+
+		case waitEntry := <-q.waitingForAddCh:
+			if waitEntry.readyAt.After(q.clock.Now()) {
+				q.waitingForAdd = insert(q.waitingForAdd, q.waitingTimeByEntry, waitEntry)
+			} else {
+				q.Add(waitEntry.data)
+			}
+
+			drained := false
+			for !drained {
+				select {
+				case waitEntry := <-q.waitingForAddCh:
+					if waitEntry.readyAt.After(q.clock.Now()) {
+						q.waitingForAdd = insert(q.waitingForAdd, q.waitingTimeByEntry, waitEntry)
+					} else {
+						q.Add(waitEntry.data)
+					}
+				default:
+					drained = true
+				}
+			}
+		}
+	}
+}
+
+// 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]
+	if exists && existingTime.Before(entry.readyAt) {
+		return entries
+	}
+
+	// if the entry exists and is scheduled for later, go ahead and remove the entry
+	if exists {
+		if existingIndex := findEntryIndex(entries, existingTime, entry.data); existingIndex >= 0 && existingIndex < len(entries) {
+			entries = append(entries[:existingIndex], entries[existingIndex+1:]...)
+		}
+	}
+
+	insertionIndex := sort.Search(len(entries), func(i int) bool {
+		return entry.readyAt.Before(entries[i].readyAt)
+	})
+
+	// grow by 1
+	entries = append(entries, waitFor{})
+	// shift items from the insertion point to the end
+	copy(entries[insertionIndex+1:], entries[insertionIndex:])
+	// insert the record
+	entries[insertionIndex] = entry
+
+	knownEntries[entry.data] = entry.readyAt
+
+	return entries
+}
+
+// findEntryIndex returns the index for an existing entry
+func findEntryIndex(entries []waitFor, existingTime time.Time, data t) int {
+	index := sort.Search(len(entries), func(i int) bool {
+		return entries[i].readyAt.After(existingTime) || existingTime == entries[i].readyAt
+	})
+
+	// we know this is the earliest possible index, but there could be multiple with the same time
+	// iterate from here to find the dupe
+	for ; index < len(entries); index++ {
+		if entries[index].data == data {
+			break
+		}
+	}
+
+	return index
+}