1 // ========================================================================
2 // Copyright (c) 2004-2009 Mort Bay Consulting Pty. Ltd.
3 // ------------------------------------------------------------------------
4 // All rights reserved. This program and the accompanying materials
5 // are made available under the terms of the Eclipse Public License v1.0
6 // and Apache License v2.0 which accompanies this distribution.
7 // The Eclipse Public License is available at
8 // http://www.eclipse.org/legal/epl-v10.html
9 // The Apache License v2.0 is available at
10 // http://www.opensource.org/licenses/apache2.0.php
11 // You may elect to redistribute this code under either of these licenses.
12 // ========================================================================
13
14
15 package org.eclipse.jetty.util.thread;
16
17 import java.util.concurrent.ConcurrentLinkedQueue;
18 import java.util.concurrent.Executor;
19 import java.util.concurrent.RejectedExecutionException;
20 import java.util.concurrent.TimeUnit;
21 import java.util.concurrent.atomic.AtomicInteger;
22 import java.util.concurrent.atomic.AtomicLong;
23
24 import org.eclipse.jetty.util.BlockingArrayQueue;
25 import org.eclipse.jetty.util.component.AbstractLifeCycle;
26 import org.eclipse.jetty.util.component.LifeCycle;
27 import org.eclipse.jetty.util.log.Log;
28
29
30 public class QueuedThreadPool extends AbstractLifeCycle implements ThreadPool, Executor
31 {
32 private final AtomicInteger _threadsStarted = new AtomicInteger();
33 private final AtomicInteger _threadsIdle = new AtomicInteger();
34 private final AtomicLong _lastShrink = new AtomicLong();
35 private final ConcurrentLinkedQueue<Thread> _threads=new ConcurrentLinkedQueue<Thread>();
36 private final Object _joinLock = new Object();
37 private BlockingArrayQueue<Runnable> _jobs;
38 private String _name;
39 private int _maxIdleTimeMs=60000;
40 private int _maxThreads=254;
41 private int _minThreads=8;
42 private int _maxQueued=-1;
43 private int _priority=Thread.NORM_PRIORITY;
44 private boolean _daemon=false;
45 private int _maxStopTime=100;
46
47 /* ------------------------------------------------------------------- */
48 /* Construct
49 */
50 public QueuedThreadPool()
51 {
52 _name="qtp"+super.hashCode();
53 }
54
55 /* ------------------------------------------------------------------- */
56 /* Construct
57 */
58 public QueuedThreadPool(int maxThreads)
59 {
60 this();
61 setMaxThreads(maxThreads);
62 }
63
64 /* ------------------------------------------------------------ */
65 @Override
66 protected void doStart() throws Exception
67 {
68 super.doStart();
69 _threadsStarted.set(0);
70
71 _jobs=_maxQueued>0 ?new BlockingArrayQueue<Runnable>(_minThreads,_minThreads,_maxQueued)
72 :new BlockingArrayQueue<Runnable>(_minThreads,_minThreads);
73
74 int threads=_threadsStarted.get();
75 while (isRunning() && threads<_minThreads)
76 {
77 startThread(threads);
78 threads=_threadsStarted.get();
79 }
80 }
81
82 /* ------------------------------------------------------------ */
83 @Override
84 protected void doStop() throws Exception
85 {
86 super.doStop();
87 long start=System.currentTimeMillis();
88
89 // let jobs complete naturally for a while
90 while (_threadsStarted.get()>0 && (System.currentTimeMillis()-start) < (_maxStopTime/2))
91 Thread.sleep(1);
92
93 // kill queued jobs and flush out idle jobs
94 _jobs.clear();
95 Runnable noop = new Runnable(){public void run(){}};
96 for (int i=_threadsIdle.get();i-->0;)
97 _jobs.offer(noop);
98 Thread.yield();
99
100 // interrupt remaining threads
101 if (_threadsStarted.get()>0)
102 for (Thread thread : _threads)
103 thread.interrupt();
104
105 // wait for remaining threads to die
106 while (_threadsStarted.get()>0 && (System.currentTimeMillis()-start) < _maxStopTime)
107 {
108 Thread.sleep(1);
109 }
110
111 if (_threads.size()>0)
112 Log.warn(_threads.size()+" threads could not be stopped");
113
114 synchronized (_joinLock)
115 {
116 _joinLock.notifyAll();
117 }
118 }
119
120 /* ------------------------------------------------------------ */
121 /**
122 * Delegated to the named or anonymous Pool.
123 */
124 public void setDaemon(boolean daemon)
125 {
126 _daemon=daemon;
127 }
128
129 /* ------------------------------------------------------------ */
130 /** Set the maximum thread idle time.
131 * Threads that are idle for longer than this period may be
132 * stopped.
133 * Delegated to the named or anonymous Pool.
134 * @see #getMaxIdleTimeMs
135 * @param maxIdleTimeMs Max idle time in ms.
136 */
137 public void setMaxIdleTimeMs(int maxIdleTimeMs)
138 {
139 _maxIdleTimeMs=maxIdleTimeMs;
140 }
141
142 /* ------------------------------------------------------------ */
143 /**
144 * @param stopTimeMs maximum total time that stop() will wait for threads to die.
145 */
146 public void setMaxStopTimeMs(int stopTimeMs)
147 {
148 _maxStopTime = stopTimeMs;
149 }
150
151 /* ------------------------------------------------------------ */
152 /** Set the maximum number of threads.
153 * Delegated to the named or anonymous Pool.
154 * @see #getMaxThreads
155 * @param maxThreads maximum number of threads.
156 */
157 public void setMaxThreads(int maxThreads)
158 {
159 if (isStarted() && maxThreads<_minThreads)
160 throw new IllegalArgumentException("!minThreads<maxThreads");
161 _maxThreads=maxThreads;
162 }
163
164 /* ------------------------------------------------------------ */
165 /** Set the minimum number of threads.
166 * Delegated to the named or anonymous Pool.
167 * @see #getMinThreads
168 * @param minThreads minimum number of threads
169 */
170 public void setMinThreads(int minThreads)
171 {
172 if (isStarted() && (minThreads<=0 || minThreads>_maxThreads))
173 throw new IllegalArgumentException("!0<=minThreads<maxThreads");
174 _minThreads=minThreads;
175
176 int threads=_threadsStarted.get();
177 while (isStarted() && threads<_minThreads)
178 {
179 startThread(threads);
180 threads=_threadsStarted.get();
181 }
182 }
183
184 /* ------------------------------------------------------------ */
185 /**
186 * @param name Name of the BoundedThreadPool to use when naming Threads.
187 */
188 public void setName(String name)
189 {
190 if (isRunning())
191 throw new IllegalStateException("started");
192 _name= name;
193 }
194
195 /* ------------------------------------------------------------ */
196 /** Set the priority of the pool threads.
197 * @param priority the new thread priority.
198 */
199 public void setThreadsPriority(int priority)
200 {
201 _priority=priority;
202 }
203
204 /* ------------------------------------------------------------ */
205 /**
206 * @return maximum queue size
207 */
208 public int getMaxQueued()
209 {
210 if (isRunning())
211 throw new IllegalStateException("started");
212 return _maxQueued;
213 }
214
215 /* ------------------------------------------------------------ */
216 /** Get the maximum thread idle time.
217 * Delegated to the named or anonymous Pool.
218 * @see #setMaxIdleTimeMs
219 * @return Max idle time in ms.
220 */
221 public int getMaxIdleTimeMs()
222 {
223 return _maxIdleTimeMs;
224 }
225
226 /* ------------------------------------------------------------ */
227 /**
228 * @return maximum total time that stop() will wait for threads to die.
229 */
230 public int getMaxStopTimeMs()
231 {
232 return _maxStopTime;
233 }
234
235 /* ------------------------------------------------------------ */
236 /** Set the maximum number of threads.
237 * Delegated to the named or anonymous Pool.
238 * @see #setMaxThreads
239 * @return maximum number of threads.
240 */
241 public int getMaxThreads()
242 {
243 return _maxThreads;
244 }
245
246 /* ------------------------------------------------------------ */
247 /** Get the minimum number of threads.
248 * Delegated to the named or anonymous Pool.
249 * @see #setMinThreads
250 * @return minimum number of threads.
251 */
252 public int getMinThreads()
253 {
254 return _minThreads;
255 }
256
257 /* ------------------------------------------------------------ */
258 /**
259 * @return The name of the BoundedThreadPool.
260 */
261 public String getName()
262 {
263 return _name;
264 }
265
266 /* ------------------------------------------------------------ */
267 /** Get the priority of the pool threads.
268 * @return the priority of the pool threads.
269 */
270 public int getThreadsPriority()
271 {
272 return _priority;
273 }
274
275 /* ------------------------------------------------------------ */
276 /**
277 * Delegated to the named or anonymous Pool.
278 */
279 public boolean isDaemon()
280 {
281 return _daemon;
282 }
283
284
285 /* ------------------------------------------------------------ */
286 public boolean dispatch(Runnable job)
287 {
288 if (isRunning())
289 {
290 final int jobQ = _jobs.size();
291 final int idle = getIdleThreads();
292 if(_jobs.offer(job))
293 {
294 // If we had no idle threads or the jobQ is greater than the idle threads
295 if (idle==0 || jobQ>idle)
296 {
297 int threads=_threadsStarted.get();
298 if (threads<_maxThreads)
299 startThread(threads);
300 }
301 return true;
302 }
303 }
304 return false;
305 }
306
307 /* ------------------------------------------------------------ */
308 public void execute(Runnable job)
309 {
310 if (!dispatch(job))
311 throw new RejectedExecutionException();
312 }
313
314 /* ------------------------------------------------------------ */
315 /**
316 * Blocks until the thread pool is {@link LifeCycle#stop stopped}.
317 */
318 public void join() throws InterruptedException
319 {
320 synchronized (_joinLock)
321 {
322 while (isRunning())
323 _joinLock.wait();
324 }
325
326 while (isStopping())
327 Thread.sleep(1);
328 }
329
330 /* ------------------------------------------------------------ */
331 /**
332 * @return The total number of threads currently in the pool
333 */
334 public int getThreads()
335 {
336 return _threadsStarted.get();
337 }
338
339 /* ------------------------------------------------------------ */
340 /**
341 * @return The number of idle threads in the pool
342 */
343 public int getIdleThreads()
344 {
345 return _threadsIdle.get();
346 }
347
348 /* ------------------------------------------------------------ */
349 /**
350 * @return True if the pool is at maxThreads and there are more queued jobs than idle threads
351 */
352 public boolean isLowOnThreads()
353 {
354 return _threadsStarted.get()==_maxThreads && _jobs.size()>_threadsIdle.get();
355 }
356
357 /* ------------------------------------------------------------ */
358 private boolean startThread(int threads)
359 {
360 final int next=threads+1;
361 if (!_threadsStarted.compareAndSet(threads,next))
362 return false;
363
364 boolean started=false;
365 try
366 {
367 Thread thread=newThread(_runnable);
368 thread.setDaemon(_daemon);
369 thread.setPriority(_priority);
370 thread.setName(_name+"-"+thread.getId());
371 _threads.add(thread);
372
373 thread.start();
374 started=true;
375 }
376 finally
377 {
378 if (!started)
379 _threadsStarted.decrementAndGet();
380 }
381 return started;
382 }
383
384 /* ------------------------------------------------------------ */
385 protected Thread newThread(Runnable runnable)
386 {
387 return new Thread(runnable);
388 }
389
390 /* ------------------------------------------------------------ */
391 private Runnable _runnable = new Runnable()
392 {
393 public void run()
394 {
395 boolean shrink=false;
396 try
397 {
398 Runnable job=_jobs.poll();
399 while (isRunning())
400 {
401 // Job loop
402 while (job!=null && isRunning())
403 {
404 job.run();
405 job=_jobs.poll();
406 }
407
408 // Idle loop
409 try
410 {
411 _threadsIdle.incrementAndGet();
412
413 while (isRunning() && job==null)
414 {
415 if (_maxIdleTimeMs<=0)
416 job=_jobs.take();
417 else
418 {
419 job=_jobs.poll(_maxIdleTimeMs,TimeUnit.MILLISECONDS);
420
421 if (job==null)
422 {
423 // maybe we should shrink?
424 final int size=_threadsStarted.get();
425 if (size>_minThreads)
426 {
427 long last=_lastShrink.get();
428 long now=System.currentTimeMillis();
429 if (last==0 || (now-last)>_maxIdleTimeMs)
430 {
431 shrink=_lastShrink.compareAndSet(last,now) &&
432 _threadsStarted.compareAndSet(size,size-1);
433 if (shrink)
434 return;
435 }
436 }
437 }
438 }
439 }
440 }
441 finally
442 {
443 _threadsIdle.decrementAndGet();
444 }
445 }
446 }
447 catch(InterruptedException e)
448 {
449 Log.ignore(e);
450 }
451 catch(Exception e)
452 {
453 Log.warn(e);
454 }
455 finally
456 {
457 if (!shrink)
458 _threadsStarted.decrementAndGet();
459 _threads.remove(Thread.currentThread());
460 }
461 }
462 };
463 }