QuartzSchedulerThread


1   
2   /* 
3    * Copyright 2004-2005 OpenSymphony 
4    * 
5    * Licensed under the Apache License, Version 2.0 (the "License"); you may not 
6    * use this file except in compliance with the License. You may obtain a copy 
7    * of the License at 
8    * 
9    *   http://www.apache.org/licenses/LICENSE-2.0 
10   *   
11   * Unless required by applicable law or agreed to in writing, software 
12   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 
13   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 
14   * License for the specific language governing permissions and limitations 
15   * under the License.
16   * 
17   */
18  
19  /*
20   * Previously Copyright (c) 2001-2004 James House
21   */
22  package org.quartz.core;
23  
24  import org.apache.commons.logging.Log;
25  import org.apache.commons.logging.LogFactory;
26  import org.quartz.JobPersistenceException;
27  import org.quartz.SchedulerException;
28  import org.quartz.Trigger;
29  import org.quartz.spi.TriggerFiredBundle;
30  
31  import java.util.Random  ;
32  
33  /**
34   * <p>
35   * The thread responsible for performing the work of firing <code>{@link Trigger}</code>
36   * s that are registered with the <code>{@link QuartzScheduler}</code>.
37   * </p>
38   * 
39   * @see QuartzScheduler
40   * @see org.quartz.Job
41   * @see Trigger
42   * 
43   * @author James House
44   */
45  public class QuartzSchedulerThread extends Thread   {
46      /*
47       * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
48       * 
49       * Data members.
50       * 
51       * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
52       */
53      private QuartzScheduler qs;
54  
55      private QuartzSchedulerResources qsRsrcs;
56  
57      private Object   pauseLock = new Object  ();
58  
59      private Object   idleLock = new Object  ();
60  
61      private boolean signaled;
62  
63      private boolean paused;
64  
65      private boolean halted;
66  
67      private SchedulingContext ctxt = null;
68  
69      private Random   random = new Random  (System.currentTimeMillis());
70  
71      // When the scheduler finds there is no current trigger to fire, how long
72      // it should wait until checking again...
73      private static long DEFAULT_IDLE_WAIT_TIME = 30L * 1000L;
74  
75      private long idleWaitTime = DEFAULT_IDLE_WAIT_TIME;
76  
77      private int idleWaitVariablness = 7 * 1000;
78  
79      private long dbFailureRetryInterval = 15L * 1000L;
80  
81      private final Log log = LogFactory.getLog(getClass());
82  
83      /*
84       * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
85       * 
86       * Constructors.
87       * 
88       * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
89       */
90  
91      /**
92       * <p>
93       * Construct a new <code>QuartzSchedulerThread</code> for the given
94       * <code>QuartzScheduler</code> as a non-daemon <code>Thread</code>
95       * with normal priority.
96       * </p>
97       */
98      QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs,
99              SchedulingContext ctxt) {
100         this(qs, qsRsrcs, ctxt, qsRsrcs.getMakeSchedulerThreadDaemon(), Thread.NORM_PRIORITY);
101     }
102 
103     /**
104      * <p>
105      * Construct a new <code>QuartzSchedulerThread</code> for the given
106      * <code>QuartzScheduler</code> as a <code>Thread</code> with the given
107      * attributes.
108      * </p>
109      */
110     QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs,
111             SchedulingContext ctxt, boolean setDaemon, int threadPrio) {
112         super(qs.getSchedulerThreadGroup(), qsRsrcs.getThreadName());
113         this.qs = qs;
114         this.qsRsrcs = qsRsrcs;
115         this.ctxt = ctxt;
116         this.setDaemon(setDaemon);
117         this.setPriority(threadPrio);
118 
119         // start the underlying thread, but put this object into the 'paused'
120         // state
121         // so processing doesn't start yet...
122         paused = true;
123         halted = false;
124         this.start();
125     }
126 
127     /*
128      * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
129      * 
130      * Interface.
131      * 
132      * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
133      */
134 
135     void setIdleWaitTime(long waitTime) {
136         idleWaitTime = waitTime;
137         idleWaitVariablness = (int) (waitTime * 0.2);
138     }
139 
140     private long getDbFailureRetryInterval() {
141         return dbFailureRetryInterval;
142     }
143 
144     public void setDbFailureRetryInterval(long dbFailureRetryInterval) {
145         this.dbFailureRetryInterval = dbFailureRetryInterval;
146     }
147 
148     private long getRandomizedIdleWaitTime() {
149         return idleWaitTime - random.nextInt(idleWaitVariablness);
150     }
151 
152     /**
153      * <p>
154      * Signals the main processing loop to pause at the next possible point.
155      * </p>
156      */
157     void togglePause(boolean pause) {
158         synchronized (pauseLock) {
159             paused = pause;
160 
161             if (paused) {
162                 signalSchedulingChange();
163             } else {
164                 pauseLock.notify();
165             }
166         }
167     }
168 
169     /**
170      * <p>
171      * Signals the main processing loop to pause at the next possible point.
172      * </p>
173      */
174     void halt() {
175         synchronized (pauseLock) {
176             halted = true;
177 
178             if (paused) {
179                 pauseLock.notify();
180             } else {
181                 signalSchedulingChange();
182             }
183         }
184     }
185 
186     boolean isPaused() {
187         return paused;
188     }
189 
190     /**
191      * <p>
192      * Signals the main processing loop that a change in scheduling has been
193      * made - in order to interrupt any sleeping that may be occuring while
194      * waiting for the fire time to arrive.
195      * </p>
196      */
197     void signalSchedulingChange() {
198         signaled = true;
199     }
200 
201     /**
202      * <p>
203      * The main processing loop of the <code>QuartzSchedulerThread</code>.
204      * </p>
205      */
206     public void run() {
207         boolean lastAcquireFailed = false;
208         
209         while (!halted) {
210             try {
211                 // check if we're supposed to pause...
212                 synchronized (pauseLock) {
213                     while (paused && !halted) {
214                         try {
215                             // wait until togglePause(false) is called...
216                             pauseLock.wait(100L);
217                         } catch (InterruptedException   ignore) {
218                         }
219                     }
220     
221                     if (halted) {
222                         break;
223                     }
224                 }
225 
226                 int availTreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
227                 if(availTreadCount > 0) {
228 
229                     Trigger trigger = null;
230 
231                     long now = System.currentTimeMillis();
232 
233                     signaled = false;
234                     try {
235                         trigger = qsRsrcs.getJobStore().acquireNextTrigger(
236                                 ctxt, now + idleWaitTime);
237                         lastAcquireFailed = false;
238                     } catch (JobPersistenceException jpe) {
239                         if(!lastAcquireFailed) {
240                             qs.notifySchedulerListenersError(
241                                 "An error occured while scanning for the next trigger to fire.",
242                                 jpe);
243                         }
244                         lastAcquireFailed = true;
245                     } catch (RuntimeException   e) {
246                         if(!lastAcquireFailed) {
247                             getLog().error("quartzSchedulerThreadLoop: RuntimeException "
248                                     +e.getMessage(), e);
249                         }
250                         lastAcquireFailed = true;
251                     }
252 
253                     if (trigger != null) {
254 
255                         now = System.currentTimeMillis();
256                         long triggerTime = trigger.getNextFireTime().getTime();
257                         long timeUntilTrigger = triggerTime - now;
258                         long spinInterval = 10;
259 
260                         // this looping may seem a bit silly, but it's the
261                         // current work-around
262                         // for a dead-lock that can occur if the Thread.sleep()
263                         // is replaced with
264                         // a obj.wait() that gets notified when the signal is
265                         // set...
266                         // so to be able to detect the signal change without
267                         // sleeping the entire
268                         // timeUntilTrigger, we spin here... don't worry
269                         // though, this spinning
270                         // doesn't even register 0.2% cpu usage on a pentium 4.
271                         int numPauses = (int) (timeUntilTrigger / spinInterval);
272                         while (numPauses >= 0 && !signaled) {
273 
274                             try {
275                                 Thread.sleep(spinInterval);
276                             } catch (InterruptedException   ignore) {
277                             }
278 
279                             now = System.currentTimeMillis();
280                             timeUntilTrigger = triggerTime - now;
281                             numPauses = (int) (timeUntilTrigger / spinInterval);
282                         }
283                         if (signaled) {
284                             try {
285                                 qsRsrcs.getJobStore().releaseAcquiredTrigger(
286                                         ctxt, trigger);
287                             } catch (JobPersistenceException jpe) {
288                                 qs.notifySchedulerListenersError(
289                                         "An error occured while releasing trigger '"
290                                                 + trigger.getFullName() + "'",
291                                         jpe);
292                                 // db connection must have failed... keep
293                                 // retrying until it's up...
294                                 releaseTriggerRetryLoop(trigger);
295                             } catch (RuntimeException   e) {
296                                 getLog().error(
297                                     "releaseTriggerRetryLoop: RuntimeException "
298                                     +e.getMessage(), e);
299                                 // db connection must have failed... keep
300                                 // retrying until it's up...
301                                 releaseTriggerRetryLoop(trigger);
302                             }
303                             signaled = false;
304                             continue;
305                         }
306 
307                         // set trigger to 'executing'
308                         TriggerFiredBundle bndle = null;
309 
310                         synchronized(pauseLock) {
311                             if(!halted) {
312                                 try {
313                                     bndle = qsRsrcs.getJobStore().triggerFired(ctxt,
314                                             trigger);
315                                 } catch (SchedulerException se) {
316                                     qs.notifySchedulerListenersError(
317                                             "An error occured while firing trigger '"
318                                                     + trigger.getFullName() + "'", se);
319                                 } catch (RuntimeException   e) {
320                                     getLog().error(
321                                         "RuntimeException while firing trigger " +
322                                         trigger.getFullName(), e);
323                                     // db connection must have failed... keep
324                                     // retrying until it's up...
325                                     releaseTriggerRetryLoop(trigger);
326                                 }
327                             }
328 
329                             // it's possible to get 'null' if the trigger was paused,
330                             // blocked, or other similar occurances that prevent it being
331                             // fired at this time...  or if the scheduler was shutdown (halted)
332                             if (bndle == null) {
333                                 try {
334                                     qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt,
335                                             trigger);
336                                 } catch (SchedulerException se) {
337                                     qs.notifySchedulerListenersError(
338                                             "An error occured while releasing trigger '"
339                                                     + trigger.getFullName() + "'", se);
340                                     // db connection must have failed... keep retrying
341                                     // until it's up...
342                                     releaseTriggerRetryLoop(trigger);
343                                 }
344                                 continue;
345                             }
346 
347                             // TODO: improvements:
348                             //
349                             // 2- make sure we can get a job runshell before firing trigger, or
350                             //   don't let that throw an exception (right now it never does,
351                             //   but the signature says it can).
352                             // 3- acquire more triggers at a time (based on num threads available?)
353 
354 
355                             JobRunShell shell = null;
356                             try {
357                                 shell = qsRsrcs.getJobRunShellFactory().borrowJobRunShell();
358                                 shell.initialize(qs, bndle);
359                             } catch (SchedulerException se) {
360                                 try {
361                                     qsRsrcs.getJobStore().triggeredJobComplete(ctxt,
362                                             trigger, bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);
363                                 } catch (SchedulerException se2) {
364                                     qs.notifySchedulerListenersError(
365                                             "An error occured while placing job's triggers in error state '"
366                                                     + trigger.getFullName() + "'", se2);
367                                     // db connection must have failed... keep retrying
368                                     // until it's up...
369                                     errorTriggerRetryLoop(bndle);
370                                 }
371                                 continue;
372                             }
373 
374                             if (qsRsrcs.getThreadPool().runInThread(shell) == false) {
375                                 try {
376                                     // this case should never happen, as it is indicative of the
377                                     // scheduler being shutdown or a bug in the thread pool or
378                                     // a thread pool being used concurrently - which the docs
379                                     // say not to do...
380                                     getLog().error("ThreadPool.runInThread() return false!");
381                                     qsRsrcs.getJobStore().triggeredJobComplete(ctxt,
382                                             trigger, bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);
383                                 } catch (SchedulerException se2) {
384                                     qs.notifySchedulerListenersError(
385                                             "An error occured while placing job's triggers in error state '"
386                                                     + trigger.getFullName() + "'", se2);
387                                     // db connection must have failed... keep retrying
388                                     // until it's up...
389                                     releaseTriggerRetryLoop(trigger);
390                                 }
391                             }
392                         }
393 
394                         continue;
395                     }
396                 } else { // if(availTreadCount > 0)
397                     continue; // should never happen, if threadPool.blockForAvailableThreads() follows contract
398                 }
399 
400                 // this looping may seem a bit silly, but it's the current
401                 // work-around
402                 // for a dead-lock that can occur if the Thread.sleep() is replaced
403                 // with
404                 // a obj.wait() that gets notified when the signal is set...
405                 // so to be able to detect the signal change without sleeping the
406                 // entier
407                 // getRandomizedIdleWaitTime(), we spin here... don't worry though,
408                 // the
409                 // CPU usage of this spinning can't even be measured on a pentium
410                 // 4.
411                 long now = System.currentTimeMillis();
412                 long waitTime = now + getRandomizedIdleWaitTime();
413                 long timeUntilContinue = waitTime - now;
414                 long spinInterval = 10;
415                 int numPauses = (int) (timeUntilContinue / spinInterval);
416     
417                 while (numPauses > 0 && !signaled) {
418     
419                     try {
420                         Thread.sleep(10L);
421                     } catch (InterruptedException   ignore) {
422                     }
423     
424                     now = System.currentTimeMillis();
425                     timeUntilContinue = waitTime - now;
426                     numPauses = (int) (timeUntilContinue / spinInterval);
427                 }
428             } catch(RuntimeException   re) {
429                 getLog().error("Runtime error occured in main trigger firing loop.", re);
430             }
431         } // loop...
432 
433         // drop references to scheduler stuff to aid garbage collection...
434         qs = null;
435         qsRsrcs = null;
436     }
437 
438     public void errorTriggerRetryLoop(TriggerFiredBundle bndle) {
439         int retryCount = 0;
440         try {
441             while (!halted) {
442                 try {
443                     Thread.sleep(getDbFailureRetryInterval()); // retry every N
444                     // seconds (the db
445                     // connection must
446                     // be failed)
447                     retryCount++;
448                     qsRsrcs.getJobStore().triggeredJobComplete(ctxt,
449                             bndle.getTrigger(), bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);
450                     retryCount = 0;
451                     break;
452                 } catch (JobPersistenceException jpe) {
453                     if(retryCount % 4 == 0) {
454                         qs.notifySchedulerListenersError(
455                             "An error occured while releasing trigger '"
456                                     + bndle.getTrigger().getFullName() + "'", jpe);
457                     }
458                 } catch (RuntimeException   e) {
459                     getLog().error("releaseTriggerRetryLoop: RuntimeException "+e.getMessage(), e);
460                 } catch (InterruptedException   e) {
461                     getLog().error("releaseTriggerRetryLoop: InterruptedException "+e.getMessage(), e);
462                 }
463             }
464         } finally {
465             if(retryCount == 0) {
466                 getLog().info("releaseTriggerRetryLoop: connection restored.");
467             }
468         }
469     }
470     
471     public void releaseTriggerRetryLoop(Trigger trigger) {
472         int retryCount = 0;
473         try {
474             while (!halted) {
475                 try {
476                     Thread.sleep(getDbFailureRetryInterval()); // retry every N
477                     // seconds (the db
478                     // connection must
479                     // be failed)
480                     retryCount++;
481                     qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt, trigger);
482                     retryCount = 0;
483                     break;
484                 } catch (JobPersistenceException jpe) {
485                     if(retryCount % 4 == 0) {
486                         qs.notifySchedulerListenersError(
487                             "An error occured while releasing trigger '"
488                                     + trigger.getFullName() + "'", jpe);
489                     }
490                 } catch (RuntimeException   e) {
491                     getLog().error("releaseTriggerRetryLoop: RuntimeException "+e.getMessage(), e);
492                 } catch (InterruptedException   e) {
493                     getLog().error("releaseTriggerRetryLoop: InterruptedException "+e.getMessage(), e);
494                 }
495             }
496         } finally {
497             if(retryCount == 0) {
498                 getLog().info("releaseTriggerRetryLoop: connection restored.");
499             }
500         }
501     }
502     
503     public Log getLog() {
504         return log;
505     }
506 
507 } // end of QuartzSchedulerThread
508
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags