Author: marco.rietveld Date: 2011-05-19 11:34:47 -0400 (Thu, 19 May 2011) New Revision: 6942 Modified: jbpm3/branches/jbpm-3.2-soa/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java Log: Experiment with ThreadMXBean in order to see if this helps to run JobExecutor tests. Modified: jbpm3/branches/jbpm-3.2-soa/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java =================================================================== --- jbpm3/branches/jbpm-3.2-soa/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java 2011-05-19 13:03:08 UTC (rev 6941) +++ jbpm3/branches/jbpm-3.2-soa/core/src/main/java/org/jbpm/db/AbstractDbTestCase.java 2011-05-19 15:34:47 UTC (rev 6942) @@ -21,7 +21,11 @@ */ package org.jbpm.db; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadMXBean; import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -31,7 +35,6 @@ import org.hibernate.cfg.Environment; import org.hibernate.criterion.Projections; import org.hibernate.criterion.Restrictions; - import org.jbpm.AbstractJbpmTestCase; import org.jbpm.JbpmConfiguration; import org.jbpm.JbpmContext; @@ -205,9 +208,10 @@ } } - protected void startJobExecutor() { + protected JobExecutor startJobExecutor() { jobExecutor = getJbpmConfiguration().getJobExecutor(); jobExecutor.start(); + return jobExecutor; } /** @@ -260,6 +264,72 @@ } } + private long getUserTimeInMilliseconds( HashSet<Long> threadIdSet ) { + ThreadMXBean bean = ManagementFactory.getThreadMXBean( ); + if ( ! bean.isThreadCpuTimeSupported( ) ) { + return 0L; + } + + long totalThreadTimeInNanoSeconds = 0; + Long [] threadId = new Long[0]; + threadId = threadIdSet.toArray(threadId); + + for ( int i = 0; i < threadId.length; ++i ) { + long threadTime = bean.getThreadUserTime( threadId[i] ); + if ( threadTime != -1 ) { + totalThreadTimeInNanoSeconds += threadTime; + } + } + + return totalThreadTimeInNanoSeconds/1000000; + } + + protected void newWaitForJobs(final long timeout, Map threadMap) { + + // Gather thread Id's + Iterator<Thread> threadMapIter = threadMap.values().iterator(); + HashSet<Long> threadIdSet = new HashSet<Long>(); + while(threadMapIter.hasNext()) { + threadIdSet.add(threadMapIter.next().getId()); + } + + long waitPeriod = 500; + for (int currentCount, previousCount = 0; (currentCount = getNbrOfJobsAvailable()) > 0;) { + + long elapsedTime = getUserTimeInMilliseconds(threadIdSet); + + if (elapsedTime > timeout) { + fail("test execution exceeded threshold of " + timeout + " ms"); + } + + if (currentCount < previousCount) { + long elapsedTimePerJob = (elapsedTime) / (previousCount - currentCount); + waitPeriod = currentCount * elapsedTimePerJob; + if (waitPeriod < 500) waitPeriod = 500; + } + + if (waitPeriod > 5000) { + waitPeriod = 5000; + } + else { + long remainingTime = timeout - elapsedTime; + waitPeriod = (waitPeriod > remainingTime) ? remainingTime : waitPeriod; + } + + if (log.isDebugEnabled()) { + log.debug("waiting " + waitPeriod + " ms for " + currentCount + " jobs"); + } + try { + Thread.sleep(waitPeriod); + } + catch (InterruptedException e) { + fail("wait for jobs got interrupted"); + } + + previousCount = currentCount; + } + } + protected int getNbrOfJobsAvailable() { if (session != null) { return getJobCount(session); @@ -305,7 +375,7 @@ * time has elapsed. The current jBPM context is closed before waiting and a new one is opened * after processing the jobs. */ - protected void processJobs(long timeout) { + protected void oldProcessJobs(long timeout) { closeJbpmContext(); try { startJobExecutor(); @@ -317,6 +387,18 @@ } } + protected void processJobs(long timeout) { + closeJbpmContext(); + try { + JobExecutor jobExecutor = startJobExecutor(); + newWaitForJobs(timeout, jobExecutor.getThreads()); + } + finally { + stopJobExecutor(); + createJbpmContext(); + } + } + protected void stopJobExecutor() { if (jobExecutor != null) { try {