Skip to content

Commit 1068ebe

Browse files
ninthanintha
committed
feat: add time wheel
1 parent 69deca1 commit 1068ebe

3 files changed

Lines changed: 223 additions & 10 deletions

File tree

‎base/src/main/java/vproxy/base/util/time/impl/TimeElemImpl.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@ public T get() {
1818
return elem;
1919
}
2020

21+
@Override
2122
public void removeSelf() {
22-
queue.queue.remove(this);
23+
queue.remove(this);
2324
}
2425
}

‎base/src/main/java/vproxy/base/util/time/impl/TimeQueueImpl.java‎

Lines changed: 102 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,37 +3,130 @@
33
import vproxy.base.util.time.TimeElem;
44
import vproxy.base.util.time.TimeQueue;
55

6-
import java.util.PriorityQueue;
6+
import java.util.*;
77

88
public class TimeQueueImpl<T> implements TimeQueue<T> {
9-
PriorityQueue<TimeElemImpl<T>> queue = new PriorityQueue<>((a, b) -> (int) (a.triggerTime - b.triggerTime));
9+
private static final int TIME_WHEEL_LEVEL = 4;
10+
private static final int MAX_TIME_WHEEL_INTERVAL = 1 << (TIME_WHEEL_LEVEL * TimeWheel.WHEEL_SIZE_POWER);
11+
12+
private final PriorityQueue<TimeElemImpl<T>> queue = new PriorityQueue<>(Comparator.comparingLong(x -> x.triggerTime));
13+
14+
private final ArrayList<TimeWheel<T>> timeWheels;
15+
16+
private long lastTickTimestamp;
17+
18+
public TimeQueueImpl() {
19+
this(System.currentTimeMillis());
20+
}
21+
22+
public TimeQueueImpl(long currentTimestamp) {
23+
this.timeWheels = new ArrayList<>(TIME_WHEEL_LEVEL);
24+
for (int i = 0; i < TIME_WHEEL_LEVEL; i++) {
25+
this.timeWheels.add(new TimeWheel<>(1 << (i * TimeWheel.WHEEL_SIZE_POWER), currentTimestamp));
26+
}
27+
this.lastTickTimestamp = currentTimestamp;
28+
}
1029

1130
@Override
1231
public TimeElem<T> add(long currentTimestamp, int timeout, T elem) {
13-
TimeElemImpl<T> event = new TimeElemImpl<>(currentTimestamp + timeout, elem, this);
14-
queue.add(event);
32+
final TimeElemImpl<T> event = new TimeElemImpl<>(currentTimestamp + timeout, elem, this);
33+
addTimeElem(event, currentTimestamp);
1534
return event;
1635
}
1736

37+
private void addTimeElem(TimeElemImpl<T> event, long currentTimestamp) {
38+
long timeout = event.triggerTime - currentTimestamp;
39+
if (timeout >= MAX_TIME_WHEEL_INTERVAL) {
40+
queue.add(event);
41+
}
42+
// already timeout task put into the lowest time wheel
43+
else if (timeout <= 0) {
44+
this.timeWheels.get(0).add(event, currentTimestamp);
45+
}
46+
// long timeout task put into queue
47+
else {
48+
var index = findTimeWheelIndex(timeout);
49+
this.timeWheels.get(index).add(event, currentTimestamp);
50+
}
51+
}
52+
1853
@Override
1954
public T poll() {
20-
TimeElemImpl<T> elem = queue.poll();
21-
if (elem == null)
55+
TimeElem<T> elem = timeWheels.get(0).poll();
56+
if (elem == null) {
2257
return null;
23-
return elem.elem;
58+
}
59+
return elem.get();
2460
}
2561

2662
@Override
2763
public boolean isEmpty() {
28-
return queue.isEmpty();
64+
for (TimeWheel<T> timeWheel : timeWheels) {
65+
if (!timeWheel.isEmpty()) {
66+
return false;
67+
}
68+
}
69+
return true;
2970
}
3071

3172
@Override
3273
public int nextTime(long currentTimestamp) {
74+
tickTimeWheel(currentTimestamp);
75+
for (TimeWheel<T> timeWheel : timeWheels) {
76+
if (timeWheel.isEmpty()) {
77+
continue;
78+
}
79+
return timeWheel.nextTime(currentTimestamp);
80+
}
81+
3382
TimeElemImpl<T> elem = queue.peek();
34-
if (elem == null)
83+
if (elem == null) {
3584
return Integer.MAX_VALUE;
85+
}
3686
long triggerTime = elem.triggerTime;
3787
return Math.max((int) (triggerTime - currentTimestamp), 0);
3888
}
89+
90+
private void tickTimeWheel(long currentTimestamp) {
91+
for (int i = TIME_WHEEL_LEVEL - 1; i > 0; i--) {
92+
final var wheel = timeWheels.get(i);
93+
while (wheel.tryTick(currentTimestamp)) {
94+
final Collection<TimeElemImpl<T>> events = wheel.tick(currentTimestamp);
95+
for (TimeElemImpl<T> event : events) {
96+
addTimeElem(event, currentTimestamp);
97+
}
98+
}
99+
}
100+
101+
// move elements from queue to time wheels
102+
while (!queue.isEmpty()) {
103+
final TimeElemImpl<T> elem = queue.peek();
104+
long timeout = elem.triggerTime - currentTimestamp;
105+
if (timeout >= MAX_TIME_WHEEL_INTERVAL) {
106+
break;
107+
}
108+
109+
addTimeElem(elem, currentTimestamp);
110+
queue.poll();
111+
}
112+
113+
this.lastTickTimestamp = currentTimestamp;
114+
}
115+
116+
public void remove(TimeElemImpl<T> elem) {
117+
long timeout = elem.triggerTime - this.lastTickTimestamp;
118+
if (timeout >= MAX_TIME_WHEEL_INTERVAL) {
119+
queue.remove(elem);
120+
} else {
121+
timeWheels.get(findTimeWheelIndex(timeout)).remove(elem);
122+
}
123+
}
124+
125+
private static int findTimeWheelIndex(long timeout) {
126+
if (timeout <= 0) {
127+
return 0;
128+
}
129+
int bits = 63 - Long.numberOfLeadingZeros(timeout);
130+
return bits / TimeWheel.WHEEL_SIZE_POWER;
131+
}
39132
}

‎base/src/main/java/vproxy/base/util/time/impl/TimeWheel.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package vproxy.base.util.time.impl;
2+
3+
import vproxy.base.util.time.TimeElem;
4+
5+
import java.util.*;
6+
7+
public class TimeWheel<T> {
8+
public static final int WHEEL_SIZE_POWER = 5;
9+
public static final int WHEEL_SIZE = 1 << WHEEL_SIZE_POWER;
10+
11+
private final PriorityQueue<TimeElemImpl<T>>[] slots = new PriorityQueue[WHEEL_SIZE];
12+
/**
13+
* min time unit in this time wheel
14+
*/
15+
public final long tickDuration;
16+
/**
17+
* the time wheel max time interval. interval = tickDuration * WHEEL_SIZE
18+
*/
19+
public final long interval;
20+
public final long startTimestamp;
21+
private int tickIndex;
22+
private long elemNum;
23+
private long currentTime;
24+
25+
public TimeWheel(long tickDuration, long timestamp) {
26+
this.tickDuration = tickDuration;
27+
this.interval = this.tickDuration * WHEEL_SIZE;
28+
this.startTimestamp = timestamp;
29+
this.currentTime = timestamp;
30+
this.tickIndex = 0;
31+
this.elemNum = 0;
32+
33+
for (int i = 0; i < slots.length; i++) {
34+
slots[i] = new PriorityQueue<>(Comparator.comparingLong(x -> x.triggerTime));
35+
}
36+
}
37+
38+
public void add(TimeElemImpl<T> elem, long timestamp) {
39+
if (elem.triggerTime <= timestamp) {
40+
slots[tickIndex].add(elem);
41+
} else {
42+
slots[findSlotIndex(elem.triggerTime)].add(elem);
43+
}
44+
elemNum++;
45+
}
46+
47+
private int findSlotIndex(long timestamp) {
48+
long timeout = timestamp - startTimestamp;
49+
return (int) ((timeout & (interval - 1)) / tickDuration);
50+
}
51+
52+
/**
53+
* return true if it can move.
54+
*/
55+
public boolean tryTick(long timestamp) {
56+
return timestamp - currentTime >= tickDuration;
57+
}
58+
59+
/**
60+
* move the tick index to point the next slot.
61+
*/
62+
public Collection<TimeElemImpl<T>> tick(long timestamp) {
63+
if (!tryTick(timestamp)) {
64+
return Collections.emptyList();
65+
}
66+
67+
int oldIndex = tickIndex;
68+
int nextIndex = (oldIndex + 1) & (WHEEL_SIZE - 1);
69+
if (!slots[oldIndex].isEmpty()) {
70+
slots[nextIndex].addAll(slots[oldIndex]);
71+
}
72+
this.tickIndex = nextIndex;
73+
final PriorityQueue<TimeElemImpl<T>> queue = slots[tickIndex];
74+
slots[tickIndex] = new PriorityQueue<>(Comparator.comparingLong(x -> x.triggerTime));
75+
76+
elemNum -= queue.size();
77+
currentTime += tickDuration;
78+
return queue;
79+
}
80+
81+
public TimeElem<T> poll() {
82+
var elem = slots[tickIndex].poll();
83+
if (elem != null) {
84+
elemNum--;
85+
}
86+
return elem;
87+
}
88+
89+
public boolean isEmpty() {
90+
return elemNum == 0;
91+
}
92+
93+
public long size() {
94+
return elemNum;
95+
}
96+
97+
public int nextTime(long timestamp) {
98+
for (int i = tickIndex; i < tickIndex + WHEEL_SIZE; i++) {
99+
final int index = i & (WHEEL_SIZE - 1);
100+
if (slots[index].isEmpty()) {
101+
continue;
102+
}
103+
104+
long triggerTime = slots[index].peek().triggerTime;
105+
int nextTime = Math.max((int) (triggerTime - timestamp), 0);
106+
if (nextTime == 0 && index != tickIndex){
107+
slots[tickIndex].add(slots[index].poll());
108+
}
109+
return nextTime;
110+
}
111+
return Integer.MAX_VALUE;
112+
}
113+
114+
public void remove(TimeElemImpl<T> elem) {
115+
if (slots[findSlotIndex(elem.triggerTime)].remove(elem)) {
116+
elemNum--;
117+
}
118+
}
119+
}

0 commit comments

Comments
 (0)