View Javadoc
1   /*
2    * Copyright 2012 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package io.netty.util.concurrent;
17  
18  import io.netty.util.internal.logging.InternalLogger;
19  import io.netty.util.internal.logging.InternalLoggerFactory;
20  
21  import java.security.AccessController;
22  import java.security.PrivilegedAction;
23  import java.util.Queue;
24  import java.util.concurrent.BlockingQueue;
25  import java.util.concurrent.Executors;
26  import java.util.concurrent.LinkedBlockingQueue;
27  import java.util.concurrent.RejectedExecutionException;
28  import java.util.concurrent.ThreadFactory;
29  import java.util.concurrent.TimeUnit;
30  import java.util.concurrent.atomic.AtomicBoolean;
31  
32  /**
33   * Single-thread singleton {@link EventExecutor}.  It starts the thread automatically and stops it when there is no
34   * task pending in the task queue for 1 second.  Please note it is not scalable to schedule large number of tasks to
35   * this executor; use a dedicated executor.
36   */
37  public final class GlobalEventExecutor extends AbstractScheduledEventExecutor {
38  
39      private static final InternalLogger logger = InternalLoggerFactory.getInstance(GlobalEventExecutor.class);
40  
41      private static final long SCHEDULE_QUIET_PERIOD_INTERVAL = TimeUnit.SECONDS.toNanos(1);
42  
43      public static final GlobalEventExecutor INSTANCE = new GlobalEventExecutor();
44  
45      final BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<Runnable>();
46      final ScheduledFutureTask<Void> quietPeriodTask = new ScheduledFutureTask<Void>(
47              this, Executors.<Void>callable(new Runnable() {
48          @Override
49          public void run() {
50              // NOOP
51          }
52      }, null), ScheduledFutureTask.deadlineNanos(SCHEDULE_QUIET_PERIOD_INTERVAL), -SCHEDULE_QUIET_PERIOD_INTERVAL);
53  
54      // because the GlobalEventExecutor is a singleton, tasks submitted to it can come from arbitrary threads and this
55      // can trigger the creation of a thread from arbitrary thread groups; for this reason, the thread factory must not
56      // be sticky about its thread group
57      // visible for testing
58      final ThreadFactory threadFactory =
59              new DefaultThreadFactory(DefaultThreadFactory.toPoolName(getClass()), false, Thread.NORM_PRIORITY, null);
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      }
69  
70      @Override
71      public EventExecutorGroup parent() {
72          return null;
73      }
74  
75      /**
76       * Take the next {@link Runnable} from the task queue and so will block if no task is currently present.
77       *
78       * @return {@code null} if the executor thread has been interrupted or waken up.
79       */
80      Runnable takeTask() {
81          BlockingQueue<Runnable> taskQueue = this.taskQueue;
82          for (;;) {
83              ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
84              if (scheduledTask == null) {
85                  Runnable task = null;
86                  try {
87                      task = taskQueue.take();
88                  } catch (InterruptedException e) {
89                      // Ignore
90                  }
91                  return task;
92              } else {
93                  long delayNanos = scheduledTask.delayNanos();
94                  Runnable task;
95                  if (delayNanos > 0) {
96                      try {
97                          task = taskQueue.poll(delayNanos, TimeUnit.NANOSECONDS);
98                      } catch (InterruptedException e) {
99                          return null;
100                     }
101                 } else {
102                     task = taskQueue.poll();
103                 }
104 
105                 if (task == null) {
106                     fetchFromScheduledTaskQueue();
107                     task = taskQueue.poll();
108                 }
109 
110                 if (task != null) {
111                     return task;
112                 }
113             }
114         }
115     }
116 
117     private void fetchFromScheduledTaskQueue() {
118         long nanoTime = AbstractScheduledEventExecutor.nanoTime();
119         Runnable scheduledTask = pollScheduledTask(nanoTime);
120         while (scheduledTask != null) {
121             taskQueue.add(scheduledTask);
122             scheduledTask = pollScheduledTask(nanoTime);
123         }
124     }
125 
126     /**
127      * Return the number of tasks that are pending for processing.
128      *
129      * <strong>Be aware that this operation may be expensive as it depends on the internal implementation of the
130      * SingleThreadEventExecutor. So use it was care!</strong>
131      */
132     public int pendingTasks() {
133         return taskQueue.size();
134     }
135 
136     /**
137      * Add a task to the task queue, or throws a {@link RejectedExecutionException} if this instance was shutdown
138      * before.
139      */
140     private void addTask(Runnable task) {
141         if (task == null) {
142             throw new NullPointerException("task");
143         }
144         taskQueue.add(task);
145     }
146 
147     @Override
148     public boolean inEventLoop(Thread thread) {
149         return thread == this.thread;
150     }
151 
152     @Override
153     public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
154         return terminationFuture();
155     }
156 
157     @Override
158     public Future<?> terminationFuture() {
159         return terminationFuture;
160     }
161 
162     @Override
163     @Deprecated
164     public void shutdown() {
165         throw new UnsupportedOperationException();
166     }
167 
168     @Override
169     public boolean isShuttingDown() {
170         return false;
171     }
172 
173     @Override
174     public boolean isShutdown() {
175         return false;
176     }
177 
178     @Override
179     public boolean isTerminated() {
180         return false;
181     }
182 
183     @Override
184     public boolean awaitTermination(long timeout, TimeUnit unit) {
185         return false;
186     }
187 
188     /**
189      * Waits until the worker thread of this executor has no tasks left in its task queue and terminates itself.
190      * Because a new worker thread will be started again when a new task is submitted, this operation is only useful
191      * when you want to ensure that the worker thread is terminated <strong>after</strong> your application is shut
192      * down and there's no chance of submitting a new task afterwards.
193      *
194      * @return {@code true} if and only if the worker thread has been terminated
195      */
196     public boolean awaitInactivity(long timeout, TimeUnit unit) throws InterruptedException {
197         if (unit == null) {
198             throw new NullPointerException("unit");
199         }
200 
201         final Thread thread = this.thread;
202         if (thread == null) {
203             throw new IllegalStateException("thread was not started");
204         }
205         thread.join(unit.toMillis(timeout));
206         return !thread.isAlive();
207     }
208 
209     @Override
210     public void execute(Runnable task) {
211         if (task == null) {
212             throw new NullPointerException("task");
213         }
214 
215         addTask(task);
216         if (!inEventLoop()) {
217             startThread();
218         }
219     }
220 
221     private void startThread() {
222         if (started.compareAndSet(false, true)) {
223             final Thread t = threadFactory.newThread(taskRunner);
224             // Set to null to ensure we not create classloader leaks by holds a strong reference to the inherited
225             // classloader.
226             // See:
227             // - https://github.com/netty/netty/issues/7290
228             // - https://bugs.openjdk.java.net/browse/JDK-7008595
229             AccessController.doPrivileged(new PrivilegedAction<Void>() {
230                 @Override
231                 public Void run() {
232                     t.setContextClassLoader(null);
233                     return null;
234                 }
235             });
236 
237             // Set the thread before starting it as otherwise inEventLoop() may return false and so produce
238             // an assert error.
239             // See https://github.com/netty/netty/issues/4357
240             thread = t;
241             t.start();
242         }
243     }
244 
245     final class TaskRunner implements Runnable {
246         @Override
247         public void run() {
248             for (;;) {
249                 Runnable task = takeTask();
250                 if (task != null) {
251                     try {
252                         task.run();
253                     } catch (Throwable t) {
254                         logger.warn("Unexpected exception from the global event executor: ", t);
255                     }
256 
257                     if (task != quietPeriodTask) {
258                         continue;
259                     }
260                 }
261 
262                 Queue<ScheduledFutureTask<?>> scheduledTaskQueue = GlobalEventExecutor.this.scheduledTaskQueue;
263                 // Terminate if there is no task in the queue (except the noop task).
264                 if (taskQueue.isEmpty() && (scheduledTaskQueue == null || scheduledTaskQueue.size() == 1)) {
265                     // Mark the current thread as stopped.
266                     // The following CAS must always success and must be uncontended,
267                     // because only one thread should be running at the same time.
268                     boolean stopped = started.compareAndSet(true, false);
269                     assert stopped;
270 
271                     // Check if there are pending entries added by execute() or schedule*() while we do CAS above.
272                     if (taskQueue.isEmpty() && (scheduledTaskQueue == null || scheduledTaskQueue.size() == 1)) {
273                         // A) No new task was added and thus there's nothing to handle
274                         //    -> safe to terminate because there's nothing left to do
275                         // B) A new thread started and handled all the new tasks.
276                         //    -> safe to terminate the new thread will take care the rest
277                         break;
278                     }
279 
280                     // There are pending tasks added again.
281                     if (!started.compareAndSet(false, true)) {
282                         // startThread() started a new thread and set 'started' to true.
283                         // -> terminate this thread so that the new thread reads from taskQueue exclusively.
284                         break;
285                     }
286 
287                     // New tasks were added, but this worker was faster to set 'started' to true.
288                     // i.e. a new worker thread was not started by startThread().
289                     // -> keep this thread alive to handle the newly added entries.
290                 }
291             }
292         }
293     }
294 }