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