Scheduler


1   /**
2    * Redistribution and use of this software and associated documentation
3    * ("Software"), with or without modification, are permitted provided
4    * that the following conditions are met:
5    *
6    * 1. Redistributions of source code must retain copyright
7    *    statements and notices.  Redistributions must also contain a
8    *    copy of this document.
9    *
10   * 2. Redistributions in binary form must reproduce the
11   *    above copyright notice, this list of conditions and the
12   *    following disclaimer in the documentation and/or other
13   *    materials provided with the distribution.
14   *
15   * 3. The name "Exolab" must not be used to endorse or promote
16   *    products derived from this Software without prior written
17   *    permission of Exoffice Technologies.  For written permission,
18   *    please contact info@exolab.org.
19   *
20   * 4. Products derived from this Software may not be called "Exolab"
21   *    nor may "Exolab" appear in their names without prior written
22   *    permission of Exoffice Technologies. Exolab is a registered
23   *    trademark of Exoffice Technologies.
24   *
25   * 5. Due credit should be given to the Exolab Project
26   *    (http://www.exolab.org/).
27   *
28   * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
29   * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
30   * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
31   * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
32   * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
33   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
34   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
35   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
36   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
37   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
38   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
39   * OF THE POSSIBILITY OF SUCH DAMAGE.
40   *
41   * Copyright 2000-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
42   */
43  package org.exolab.jms.scheduler;
44  
45  import java.util.HashMap  ;
46  import java.util.LinkedList  ;
47  
48  import org.apache.commons.logging.Log;
49  import org.apache.commons.logging.LogFactory;
50  
51  import org.exolab.jms.service.BasicService;
52  import org.exolab.jms.service.ServiceException;
53  import org.exolab.jms.common.threads.ThreadPool;
54  import org.exolab.jms.config.Configuration;
55  import org.exolab.jms.config.ConfigurationManager;
56  import org.exolab.jms.config.SchedulerConfiguration;
57  import org.exolab.jms.threads.ThreadPoolManager;
58  
59  
60  /**
61   * The scheduler is responsible for executing {@link Runnable} objects
62   * using a thread pool. Clients can add these objects to the scheduler
63   * and the scheduler will, in fifo order, execute them. If there are no
64   * threads currently available, the runnable will wait for one to become
65   * available.
66   * <p>
67   * A client can add or remove {@link Runnable} objects.
68   *
69   * @version     $Revision: 1.1 $ $Date: 2004/11/26 01:50:44 $
70   * @author      <a HREF="mailto:mourikis@intalio.com">Jim Mourikis</a>
71   * @author      <a HREF="mailto:tma@netspace.net.au">Tim Anderson</a>
72   * @author      <a HREF="mailto:jima@comware.com.au">Jim Alateras</a>
73   * @see         java.lang.Runnable
74   * @see         org.exolab.jms.service.BasicService
75   * @see         org.exolab.jms.common.threads.ThreadPool
76   */
77  public class Scheduler extends BasicService {
78  
79      /**
80       * The thread pool used by the scheduler
81       */
82      private ThreadPool _threads = null;
83  
84      /**
85       * The queue of Runnable instances
86       */
87      private LinkedList   _queue = new LinkedList  ();;
88  
89      /**
90       * HashMap of Runnable->Integer, representing the number of times a
91       * Runnable object exists in the queue
92       */
93      private HashMap   _references = new HashMap  ();
94  
95      /**
96       * If true, shuts down the scheduler
97       */
98      private volatile boolean _stop = false;
99  
100     /**
101      * This attribute holds the number of threads that the scheduler
102      * will use in its pool. It defaults to 6 and has a minimum value
103      * of 2
104      */
105     private int _threadCount = 6;
106 
107     /**
108      * This is the minimum number of threads that can be used to
109      * configure the scheduler. If a lower nmber is specified then
110      * it defaults to this value
111      */
112     private final static int MIN_THREAD_COUNT = 2;
113 
114     /**
115      * Unique name identifyting this sevice
116      */
117     private static final String   SCHEDULER_NAME = "Scheduler";
118 
119     /**
120      * This is the singleton instance of this class
121      */
122     private static Scheduler _instance = null;
123 
124     /**
125      * The logger
126      */
127     private static final Log _log = LogFactory.getLog(Scheduler.class);
128 
129     /**
130      * Creates the singleton instance
131      *
132      * @throws ServiceException if the scheduler can't be created
133      */
134     public static Scheduler createInstance() throws ServiceException {
135         _instance = new Scheduler();
136         return _instance;
137     }
138 
139     /**
140      * Returns the singleton instance
141      *
142      * @return the singleton instance, or <code>null</code> if it hasn't
143      * been initialised
144      */
145     public static Scheduler instance() {
146         return _instance;
147     }
148 
149     /**
150      * Construct an instance of the scheduler. This will read the information
151      * from the configuration file and then set up a thread pool.
152      *
153      * @throws ServiceException if the thread pool can't be initialised
154      */
155     private Scheduler() throws ServiceException {
156         super(SCHEDULER_NAME);
157 
158         // access the configuration file.
159         Configuration config = ConfigurationManager.getConfig();
160         SchedulerConfiguration sched_config =
161             config.getSchedulerConfiguration();
162 
163         int count = sched_config.getMaxThreads();
164         if (count < MIN_THREAD_COUNT) {
165             count = MIN_THREAD_COUNT;
166         }
167         _threadCount = count;
168 
169         // create the thread pool
170         _threads = ThreadPoolManager.instance().createThreadPool(
171             SCHEDULER_NAME, _threadCount);
172     }
173 
174     /**
175      * Add a Runnable object to the scheduler queue.
176      * When a thread becomes available, it will be executed.
177      *
178      * @param       runner              the object to execute
179      */
180     public void add(Runnable   runner) {
181         synchronized (_queue) {
182             _queue.addLast(runner);
183             addReference(runner);
184             _queue.notify();
185         }
186     }
187 
188     /**
189      * Remove a Runnable object from the scheduler queue.
190      *
191      * @param       runner              the object to remove
192      * @return      boolean             <tt>true</tt> if the object was
193      *                                  removed, <tt>false</tt> if it is
194      *                                  already running or doesn't exist
195      */
196     public boolean remove(Runnable   runner) {
197         boolean result = false;
198         synchronized (_queue) {
199             result = _queue.remove(runner);
200         }
201         return result;
202     }
203 
204     /**
205      * Returns if a Runnable object exists in the scheduler queue.
206      *
207      * @param       runner              the object to remove
208      * @return      boolean             <tt>true</tt> if the object exists,
209      *                                  <tt>false</tt> if it is already
210      *                                  running or doesn't exist
211      */
212     public boolean contains(Runnable   runner) {
213         boolean result = false;
214         synchronized (_queue) {
215             result = (_references.get(runner) != null);
216         }
217         return result;
218     }
219 
220     /**
221      * Returns true if the scheduler queue is empty
222      *
223      * @return <tt>true</tt> if the scheduler queue is empty
224      */
225     public boolean isEmpty() {
226         boolean result = false;
227         synchronized (_queue) {
228             result = _queue.isEmpty();
229         }
230         return result;
231     }
232 
233     /**
234      * Start the scheduler
235      * This can only be terminated by invoking {@link #stop}
236      */
237     public void run() {
238         while (!_stop) {
239             Runnable   runner = next();
240             if (!_stop && runner != null) {
241                 try {
242                     _threads.execute(runner);
243                 } catch (Exception   exception) {
244                     _log.error(exception);
245                 }
246             }
247         }
248     }
249 
250     // override BasicService.stop
251     public void stop() throws ServiceException {
252         // TODO - need a safer way of shutting down threads.
253         _threads.stopRequestAllWorkers();
254         _stop = true;
255         super.stop();
256     }
257 
258     /**
259      * Return the next object in the queue to execute
260      * This method blocks until an object becomes available.
261      *
262      * @return      Runnable            the next object to execute
263      */
264     protected Runnable   next() {
265         Runnable   result = null;
266         synchronized (_queue) {
267             while (!_stop && _queue.isEmpty()) {
268                 try {
269                     _queue.wait();
270                 } catch (InterruptedException   ignore) {
271                     // do nothing.
272                 }
273             }
274             if (!_stop) {
275                 result = (Runnable  ) _queue.removeFirst();
276                 removeReference(result);
277             }
278         }
279         return result;
280     }
281 
282     /**
283      * Increment the reference count to a queued Runnable object,
284      * to enable contains() to be as efficient as possible.
285      */
286     private void addReference(Runnable   runner) {
287         Integer   count = (Integer  ) _references.get(runner);
288         if (count != null) {
289             count = new Integer  (count.intValue() + 1);
290             _references.put(runner, count);
291         } else {
292             _references.put(runner, new Integer  (1));
293         }
294     }
295 
296     /**
297      * Decrement the reference count to a queued Runnable object,
298      * removing it if no more references remain.
299      */
300     private void removeReference(Runnable   runner) {
301         // decrement the no. of references to the Runnable object,
302         // removing it when the count reaches 0
303         Integer   count = (Integer  ) _references.get(runner);
304         if (count.intValue() <= 1) {
305             _references.remove(runner);
306         } else {
307             _references.put(runner, new Integer  (count.intValue() - 1));
308         }
309     }
310 
311 }
312
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags