diff --git a/staging/src/k8s.io/apiserver/BUILD b/staging/src/k8s.io/apiserver/BUILD index 77aa4df6c7b..f49345d725c 100644 --- a/staging/src/k8s.io/apiserver/BUILD +++ b/staging/src/k8s.io/apiserver/BUILD @@ -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", diff --git a/staging/src/k8s.io/apiserver/pkg/util/clock/BUILD b/staging/src/k8s.io/apiserver/pkg/util/clock/BUILD deleted file mode 100644 index 27f854d70c5..00000000000 --- a/staging/src/k8s.io/apiserver/pkg/util/clock/BUILD +++ /dev/null @@ -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"], -) diff --git a/staging/src/k8s.io/apiserver/pkg/util/clock/clock.go b/staging/src/k8s.io/apiserver/pkg/util/clock/clock.go deleted file mode 100644 index 710c7cfbd92..00000000000 --- a/staging/src/k8s.io/apiserver/pkg/util/clock/clock.go +++ /dev/null @@ -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() { -} diff --git a/staging/src/k8s.io/apiserver/pkg/util/clock/clock_test.go b/staging/src/k8s.io/apiserver/pkg/util/clock/clock_test.go deleted file mode 100644 index a4eab1b1426..00000000000 --- a/staging/src/k8s.io/apiserver/pkg/util/clock/clock_test.go +++ /dev/null @@ -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) - } -} diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD index 1e07452de35..56d603af543 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/BUILD @@ -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", ], ) diff --git a/staging/src/k8s.io/apiserver/pkg/util/clock/event_clock.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock.go similarity index 79% rename from staging/src/k8s.io/apiserver/pkg/util/clock/event_clock.go rename to staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock.go index 122e43b1103..f98114e3163 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/clock/event_clock.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock.go @@ -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) + } +} diff --git a/staging/src/k8s.io/apiserver/pkg/util/clock/event_clock_test.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock_test.go similarity index 94% rename from staging/src/k8s.io/apiserver/pkg/util/clock/event_clock_test.go rename to staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock_test.go index ce9f9f97a92..a35173a3b08 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/clock/event_clock_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/event_clock_test.go @@ -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) }) } diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go index aa7e726ff4b..b48448de34e 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fairqueuing.go @@ -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" diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fq_test.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fq_test.go index a5e22c2b3b4..964328cbfcc 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fq_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/fq_test.go @@ -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) diff --git a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/integrator.go b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/integrator.go index 2d5b2bfbee4..f05b65489db 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/integrator.go +++ b/staging/src/k8s.io/apiserver/pkg/util/flowcontrol/fairqueuing/integrator.go @@ -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