Author: jbarrez Date: 2009-06-22 10:29:53 -0400 (Mon, 22 Jun 2009) New Revision: 5080 Removed: jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/OptimisticLockTestGround.java Modified: jbpm4/trunk/modules/test-concurrent/src/main/java/org/jbpm/test/concurrent/SynchronizableCommandExecutor.java jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/AsyncForkTest.java jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/TimerVsSignalConcurrencyTest.java jbpm4/trunk/modules/test-concurrent/src/test/resources/jbpm.hibernate.cfg.xml Log: Work in progress: concurrency testing Modified: jbpm4/trunk/modules/test-concurrent/src/main/java/org/jbpm/test/concurrent/SynchronizableCommandExecutor.java =================================================================== --- jbpm4/trunk/modules/test-concurrent/src/main/java/org/jbpm/test/concurrent/SynchronizableCommandExecutor.java 2009-06-22 14:26:18 UTC (rev 5079) +++ jbpm4/trunk/modules/test-concurrent/src/main/java/org/jbpm/test/concurrent/SynchronizableCommandExecutor.java 2009-06-22 14:29:53 UTC (rev 5080) @@ -80,6 +80,8 @@ private CyclicBarrier afterExecutionBarrier; + private List<SynchronizableCommandExecutor> executorsSyncedAfterExecution; + private CyclicBarrier beforeExecutionBarrier; /** @@ -187,11 +189,13 @@ } if (afterExecutionBarrier == null ) { - afterExecutionBarrier = new CyclicBarrier(2); + setSyncPointForAfterExecution(new CyclicBarrier(2)); } else { - afterExecutionBarrier = new CyclicBarrier(afterExecutionBarrier.getParties() + 1); + setSyncPointForAfterExecution(new CyclicBarrier(afterExecutionBarrier.getParties() + 1)); } + setSyncPointForAfterExecution(afterExecutionBarrier); + return this; } @@ -200,6 +204,11 @@ throw new RuntimeException("Cannot set synchronization point once the thread has been started"); } + if (executorsSyncedAfterExecution == null) { + executorsSyncedAfterExecution = new ArrayList<SynchronizableCommandExecutor>(); + } + executorsSyncedAfterExecution.add(otherExecutor); + int threadsInvolved = 0; if (afterExecutionBarrier != null) { threadsInvolved += afterExecutionBarrier.getParties(); @@ -213,12 +222,21 @@ threadsInvolved++; } - this.afterExecutionBarrier = new CyclicBarrier(threadsInvolved); - otherExecutor.afterExecutionBarrier = this.afterExecutionBarrier; + setSyncPointForAfterExecution(new CyclicBarrier(threadsInvolved)); return this; } + private void setSyncPointForAfterExecution(CyclicBarrier syncpoint) { + this.afterExecutionBarrier = syncpoint; + if (executorsSyncedAfterExecution == null) { + executorsSyncedAfterExecution = new ArrayList<SynchronizableCommandExecutor>(); + } + for (SynchronizableCommandExecutor executor : executorsSyncedAfterExecution) { + executor.afterExecutionBarrier = this.afterExecutionBarrier; + } + } + public SynchronizableCommandExecutor synchroniseBeforeExecution() { if (isAlive()) { throw new RuntimeException("Cannot set synchronization point once the JobExecutorEmulator has been started"); Modified: jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/AsyncForkTest.java =================================================================== --- jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/AsyncForkTest.java 2009-06-22 14:26:18 UTC (rev 5079) +++ jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/AsyncForkTest.java 2009-06-22 14:29:53 UTC (rev 5080) @@ -6,10 +6,24 @@ import org.jbpm.api.ProcessInstance; import org.jbpm.api.job.Job; - +/** + * Concurrency test case: when using an async fork with differen outgoing + * transitions, potentially conflicts can occur (eg when the outgoing paths + * come together in the Join activity at the same time). + * + * @author jbarrez + */ public class AsyncForkTest extends ConcurrentJbpmTestCase { - public void testAsyncForkNoOptimisticLockingFailure() { + /** + * Test case using an async fork with 2 outgoing transactions. + * In jBPM3, a StaleStateException was thrown when the different paths + * came together in the Join at the same time. + * + * However, the jBPM4 Join activity is designed with concurrency in mind, + * and the StaleStateException should not occur when this scenario happens. + */ + public void testAsyncForkNoOptimisticLockingFailure() { deployJpdlXmlString( "<process name='asyncFork'>" + " <start>" + @@ -26,12 +40,19 @@ " <custom name='pathB' class='org.jbpm.test.concurrent.PassThroughActivity' >" + " <transition to='theJoin' />" + " </custom>" + - " <join name='theJoin'>" + - " <transition to='test' />" + + // Can't test with default lock-mode (upgrade). SELECT ... FOR UPGRADE + // will block transactions at database level with no decent approach + // to check if the thread is blocking. So we use the default lockmode, + // which is the standard Hibernate optimistic locking. + // + // Note: not using lockmode upgrade can cause the Join logic to + // work with incorrect data: ie it could be that an incoming + // transition is not seen as the 'last' one, due to a concurrent + // read of data. This siutation is avoided in the test by executing + // the Join activity logic of the last transition only after the other one. + " <join name='theJoin' lockmode='none'>" + + " <transition to='end' />" + " </join>" + - " <state name='test' >" + - " <transition to='end' /> " + - " </state>" + " <end name='end' />" + "</process>" ); @@ -40,19 +61,30 @@ final List<Job> jobs = managementService.createJobQuery().processInstanceId(processInstance.getId()).list(); assertEquals(2, jobs.size()); - SynchronizableCommandExecutor executor1 = new SynchronizableCommandExecutor(environmentFactory, jobs.get(0)); - SynchronizableCommandExecutor executor2 = new SynchronizableCommandExecutor(environmentFactory, jobs.get(1)); + SynchronizableCommandExecutor executor1 = startThreadAndSyncAfterExecution(jobs.get(0)); + SynchronizableCommandExecutor executor2 = startThreadAndSyncAfterExecution(jobs.get(1)); - executor1.synchroniseAfterExecution(executor2); - executor1.synchroniseAfterExecution(); - - executor1.start(); - executor2.start(); + try { + executor1.join(); + executor2.join(); + } catch (InterruptedException e) { + e.printStackTrace(); + } if (executor1.getException() instanceof StaleStateException || executor2.getException() instanceof StaleStateException) { fail("A StaleStaeException was thrown, altough this shouldn't happen"); } + + assertProcessInstanceEnded(processInstance); } + + private SynchronizableCommandExecutor startThreadAndSyncAfterExecution(Job job) { + SynchronizableCommandExecutor executor = new SynchronizableCommandExecutor(environmentFactory, job); + executor.synchroniseAfterExecution(); + executor.start(); + executor.waitUntilExecutionFinished(false); + return executor; + } } Deleted: jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/OptimisticLockTestGround.java =================================================================== --- jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/OptimisticLockTestGround.java 2009-06-22 14:26:18 UTC (rev 5079) +++ jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/OptimisticLockTestGround.java 2009-06-22 14:29:53 UTC (rev 5080) @@ -1,255 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2005, JBoss Inc., and individual contributors as indicated - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -/** - * - */ -package org.jbpm.test.concurrent; - -import java.util.List; -import java.util.concurrent.Semaphore; - -import org.jbpm.api.Execution; -import org.jbpm.api.ProcessInstance; -import org.jbpm.api.env.Environment; -import org.jbpm.api.env.EnvironmentFactory; -import org.jbpm.api.job.Job; -import org.jbpm.pvm.internal.cmd.ExecuteJobCmd; -import org.jbpm.pvm.internal.tx.StandardTransaction; -import org.jbpm.test.JbpmTestCase; - - -/** - * Class to test different approaches to the concurrency problem. - * - * Doesnt work anymore, but dont delete yet, I need some stuff in here for later! - * * @author Joram Barrez - */ -public class OptimisticLockTestGround extends JbpmTestCase { - - private EnvironmentFactory environmentFactory; - - protected void setUp() throws Exception { - super.setUp(); - this.environmentFactory = (EnvironmentFactory) processEngine; // Better way to do this? - } - - - public void testMe() throws Exception { - deployJpdlXmlString( - "<process name='asyncFork'>" + - " <start>" + - " <transition to='theFork' />" + - " </start>" + - " <fork name='theFork'>" + - " <on event='end' continue='async' />" + - " <transition to='pathA' />" + - " <transition to='pathB' />" + - " </fork>" + - " <custom name='pathA' class='org.jbpm.test.concurrent.PassThroughActivity' >" + - " <transition to='theJoin' />" + - " </custom>" + - " <custom name='pathB' class='org.jbpm.test.concurrent.PassThroughActivity' >" + - " <transition to='theJoin' />" + - " </custom>" + - " <join name='theJoin' lockmode='none'>" + - " <transition to='end' />" + - " </join>" + - " <end name='end' />" + - "</process>" - ); - - ProcessInstance processInstance = executionService.startProcessInstanceByKey("asyncFork"); - final List<Job> jobs = managementService.createJobQuery().processInstanceId(processInstance.getId()).list(); - assertEquals(2, jobs.size()); - - JobExecutorEmulator jobExecutorEmulator1 = new JobExecutorEmulator(jobs.get(0)); - jobExecutorEmulator1.start(); - - JobExecutorEmulator jobExecutorEmulator2 = new JobExecutorEmulator(jobs.get(1)); - jobExecutorEmulator2.start(); - - // TODO check if transitions are marked for rollback -> exception happened - - jobExecutorEmulator1.join(); - if (jobExecutorEmulator1.getException() != null) { - fail("Error while executing job: " + jobExecutorEmulator1.getException().getMessage()); - } - - jobExecutorEmulator2.join(); - if (jobExecutorEmulator2.getException() != null) { - fail("Error while executing job: " + jobExecutorEmulator2.getException().getMessage()); - } - - } - - public void testMeToo() throws Exception { - - semaphore = new Semaphore(0); - - deployJpdlXmlString( - "<process name='timer_vs_signal'>" + - " <start>" + - " <transition to='wait' />" + - " </start>" + - " <state name='wait'>" + - " <transition name='timeout' to='end'>" + - " <timer duedate='1 second' />" + - " </transition>" + - " <transition to='end' name='go on' />" + - " </state>" + - " <end name='end' />" + - "</process>" - ); - - ProcessInstance processInstance = executionService.startProcessInstanceByKey("timer_vs_signal"); - final List<Job> jobs = managementService.createJobQuery().processInstanceId(processInstance.getId()).list(); - assertEquals(1, jobs.size()); - - JobExecutorEmulator jobExecutorEmulator1 = new JobExecutorEmulator(jobs.get(0)); - jobExecutorEmulator1.start(); - - while (!semaphore.hasQueuedThreads()) { - synchronized (this) { - System.out.println("------------------> waiting ..."); - wait(20L); - } - } - - // cause conflict - System.out.println("---------------------------> Causing MayHem"); - - Execution executionAtState = processInstance.findActiveExecutionIn("wait"); - assertNotNull(executionAtState); - executionService.signalExecutionById(executionAtState.getId(), "go on"); - semaphore.release(); - System.out.println("----------------------------> Released semaphore"); - - // TODO check if transitions are marked for rollback -> exception happened - - jobExecutorEmulator1.join(); - if (jobExecutorEmulator1.getException() != null) { - fail("Error while executing job: " + jobExecutorEmulator1.getException().getMessage()); - } - - } - - // Todo refactor - private static Semaphore semaphore = new Semaphore(1); // binary semaphore - - - private class JobExecutorEmulator extends Thread { - - private Long jobId; - - private Exception exception; - - public JobExecutorEmulator(Job job) { - this.jobId = job.getDbid(); - } - - public void run() { - - Environment environment = environmentFactory.openEnvironment(); - StandardTransaction standardTransaction = environment.get(StandardTransaction.class); - standardTransaction.begin(); - - try { - - System.out.println(Thread.currentThread().getName() + "----------------------------------------> executing job " + jobId); - ExecuteJobCmd executeJobCmd = new ExecuteJobCmd(jobId); - executeJobCmd.execute(environment); - - System.out.println(Thread.currentThread().getName() +"-------------------------------------------------------> DONE EXECUTING JOB " + jobId); - - } catch (Exception e) { - standardTransaction.setRollbackOnly(); - this.exception = e; - - System.out.println(Thread.currentThread().getName() +" -----------------------------------------------------> IM A FAILURE"); - } finally { - - - // error: both threads entering at the same time! - /* - boolean acquired = false; - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> BEFORE SYNC BLOCK"); - synchronized (semaphore) { - acquired = semaphore.tryAcquire(); - } - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> AFTER SYNC BLOCK"); - if (acquired) { - - try { - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> ACQUIRING"); - synchronized (semaphore) { - semaphore.wait(); - } - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> FREED FROM ACQUIRED"); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - */ - - - if (!semaphore.hasQueuedThreads()) { - try { - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> ACQUIRING"); - semaphore.acquire(); - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> FREED FROM ACQUIRED"); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - - - - if (standardTransaction.isRollbackOnly()) { - exception = new Exception("Transaction was rollbacked"); - } - - System.out.println(Thread.currentThread().getName() + "-------------------------------------------------> COMPLETING "); - standardTransaction.complete(); - - System.out.println(Thread.currentThread().getName() + "------------------------------------------------------> RELEASING"); - - /*if (!acquired) { - semaphore.release(); - synchronized (semaphore) { - semaphore.notify(); - } - }*/ - - } - environment.close(); - System.out.println(Thread.currentThread().getName() + " has finished"); - } - - - public Exception getException() { - return exception; - } - - - } - -} Modified: jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/TimerVsSignalConcurrencyTest.java =================================================================== --- jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/TimerVsSignalConcurrencyTest.java 2009-06-22 14:26:18 UTC (rev 5079) +++ jbpm4/trunk/modules/test-concurrent/src/test/java/org/jbpm/test/concurrent/TimerVsSignalConcurrencyTest.java 2009-06-22 14:29:53 UTC (rev 5080) @@ -34,11 +34,18 @@ /** + * Concurrency test case: when a timer is defined on an activity, a job + * will be executed by the jobExecutor. But when at the same time a + * signal is done on the same activity, a conflicting situation occurs. + * + * This test case will mimic this behaviour to understand how the different + * databases react on such a conflict. + * * @author Joram Barrez */ public class TimerVsSignalConcurrencyTest extends ConcurrentJbpmTestCase { - public void testStaleObjectExceptionThrown() throws Exception { + public void testStaleObjectExceptionThrown() { deployJpdlXmlString( "<process name='timer_vs_signal'>" + @@ -76,13 +83,22 @@ // Best effort: wait 1 sec and see if the staleObjectException has been caused synchronized (this) { - wait(1000L); + try { + wait(1000L); + } catch (InterruptedException e) { + e.printStackTrace(); + } } jobExecutorEmulator.goOn(); - jobExecutorEmulator.join(); - signalThread.join(); + try { + jobExecutorEmulator.join(); + signalThread.join(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + if (!(jobExecutorEmulator.getException() instanceof StaleStateException Modified: jbpm4/trunk/modules/test-concurrent/src/test/resources/jbpm.hibernate.cfg.xml =================================================================== --- jbpm4/trunk/modules/test-concurrent/src/test/resources/jbpm.hibernate.cfg.xml 2009-06-22 14:26:18 UTC (rev 5079) +++ jbpm4/trunk/modules/test-concurrent/src/test/resources/jbpm.hibernate.cfg.xml 2009-06-22 14:29:53 UTC (rev 5080) @@ -22,7 +22,7 @@ <property name="hibernate.connection.password">sa</property> <property name="hibernate.default_schema">PUBLIC</property> <property name="hibernate.dialect">org.hibernate.dialect.H2Dialect</property> - + <!-- POSTGRES config <property name="hibernate.connection.driver_class">org.h2.Driver</property> @@ -30,7 +30,7 @@ <property name="hibernate.connection.username">postgres</property> <property name="hibernate.connection.password">postgres</property> <property name="hibernate.dialect">org.hibernate.dialect.PostgreSQLDialect</property> - --> + --> <property name="hibernate.hbm2ddl.auto">create-drop</property> <property name="hibernate.format_sql">true</property>