1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.util.concurrent;
17
18 import io.netty.util.internal.ThreadExecutorMap;
19 import io.netty.util.internal.logging.InternalLogger;
20 import io.netty.util.internal.logging.InternalLoggerFactory;
21
22 import java.security.AccessController;
23 import java.security.PrivilegedAction;
24 import java.util.Queue;
25 import java.util.concurrent.BlockingQueue;
26 import java.util.concurrent.Executors;
27 import java.util.concurrent.LinkedBlockingQueue;
28 import java.util.concurrent.RejectedExecutionException;
29 import java.util.concurrent.ThreadFactory;
30 import java.util.concurrent.TimeUnit;
31 import java.util.concurrent.atomic.AtomicBoolean;
32
33
34
35
36
37
38 public final class GlobalEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
39
40 private static final InternalLogger logger = InternalLoggerFactory.getInstance(GlobalEventExecutor.class);
41
42 private static final long SCHEDULE_QUIET_PERIOD_INTERVAL = TimeUnit.SECONDS.toNanos(1);
43
44 public static final GlobalEventExecutor INSTANCE = new GlobalEventExecutor();
45
46 final BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<Runnable>();
47 final ScheduledFutureTask<Void> quietPeriodTask = new ScheduledFutureTask<Void>(
48 this, Executors.<Void>callable(new Runnable() {
49 @Override
50 public void run() {
51
52 }
53 }, null), ScheduledFutureTask.deadlineNanos(SCHEDULE_QUIET_PERIOD_INTERVAL), -SCHEDULE_QUIET_PERIOD_INTERVAL);
54
55
56
57
58
59 final ThreadFactory threadFactory;
60 private final TaskRunner taskRunner = new TaskRunner();
61 private final AtomicBoolean started = new AtomicBoolean();
62 volatile Thread thread;
63
64 private final Future<?> terminationFuture = new FailedFuture<Object>(this, new UnsupportedOperationException());
65
66 private GlobalEventExecutor() {
67 scheduledTaskQueue().add(quietPeriodTask);
68 threadFactory = ThreadExecutorMap.apply(new DefaultThreadFactory(
69 DefaultThreadFactory.toPoolName(getClass()), false, Thread.NORM_PRIORITY, null), this);
70 }
71
72
73
74
75
76
77 Runnable takeTask() {
78 BlockingQueue<Runnable> taskQueue = this.taskQueue;
79 for (;;) {
80 ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
81 if (scheduledTask == null) {
82 Runnable task = null;
83 try {
84 task = taskQueue.take();
85 } catch (InterruptedException e) {
86
87 }
88 return task;
89 } else {
90 long delayNanos = scheduledTask.delayNanos();
91 Runnable task;
92 if (delayNanos > 0) {
93 try {
94 task = taskQueue.poll(delayNanos, TimeUnit.NANOSECONDS);
95 } catch (InterruptedException e) {
96
97 return null;
98 }
99 } else {
100 task = taskQueue.poll();
101 }
102
103 if (task == null) {
104 fetchFromScheduledTaskQueue();
105 task = taskQueue.poll();
106 }
107
108 if (task != null) {
109 return task;
110 }
111 }
112 }
113 }
114
115 private void fetchFromScheduledTaskQueue() {
116 long nanoTime = AbstractScheduledEventExecutor.nanoTime();
117 Runnable scheduledTask = pollScheduledTask(nanoTime);
118 while (scheduledTask != null) {
119 taskQueue.add(scheduledTask);
120 scheduledTask = pollScheduledTask(nanoTime);
121 }
122 }
123
124
125
126
127
128
129
130 public int pendingTasks() {
131 return taskQueue.size();
132 }
133
134
135
136
137
138 private void addTask(Runnable task) {
139 if (task == null) {
140 throw new NullPointerException("task");
141 }
142 taskQueue.add(task);
143 }
144
145 @Override
146 public boolean inEventLoop(Thread thread) {
147 return thread == this.thread;
148 }
149
150 @Override
151 public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
152 return terminationFuture();
153 }
154
155 @Override
156 public Future<?> terminationFuture() {
157 return terminationFuture;
158 }
159
160 @Override
161 @Deprecated
162 public void shutdown() {
163 throw new UnsupportedOperationException();
164 }
165
166 @Override
167 public boolean isShuttingDown() {
168 return false;
169 }
170
171 @Override
172 public boolean isShutdown() {
173 return false;
174 }
175
176 @Override
177 public boolean isTerminated() {
178 return false;
179 }
180
181 @Override
182 public boolean awaitTermination(long timeout, TimeUnit unit) {
183 return false;
184 }
185
186
187
188
189
190
191
192
193
194 public boolean awaitInactivity(long timeout, TimeUnit unit) throws InterruptedException {
195 if (unit == null) {
196 throw new NullPointerException("unit");
197 }
198
199 final Thread thread = this.thread;
200 if (thread == null) {
201 throw new IllegalStateException("thread was not started");
202 }
203 thread.join(unit.toMillis(timeout));
204 return !thread.isAlive();
205 }
206
207 @Override
208 public void execute(Runnable task) {
209 if (task == null) {
210 throw new NullPointerException("task");
211 }
212
213 addTask(task);
214 if (!inEventLoop()) {
215 startThread();
216 }
217 }
218
219 private void startThread() {
220 if (started.compareAndSet(false, true)) {
221 final Thread t = threadFactory.newThread(taskRunner);
222
223
224
225
226
227 AccessController.doPrivileged(new PrivilegedAction<Void>() {
228 @Override
229 public Void run() {
230 t.setContextClassLoader(null);
231 return null;
232 }
233 });
234
235
236
237
238 thread = t;
239 t.start();
240 }
241 }
242
243 final class TaskRunner implements Runnable {
244 @Override
245 public void run() {
246 for (;;) {
247 Runnable task = takeTask();
248 if (task != null) {
249 try {
250 task.run();
251 } catch (Throwable t) {
252 logger.warn("Unexpected exception from the global event executor: ", t);
253 }
254
255 if (task != quietPeriodTask) {
256 continue;
257 }
258 }
259
260 Queue<ScheduledFutureTask<?>> scheduledTaskQueue = GlobalEventExecutor.this.scheduledTaskQueue;
261
262 if (taskQueue.isEmpty() && (scheduledTaskQueue == null || scheduledTaskQueue.size() == 1)) {
263
264
265
266 boolean stopped = started.compareAndSet(true, false);
267 assert stopped;
268
269
270 if (taskQueue.isEmpty() && (scheduledTaskQueue == null || scheduledTaskQueue.size() == 1)) {
271
272
273
274
275 break;
276 }
277
278
279 if (!started.compareAndSet(false, true)) {
280
281
282 break;
283 }
284
285
286
287
288 }
289 }
290 }
291 }
292 }