Author: alex.guizar(a)jboss.com Date: 2010-03-24 18:28:43 -0400 (Wed, 24 Mar 2010) New Revision: 6233 Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/default.jbpm.cfg.xml jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/async.xml jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/configuration.xml Log: SOA-1998: interpret JobExecutor.retries property as number of attempts to execute a failed job; document all job executor properties in "Asynchronous continuations" chapter Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutor.java 2010-03-24 22:28:43 UTC (rev 6233) @@ -28,25 +28,25 @@ private int retries = 1; protected int idleInterval; protected int maxIdleInterval; - /** @deprecated this field was never used */ + /** @deprecated property has no effect */ protected int historyMaxSize; protected int maxLockTime; protected int lockMonitorInterval; + /** @deprecated property has no effect */ protected int lockBufferTime; private ThreadGroup threadGroup; /** @deprecated call {@link #getThreads()} instead */ protected Map threads; + /** @deprecated call {@link #getThreads()} instead */ protected LockMonitorThread lockMonitorThread; protected Map monitoredJobIds = new Hashtable(); + protected boolean isStarted; - protected boolean isStarted = false; - - /** @deprecated this field should never have been exposed */ + /** @deprecated this field was just an aid for generating thread names */ protected static String hostName; - private static String hostAddress; public synchronized void start() { if (!isStarted) { @@ -190,7 +190,7 @@ void startLockMonitorThread() { String threadName = getLockMonitorThreadName(); - lockMonitorThread = new LockMonitorThread(threadName, this); + Thread lockMonitorThread = new LockMonitorThread(threadName, this); log.debug("starting " + threadName); lockMonitorThread.start(); @@ -201,15 +201,12 @@ } private static String getHostAddress() { - if (hostAddress == null) { - try { - hostAddress = InetAddress.getLocalHost().getHostAddress(); - } - catch (UnknownHostException e) { - hostAddress = "127.0.0.1"; - } + try { + return InetAddress.getLocalHost().getHostAddress(); } - return hostAddress; + catch (UnknownHostException e) { + return "127.0.0.1"; + } } public Set getMonitoredJobIds() { @@ -225,19 +222,21 @@ } /** - * @throws UnsupportedOperationException to prevent invocation - * @deprecated <code>monitoredJobIds</code> is an internal control field + * Method has no effect. + * + * @deprecated call {@link #addMonitoredJobId(String, long)} or + * {@link #removeMonitoredJobId(String)} to manipulate the set of monitored + * jobs */ public void setMonitoredJobIds(Map monitoredJobIds) { - throw new UnsupportedOperationException(); } - /** @deprecated this property was never used */ + /** @deprecated property has no effect */ public int getHistoryMaxSize() { return historyMaxSize; } - /** @deprecated this property was never used */ + /** @deprecated property has no effect */ public void setHistoryMaxSize(int historyMaxSize) { this.historyMaxSize = historyMaxSize; } @@ -258,11 +257,12 @@ } /** - * @throws UnsupportedOperationException to prevent invocation - * @deprecated <code>isStarted</code> is an internal control field + * This method has no effect. + * + * @deprecated call {@link #start()} or {@link #stop()} to control this job + * executor. */ public void setStarted(boolean isStarted) { - throw new UnsupportedOperationException(); } public JbpmConfiguration getJbpmConfiguration() { @@ -331,10 +331,12 @@ this.maxLockTime = maxLockTime; } + /** @deprecated property has no effect */ public int getLockBufferTime() { return lockBufferTime; } + /** @deprecated property has no effect */ public void setLockBufferTime(int lockBufferTime) { this.lockBufferTime = lockBufferTime; } Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/JobExecutorThread.java 2010-03-24 22:28:43 UTC (rev 6233) @@ -21,23 +21,13 @@ public class JobExecutorThread extends Thread { - final JobExecutor jobExecutor; - final JbpmConfiguration jbpmConfiguration; - final int idleInterval; - final int maxIdleInterval; - final long maxLockTime; - - volatile boolean isActive = true; + private final JobExecutor jobExecutor; + private volatile boolean isActive = true; private final Random random = new Random(); public JobExecutorThread(String name, JobExecutor jobExecutor) { super(jobExecutor.getThreadGroup(), name); this.jobExecutor = jobExecutor; - - jbpmConfiguration = jobExecutor.getJbpmConfiguration(); - idleInterval = jobExecutor.getIdleInterval(); - maxIdleInterval = jobExecutor.getMaxIdleInterval(); - maxLockTime = jobExecutor.getMaxLockTime(); } /** @@ -48,71 +38,61 @@ long maxLockTime, int maxHistory) { super(jobExecutor.getThreadGroup(), name); this.jobExecutor = jobExecutor; - - this.jbpmConfiguration = jbpmConfiguration; - this.idleInterval = idleInterval; - this.maxIdleInterval = maxIdleInterval; - this.maxLockTime = maxLockTime; } public void run() { - int currentIdleInterval = idleInterval; + int currentIdleInterval = jobExecutor.getIdleInterval(); while (isActive) { - try { - Collection acquiredJobs = acquireJobs(); - for (Iterator i = acquiredJobs.iterator(); i.hasNext() && isActive;) { - Job job = (Job) i.next(); - try { - executeJob(job); - } - catch (Exception e) { - saveJobException(job, e); - throw e; - } + // acquire jobs; on exception, call returns empty collection + Collection acquiredJobs = acquireJobs(); + // execute jobs + boolean success = true; + for (Iterator i = acquiredJobs.iterator(); i.hasNext() && isActive;) { + Job job = (Job) i.next(); + try { + executeJob(job); } - if (isActive) { - long waitPeriod = getWaitPeriod(currentIdleInterval); - if (waitPeriod > 0) { - synchronized (jobExecutor) { - jobExecutor.wait(waitPeriod); - } - } + catch (Exception e) { + // on exception, call returns normally + saveJobException(job, e); + success = false; + break; } - // no exception, reset current idle interval - currentIdleInterval = idleInterval; } - catch (InterruptedException e) { - if (log.isDebugEnabled()) log.debug(getName() + " got interrupted"); - } - catch (Exception e) { - // if this is a locking exception, keep it quiet - if (DbPersistenceService.isLockingException(e)) { - StaleObjectLogConfigurer.getStaleObjectExceptionsLog() - .error("job execution failure", e); - } - else { - log.error("job execution failure", e); - } - // wait after error - if (isActive) { - // wait a random period, at least half the current idle interval - int waitPeriod = currentIdleInterval / 2; - waitPeriod = random.nextInt(waitPeriod) + waitPeriod; - try { - // sleep instead of waiting on jobExecutor to prevent - // DbMessageService from waking up this thread - sleep(waitPeriod); - // after an exception, double the current idle interval to avoid - // throwing exceptions continuously during anomalous conditions + + // if still active wait for next batch + if (isActive) { + try { + if (success) { + // reset current idle interval + currentIdleInterval = jobExecutor.getIdleInterval(); + // wait for next due job + long waitPeriod = getWaitPeriod(currentIdleInterval); + if (waitPeriod > 0) { + synchronized (jobExecutor) { + jobExecutor.wait(waitPeriod); + } + } + } + else { + // wait a random period, at least half the current idle interval + int waitPeriod = currentIdleInterval / 2; + // sleep instead of waiting on jobExecutor + // to prevent DbMessageService from waking up this thread + sleep(waitPeriod + random.nextInt(waitPeriod)); + // after an exception, double the current idle interval + // to avoid continuous failures during anomalous conditions currentIdleInterval *= 2; + // enforce maximum idle interval + int maxIdleInterval = jobExecutor.getMaxIdleInterval(); if (currentIdleInterval > maxIdleInterval || currentIdleInterval < 0) { currentIdleInterval = maxIdleInterval; } } - catch (InterruptedException ie) { - if (log.isDebugEnabled()) log.debug(getName() + " got interrupted"); - } } + catch (InterruptedException e) { + if (log.isDebugEnabled()) log.debug(getName() + " got interrupted"); + } } } log.info(getName() + " leaves cyberspace"); @@ -123,7 +103,7 @@ Collection jobs; // acquire monitor before creating context and allocating resources synchronized (jobExecutor) { - JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); + JbpmContext jbpmContext = jobExecutor.getJbpmConfiguration().createJbpmContext(); try { String lockOwner = getName(); JobSession jobSession = jbpmContext.getJobSession(); @@ -140,13 +120,21 @@ } Date lockTime = new Date(); - int retries = jobExecutor.getRetries(); for (Iterator i = jobs.iterator(); i.hasNext();) { Job job = (Job) i.next(); job.setLockOwner(lockOwner); job.setLockTime(lockTime); - // if job has not failed already, give it the configured retries - if (job.getException() == null) job.setRetries(retries); + // has job failed previously? + if (job.getException() == null) { + // initialize retry count to configured value + job.setRetries(jobExecutor.getRetries()); + } + else { + // decrease retry count + int retries = job.getRetries() - 1; + job.setRetries(retries); + if (debug) log.debug(job + " has " + retries + " retries remaining"); + } } if (debug) log.debug("acquired " + jobs); } @@ -178,7 +166,7 @@ } protected void executeJob(Job job) throws Exception { - JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); + JbpmContext jbpmContext = jobExecutor.getJbpmConfiguration().createJbpmContext(); try { JobSession jobSession = jbpmContext.getJobSession(); jobSession.reattachJob(job); @@ -188,38 +176,35 @@ jbpmContext.addAutoSaveProcessInstance(job.getProcessInstance()); if (log.isDebugEnabled()) log.debug("executing " + job); - try { - if (job.execute(jbpmContext)) { - jobSession.deleteJob(job); - } - } - catch (Exception e) { - // prevent unsafe use of the session after an exception occurs - jbpmContext.setRollbackOnly(); - throw e; - } - catch (Error e) { - jbpmContext.setRollbackOnly(); - throw e; - } - - // if this job is locked too long - long totalLockTimeInMillis = System.currentTimeMillis() - job.getLockTime().getTime(); - if (totalLockTimeInMillis > maxLockTime) { - jbpmContext.setRollbackOnly(); - } + if (job.execute(jbpmContext)) jobSession.deleteJob(job); } + catch (Exception e) { + jbpmContext.setRollbackOnly(); + throw e; + } + catch (Error e) { + jbpmContext.setRollbackOnly(); + throw e; + } finally { jbpmContext.close(); } } private void saveJobException(Job job, Exception exception) { + // if this is a locking exception, keep it quiet + if (DbPersistenceService.isLockingException(exception)) { + StaleObjectLogConfigurer.getStaleObjectExceptionsLog() + .error("failed to execute " + job, exception); + } + else { + log.error("failed to execute " + job, exception); + } + boolean debug = log.isDebugEnabled(); - JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); + JbpmContext jbpmContext = jobExecutor.getJbpmConfiguration().createJbpmContext(); try { - if (debug) log.debug("saving exception thrown by " + job); - // do not reattach existing job, it may contain undesired updates + // do not reattach existing job as it contains undesired updates JobSession jobSession = jbpmContext.getJobSession(); job = jobSession.loadJob(job.getId()); @@ -227,11 +212,6 @@ StringWriter out = new StringWriter(); exception.printStackTrace(new PrintWriter(out)); job.setException(out.toString()); - - // decrease retry count - int retries = job.getRetries() - 1; - job.setRetries(retries); - if (debug) log.debug(job + " has " + retries + " retries remaining"); } catch (RuntimeException e) { jbpmContext.setRollbackOnly(); @@ -261,21 +241,25 @@ } protected Date getNextDueDate() { - Date nextDueDate = null; - String lockOwner = getName(); - JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); + Date nextDueDate; + JbpmContext jbpmContext = jobExecutor.getJbpmConfiguration().createJbpmContext(); try { - JobSession jobSession = jbpmContext.getJobSession(); - Collection jobIdsToIgnore = jobExecutor.getMonitoredJobIds(); - Job job = jobSession.getFirstDueJob(lockOwner, jobIdsToIgnore); + String lockOwner = getName(); + Job job = jbpmContext.getJobSession() + .getFirstDueJob(lockOwner, jobExecutor.getMonitoredJobIds()); if (job != null) { - nextDueDate = job.getDueDate(); jobExecutor.addMonitoredJobId(lockOwner, job.getId()); + nextDueDate = job.getDueDate(); } + else { + nextDueDate = null; + if (log.isDebugEnabled()) log.debug("no due job found"); + } } catch (RuntimeException e) { jbpmContext.setRollbackOnly(); - throw e; + nextDueDate = null; + if (log.isDebugEnabled()) log.debug("failed to determine next due date", e); } catch (Error e) { jbpmContext.setRollbackOnly(); @@ -286,11 +270,8 @@ jbpmContext.close(); } catch (RuntimeException e) { - if (!DbPersistenceService.isLockingException(e)) throw e; - // keep locking exception quiet - StaleObjectLogConfigurer.getStaleObjectExceptionsLog() - .error("failed to determine next due date", e); nextDueDate = null; + if (log.isDebugEnabled()) log.debug("failed to determine next due date", e); } } return nextDueDate; @@ -304,7 +285,8 @@ } /** - * Indicates that this thread should stop running. Execution will cease shortly afterwards. + * Indicates that this thread should stop running. Execution will cease + * shortly afterwards. */ public void deactivate() { if (isActive) { @@ -313,5 +295,5 @@ } } - private static Log log = LogFactory.getLog(JobExecutorThread.class); + private static final Log log = LogFactory.getLog(JobExecutorThread.class); } Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/java/org/jbpm/job/executor/LockMonitorThread.java 2010-03-24 22:28:43 UTC (rev 6233) @@ -17,32 +17,25 @@ public static final String DEFAULT_NAME = "Monitor"; - final JbpmConfiguration jbpmConfiguration; - final int lockMonitorInterval; - final int maxLockTime; - final int lockBufferTime; + private final JobExecutor jobExecutor; + private volatile boolean isActive = true; - volatile boolean isActive = true; - public LockMonitorThread(JobExecutor jobExecutor) { this(DEFAULT_NAME, jobExecutor); } public LockMonitorThread(String name, JobExecutor jobExecutor) { super(jobExecutor.getThreadGroup(), name); - jbpmConfiguration = jobExecutor.getJbpmConfiguration(); - lockMonitorInterval = jobExecutor.getLockMonitorInterval(); - maxLockTime = jobExecutor.getMaxLockTime(); - lockBufferTime = jobExecutor.getLockBufferTime(); + this.jobExecutor = jobExecutor; } - /** @deprecated As of jBPM 3.2.6, replaced by {@link #LockMonitorThread(JobExecutor)} */ + /** + * @deprecated As of jBPM 3.2.6, replaced by + * {@link #LockMonitorThread(JobExecutor)} + */ public LockMonitorThread(JbpmConfiguration jbpmConfiguration, int lockMonitorInterval, - int maxLockTime, int lockBufferTime) { - this.jbpmConfiguration = jbpmConfiguration; - this.lockMonitorInterval = lockMonitorInterval; - this.maxLockTime = maxLockTime; - this.lockBufferTime = lockBufferTime; + int maxLockTime, int lockBufferTime) { + jobExecutor = jbpmConfiguration.getJobExecutor(); } public void run() { @@ -55,7 +48,7 @@ } if (isActive) { try { - sleep(lockMonitorInterval); + sleep(jobExecutor.getLockMonitorInterval()); } catch (InterruptedException e) { log.info(getName() + " got interrupted"); @@ -66,15 +59,13 @@ } protected void unlockOverdueJobs() { - JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(); + JbpmContext jbpmContext = jobExecutor.getJbpmConfiguration().createJbpmContext(); try { - Date threshold = new Date(System.currentTimeMillis() - maxLockTime - lockBufferTime); + Date threshold = new Date(System.currentTimeMillis() - jobExecutor.getMaxLockTime()); List overdueJobs = jbpmContext.getJobSession().findJobsWithOverdueLockTime(threshold); for (Iterator i = overdueJobs.iterator(); i.hasNext();) { Job job = (Job) i.next(); - if (log.isDebugEnabled()) { - log.debug("unlocking " + job + " owned by " + job.getLockOwner()); - } + if (log.isDebugEnabled()) log.debug("unlocking " + job); job.setLockOwner(null); job.setLockTime(null); } @@ -84,7 +75,7 @@ if (!DbPersistenceService.isLockingException(e)) throw e; // keep locking exception quiet StaleObjectLogConfigurer.getStaleObjectExceptionsLog() - .error("could not unlock overdue jobs", e); + .error("could not unlock overdue jobs", e); } catch (Error e) { jbpmContext.setRollbackOnly(); @@ -98,7 +89,7 @@ if (!DbPersistenceService.isLockingException(e)) throw e; // keep locking exception quiet StaleObjectLogConfigurer.getStaleObjectExceptionsLog() - .error("could not unlock overdue jobs", e); + .error("could not unlock overdue jobs", e); } } } @@ -111,7 +102,8 @@ } /** - * Indicates that this thread should stop running. Execution will cease shortly afterwards. + * Indicates that this thread should stop running. Execution will cease + * shortly afterwards. */ public void deactivate() { if (isActive) { Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/db/hibernate.queries.hbm.xml 2010-03-24 22:28:43 UTC (rev 6233) @@ -10,7 +10,7 @@ <query name="GraphSession.AllProcessNames"> <![CDATA[ - select distinct (pd.name) + select distinct pd.name from org.jbpm.graph.def.ProcessDefinition pd ]]> </query> @@ -33,7 +33,7 @@ <query name="GraphSession.findLatestProcessDefinitionQuery"> <![CDATA[ select pd - from org.jbpm.graph.def.ProcessDefinition as pd + from org.jbpm.graph.def.ProcessDefinition pd where pd.name = :name order by pd.version desc ]]> @@ -42,7 +42,7 @@ <query name="GraphSession.findProcessDefinitionByNameAndVersion"> <![CDATA[ select pd - from org.jbpm.graph.def.ProcessDefinition as pd + from org.jbpm.graph.def.ProcessDefinition pd where pd.name = :name and pd.version = :version ]]> @@ -51,7 +51,7 @@ <query name="GraphSession.findAllProcessDefinitions"> <![CDATA[ select pd - from org.jbpm.graph.def.ProcessDefinition as pd + from org.jbpm.graph.def.ProcessDefinition pd order by pd.name, pd.version desc ]]> </query> @@ -59,7 +59,7 @@ <query name="GraphSession.findAllProcessDefinitionVersions"> <![CDATA[ select pd - from org.jbpm.graph.def.ProcessDefinition as pd + from org.jbpm.graph.def.ProcessDefinition pd where pd.name = :name order by pd.version desc ]]> @@ -68,16 +68,15 @@ <query name="GraphSession.findLatestProcessDefinitions"> <![CDATA[ select pd.name, max(pd.version) - from org.jbpm.graph.def.ProcessDefinition as pd + from org.jbpm.graph.def.ProcessDefinition pd group by pd.name ]]> - <!-- better alternative, if your database supports ansi sql tuple syntax --> - <!-- + <!-- better alternative, if all databases supported ansi sql tuple syntax select pd - from org.jbpm.graph.def.ProcessDefinition as pd - where ( pd.name, pd.version ) in ( + from org.jbpm.graph.def.ProcessDefinition pd + where (pd.name, pd.version) in ( select pd.name, max(pd.version) - from org.jbpm.graph.def.ProcessDefinition as pd + from org.jbpm.graph.def.ProcessDefinition pd group by pd.name ) --> @@ -86,7 +85,7 @@ <query name="GraphSession.findAllProcessInstancesForDefinition"> <![CDATA[ select pi - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi where pi.processDefinition.id = :processDefinitionId order by pi.start desc ]]> @@ -95,7 +94,7 @@ <query name="GraphSession.findAllProcessInstanceIdsForDefinition"> <![CDATA[ select pi.id - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi where pi.processDefinition.id = :processDefinitionId ]]> </query> @@ -103,7 +102,7 @@ <query name="GraphSession.findReferencingProcessStates"> <![CDATA[ select ps - from org.jbpm.graph.node.ProcessState as ps + from org.jbpm.graph.node.ProcessState ps where ps.subProcessDefinition = :subProcessDefinition ]]> </query> @@ -114,7 +113,7 @@ <query name="GraphSession.findAllRunningProcessInstancesWithProcessName"> <![CDATA[ select pi - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi where pi.processDefinition.name = :processDefinitionName and pi.end is null order by pi.start desc @@ -124,7 +123,7 @@ <query name="GraphSession.findAllProcessInstancesWithProcessName"> <![CDATA[ select pi - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi where pi.processDefinition.name = :processDefinitionName order by pi.start desc ]]> @@ -133,7 +132,7 @@ <query name="GraphSession.findAllRunningProcessInstances"> <![CDATA[ select pi - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi where pi.end is null order by pi.start desc ]]> @@ -142,7 +141,7 @@ <query name="GraphSession.findAllProcessInstances"> <![CDATA[ select pi - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi order by pi.start desc ]]> </query> @@ -150,7 +149,7 @@ <query name="GraphSession.findSubProcessInstances"> <![CDATA[ select pi - from org.jbpm.graph.exe.ProcessInstance as pi + from org.jbpm.graph.exe.ProcessInstance pi where pi.superProcessToken is not null and pi.superProcessToken.processInstance = :processInstance order by pi.start desc @@ -188,7 +187,7 @@ <query name="GraphSession.findLogsForProcessInstance"> <![CDATA[ select pl - from org.jbpm.logging.log.ProcessLog as pl + from org.jbpm.logging.log.ProcessLog pl where pl.token.processInstance = :processInstance ]]> </query> @@ -228,11 +227,9 @@ <query name="GraphSession.findActiveNodesByProcessInstance"> <![CDATA[ select node - from org.jbpm.graph.def.Node node, - org.jbpm.graph.exe.Token token, - org.jbpm.graph.exe.ProcessInstance processInstance + from org.jbpm.graph.exe.Token token + join token.node node where token.processInstance = :processInstance - and token.node = node and token.isSuspended = false and token.parent is not null and token.end is null @@ -242,7 +239,7 @@ <query name="GraphSession.findAllSwimlaneNames"> <![CDATA[ select distinct swimlane.name - from org.jbpm.taskmgmt.def.Swimlane as swimlane + from org.jbpm.taskmgmt.def.Swimlane swimlane ]]> </query> @@ -252,7 +249,7 @@ <query name="LoggingSession.findLogsByToken"> <![CDATA[ select pl - from org.jbpm.logging.log.ProcessLog as pl + from org.jbpm.logging.log.ProcessLog pl where pl.token = :token order by pl.index ]]> @@ -264,7 +261,7 @@ <query name="JobSession.getFirstAcquirableJob"> <![CDATA[ select job - from org.jbpm.job.Job as job + from org.jbpm.job.Job job where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.retries > 0 and job.dueDate <= :now @@ -276,7 +273,7 @@ <query name="JobSession.findExclusiveJobs"> <![CDATA[ select job - from org.jbpm.job.Job as job + from org.jbpm.job.Job job where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.retries > 0 and job.dueDate <= :now @@ -290,7 +287,7 @@ <query name="JobSession.getFirstDueJob"> <![CDATA[ select job - from org.jbpm.job.Job as job + from org.jbpm.job.Job job where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.retries > 0 and job.isSuspended = false @@ -301,7 +298,7 @@ <query name="JobSession.getFirstDueJobExcludingMonitoredJobs"> <![CDATA[ select job - from org.jbpm.job.Job as job + from org.jbpm.job.Job job where (job.lockOwner is null or job.lockOwner = :lockOwner) and job.id not in (:monitoredJobIds) and job.retries > 0 @@ -384,7 +381,7 @@ <query name="JobSession.findJobsWithOverdueLockTime"> <![CDATA[ select job - from org.jbpm.job.Job as job + from org.jbpm.job.Job job where job.lockTime < :threshold ]]> </query> @@ -395,7 +392,7 @@ <query name="TaskMgmtSession.findTaskInstancesByActorId"> <![CDATA[ select ti - from org.jbpm.taskmgmt.exe.TaskInstance as ti + from org.jbpm.taskmgmt.exe.TaskInstance ti where ti.actorId = :actorId and ti.isSuspended = false and ti.isOpen = true @@ -405,7 +402,7 @@ <query name="TaskMgmtSession.findTaskInstancesByActorIds"> <![CDATA[ select ti - from org.jbpm.taskmgmt.exe.TaskInstance as ti + from org.jbpm.taskmgmt.exe.TaskInstance ti where ti.actorId in (:actorIds) and ti.isSuspended = false and ti.isOpen = true Modified: jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/default.jbpm.cfg.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/default.jbpm.cfg.xml 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/core/src/main/resources/org/jbpm/default.jbpm.cfg.xml 2010-03-24 22:28:43 UTC (rev 6233) @@ -66,12 +66,11 @@ <property name="jbpmConfiguration"><ref bean="jbpmConfiguration" /></property> <property name="name"><string value="JbpmJobExecutor" /></property> <property name="nbrOfThreads"><int value="1" /></property> - <property name="retries"><int value="3" /></property> - <property name="idleInterval"><int value="5000" /></property> - <property name="maxIdleInterval"><int value="3600000" /><!-- 1 hour --></property> - <property name="maxLockTime"><int value="600000" /><!-- 10 minutes --></property> - <property name="lockMonitorInterval"><int value="60000" /><!-- 1 minute --></property> - <property name="lockBufferTime"><int value="5000" /><!-- 5 seconds --></property> + <property name="retries"><int value="2" /></property> + <property name="idleInterval"><!-- 5 seconds --><int value="5000" /></property> + <property name="maxIdleInterval"><!-- 1 hour --><int value="3600000" /></property> + <property name="maxLockTime"><!-- 10 minutes --><int value="600000" /></property> + <property name="lockMonitorInterval"><!-- 1 minute --><int value="60000" /></property> </bean> </jbpm-configuration> Modified: jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/async.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/async.xml 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/async.xml 2010-03-24 22:28:43 UTC (rev 6233) @@ -7,14 +7,13 @@ <para>jBPM is based on Graph Oriented Programming (GOP). Basically, GOP specifies a simple state machine that can handle concurrent paths of execution. But in the execution algorithm specified in GOP, all state transitions are done in a single operation in the - thread of the client. If you're not familiar with the execution algorithm defined in - <xref linkend="graphorientedprogramming" />, please read that first. By default, this + thread of the client. By default, this performing state transitions in the thread of the client is a good approach cause it fits naturally with server side transactions. The process execution moves from one wait state to another wait state in one transaction. </para> - <para>But in some situations, a developer might want to fine-tune the transaction + <para>In some situations, a developer might want to fine-tune the transaction demarcation in the process definition. In jPDL, it is possible to specify that the process execution should continue asynchronously with the attribute <literal>async=&quot;true&quot;</literal>. <literal>async=&quot;true&quot;</literal> can be specified on all node types and all action types. @@ -88,17 +87,18 @@ <para>In jBPM, asynchronous continuations are realized by using an asynchronous messaging system. When the process execution arrives at a point that should be executed - asynchronously, jBPM will suspend the execution, produces a command message and send it + asynchronously, jBPM will suspend the execution, produces a job message and send it to the command executor. The command executor is a separate component that, upon receipt of a message, will resume the execution of the process where it got suspended. </para> <para>jBPM can be configured to use a JMS provider or its built-in asynchronous messaging - system. The built-in messaging system is quite limited in functionality, but allowes + system. The built-in messaging system is quite limited in functionality, but allows this feature to be supported on environments where JMS is unavailable. </para> </section> + <section id="thejobexecutor"> <title>The job executor</title> @@ -110,7 +110,7 @@ <para>These job messages are produced by the process execution. During process execution, for each node or action that has to be executed asynchronously, a <literal>Job</literal> - (POJO) will be dispatched to the <literal>MessageService</literal>. The message service + object will be dispatched to the <literal>MessageService</literal>. The message service is associated with the <literal>JbpmContext</literal> and it just collects all the messages that have to be sent. </para> @@ -143,7 +143,7 @@ acquires a job by putting its name into the owner field of the job. Each thread has a unique name based on ip-address and sequence number. Hibernate's optimistic locking is enabled on <literal>Job</literal>-objects. So if 2 threads try to acquire a job concurrently, one of - them will get a StaleObjectException and rollback. Only the first one will succeed. The + them will get a <literal>StaleStateException</literal> and rollback. Only the first one will succeed. The thread that succeeds in acquiring a job is now responsible for executing it in a separate transaction. </para> @@ -178,34 +178,104 @@ </para> </section> + <section id="configuringjobexecutor"> + <title>Configuring the job executor</title> + + <para>The job executor exposes a few properties you can tune. To change any of the + values described below, edit the <literal>jbpm.job.executor</literal> bean in + <literal>jbpm.cfg.xml</literal>.</para> + + <table><title>Job executor properties</title> + <tgroup cols='3'> + <colspec colname='attr' colwidth="*" /> + <colspec colname='desc' colwidth="4*" /> + <colspec colname="def" colwidth="*"/> + + <thead> + <row> + <entry>Property</entry> + <entry>Description</entry> + <entry>Default</entry> + </row> + </thead> + + <![CDATA[ + <property name="nbrOfThreads"><int value="1" /></property> + <property name="retries"><int value="3" /></property> + <property name="idleInterval"><int value="5000" /></property> + <property name="maxIdleInterval"><int value="3600000" /><!-- 1 hour --></property> + <property name="maxLockTime"><int value="600000" /><!-- 10 minutes --></property> + <property name="lockMonitorInterval"><int value="60000" /><!-- 1 minute --></property> + <property name="lockBufferTime"><int value="5000" /><!-- 5 seconds --></property>]]> + + <tbody> + <row> + <entry>nbrOfThreads</entry> + <entry>Size of the job executor thread pool</entry> + <entry>1</entry> + </row> + <row> + <entry>retries</entry> + <entry>Number of times the job executor retries a failed job</entry> + <entry>2</entry> + </row> + <row> + <entry>idleInterval</entry> + <entry>Period between checks for new jobs (milliseconds)</entry> + <entry>5 seconds</entry> + </row> + <row> + <entry>maxIdleInterval</entry> + <entry>When a job fails, the affected thread pauses for a period initially equal + to <literal>idleInterval</literal>, which is increased twofold until it reaches + <literal>maxIdleInterval</literal> (milliseconds)</entry> + <entry>1 hour</entry> + </row> + <row> + <entry>maxLockTime</entry> + <entry>Amount of time a job executor thread is allowed to hold a job before + the job is released and offered to other threads (milliseconds)</entry> + <entry>10 minutes</entry> + </row> + <row> + <entry>lockMonitorInterval</entry> + <entry>Period between checks for job lock times (milliseconds)</entry> + <entry>1 minute</entry> + </row> + </tbody> + </tgroup> + </table> + + </section> + <section id="jbpmsbuiltinasynchronousmessaging"> <title>jBPM's built-in asynchronous messaging</title> <para>When using jBPM's built-in asynchronous messaging, job messages will be sent by persisting - them to the database. This message persisting can be done in the same transaction/JDBC + them to the database. This message persisting is done in the same transaction/JDBC connection as the jBPM process updates. </para> <para>The job messages will be stored in the <literal>JBPM_JOB</literal> table. </para> - <para>The POJO command executor (<literal>org.jbpm.msg.command.CommandExecutor</literal>) + <para>The job executor (<literal>org.jbpm.job.executor.JobExecutor</literal>) will read the messages from the database table and execute them. So the typical - transaction of the POJO command executor looks like this: 1) read next command message - 2) execute command message 3) delete command message. + transaction of the job executor looks like this: 1) read next job message + 2) execute job 3) delete job message. </para> - <para>If execution of a command message fails, the transaction will be rolled back. - After that, a new transaction will be started that adds the error message to the - message in the database. The command executor filters out all messages that contain + <para>If execution of a job fails, the transaction will be rolled back. + After that, a new transaction will be started that adds the exception stack trace to the + job message in the database. The job executor filters out all messages that contain an exception. </para> - <figure id="pojo.command.executor.image"> - <title>POJO command executor transactions</title> + <figure id="job.executor.image"> + <title>Job executor transactions</title> <mediaobject><imageobject><imagedata align="center" fileref="images/pojo.command.executor.gif"/></imageobject></mediaobject> </figure> <para>If for some reason or another, the transaction that adds the exception to the - command message would fail, it is rolled back as well. In that case, the - message remains in the queue without an exception so it will be retried later. + job message would fail, it is rolled back as well. In that case, the + message remains in the database without an exception so it will be retried later. </para> <para>Limitation: beware that jBPM's built-in asynchronous messaging system does - not support multinode locking. So you cannot just deploy the POJO command + not support multinode locking. So you cannot just deploy the job executor multiple times and have them configured to use the same database. </para> </section> Modified: jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/configuration.xml =================================================================== --- jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/configuration.xml 2010-03-19 11:20:08 UTC (rev 6232) +++ jbpm3/branches/jbpm-3.2-soa/modules/userguide/src/main/docbook/en/modules/configuration.xml 2010-03-24 22:28:43 UTC (rev 6233) @@ -44,50 +44,70 @@ <para>Here's the default configuration for the JbpmContext: </para> -<programlisting>&lt;jbpm-configuration&gt; +<programlisting><![CDATA[<jbpm-configuration> + + <jbpm-context> + <service name="authentication" factory="org.jbpm.security.authentication.DefaultAuthenticationServiceFactory" /> + <service name="logging" factory="org.jbpm.logging.db.DbLoggingServiceFactory" /> + <service name="message" factory="org.jbpm.msg.db.DbMessageServiceFactory" /> + <service name="persistence" factory="org.jbpm.persistence.db.DbPersistenceServiceFactory" /> + <service name="scheduler" factory="org.jbpm.scheduler.db.DbSchedulerServiceFactory" /> + <service name="tx" factory="org.jbpm.tx.TxServiceFactory" /> + </jbpm-context> - &lt;jbpm-context&gt; - &lt;service name='persistence' factory='org.jbpm.persistence.db.DbPersistenceServiceFactory' /&gt; - &lt;service name='message' factory='org.jbpm.msg.db.DbMessageServiceFactory' /&gt; - &lt;service name='scheduler' factory='org.jbpm.scheduler.db.DbSchedulerServiceFactory' /&gt; - &lt;service name='logging' factory='org.jbpm.logging.db.DbLoggingServiceFactory' /&gt; - &lt;service name='authentication' factory='org.jbpm.security.authentication.DefaultAuthenticationServiceFactory' /&gt; - &lt;/jbpm-context&gt; + <!-- database persistence configuration --> + <string name="resource.hibernate.cfg.xml" value="hibernate.cfg.xml" /> + <!-- additional database persistence properties + <string name="resource.hibernate.properties" value="hibernate.properties" /> + --> - &lt;!-- configuration resource files pointing to default configuration files in jbpm-{version}.jar --&gt; - &lt;string name='resource.hibernate.cfg.xml' value='hibernate.cfg.xml' /&gt; + <!-- entries pointing to default configuration resources in jbpm-jpdl.jar --> + <string name="resource.business.calendar" value="org/jbpm/calendar/jbpm.business.calendar.properties" /> + <string name="resource.default.modules" value="org/jbpm/graph/def/jbpm.default.modules.properties" /> + <string name="resource.converter" value="org/jbpm/db/hibernate/jbpm.converter.properties" /> + <string name="resource.action.types" value="org/jbpm/graph/action/action.types.xml" /> + <string name="resource.node.types" value="org/jbpm/graph/node/node.types.xml" /> + <string name="resource.parsers" value="org/jbpm/jpdl/par/jbpm.parsers.xml" /> + <string name="resource.varmapping" value="org/jbpm/context/exe/jbpm.varmapping.xml" /> + <string name="resource.mail.templates" value="jbpm.mail.templates.xml" /> - &lt;!-- &lt;string name='resource.hibernate.properties' value='hibernate.properties' /&gt; --&gt; - &lt;string name='resource.business.calendar' value='org/jbpm/calendar/jbpm.business.calendar.properties' /&gt; - &lt;string name='resource.default.modules' value='org/jbpm/graph/def/jbpm.default.modules.properties' /&gt; - &lt;string name='resource.converter' value='org/jbpm/db/hibernate/jbpm.converter.properties' /&gt; - &lt;string name='resource.action.types' value='org/jbpm/graph/action/action.types.xml' /&gt; - &lt;string name='resource.node.types' value='org/jbpm/graph/node/node.types.xml' /&gt; - &lt;string name='resource.parsers' value='org/jbpm/jpdl/par/jbpm.parsers.xml' /&gt; - &lt;string name='resource.varmapping' value='org/jbpm/context/exe/jbpm.varmapping.xml' /&gt; - &lt;string name='resource.mail.templates' value='jbpm.mail.templates.xml' /&gt; + <!-- alternate delegation class loader + <string name="jbpm.class.loader" value="context" /> + --> + <bean name="process.class.loader.factory" class="org.jbpm.instantiation.SharedProcessClassLoaderFactory" singleton="true" /> + <bean name="jbpm.task.instance.factory" class="org.jbpm.taskmgmt.impl.DefaultTaskInstanceFactoryImpl" singleton="true" /> - &lt;int name='jbpm.byte.block.size' value="1024" singleton="true" /&gt; - &lt;bean name='jbpm.task.instance.factory' class='org.jbpm.taskmgmt.impl.DefaultTaskInstanceFactoryImpl' singleton='true' /&gt; - &lt;bean name='jbpm.variable.resolver' class='org.jbpm.jpdl.el.impl.JbpmVariableResolver' singleton='true' /&gt; - &lt;string name='jbpm.mail.smtp.host' value='localhost' /&gt; - &lt;bean name='jbpm.mail.address.resolver' class='org.jbpm.identity.mail.IdentityAddressResolver' singleton='true' /&gt; - &lt;string name='jbpm.mail.from.address' value='jbpm@noreply' /&gt; + <!-- make sure the block size matches the length in ByteArray.hbm.xml --> + <int name="jbpm.byte.block.size" value="1024" singleton="true" /> - &lt;bean name='jbpm.job.executor' class='org.jbpm.job.executor.JobExecutor'&gt; - &lt;field name='jbpmConfiguration'&gt;&lt;ref bean='jbpmConfiguration' /&gt;&lt;/field&gt; - &lt;field name='name'&gt;&lt;string value='JbpmJobExecutor' /&gt;&lt;/field&gt; - &lt;field name='nbrOfThreads'&gt;&lt;int value='1' /&gt;&lt;/field&gt; - &lt;field name='idleInterval'&gt;&lt;int value='5000' /&gt;&lt;/field&gt; - &lt;field name='maxIdleInterval'&gt;&lt;int value='3600000' /&gt;&lt;/field&gt; &lt;!-- 1 hour --&gt; - &lt;field name='historyMaxSize'&gt;&lt;int value='20' /&gt;&lt;/field&gt; - &lt;field name='maxLockTime'&gt;&lt;int value='600000' /&gt;&lt;/field&gt; &lt;!-- 10 minutes --&gt; - &lt;field name='lockMonitorInterval'&gt;&lt;int value='60000' /&gt;&lt;/field&gt; &lt;!-- 1 minute --&gt; - &lt;field name='lockBufferTime'&gt;&lt;int value='5000' /&gt;&lt;/field&gt; &lt;!-- 5 seconds --&gt; - &lt;/bean&gt; + <bean name="jbpm.expression.evaluator" class="org.jbpm.jpdl.el.impl.ExpressionEvaluatorImpl" singleton="true" /> + <bean name="jbpm.variable.resolver" class="org.jbpm.jpdl.el.impl.JbpmVariableResolver" singleton="true" /> + <!-- custom function mapper + <bean name="jbpm.function.mapper" class="org.example.FunctionMapperImpl" /> + --> + <!-- user code interceptor + <bean name="jbpm.user.code.interceptor" class="org.example.UserCodeInterceptorImpl" /> + --> -&lt;/jbpm-configuration&gt;</programlisting> + <!-- mail settings --> + <string name="jbpm.mail.smtp.host" value="localhost" /> + <string name="jbpm.mail.from.address" value="jbpm@noreply" /> + <bean name="jbpm.mail.address.resolver" class="org.jbpm.identity.mail.IdentityAddressResolver" singleton="true" /> + + <bean name="jbpm.job.executor" class="org.jbpm.job.executor.JobExecutor"> + <property name="jbpmConfiguration"><ref bean="jbpmConfiguration" /></property> + <property name="name"><string value="JbpmJobExecutor" /></property> + <property name="nbrOfThreads"><int value="1" /></property> + <property name="retries"><int value="3" /></property> + <property name="idleInterval"><int value="5000" /></property> + <property name="maxIdleInterval"><int value="3600000" /><!-- 1 hour --></property> + <property name="maxLockTime"><int value="600000" /><!-- 10 minutes --></property> + <property name="lockMonitorInterval"><int value="60000" /><!-- 1 minute --></property> + <property name="lockBufferTime"><int value="5000" /><!-- 5 seconds --></property> + </bean> +</jbpm-configuration>]]></programlisting> + <para>In this configuration file you can see 3 parts: </para> @@ -103,13 +123,13 @@ Then you update the reference in this file and jbpm will use your customized version of that configuration file. </para></listitem> - <listitem><para>The third part are some miscallanious configurations used in jbpm. + <listitem><para>The third part are some miscellaneous configurations used in jbpm. These configuration options are described in the chapters that cover the specific topic. </para></listitem> </itemizedlist> - <para>The default configured set of services is targetted at a simple webapp environment + <para>The default configured set of services is targeted at a simple webapp environment and minimal dependencies. The persistence service will obtain a jdbc connection and all the other services will use the same connection to perform their services. So all of your workflow operations are centralized into 1 transaction on a JDBC connection without the @@ -138,7 +158,7 @@ <para>Note that the <literal>XxxForUpdate</literal> methods will register the loaded object for auto-save so that you don't have to call one of the save methods - explicitely. + explicitly. </para> <para>It's possible to specify multiple <literal>jbpm-context</literal>s, but then you have