Merge pull request kubernetes#87362 from MikeSpreitzer/limited-cancel2

Simplified and corrected logic around context cancelation in refactored QueueSet

Author: k8s-ci-robot

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/interface.go

@@ -21,9 +21,22 @@ import (
 	"time"
 )
 
-// QueueSetFactory is used to create QueueSet objects.
+// QueueSetFactory is used to create QueueSet objects.  Creation, like
+// config update, is done in two phases: the first phase consumes the
+// QueuingConfig and the second consumes the DispatchingConfig.  They
+// are separated so that errors from the first phase can be found
+// before committing to a concurrency allotment for the second.
 type QueueSetFactory interface {
-	NewQueueSet(config QueueSetConfig) (QueueSet, error)
+	// BeginConstruction does the first phase of creating a QueueSet
+	BeginConstruction(QueuingConfig) (QueueSetCompleter, error)
+}
+
+// QueueSetCompleter finishes the two-step process of creating or
+// reconfiguring a QueueSet
+type QueueSetCompleter interface {
+	// Complete returns a QueueSet configured by the given
+	// dispatching configuration.
+	Complete(DispatchingConfig) QueueSet
 }
 
 // QueueSet is the abstraction for the queuing and dispatching
@@ -34,19 +47,27 @@ type QueueSetFactory interface {
 // .  Some day we may have connections between priority levels, but
 // today is not that day.
 type QueueSet interface {
-	// SetConfiguration updates the configuration
-	SetConfiguration(QueueSetConfig) error
-
-	// Quiesce controls whether the QueueSet is operating normally or is quiescing.
-	// A quiescing QueueSet drains as normal but does not admit any
-	// new requests. Passing a non-nil handler means the system should
-	// be quiescing, a nil handler means the system should operate
-	// normally. A call to Wait while the system is quiescing
-	// will be rebuffed by returning tryAnother=true. If all the
-	// queues have no requests waiting nor executing while the system
-	// is quiescing then the handler will eventually be called with no
-	// locks held (even if the system becomes non-quiescing between the
-	// triggering state and the required call).
+	// BeginConfigChange starts the two-step process of updating
+	// the configuration.  No change is made until Complete is
+	// called.  If `C := X.BeginConstruction(q)` then
+	// `C.Complete(d)` returns the same value `X`.  If the
+	// QueuingConfig's DesiredNumQueues field is zero then the other
+	// queuing-specific config parameters are not changed, so that the
+	// queues continue draining as before.
+	BeginConfigChange(QueuingConfig) (QueueSetCompleter, error)
+
+	// Quiesce controls whether the QueueSet is operating normally or
+	// is quiescing.  A quiescing QueueSet drains as normal but does
+	// not admit any new requests. Passing a non-nil handler means the
+	// system should be quiescing, a nil handler means the system
+	// should operate normally. A call to Wait while the system is
+	// quiescing will be rebuffed by returning tryAnother=true. If all
+	// the queues have no requests waiting nor executing while the
+	// system is quiescing then the handler will eventually be called
+	// with no locks held (even if the system becomes non-quiescing
+	// between the triggering state and the required call).  In Go
+	// Memory Model terms, the triggering state happens before the
+	// call to the EmptyHandler.
 	Quiesce(EmptyHandler)
 
 	// Wait uses the given hashValue as the source of entropy as it
@@ -56,34 +77,47 @@ type QueueSet interface {
 	// tryAnother==true at return then the QueueSet has become
 	// undesirable and the client should try to find a different
 	// QueueSet to use; execute and afterExecution are irrelevant in
-	// this case.  Otherwise, if execute then the client should start
-	// executing the request and, once the request finishes execution
-	// or is canceled, call afterExecution().  Otherwise the client
-	// should not execute the request and afterExecution is
-	// irrelevant.
+	// this case.  In the terms of the Go Memory Model, there was a
+	// call to Quiesce with a non-nil handler that happened before
+	// this return from Wait.  Otherwise, if execute then the client
+	// should start executing the request and, once the request
+	// finishes execution or is canceled, call afterExecution().
+	// Otherwise the client should not execute the request and
+	// afterExecution is irrelevant.  Canceling the context while the
+	// request is waiting in its queue will cut short that wait and
+	// cause a return with tryAnother and execute both false; later
+	// cancellations are the caller's problem.
 	Wait(ctx context.Context, hashValue uint64, descr1, descr2 interface{}) (tryAnother, execute bool, afterExecution func())
 }
 
-// QueueSetConfig defines the configuration of a QueueSet.
-type QueueSetConfig struct {
+// QueuingConfig defines the configuration of the queuing aspect of a QueueSet.
+type QueuingConfig struct {
 	// Name is used to identify a queue set, allowing for descriptive information about its intended use
 	Name string
-	// ConcurrencyLimit is the maximum number of requests of this QueueSet that may be executing at a time
-	ConcurrencyLimit int
+
 	// DesiredNumQueues is the number of queues that the API says
 	// should exist now.  This may be zero, in which case
 	// QueueLengthLimit, HandSize, and RequestWaitLimit are ignored.
 	DesiredNumQueues int
+
 	// QueueLengthLimit is the maximum number of requests that may be waiting in a given queue at a time
 	QueueLengthLimit int
+
 	// HandSize is a parameter of shuffle sharding.  Upon arrival of a request, a queue is chosen by randomly
 	// dealing a "hand" of this many queues and then picking one of minimum length.
 	HandSize int
+
 	// RequestWaitLimit is the maximum amount of time that a request may wait in a queue.
 	// If, by the end of that time, the request has not been dispatched then it is rejected.
 	RequestWaitLimit time.Duration
 }
 
+// DispatchingConfig defines the configuration of the dispatching aspect of a QueueSet.
+type DispatchingConfig struct {
+	// ConcurrencyLimit is the maximum number of requests of this QueueSet that may be executing at a time
+	ConcurrencyLimit int
+}
+
 // EmptyHandler is used to notify the callee when all the queues
 // of a QueueSet have been drained.
 type EmptyHandler interface {

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/promise/interface.go

@@ -16,27 +16,114 @@ limitations under the License.
 
 package promise
 
-// Mutable is a variable that is initially not set and can be set one
-// or more times (unlike a traditional "promise", which can be written
-// only once).
-type Mutable interface {
+// This file defines interfaces for promises and futures and related
+// things.  These are about coordination among multiple goroutines and
+// so are safe for concurrent calls --- although moderated in some
+// cases by a requirement that the caller hold a certain lock.
 
+// Readable represents a variable that is initially not set and later
+// becomes set.  Some instances may be set to multiple values in
+// series.  A Readable for a variable that can only get one value is
+// commonly known as a "future".
+type Readable interface {
+	// Get reads the current value of this variable.  If this variable
+	// is not set yet then this call blocks until this variable gets a
+	// value.
+	Get() interface{}
+
+	// IsSet returns immediately with an indication of whether this
+	// variable has been set.
+	IsSet() bool
+}
+
+// LockingReadable is a Readable whose implementation is protected by
+// a lock
+type LockingReadable interface {
+	Readable
+
+	// GetLocked is like Get but the caller must already hold the
+	// lock.  GetLocked may release, and later re-acquire, the lock
+	// any number of times.  Get may acquire, and later release, the
+	// lock any number of times.
+	GetLocked() interface{}
+
+	// IsSetLocked is like IsSet but the caller must already hold the
+	// lock.  IsSetLocked may release, and later re-acquire, the lock
+	// any number of times.  IsSet may acquire, and later release, the
+	// lock any number of times.
+	IsSetLocked() bool
+}
+
+// WriteOnceOnly represents a variable that is initially not set and
+// can be set once.
+type WriteOnceOnly interface {
+	// Set normally writes a value into this variable, unblocks every
+	// goroutine waiting for this variable to have a value, and
+	// returns true.  In the unhappy case that this variable is
+	// already set, this method returns false without modifying the
+	// variable's value.
+	Set(interface{}) bool
+}
+
+// WriteOnce represents a variable that is initially not set and can
+// be set once and is readable.  This is the common meaning for
+// "promise".
+type WriteOnce interface {
+	Readable
+	WriteOnceOnly
+}
+
+// LockingWriteOnceOnly is a WriteOnceOnly whose implementation is
+// protected by a lock.
+type LockingWriteOnceOnly interface {
+	WriteOnceOnly
+
+	// SetLocked is like Set but the caller must already hold the
+	// lock.  SetLocked may release, and later re-acquire, the lock
+	// any number of times.  Set may acquire, and later release, the
+	// lock any number of times
+	SetLocked(interface{}) bool
+}
+
+// LockingWriteOnce is a WriteOnce whose implementation is protected
+// by a lock.
+type LockingWriteOnce interface {
+	LockingReadable
+	LockingWriteOnceOnly
+}
+
+// WriteMultipleOnly represents a variable that is initially not set
+// and can be set one or more times (unlike a traditional "promise",
+// which can be written only once).
+type WriteMultipleOnly interface {
 	// Set writes a value into this variable and unblocks every
 	// goroutine waiting for this variable to have a value
 	Set(interface{})
+}
 
-	// Get reads the value of this variable.  If this variable is
-	// not set yet then this call blocks until this variable gets a value.
-	Get() interface{}
+// WriteMultiple represents a variable that is initially not set and
+// can be set one or more times (unlike a traditional "promise", which
+// can be written only once) and is readable.
+type WriteMultiple interface {
+	Readable
+	WriteMultipleOnly
 }
 
-// LockingMutable is a Mutable whose implementation is protected by a lock
-type LockingMutable interface {
-	Mutable
+// LockingWriteMultipleOnly is a WriteMultipleOnly whose
+// implementation is protected by a lock.
+type LockingWriteMultipleOnly interface {
+	WriteMultipleOnly
 
-	// SetLocked is like Set but the caller must already hold the lock
+	// SetLocked is like Set but the caller must already hold the
+	// lock.  SetLocked may release, and later re-acquire, the lock
+	// any number of times.  Set may acquire, and later release, the
+	// lock any number of times
 	SetLocked(interface{})
+}
 
-	// GetLocked is like Get but the caller must already hold the lock
-	GetLocked() interface{}
+// LockingWriteMultiple is a WriteMultiple whose implementation is
+// protected by a lock.
+type LockingWriteMultiple interface {
+	LockingReadable
+	LockingWriteMultipleOnly
 }

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/promise/lockingpromise/lockingpromise.go

@@ -23,57 +23,102 @@ import (
 	"k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/promise"
 )
 
-// lockingPromise implements LockingMutable based on a condition
-// variable.  This implementation tracks active goroutines: the given
-// counter is decremented for a goroutine waiting for this varible to
-// be set and incremented when such a goroutine is unblocked.
-type lockingPromise struct {
+// promisoid is the data and behavior common to all the promise-like
+// abstractions implemented here.  This implementation is based on a
+// condition variable.  This implementation tracks active goroutines:
+// the given counter is decremented for a goroutine waiting for this
+// varible to be set and incremented when such a goroutine is
+// unblocked.
+type promisoid struct {
 	lock          sync.Locker
 	cond          sync.Cond
 	activeCounter counter.GoRoutineCounter // counter of active goroutines
-	waitingCount  int                      // number of goroutines idle due to this mutable being unset
+	waitingCount  int                      // number of goroutines idle due to this being unset
 	isSet         bool
 	value         interface{}
 }
 
-var _ promise.LockingMutable = &lockingPromise{}
+func (pr *promisoid) Get() interface{} {
+	pr.lock.Lock()
+	defer pr.lock.Unlock()
+	return pr.GetLocked()
+}
 
-// NewLockingPromise makes a new promise.LockingMutable
-func NewLockingPromise(lock sync.Locker, activeCounter counter.GoRoutineCounter) promise.LockingMutable {
-	return &lockingPromise{
-		lock:          lock,
-		cond:          *sync.NewCond(lock),
-		activeCounter: activeCounter,
+func (pr *promisoid) GetLocked() interface{} {
+	if !pr.isSet {
+		pr.waitingCount++
+		pr.activeCounter.Add(-1)
+		pr.cond.Wait()
 	}
+	return pr.value
 }
 
-func (lp *lockingPromise) Set(value interface{}) {
-	lp.lock.Lock()
-	defer lp.lock.Unlock()
-	lp.SetLocked(value)
+func (pr *promisoid) IsSet() bool {
+	pr.lock.Lock()
+	defer pr.lock.Unlock()
+	return pr.IsSetLocked()
 }
 
-func (lp *lockingPromise) Get() interface{} {
-	lp.lock.Lock()
-	defer lp.lock.Unlock()
-	return lp.GetLocked()
+func (pr *promisoid) IsSetLocked() bool {
+	return pr.isSet
 }
 
-func (lp *lockingPromise) SetLocked(value interface{}) {
-	lp.isSet = true
-	lp.value = value
-	if lp.waitingCount > 0 {
-		lp.activeCounter.Add(lp.waitingCount)
-		lp.waitingCount = 0
-		lp.cond.Broadcast()
+func (pr *promisoid) SetLocked(value interface{}) {
+	pr.isSet = true
+	pr.value = value
+	if pr.waitingCount > 0 {
+		pr.activeCounter.Add(pr.waitingCount)
+		pr.waitingCount = 0
+		pr.cond.Broadcast()
 	}
 }
 
-func (lp *lockingPromise) GetLocked() interface{} {
-	if !lp.isSet {
-		lp.waitingCount++
-		lp.activeCounter.Add(-1)
-		lp.cond.Wait()
+type writeOnce struct {
+	promisoid
+}
+
+var _ promise.LockingWriteOnce = &writeOnce{}
+
+// NewWriteOnce makes a new promise.LockingWriteOnce
+func NewWriteOnce(lock sync.Locker, activeCounter counter.GoRoutineCounter) promise.LockingWriteOnce {
+	return &writeOnce{promisoid{
+		lock:          lock,
+		cond:          *sync.NewCond(lock),
+		activeCounter: activeCounter,
+	}}
+}
+
+func (wr *writeOnce) Set(value interface{}) bool {
+	wr.lock.Lock()
+	defer wr.lock.Unlock()
+	return wr.SetLocked(value)
+}
+
+func (wr *writeOnce) SetLocked(value interface{}) bool {
+	if wr.isSet {
+		return false
 	}
-	return lp.value
+	wr.promisoid.SetLocked(value)
+	return true
+}
+
+type writeMultiple struct {
+	promisoid
+}
+
+var _ promise.LockingWriteMultiple = &writeMultiple{}
+
+// NewWriteMultiple makes a new promise.LockingWriteMultiple
+func NewWriteMultiple(lock sync.Locker, activeCounter counter.GoRoutineCounter) promise.LockingWriteMultiple {
+	return &writeMultiple{promisoid{
+		lock:          lock,
+		cond:          *sync.NewCond(lock),
+		activeCounter: activeCounter,
+	}}
+}
+
+func (wr *writeMultiple) Set(value interface{}) {
+	wr.lock.Lock()
+	defer wr.lock.Unlock()
+	wr.SetLocked(value)
 }