From fd7d9b244744df4c2c1460c8c24431532a9a45b3 Mon Sep 17 00:00:00 2001 From: Tom Pantelis Date: Fri, 3 Apr 2015 05:39:56 -0400 Subject: [PATCH] Remove recorded modification Futures from TransactionContext This removes the recorded batched modification message Futures from TransactionContext to avoid the overhead of waiting for the Futures to complete on ready and transforming them. The Futures were recorded to ensure all modification messages were successfully processed before committing the transaction but we can do that by including the count of modification messages sent in the ready message for the transaction actor to verify (later patch). Change-Id: I8d6ad80cac3e8e13fde48fbf5c57a476cca003dd Signed-off-by: Tom Pantelis --- .../datastore/AbstractTransactionContext.java | 25 +----- .../cluster/datastore/TransactionContext.java | 3 - .../datastore/TransactionContextImpl.java | 60 ++++----------- .../cluster/datastore/TransactionProxy.java | 13 ---- .../WriteOnlyTransactionContextImpl.java | 5 +- .../PreLithiumTransactionContextImpl.java | 14 ++-- .../datastore/TransactionProxyTest.java | 77 ------------------- 7 files changed, 23 insertions(+), 174 deletions(-) diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContext.java index d94e1c691e..81605d8c8f 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContext.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContext.java @@ -7,40 +7,17 @@ */ package org.opendaylight.controller.cluster.datastore; -import com.google.common.collect.ImmutableList; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier; -import scala.concurrent.Future; abstract class AbstractTransactionContext implements TransactionContext { - private final List> recordedOperationFutures = new ArrayList<>(); private final TransactionIdentifier identifier; protected AbstractTransactionContext(TransactionIdentifier identifier) { this.identifier = identifier; } - @Override - public final void copyRecordedOperationFutures(Collection> target) { - target.addAll(recordedOperationFutures); - } - protected final TransactionIdentifier getIdentifier() { return identifier; } - - protected final Collection> copyRecordedOperationFutures() { - return ImmutableList.copyOf(recordedOperationFutures); - } - - protected final int recordedOperationCount() { - return recordedOperationFutures.size(); - } - - protected final void recordOperationFuture(Future future) { - recordedOperationFutures.add(future); - } -} +} \ No newline at end of file diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContext.java index a5a7494e1a..bc6e5f229f 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContext.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContext.java @@ -10,7 +10,6 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; import com.google.common.base.Optional; import com.google.common.util.concurrent.SettableFuture; -import java.util.Collection; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import scala.concurrent.Future; @@ -33,6 +32,4 @@ interface TransactionContext { void readData(final YangInstanceIdentifier path, SettableFuture>> proxyFuture); void dataExists(YangInstanceIdentifier path, SettableFuture proxyFuture); - - void copyRecordedOperationFutures(Collection> target); } diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContextImpl.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContextImpl.java index c61682d8ef..b9900889b1 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContextImpl.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionContextImpl.java @@ -11,9 +11,7 @@ import akka.actor.ActorSelection; import akka.dispatch.Mapper; import akka.dispatch.OnComplete; import com.google.common.base.Optional; -import com.google.common.collect.Lists; import com.google.common.util.concurrent.SettableFuture; -import java.util.List; import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; @@ -93,51 +91,30 @@ public class TransactionContextImpl extends AbstractTransactionContext { @Override public Future readyTransaction() { - LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending", - getIdentifier(), recordedOperationCount()); + LOG.debug("Tx {} readyTransaction called", getIdentifier()); // Send the remaining batched modifications if any. - sendAndRecordBatchedModifications(); + sendBatchedModifications(); // Send the ReadyTransaction message to the Tx actor. Future readyReplyFuture = executeOperationAsync(ReadyTransaction.INSTANCE); - return combineRecordedOperationsFutures(readyReplyFuture); + return transformReadyReply(readyReplyFuture); } - protected Future combineRecordedOperationsFutures(final Future withLastReplyFuture) { - // Combine all the previously recorded put/merge/delete operation reply Futures and the - // ReadyTransactionReply Future into one Future. If any one fails then the combined - // Future will fail. We need all prior operations and the ready operation to succeed - // in order to attempt commit. + protected Future transformReadyReply(final Future readyReplyFuture) { + // Transform the last reply Future into a Future that returns the cohort actor path from + // the last reply message. That's the end result of the ready operation. - List> futureList = Lists.newArrayListWithCapacity(recordedOperationCount() + 1); - copyRecordedOperationFutures(futureList); - futureList.add(withLastReplyFuture); - - Future> combinedFutures = akka.dispatch.Futures.sequence(futureList, - actorContext.getClientDispatcher()); - - // Transform the combined Future into a Future that returns the cohort actor path from - // the ReadyTransactionReply. That's the end result of the ready operation. - - return combinedFutures.transform(new Mapper, ActorSelection>() { + return readyReplyFuture.transform(new Mapper() { @Override - public ActorSelection checkedApply(Iterable notUsed) { - LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded", - getIdentifier()); - - // At this point all the Futures succeeded and we need to extract the cohort - // actor path from the ReadyTransactionReply. For the recorded operations, they - // don't return any data so we're only interested that they completed - // successfully. We could be paranoid and verify the correct reply types but - // that really should never happen so it's not worth the overhead of - // de-serializing each reply. - - // Note the Future get call here won't block as it's complete. - Object serializedReadyReply = withLastReplyFuture.value().get().get(); + public ActorSelection checkedApply(Object serializedReadyReply) { + LOG.debug("Tx {} readyTransaction", getIdentifier()); + + // At this point the rwady operation succeeded and we need to extract the cohort + // actor path from the reply. if (serializedReadyReply instanceof ReadyTransactionReply) { return actorContext.actorSelection(((ReadyTransactionReply)serializedReadyReply).getCohortPath()); } else if(serializedReadyReply instanceof BatchedModificationsReply) { @@ -169,14 +146,7 @@ public class TransactionContextImpl extends AbstractTransactionContext { if(batchedModifications.getModifications().size() >= actorContext.getDatastoreContext().getShardBatchedModificationCount()) { - sendAndRecordBatchedModifications(); - } - } - - private void sendAndRecordBatchedModifications() { - Future sentFuture = sendBatchedModifications(); - if(sentFuture != null) { - recordOperationFuture(sentFuture); + sendBatchedModifications(); } } @@ -232,7 +202,7 @@ public class TransactionContextImpl extends AbstractTransactionContext { // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the // public API contract. - sendAndRecordBatchedModifications(); + sendBatchedModifications(); OnComplete onComplete = new OnComplete() { @Override @@ -274,7 +244,7 @@ public class TransactionContextImpl extends AbstractTransactionContext { // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the // public API contract. - sendAndRecordBatchedModifications(); + sendBatchedModifications(); OnComplete onComplete = new OnComplete() { @Override diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java index 5f9cc83618..0fd37b9ece 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java @@ -158,19 +158,6 @@ public class TransactionProxy extends AbstractDOMStoreTransaction> getRecordedOperationFutures() { - List> recordedOperationFutures = Lists.newArrayList(); - for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) { - TransactionContext transactionContext = txFutureCallback.getTransactionContext(); - if (transactionContext != null) { - transactionContext.copyRecordedOperationFutures(recordedOperationFutures); - } - } - - return recordedOperationFutures; - } - @VisibleForTesting boolean hasTransactionContext() { for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) { diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/WriteOnlyTransactionContextImpl.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/WriteOnlyTransactionContextImpl.java index e1313540c4..b9fe90dc13 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/WriteOnlyTransactionContextImpl.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/WriteOnlyTransactionContextImpl.java @@ -32,13 +32,12 @@ public class WriteOnlyTransactionContextImpl extends TransactionContextImpl { @Override public Future readyTransaction() { - LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending", - getIdentifier(), recordedOperationCount()); + LOG.debug("Tx {} readyTransaction called", getIdentifier()); // Send the remaining batched modifications if any. Future lastModificationsFuture = sendBatchedModifications(true); - return combineRecordedOperationsFutures(lastModificationsFuture); + return transformReadyReply(lastModificationsFuture); } } diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/compat/PreLithiumTransactionContextImpl.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/compat/PreLithiumTransactionContextImpl.java index c3450333a4..9509b06ba7 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/compat/PreLithiumTransactionContextImpl.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/compat/PreLithiumTransactionContextImpl.java @@ -45,32 +45,28 @@ public class PreLithiumTransactionContextImpl extends TransactionContextImpl { @Override public void deleteData(YangInstanceIdentifier path) { - recordOperationFuture(executeOperationAsync( - new DeleteData(path, getRemoteTransactionVersion()))); + executeOperationAsync(new DeleteData(path, getRemoteTransactionVersion())); } @Override public void mergeData(YangInstanceIdentifier path, NormalizedNode data) { - recordOperationFuture(executeOperationAsync( - new MergeData(path, data, getRemoteTransactionVersion()))); + executeOperationAsync(new MergeData(path, data, getRemoteTransactionVersion())); } @Override public void writeData(YangInstanceIdentifier path, NormalizedNode data) { - recordOperationFuture(executeOperationAsync( - new WriteData(path, data, getRemoteTransactionVersion()))); + executeOperationAsync(new WriteData(path, data, getRemoteTransactionVersion())); } @Override public Future readyTransaction() { - LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending", - getIdentifier(), recordedOperationCount()); + LOG.debug("Tx {} readyTransaction called", getIdentifier()); // Send the ReadyTransaction message to the Tx actor. Future lastReplyFuture = executeOperationAsync(ReadyTransaction.INSTANCE); - return combineRecordedOperationsFutures(lastReplyFuture); + return transformReadyReply(lastReplyFuture); } @Override diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java index 29a5b09c5c..b95eaf64d7 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java @@ -3,7 +3,6 @@ package org.opendaylight.controller.cluster.datastore; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; @@ -41,7 +40,6 @@ import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedEx import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; -import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; @@ -61,10 +59,7 @@ import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import scala.concurrent.Await; -import scala.concurrent.Future; import scala.concurrent.Promise; -import scala.concurrent.duration.Duration; @SuppressWarnings("resource") public class TransactionProxyTest extends AbstractTransactionProxyTest { @@ -313,29 +308,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { transactionProxy.exists(TestModel.TEST_PATH); } - private void verifyRecordingOperationFutures(List> futures, - Class... expResultTypes) throws Exception { - assertEquals("getRecordingOperationFutures size", expResultTypes.length, futures.size()); - - int i = 0; - for( Future future: futures) { - assertNotNull("Recording operation Future is null", future); - - Class expResultType = expResultTypes[i++]; - if(Throwable.class.isAssignableFrom(expResultType)) { - try { - Await.result(future, Duration.create(5, TimeUnit.SECONDS)); - fail("Expected exception from recording operation Future"); - } catch(Exception e) { - // Expected - } - } else { - assertEquals(String.format("Recording operation %d Future result type", i +1 ), expResultType, - Await.result(future, Duration.create(5, TimeUnit.SECONDS)).getClass()); - } - } - } - @Test public void testWrite() throws Exception { dataStoreContextBuilder.shardBatchedModificationCount(1); @@ -405,9 +377,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { transactionProxy.ready(); verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); } @Test(expected=IllegalStateException.class) @@ -479,9 +448,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), @@ -511,8 +477,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures()); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); List batchedModifications = captureBatchedModifications(actorRef); @@ -544,9 +508,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); List batchedModifications = captureBatchedModifications(actorRef); @@ -561,33 +522,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { isA(ReadyTransaction.SERIALIZABLE_CLASS)); } - @Test - public void testReadyWithRecordingOperationFailure() throws Exception { - dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true); - - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - - expectFailedBatchedModifications(actorRef); - - doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); - - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - - DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - - verifyCohortFutures(proxy, TestException.class); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), TestException.class); - } - @Test public void testReadyWithReplyFailure() throws Exception { dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); @@ -1245,14 +1179,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { boolean optimizedWriteOnly = type == WRITE_ONLY && dataStoreContextBuilder.build().isWriteOnlyTransactionOptimizationsEnabled(); verifyBatchedModifications(batchedModifications.get(2), optimizedWriteOnly, new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2)); - - if(optimizedWriteOnly) { - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class); - } else { - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); - } } @Test @@ -1357,9 +1283,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { inOrder.verify(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); } @Test -- 2.36.6