1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.documentapi.messagebus;
import com.yahoo.concurrent.SystemTimer;
import com.yahoo.concurrent.Timer;
import java.util.*;
import java.util.concurrent.RejectedExecutionException;
/**
* Simple, lightweight event scheduler that does not maintain any executor
* threads of its own, but rather makes it the responsibility of the caller
* to run the events as the queue hands them over.
*
* Fully thread safe for multiple readers and writers.
*/
public class ScheduledEventQueue {
private final Set<Entry> tasks = new TreeSet<Entry>();
private long sequenceCounter = 0;
private Timer timer;
private volatile boolean waiting = false;
private volatile boolean shutdown = false;
private static class Entry implements Comparable<Entry> {
private Runnable task;
private long timestamp;
private long sequenceId;
public Entry(Runnable task, long timestamp, long sequenceId) {
this.task = task;
this.timestamp = timestamp;
this.sequenceId = sequenceId;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Entry entry = (Entry) o;
if (sequenceId != entry.sequenceId) return false;
if (timestamp != entry.timestamp) return false;
if (!task.equals(entry.task)) return false;
return true;
}
@Override
public int hashCode() {
return java.util.Objects.hash(sequenceId, timestamp, task);
}
@Override
public int compareTo(Entry o) {
if (timestamp < o.timestamp) return -1;
if (timestamp > o.timestamp) return 1;
if (sequenceId < o.sequenceId) return -1;
if (sequenceId > o.sequenceId) return 1;
return 0;
}
public Runnable getTask() {
return task;
}
public long getTimestamp() {
return timestamp;
}
public long getSequenceId() {
return sequenceId;
}
}
public ScheduledEventQueue() {
this.timer = SystemTimer.INSTANCE;
}
public ScheduledEventQueue(Timer timer) {
this.timer = timer;
}
public void pushTask(Runnable task) {
synchronized (tasks) {
if (shutdown) {
throw new RejectedExecutionException("Tasks can't be scheduled since queue has been shut down.");
}
tasks.add(new Entry(task, 0, sequenceCounter++));
tasks.notifyAll();
}
}
public void pushTask(Runnable task, long milliSecondsToWait) {
synchronized (tasks) {
if (shutdown) {
throw new RejectedExecutionException("Tasks can't be scheduled since queue has been shut down.");
}
tasks.add(new Entry(task, timer.milliTime() + milliSecondsToWait, sequenceCounter++));
tasks.notifyAll();
}
}
public boolean isWaiting() {
synchronized (tasks) {
return waiting;
}
}
/**
* Waits until the queue has a task that is ready for scheduling, removes that
* task from the queue and returns it.
*
* @return The next task.
*/
public Runnable getNextTask() {
try {
while (true) {
synchronized (tasks) {
Iterator<Entry> iter = tasks.iterator();
if (!iter.hasNext()) {
if (shutdown) {
return null;
}
// Set flag for unit tests to coordinate with.
waiting = true;
tasks.wait();
waiting = false;
continue;
}
Entry retEntry = iter.next();
long timeNow = timer.milliTime();
if (retEntry.getTimestamp() > timeNow) {
waiting = true;
tasks.wait(retEntry.getTimestamp() - timeNow);
waiting = false;
continue;
}
iter.remove();
return retEntry.getTask();
}
}
} catch (InterruptedException e) {
return null;
}
}
/**
* If there is a task ready for scheduling, remove it from the queue and return it.
*
* @return The next task.
*/
public Runnable popTask() {
synchronized (tasks) {
Iterator<Entry> iter = tasks.iterator();
if (!iter.hasNext()) {
return null;
}
Entry retEntry = iter.next();
if (retEntry.getTimestamp() > timer.milliTime()) {
return null;
}
iter.remove();
return retEntry.getTask();
}
}
/** For unit testing only */
public void wakeTasks() {
synchronized (tasks) {
tasks.notifyAll();
}
}
public void shutdown() {
synchronized (tasks) {
shutdown = true;
tasks.notifyAll();
}
}
public boolean isShutdown() {
synchronized (tasks) {
return shutdown;
}
}
public long size() {
synchronized (tasks) {
return tasks.size();
}
}
}
|
