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