CmsSchedulerThreadPool


1   /*
2    * File   : $Source: /usr/local/cvs/opencms/src/org/opencms/scheduler/CmsSchedulerThreadPool.java,v $
3    * Date   : $Date: 2006/03/27 14:52:20 $
4    * Version: $Revision: 1.12 $
5    *
6    * This library is part of OpenCms -
7    * the Open Source Content Mananagement System
8    *
9    * Copyright (c) 2005 Alkacon Software GmbH (http://www.alkacon.com)
10   *
11   * This library is free software; you can redistribute it and/or
12   * modify it under the terms of the GNU Lesser General Public
13   * License as published by the Free Software Foundation; either
14   * version 2.1 of the License, or (at your option) any later version.
15   *
16   * This library is distributed in the hope that it will be useful,
17   * but WITHOUT ANY WARRANTY; without even the implied warranty of
18   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
19   * Lesser General Public License for more details.
20   *
21   * For further information about Alkacon Software GmbH, please see the
22   * company website: http://www.alkacon.com
23   *
24   * For further information about OpenCms, please see the
25   * project website: http://www.opencms.org
26   * 
27   * You should have received a copy of the GNU Lesser General Public
28   * License along with this library; if not, write to the Free Software
29   * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
30   * 
31   * 
32   * This library is based to some extend on code from the 
33   * OpenSymphony Quartz project. Original copyright notice:
34   * 
35   * Copyright James House (c) 2001-2005
36   * 
37   * All rights reserved.
38   * 
39   * Redistribution and use in source and binary forms, with or without
40   * modification, are permitted provided that the following conditions are met: 1.
41   * Redistributions of source code must retain the above copyright notice, this
42   * list of conditions and the following disclaimer. 2. Redistributions in
43   * binary form must reproduce the above copyright notice, this list of
44   * conditions and the following disclaimer in the documentation and/or other
45   * materials provided with the distribution.
46   * 
47   * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND ANY
48   * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
49   * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
50   * DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE FOR ANY
51   * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
52   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
53   * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
54   * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
55   * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
56   * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
57   */
58  
59  package org.opencms.scheduler;
60  
61  import org.opencms.main.CmsLog;
62  
63  import org.apache.commons.logging.Log;
64  
65  import org.quartz.SchedulerConfigException;
66  import org.quartz.spi.ThreadPool;
67  
68  /**
69   * Simple thread pool used for the Quartz scheduler in OpenCms.<p>
70   * 
71   * @author Alexander Kandzior 
72   * @author James House
73   * @author Juergen Donnerstag
74   *
75   * @version $Revision: 1.12 $ 
76   * 
77   * @since 6.0.0 
78   */
79  public class CmsSchedulerThreadPool implements ThreadPool {
80  
81      /** The log object for this class. */
82      private static final Log LOG = CmsLog.getLog(CmsSchedulerThreadPool.class);
83  
84      private int m_currentThreadCount;
85  
86      private boolean m_inheritGroup;
87  
88      private boolean m_inheritLoader;
89  
90      private int m_initialThreadCount;
91  
92      private boolean m_isShutdown;
93  
94      private boolean m_makeThreadsDaemons;
95  
96      private int m_maxThreadCount;
97  
98      private Runnable   m_nextRunnable;
99  
100     private Object   m_nextRunnableLock;
101 
102     private ThreadGroup   m_threadGroup;
103 
104     private String   m_threadNamePrefix;
105 
106     private int m_threadPriority;
107 
108     private CmsSchedulerThread[] m_workers;
109 
110     /**
111      * Create a new <code>CmsSchedulerThreadPool</code> with default values.
112      * 
113      * This will create a pool with 0 initial and 10 maximum threads running 
114      * in normal priority.<p>
115      * 
116      * @see #CmsSchedulerThreadPool(int, int, int)
117      */
118     public CmsSchedulerThreadPool() {
119 
120         this(0, 10, Thread.NORM_PRIORITY);
121     }
122 
123     /**
124      * Create a new <code>CmsSchedulerThreadPool</code> with the specified number
125      * of threads that have the given priority.
126      * 
127      * The OpenCms scheduler thread pool will initially start with provided number of
128      * active scheduler threads.
129      * When a thread is requested by the scheduler, and no "free" threads are available,
130      * a new thread will be added to the pool and used for execution. The pool 
131      * will be allowed to grow until it has reached the configured number 
132      * of maximum threads.<p>
133      * 
134      * @param initialThreadCount the initial number of threads for the pool
135      * @param maxThreadCount maximum number of threads the pool is allowed to grow
136      * @param threadPriority the thread priority for the scheduler threads
137      * 
138      * @see java.lang.Thread
139      */
140     public CmsSchedulerThreadPool(int initialThreadCount, int maxThreadCount, int threadPriority) {
141 
142         m_inheritGroup = true;
143         m_inheritLoader = true;
144         m_nextRunnableLock = new Object  ();
145         m_threadNamePrefix = "OpenCms: Scheduler Thread ";
146         m_makeThreadsDaemons = true;
147         m_initialThreadCount = initialThreadCount;
148         m_currentThreadCount = 0;
149         m_maxThreadCount = maxThreadCount;
150         m_threadPriority = threadPriority;
151     }
152 
153     /**
154      * @see org.quartz.spi.ThreadPool#getPoolSize()
155      */
156     public int getPoolSize() {
157 
158         return m_currentThreadCount;
159     }
160 
161     /**
162      * Returns the thread priority of the threads in the scheduler pool.<p>
163      * 
164      * @return the thread priority of the threads in the scheduler pool 
165      */
166     public int getThreadPriority() {
167 
168         return m_threadPriority;
169     }
170 
171     /**
172      * @see org.quartz.spi.ThreadPool#initialize()
173      */
174     public void initialize() throws SchedulerConfigException {
175 
176         if (m_maxThreadCount <= 0 || m_maxThreadCount > 200) {
177             throw new SchedulerConfigException(Messages.get().getBundle().key(Messages.ERR_MAX_THREAD_COUNT_BOUNDS_0));
178         }
179         if (m_initialThreadCount < 0 || m_initialThreadCount > m_maxThreadCount) {
180             throw new SchedulerConfigException(Messages.get().getBundle().key(Messages.ERR_INIT_THREAD_COUNT_BOUNDS_0));
181         }
182         if (m_threadPriority <= 0 || m_threadPriority > 9) {
183             throw new SchedulerConfigException(Messages.get().getBundle().key(Messages.ERR_SCHEDULER_PRIORITY_BOUNDS_0));
184         }
185 
186         if (m_inheritGroup) {
187             m_threadGroup = Thread.currentThread().getThreadGroup();
188         } else {
189             // follow the threadGroup tree to the root thread group
190             m_threadGroup = Thread.currentThread().getThreadGroup();
191             ThreadGroup   parent = m_threadGroup;
192             while (!parent.getName().equals("main")) {
193                 m_threadGroup = parent;
194                 parent = m_threadGroup.getParent();
195             }
196             m_threadGroup = new ThreadGroup  (parent, this.getClass().getName());
197         }
198 
199         if (m_inheritLoader) {
200             LOG.debug(Messages.get().getBundle().key(
201                 Messages.LOG_USING_THREAD_CLASSLOADER_1,
202                 Thread.currentThread().getName()));
203         }
204 
205         // create the worker threads and start them
206         m_workers = new CmsSchedulerThread[m_maxThreadCount];
207         for (int i = 0; i < m_initialThreadCount; ++i) {
208             growThreadPool();
209         }
210     }
211 
212     /**
213      * Run the given <code>Runnable</code> object in the next available
214      * <code>Thread</code>.<p>
215      * 
216      * If while waiting the thread pool is asked to
217      * shut down, the Runnable is executed immediately within a new additional
218      * thread.<p>
219      * 
220      * @param runnable the <code>Runnable</code> to run
221      * @return true if the <code>Runnable</code> was run
222      */
223     public boolean runInThread(Runnable   runnable) {
224 
225         if (runnable == null) {
226             return false;
227         }
228 
229         if (m_isShutdown) {
230             LOG.debug(Messages.get().getBundle().key(Messages.LOG_THREAD_POOL_UNAVAILABLE_0));
231             return false;
232         }
233 
234         if ((m_currentThreadCount == 0) || (m_nextRunnable != null)) {
235             // try to grow the thread pool since other runnables are already waiting
236             growThreadPool();
237         }
238 
239         synchronized (m_nextRunnableLock) {
240 
241             // wait until a worker thread has taken the previous Runnable
242             // or until the thread pool is asked to shutdown
243             while ((m_nextRunnable != null) && !m_isShutdown) {
244                 try {
245                     m_nextRunnableLock.wait(1000);
246                 } catch (InterruptedException   e) {
247                     // can be ignores
248                 }
249             }
250 
251             // during normal operation, not shutdown, set the nextRunnable
252             // and notify the worker threads waiting (getNextRunnable())
253             if (!m_isShutdown) {
254                 m_nextRunnable = runnable;
255                 m_nextRunnableLock.notifyAll();
256             }
257         }
258 
259         // if the thread pool is going down, execute the Runnable
260         // within a new additional worker thread (no thread from the pool)
261         // note: the synchronized section should be as short (time) as
262         // possible as starting a new thread is not a quick action
263         if (m_isShutdown) {
264             new CmsSchedulerThread(
265                 this,
266                 m_threadGroup,
267                 m_threadNamePrefix + "(final)",
268                 m_threadPriority,
269                 false,
270                 runnable);
271         }
272 
273         return true;
274     }
275 
276     /**
277      * Terminate any worker threads in this thread group.<p>
278      * 
279      * Jobs currently in progress will be allowed to complete.<p>
280      */
281     public void shutdown() {
282 
283         shutdown(true);
284     }
285 
286     /**
287      * Terminate all threads in this thread group.<p>
288      * 
289      * @param waitForJobsToComplete if true,, all current jobs will be allowed to complete
290      */
291     public void shutdown(boolean waitForJobsToComplete) {
292 
293         m_isShutdown = true;
294 
295         // signal each scheduler thread to shut down
296         for (int i = 0; i < m_currentThreadCount; i++) {
297             if (m_workers[i] != null) {
298                 m_workers[i].shutdown();
299             }
300         }
301 
302         // give waiting (wait(1000)) worker threads a chance to shut down
303         // active worker threads will shut down after finishing their
304         // current job
305         synchronized (m_nextRunnableLock) {
306             m_nextRunnableLock.notifyAll();
307         }
308 
309         if (waitForJobsToComplete) {
310             // wait until all worker threads are shut down
311             int alive = m_currentThreadCount;
312             while (alive > 0) {
313                 alive = 0;
314                 for (int i = 0; i < m_currentThreadCount; i++) {
315                     if (m_workers[i].isAlive()) {
316                         try {
317                             if (LOG.isDebugEnabled()) {
318                                 LOG.debug(Messages.get().getBundle().key(
319                                     Messages.LOG_THREAD_POOL_WAITING_1,
320                                     new Integer  (i)));
321                             }
322 
323                             // note: with waiting infinite - join(0) - the application 
324                             // may appear to 'hang' 
325                             // waiting for a finite time however requires an additional loop (alive)
326                             alive++;
327                             m_workers[i].join(200);
328                         } catch (InterruptedException   e) {
329                             // can be ignored
330                         }
331                     }
332                 }
333             }
334 
335             int activeCount = m_threadGroup.activeCount();
336             if (activeCount > 0 && LOG.isInfoEnabled()) {
337                 LOG.info(Messages.get().getBundle().key(
338                     Messages.LOG_THREAD_POOL_STILL_ACTIVE_1,
339                     new Integer  (activeCount)));
340             }
341             if (LOG.isDebugEnabled()) {
342                 LOG.debug(Messages.get().getBundle().key(Messages.LOG_THREAD_POOL_SHUTDOWN_0));
343             }
344         }
345     }
346 
347     /**
348      * Dequeue the next pending <code>Runnable</code>.<p>
349      * 
350      * @return the next pending <code>Runnable</code>
351      * @throws InterruptedException if something goes wrong
352      */
353     protected Runnable   getNextRunnable() throws InterruptedException   {
354 
355         Runnable   toRun = null;
356 
357         // Wait for new Runnable (see runInThread()) and notify runInThread()
358         // in case the next Runnable is already waiting.
359         synchronized (m_nextRunnableLock) {
360             if (m_nextRunnable == null) {
361                 m_nextRunnableLock.wait(1000);
362             }
363 
364             if (m_nextRunnable != null) {
365                 toRun = m_nextRunnable;
366                 m_nextRunnable = null;
367                 m_nextRunnableLock.notifyAll();
368             }
369         }
370 
371         return toRun;
372     }
373 
374     /**
375      * Grows the thread pool by one new thread if the maximum pool size 
376      * has not been reached.<p>
377      */
378     private void growThreadPool() {
379 
380         if (m_currentThreadCount < m_maxThreadCount) {
381             // if maximum number is not reached grow the thread pool
382             synchronized (m_nextRunnableLock) {
383                 m_workers[m_currentThreadCount] = new CmsSchedulerThread(this, m_threadGroup, m_threadNamePrefix
384                     + m_currentThreadCount, m_threadPriority, m_makeThreadsDaemons);
385                 if (m_inheritLoader) {
386                     m_workers[m_currentThreadCount].setContextClassLoader(Thread.currentThread().getContextClassLoader());
387                 }
388                 // increas the current size
389                 m_currentThreadCount++;
390                 // notify the waiting threads
391                 m_nextRunnableLock.notifyAll();
392             }
393         }
394     }
395 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags