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

refactor/inline event-clock to fairqueuing package

Browse Source
This commit is contained in:
yue9944882
2019-08-30 13:12:37 +08:00
committed by Mike Spreitzer
parent ea845aa59a
commit 21439b64d1
10 changed files with 78 additions and 802 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
staging/src/k8s.io/apiserver/BUILD
View File

@@ -39,7 +39,6 @@ filegroup(
"//staging/src/k8s.io/apiserver/pkg/registry:all-srcs",
"//staging/src/k8s.io/apiserver/pkg/server:all-srcs",
"//staging/src/k8s.io/apiserver/pkg/storage:all-srcs",
"//staging/src/k8s.io/apiserver/pkg/util/clock:all-srcs",
"//staging/src/k8s.io/apiserver/pkg/util/dryrun:all-srcs",
"//staging/src/k8s.io/apiserver/pkg/util/feature:all-srcs",
"//staging/src/k8s.io/apiserver/pkg/util/flowcontrol:all-srcs",

35
staging/src/k8s.io/apiserver/pkg/util/clock/BUILD
View File

@@ -1,35 +0,0 @@
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = [
"clock.go",
"event_clock.go",
],
importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/util/clock",
importpath = "k8s.io/apiserver/pkg/util/clock",
visibility = ["//visibility:public"],
)
go_test(
name = "go_default_test",
srcs = [
"clock_test.go",
"event_clock_test.go",
],
embed = [":go_default_library"],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
visibility = ["//visibility:public"],
)

408
staging/src/k8s.io/apiserver/pkg/util/clock/clock.go
View File

@@ -1,408 +0,0 @@
/*
Copyright 2014 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 clock
import (
"sync"
"time"
)
// PassiveClock allows for injecting fake or real clocks into code
// that needs to read the current time but does not support scheduling
// activity in the future.
type PassiveClock interface {
Now() time.Time
Since(time.Time) time.Duration
}
// Clock allows for injecting fake or real clocks into code that
// needs to do arbitrary things based on time.
type Clock interface {
PassiveClock
After(time.Duration) <-chan time.Time
NewTimer(time.Duration) Timer
Sleep(time.Duration)
NewTicker(time.Duration) Ticker
}
// RealClock really calls time.Now()
type RealClock struct{}
// Now returns the current time.
func (RealClock) Now() time.Time {
return time.Now()
}
// Since returns time since the specified timestamp.
func (RealClock) Since(ts time.Time) time.Duration {
return time.Since(ts)
}
// After is the same as time.After(d).
func (RealClock) After(d time.Duration) <-chan time.Time {
return time.After(d)
}
// EventAfterDuration schedules an EventFunc
func (RealClock) EventAfterDuration(f EventFunc, d time.Duration) {
ch := time.After(d)
go func() {
select {
case t := <-ch:
f(t)
}
}()
}
// EventAfterTime schedules an EventFunc
func (r RealClock) EventAfterTime(f EventFunc, t time.Time) {
now := time.Now()
d := t.Sub(now)
if d <= 0 {
go f(now)
} else {
r.EventAfterDuration(f, d)
}
}
// NewTimer is time.NewTime
func (RealClock) NewTimer(d time.Duration) Timer {
return &realTimer{
timer: time.NewTimer(d),
}
}
// NewTicker is time.NewTicker
func (RealClock) NewTicker(d time.Duration) Ticker {
return &realTicker{
ticker: time.NewTicker(d),
}
}
// Sleep is time.Sleep
func (RealClock) Sleep(d time.Duration) {
time.Sleep(d)
}
// FakePassiveClock implements PassiveClock, but returns an arbitrary time.
type FakePassiveClock struct {
lock sync.RWMutex
time time.Time
}
// FakeClock implements Clock, but returns an arbitrary time.
type FakeClock struct {
FakePassiveClock
// waiters are waiting for the fake time to pass their specified time
waiters []fakeClockWaiter
}
type fakeClockWaiter struct {
targetTime time.Time
stepInterval time.Duration
skipIfBlocked bool
destChan chan time.Time
}
// NewFakePassiveClock creates one
func NewFakePassiveClock(t time.Time) *FakePassiveClock {
return &FakePassiveClock{
time: t,
}
}
// NewFakeClock creates one
func NewFakeClock(t time.Time) *FakeClock {
return &FakeClock{
FakePassiveClock: *NewFakePassiveClock(t),
}
}
// Now returns f's time.
func (f *FakePassiveClock) Now() time.Time {
f.lock.RLock()
defer f.lock.RUnlock()
return f.time
}
// Since returns time since the time in f.
func (f *FakePassiveClock) Since(ts time.Time) time.Duration {
f.lock.RLock()
defer f.lock.RUnlock()
return f.time.Sub(ts)
}
// After is fake version of time.After(d).
func (f *FakeClock) After(d time.Duration) <-chan time.Time {
f.lock.Lock()
defer f.lock.Unlock()
stopTime := f.time.Add(d)
ch := make(chan time.Time, 1) // Don't block!
f.waiters = append(f.waiters, fakeClockWaiter{
targetTime: stopTime,
destChan: ch,
})
return ch
}
// NewTimer is fake version of time.NewTimer(d).
func (f *FakeClock) NewTimer(d time.Duration) Timer {
f.lock.Lock()
defer f.lock.Unlock()
stopTime := f.time.Add(d)
ch := make(chan time.Time, 1) // Don't block!
timer := &fakeTimer{
fakeClock: f,
waiter: fakeClockWaiter{
targetTime: stopTime,
destChan: ch,
},
}
f.waiters = append(f.waiters, timer.waiter)
return timer
}
// NewTicker creates one
func (f *FakeClock) NewTicker(d time.Duration) Ticker {
f.lock.Lock()
defer f.lock.Unlock()
tickTime := f.time.Add(d)
ch := make(chan time.Time, 1) // hold one tick
f.waiters = append(f.waiters, fakeClockWaiter{
targetTime: tickTime,
stepInterval: d,
skipIfBlocked: true,
destChan: ch,
})
return &fakeTicker{
c: ch,
}
}
// Step moves clock by Duration, notify anyone that's called After, Tick, or NewTimer
func (f *FakeClock) Step(d time.Duration) {
f.lock.Lock()
defer f.lock.Unlock()
f.setTimeLocked(f.time.Add(d))
}
// SetTime sets the time.
func (f *FakeClock) SetTime(t time.Time) {
f.lock.Lock()
defer f.lock.Unlock()
f.setTimeLocked(t)
}
// Actually changes the time and checks any waiters. f must be write-locked.
func (f *FakeClock) setTimeLocked(t time.Time) {
f.time = t
newWaiters := make([]fakeClockWaiter, 0, len(f.waiters))
for i := range f.waiters {
w := &f.waiters[i]
if !w.targetTime.After(t) {
if w.skipIfBlocked {
select {
case w.destChan <- t:
default:
}
} else {
w.destChan <- t
}
if w.stepInterval > 0 {
for !w.targetTime.After(t) {
w.targetTime = w.targetTime.Add(w.stepInterval)
}
newWaiters = append(newWaiters, *w)
}
} else {
newWaiters = append(newWaiters, f.waiters[i])
}
}
f.waiters = newWaiters
}
// HasWaiters returns true if After has been called on f but not yet satisfied (so you can
// write race-free tests).
func (f *FakeClock) HasWaiters() bool {
f.lock.RLock()
defer f.lock.RUnlock()
return len(f.waiters) > 0
}
// Sleep advances the clock
func (f *FakeClock) Sleep(d time.Duration) {
f.Step(d)
}
// IntervalClock implements Clock, but each invocation of Now steps the clock forward the specified duration
type IntervalClock struct {
Time time.Time
Duration time.Duration
}
// Now returns i's time.
func (i *IntervalClock) Now() time.Time {
i.Time = i.Time.Add(i.Duration)
return i.Time
}
// Since returns time since the time in i.
func (i *IntervalClock) Since(ts time.Time) time.Duration {
return i.Time.Sub(ts)
}
// After is Unimplemented, will panic.
// TODO: make interval clock use FakeClock so this can be implemented.
func (*IntervalClock) After(d time.Duration) <-chan time.Time {
panic("IntervalClock doesn't implement After")
}
// NewTimer is Unimplemented, will panic.
// TODO: make interval clock use FakeClock so this can be implemented.
func (*IntervalClock) NewTimer(d time.Duration) Timer {
panic("IntervalClock doesn't implement NewTimer")
}
// NewTicker is Unimplemented, will panic.
// TODO: make interval clock use FakeClock so this can be implemented.
func (*IntervalClock) NewTicker(d time.Duration) Ticker {
panic("IntervalClock doesn't implement NewTicker")
}
// Sleep is like time.Sleep
func (*IntervalClock) Sleep(d time.Duration) {
panic("IntervalClock doesn't implement Sleep")
}
// Timer allows for injecting fake or real timers into code that
// needs to do arbitrary things based on time.
type Timer interface {
C() <-chan time.Time
Stop() bool
Reset(d time.Duration) bool
}
// realTimer is backed by an actual time.Timer.
type realTimer struct {
timer *time.Timer
}
// C returns the underlying timer's channel.
func (r *realTimer) C() <-chan time.Time {
return r.timer.C
}
// Stop calls Stop() on the underlying timer.
func (r *realTimer) Stop() bool {
return r.timer.Stop()
}
// Reset calls Reset() on the underlying timer.
func (r *realTimer) Reset(d time.Duration) bool {
return r.timer.Reset(d)
}
// fakeTimer implements Timer based on a FakeClock.
type fakeTimer struct {
fakeClock *FakeClock
waiter fakeClockWaiter
}
// C returns the channel that notifies when this timer has fired.
func (f *fakeTimer) C() <-chan time.Time {
return f.waiter.destChan
}
// Stop conditionally stops the timer. If the timer has neither fired
// nor been stopped then this call stops the timer and returns true,
// otherwise this call returns false. This is like time.Timer::Stop.
func (f *fakeTimer) Stop() bool {
f.fakeClock.lock.Lock()
defer f.fakeClock.lock.Unlock()
// The timer has already fired or been stopped, unless it is found
// among the clock's waiters.
stopped := false
oldWaiters := f.fakeClock.waiters
newWaiters := make([]fakeClockWaiter, 0, len(oldWaiters))
seekChan := f.waiter.destChan
for i := range oldWaiters {
// Identify the timer's fakeClockWaiter by the identity of the
// destination channel, nothing else is necessarily unique and
// constant since the timer's creation.
if oldWaiters[i].destChan == seekChan {
stopped = true
} else {
newWaiters = append(newWaiters, oldWaiters[i])
}
}
f.fakeClock.waiters = newWaiters
return stopped
}
// Reset conditionally updates the firing time of the timer. If the
// timer has neither fired nor been stopped then this call resets the
// timer to the fake clock's "now" + d and returns true, otherwise
// this call returns false. This is like time.Timer::Reset.
func (f *fakeTimer) Reset(d time.Duration) bool {
f.fakeClock.lock.Lock()
defer f.fakeClock.lock.Unlock()
waiters := f.fakeClock.waiters
seekChan := f.waiter.destChan
for i := range waiters {
if waiters[i].destChan == seekChan {
waiters[i].targetTime = f.fakeClock.time.Add(d)
return true
}
}
return false
}
// Ticker is for ticking implementations
type Ticker interface {
C() <-chan time.Time
Stop()
}
type realTicker struct {
ticker *time.Ticker
}
func (t *realTicker) C() <-chan time.Time {
return t.ticker.C
}
func (t *realTicker) Stop() {
t.ticker.Stop()
}
type fakeTicker struct {
c <-chan time.Time
}
func (t *fakeTicker) C() <-chan time.Time {
return t.c
}
func (t *fakeTicker) Stop() {
}

328
staging/src/k8s.io/apiserver/pkg/util/clock/clock_test.go
View File

@@ -1,328 +0,0 @@
/*
Copyright 2019 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 clock
import (
"testing"
"time"
)
var (
_ = Clock(RealClock{})
_ = Clock(&FakeClock{})
_ = Clock(&IntervalClock{})
_ = Timer(&realTimer{})
_ = Timer(&fakeTimer{})
_ = Ticker(&realTicker{})
_ = Ticker(&fakeTicker{})
)
type SettablePassiveClock interface {
PassiveClock
SetTime(time.Time)
}
func exercisePassiveClock(t *testing.T, pc SettablePassiveClock) {
t1 := time.Now()
t2 := t1.Add(time.Hour)
pc.SetTime(t1)
tx := pc.Now()
if tx != t1 {
t.Errorf("SetTime(%#+v); Now() => %#+v", t1, tx)
}
dx := pc.Since(t1)
if dx != 0 {
t.Errorf("Since() => %v", dx)
}
pc.SetTime(t2)
dx = pc.Since(t1)
if dx != time.Hour {
t.Errorf("Since() => %v", dx)
}
tx = pc.Now()
if tx != t2 {
t.Errorf("Now() => %#+v", tx)
}
}
func TestFakeClock(t *testing.T) {
startTime := time.Now()
tc := NewFakeClock(startTime)
exercisePassiveClock(t, tc)
tc.SetTime(startTime)
tc.Step(time.Second)
now := tc.Now()
if now.Sub(startTime) != time.Second {
t.Errorf("input: %s now=%s gap=%s expected=%s", startTime, now, now.Sub(startTime), time.Second)
}
tt := tc.Now()
tc.SetTime(tt.Add(time.Hour))
if tc.Since(tt) != time.Hour {
t.Errorf("input: %s now=%s gap=%s expected=%s", tt, tc.Now(), tc.Since(tt), time.Hour)
}
}
func TestFakeClockSleep(t *testing.T) {
startTime := time.Now()
tc := NewFakeClock(startTime)
tc.Sleep(time.Duration(1) * time.Hour)
now := tc.Now()
if now.Sub(startTime) != time.Hour {
t.Errorf("Fake sleep failed, expected time to advance by one hour, instead, its %v", now.Sub(startTime))
}
}
func TestFakeAfter(t *testing.T) {
tc := NewFakeClock(time.Now())
if tc.HasWaiters() {
t.Errorf("unexpected waiter?")
}
oneSec := tc.After(time.Second)
if !tc.HasWaiters() {
t.Errorf("unexpected lack of waiter?")
}
oneOhOneSec := tc.After(time.Second + time.Millisecond)
twoSec := tc.After(2 * time.Second)
select {
case <-oneSec:
t.Errorf("unexpected channel read")
case <-oneOhOneSec:
t.Errorf("unexpected channel read")
case <-twoSec:
t.Errorf("unexpected channel read")
default:
}
tc.Step(999 * time.Millisecond)
select {
case <-oneSec:
t.Errorf("unexpected channel read")
case <-oneOhOneSec:
t.Errorf("unexpected channel read")
case <-twoSec:
t.Errorf("unexpected channel read")
default:
}
tc.Step(time.Millisecond)
select {
case <-oneSec:
// Expected!
case <-oneOhOneSec:
t.Errorf("unexpected channel read")
case <-twoSec:
t.Errorf("unexpected channel read")
default:
t.Errorf("unexpected non-channel read")
}
tc.Step(time.Millisecond)
select {
case <-oneSec:
// should not double-trigger!
t.Errorf("unexpected channel read")
case <-oneOhOneSec:
// Expected!
case <-twoSec:
t.Errorf("unexpected channel read")
default:
t.Errorf("unexpected non-channel read")
}
}
func TestFakeTimer(t *testing.T) {
tc := NewFakeClock(time.Now())
if tc.HasWaiters() {
t.Errorf("unexpected waiter?")
}
oneSec := tc.NewTimer(time.Second)
twoSec := tc.NewTimer(time.Second * 2)
treSec := tc.NewTimer(time.Second * 3)
if !tc.HasWaiters() {
t.Errorf("unexpected lack of waiter?")
}
select {
case <-oneSec.C():
t.Errorf("unexpected channel read")
case <-twoSec.C():
t.Errorf("unexpected channel read")
case <-treSec.C():
t.Errorf("unexpected channel read")
default:
}
tc.Step(999999999 * time.Nanosecond) // t=.999,999,999
select {
case <-oneSec.C():
t.Errorf("unexpected channel read")
case <-twoSec.C():
t.Errorf("unexpected channel read")
case <-treSec.C():
t.Errorf("unexpected channel read")
default:
}
tc.Step(time.Nanosecond) // t=1
select {
case <-twoSec.C():
t.Errorf("unexpected channel read")
case <-treSec.C():
t.Errorf("unexpected channel read")
default:
}
select {
case <-oneSec.C():
// Expected!
default:
t.Errorf("unexpected channel non-read")
}
tc.Step(time.Nanosecond) // t=1.000,000,001
select {
case <-oneSec.C():
t.Errorf("unexpected channel read")
case <-twoSec.C():
t.Errorf("unexpected channel read")
case <-treSec.C():
t.Errorf("unexpected channel read")
default:
}
if oneSec.Stop() {
t.Errorf("Expected oneSec.Stop() to return false")
}
if !twoSec.Stop() {
t.Errorf("Expected twoSec.Stop() to return true")
}
tc.Step(time.Second) // t=2.000,000,001
select {
case <-oneSec.C():
t.Errorf("unexpected channel read")
case <-twoSec.C():
t.Errorf("unexpected channel read")
case <-treSec.C():
t.Errorf("unexpected channel read")
default:
}
if twoSec.Reset(time.Second) {
t.Errorf("Expected twoSec.Reset() to return false")
}
if !treSec.Reset(time.Second) {
t.Errorf("Expected treSec.Reset() to return true")
}
tc.Step(time.Nanosecond * 999999999) // t=3.0
select {
case <-oneSec.C():
t.Errorf("unexpected channel read")
case <-twoSec.C():
t.Errorf("unexpected channel read")
case <-treSec.C():
t.Errorf("unexpected channel read")
default:
}
tc.Step(time.Nanosecond) // t=3.000,000,001
select {
case <-oneSec.C():
t.Errorf("unexpected channel read")
case <-twoSec.C():
t.Errorf("unexpected channel read")
default:
}
select {
case <-treSec.C():
// Expected!
default:
t.Errorf("unexpected channel non-read")
}
}
func TestFakeTick(t *testing.T) {
tc := NewFakeClock(time.Now())
if tc.HasWaiters() {
t.Errorf("unexpected waiter?")
}
oneSec := tc.NewTicker(time.Second).C()
if !tc.HasWaiters() {
t.Errorf("unexpected lack of waiter?")
}
oneOhOneSec := tc.NewTicker(time.Second + time.Millisecond).C()
twoSec := tc.NewTicker(2 * time.Second).C()
select {
case <-oneSec:
t.Errorf("unexpected channel read")
case <-oneOhOneSec:
t.Errorf("unexpected channel read")
case <-twoSec:
t.Errorf("unexpected channel read")
default:
}
tc.Step(999 * time.Millisecond) // t=.999
select {
case <-oneSec:
t.Errorf("unexpected channel read")
case <-oneOhOneSec:
t.Errorf("unexpected channel read")
case <-twoSec:
t.Errorf("unexpected channel read")
default:
}
tc.Step(time.Millisecond) // t=1.000
select {
case <-oneSec:
// Expected!
case <-oneOhOneSec:
t.Errorf("unexpected channel read")
case <-twoSec:
t.Errorf("unexpected channel read")
default:
t.Errorf("unexpected non-channel read")
}
tc.Step(time.Millisecond) // t=1.001
select {
case <-oneSec:
// should not double-trigger!
t.Errorf("unexpected channel read")
case <-oneOhOneSec:
// Expected!
case <-twoSec:
t.Errorf("unexpected channel read")
default:
t.Errorf("unexpected non-channel read")
}
tc.Step(time.Second) // t=2.001
tc.Step(time.Second) // t=3.001
tc.Step(time.Second) // t=4.001
tc.Step(time.Second) // t=5.001
// The one second ticker should not accumulate ticks
accumulatedTicks := 0
drained := false
for !drained {
select {
case <-oneSec:
accumulatedTicks++
default:
drained = true
}
}
if accumulatedTicks != 1 {
t.Errorf("unexpected number of accumulated ticks: %d", accumulatedTicks)
}
}

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

@@ -4,6 +4,7 @@ go_library(
name = "go_default_library",
srcs = [
"dummy.go",
"event_clock.go",
"fairqueuing.go",
"integrator.go",
"interface.go",
@@ -14,8 +15,8 @@ go_library(
importpath = "k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing",
visibility = ["//visibility:public"],
deps = [
"//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/waitgroup:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/clock:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/shufflesharding:go_default_library",
"//vendor/k8s.io/klog:go_default_library",
@@ -25,6 +26,7 @@ go_library(
go_test(
name = "go_default_test",
srcs = [
"event_clock_test.go",
"fairqueuing_test.go",
"fq_test.go",
],
@@ -32,7 +34,6 @@ go_test(
deps = [
"//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/waitgroup:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/clock:go_default_library",
],
)

83
staging/src/k8s.io/apiserver/pkg/util/clock/event_clock.go → staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock.go
View File

@@ -14,13 +14,16 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package clock
package fairqueuing
import (
"container/heap"
"math/rand"
"sync"
"testing"
"time"
"k8s.io/apimachinery/pkg/util/clock"
)
// EventFunc does some work that needs to be done at or after the
@@ -30,10 +33,47 @@ import (
// no other work is left to be completed in goroutines.
type EventFunc func(time.Time)
type SettablePassiveClock interface {
clock.PassiveClock
SetTime(time.Time)
}
type EventClock interface {
clock.PassiveClock
EventAfterDuration(f EventFunc, d time.Duration)
EventAfterTime(f EventFunc, t time.Time)
}
type RealEventClock struct {
clock.RealClock
}
// EventAfterDuration schedules an EventFunc
func (RealEventClock) EventAfterDuration(f EventFunc, d time.Duration) {
ch := time.After(d)
go func() {
select {
case t := <-ch:
f(t)
}
}()
}
// EventAfterTime schedules an EventFunc
func (r RealEventClock) EventAfterTime(f EventFunc, t time.Time) {
now := time.Now()
d := t.Sub(now)
if d <= 0 {
go f(now)
} else {
r.EventAfterDuration(f, d)
}
}
// FakeEventClock is one whose time does not pass implicitly but
// rather is explicitly set by invocations of its SetTime method
type FakeEventClock struct {
FakePassiveClock
clock.FakePassiveClock
// waiters is a heap of waiting work, sorted by time
waiters eventWaiterHeap
@@ -75,7 +115,7 @@ func NewFakeEventClock(t time.Time, clientWG *sync.WaitGroup, fuzz time.Duration
r.Uint64()
}
return &FakeEventClock{
FakePassiveClock: *NewFakePassiveClock(t),
FakePassiveClock: *clock.NewFakePassiveClock(t),
clientWG: clientWG,
fuzz: fuzz,
rand: r,
@@ -112,13 +152,12 @@ func (fec *FakeEventClock) Run(limit *time.Time) {
// be started by the given time --- including any further events they
// schedule
func (fec *FakeEventClock) SetTime(t time.Time) {
fec.lock.Lock()
fec.time = t
fec.FakePassiveClock.SetTime(t)
for {
// This loop is because events run at a given time may schedule more
// events to run at that or an earlier time.
// Events should not advance the clock. But just in case they do...
now := fec.time
now := fec.Now()
var wg sync.WaitGroup
foundSome := false
for len(fec.waiters) > 0 && !now.Before(fec.waiters[0].targetTime) {
@@ -130,19 +169,14 @@ func (fec *FakeEventClock) SetTime(t time.Time) {
if !foundSome {
break
}
fec.lock.Unlock()
wg.Wait()
fec.lock.Lock()
}
fec.lock.Unlock()
}
// EventAfterDuration schedules the given function to be invoked once
// the given duration has passed.
func (fec *FakeEventClock) EventAfterDuration(f EventFunc, d time.Duration) {
fec.lock.Lock()
defer fec.lock.Unlock()
now := fec.time
now := fec.Now()
fd := time.Duration(float32(fec.fuzz) * fec.rand.Float32())
heap.Push(&fec.waiters, eventWaiter{targetTime: now.Add(d + fd), f: f})
}
@@ -150,8 +184,6 @@ func (fec *FakeEventClock) EventAfterDuration(f EventFunc, d time.Duration) {
// EventAfterTime schedules the given function to be invoked once
// the given time has arrived.
func (fec *FakeEventClock) EventAfterTime(f EventFunc, t time.Time) {
fec.lock.Lock()
defer fec.lock.Unlock()
fd := time.Duration(float32(fec.fuzz) * fec.rand.Float32())
heap.Push(&fec.waiters, eventWaiter{targetTime: t.Add(fd), f: f})
}
@@ -173,3 +205,26 @@ func (ewh *eventWaiterHeap) Pop() interface{} {
*ewh = old[:n-1]
return x
}
func exercisePassiveClock(t *testing.T, pc SettablePassiveClock) {
t1 := time.Now()
t2 := t1.Add(time.Hour)
pc.SetTime(t1)
tx := pc.Now()
if tx != t1 {
t.Errorf("SetTime(%#+v); Now() => %#+v", t1, tx)
}
dx := pc.Since(t1)
if dx != 0 {
t.Errorf("Since() => %v", dx)
}
pc.SetTime(t2)
dx = pc.Since(t1)
if dx != time.Hour {
t.Errorf("Since() => %v", dx)
}
tx = pc.Now()
if tx != t2 {
t.Errorf("Now() => %#+v", tx)
}
}

10
staging/src/k8s.io/apiserver/pkg/util/clock/event_clock_test.go → staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock_test.go
View File

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package clock
package fairqueuing
import (
"math/rand"
@@ -23,12 +23,6 @@ import (
"time"
)
type EventClock interface {
PassiveClock
EventAfterDuration(f EventFunc, d time.Duration)
EventAfterTime(f EventFunc, t time.Time)
}
type TestableEventClock interface {
EventClock
SetTime(time.Time)
@@ -154,5 +148,5 @@ func exerciseEventClock(t *testing.T, ec EventClock, relax func(time.Duration))
}
func TestRealEventClock(t *testing.T) {
exerciseEventClock(t, RealClock{}, func(d time.Duration) { time.Sleep(d) })
exerciseEventClock(t, RealEventClock{}, func(d time.Duration) { time.Sleep(d) })
}

2
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go
View File

@@ -21,8 +21,8 @@ import (
"sync"
"time"
"k8s.io/apimachinery/pkg/util/clock"
"k8s.io/apimachinery/pkg/util/waitgroup"
"k8s.io/apiserver/pkg/util/clock"
"k8s.io/apiserver/pkg/util/flowcontrol/metrics"
"k8s.io/apiserver/pkg/util/shufflesharding"
"k8s.io/klog"

6
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fq_test.go
View File

@@ -21,8 +21,6 @@ import (
"sync"
"testing"
"time"
"k8s.io/apiserver/pkg/util/clock"
)
type uniformScenario []uniformClient
@@ -41,7 +39,7 @@ type uniformClient struct {
func exerciseQueueSetUniformScenario(t *testing.T, qs QueueSet, sc uniformScenario, handSize int32, totalDuration time.Duration, expectPass bool) {
wg := new(sync.WaitGroup)
now := time.Now()
clk := clock.NewFakeEventClock(now, wg, 0, nil)
clk := NewFakeEventClock(now, wg, 0, nil)
t.Logf("%s: Start", clk.Now().Format("2006-01-02 15:04:05.000000000"))
integrators := make([]Integrator, len(sc))
for i, uc := range sc {
@@ -106,7 +104,7 @@ func TestDummy(t *testing.T) {
}, 1, time.Second*10, false)
}
func ClockWait(clk *clock.FakeEventClock, wg *sync.WaitGroup, duration time.Duration) {
func ClockWait(clk *FakeEventClock, wg *sync.WaitGroup, duration time.Duration) {
dunch := make(chan struct{})
clk.EventAfterDuration(func(time.Time) {
wg.Add(1)

2
staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/integrator.go
View File

@@ -21,7 +21,7 @@ import (
"sync"
"time"
"k8s.io/apiserver/pkg/util/clock"
"k8s.io/apimachinery/pkg/util/clock"
)
// Integrator computes the integral of some variable X over time as