/*
* Copyright 2004-2005 OpenSymphony
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
* use this file except in compliance with the License. You may obtain a copy
* of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*
*/
/*
* Previously Copyright (c) 2001-2004 James House
*/
package org.quartz.core;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.quartz.JobPersistenceException;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.spi.TriggerFiredBundle;
import java.util.Random;
/**
*
* The thread responsible for performing the work of firing {@link Trigger}
* s that are registered with the {@link QuartzScheduler}.
*
*
* @see QuartzScheduler
* @see org.quartz.Job
* @see Trigger
*
* @author James House
*/
public class QuartzSchedulerThread extends Thread {
/*
* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*
* Data members.
*
* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*/
private QuartzScheduler qs;
private QuartzSchedulerResources qsRsrcs;
private Object pauseLock = new Object();
private Object idleLock = new Object();
private boolean signaled;
private boolean paused;
private boolean halted;
private SchedulingContext ctxt = null;
private Random random = new Random(System.currentTimeMillis());
// When the scheduler finds there is no current trigger to fire, how long
// it should wait until checking again...
private static long DEFAULT_IDLE_WAIT_TIME = 30L * 1000L;
private long idleWaitTime = DEFAULT_IDLE_WAIT_TIME;
private int idleWaitVariablness = 7 * 1000;
private long dbFailureRetryInterval = 15L * 1000L;
private final Log log = LogFactory.getLog(getClass());
/*
* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*
* Constructors.
*
* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*/
/**
*
* Construct a new QuartzSchedulerThread for the given
* QuartzScheduler as a non-daemon Thread
* with normal priority.
*
*/
QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs,
SchedulingContext ctxt) {
this(qs, qsRsrcs, ctxt, qsRsrcs.getMakeSchedulerThreadDaemon(), Thread.NORM_PRIORITY);
}
/**
*
* Construct a new QuartzSchedulerThread for the given
* QuartzScheduler as a Thread with the given
* attributes.
*
*/
QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs,
SchedulingContext ctxt, boolean setDaemon, int threadPrio) {
super(qs.getSchedulerThreadGroup(), qsRsrcs.getThreadName());
this.qs = qs;
this.qsRsrcs = qsRsrcs;
this.ctxt = ctxt;
this.setDaemon(setDaemon);
this.setPriority(threadPrio);
// start the underlying thread, but put this object into the 'paused'
// state
// so processing doesn't start yet...
paused = true;
halted = false;
this.start();
}
/*
* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*
* Interface.
*
* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
*/
void setIdleWaitTime(long waitTime) {
idleWaitTime = waitTime;
idleWaitVariablness = (int) (waitTime * 0.2);
}
private long getDbFailureRetryInterval() {
return dbFailureRetryInterval;
}
public void setDbFailureRetryInterval(long dbFailureRetryInterval) {
this.dbFailureRetryInterval = dbFailureRetryInterval;
}
private long getRandomizedIdleWaitTime() {
return idleWaitTime - random.nextInt(idleWaitVariablness);
}
/**
*
* Signals the main processing loop to pause at the next possible point.
*
*/
void togglePause(boolean pause) {
synchronized (pauseLock) {
paused = pause;
if (paused) {
signalSchedulingChange();
} else {
pauseLock.notify();
}
}
}
/**
*
* Signals the main processing loop to pause at the next possible point.
*
*/
void halt() {
synchronized (pauseLock) {
halted = true;
if (paused) {
pauseLock.notify();
} else {
signalSchedulingChange();
}
}
}
boolean isPaused() {
return paused;
}
/**
*
* Signals the main processing loop that a change in scheduling has been
* made - in order to interrupt any sleeping that may be occuring while
* waiting for the fire time to arrive.
*
*/
void signalSchedulingChange() {
signaled = true;
}
/**
*
* The main processing loop of the QuartzSchedulerThread.
*
*/
public void run() {
boolean lastAcquireFailed = false;
while (!halted) {
try {
// check if we're supposed to pause...
synchronized (pauseLock) {
while (paused && !halted) {
try {
// wait until togglePause(false) is called...
pauseLock.wait(100L);
} catch (InterruptedException ignore) {
}
}
if (halted) {
break;
}
}
int availTreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
if(availTreadCount > 0) {
Trigger trigger = null;
long now = System.currentTimeMillis();
signaled = false;
try {
trigger = qsRsrcs.getJobStore().acquireNextTrigger(
ctxt, now + idleWaitTime);
lastAcquireFailed = false;
} catch (JobPersistenceException jpe) {
if(!lastAcquireFailed) {
qs.notifySchedulerListenersError(
"An error occured while scanning for the next trigger to fire.",
jpe);
}
lastAcquireFailed = true;
} catch (RuntimeException e) {
if(!lastAcquireFailed) {
getLog().error("quartzSchedulerThreadLoop: RuntimeException "
+e.getMessage(), e);
}
lastAcquireFailed = true;
}
if (trigger != null) {
now = System.currentTimeMillis();
long triggerTime = trigger.getNextFireTime().getTime();
long timeUntilTrigger = triggerTime - now;
long spinInterval = 10;
// this looping may seem a bit silly, but it's the
// current work-around
// for a dead-lock that can occur if the Thread.sleep()
// is replaced with
// a obj.wait() that gets notified when the signal is
// set...
// so to be able to detect the signal change without
// sleeping the entire
// timeUntilTrigger, we spin here... don't worry
// though, this spinning
// doesn't even register 0.2% cpu usage on a pentium 4.
int numPauses = (int) (timeUntilTrigger / spinInterval);
while (numPauses >= 0 && !signaled) {
try {
Thread.sleep(spinInterval);
} catch (InterruptedException ignore) {
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
numPauses = (int) (timeUntilTrigger / spinInterval);
}
if (signaled) {
try {
qsRsrcs.getJobStore().releaseAcquiredTrigger(
ctxt, trigger);
} catch (JobPersistenceException jpe) {
qs.notifySchedulerListenersError(
"An error occured while releasing trigger '"
+ trigger.getFullName() + "'",
jpe);
// db connection must have failed... keep
// retrying until it's up...
releaseTriggerRetryLoop(trigger);
} catch (RuntimeException e) {
getLog().error(
"releaseTriggerRetryLoop: RuntimeException "
+e.getMessage(), e);
// db connection must have failed... keep
// retrying until it's up...
releaseTriggerRetryLoop(trigger);
}
signaled = false;
continue;
}
// set trigger to 'executing'
TriggerFiredBundle bndle = null;
synchronized(pauseLock) {
if(!halted) {
try {
bndle = qsRsrcs.getJobStore().triggerFired(ctxt,
trigger);
} catch (SchedulerException se) {
qs.notifySchedulerListenersError(
"An error occured while firing trigger '"
+ trigger.getFullName() + "'", se);
} catch (RuntimeException e) {
getLog().error(
"RuntimeException while firing trigger " +
trigger.getFullName(), e);
// db connection must have failed... keep
// retrying until it's up...
releaseTriggerRetryLoop(trigger);
}
}
// it's possible to get 'null' if the trigger was paused,
// blocked, or other similar occurances that prevent it being
// fired at this time... or if the scheduler was shutdown (halted)
if (bndle == null) {
try {
qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt,
trigger);
} catch (SchedulerException se) {
qs.notifySchedulerListenersError(
"An error occured while releasing trigger '"
+ trigger.getFullName() + "'", se);
// db connection must have failed... keep retrying
// until it's up...
releaseTriggerRetryLoop(trigger);
}
continue;
}
// TODO: improvements:
//
// 2- make sure we can get a job runshell before firing trigger, or
// don't let that throw an exception (right now it never does,
// but the signature says it can).
// 3- acquire more triggers at a time (based on num threads available?)
JobRunShell shell = null;
try {
shell = qsRsrcs.getJobRunShellFactory().borrowJobRunShell();
shell.initialize(qs, bndle);
} catch (SchedulerException se) {
try {
qsRsrcs.getJobStore().triggeredJobComplete(ctxt,
trigger, bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);
} catch (SchedulerException se2) {
qs.notifySchedulerListenersError(
"An error occured while placing job's triggers in error state '"
+ trigger.getFullName() + "'", se2);
// db connection must have failed... keep retrying
// until it's up...
errorTriggerRetryLoop(bndle);
}
continue;
}
if (qsRsrcs.getThreadPool().runInThread(shell) == false) {
try {
// this case should never happen, as it is indicative of the
// scheduler being shutdown or a bug in the thread pool or
// a thread pool being used concurrently - which the docs
// say not to do...
getLog().error("ThreadPool.runInThread() return false!");
qsRsrcs.getJobStore().triggeredJobComplete(ctxt,
trigger, bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);
} catch (SchedulerException se2) {
qs.notifySchedulerListenersError(
"An error occured while placing job's triggers in error state '"
+ trigger.getFullName() + "'", se2);
// db connection must have failed... keep retrying
// until it's up...
releaseTriggerRetryLoop(trigger);
}
}
}
continue;
}
} else { // if(availTreadCount > 0)
continue; // should never happen, if threadPool.blockForAvailableThreads() follows contract
}
// this looping may seem a bit silly, but it's the current
// work-around
// for a dead-lock that can occur if the Thread.sleep() is replaced
// with
// a obj.wait() that gets notified when the signal is set...
// so to be able to detect the signal change without sleeping the
// entier
// getRandomizedIdleWaitTime(), we spin here... don't worry though,
// the
// CPU usage of this spinning can't even be measured on a pentium
// 4.
long now = System.currentTimeMillis();
long waitTime = now + getRandomizedIdleWaitTime();
long timeUntilContinue = waitTime - now;
long spinInterval = 10;
int numPauses = (int) (timeUntilContinue / spinInterval);
while (numPauses > 0 && !signaled) {
try {
Thread.sleep(10L);
} catch (InterruptedException ignore) {
}
now = System.currentTimeMillis();
timeUntilContinue = waitTime - now;
numPauses = (int) (timeUntilContinue / spinInterval);
}
} catch(RuntimeException re) {
getLog().error("Runtime error occured in main trigger firing loop.", re);
}
} // loop...
// drop references to scheduler stuff to aid garbage collection...
qs = null;
qsRsrcs = null;
}
public void errorTriggerRetryLoop(TriggerFiredBundle bndle) {
int retryCount = 0;
try {
while (!halted) {
try {
Thread.sleep(getDbFailureRetryInterval()); // retry every N
// seconds (the db
// connection must
// be failed)
retryCount++;
qsRsrcs.getJobStore().triggeredJobComplete(ctxt,
bndle.getTrigger(), bndle.getJobDetail(), Trigger.INSTRUCTION_SET_ALL_JOB_TRIGGERS_ERROR);
retryCount = 0;
break;
} catch (JobPersistenceException jpe) {
if(retryCount % 4 == 0) {
qs.notifySchedulerListenersError(
"An error occured while releasing trigger '"
+ bndle.getTrigger().getFullName() + "'", jpe);
}
} catch (RuntimeException e) {
getLog().error("releaseTriggerRetryLoop: RuntimeException "+e.getMessage(), e);
} catch (InterruptedException e) {
getLog().error("releaseTriggerRetryLoop: InterruptedException "+e.getMessage(), e);
}
}
} finally {
if(retryCount == 0) {
getLog().info("releaseTriggerRetryLoop: connection restored.");
}
}
}
public void releaseTriggerRetryLoop(Trigger trigger) {
int retryCount = 0;
try {
while (!halted) {
try {
Thread.sleep(getDbFailureRetryInterval()); // retry every N
// seconds (the db
// connection must
// be failed)
retryCount++;
qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt, trigger);
retryCount = 0;
break;
} catch (JobPersistenceException jpe) {
if(retryCount % 4 == 0) {
qs.notifySchedulerListenersError(
"An error occured while releasing trigger '"
+ trigger.getFullName() + "'", jpe);
}
} catch (RuntimeException e) {
getLog().error("releaseTriggerRetryLoop: RuntimeException "+e.getMessage(), e);
} catch (InterruptedException e) {
getLog().error("releaseTriggerRetryLoop: InterruptedException "+e.getMessage(), e);
}
}
} finally {
if(retryCount == 0) {
getLog().info("releaseTriggerRetryLoop: connection restored.");
}
}
}
public Log getLog() {
return log;
}
} // end of QuartzSchedulerThread