github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 02:06:23 +00:00
Code Issues Projects Releases Wiki Activity

add fair queuing implementation and passing test suite

Browse Source
This commit is contained in:
Aaron Prindle
2019-08-16 14:34:06 -07:00
committed by Mike Spreitzer
parent 7a156d2914
commit fb9dc1b2bb
4 changed files with 914 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD
View File

@@ -4,13 +4,28 @@ go_library(
name = "go_default_library",
srcs = [
"dummy.go",
"fairqueuing.go",
"integrator.go",
"interface.go",
"no-restraint.go",
"types.go",
],
importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing",
importpath = "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing",
visibility = ["//visibility:public"],
deps = [
"//staging/src/k8s.io/apiserver/pkg/util/clock:go_default_library",
"//vendor/k8s.io/klog:go_default_library",
],
)
go_test(
name = "go_default_test",
srcs = [
"fairqueuing_test.go",
"fq_test.go",
],
embed = [":go_default_library"],
deps = ["//staging/src/k8s.io/apiserver/pkg/util/clock:go_default_library"],
)
@@ -27,10 +42,3 @@ filegroup(
tags = ["automanaged"],
visibility = ["//visibility:public"],
)
go_test(
name = "go_default_test",
srcs = ["fq_test.go"],
embed = [":go_default_library"],
deps = ["//staging/src/k8s.io/apiserver/pkg/util/clock:go_default_library"],
)

632
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go Normal file
View File

@@ -0,0 +1,632 @@
/*
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 fairqueuing
import (
"math"
"sync"
"time"
"k8s.io/apiserver/pkg/util/clock"
"k8s.io/klog"
)
// queueSetFactoryImpl implements the QueueSetFactory interface
// queueSetFactoryImpl makes QueueSetSystem objects.
// This filter makes a QueueSetSystem for each priority level.
type queueSetFactoryImpl struct {
// wg can be nil and is ignored in that case
wg *sync.WaitGroup
clk clock.PassiveClock
}
// NewQueueSetFactory creates a new NewQueueSetFactory object
func NewQueueSetFactory(clk clock.PassiveClock, wg *sync.WaitGroup) QueueSetFactory {
return &queueSetFactoryImpl{
wg: wg,
clk: clk,
}
}
// NewQueueSet creates a new QueueSetSystem object
// There is a new QueueSet created for each priority level.
func (qsf queueSetFactoryImpl) NewQueueSet(concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration) QueueSet {
return newQueueSetImpl(concurrencyLimit, desiredNumQueues,
queueLengthLimit, requestWaitLimit, qsf.clk, qsf.wg)
}
// queueSetImpl is a fair queuing implementation designed for the kube-apiserver.
// FQ is designed for
// 1) dispatching requests to be served rather than packets to be transmitted
// 2) serving multiple requests at once
// 3) accounting for unknown and varying service times
// implementation of:
// https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/20190228-priority-and-fairness.md
type queueSetImpl struct {
lock sync.Mutex
wg *sync.WaitGroup
queues []*Queue
clk clock.PassiveClock
vt float64
estimatedServiceTime float64
lastRealTime time.Time
robinIdx int
// numRequestsEnqueued is the number of packets currently enqueued
// (eg: incremeneted on Enqueue, decremented on Dequue)
numRequestsEnqueued int
concurrencyLimit int
desiredNumQueues int
queueLengthLimit int
requestWaitLimit time.Duration
quiescent bool
emptyHandler EmptyHandler
}
// initQueues is a helper method for initializing an array of n queues
func initQueues(numQueues int) []*Queue {
fqqueues := make([]*Queue, numQueues, numQueues)
for i := 0; i < numQueues; i++ {
fqqueues[i] = &Queue{Index: i, Requests: make([]*Request, 0)}
}
return fqqueues
}
// newQueueSetImpl creates a new queueSetImpl from passed in parameters and
func newQueueSetImpl(concurrencyLimit, desiredNumQueues, queueLengthLimit int,
requestWaitLimit time.Duration, clk clock.PassiveClock, wg *sync.WaitGroup) *queueSetImpl {
fq := &queueSetImpl{
wg: wg,
queues: initQueues(desiredNumQueues),
clk: clk,
vt: 0,
lastRealTime: clk.Now(),
desiredNumQueues: desiredNumQueues,
concurrencyLimit: concurrencyLimit,
queueLengthLimit: queueLengthLimit,
requestWaitLimit: requestWaitLimit,
}
return fq
}
// LockAndSyncTime is used to ensure that the virtual time of a queueSetImpl
// is synced everytime its fields are accessed
func (qs *queueSetImpl) LockAndSyncTime() {
qs.lock.Lock()
qs.synctime()
}
// SetConfiguration is used to set the configuration for a queueSetImpl
// update handling for when fields are updated is handled here as well -
// eg: if desiredNumQueues is increased, SetConfiguration reconciles by
// adding more queues.
func (qs *queueSetImpl) SetConfiguration(concurrencyLimit, desiredNumQueues, queueLengthLimit int, requestWaitLimit time.Duration) {
// TODO(aaron-prindle) verify updating queues makes sense here vs elsewhere
// lock required as method can change Queues which has its indexes and length used
// concurrently
qs.lock.Lock()
defer qs.lock.Unlock()
// Adding queues is the only thing that requires immediate action
// Removing queues is handled by omitting indexes >desiredNumQueues from
// chooseQueueIdx
numQueues := len(qs.queues)
if desiredNumQueues > numQueues {
qs.addQueues(desiredNumQueues - numQueues)
}
qs.concurrencyLimit = concurrencyLimit
qs.desiredNumQueues = desiredNumQueues
qs.queueLengthLimit = queueLengthLimit
qs.requestWaitLimit = requestWaitLimit
}
// TimeoutOldRequestsAndRejectOrEnqueue encapsulates the lock sharing logic required
// to validated and enqueue a request for the queueSetImpl/QueueSetSystem:
// 1) Start with shuffle sharding, to pick a queue.
// 2) Reject old requests that have been waiting too long
// 3) Reject current request if there is not enough concurrency shares and
// we are at max queue length
// 4) If not rejected, create a packet and enqueue
// returns true on a successful enqueue
// returns false in the case that there is no available concurrency or
// the queuelengthlimit has been reached
// NOTE: This function must only be called with the QueueSet locked
func (qs *queueSetImpl) TimeoutOldRequestsAndRejectOrEnqueue(hashValue uint64, handSize int32) *Request {
// Start with the shuffle sharding, to pick a queue.
queueIdx := qs.ChooseQueueIdx(hashValue, int(handSize))
queue := qs.queues[queueIdx]
// The next step is the logic to reject requests that have been waiting too long
qs.removeTimedOutRequestsFromQueue(queue)
// NOTE: currently timeout is only checked for each new request. This means that there can be
// requests that are in the queue longer than the timeout if there are no new requests
// We think this is a fine tradeoff
// Create a packet and enqueue
pkt := &Request{
DequeueChannel: make(chan bool, 1),
EnqueueTime: qs.clk.Now(),
Queue: queue,
}
if ok := qs.rejectOrEnqueue(pkt); !ok {
return nil
}
return pkt
}
// removeTimedOutRequestsFromQueue rejects old requests that have been enqueued
// past the requestWaitLimit
func (qs *queueSetImpl) removeTimedOutRequestsFromQueue(queue *Queue) {
timeoutIdx := -1
now := qs.clk.Now()
pkts := queue.Requests
// pkts are sorted oldest -> newest
// can short circuit loop (break) if oldest packets are not timing out
// as newer packets also will not have timed out
// now - requestWaitLimit = waitLimit
waitLimit := now.Add(-qs.requestWaitLimit)
for i, pkt := range pkts {
if waitLimit.After(pkt.EnqueueTime) {
if qs.wg != nil {
qs.wg.Add(1)
}
pkt.DequeueChannel <- false
close(pkt.DequeueChannel)
// // TODO(aaron-prindle) verify this makes sense here
// get idx for timed out packets
timeoutIdx = i
} else {
break
}
}
// remove timed out packets from queue
if timeoutIdx != -1 {
// timeoutIdx + 1 to remove the last timeout pkt
removeIdx := timeoutIdx + 1
// remove all the timeout packets
queue.Requests = pkts[removeIdx:]
// decrement the # of requestsEnqueued
qs.numRequestsEnqueued -= removeIdx
}
}
// GetRequestsExecuting gets the # of requests which are "executing":
// this is the# of requests/packets which have been dequeued but have not had
// finished (via the FinishRequest method invoked after service)
func (qs *queueSetImpl) GetRequestsExecuting() int {
total := 0
for _, queue := range qs.queues {
total += queue.RequestsExecuting
}
return total
}
func shuffleDealAndPick(v, nq uint64,
lengthOfQueue func(int) int,
mr func(int /*in [0, nq-1]*/) int, /*in [0, numQueues-1] and excluding previously determined members of I*/
nRem, minLen, bestIdx int) int {
if nRem < 1 {
return bestIdx
}
vNext := v / nq
ai := int(v - nq*vNext)
ii := mr(ai)
mrNext := func(a int /*in [0, nq-2]*/) int /*in [0, numQueues-1] and excluding I[0], I[1], ... ii*/ {
if a < ai {
return mr(a)
}
return mr(a + 1)
}
lenI := lengthOfQueue(ii)
if lenI < minLen {
minLen = lenI
bestIdx = ii
}
return shuffleDealAndPick(vNext, nq-1, lengthOfQueue, mrNext, nRem-1, minLen, bestIdx)
}
// ChooseQueueIdx uses shuffle sharding to select an queue index
// using a 'hashValue'. The 'hashValue' derives a hand from a set range of
// indexes (range 'desiredNumQueues') and returns the queue with the least queued packets
// from a dealt hand (of size 'handSize')
func (qs *queueSetImpl) ChooseQueueIdx(hashValue uint64, handSize int) int {
// TODO(aaron-prindle) currently a lock is held for this in a larger anonymous function
// verify that makes sense...
// desiredNumQueues is used here instead of numQueues to omit quiesce queues
return shuffleDealAndPick(hashValue, uint64(qs.desiredNumQueues),
func(idx int) int { return len(qs.queues[idx].Requests) },
func(i int) int { return i }, handSize, math.MaxInt32, -1)
}
// rejectOrEnqueue rejects or enqueues the newly arrived request if
// resource criteria isn't met
func (qs *queueSetImpl) rejectOrEnqueue(packet *Request) bool {
queue := packet.Queue
curQueueLength := len(queue.Requests)
// rejects the newly arrived request if resource criteria not met
if qs.GetRequestsExecuting() >= qs.concurrencyLimit &&
curQueueLength >= qs.queueLengthLimit {
return false
}
qs.enqueue(packet)
return true
}
// enqueues a packet into an queueSetImpl
func (qs *queueSetImpl) enqueue(packet *Request) {
queue := packet.Queue
queue.Enqueue(packet)
qs.updateQueueVirStartTime(packet, queue)
qs.numRequestsEnqueued++
}
// Enqueue enqueues a packet directly into an queueSetImpl w/ no restriction
func (qs *queueSetImpl) Enqueue(packet *Request) bool {
qs.LockAndSyncTime()
defer qs.lock.Unlock()
qs.enqueue(packet)
return true
}
// synctime is used to sync the time of the queueSetImpl by looking at the elapsed
// time since the last sync and this value based on the 'virtualtime ratio'
// which scales inversely to the # of active flows
func (qs *queueSetImpl) synctime() {
realNow := qs.clk.Now()
timesincelast := realNow.Sub(qs.lastRealTime).Seconds()
qs.lastRealTime = realNow
qs.vt += timesincelast * qs.getvirtualtimeratio()
}
func (qs *queueSetImpl) getvirtualtimeratio() float64 {
NEQ := 0
reqs := 0
for _, queue := range qs.queues {
reqs += queue.RequestsExecuting
// It might be best to delete this line. If everything is working
// correctly, there will be no waiting packets if reqs < C on current
// line 85; if something is going wrong, it is more accurate to say
// that virtual time advanced due to the requests actually executing.
// reqs += len(queue.Requests)
if len(queue.Requests) > 0 || queue.RequestsExecuting > 0 {
NEQ++
}
}
// no active flows, virtual time does not advance (also avoid div by 0)
if NEQ == 0 {
return 0
}
return math.Min(float64(reqs), float64(qs.concurrencyLimit)) / float64(NEQ)
}
// updateQueueVirStartTime updates the virtual start time for a queue
// this is done when a new packet is enqueued. For more info see:
// https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/20190228-priority-and-fairness.md#dispatching
func (qs *queueSetImpl) updateQueueVirStartTime(packet *Request, queue *Queue) {
// When a request arrives to an empty queue with no requests executing:
// len(queue.Requests) == 1 as enqueue has just happened prior (vs == 0)
if len(queue.Requests) == 1 && queue.RequestsExecuting == 0 {
// the queue’s virtual start time is set to the virtual time.
queue.VirStart = qs.vt
}
}
// removeQueueAndUpdateIndexes uses reslicing to remove an index from a slice
// and then updates the 'Index' field of the queues to be correct
func removeQueueAndUpdateIndexes(queues []*Queue, index int) []*Queue {
removedQueues := removeIndex(queues, index)
for i := index; i < len(removedQueues); i++ {
removedQueues[i].Index--
}
return removedQueues
}
// removeIndex uses reslicing to remove an index from a slice
func removeIndex(s []*Queue, index int) []*Queue {
return append(s[:index], s[index+1:]...)
}
// FinishRequestAndDequeueWithChannelAsMuchAsPossible is a convenience method which calls finishRequest
// for a given packet and then dequeues as many packets as possible
// and updates that packet's channel signifying it is is dequeued
// this is a callback used for the filter that the queueSetImpl supports
func (qs *queueSetImpl) FinishRequestAndDequeueWithChannelAsMuchAsPossible(pkt *Request) {
qs.LockAndSyncTime()
defer qs.lock.Unlock()
qs.finishRequest(pkt)
qs.DequeueWithChannelAsMuchAsPossible()
}
// FinishRequest is a callback that should be used when a previously dequeud packet
// has completed it's service. This callback updates imporatnt state in the
// queueSetImpl
func (qs *queueSetImpl) finishRequest(p *Request) {
S := qs.clk.Since(p.StartTime).Seconds()
// When a request finishes being served, and the actual service time was S,
// the queue’s virtual start time is decremented by G - S.
p.Queue.VirStart -= qs.estimatedServiceTime - S
// request has finished, remove from requests executing
p.Queue.RequestsExecuting--
// Logic to remove quiesced queues
// >= as QueueIdx=25 is out of bounds for desiredNumQueues=25 [0...24]
if p.Queue.Index >= qs.desiredNumQueues &&
len(p.Queue.Requests) == 0 &&
p.Queue.RequestsExecuting == 0 {
qs.queues = removeQueueAndUpdateIndexes(qs.queues, p.Queue.Index)
// At this point, if the qs is quiescing,
// has zero requests executing, and has zero requests enqueued
// then a call to the EmptyHandler should be forked.
if qs.quiescent && qs.numRequestsEnqueued == 0 &&
qs.GetRequestsExecuting() == 0 {
// then a call to the EmptyHandler should be forked.
go qs.emptyHandler.HandleEmpty()
}
}
}
// dequeue dequeues a packet from the queueSetImpl
func (qs *queueSetImpl) dequeue() (*Request, bool) {
queue := qs.selectQueue()
if queue == nil {
return nil, false
}
packet, ok := queue.Dequeue()
if ok {
// When a request is dequeued for service -> qs.VirStart += G
queue.VirStart += qs.estimatedServiceTime
packet.StartTime = qs.clk.Now()
// request dequeued, service has started
queue.RequestsExecuting++
} else {
// TODO(aaron-prindle) verify this statement is needed...
return nil, false
}
qs.numRequestsEnqueued--
return packet, ok
}
// Dequeue dequeues a packet from the queueSetImpl
func (qs *queueSetImpl) Dequeue() (*Request, bool) {
qs.LockAndSyncTime()
defer qs.lock.Unlock()
return qs.dequeue()
}
// isEmpty is a convenience method that returns 'true' when all of the queues
// in an queueSetImpl have no packets (and is "empty")
func (qs *queueSetImpl) isEmpty() bool {
return qs.numRequestsEnqueued == 0
}
// DequeueWithChannelAsMuchAsPossible runs a loop, as long as there
// are non-empty queues and the number currently executing is less than the
// assured concurrency value. The body of the loop uses the fair queuing
// technique to pick a queue, dequeue the request at the head of that
// queue, increment the count of the number executing, and send `{true,
// handleCompletion(that dequeued request)}` to the request's channel.
func (qs *queueSetImpl) DequeueWithChannelAsMuchAsPossible() {
for !qs.isEmpty() && qs.GetRequestsExecuting() < qs.concurrencyLimit {
_, ok := qs.dequeueWithChannel()
// TODO(aaron-prindle) verify checking ok makes senes
if !ok {
break
}
}
}
// dequeueWithChannel is convenience method for dequeueing packets that
// require a message to be sent through the packets channel
// this is a required pattern for the QueueSetSystem the queueSetImpl supports
func (qs *queueSetImpl) dequeueWithChannel() (*Request, bool) {
pkt, ok := qs.dequeue()
if !ok {
return nil, false
}
if qs.wg != nil {
qs.wg.Add(1)
}
pkt.DequeueChannel <- true
return pkt, ok
}
func (qs *queueSetImpl) roundrobinqueue() int {
// TODO(aaron-prindle) verify this is modified on quiesce...
qs.robinIdx = (qs.robinIdx + 1) % len(qs.queues)
return qs.robinIdx
}
// selectQueue selects the minimum virtualfinish time from the set of queues
// the starting queue is selected via roundrobin
// TODO(aaron-prindle) verify that the roundrobin usage is correct
// unsure if the code currently prioritizes the correct queues for ties
func (qs *queueSetImpl) selectQueue() *Queue {
minvirfinish := math.Inf(1)
var minqueue *Queue
var minidx int
for range qs.queues {
// TODO(aaron-prindle) how should this work with queue deletion?
idx := qs.roundrobinqueue()
queue := qs.queues[idx]
if len(queue.Requests) != 0 {
curvirfinish := queue.GetVirtualFinish(0, qs.estimatedServiceTime)
if curvirfinish < minvirfinish {
minvirfinish = curvirfinish
minqueue = queue
minidx = idx
}
}
}
qs.robinIdx = minidx
return minqueue
}
// AddQueues adds additional queues to the queueSetImpl
// the complementary DeleteQueues is not an explicit fxn as queue deletion requires draining
// the queues first, queue deletion is done for the proper cases
// in the the FinishRequest function
func (qs *queueSetImpl) addQueues(n int) {
for i := 0; i < n; i++ {
qs.queues = append(qs.queues, &Queue{
Requests: []*Request{},
})
}
}
// ===========================================================================
// ===========================================================================
// Quiesce controls whether this system is quiescing. Passing a
// non-nil handler means the system should become quiescent, a nil
// handler means the system should become non-quiescent. A call
// to Wait while the system is quiescent will be rebuffed by
// returning `quiescent=true`. If all the queues have no requests
// waiting nor executing while the system is quiescent then the
// handler will eventually be called with no locks held (even if
// the system becomes non-quiescent between the triggering state
// and the required call).
//
// The filter uses this for a priority level that has become
// undesired, setting a handler that will cause the priority level
// to eventually be removed from the filter if the filter still
// wants that. If the filter later changes its mind and wants to
// preserve the priority level then the filter can use this to
// cancel the handler registration.
func (qs *queueSetImpl) Quiesce(eh EmptyHandler) {
qs.lock.Lock()
defer qs.lock.Unlock()
if eh == nil {
qs.quiescent = false
return
}
// Here we check whether there are any requests queued or executing and
// if not then fork an invocation of the EmptyHandler.
if qs.numRequestsEnqueued == 0 && qs.GetRequestsExecuting() == 0 {
// fork an invocation of the EmptyHandler.
go func() {
eh.HandleEmpty()
}()
}
qs.quiescent = true
}
// Wait in the happy case, shuffle shards the given request into
// a queue and eventually dispatches the request from that queue.
// Dispatching means to return with `quiescent==false` and
// `execute==true`. In one unhappy case the request is
// immediately rebuffed with `quiescent==true` (which tells the
// filter that there has been a timing splinter and the filter
// re-calcuates the priority level to use); in all other cases
// `quiescent` will be returned `false` (even if the system is
// quiescent by then). In the non-quiescent unhappy cases the
// request is eventually rejected, which means to return with
// `execute=false`. In the happy case the caller is required to
// invoke the returned `afterExecution` after the request is done
// executing. The hash value and hand size are used to do the
// shuffle sharding.
func (qs *queueSetImpl) Wait(hashValue uint64, handSize int32) (quiescent, execute bool, afterExecution func()) {
// TODO(aaron-prindle) verify what should/shouldn't be locked!!!!
// TODO(aaron-prindle) collapse all of FQ into one layer/lock (vs 3)
// currently able to collapse to 1 impl layer and 2 locks...
qs.LockAndSyncTime()
// TODO(aaron-prindle) verify and test quiescent
// A call to Wait while the system is quiescent will be rebuffed by
// returning `quiescent=true`.
if qs.quiescent {
qs.lock.Unlock()
return true, false, func() {}
}
// ========================================================================
// Step 1:
// 1) Start with shuffle sharding, to pick a queue.
// 2) Reject old requests that have been waiting too long
// 3) Reject current request if there is not enough concurrency shares and
// we are at max queue length
// 4) If not rejected, create a packet and enqueue
pkt := qs.TimeoutOldRequestsAndRejectOrEnqueue(hashValue, handSize)
// pkt == nil means that the request was rejected - no remaining
// concurrency shares and at max queue length already
if pkt == nil {
qs.lock.Unlock()
return false, false, func() {}
}
// ========================================================================
// ------------------------------------------------------------------------
// Step 2:
// 1) The next step is to invoke the method that dequeues as much as possible.
// This method runs a loop, as long as there
// are non-empty queues and the number currently executing is less than the
// assured concurrency value. The body of the loop uses the fair queuing
// technique to pick a queue, dequeue the request at the head of that
// queue, increment the count of the number executing, and send `{true,
// handleCompletion(that dequeued request)}` to the request's channel.
qs.DequeueWithChannelAsMuchAsPossible()
// ------------------------------------------------------------------------
qs.lock.Unlock()
// ************************************************************************
// Step 3:
// After that method finishes its loop and returns, the final step in Wait
// is to `select` on either request timeout or receipt of a record on the
// newly arrived request's channel, and return appropriately. If a record
// has been sent to the request's channel then this `select` will
// immediately complete
if qs.wg != nil {
qs.wg.Done()
}
select {
case execute := <-pkt.DequeueChannel:
if execute {
// execute
return false, true, func() {
qs.FinishRequestAndDequeueWithChannelAsMuchAsPossible(pkt)
}
}
// timed out
klog.V(5).Infof("pkt.DequeueChannel timed out\n")
return false, false, func() {}
}
// ************************************************************************
}

195
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing_test.go Normal file
View File

@@ -0,0 +1,195 @@
/*
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 fairqueuing
import (
"fmt"
"math"
"math/rand"
"runtime"
"testing"
"time"
"k8s.io/apiserver/pkg/util/clock"
)
// adapted from https://github.com/tadglines/wfq/blob/master/wfq_test.go
type flowDesc struct {
// In
ftotal float64 // Total units in flow
imin float64 // Min testRequest servicetime
imax float64 // Max testRequest servicetime
// Out
idealPercent float64
actualPercent float64
}
func genFlow(fq *queueSetImpl, desc *flowDesc, key int) {
for i, t := 1, float64(0); t < desc.ftotal; i++ {
it := new(Request)
it.QueueIdx = key
it.Queue = fq.queues[key]
if desc.imin == desc.imax {
it.servicetime = desc.imax
} else {
it.servicetime = desc.imin + rand.Float64()*(desc.imax-desc.imin)
}
if float64(t)+it.servicetime > desc.ftotal {
it.servicetime = desc.ftotal - float64(t)
}
t += it.servicetime
it.seq = i
// new packet
fq.Enqueue(it)
}
}
func consumeQueue(t *testing.T, fq *queueSetImpl, descs []flowDesc) (float64, error) {
active := make(map[int]bool)
var total float64
acnt := make(map[int]float64)
cnt := make(map[int]float64)
seqs := make(map[int]int)
for it, ok := fq.Dequeue(); ok; it, ok = fq.Dequeue() {
// callback to update virtualtime w/ correct service time for request
fq.finishRequest(it)
seq := seqs[it.QueueIdx]
if seq+1 != it.seq {
return 0, fmt.Errorf("testRequest for flow %d came out of queue out-of-order: expected %d, got %d", it.QueueIdx, seq+1, it.seq)
}
seqs[it.QueueIdx] = it.seq
// set the flow this item is a part of to active
active[it.QueueIdx] = true
cnt[it.QueueIdx] += it.servicetime
// if # of active flows is equal to the # of total flows, add to total
if len(active) == len(descs) {
acnt[it.QueueIdx] += it.servicetime
total += it.servicetime
}
// if all items have been processed from the flow, remove it from active
if cnt[it.QueueIdx] == descs[it.QueueIdx].ftotal {
delete(active, it.QueueIdx)
}
}
if total == 0 {
t.Fatalf("expected 'total' to be nonzero")
}
var variance float64
for key := 0; key < len(descs); key++ {
// flows in this test have same expected # of requests
// idealPercent = total-all-active/len(flows) / total-all-active
// "how many bytes/requests you expect for this flow - all-active"
descs[key].idealPercent = float64(100) / float64(len(descs))
// actualPercent = requests-for-this-flow-all-active / total-reqs
// "how many bytes/requests you got for this flow - all-active"
descs[key].actualPercent = (acnt[key] / total) * 100
x := descs[key].idealPercent - descs[key].actualPercent
x *= x
variance += x
}
variance /= float64(len(descs))
stdDev := math.Sqrt(variance)
return stdDev, nil
}
func TestSingleFlow(t *testing.T) {
var flows = []flowDesc{
{100, 1, 1, 0, 0},
}
flowStdDevTest(t, flows, 0)
}
func TestUniformMultiFlow(t *testing.T) {
var flows = []flowDesc{
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
}
// .35 was expectedStdDev used in
// https://github.com/tadglines/wfq/blob/master/wfq_test.go
flowStdDevTest(t, flows, .041)
}
func TestOneBurstingFlow(t *testing.T) {
var flows = []flowDesc{
{1000, 1, 1, 0, 0},
{100, 1, 1, 0, 0},
}
flowStdDevTest(t, flows, 0)
}
func flowStdDevTest(t *testing.T, flows []flowDesc, expectedStdDev float64) {
runtime.GOMAXPROCS(runtime.NumCPU())
// queues := initQueues(len(flows))
// a fake clock that returns the current time is used for enqueing which
// returns the same time (now)
// this simulates all queued requests coming at the same time
now := time.Now()
fc := clock.NewFakeClock(now)
// fqqueues := make([]*Queue, len(queues), len(queues))
// for i := range queues {
// fqqueues[i] = queues[i]
// }
fq := newQueueSetImpl(20000, len(flows), 20000, 5*time.Second, fc, nil)
for n := 0; n < len(flows); n++ {
genFlow(fq, &flows[n], n)
}
// prior to dequeing, we switch to an interval clock which will simulate
// each dequeue happening at a fixed interval of time
ic := &clock.IntervalClock{
Time: now,
Duration: time.Millisecond,
}
fq.clk = ic
stdDev, err := consumeQueue(t, fq, flows)
if err != nil {
t.Fatal(err.Error())
}
if stdDev > expectedStdDev {
for k, d := range flows {
t.Logf("For flow %d: Expected %v%%, got %v%%", k, d.idealPercent, d.actualPercent)
}
t.Fatalf("StdDev was expected to be < %f but got %v", expectedStdDev, stdDev)
}
}

72
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/types.go Normal file
View File

@@ -0,0 +1,72 @@
/*
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 fairqueuing
import (
"time"
)
// Request is a temporary container for "requests" with additional tracking fields
// required for the functionality FQScheduler
type Request struct {
//TODO(aaron-prindle) seq is only used for testing, this was abstracted
// via an interface before, keeping this for now
seq int
QueueIdx int
servicetime float64
Queue *Queue
StartTime time.Time
DequeueChannel chan bool
EnqueueTime time.Time
}
// Queue is an array of packets with additional metadata required for
// the FQScheduler
type Queue struct {
Requests []*Request
VirStart float64
RequestsExecuting int
Index int
}
// Enqueue enqueues a packet into the queue
func (q *Queue) Enqueue(packet *Request) {
q.Requests = append(q.Requests, packet)
}
// Dequeue dequeues a packet from the queue
func (q *Queue) Dequeue() (*Request, bool) {
if len(q.Requests) == 0 {
return nil, false
}
packet := q.Requests[0]
q.Requests = q.Requests[1:]
return packet, true
}
// GetVirtualFinish returns the expected virtual finish time of the packet at
// index J in the queue with estimated finish time G
func (q *Queue) GetVirtualFinish(J int, G float64) float64 {
// The virtual finish time of request number J in the queue
// (counting from J=1 for the head) is J * G + (virtual start time).
// counting from J=1 for the head (eg: queue.Requests[0] -> J=1) - J+1
jg := float64(J+1) * float64(G)
return jg + q.VirStart
}