[infinispan-commits] Infinispan SVN: r601 - in trunk/core/src/test/java/org/infinispan: test and 1 other directories.
infinispan-commits at lists.jboss.org
infinispan-commits at lists.jboss.org
Tue Jul 21 12:23:23 EDT 2009
Author: mircea.markus
Date: 2009-07-21 12:23:23 -0400 (Tue, 21 Jul 2009)
New Revision: 601
Added:
trunk/core/src/test/java/org/infinispan/test/PerCacheExecutorThread.java
trunk/core/src/test/java/org/infinispan/tx/LocalDeadlockDetectionTest.java
trunk/core/src/test/java/org/infinispan/tx/ReplDeadlockDetectionTest.java
Removed:
trunk/core/src/test/java/org/infinispan/tx/DeadlockDetectionTest.java
Modified:
trunk/core/src/test/java/org/infinispan/distribution/DeadlockDetectionDistributionTest.java
trunk/core/src/test/java/org/infinispan/test/AbstractCacheTest.java
Log:
[ISPN-38] - (eager deadlock detection) - more detailed UT for local caches
Modified: trunk/core/src/test/java/org/infinispan/distribution/DeadlockDetectionDistributionTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/distribution/DeadlockDetectionDistributionTest.java 2009-07-21 14:29:24 UTC (rev 600)
+++ trunk/core/src/test/java/org/infinispan/distribution/DeadlockDetectionDistributionTest.java 2009-07-21 16:23:23 UTC (rev 601)
@@ -1,7 +1,7 @@
package org.infinispan.distribution;
import org.infinispan.config.Configuration;
-import org.infinispan.tx.DeadlockDetectionTest;
+import org.infinispan.tx.ReplDeadlockDetectionTest;
import static org.testng.Assert.fail;
import org.testng.annotations.Test;
@@ -11,7 +11,7 @@
* @author Mircea.Markus at jboss.com
*/
@Test(groups = "functional", enabled = false, testName = "tx.DeadlockDetectionDistributionTest")
-public class DeadlockDetectionDistributionTest extends DeadlockDetectionTest {
+public class DeadlockDetectionDistributionTest extends ReplDeadlockDetectionTest {
public DeadlockDetectionDistributionTest() {
cacheMode = Configuration.CacheMode.DIST_SYNC;
@@ -21,4 +21,26 @@
public void testDeadlockDetectedTwoTransactions() throws Exception {
fail("This test should be updated to make sure tx replicate on opposite nodes");
}
+
+
+ //following methods are overridden as TestNG will otherwise run them even if I mark the class as enabled = false
+ @Override
+ public void testDeadlockDetectionAndAsyncCaches() {
+ throw new IllegalStateException("TODO - please implement me!!!"); //todo implement!!!
+ }
+
+ @Override
+ public void testExpectedInnerStructure() {
+ throw new IllegalStateException("TODO - please implement me!!!"); //todo implement!!!
+ }
+
+ @Override
+ public void testDeadlockDetectedOneTx() throws Exception {
+ throw new IllegalStateException("TODO - please implement me!!!"); //todo implement!!!
+ }
+
+ @Override
+ public void testLockReleasedWhileTryingToAcquire() throws Exception {
+ throw new IllegalStateException("TODO - please implement me!!!"); //todo implement!!!
+ }
}
Modified: trunk/core/src/test/java/org/infinispan/test/AbstractCacheTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/test/AbstractCacheTest.java 2009-07-21 14:29:24 UTC (rev 600)
+++ trunk/core/src/test/java/org/infinispan/test/AbstractCacheTest.java 2009-07-21 16:23:23 UTC (rev 601)
@@ -118,4 +118,9 @@
configuration.setFetchInMemoryState(false);
return configuration;
}
+
+
+ protected boolean xor(boolean b1, boolean b2) {
+ return (b1 || b2) && !(b1 && b2);
+ }
}
Added: trunk/core/src/test/java/org/infinispan/test/PerCacheExecutorThread.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/test/PerCacheExecutorThread.java (rev 0)
+++ trunk/core/src/test/java/org/infinispan/test/PerCacheExecutorThread.java 2009-07-21 16:23:23 UTC (rev 601)
@@ -0,0 +1,216 @@
+package org.infinispan.test;
+
+import org.infinispan.Cache;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
+
+import javax.transaction.TransactionManager;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * Utility class that can be used for writing tests that need to access a cache instance from multiple threads.
+ *
+ * @author Mircea.Markus at jboss.com
+ * @see Operations
+ * @see PerCacheExecutorThread.OperationsResult
+ */
+public final class PerCacheExecutorThread extends Thread {
+
+ private static Log log = LogFactory.getLog(PerCacheExecutorThread.class);
+
+ private Cache<Object, Object> cache;
+ private BlockingQueue<Object> toExecute = new ArrayBlockingQueue<Object>(1);
+ private volatile Object response;
+ private CountDownLatch responseLatch = new CountDownLatch(1);
+
+ private volatile Object key, value;
+
+ public void setKeyValue(Object key, Object value) {
+ this.key = key;
+ this.value = value;
+ }
+
+ public PerCacheExecutorThread(Cache<Object, Object> cache, int index) {
+ super("PerCacheExecutorThread-" + index);
+ this.cache = cache;
+ start();
+ }
+
+ public Object execute(Operations op) {
+ try {
+ responseLatch = new CountDownLatch(1);
+ toExecute.put(op);
+ responseLatch.await();
+ return response;
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Unexpected", e);
+ }
+ }
+
+ public void executeNoResponse(Operations op) {
+ try {
+ responseLatch = null;
+ response = null;
+ toExecute.put(op);
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Unexpected", e);
+ }
+ }
+
+ @Override
+ public void run() {
+ Operations operation;
+ boolean run = true;
+ while (run) {
+ try {
+ operation = (Operations) toExecute.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ System.out.println("about to process operation " + operation);
+ switch (operation) {
+ case BEGGIN_TX: {
+ TransactionManager txManager = TestingUtil.getTransactionManager(cache);
+ try {
+ txManager.begin();
+ setResponse(OperationsResult.BEGGIN_TX_OK);
+ } catch (Exception e) {
+ log.trace("Failure on beggining tx", e);
+ setResponse(e);
+ }
+ break;
+ }
+ case COMMIT_TX: {
+ TransactionManager txManager = TestingUtil.getTransactionManager(cache);
+ try {
+ txManager.commit();
+ setResponse(OperationsResult.COMMIT_TX_OK);
+ } catch (Exception e) {
+ log.trace("Exception while committing tx", e);
+ setResponse(e);
+ }
+ break;
+ }
+ case PUT_KEY_VALUE: {
+ try {
+ cache.put(key, value);
+ log.trace("Successfully exucuted putKeyValue(" + key + ", " + value + ")");
+ setResponse(OperationsResult.PUT_KEY_VALUE_OK);
+ } catch (Exception e) {
+ log.trace("Exception while executing putKeyValue(" + key + ", " + value + ")", e);
+ setResponse(e);
+ }
+ break;
+ }
+ case REMOVE_KEY: {
+ try {
+ cache.remove(key);
+ log.trace("Successfully exucuted remove(" + key + ")");
+ setResponse(OperationsResult.REMOVE_KEY_OK);
+ } catch (Exception e) {
+ log.trace("Exception while executing remove(" + key + ")", e);
+ setResponse(e);
+ }
+ break;
+ }
+ case REPLACE_KEY_VALUE: {
+ try {
+ cache.replace(key, value);
+ log.trace("Successfully exucuted replace(" + key + "," + value + ")");
+ setResponse(OperationsResult.REPLACE_KEY_VALUE_OK);
+ } catch (Exception e) {
+ log.trace("Exception while executing replace(" + key + "," + value + ")", e);
+ setResponse(e);
+ }
+ break;
+ }
+ case STOP_THREAD: {
+ System.out.println("Exiting...");
+ toExecute = null;
+ run = false;
+ break;
+ }
+ default : {
+ setResponse(new IllegalStateException("Unknown operation!" + operation));
+ }
+ }
+ if (responseLatch != null) responseLatch.countDown();
+ }
+ setResponse("EXIT");
+ }
+
+ private void setResponse(Object e) {
+ log.trace("setResponse to " + e);
+ response = e;
+ }
+
+ public void stopThread() {
+ execute(Operations.STOP_THREAD);
+ while (!this.getState().equals(State.TERMINATED)) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+ }
+
+ public Object lastResponse() {
+ return response;
+ }
+
+ public void clearResponse() {
+ response = null;
+ }
+
+ public Object waitForResponse() {
+ while (response == null) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return response;
+ }
+
+ /**
+ * Defines allowed operations for {@link PerCacheExecutorThread}.
+ *
+ * @author Mircea.Markus at jboss.com
+ */
+ public static enum Operations {
+ BEGGIN_TX, COMMIT_TX, PUT_KEY_VALUE, REMOVE_KEY, REPLACE_KEY_VALUE, STOP_THREAD;
+ public OperationsResult getCorrespondingOkResult() {
+ switch (this) {
+ case BEGGIN_TX:
+ return OperationsResult.BEGGIN_TX_OK;
+ case COMMIT_TX:
+ return OperationsResult.COMMIT_TX_OK;
+ case PUT_KEY_VALUE:
+ return OperationsResult.PUT_KEY_VALUE_OK;
+ case REMOVE_KEY:
+ return OperationsResult.REMOVE_KEY_OK;
+ case REPLACE_KEY_VALUE:
+ return OperationsResult.REPLACE_KEY_VALUE_OK;
+ case STOP_THREAD:
+ return OperationsResult.STOP_THREAD_OK;
+ default:
+ throw new IllegalStateException("Unrecognized operation: " + this);
+ }
+ }
+
+ }
+
+ /**
+ * Defines operation results returned by {@link PerCacheExecutorThread}.
+ *
+ * @author Mircea.Markus at jboss.com
+ */
+ public static enum OperationsResult {
+ BEGGIN_TX_OK, COMMIT_TX_OK, PUT_KEY_VALUE_OK, REMOVE_KEY_OK, REPLACE_KEY_VALUE_OK, STOP_THREAD_OK
+
+ }
+}
Property changes on: trunk/core/src/test/java/org/infinispan/test/PerCacheExecutorThread.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Deleted: trunk/core/src/test/java/org/infinispan/tx/DeadlockDetectionTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/tx/DeadlockDetectionTest.java 2009-07-21 14:29:24 UTC (rev 600)
+++ trunk/core/src/test/java/org/infinispan/tx/DeadlockDetectionTest.java 2009-07-21 16:23:23 UTC (rev 601)
@@ -1,557 +0,0 @@
-package org.infinispan.tx;
-
-import org.infinispan.Cache;
-import org.infinispan.api.mvcc.LockAssert;
-import org.infinispan.commands.ReplicableCommand;
-import org.infinispan.config.Configuration;
-import org.infinispan.config.ConfigurationException;
-import org.infinispan.context.impl.NonTxInvocationContext;
-import org.infinispan.interceptors.DeadlockDetectingInterceptor;
-import org.infinispan.interceptors.InterceptorChain;
-import org.infinispan.manager.CacheManager;
-import org.infinispan.remoting.ReplicationException;
-import org.infinispan.remoting.responses.Response;
-import org.infinispan.remoting.rpc.ResponseFilter;
-import org.infinispan.remoting.rpc.ResponseMode;
-import org.infinispan.remoting.rpc.RpcManager;
-import org.infinispan.remoting.transport.Address;
-import org.infinispan.remoting.transport.Transport;
-import org.infinispan.statetransfer.StateTransferException;
-import org.infinispan.test.MultipleCacheManagersTest;
-import org.infinispan.test.TestingUtil;
-import org.infinispan.test.fwk.TestCacheManagerFactory;
-import org.infinispan.transaction.lookup.DummyTransactionManagerLookup;
-import org.infinispan.util.concurrent.NotifyingNotifiableFuture;
-import org.infinispan.util.concurrent.locks.DeadlockDetectingLockManager;
-import org.infinispan.util.concurrent.locks.LockManager;
-import org.infinispan.util.concurrent.locks.DeadlockDetectedException;
-import org.infinispan.util.logging.Log;
-import org.infinispan.util.logging.LogFactory;
-import static org.testng.Assert.assertEquals;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.Test;
-
-import javax.transaction.TransactionManager;
-import javax.transaction.RollbackException;
-import java.util.List;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
-
-/**
- * Functional test for deadlock detection.
- *
- * @author Mircea.Markus at jboss.com
- * TODO - add test deadlock with distribution
- */
- at Test(testName = "tx.DeadlockDetectionTest", groups = "functional")
-public class DeadlockDetectionTest extends MultipleCacheManagersTest {
-
- protected ControlledRpcManager controlledRpcManager1;
- protected ControlledRpcManager controlledRpcManager2;
- protected CountDownLatch replicationLatch;
- protected ExecutorThread t1;
- protected ExecutorThread t2;
- protected DeadlockDetectingLockManager ddLm1;
- protected DeadlockDetectingLockManager ddLm2;
-
- protected Configuration.CacheMode cacheMode = Configuration.CacheMode.REPL_SYNC;
-
- protected void createCacheManagers() throws Throwable {
- Configuration config = getDefaultClusteredConfig(cacheMode);
- config.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName());
- config.setEnableDeadlockDetection(true);
- config.setSyncCommitPhase(true);
- config.setSyncRollbackPhase(true);
- config.setUseLockStriping(false);
- assert config.isEnableDeadlockDetection();
- createClusteredCaches(2, "test", config);
- assert config.isEnableDeadlockDetection();
-
- assert cache(0, "test").getConfiguration().isEnableDeadlockDetection();
- assert cache(1, "test").getConfiguration().isEnableDeadlockDetection();
- assert !cache(0, "test").getConfiguration().isExposeJmxStatistics();
- assert !cache(1, "test").getConfiguration().isExposeJmxStatistics();
-
- ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(0, "test"))).setExposeJmxStats(true);
- ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(1, "test"))).setExposeJmxStats(true);
-
- RpcManager rpcManager1 = TestingUtil.extractComponent(cache(0, "test"), RpcManager.class);
- RpcManager rpcManager2 = TestingUtil.extractComponent(cache(1, "test"), RpcManager.class);
-
- controlledRpcManager1 = new ControlledRpcManager(rpcManager1);
- controlledRpcManager2 = new ControlledRpcManager(rpcManager2);
- TestingUtil.replaceComponent(cache(0, "test"), RpcManager.class, controlledRpcManager1, true);
- TestingUtil.replaceComponent(cache(1, "test"), RpcManager.class, controlledRpcManager2, true);
-
- assert TestingUtil.extractComponent(cache(0, "test"), RpcManager.class) instanceof ControlledRpcManager;
- assert TestingUtil.extractComponent(cache(1, "test"), RpcManager.class) instanceof ControlledRpcManager;
-
- ddLm1 = (DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(0, "test"));
- ddLm2 = (DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(1, "test"));
- }
-
-
- @BeforeMethod
- public void beforeMethod() {
- t1 = new ExecutorThread(cache(0, "test"), 1);
- t2 = new ExecutorThread(cache(1, "test"), 2);
- replicationLatch = new CountDownLatch(1);
- controlledRpcManager1.setReplicationLatch(replicationLatch);
- controlledRpcManager2.setReplicationLatch(replicationLatch);
- log.trace("_________________________ Here is beggins");
- }
-
- @AfterMethod
- public void afterMethod() {
- t1.stopThread();
- t2.stopThread();
- ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(0, "test"))).resetStatistics();
- ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(1, "test"))).resetStatistics();
- }
-
- public void testDeadlockDetectionAndAsyncCaches() {
- Configuration config = getDefaultClusteredConfig(Configuration.CacheMode.REPL_ASYNC);
- config.setEnableDeadlockDetection(true);
- config.setUseLockStriping(false);
- CacheManager cm = TestCacheManagerFactory.createClusteredCacheManager();
- cm.defineCache("test", config);
- try {
- cm.getCache("test");
- assert false : "Exception expected";
- } catch (ConfigurationException e) {
- //expected
- System.out.println("Error message is " + e.getMessage());
- }
- cm.stop();
- }
-
- public void testExpectedInnerStructure() {
- LockManager lockManager = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
- assert lockManager instanceof DeadlockDetectingLockManager;
-
- InterceptorChain ic = TestingUtil.extractComponent(cache(0, "test"), InterceptorChain.class);
- assert ic.containsInterceptorType(DeadlockDetectingInterceptor.class);
- }
-
- public void testDeadlockDetectedTwoTransactions() throws Exception {
- t1.setKeyValue("key", "value1");
- t2.setKeyValue("key", "value2");
- assert OperationsResult.BEGGIN_TX_OK == t1.execute(Operations.BEGGIN_TX);
- assert OperationsResult.BEGGIN_TX_OK == t2.execute(Operations.BEGGIN_TX);
- System.out.println("After beggin");
-
- t1.execute(Operations.PUT_KEY_VALUE);
- t2.execute(Operations.PUT_KEY_VALUE);
- System.out.println("After put key value");
-
- t1.clearResponse();
- t2.clearResponse();
-
- t1.executeNoResponse(Operations.COMMIT_TX);
- t2.executeNoResponse(Operations.COMMIT_TX);
-
- System.out.println("Now replication is triggered");
- replicationLatch.countDown();
-
-
- Object t1Commit = t1.waitForResponse();
- Object t2Commit = t2.waitForResponse();
- System.out.println("After commit: " + t1Commit + ", " + t2Commit);
-
- assert xor(t1Commit instanceof Exception, t2Commit instanceof Exception) : "only one thread must be failing " + t1Commit + "," + t2Commit;
- System.out.println("t2Commit = " + t2Commit);
- System.out.println("t1Commit = " + t1Commit);
-
- if (t1Commit instanceof Exception) {
- System.out.println("t1 rolled back");
- Object o = cache(0, "test").get("key");
- assert o != null;
- assert o.equals("value2");
- } else {
- System.out.println("t2 rolled back");
- Object o = cache(0, "test").get("key");
- assert o != null;
- assert o.equals("value1");
- o = cache(1, "test").get("key");
- assert o != null;
- assert o.equals("value1");
- }
-
- assert ddLm1.getDetectedDeadlocks() + ddLm2.getDetectedDeadlocks() >= 1;
-
- LockManager lm1 = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
- assert !lm1.isLocked("key") : "It is locked by " + lm1.getOwner("key");
- LockManager lm2 = TestingUtil.extractComponent(cache(1, "test"), LockManager.class);
- assert !lm2.isLocked("key") : "It is locked by " + lm2.getOwner("key");
- LockAssert.assertNoLocks(cache(0, "test"));
- }
-
- public void testLocalVsLocalTxDeadlock() {
- CacheManager cm = null;
- try {
- cm = TestCacheManagerFactory.createLocalCacheManager();
- Configuration configuration = new Configuration();
- configuration.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName());
- configuration.setEnableDeadlockDetection(true);
- configuration.setUseLockStriping(false);
- configuration.setExposeJmxStatistics(true);
- cm.defineCache("test", configuration);
- Cache localCache = cm.getCache("test");
- DeadlockDetectingLockManager lockManager = (DeadlockDetectingLockManager) TestingUtil.extractLockManager(localCache);
-
- ExecutorThread t1 = new ExecutorThread(localCache, 0);
- ExecutorThread t2 = new ExecutorThread(localCache, 1);
-
-
- assert OperationsResult.BEGGIN_TX_OK == t1.execute(Operations.BEGGIN_TX);
- assert OperationsResult.BEGGIN_TX_OK == t2.execute(Operations.BEGGIN_TX);
- System.out.println("After beggin");
-
- t1.setKeyValue("k1", "value_1_t1");
- t2.setKeyValue("k2", "value_2_t2");
-
- assert OperationsResult.PUT_KEY_VALUE_OK == t1.execute(Operations.PUT_KEY_VALUE);
- assert OperationsResult.PUT_KEY_VALUE_OK == t2.execute(Operations.PUT_KEY_VALUE);
-
- System.out.println("After first PUT");
- assert lockManager.isLocked("k1");
- assert lockManager.isLocked("k2");
-
-
- t1.setKeyValue("k2", "value_2_t1");
- t2.setKeyValue("k1", "value_1_t2");
- t1.executeNoResponse(Operations.PUT_KEY_VALUE);
- t2.executeNoResponse(Operations.PUT_KEY_VALUE);
-
- Object response1 = t1.waitForResponse();
- Object response2 = t2.waitForResponse();
-
- assert xor(response1 instanceof DeadlockDetectedException, response2 instanceof DeadlockDetectedException) : "expected one and only one exception: " + response1 + ", " + response2;
- assert xor(response1 == OperationsResult.PUT_KEY_VALUE_OK, response2 == OperationsResult.PUT_KEY_VALUE_OK) : "expected one and only one exception: " + response1 + ", " + response2;
-
- assert lockManager.isLocked("k1");
- assert lockManager.isLocked("k2");
- assert lockManager.getOwner("k1") == lockManager.getOwner("k2");
-
- if (response1 instanceof Exception) {
- assert OperationsResult.COMMIT_TX_OK == t2.execute(Operations.COMMIT_TX);
- assertEquals("value_1_t2", localCache.get("k1"));
- assertEquals("value_2_t2", localCache.get("k2"));
- assert t1.execute(Operations.COMMIT_TX) instanceof RollbackException;
- } else {
- assert OperationsResult.COMMIT_TX_OK == t1.execute(Operations.COMMIT_TX);
- assertEquals("value_1_t1", localCache.get("k1"));
- assertEquals("value_2_t1", localCache.get("k2"));
- assert t2.execute(Operations.COMMIT_TX) instanceof RollbackException;
- }
- assert lockManager.getNumberOfLocksHeld() == 0;
- assertEquals(lockManager.getDetectedDeadlocks(), 1);
- } finally {
- TestingUtil.killCacheManagers(cm);
- }
- }
-
-
- public void testDeadlockDetectedOneTx() throws Exception {
- t1.setKeyValue("key", "value1");
-
- LockManager lm2 = TestingUtil.extractComponent(cache(1, "test"), LockManager.class);
- NonTxInvocationContext ctx = cache(1, "test").getAdvancedCache().getInvocationContextContainer().createNonTxInvocationContext();
- lm2.lockAndRecord("key", ctx);
- assert lm2.isLocked("key");
-
-
- assert OperationsResult.BEGGIN_TX_OK == t1.execute(Operations.BEGGIN_TX) : "but received " + t1.lastResponse();
- t1.execute(Operations.PUT_KEY_VALUE);
-
- t1.clearResponse();
- t1.executeNoResponse(Operations.COMMIT_TX);
-
- replicationLatch.countDown();
- System.out.println("Now replication is triggered");
-
- t1.waitForResponse();
-
-
- Object t1CommitRsp = t1.lastResponse();
-
- assert t1CommitRsp instanceof Exception : "expected exception, received " + t1.lastResponse();
-
- LockManager lm1 = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
- assert !lm1.isLocked("key") : "It is locked by " + lm1.getOwner("key");
-
- lm2.unlock("key", ctx.getLockOwner());
- assert !lm2.isLocked("key");
- assert !lm1.isLocked("key");
- }
-
- public void testLockReleasedWhileTryingToAcquire() throws Exception {
- t1.setKeyValue("key", "value1");
-
- LockManager lm2 = TestingUtil.extractComponent(cache(1, "test"), LockManager.class);
- NonTxInvocationContext ctx = cache(1, "test").getAdvancedCache().getInvocationContextContainer().createNonTxInvocationContext();
- lm2.lockAndRecord("key", ctx);
- assert lm2.isLocked("key");
-
-
- assert OperationsResult.BEGGIN_TX_OK == t1.execute(Operations.BEGGIN_TX) : "but received " + t1.lastResponse();
- t1.execute(Operations.PUT_KEY_VALUE);
-
- t1.clearResponse();
- t1.executeNoResponse(Operations.COMMIT_TX);
-
- replicationLatch.countDown();
-
- Thread.sleep(3000); //just to make sure the remote tx thread managed to spin around for some times.
- lm2.unlock("key", ctx.getLockOwner());
-
- t1.waitForResponse();
-
-
- Object t1CommitRsp = t1.lastResponse();
-
- assert t1CommitRsp == OperationsResult.COMMIT_TX_OK : "expected true, received " + t1.lastResponse();
-
- LockManager lm1 = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
- assert !lm1.isLocked("key") : "It is locked by " + lm1.getOwner("key");
-
- assert !lm2.isLocked("key");
- assert !lm1.isLocked("key");
- }
-
- public static enum Operations {
- BEGGIN_TX, COMMIT_TX, PUT_KEY_VALUE, STOP_THREAD
- }
-
- public static enum OperationsResult {
- BEGGIN_TX_OK, COMMIT_TX_OK, PUT_KEY_VALUE_OK, STOP_THREAD_OK
- }
-
- public static final class ExecutorThread extends Thread {
-
- private static Log log = LogFactory.getLog(ExecutorThread.class);
-
- private Cache<Object, Object> cache;
- private BlockingQueue<Object> toExecute = new ArrayBlockingQueue<Object>(1);
- private volatile Object response;
- private CountDownLatch responseLatch = new CountDownLatch(1);
-
- private volatile Object key, value;
-
- public void setKeyValue(Object key, Object value) {
- this.key = key;
- this.value = value;
- }
-
- public ExecutorThread(Cache<Object, Object> cache, int index) {
- super("ExecutorThread-" + index);
- this.cache = cache;
- start();
- }
-
- public Object execute(Operations op) {
- try {
- responseLatch = new CountDownLatch(1);
- toExecute.put(op);
- responseLatch.await();
- return response;
- } catch (InterruptedException e) {
- throw new RuntimeException("Unexpected", e);
- }
- }
-
- public void executeNoResponse(Operations op) {
- try {
- responseLatch = null;
- response = null;
- toExecute.put(op);
- } catch (InterruptedException e) {
- throw new RuntimeException("Unexpected", e);
- }
- }
-
- @Override
- public void run() {
- Operations operation;
- boolean run = true;
- while (run) {
- try {
- operation = (Operations) toExecute.take();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- System.out.println("about to process operation " + operation);
- switch (operation) {
- case BEGGIN_TX: {
- TransactionManager txManager = TestingUtil.getTransactionManager(cache);
- try {
- txManager.begin();
- setResponse(OperationsResult.BEGGIN_TX_OK);
- } catch (Exception e) {
- log.trace("Failure on beggining tx", e);
- setResponse(e);
- }
- break;
- }
- case COMMIT_TX: {
- TransactionManager txManager = TestingUtil.getTransactionManager(cache);
- try {
- txManager.commit();
- setResponse(OperationsResult.COMMIT_TX_OK);
- } catch (Exception e) {
- log.trace("Exception while committing tx", e);
- setResponse(e);
- }
- break;
- }
- case PUT_KEY_VALUE: {
- try {
- cache.put(key, value);
- log.trace("Successfully exucuted putKeyValue(" + key + ", " + value + ")");
- setResponse(OperationsResult.PUT_KEY_VALUE_OK);
- } catch (Exception e) {
- log.trace("Exception while executing putKeyValue(" + key + ", " + value + ")", e);
- setResponse(e);
- }
- break;
- }
- case STOP_THREAD: {
- System.out.println("Exiting...");
- toExecute = null;
- run = false;
- break;
- }
- }
- if (responseLatch != null) responseLatch.countDown();
- }
- setResponse("EXIT");
- }
-
- private void setResponse(Object e) {
- log.trace("setResponse to " + e);
- response = e;
- }
-
- public void stopThread() {
- execute(Operations.STOP_THREAD);
- while (!this.getState().equals(State.TERMINATED)) {
- try {
- Thread.sleep(50);
- } catch (InterruptedException e) {
- throw new IllegalStateException(e);
- }
- }
- }
-
- public Object lastResponse() {
- return response;
- }
-
- public void clearResponse() {
- response = null;
- }
-
- public Object waitForResponse() {
- while (response == null) {
- try {
- Thread.sleep(50);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- }
- return response;
- }
- }
-
- protected boolean xor(boolean b1, boolean b2) {
- return (b1 || b2) && !(b1 && b2);
- }
-
- public static final class ControlledRpcManager implements RpcManager {
-
- private volatile CountDownLatch replicationLatch;
-
- public ControlledRpcManager(RpcManager realOne) {
- this.realOne = realOne;
- }
-
- private RpcManager realOne;
-
- public void setReplicationLatch(CountDownLatch replicationLatch) {
- this.replicationLatch = replicationLatch;
- }
-
- public List<Response> invokeRemotely(List<Address> recipients, ReplicableCommand rpcCommand, ResponseMode mode, long timeout, boolean usePriorityQueue, ResponseFilter responseFilter) {
- return realOne.invokeRemotely(recipients, rpcCommand, mode, timeout, usePriorityQueue, responseFilter);
- }
-
- public List<Response> invokeRemotely(List<Address> recipients, ReplicableCommand rpcCommand, ResponseMode mode, long timeout, boolean usePriorityQueue) {
- return realOne.invokeRemotely(recipients, rpcCommand, mode, timeout, usePriorityQueue);
- }
-
- public List<Response> invokeRemotely(List<Address> recipients, ReplicableCommand rpcCommand, ResponseMode mode, long timeout) throws Exception {
- return realOne.invokeRemotely(recipients, rpcCommand, mode, timeout);
- }
-
- public void retrieveState(String cacheName, long timeout) throws StateTransferException {
- realOne.retrieveState(cacheName, timeout);
- }
-
- public void broadcastRpcCommand(ReplicableCommand rpc, boolean sync) throws ReplicationException {
- waitFirst();
- realOne.broadcastRpcCommand(rpc, sync);
- }
-
- public void broadcastRpcCommand(ReplicableCommand rpc, boolean sync, boolean usePriorityQueue) throws ReplicationException {
- waitFirst();
- realOne.broadcastRpcCommand(rpc, sync, usePriorityQueue);
- }
-
- private void waitFirst() {
- System.out.println(Thread.currentThread().getName() + " -- replication trigger called!");
- try {
- replicationLatch.await();
- } catch (Exception e) {
- throw new RuntimeException("Unexpected exception!", e);
- }
- }
-
- public void broadcastRpcCommandInFuture(ReplicableCommand rpc, NotifyingNotifiableFuture<Object> future) {
- realOne.broadcastRpcCommandInFuture(rpc, future);
- }
-
- public void broadcastRpcCommandInFuture(ReplicableCommand rpc, boolean usePriorityQueue, NotifyingNotifiableFuture<Object> future) {
- realOne.broadcastRpcCommandInFuture(rpc, usePriorityQueue, future);
- }
-
- public void invokeRemotely(List<Address> recipients, ReplicableCommand rpc, boolean sync) throws ReplicationException {
- realOne.invokeRemotely(recipients, rpc, sync);
- }
-
- public void invokeRemotely(List<Address> recipients, ReplicableCommand rpc, boolean sync, boolean usePriorityQueue) throws ReplicationException {
- realOne.invokeRemotely(recipients, rpc, sync, usePriorityQueue);
- }
-
- public void invokeRemotelyInFuture(List<Address> recipients, ReplicableCommand rpc, NotifyingNotifiableFuture<Object> future) {
- realOne.invokeRemotelyInFuture(recipients, rpc, future);
- }
-
- public void invokeRemotelyInFuture(List<Address> recipients, ReplicableCommand rpc, boolean usePriorityQueue, NotifyingNotifiableFuture<Object> future) {
- realOne.invokeRemotelyInFuture(recipients, rpc, usePriorityQueue, future);
- }
-
- public void invokeRemotelyInFuture(List<Address> recipients, ReplicableCommand rpc, boolean usePriorityQueue, NotifyingNotifiableFuture<Object> future, long timeout) {
- realOne.invokeRemotelyInFuture(recipients, rpc, usePriorityQueue, future, timeout);
- }
-
- public Transport getTransport() {
- return realOne.getTransport();
- }
-
- public Address getCurrentStateTransferSource() {
- return realOne.getCurrentStateTransferSource();
- }
- }
-}
Added: trunk/core/src/test/java/org/infinispan/tx/LocalDeadlockDetectionTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/tx/LocalDeadlockDetectionTest.java (rev 0)
+++ trunk/core/src/test/java/org/infinispan/tx/LocalDeadlockDetectionTest.java 2009-07-21 16:23:23 UTC (rev 601)
@@ -0,0 +1,193 @@
+package org.infinispan.tx;
+
+import org.infinispan.config.Configuration;
+import org.infinispan.manager.CacheManager;
+import org.infinispan.test.PerCacheExecutorThread;
+import org.infinispan.test.SingleCacheManagerTest;
+import org.infinispan.test.TestingUtil;
+import org.infinispan.test.fwk.TestCacheManagerFactory;
+import org.infinispan.transaction.lookup.DummyTransactionManagerLookup;
+import org.infinispan.util.concurrent.locks.DeadlockDetectedException;
+import org.infinispan.util.concurrent.locks.DeadlockDetectingLockManager;
+import static org.testng.Assert.assertEquals;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import javax.transaction.RollbackException;
+
+/**
+ * Tests deadlock detection functionality for local caches.
+ *
+ * @author Mircea.Markus at jboss.com
+ */
+ at Test(groups = "functional", testName = "tx.LocalDeadlockDetectionTest")
+public class LocalDeadlockDetectionTest extends SingleCacheManagerTest {
+
+ private PerCacheExecutorThread t1;
+ private PerCacheExecutorThread t2;
+ private DeadlockDetectingLockManager lockManager;
+ private Object response1;
+ private Object response2;
+
+ protected CacheManager createCacheManager() throws Exception {
+ cacheManager = TestCacheManagerFactory.createLocalCacheManager();
+ Configuration configuration = new Configuration();
+ configuration.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName());
+ configuration.setEnableDeadlockDetection(true);
+ configuration.setUseLockStriping(false);
+ configuration.setExposeJmxStatistics(true);
+ cacheManager.defineCache("test", configuration);
+ cache = cacheManager.getCache("test");
+ lockManager = (DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache);
+ return cacheManager;
+ }
+
+ @BeforeMethod
+ public void startExecutors() {
+ t1 = new PerCacheExecutorThread(cache, 0);
+ t2 = new PerCacheExecutorThread(cache, 1);
+ lockManager.resetStatistics();
+ }
+
+
+ @AfterMethod
+ public void stopExecutors() {
+ t1.stopThread();
+ t2.stopThread();
+ }
+
+
+ public void testDldPutAndPut() {
+ testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations.PUT_KEY_VALUE,
+ PerCacheExecutorThread.Operations.PUT_KEY_VALUE );
+ if (response1 instanceof Exception) {
+ assertEquals("value_1_t2", cache.get("k1"));
+ assertEquals("value_2_t2", cache.get("k2"));
+ } else {
+ assertEquals("value_1_t1", cache.get("k1"));
+ assertEquals("value_2_t1", cache.get("k2"));
+ }
+ }
+
+ public void testDldPutAndRemove() {
+ testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations.PUT_KEY_VALUE,
+ PerCacheExecutorThread.Operations.REMOVE_KEY );
+ if (response1 instanceof Exception) {
+ assertEquals(cache.get("k1"), null);
+ assertEquals("value_2_t2", cache.get("k2"));
+ } else {
+ assertEquals("value_1_t1", cache.get("k1"));
+ assertEquals(null, cache.get("k2"));
+ }
+ }
+
+ public void testDldRemoveAndPut() {
+ testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations.REMOVE_KEY,
+ PerCacheExecutorThread.Operations.PUT_KEY_VALUE );
+ if (response1 instanceof Exception) {
+ System.out.println("t1 failure");
+ assertEquals(cache.get("k1"), "value_1_t2");
+ assertEquals(cache.get("k2"), null);
+ } else {
+ System.out.println("t2 failure");
+ assertEquals(cache.get("k1"), null);
+ assertEquals(cache.get("k2"), "value_2_t1");
+ }
+ }
+
+ public void testDldRemoveAndRemove() {
+ testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations.REMOVE_KEY,
+ PerCacheExecutorThread.Operations.REMOVE_KEY );
+ if (response1 instanceof Exception) {
+ System.out.println("t1 failure");
+ assertEquals(cache.get("k1"), null);
+ assertEquals(cache.get("k2"), null);
+ } else {
+ System.out.println("t2 failure");
+ assertEquals(cache.get("k1"), null);
+ assertEquals(cache.get("k2"), null);
+ }
+ }
+
+ public void testDldPutAndReplace() {
+
+ cache.put("k1", "initial_1");
+ cache.put("k2", "initial_2");
+
+ testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations.PUT_KEY_VALUE,
+ PerCacheExecutorThread.Operations.REPLACE_KEY_VALUE);
+ if (response1 instanceof Exception) {
+ System.out.println("t1 failure");
+ assertEquals(cache.get("k1"), "value_1_t2");
+ assertEquals(cache.get("k2"), "value_2_t2");
+ } else {
+ System.out.println("t2 failure");
+ assertEquals(cache.get("k1"), "value_1_t1");
+ assertEquals(cache.get("k2"), "value_2_t1");
+ }
+ }
+
+ public void testDldReplaceAndPut() {
+
+ cache.put("k1", "initial_1");
+ cache.put("k2", "initial_2");
+
+ testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations.REPLACE_KEY_VALUE,
+ PerCacheExecutorThread.Operations.PUT_KEY_VALUE);
+ if (response1 instanceof Exception) {
+ System.out.println("t1 failure");
+ assertEquals(cache.get("k1"), "value_1_t2");
+ assertEquals(cache.get("k2"), "value_2_t2");
+ } else {
+ System.out.println("t2 failure");
+ assertEquals(cache.get("k1"), "value_1_t1");
+ assertEquals(cache.get("k2"), "value_2_t1");
+ }
+ }
+
+
+ private void testLocalVsLocalTxDeadlock(PerCacheExecutorThread.Operations firstOperation, PerCacheExecutorThread.Operations secondOperation) {
+
+ assert PerCacheExecutorThread.OperationsResult.BEGGIN_TX_OK == t1.execute(PerCacheExecutorThread.Operations.BEGGIN_TX);
+ assert PerCacheExecutorThread.OperationsResult.BEGGIN_TX_OK == t2.execute(PerCacheExecutorThread.Operations.BEGGIN_TX);
+ System.out.println("After beggin");
+
+ t1.setKeyValue("k1", "value_1_t1");
+ t2.setKeyValue("k2", "value_2_t2");
+
+ assert firstOperation.getCorrespondingOkResult() == t1.execute(firstOperation);
+ assert firstOperation.getCorrespondingOkResult() == t2.execute(firstOperation);
+
+ System.out.println("After first PUT");
+ assert lockManager.isLocked("k1");
+ assert lockManager.isLocked("k2");
+
+
+ t1.setKeyValue("k2", "value_2_t1");
+ t2.setKeyValue("k1", "value_1_t2");
+ t1.executeNoResponse(secondOperation);
+ t2.executeNoResponse(secondOperation);
+
+ response1 = t1.waitForResponse();
+ response2 = t2.waitForResponse();
+
+ assert xor(response1 instanceof DeadlockDetectedException, response2 instanceof DeadlockDetectedException) : "expected one and only one exception: " + response1 + ", " + response2;
+ assert xor(response1 == secondOperation.getCorrespondingOkResult(), response2 == secondOperation.getCorrespondingOkResult()) : "expected one and only one exception: " + response1 + ", " + response2;
+
+ assert lockManager.isLocked("k1");
+ assert lockManager.isLocked("k2");
+ assert lockManager.getOwner("k1") == lockManager.getOwner("k2");
+
+ if (response1 instanceof Exception) {
+ assert PerCacheExecutorThread.OperationsResult.COMMIT_TX_OK == t2.execute(PerCacheExecutorThread.Operations.COMMIT_TX);
+ assert t1.execute(PerCacheExecutorThread.Operations.COMMIT_TX) instanceof RollbackException;
+ } else {
+ assert PerCacheExecutorThread.OperationsResult.COMMIT_TX_OK == t1.execute(PerCacheExecutorThread.Operations.COMMIT_TX);
+ assert t2.execute(PerCacheExecutorThread.Operations.COMMIT_TX) instanceof RollbackException;
+ }
+ assert lockManager.getNumberOfLocksHeld() == 0;
+ assertEquals(lockManager.getDetectedDeadlocks(), 1);
+ }
+
+}
Property changes on: trunk/core/src/test/java/org/infinispan/tx/LocalDeadlockDetectionTest.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Copied: trunk/core/src/test/java/org/infinispan/tx/ReplDeadlockDetectionTest.java (from rev 598, trunk/core/src/test/java/org/infinispan/tx/DeadlockDetectionTest.java)
===================================================================
--- trunk/core/src/test/java/org/infinispan/tx/ReplDeadlockDetectionTest.java (rev 0)
+++ trunk/core/src/test/java/org/infinispan/tx/ReplDeadlockDetectionTest.java 2009-07-21 16:23:23 UTC (rev 601)
@@ -0,0 +1,335 @@
+package org.infinispan.tx;
+
+import org.infinispan.api.mvcc.LockAssert;
+import org.infinispan.commands.ReplicableCommand;
+import org.infinispan.config.Configuration;
+import org.infinispan.config.ConfigurationException;
+import org.infinispan.context.impl.NonTxInvocationContext;
+import org.infinispan.interceptors.DeadlockDetectingInterceptor;
+import org.infinispan.interceptors.InterceptorChain;
+import org.infinispan.manager.CacheManager;
+import org.infinispan.remoting.ReplicationException;
+import org.infinispan.remoting.responses.Response;
+import org.infinispan.remoting.rpc.ResponseFilter;
+import org.infinispan.remoting.rpc.ResponseMode;
+import org.infinispan.remoting.rpc.RpcManager;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.remoting.transport.Transport;
+import org.infinispan.statetransfer.StateTransferException;
+import org.infinispan.test.MultipleCacheManagersTest;
+import org.infinispan.test.TestingUtil;
+import org.infinispan.test.PerCacheExecutorThread;
+import org.infinispan.test.fwk.TestCacheManagerFactory;
+import org.infinispan.transaction.lookup.DummyTransactionManagerLookup;
+import org.infinispan.util.concurrent.NotifyingNotifiableFuture;
+import org.infinispan.util.concurrent.locks.DeadlockDetectingLockManager;
+import org.infinispan.util.concurrent.locks.LockManager;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * Functional test for deadlock detection.
+ *
+ * @author Mircea.Markus at jboss.com
+ *
+ */
+ at Test(testName = "tx.ReplDeadlockDetectionTest", groups = "functional")
+public class ReplDeadlockDetectionTest extends MultipleCacheManagersTest {
+
+ protected ControlledRpcManager controlledRpcManager1;
+ protected ControlledRpcManager controlledRpcManager2;
+ protected CountDownLatch replicationLatch;
+ protected PerCacheExecutorThread t1;
+ protected PerCacheExecutorThread t2;
+ protected DeadlockDetectingLockManager ddLm1;
+ protected DeadlockDetectingLockManager ddLm2;
+
+ protected Configuration.CacheMode cacheMode = Configuration.CacheMode.REPL_SYNC;
+
+ protected void createCacheManagers() throws Throwable {
+ Configuration config = getDefaultClusteredConfig(cacheMode);
+ config.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName());
+ config.setEnableDeadlockDetection(true);
+ config.setSyncCommitPhase(true);
+ config.setSyncRollbackPhase(true);
+ config.setUseLockStriping(false);
+ assert config.isEnableDeadlockDetection();
+ createClusteredCaches(2, "test", config);
+ assert config.isEnableDeadlockDetection();
+
+ assert cache(0, "test").getConfiguration().isEnableDeadlockDetection();
+ assert cache(1, "test").getConfiguration().isEnableDeadlockDetection();
+ assert !cache(0, "test").getConfiguration().isExposeJmxStatistics();
+ assert !cache(1, "test").getConfiguration().isExposeJmxStatistics();
+
+ ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(0, "test"))).setExposeJmxStats(true);
+ ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(1, "test"))).setExposeJmxStats(true);
+
+ RpcManager rpcManager1 = TestingUtil.extractComponent(cache(0, "test"), RpcManager.class);
+ RpcManager rpcManager2 = TestingUtil.extractComponent(cache(1, "test"), RpcManager.class);
+
+ controlledRpcManager1 = new ControlledRpcManager(rpcManager1);
+ controlledRpcManager2 = new ControlledRpcManager(rpcManager2);
+ TestingUtil.replaceComponent(cache(0, "test"), RpcManager.class, controlledRpcManager1, true);
+ TestingUtil.replaceComponent(cache(1, "test"), RpcManager.class, controlledRpcManager2, true);
+
+ assert TestingUtil.extractComponent(cache(0, "test"), RpcManager.class) instanceof ControlledRpcManager;
+ assert TestingUtil.extractComponent(cache(1, "test"), RpcManager.class) instanceof ControlledRpcManager;
+
+ ddLm1 = (DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(0, "test"));
+ ddLm2 = (DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(1, "test"));
+ }
+
+
+ @BeforeMethod
+ public void beforeMethod() {
+ t1 = new PerCacheExecutorThread(cache(0, "test"), 1);
+ t2 = new PerCacheExecutorThread(cache(1, "test"), 2);
+ replicationLatch = new CountDownLatch(1);
+ controlledRpcManager1.setReplicationLatch(replicationLatch);
+ controlledRpcManager2.setReplicationLatch(replicationLatch);
+ log.trace("_________________________ Here is beggins");
+ }
+
+ @AfterMethod
+ public void afterMethod() {
+ t1.stopThread();
+ t2.stopThread();
+ ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(0, "test"))).resetStatistics();
+ ((DeadlockDetectingLockManager) TestingUtil.extractLockManager(cache(1, "test"))).resetStatistics();
+ }
+
+ public void testDeadlockDetectionAndAsyncCaches() {
+ Configuration config = getDefaultClusteredConfig(Configuration.CacheMode.REPL_ASYNC);
+ config.setEnableDeadlockDetection(true);
+ config.setUseLockStriping(false);
+ CacheManager cm = TestCacheManagerFactory.createClusteredCacheManager();
+ cm.defineCache("test", config);
+ try {
+ cm.getCache("test");
+ assert false : "Exception expected";
+ } catch (ConfigurationException e) {
+ //expected
+ System.out.println("Error message is " + e.getMessage());
+ }
+ cm.stop();
+ }
+
+ public void testExpectedInnerStructure() {
+ LockManager lockManager = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
+ assert lockManager instanceof DeadlockDetectingLockManager;
+
+ InterceptorChain ic = TestingUtil.extractComponent(cache(0, "test"), InterceptorChain.class);
+ assert ic.containsInterceptorType(DeadlockDetectingInterceptor.class);
+ }
+
+ public void testDeadlockDetectedTwoTransactions() throws Exception {
+ t1.setKeyValue("key", "value1");
+ t2.setKeyValue("key", "value2");
+ assert PerCacheExecutorThread.OperationsResult.BEGGIN_TX_OK == t1.execute(PerCacheExecutorThread.Operations.BEGGIN_TX);
+ assert PerCacheExecutorThread.OperationsResult.BEGGIN_TX_OK == t2.execute(PerCacheExecutorThread.Operations.BEGGIN_TX);
+ System.out.println("After beggin");
+
+ t1.execute(PerCacheExecutorThread.Operations.PUT_KEY_VALUE);
+ t2.execute(PerCacheExecutorThread.Operations.PUT_KEY_VALUE);
+ System.out.println("After put key value");
+
+ t1.clearResponse();
+ t2.clearResponse();
+
+ t1.executeNoResponse(PerCacheExecutorThread.Operations.COMMIT_TX);
+ t2.executeNoResponse(PerCacheExecutorThread.Operations.COMMIT_TX);
+
+ System.out.println("Now replication is triggered");
+ replicationLatch.countDown();
+
+
+ Object t1Commit = t1.waitForResponse();
+ Object t2Commit = t2.waitForResponse();
+ System.out.println("After commit: " + t1Commit + ", " + t2Commit);
+
+ assert xor(t1Commit instanceof Exception, t2Commit instanceof Exception) : "only one thread must be failing " + t1Commit + "," + t2Commit;
+ System.out.println("t2Commit = " + t2Commit);
+ System.out.println("t1Commit = " + t1Commit);
+
+ if (t1Commit instanceof Exception) {
+ System.out.println("t1 rolled back");
+ Object o = cache(0, "test").get("key");
+ assert o != null;
+ assert o.equals("value2");
+ } else {
+ System.out.println("t2 rolled back");
+ Object o = cache(0, "test").get("key");
+ assert o != null;
+ assert o.equals("value1");
+ o = cache(1, "test").get("key");
+ assert o != null;
+ assert o.equals("value1");
+ }
+
+ assert ddLm1.getDetectedDeadlocks() + ddLm2.getDetectedDeadlocks() >= 1;
+
+ LockManager lm1 = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
+ assert !lm1.isLocked("key") : "It is locked by " + lm1.getOwner("key");
+ LockManager lm2 = TestingUtil.extractComponent(cache(1, "test"), LockManager.class);
+ assert !lm2.isLocked("key") : "It is locked by " + lm2.getOwner("key");
+ LockAssert.assertNoLocks(cache(0, "test"));
+ }
+
+ public void testDeadlockDetectedOneTx() throws Exception {
+ t1.setKeyValue("key", "value1");
+
+ LockManager lm2 = TestingUtil.extractComponent(cache(1, "test"), LockManager.class);
+ NonTxInvocationContext ctx = cache(1, "test").getAdvancedCache().getInvocationContextContainer().createNonTxInvocationContext();
+ lm2.lockAndRecord("key", ctx);
+ assert lm2.isLocked("key");
+
+
+ assert PerCacheExecutorThread.OperationsResult.BEGGIN_TX_OK == t1.execute(PerCacheExecutorThread.Operations.BEGGIN_TX) : "but received " + t1.lastResponse();
+ t1.execute(PerCacheExecutorThread.Operations.PUT_KEY_VALUE);
+
+ t1.clearResponse();
+ t1.executeNoResponse(PerCacheExecutorThread.Operations.COMMIT_TX);
+
+ replicationLatch.countDown();
+ System.out.println("Now replication is triggered");
+
+ t1.waitForResponse();
+
+
+ Object t1CommitRsp = t1.lastResponse();
+
+ assert t1CommitRsp instanceof Exception : "expected exception, received " + t1.lastResponse();
+
+ LockManager lm1 = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
+ assert !lm1.isLocked("key") : "It is locked by " + lm1.getOwner("key");
+
+ lm2.unlock("key", ctx.getLockOwner());
+ assert !lm2.isLocked("key");
+ assert !lm1.isLocked("key");
+ }
+
+ public void testLockReleasedWhileTryingToAcquire() throws Exception {
+ t1.setKeyValue("key", "value1");
+
+ LockManager lm2 = TestingUtil.extractComponent(cache(1, "test"), LockManager.class);
+ NonTxInvocationContext ctx = cache(1, "test").getAdvancedCache().getInvocationContextContainer().createNonTxInvocationContext();
+ lm2.lockAndRecord("key", ctx);
+ assert lm2.isLocked("key");
+
+
+ assert PerCacheExecutorThread.OperationsResult.BEGGIN_TX_OK == t1.execute(PerCacheExecutorThread.Operations.BEGGIN_TX) : "but received " + t1.lastResponse();
+ t1.execute(PerCacheExecutorThread.Operations.PUT_KEY_VALUE);
+
+ t1.clearResponse();
+ t1.executeNoResponse(PerCacheExecutorThread.Operations.COMMIT_TX);
+
+ replicationLatch.countDown();
+
+ Thread.sleep(3000); //just to make sure the remote tx thread managed to spin around for some times.
+ lm2.unlock("key", ctx.getLockOwner());
+
+ t1.waitForResponse();
+
+
+ Object t1CommitRsp = t1.lastResponse();
+
+ assert t1CommitRsp == PerCacheExecutorThread.OperationsResult.COMMIT_TX_OK : "expected true, received " + t1.lastResponse();
+
+ LockManager lm1 = TestingUtil.extractComponent(cache(0, "test"), LockManager.class);
+ assert !lm1.isLocked("key") : "It is locked by " + lm1.getOwner("key");
+
+ assert !lm2.isLocked("key");
+ assert !lm1.isLocked("key");
+ }
+
+ public static final class ControlledRpcManager implements RpcManager {
+
+ private volatile CountDownLatch replicationLatch;
+
+ public ControlledRpcManager(RpcManager realOne) {
+ this.realOne = realOne;
+ }
+
+ private RpcManager realOne;
+
+ public void setReplicationLatch(CountDownLatch replicationLatch) {
+ this.replicationLatch = replicationLatch;
+ }
+
+ public List<Response> invokeRemotely(List<Address> recipients, ReplicableCommand rpcCommand, ResponseMode mode, long timeout, boolean usePriorityQueue, ResponseFilter responseFilter) {
+ return realOne.invokeRemotely(recipients, rpcCommand, mode, timeout, usePriorityQueue, responseFilter);
+ }
+
+ public List<Response> invokeRemotely(List<Address> recipients, ReplicableCommand rpcCommand, ResponseMode mode, long timeout, boolean usePriorityQueue) {
+ return realOne.invokeRemotely(recipients, rpcCommand, mode, timeout, usePriorityQueue);
+ }
+
+ public List<Response> invokeRemotely(List<Address> recipients, ReplicableCommand rpcCommand, ResponseMode mode, long timeout) throws Exception {
+ return realOne.invokeRemotely(recipients, rpcCommand, mode, timeout);
+ }
+
+ public void retrieveState(String cacheName, long timeout) throws StateTransferException {
+ realOne.retrieveState(cacheName, timeout);
+ }
+
+ public void broadcastRpcCommand(ReplicableCommand rpc, boolean sync) throws ReplicationException {
+ waitFirst();
+ realOne.broadcastRpcCommand(rpc, sync);
+ }
+
+ public void broadcastRpcCommand(ReplicableCommand rpc, boolean sync, boolean usePriorityQueue) throws ReplicationException {
+ waitFirst();
+ realOne.broadcastRpcCommand(rpc, sync, usePriorityQueue);
+ }
+
+ private void waitFirst() {
+ System.out.println(Thread.currentThread().getName() + " -- replication trigger called!");
+ try {
+ replicationLatch.await();
+ } catch (Exception e) {
+ throw new RuntimeException("Unexpected exception!", e);
+ }
+ }
+
+ public void broadcastRpcCommandInFuture(ReplicableCommand rpc, NotifyingNotifiableFuture<Object> future) {
+ realOne.broadcastRpcCommandInFuture(rpc, future);
+ }
+
+ public void broadcastRpcCommandInFuture(ReplicableCommand rpc, boolean usePriorityQueue, NotifyingNotifiableFuture<Object> future) {
+ realOne.broadcastRpcCommandInFuture(rpc, usePriorityQueue, future);
+ }
+
+ public void invokeRemotely(List<Address> recipients, ReplicableCommand rpc, boolean sync) throws ReplicationException {
+ realOne.invokeRemotely(recipients, rpc, sync);
+ }
+
+ public void invokeRemotely(List<Address> recipients, ReplicableCommand rpc, boolean sync, boolean usePriorityQueue) throws ReplicationException {
+ realOne.invokeRemotely(recipients, rpc, sync, usePriorityQueue);
+ }
+
+ public void invokeRemotelyInFuture(List<Address> recipients, ReplicableCommand rpc, NotifyingNotifiableFuture<Object> future) {
+ realOne.invokeRemotelyInFuture(recipients, rpc, future);
+ }
+
+ public void invokeRemotelyInFuture(List<Address> recipients, ReplicableCommand rpc, boolean usePriorityQueue, NotifyingNotifiableFuture<Object> future) {
+ realOne.invokeRemotelyInFuture(recipients, rpc, usePriorityQueue, future);
+ }
+
+ public void invokeRemotelyInFuture(List<Address> recipients, ReplicableCommand rpc, boolean usePriorityQueue, NotifyingNotifiableFuture<Object> future, long timeout) {
+ realOne.invokeRemotelyInFuture(recipients, rpc, usePriorityQueue, future, timeout);
+ }
+
+ public Transport getTransport() {
+ return realOne.getTransport();
+ }
+
+ public Address getCurrentStateTransferSource() {
+ return realOne.getCurrentStateTransferSource();
+ }
+ }
+}
Property changes on: trunk/core/src/test/java/org/infinispan/tx/ReplDeadlockDetectionTest.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
More information about the infinispan-commits
mailing list