all: use AbsTime.Add instead of conversion (#25417)

pull/25448/head
Seungbae Yu 2 years ago committed by GitHub
parent 9ad508018e
commit 029059947a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 4
      common/mclock/simclock.go
  2. 4
      common/prque/lazyqueue.go
  3. 2
      les/distributor.go
  4. 2
      les/flowcontrol/control.go
  5. 2
      les/utils/timeutils_test.go
  6. 8
      les/vflux/server/balance.go

@ -58,7 +58,7 @@ func (s *Simulated) Run(d time.Duration) {
s.mu.Lock() s.mu.Lock()
s.init() s.init()
end := s.now + AbsTime(d) end := s.now.Add(d)
var do []func() var do []func()
for len(s.scheduled) > 0 && s.scheduled[0].at <= end { for len(s.scheduled) > 0 && s.scheduled[0].at <= end {
ev := heap.Pop(&s.scheduled).(*simTimer) ev := heap.Pop(&s.scheduled).(*simTimer)
@ -134,7 +134,7 @@ func (s *Simulated) AfterFunc(d time.Duration, fn func()) Timer {
func (s *Simulated) schedule(d time.Duration, fn func()) *simTimer { func (s *Simulated) schedule(d time.Duration, fn func()) *simTimer {
s.init() s.init()
at := s.now + AbsTime(d) at := s.now.Add(d)
ev := &simTimer{do: fn, at: at, s: s} ev := &simTimer{do: fn, at: at, s: s}
heap.Push(&s.scheduled, ev) heap.Push(&s.scheduled, ev)
s.cond.Broadcast() s.cond.Broadcast()

@ -87,13 +87,13 @@ func (q *LazyQueue) Refresh() {
// refresh re-evaluates items in the older queue and swaps the two queues // refresh re-evaluates items in the older queue and swaps the two queues
func (q *LazyQueue) refresh(now mclock.AbsTime) { func (q *LazyQueue) refresh(now mclock.AbsTime) {
q.maxUntil = now + mclock.AbsTime(q.period) q.maxUntil = now.Add(q.period)
for q.queue[0].Len() != 0 { for q.queue[0].Len() != 0 {
q.Push(heap.Pop(q.queue[0]).(*item).value) q.Push(heap.Pop(q.queue[0]).(*item).value)
} }
q.queue[0], q.queue[1] = q.queue[1], q.queue[0] q.queue[0], q.queue[1] = q.queue[1], q.queue[0]
q.indexOffset = 1 - q.indexOffset q.indexOffset = 1 - q.indexOffset
q.maxUntil += mclock.AbsTime(q.period) q.maxUntil = q.maxUntil.Add(q.period)
} }
// Push adds an item to the queue // Push adds an item to the queue

@ -256,7 +256,7 @@ func (d *requestDistributor) queue(r *distReq) chan distPeer {
if r.reqOrder == 0 { if r.reqOrder == 0 {
d.lastReqOrder++ d.lastReqOrder++
r.reqOrder = d.lastReqOrder r.reqOrder = d.lastReqOrder
r.waitForPeers = d.clock.Now() + mclock.AbsTime(waitForPeers) r.waitForPeers = d.clock.Now().Add(waitForPeers)
} }
// Assign the timestamp when the request is queued no matter it's // Assign the timestamp when the request is queued no matter it's
// a new one or re-queued one. // a new one or re-queued one.

@ -182,7 +182,7 @@ func (node *ClientNode) UpdateParams(params ServerParams) {
return return
} }
} }
node.updateSchedule = append(node.updateSchedule, scheduledUpdate{time: now + mclock.AbsTime(DecParamDelay), params: params}) node.updateSchedule = append(node.updateSchedule, scheduledUpdate{time: now.Add(DecParamDelay), params: params})
} }
} }

@ -37,7 +37,7 @@ func TestUpdateTimer(t *testing.T) {
if updated := timer.Update(func(diff time.Duration) bool { return true }); !updated { if updated := timer.Update(func(diff time.Duration) bool { return true }); !updated {
t.Fatalf("Doesn't update the clock when reaching the threshold") t.Fatalf("Doesn't update the clock when reaching the threshold")
} }
if updated := timer.UpdateAt(sim.Now()+mclock.AbsTime(time.Second), func(diff time.Duration) bool { return true }); !updated { if updated := timer.UpdateAt(sim.Now().Add(time.Second), func(diff time.Duration) bool { return true }); !updated {
t.Fatalf("Doesn't update the clock when reaching the threshold") t.Fatalf("Doesn't update the clock when reaching the threshold")
} }
timer = NewUpdateTimer(sim, 0) timer = NewUpdateTimer(sim, 0)

@ -356,7 +356,7 @@ func (n *nodeBalance) estimatePriority(capacity uint64, addBalance int64, future
b = n.reducedBalance(b, now, future, capacity, avgReqCost) b = n.reducedBalance(b, now, future, capacity, avgReqCost)
} }
if bias > 0 { if bias > 0 {
b = n.reducedBalance(b, now+mclock.AbsTime(future), bias, capacity, 0) b = n.reducedBalance(b, now.Add(future), bias, capacity, 0)
} }
pri := n.balanceToPriority(now, b, capacity) pri := n.balanceToPriority(now, b, capacity)
// Ensure that biased estimates are always lower than actual priorities, even if // Ensure that biased estimates are always lower than actual priorities, even if
@ -512,7 +512,7 @@ func (n *nodeBalance) scheduleCheck(now mclock.AbsTime) {
n.updateAfter(0) n.updateAfter(0)
return return
} }
if n.nextUpdate == 0 || n.nextUpdate > now+mclock.AbsTime(d) { if n.nextUpdate == 0 || n.nextUpdate > now.Add(d) {
if d > time.Second { if d > time.Second {
// Note: if the scheduled update is not in the very near future then we // Note: if the scheduled update is not in the very near future then we
// schedule the update a bit earlier. This way we do need to update a few // schedule the update a bit earlier. This way we do need to update a few
@ -520,7 +520,7 @@ func (n *nodeBalance) scheduleCheck(now mclock.AbsTime) {
// brings the expected firing time a little bit closer. // brings the expected firing time a little bit closer.
d = ((d - time.Second) * 7 / 8) + time.Second d = ((d - time.Second) * 7 / 8) + time.Second
} }
n.nextUpdate = now + mclock.AbsTime(d) n.nextUpdate = now.Add(d)
n.updateAfter(d) n.updateAfter(d)
} }
} else { } else {
@ -629,7 +629,7 @@ func (n *nodeBalance) reducedBalance(b balance, start mclock.AbsTime, dt time.Du
// since the costs are applied continuously during the dt time period we calculate // since the costs are applied continuously during the dt time period we calculate
// the expiration offset at the middle of the period // the expiration offset at the middle of the period
var ( var (
at = start + mclock.AbsTime(dt/2) at = start.Add(dt / 2)
dtf = float64(dt) dtf = float64(dt)
) )
if !b.pos.IsZero() { if !b.pos.IsZero() {

Loading…
Cancel
Save