JBossDeveloper

Sorry, premature 'send' of above. Here's missing source code:

/**
 *
 * @param ltesterUserName
 * @return current userSession or null
 */
 private UserSession getCurrentUserSession(final String ltesterUserName) throws Exception {

 LOG.debug("start getCurrentUserSession method user=" + ltesterUserName);

 try {
 return (UserSession) txTemplate.executeWithException(new TransactionCallbackWithException() {

 public Object doInTransactionWithException(TransactionStatus status) throws Exception {

 //add user session, throw exception if one does not exist

 LOG.debug("before findByName user=" + ltesterUserName);
 final User usr = getUserDao().findByNameChecked(ltesterUserName);

 LOG.debug("before usr.getCurrentUserSession user=" + ltesterUserName);
 final UserSession us = usr.getCurrentUserSession();
 LOG.debug("after usr.getCurrentUserSession user=" + ltesterUserName);
 return us;
 }
 });
 } finally {
 LOG.debug("end getCurrentUserSession user=" + ltesterUserName);
 }

 }

OK will do: look at the test suite.

Good idea to try PESSIMISTIC locking. Results are:

PESSIMISTIC & REPEATABLE_READ == pass
MVCC & REPEATABLE_READ == fail(-fast ;-)

I've debugged the failure down to:
-on commit Hibernate calls CollectionAction.evict() (as you mentioned it should)
-downstream, CacheInvocationDelegate.removeNode(collection Fqn) calls invoke.invoke *which returns false*. I assume that means that for whatever reason invocation of RemoveNodeCommand fails.

Is there someone who may be able to help me track this down in your absence? We're nearing some milestone dates here ;-

thanks
-nikita

Turning up userCount under REPEATABLE_READ and PESSIMISTIC locking yielded the same failure. So, looks like there are 2 distinct problems:

1) One we've been debugging all along: cache left in incorrect state when one thread wants to remove node while another wants to putForExternalRead (not clear whether the problem is that latter wins or former fails on invocation of RemoveNodeCommand)

2) Under REPEATABLE_READ + MVCC: CacheInvocationDelegate.removeNode(Fqn) returns false and fails to remove() node in question. This happens w/o any concurrency (single thread)

Thought latter seems easier to debug, I can (prefer to) live w/o REPEATABLE_READ, so still problem #1 is the one to solve for me ;-

"1) Do you *really* need to cache this collection? "

No, for a short while we could probably live w/o that. However, there are several other collections that we do need to cache (that's pretty much the point of the project) and thus far I don't see why these failures won't show up there as well. Perhaps I will do that now, and wait and see if someone else chimes in with a take in the meantime.

Here's my first take on a unit test that builds/runs as part of org.hibernate.test.cache.jbc2 maven project.

The problem is that this test (since it tests concurrency) requires JTATransactionManager impl that can handle multiple concurrent Transactions. And SimpleJtaTransactionManagerImpl is only suitable for single thread tests. So testManyUsers() fails for all the wrong reasons.

Does anyone know which TX Manager impl I should be using in this test environment? If one isn't available (then how is concurrency under JTA being tested?) - any hints on the simplest way to plug in a real impl?

package org.hibernate.test.cache.jbc2.functional;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import junit.framework.Test;
import org.hibernate.FlushMode;
import org.hibernate.Session;
import org.hibernate.exception.ExceptionUtils;
import org.hibernate.junit.functional.FunctionalTestClassTestSuite;
import org.hibernate.test.tm.SimpleJtaTransactionManagerImpl;
import org.hibernate.transaction.CMTTransactionFactory;

/**
 *
 * @author nikita_tovstoles@mba.berkeley.edu
 */
public class MVCCConcurrentWriteTest extends MVCCJBossCacheTest {

 /**
 * when USER_COUNT==1, tests pass, when >4 tests fail
 */
 final private int USER_COUNT = 5;
 final private int ITERATION_COUNT = 50;
 final private long THINK_TIME_MILLIS = 10;
 final private long LAUNCH_INTERVAL_MILLIS = 10;
 /**
 * collection of IDs of all customers participating in this test
 */
 private Set<Integer> customerIDs = new HashSet<Integer>();

 public MVCCConcurrentWriteTest(String x) {
 super(x);
 }

 @Override
 protected Class getTransactionFactoryClass() {
 return CMTTransactionFactory.class; //really want JTATransactionFactory here but it's not supported by this test harness
 // return JTATransactionFactory.class;
 }

 @Override
 public void testEmptySecondLevelCacheEntry() throws Exception {
 //do nothing
 }

 @Override
 public void testQueryCacheInvalidation() {
 //do nothing
 }

 @Override
 public void testStaleWritesLeaveCacheConsistent() {
 //do nothing
 }

 public void testSingleUser() throws Exception {
 try {
 //setup
 Customer customer = createCustomer(0);
 final Integer customerId = customer.getId();
 getCustomerIDs().add(customerId);

 assertNull("contact exists despite not being added", getFirstContact(customerId));

 final Contact contact = addContact(customerId);
 assertNotNull("contact returned by addContact is null", contact);
 assertNotNull("Contact missing after successful add call", getFirstContact(customerId));

 //read everyone's contacts
 readEveryonesFirstContact();

 removeContact(customerId);
 assertNull("contact still exists after successful remove call", getFirstContact(customerId));
 } finally {
 //cleanup
 cleanup();
 }
 }

 public void testManyUsers() throws Exception {

 try {
 //setup - create users
 for (int i = 0; i < USER_COUNT; i++) {
 Customer customer = createCustomer(0);
 getCustomerIDs().add(customer.getId());
 }

 final ExecutorService pool = Executors.newFixedThreadPool(USER_COUNT);

 Set<UserRunner> runners = new HashSet<UserRunner>();
 for (Integer customerId : getCustomerIDs()) {
 UserRunner r = new UserRunner(customerId);
 runners.add(r);
 pool.execute(r);
 Thread.sleep(LAUNCH_INTERVAL_MILLIS); //rampup
 }

 assertEquals("not all user threads launched", USER_COUNT, runners.size());

 pool.shutdown();
 boolean finishedInTime = pool.awaitTermination(2, TimeUnit.MINUTES);
 assertTrue("test took too long", finishedInTime);

 //check whether all runners suceeded
 for (UserRunner r : runners) {
 assertEquals("runner for customerId=" + r.getCustomerId() + " did not complete all iterations; cause=" +
 ExceptionUtils.getFullStackTrace(r.getCauseOfFailure()),
 ITERATION_COUNT,
 r.getCompletedIterations());
 }

 } finally {
 //TODO - move to onTearDown or similar
 //cleanup - remove users and contacts
 cleanup();
 }
 }

 public void cleanup() throws Exception {

 getCustomerIDs().clear();

 String deleteContactHQL = "delete from Contact";
 String deleteCustomerHQL = "delete from Customer";

 beginTx();
 try {

 Session session = getSessions().getCurrentSession();
 session.createQuery(deleteContactHQL).setFlushMode(FlushMode.AUTO).executeUpdate();
 session.createQuery(deleteCustomerHQL).setFlushMode(FlushMode.AUTO).executeUpdate();
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }

 }

 private Customer createCustomer(int nameSuffix) throws Exception {
 beginTx();
 try {
 Customer customer = new Customer();
 customer.setName("customer_" + nameSuffix);
 customer.setContacts(new HashSet<Contact>());

 getSessions().getCurrentSession().persist(customer);
 commitTx();
 return customer;
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * delegate method since I'm trying to figure out which txManager to use
 * given that this test runs multiple threads (SimpleJtaTxMgrImpl isn't suited for that).
 *
 * What is needed is a thread-safe JTATransactionManager impl that can handle concurrent TXs
 *
 * @throws java.lang.Exception
 */
 private void beginTx() throws Exception {
 SimpleJtaTransactionManagerImpl.getInstance().begin();
 //getSessions().getCurrentSession().beginTransaction();
 }

 /**
 * @see #beginTx()
 * @throws java.lang.Exception
 */
 private void commitTx() throws Exception {
 SimpleJtaTransactionManagerImpl.getInstance().commit();
 //getSessions().getCurrentSession().getTransaction().commit();
 }

 /**
 * @see #beginTx()
 * @throws java.lang.Exception
 */
 private void rollbackTx() throws Exception {
 SimpleJtaTransactionManagerImpl.getInstance().rollback();
 //getSessions().getCurrentSession().getTransaction().rollback();
 }

 /**
 * read first contact of every Customer participating in this test.
 * this forces concurrent cache writes of Customer.contacts Collection cache node
 *
 * @return who cares
 * @throws java.lang.Exception
 */
 private void readEveryonesFirstContact() throws Exception {
 beginTx();
 try {
 for (Integer customerId : getCustomerIDs()) {
 final Customer customer = (Customer) getSessions().getCurrentSession().load(Customer.class, customerId);
 Set<Contact> contacts = customer.getContacts();
 Contact firstContact = contacts.isEmpty() ? null : contacts.iterator().next();
 }
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * -load existing Customer
 * -get customer's contacts; return 1st one
 *
 * @param customerId
 * @return first Contact or null if customer has none
 */
 private Contact getFirstContact(Integer customerId) throws Exception {
 assert customerId != null;

 beginTx();
 try {
 final Customer customer = (Customer) getSessions().getCurrentSession().load(Customer.class, customerId);
 Set<Contact> contacts = customer.getContacts();
 Contact firstContact = contacts.isEmpty() ? null : contacts.iterator().next();
 commitTx();
 return firstContact;
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * -load existing Customer
 * -create a new Contact and add to customer's contacts
 *
 * @param customerId
 * @return added Contact
 */
 private Contact addContact(Integer customerId) throws Exception {
 assert customerId != null;

 beginTx();
 try {
 final Customer customer = (Customer) getSessions().getCurrentSession().load(Customer.class, customerId);

 Contact contact = new Contact();
 contact.setName("contact name");
 contact.setTlf("wtf is tlf?");

 contact.setCustomer(customer);
 customer.getContacts().add(contact);

 //assuming contact is persisted via cascade from customer
 commitTx();

 return contact;
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * remove existing 'contact' from customer's list of contacts
 *
 * @param contact contact to remove from customer's contacts
 * @param customerId
 * @throws IllegalStateException if customer does not own a contact
 */
 private void removeContact(Integer customerId) throws Exception {
 assert customerId != null;

 beginTx();
 try {
 Customer customer = (Customer) getSessions().getCurrentSession().load(Customer.class, customerId);
 Set<Contact> contacts = customer.getContacts();
 if (contacts.size() != 1) {
 throw new IllegalStateException("can't remove contact: customer id=" + customerId + " expected exactly 1 contact, " +
 "actual count=" + contacts.size());
 }

 Contact contact = contacts.iterator().next();
 contacts.remove(contact);
 contact.setCustomer(null);

 //explicitly delete Contact because hbm has no 'DELETE_ORPHAN' cascade?
 //getSessions().getCurrentSession().delete(contact); //appears to not be needed

 //assuming contact is persisted via cascade from customer
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * @return the customerIDs
 */
 public Set<Integer> getCustomerIDs() {
 return customerIDs;
 }

 class UserRunner implements Runnable {

 final private Integer customerId;
 private int completedIterations = 0;
 private Throwable causeOfFailure;

 public UserRunner(final Integer cId) {
 assert cId != null;
 this.customerId = cId;
 }

 private boolean contactExists() throws Exception {
 return getFirstContact(customerId) != null;
 }

 public void run() {

 //name this thread for easier log tracing
 Thread.currentThread().setName("UserRunnerThread-" + getCustomerId());
 try {
 for (int i = 0; i < ITERATION_COUNT; i++) {

 if (contactExists()) {
 throw new IllegalStateException("contact already exists before add, customerId=" + customerId);
 }

 addContact(customerId);

 if (!contactExists()) {
 throw new IllegalStateException("contact missing after successful add, customerId=" + customerId);
 }

 think();

 //read everyone's contacts
 readEveryonesFirstContact();

 think();

 removeContact(customerId);

 if (contactExists()) {
 throw new IllegalStateException("contact still exists after successful remove call, customerId=" + customerId);
 }

 think();

 ++completedIterations;
 }

 } catch (Throwable t) {
 this.causeOfFailure = t;
 }
 }

 public boolean isSuccess() {
 return ITERATION_COUNT == getCompletedIterations();
 }

 public int getCompletedIterations() {
 return completedIterations;
 }

 public Throwable getCauseOfFailure() {
 return causeOfFailure;
 }

 public Integer getCustomerId() {
 return customerId;
 }
 }

 private void think() {
 try {
 Thread.sleep(THINK_TIME_MILLIS);
 } catch (InterruptedException ex) {
 throw new RuntimeException("sleep interrupted", ex);
 }
 }

 public static Test suite() {
 return new FunctionalTestClassTestSuite(MVCCConcurrentWriteTest.class);
 }
}

c:\code\hibernate\core\trunk\cache-jbosscache2>mvn test -Dtest=MVCCConcurrentWriteTest
[INFO] Scanning for projects...
WAGON_VERSION: 1.0-beta-2
[INFO] ------------------------------------------------------------------------
[INFO] Building Hibernate JBossCache3.x Integration
[INFO] task-segment: [test]
[INFO] ------------------------------------------------------------------------
[INFO] [enforcer:enforce {execution: enforce-java}]
[INFO] [resources:resources]
[INFO] Using default encoding to copy filtered resources.
[INFO] [compiler:compile]
[INFO] Nothing to compile - all classes are up to date
[INFO] [resources:testResources]
[INFO] Using default encoding to copy filtered resources.
[INFO] [compiler:testCompile]
[INFO] Nothing to compile - all classes are up to date
[INFO] [surefire:test]
[INFO] Surefire report directory: c:\code\hibernate\core\trunk\cache-jbosscache2\target\surefire-reports

-------------------------------------------------------
T E S T S
-------------------------------------------------------
Running org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest
Tests run: 5, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 31.557 sec <<< FAILURE!

Results :

Tests in error:
testManyUsers(org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest)

Tests run: 5, Failures: 0, Errors: 1, Skipped: 0

[INFO] ------------------------------------------------------------------------
[ERROR] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] There are test failures.

Please refer to c:\code\hibernate\core\trunk\cache-jbosscache2\target\surefire-reports for the individual test results.
[INFO] ------------------------------------------------------------------------
[INFO] For more information, run Maven with the -e switch
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 34 seconds
[INFO] Finished at: Fri Mar 06 18:04:48 PST 2009
[INFO] Final Memory: 11M/20M
[INFO] ------------------------------------------------------------------------

c:\code\hibernate\core\trunk\cache-jbosscache2>

-------------------------------------------------------------------------------
Test set: org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest
-------------------------------------------------------------------------------
Tests run: 5, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 3.222 sec <<< FAILURE!
testManyUsers(org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest) Time elapsed: 1.953 sec <<< FAILURE!
junit.framework.AssertionFailedError: runner for customerId=6 did not complete all iterations; cause=java.lang.IllegalStateException: no current transaction
at org.hibernate.test.tm.SimpleJtaTransactionManagerImpl.rollback(SimpleJtaTransactionManagerImpl.java:90)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.rollbackTx(MVCCConcurrentWriteTest.java:188)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.addContact(MVCCConcurrentWriteTest.java:263)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.access$100(MVCCConcurrentWriteTest.java:24)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest$UserRunner.run(MVCCConcurrentWriteTest.java:335)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)
expected:<50> but was:<20>
at junit.framework.Assert.fail(Assert.java:47)
at junit.framework.Assert.failNotEquals(Assert.java:282)
at junit.framework.Assert.assertEquals(Assert.java:64)
at junit.framework.Assert.assertEquals(Assert.java:201)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.testManyUsers(MVCCConcurrentWriteTest.java:111)

OK, I wired up DualNodeJtaTransactionManagerImpl. The test now passes if UserCount is below 4 or less, at 5 or above the test fails. However, the failure I am seeing with this test is NOT the same as with the test in our environment. It is this:

-------------------------------------------------------------------------------
Test set: org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest
-------------------------------------------------------------------------------
Tests run: 6, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 182.113 sec <<< FAILURE!
testManyUsers(org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest) Time elapsed: 165.831 sec <<< FAILURE!
junit.framework.AssertionFailedError: Timed out waiting for user threads to finish. Their state at the time of forced shutdown: TEST CONFIG [userCount=5, iterationsPerUser=40, thinkTimeMillis=10] STATE of UserRunners: org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest$UserRunner@ffc3eb[customerId=4 iterationsCompleted=30 completedAll=false causeOfFailure=org.hibernate.cache.CacheException: org.hibernate.cache.CacheException: org.jboss.cache.lock.TimeoutException: Unable to acquire lock on Fqn [/TS/test/org/hibernate/cache/UpdateTimestampsCache/Contacts] after [15000] milliseconds for requestor [Thread[UserRunnerThread-4,5,main]]! Lock held by [GlobalTransaction:<null>:903]
at org.hibernate.cache.jbc2.timestamp.TimestampsRegionImpl.put(TimestampsRegionImpl.java:130)
at org.hibernate.cache.UpdateTimestampsCache.preinvalidate(UpdateTimestampsCache.java:70)
at org.hibernate.engine.ActionQueue.execute(ActionQueue.java:275)
at org.hibernate.engine.ActionQueue.executeActions(ActionQueue.java:263)
at org.hibernate.engine.ActionQueue.executeActions(ActionQueue.java:167)
at org.hibernate.event.def.AbstractFlushingEventListener.performExecutions(AbstractFlushingEventListener.java:321)
at org.hibernate.event.def.DefaultFlushEventListener.onFlush(DefaultFlushEventListener.java:50)
at org.hibernate.impl.SessionImpl.flush(SessionImpl.java:1027)
at org.hibernate.impl.SessionImpl.managedFlush(SessionImpl.java:365)
at org.hibernate.transaction.CacheSynchronization.beforeCompletion(CacheSynchronization.java:88)
at org.hibernate.test.cache.jbc2.functional.util.DualNodeJtaTransactionImpl.commit(DualNodeJtaTransactionImpl.java:76)
at org.hibernate.test.cache.jbc2.functional.util.DualNodeJtaTransactionManagerImpl.commit(DualNodeJtaTransactionManagerImpl.java:123)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.commitTx(MVCCConcurrentWriteTest.java:255)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.addContact(MVCCConcurrentWriteTest.java:333)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest.access$100(MVCCConcurrentWriteTest.java:29)
at org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest$UserRunner.run(MVCCConcurrentWriteTest.java:423)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)
Caused by: org.hibernate.cache.CacheException: org.jboss.cache.lock.TimeoutException: Unable to acquire lock on Fqn [/TS/test/org/hibernate/cache/UpdateTimestampsCache/Contacts] after [15000] milliseconds for requestor [Thread[UserRunnerThread-4,5,main]]! Lock held by [GlobalTransaction:<null>:903]
at org.hibernate.cache.jbc2.util.CacheHelper.put(CacheHelper.java:214)
at org.hibernate.cache.jbc2.timestamp.TimestampsRegionImpl.put(TimestampsRegionImpl.java:128)
... 18 more]
org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest$UserRunner@18bbb61[customerId=6 iterationsCompleted=28 completedAll=false causeOfFailure=]
org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest$UserRunner@1cab18[customerId=2 iterationsCompleted=28 completedAll=false causeOfFailure=]
org.hibernate.test.cache.jbc2.functional.MVCCConcurrentWriteTest$UserRunner@9cd8db[customerId=3 iterationsCompleted=30 completedAll=false causeOfFailure=]

package org.hibernate.test.cache.jbc2.functional;

import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.transaction.SystemException;
import junit.framework.Test;
import org.hibernate.FlushMode;
import org.hibernate.Session;
import org.hibernate.cfg.Configuration;
import org.hibernate.exception.ExceptionUtils;
import org.hibernate.junit.functional.FunctionalTestClassTestSuite;
import org.hibernate.stat.SecondLevelCacheStatistics;
import org.hibernate.test.cache.jbc2.functional.util.DualNodeConnectionProviderImpl;
import org.hibernate.test.cache.jbc2.functional.util.DualNodeJtaTransactionManagerImpl;
import org.hibernate.test.cache.jbc2.functional.util.DualNodeTestUtil;
import org.hibernate.test.cache.jbc2.functional.util.DualNodeTransactionManagerLookup;
import org.hibernate.transaction.CMTTransactionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author nikita_tovstoles@mba.berkeley.edu
 */
public class MVCCConcurrentWriteTest extends MVCCJBossCacheTest {

 private static final Logger LOG = LoggerFactory.getLogger(MVCCConcurrentWriteTest.class);
 /**
 * when USER_COUNT==1, tests pass, when >4 tests fail
 */
 final private int USER_COUNT = 5;
 final private int ITERATION_COUNT = 40;
 final private int THINK_TIME_MILLIS = 10;
 final private long LAUNCH_INTERVAL_MILLIS = 10;
 final private Random random = new Random();
 /**
 * kill switch used to stop all users when one fails
 */
 private static boolean TERMINATE_ALL_USERS = false;
 /**
 * collection of IDs of all customers participating in this test
 */
 private Set<Integer> customerIDs = new HashSet<Integer>();

 public MVCCConcurrentWriteTest(String x) {
 super(x);
 }

 /**
 * test that DB can be queried
 * @throws java.lang.Exception
 */
 public void testPingDb() throws Exception {
 try {
 beginTx();
 getEnvironment().getSessionFactory().getCurrentSession().createQuery("from " + Customer.class.getName()).list();
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 fail("failed to query DB; exception=" + e);
 }
 }

 @Override
 protected void prepareTest() throws Exception {
 super.prepareTest();
 }

 @Override
 protected void cleanupTest() throws Exception {
 try {
 super.cleanupTest();

 } finally {
 cleanup();
 //DualNodeJtaTransactionManagerImpl.cleanupTransactions();
 //DualNodeJtaTransactionManagerImpl.cleanupTransactionManagers();
 }
 }

 @Override
 public void configure(Configuration cfg) {
 super.configure(cfg);
 cfg.setProperty(DualNodeTestUtil.NODE_ID_PROP, DualNodeTestUtil.LOCAL);
 }

 @Override
 protected Class getConnectionProviderClass() {
 return DualNodeConnectionProviderImpl.class;
 }

 @Override
 protected Class getTransactionManagerLookupClass() {
 return DualNodeTransactionManagerLookup.class;
 }

 @Override
 protected Class getTransactionFactoryClass() {
 return CMTTransactionFactory.class;
 }

 @Override
 public void testEmptySecondLevelCacheEntry() throws Exception {
 //do nothing
 }

 @Override
 public void testQueryCacheInvalidation() {
 //do nothing
 }

 @Override
 public void testStaleWritesLeaveCacheConsistent() {
 //do nothing
 }

 public void testSingleUser() throws Exception {
 //setup
 Customer customer = createCustomer(0);
 final Integer customerId = customer.getId();
 getCustomerIDs().add(customerId);

 assertNull("contact exists despite not being added", getFirstContact(customerId));

 //check that cache was hit
 SecondLevelCacheStatistics customerSlcs = getEnvironment().getSessionFactory().getStatistics().getSecondLevelCacheStatistics(
 getPrefixedRegionName(Customer.class.getName()));
 assertEquals(customerSlcs.getPutCount(), 1);
 assertEquals(customerSlcs.getElementCountInMemory(), 1);
 assertEquals(customerSlcs.getEntries().size(), 1);

 SecondLevelCacheStatistics contactsCollectionSlcs = getEnvironment().getSessionFactory().getStatistics().getSecondLevelCacheStatistics(
 getPrefixedRegionName(Customer.class.getName() + ".contacts"));
 assertEquals(1, contactsCollectionSlcs.getPutCount());
 assertEquals(1, contactsCollectionSlcs.getElementCountInMemory());
 assertEquals(1, contactsCollectionSlcs.getEntries().size());

 final Contact contact = addContact(customerId);
 assertNotNull("contact returned by addContact is null", contact);
 assertEquals("Customer.contacts cache was not invalidated after addContact",
 0, contactsCollectionSlcs.getElementCountInMemory());

 assertNotNull("Contact missing after successful add call", getFirstContact(customerId));


 //read everyone's contacts
 readEveryonesFirstContact();

 removeContact(customerId);
 assertNull("contact still exists after successful remove call", getFirstContact(customerId));

 }

 public void testManyUsers() throws Exception {

 //setup - create users
 for (int i = 0; i < USER_COUNT; i++) {
 Customer customer = createCustomer(0);
 getCustomerIDs().add(customer.getId());
 }

 assertEquals("failed to create enough Customers", USER_COUNT, getCustomerIDs().size());

 final ExecutorService pool = Executors.newFixedThreadPool(USER_COUNT);

 Set<UserRunner> runners = new HashSet<UserRunner>();
 for (Integer customerId : getCustomerIDs()) {
 UserRunner r = new UserRunner(customerId);
 runners.add(r);
 pool.execute(r);
 LOG.info("launched " + r);
 Thread.sleep(LAUNCH_INTERVAL_MILLIS); //rampup
 }

 assertEquals("not all user threads launched", USER_COUNT, runners.size());

 pool.shutdown();
 boolean finishedInTime = pool.awaitTermination(120, TimeUnit.SECONDS);

 if (!finishedInTime) { //timed out waiting for users to finish
 fail("Timed out waiting for user threads to finish. Their state at the time of forced shutdown: " + statusOfRunnersToString(runners));
 } else {
 //if here -> pool finished before timing out
 //check whether all runners suceeded
 boolean success = true;
 for (UserRunner r : runners) {
 if (!r.isSuccess()) {
 success = false;
 break;
 }
 }
 assertTrue("at least one UserRunner failed: " + statusOfRunnersToString(runners), success);
 }
 }

 public void cleanup() throws Exception {

 getCustomerIDs().clear();

 String deleteContactHQL = "delete from Contact";
 String deleteCustomerHQL = "delete from Customer";

 beginTx();
 try {

 //Session session = getSessions().getCurrentSession();
 Session session = getEnvironment().getSessionFactory().getCurrentSession();
 session.createQuery(deleteContactHQL).setFlushMode(FlushMode.AUTO).executeUpdate();
 session.createQuery(deleteCustomerHQL).setFlushMode(FlushMode.AUTO).executeUpdate();
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }

 }

 private Customer createCustomer(int nameSuffix) throws Exception {
 beginTx();
 try {
 Customer customer = new Customer();
 customer.setName("customer_" + nameSuffix);
 customer.setContacts(new HashSet<Contact>());

 getEnvironment().getSessionFactory().getCurrentSession().persist(customer);
 commitTx();
 return customer;
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * delegate method since I'm trying to figure out which txManager to use
 * given that this test runs multiple threads (SimpleJtaTxMgrImpl isn't suited for that).
 *
 * What is needed is a thread-safe JTATransactionManager impl that can handle concurrent TXs
 *
 * @throws java.lang.Exception
 */
 private void beginTx() throws Exception {
 DualNodeJtaTransactionManagerImpl.getInstance(DualNodeTestUtil.LOCAL).begin();
 }

 /**
 * @see #beginTx()
 * @throws java.lang.Exception
 */
 private void commitTx() throws Exception {
 DualNodeJtaTransactionManagerImpl.getInstance(DualNodeTestUtil.LOCAL).commit();
 }

 /**
 * @see #beginTx()
 * @throws java.lang.Exception
 */
 private void rollbackTx() throws Exception {
 DualNodeJtaTransactionManagerImpl.getInstance(DualNodeTestUtil.LOCAL).rollback();
 }

 /**
 * read first contact of every Customer participating in this test.
 * this forces concurrent cache writes of Customer.contacts Collection cache node
 *
 * @return who cares
 * @throws java.lang.Exception
 */
 private void readEveryonesFirstContact() throws Exception {
 beginTx();
 try {
 for (Integer customerId : getCustomerIDs()) {
 final Customer customer = (Customer) getEnvironment().getSessionFactory().getCurrentSession().load(Customer.class, customerId);
 Set<Contact> contacts = customer.getContacts();
 Contact firstContact = contacts.isEmpty() ? null : contacts.iterator().next();
 }
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * -load existing Customer
 * -get customer's contacts; return 1st one
 *
 * @param customerId
 * @return first Contact or null if customer has none
 */
 private Contact getFirstContact(Integer customerId) throws Exception {
 assert customerId != null;

 beginTx();
 try {
 final Customer customer = (Customer) getEnvironment().getSessionFactory().getCurrentSession().load(Customer.class, customerId);
 Set<Contact> contacts = customer.getContacts();
 Contact firstContact = contacts.isEmpty() ? null : contacts.iterator().next();
 commitTx();
 return firstContact;
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * -load existing Customer
 * -create a new Contact and add to customer's contacts
 *
 * @param customerId
 * @return added Contact
 */
 private Contact addContact(Integer customerId) throws Exception {
 assert customerId != null;

 beginTx();
 try {
 final Customer customer = (Customer) getEnvironment().getSessionFactory().getCurrentSession().load(Customer.class, customerId);

 Contact contact = new Contact();
 contact.setName("contact name");
 contact.setTlf("wtf is tlf?");

 contact.setCustomer(customer);
 customer.getContacts().add(contact);

 //assuming contact is persisted via cascade from customer
 commitTx();

 return contact;
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * remove existing 'contact' from customer's list of contacts
 *
 * @param contact contact to remove from customer's contacts
 * @param customerId
 * @throws IllegalStateException if customer does not own a contact
 */
 private void removeContact(Integer customerId) throws Exception {
 assert customerId != null;

 beginTx();
 try {
 Customer customer = (Customer) getEnvironment().getSessionFactory().getCurrentSession().load(Customer.class, customerId);
 Set<Contact> contacts = customer.getContacts();
 if (contacts.size() != 1) {
 throw new IllegalStateException("can't remove contact: customer id=" + customerId + " expected exactly 1 contact, " +
 "actual count=" + contacts.size());
 }

 Contact contact = contacts.iterator().next();
 contacts.remove(contact);
 contact.setCustomer(null);

 //explicitly delete Contact because hbm has no 'DELETE_ORPHAN' cascade?
 //getEnvironment().getSessionFactory().getCurrentSession().delete(contact); //appears to not be needed

 //assuming contact is persisted via cascade from customer
 commitTx();
 } catch (Exception e) {
 rollbackTx();
 throw e;
 }
 }

 /**
 * @return the customerIDs
 */
 public Set<Integer> getCustomerIDs() {
 return customerIDs;
 }

 private String statusOfRunnersToString(Set<UserRunner> runners) {
 assert runners != null;

 StringBuilder sb = new StringBuilder("TEST CONFIG [userCount=" + USER_COUNT +
 ", iterationsPerUser=" + ITERATION_COUNT +
 ", thinkTimeMillis=" + THINK_TIME_MILLIS + "] " +
 " STATE of UserRunners: ");

 for (UserRunner r : runners) {
 sb.append(r.toString() + System.getProperty("line.separator"));
 }
 return sb.toString();
 }

 class UserRunner implements Runnable {

 final private Integer customerId;
 private int completedIterations = 0;
 private Throwable causeOfFailure;

 public UserRunner(final Integer cId) {
 assert cId != null;
 this.customerId = cId;
 }

 private boolean contactExists() throws Exception {
 return getFirstContact(customerId) != null;
 }

 public void run() {

 //name this thread for easier log tracing
 Thread.currentThread().setName("UserRunnerThread-" + getCustomerId());
 try {
 for (int i = 0; i < ITERATION_COUNT; i++) {

 if (contactExists()) {
 throw new IllegalStateException("contact already exists before add, customerId=" + customerId);
 }

 addContact(customerId);

 if (!contactExists()) {
 throw new IllegalStateException("contact missing after successful add, customerId=" + customerId);
 }

 thinkRandomTime();

 //read everyone's contacts
 readEveryonesFirstContact();

 thinkRandomTime();

 removeContact(customerId);

 if (contactExists()) {
 throw new IllegalStateException("contact still exists after successful remove call, customerId=" + customerId);
 }

 thinkRandomTime();

 ++completedIterations;
 }

 } catch (Throwable t) {

 this.causeOfFailure = t;
 TERMINATE_ALL_USERS = true;

 //rollback current transaction if any
 //really should not happen since above methods all follow begin-commit-rollback pattern
 try {
 if (DualNodeJtaTransactionManagerImpl.getInstance(DualNodeTestUtil.LOCAL).getTransaction() != null) {
 DualNodeJtaTransactionManagerImpl.getInstance(DualNodeTestUtil.LOCAL).rollback();
 }
 } catch (SystemException ex) {
 throw new RuntimeException("failed to rollback tx", ex);
 }
 }
 }

 public boolean isSuccess() {
 return ITERATION_COUNT == getCompletedIterations();
 }

 public int getCompletedIterations() {
 return completedIterations;
 }

 public Throwable getCauseOfFailure() {
 return causeOfFailure;
 }

 public Integer getCustomerId() {
 return customerId;
 }

 @Override
 public String toString() {
 return super.toString() +
 "[customerId=" + getCustomerId() +
 " iterationsCompleted=" + getCompletedIterations() +
 " completedAll=" + isSuccess() +
 " causeOfFailure=" + (this.causeOfFailure != null ? ExceptionUtils.getStackTrace(causeOfFailure) : "") + "] ";
 }
 }

 /**
 * sleep between 0 and THINK_TIME_MILLIS.
 * @throws RuntimeException if sleep is interruped or TERMINATE_ALL_USERS flag was set to true in the meantime
 */
 private void thinkRandomTime() {
 try {
 Thread.sleep(random.nextInt(THINK_TIME_MILLIS));
 } catch (InterruptedException ex) {
 throw new RuntimeException("sleep interrupted", ex);
 }

 if (TERMINATE_ALL_USERS) {
 throw new RuntimeException("told to terminate (because a UserRunner had failed)");
 }
 }

 public static Test suite() {
 return new FunctionalTestClassTestSuite(MVCCConcurrentWriteTest.class);
 }
}

I am starting to suspect that InvocationContext doesn't always get injected with a TransactionContext - hopefully simply because I'd messed up my Hibernate/JBC/JTA/Spring config.

Posted a specific question on relation between the two contexts here:
http://www.jboss.org/index.html?module=bb&op=viewtopic&p=4215969

I think I found a bug:

Basically, because of failSilently==true putForExternalRead ends up adding Node to MVCCInvocationContext.lookedUpNodes but subsequent evictNode() uses MVCCInvocationContext.mvccTCtx to look up the same node (and fails to find it). See details here:

http://www.jboss.org/index.html?module=bb&op=viewtopic&t=151904

I think failure to acquire stems from failure to evict locks *after tx commit*. Here's a sample exception stack:

[11:03:29:881] main [ERROR] - org.hibernate.engine.ActionQueue.afterTransactionCompletion(line:207) - could not release a cache lock
org.hibernate.cache.CacheException: java.lang.IllegalStateException: Transaction a Bitronix Transaction with GTRID [737072696E672D62746D00000120159A386A00000033], status=COMMITTED, 1 resource(s) enlisted (started Tue Mar 17 11:03:29 PDT 2009) is not in a valid state to be invoking cache operations on.
at org.hibernate.cache.jbc2.util.CacheHelper.removeAll(CacheHelper.java:380)
at org.hibernate.cache.jbc2.util.CacheHelper.removeAll(CacheHelper.java:360)
at org.hibernate.cache.jbc2.access.TransactionalAccessDelegate.evictOrRemoveAll(TransactionalAccessDelegate.java:146)
at org.hibernate.cache.jbc2.access.TransactionalAccessDelegate.evictAll(TransactionalAccessDelegate.java:142)
at org.hibernate.cache.jbc2.entity.TransactionalAccess.evictAll(TransactionalAccess.java:102)
at org.hibernate.impl.SessionFactoryImpl.evictEntity(SessionFactoryImpl.java:870)
at org.hibernate.action.BulkOperationCleanupAction.evictEntityRegions(BulkOperationCleanupAction.java:153)
at org.hibernate.action.BulkOperationCleanupAction.afterTransactionCompletion(BulkOperationCleanupAction.java:132)
at org.hibernate.engine.ActionQueue.afterTransactionCompletion(ActionQueue.java:198)
at org.hibernate.impl.SessionImpl.afterTransactionCompletion(SessionImpl.java:451)
at org.hibernate.jdbc.JDBCContext.afterTransactionCompletion(JDBCContext.java:252)
at org.hibernate.transaction.CacheSynchronization.afterCompletion(CacheSynchronization.java:117)
at bitronix.tm.BitronixTransaction.fireAfterCompletionEvent(BitronixTransaction.java:382)
at bitronix.tm.BitronixTransaction.commit(BitronixTransaction.java:188)
at bitronix.tm.BitronixTransactionManager.commit(BitronixTransactionManager.java:96)
at org.springframework.transaction.jta.JtaTransactionManager.doCommit(JtaTransactionManager.java:1028)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:732)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:701)
at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:321)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:116)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:77)
at com.doppelganger.service.tx.TxRetryAspect.doRetryIfNeeded(TxRetryAspect.java:114)
at com.doppelganger.service.tx.TxRetryAspect.retryIfNeededForTxAnnotation(TxRetryAspect.java:90)
at sun.reflect.GeneratedMethodAccessor81.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:627)
at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:616)
at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:64)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:160)
at org.springframework.aop.aspectj.AspectJAfterAdvice.invoke(AspectJAfterAdvice.java:42)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.adapter.MethodBeforeAdviceInterceptor.invoke(MethodBeforeAdviceInterceptor.java:50)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:89)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
at $Proxy96.leaveServer(Unknown Source)
at com.doppelganger.test.TestServer.leave(TestServer.java:123)
at com.doppelganger.service.userrelations.UserToUserSessionTest.testUserSessionLifecycle(UserToUserSessionTest.java:143)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at junit.framework.TestCase.runTest(TestCase.java:168)
at junit.framework.TestCase.runBare(TestCase.java:134)
at org.springframework.test.ConditionalTestCase.runBare(ConditionalTestCase.java:76)
at junit.framework.TestResult$1.protect(TestResult.java:110)
at junit.framework.TestResult.runProtected(TestResult.java:128)
at junit.framework.TestResult.run(TestResult.java:113)
at junit.framework.TestCase.run(TestCase.java:124)
at junit.framework.TestSuite.runTest(TestSuite.java:232)
at junit.framework.TestSuite.run(TestSuite.java:227)
at org.junit.internal.runners.JUnit38ClassRunner.run(JUnit38ClassRunner.java:79)
at junit.framework.JUnit4TestAdapter.run(JUnit4TestAdapter.java:39)
at org.apache.tools.ant.taskdefs.optional.junit.JUnitTestRunner.run(JUnitTestRunner.java:420)
at org.apache.tools.ant.taskdefs.optional.junit.JUnitTestRunner.launch(JUnitTestRunner.java:911)
at org.apache.tools.ant.taskdefs.optional.junit.JUnitTestRunner.main(JUnitTestRunner.java:768)
Caused by: java.lang.IllegalStateException: Transaction a Bitronix Transaction with GTRID [737072696E672D62746D00000120159A386A00000033], status=COMMITTED, 1 resource(s) enlisted (started Tue Mar 17 11:03:29 PDT 2009) is not in a valid state to be invoking cache operations on.
at org.jboss.cache.interceptors.TxInterceptor.registerTransaction(TxInterceptor.java:604)
at org.jboss.cache.interceptors.TxInterceptor.attachGlobalTransaction(TxInterceptor.java:405)
at org.jboss.cache.interceptors.TxInterceptor.attachGtxAndPassUpChain(TxInterceptor.java:260)
at org.jboss.cache.interceptors.TxInterceptor.handleDefault(TxInterceptor.java:250)
at org.jboss.cache.commands.AbstractVisitor.visitRemoveNodeCommand(AbstractVisitor.java:75)
at org.jboss.cache.commands.write.RemoveNodeCommand.acceptVisitor(RemoveNodeCommand.java:125)
at org.jboss.cache.interceptors.base.CommandInterceptor.invokeNextInterceptor(CommandInterceptor.java:116)
at org.jboss.cache.interceptors.base.CommandInterceptor.handleDefault(CommandInterceptor.java:131)
at org.jboss.cache.commands.AbstractVisitor.visitRemoveNodeCommand(AbstractVisitor.java:75)
at org.jboss.cache.commands.write.RemoveNodeCommand.acceptVisitor(RemoveNodeCommand.java:125)
at org.jboss.cache.interceptors.base.CommandInterceptor.invokeNextInterceptor(CommandInterceptor.java:116)
at org.jboss.cache.interceptors.InvocationContextInterceptor.handleAll(InvocationContextInterceptor.java:178)
at org.jboss.cache.interceptors.InvocationContextInterceptor.visitRemoveNodeCommand(InvocationContextInterceptor.java:88)
at org.jboss.cache.commands.write.RemoveNodeCommand.acceptVisitor(RemoveNodeCommand.java:125)
at org.jboss.cache.interceptors.InterceptorChain.invoke(InterceptorChain.java:287)
at org.jboss.cache.invocation.CacheInvocationDelegate.removeNode(CacheInvocationDelegate.java:409)
at org.hibernate.cache.jbc2.util.CacheHelper.removeAll(CacheHelper.java:378)
... 58 more

 boolean txValid = TransactionTable.isValid(tx);
 if (!txValid) {
 throw new IllegalStateException("Transaction " + tx + " is not in a valid state to be invoking cache operations on.");
 }

The TransactionManager.commit method completes the transaction currently
associated with the calling thread. After the commit method returns, the calling thread
is not associated with a transaction.

TransactionManager's commit(){

//do commit
finally{
 fireSynchronizationsAfterTxCompletion(); //still in commit()
}
}

 /**
 *
 * @return a valid, usable Transaction if one exists, null otherwise
 * @throws javax.transaction.SystemException
 */
 private Transaction getValidTransaction() throws SystemException
 {
 // this creates a context if one did not exist.
 if (txManager == null)
 {
 if (trace) log.trace("no transaction manager configured, setting tx as null.");
 return null;
 }
 else
 {
 Transaction tx = txManager.getTransaction();
 //is this TX still usable?
 if(tx != null && tx.getStatus() == Status.ACTIVE) //or 'if (tx != null && TransactionTable.isValid(tx))' - not sure
 {
 return tx;
 }
 else{
 return null;
 }
 }
 }