X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionContextImpl.java;h=c61682d8efe98cf1649bebc03b381b6afeeb1d76;hb=e2d9f9c57e124d46e117f17c44b77c89222fdb99;hp=c1f9c78e69ec683586147e01605ab7168f786e1b;hpb=874a18a9ce5dc09bc49922754bf8fb3e981fffb9;p=controller.git 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 c1f9c78e69..c61682d8ef 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 @@ -16,6 +16,7 @@ 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; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; import org.opendaylight.controller.cluster.datastore.messages.DataExists; import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply; @@ -40,8 +41,8 @@ import scala.concurrent.Future; public class TransactionContextImpl extends AbstractTransactionContext { private static final Logger LOG = LoggerFactory.getLogger(TransactionContextImpl.class); + private final String transactionChainId; private final ActorContext actorContext; - private final String transactionPath; private final ActorSelection actor; private final boolean isTxActorLocal; private final short remoteTransactionVersion; @@ -49,12 +50,12 @@ public class TransactionContextImpl extends AbstractTransactionContext { private final OperationCompleter operationCompleter; private BatchedModifications batchedModifications; - protected TransactionContextImpl(String transactionPath, ActorSelection actor, TransactionIdentifier identifier, - ActorContext actorContext, SchemaContext schemaContext, - boolean isTxActorLocal, short remoteTransactionVersion, OperationCompleter operationCompleter) { + protected TransactionContextImpl(ActorSelection actor, TransactionIdentifier identifier, + String transactionChainId, ActorContext actorContext, SchemaContext schemaContext, boolean isTxActorLocal, + short remoteTransactionVersion, OperationCompleter operationCompleter) { super(identifier); - this.transactionPath = transactionPath; this.actor = actor; + this.transactionChainId = transactionChainId; this.actorContext = actorContext; this.isTxActorLocal = isTxActorLocal; this.remoteTransactionVersion = remoteTransactionVersion; @@ -71,6 +72,10 @@ public class TransactionContextImpl extends AbstractTransactionContext { return actor; } + protected ActorContext getActorContext() { + return actorContext; + } + protected short getRemoteTransactionVersion() { return remoteTransactionVersion; } @@ -81,7 +86,7 @@ public class TransactionContextImpl extends AbstractTransactionContext { @Override public void closeTransaction() { - LOG.debug("Tx {} closeTransaction called", identifier); + LOG.debug("Tx {} closeTransaction called", getIdentifier()); actorContext.sendOperationAsync(getActor(), CloseTransaction.INSTANCE.toSerializable()); } @@ -89,25 +94,28 @@ public class TransactionContextImpl extends AbstractTransactionContext { @Override public Future readyTransaction() { LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending", - identifier, recordedOperationFutures.size()); + getIdentifier(), recordedOperationCount()); // Send the remaining batched modifications if any. - sendBatchedModifications(); + sendAndRecordBatchedModifications(); // Send the ReadyTransaction message to the Tx actor. - final Future replyFuture = executeOperationAsync(ReadyTransaction.INSTANCE); + Future readyReplyFuture = executeOperationAsync(ReadyTransaction.INSTANCE); + return combineRecordedOperationsFutures(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. - List> futureList = - Lists.newArrayListWithCapacity(recordedOperationFutures.size() + 1); - futureList.addAll(recordedOperationFutures); - futureList.add(replyFuture); + List> futureList = Lists.newArrayListWithCapacity(recordedOperationCount() + 1); + copyRecordedOperationFutures(futureList); + futureList.add(withLastReplyFuture); Future> combinedFutures = akka.dispatch.Futures.sequence(futureList, actorContext.getClientDispatcher()); @@ -119,7 +127,7 @@ public class TransactionContextImpl extends AbstractTransactionContext { @Override public ActorSelection checkedApply(Iterable notUsed) { LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded", - identifier); + 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 @@ -129,145 +137,113 @@ public class TransactionContextImpl extends AbstractTransactionContext { // de-serializing each reply. // Note the Future get call here won't block as it's complete. - Object serializedReadyReply = replyFuture.value().get().get(); + Object serializedReadyReply = withLastReplyFuture.value().get().get(); if (serializedReadyReply instanceof ReadyTransactionReply) { return actorContext.actorSelection(((ReadyTransactionReply)serializedReadyReply).getCohortPath()); - + } else if(serializedReadyReply instanceof BatchedModificationsReply) { + return actorContext.actorSelection(((BatchedModificationsReply)serializedReadyReply).getCohortPath()); } else if(serializedReadyReply.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)) { ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(serializedReadyReply); - String cohortPath = reply.getCohortPath(); - - // In Helium we used to return the local path of the actor which represented - // a remote ThreePhaseCommitCohort. The local path would then be converted to - // a remote path using this resolvePath method. To maintain compatibility with - // a Helium node we need to continue to do this conversion. - // At some point in the future when upgrades from Helium are not supported - // we could remove this code to resolvePath and just use the cohortPath as the - // resolved cohortPath - if(TransactionContextImpl.this.remoteTransactionVersion < - DataStoreVersions.HELIUM_1_VERSION) { - cohortPath = actorContext.resolvePath(transactionPath, cohortPath); - } - + String cohortPath = deserializeCohortPath(reply.getCohortPath()); return actorContext.actorSelection(cohortPath); - } else { // Throwing an exception here will fail the Future. throw new IllegalArgumentException(String.format("%s: Invalid reply type %s", - identifier, serializedReadyReply.getClass())); + getIdentifier(), serializedReadyReply.getClass())); } } }, TransactionProxy.SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher()); } + protected String deserializeCohortPath(String cohortPath) { + return cohortPath; + } + private void batchModification(Modification modification) { if(batchedModifications == null) { - batchedModifications = new BatchedModifications(remoteTransactionVersion); + batchedModifications = new BatchedModifications(getIdentifier().toString(), remoteTransactionVersion, + transactionChainId); } batchedModifications.addModification(modification); if(batchedModifications.getModifications().size() >= actorContext.getDatastoreContext().getShardBatchedModificationCount()) { - sendBatchedModifications(); + sendAndRecordBatchedModifications(); } } - private void sendBatchedModifications() { + private void sendAndRecordBatchedModifications() { + Future sentFuture = sendBatchedModifications(); + if(sentFuture != null) { + recordOperationFuture(sentFuture); + } + } + + protected Future sendBatchedModifications() { + return sendBatchedModifications(false); + } + + protected Future sendBatchedModifications(boolean ready) { + Future sent = null; if(batchedModifications != null) { - LOG.debug("Tx {} sending {} batched modifications", identifier, - batchedModifications.getModifications().size()); + if(LOG.isDebugEnabled()) { + LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(), + batchedModifications.getModifications().size(), ready); + } + + batchedModifications.setReady(ready); + sent = executeOperationAsync(batchedModifications); - recordedOperationFutures.add(executeOperationAsync(batchedModifications)); - batchedModifications = null; + batchedModifications = new BatchedModifications(getIdentifier().toString(), remoteTransactionVersion, + transactionChainId); } + + return sent; } @Override public void deleteData(YangInstanceIdentifier path) { - LOG.debug("Tx {} deleteData called path = {}", identifier, path); + LOG.debug("Tx {} deleteData called path = {}", getIdentifier(), path); batchModification(new DeleteModification(path)); } @Override public void mergeData(YangInstanceIdentifier path, NormalizedNode data) { - LOG.debug("Tx {} mergeData called path = {}", identifier, path); + LOG.debug("Tx {} mergeData called path = {}", getIdentifier(), path); batchModification(new MergeModification(path, data)); } @Override public void writeData(YangInstanceIdentifier path, NormalizedNode data) { - LOG.debug("Tx {} writeData called path = {}", identifier, path); + LOG.debug("Tx {} writeData called path = {}", getIdentifier(), path); batchModification(new WriteModification(path, data)); } @Override - public void readData( - final YangInstanceIdentifier path,final SettableFuture>> returnFuture ) { + public void readData(final YangInstanceIdentifier path, + final SettableFuture>> returnFuture ) { - LOG.debug("Tx {} readData called path = {}", identifier, path); + LOG.debug("Tx {} readData called path = {}", getIdentifier(), path); - // Send the remaining batched modifications if any. + // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the + // public API contract. - sendBatchedModifications(); - - // If there were any previous recorded put/merge/delete operation reply Futures then we - // must wait for them to successfully complete. This is necessary to honor the read - // uncommitted semantics of the public API contract. If any one fails then fail the read. - - if(recordedOperationFutures.isEmpty()) { - finishReadData(path, returnFuture); - } else { - LOG.debug("Tx {} readData: verifying {} previous recorded operations", - identifier, recordedOperationFutures.size()); - - // Note: we make a copy of recordedOperationFutures to be on the safe side in case - // Futures#sequence accesses the passed List on a different thread, as - // recordedOperationFutures is not synchronized. - - Future> combinedFutures = akka.dispatch.Futures.sequence( - Lists.newArrayList(recordedOperationFutures), - actorContext.getClientDispatcher()); - - OnComplete> onComplete = new OnComplete>() { - @Override - public void onComplete(Throwable failure, Iterable notUsed) - throws Throwable { - if(failure != null) { - LOG.debug("Tx {} readData: a recorded operation failed: {}", - identifier, failure); - returnFuture.setException(new ReadFailedException( - "The read could not be performed because a previous put, merge," - + "or delete operation failed", failure)); - } else { - finishReadData(path, returnFuture); - } - } - }; - - combinedFutures.onComplete(onComplete, actorContext.getClientDispatcher()); - } - - } - - private void finishReadData(final YangInstanceIdentifier path, - final SettableFuture>> returnFuture) { - - LOG.debug("Tx {} finishReadData called path = {}", identifier, path); + sendAndRecordBatchedModifications(); OnComplete onComplete = new OnComplete() { @Override public void onComplete(Throwable failure, Object readResponse) throws Throwable { if(failure != null) { - LOG.debug("Tx {} read operation failed: {}", identifier, failure); + LOG.debug("Tx {} read operation failed: {}", getIdentifier(), failure); returnFuture.setException(new ReadFailedException( "Error reading data for path " + path, failure)); } else { - LOG.debug("Tx {} read operation succeeded", identifier, failure); + LOG.debug("Tx {} read operation succeeded", getIdentifier(), failure); if (readResponse instanceof ReadDataReply) { ReadDataReply reply = (ReadDataReply) readResponse; @@ -293,64 +269,22 @@ public class TransactionContextImpl extends AbstractTransactionContext { @Override public void dataExists(final YangInstanceIdentifier path, final SettableFuture returnFuture) { - LOG.debug("Tx {} dataExists called path = {}", identifier, path); - - // Send the remaining batched modifications if any. - - sendBatchedModifications(); - - // If there were any previous recorded put/merge/delete operation reply Futures then we - // must wait for them to successfully complete. This is necessary to honor the read - // uncommitted semantics of the public API contract. If any one fails then fail this - // request. - - if(recordedOperationFutures.isEmpty()) { - finishDataExists(path, returnFuture); - } else { - LOG.debug("Tx {} dataExists: verifying {} previous recorded operations", - identifier, recordedOperationFutures.size()); - - // Note: we make a copy of recordedOperationFutures to be on the safe side in case - // Futures#sequence accesses the passed List on a different thread, as - // recordedOperationFutures is not synchronized. - - Future> combinedFutures = akka.dispatch.Futures.sequence( - Lists.newArrayList(recordedOperationFutures), - actorContext.getClientDispatcher()); - OnComplete> onComplete = new OnComplete>() { - @Override - public void onComplete(Throwable failure, Iterable notUsed) - throws Throwable { - if(failure != null) { - LOG.debug("Tx {} dataExists: a recorded operation failed: {}", - identifier, failure); - returnFuture.setException(new ReadFailedException( - "The data exists could not be performed because a previous " - + "put, merge, or delete operation failed", failure)); - } else { - finishDataExists(path, returnFuture); - } - } - }; - - combinedFutures.onComplete(onComplete, actorContext.getClientDispatcher()); - } - } + LOG.debug("Tx {} dataExists called path = {}", getIdentifier(), path); - private void finishDataExists(final YangInstanceIdentifier path, - final SettableFuture returnFuture) { + // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the + // public API contract. - LOG.debug("Tx {} finishDataExists called path = {}", identifier, path); + sendAndRecordBatchedModifications(); OnComplete onComplete = new OnComplete() { @Override public void onComplete(Throwable failure, Object response) throws Throwable { if(failure != null) { - LOG.debug("Tx {} dataExists operation failed: {}", identifier, failure); + LOG.debug("Tx {} dataExists operation failed: {}", getIdentifier(), failure); returnFuture.setException(new ReadFailedException( "Error checking data exists for path " + path, failure)); } else { - LOG.debug("Tx {} dataExists operation succeeded", identifier, failure); + LOG.debug("Tx {} dataExists operation succeeded", getIdentifier(), failure); if (response instanceof DataExistsReply) { returnFuture.set(Boolean.valueOf(((DataExistsReply) response).exists()));