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%2FShardCommitCoordinator.java;h=5177f7ed2bcbaee7d03b53a894c5279d17a3f9f2;hb=99f80f27bee37bb23e345420bf14bb7bb4793c28;hp=080d3eec23a22ef3f7cb9771d78f42e905d08dc1;hpb=df3361f6703078faff9c4b029d0021a63ebe6e6e;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java index 080d3eec23..5177f7ed2b 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardCommitCoordinator.java @@ -7,11 +7,12 @@ */ package org.opendaylight.controller.cluster.datastore; +import static java.util.Objects.requireNonNull; + import akka.actor.ActorRef; import akka.actor.Status.Failure; import akka.serialization.Serialization; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; import com.google.common.primitives.UnsignedLong; import com.google.common.util.concurrent.FutureCallback; import java.util.ArrayDeque; @@ -20,10 +21,9 @@ import java.util.Collection; import java.util.Collections; import java.util.Deque; import java.util.HashMap; -import java.util.Iterator; import java.util.LinkedList; import java.util.Map; -import javax.annotation.Nonnull; +import org.eclipse.jdt.annotation.NonNull; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; @@ -38,7 +38,8 @@ import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionRe import org.opendaylight.controller.cluster.datastore.messages.VersionedExternalizableMessage; import org.opendaylight.controller.cluster.datastore.utils.AbstractBatchedModificationsCursor; import org.opendaylight.yangtools.concepts.Identifier; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; +import org.opendaylight.yangtools.yang.common.Empty; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate; import org.slf4j.Logger; /** @@ -71,7 +72,7 @@ final class ShardCommitCoordinator { ShardCommitCoordinator(final ShardDataTree dataTree, final Logger log, final String name) { this.log = log; this.name = name; - this.dataTree = Preconditions.checkNotNull(dataTree); + this.dataTree = requireNonNull(dataTree); } int getCohortCacheSize() { @@ -103,7 +104,7 @@ final class ShardCommitCoordinator { log.debug("{}: Readying transaction {}, client version {}", name, ready.getTransactionId(), ready.getTxnClientVersion()); - final ShardDataTreeCohort cohort = ready.getTransaction().ready(); + final ShardDataTreeCohort cohort = ready.getTransaction().ready(ready.getParticipatingShardNames()); final CohortEntry cohortEntry = CohortEntry.createReady(cohort, ready.getTxnClientVersion()); cohortCache.put(cohortEntry.getTransactionId(), cohortEntry); @@ -130,7 +131,7 @@ final class ShardCommitCoordinator { */ void handleBatchedModifications(final BatchedModifications batched, final ActorRef sender, final Shard shard) { CohortEntry cohortEntry = cohortCache.get(batched.getTransactionId()); - if (cohortEntry == null) { + if (cohortEntry == null || cohortEntry.isSealed()) { cohortEntry = CohortEntry.createOpen(dataTree.newReadWriteTransaction(batched.getTransactionId()), batched.getVersion()); cohortCache.put(cohortEntry.getTransactionId(), cohortEntry); @@ -157,12 +158,12 @@ final class ShardCommitCoordinator { } if (log.isDebugEnabled()) { - log.debug("{}: Readying Tx {}, client version {}", name, - batched.getTransactionId(), batched.getVersion()); + log.debug("{}: Readying Tx {} of {} operations, client version {}", name, + batched.getTransactionId(), cohortEntry.getTotalOperationsProcessed(), batched.getVersion()); } cohortEntry.setDoImmediateCommit(batched.isDoCommitOnReady()); - cohortEntry.ready(cohortDecorator); + cohortEntry.ready(batched.getParticipatingShardNames(), cohortDecorator); if (batched.isDoCommitOnReady()) { cohortEntry.setReplySender(sender); @@ -187,7 +188,8 @@ final class ShardCommitCoordinator { */ void handleReadyLocalTransaction(final ReadyLocalTransaction message, final ActorRef sender, final Shard shard) { final TransactionIdentifier txId = message.getTransactionId(); - final ShardDataTreeCohort cohort = dataTree.newReadyCohort(txId, message.getModification()); + final ShardDataTreeCohort cohort = dataTree.newReadyCohort(txId, message.getModification(), + message.getParticipatingShardNames()); final CohortEntry cohortEntry = CohortEntry.createReady(cohort, DataStoreVersions.CURRENT_VERSION); cohortCache.put(cohortEntry.getTransactionId(), cohortEntry); cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady()); @@ -227,15 +229,17 @@ final class ShardCommitCoordinator { BatchedModifications last = newModifications.getLast(); last.setDoCommitOnReady(from.isDoCommitOnReady()); - last.setReady(from.isReady()); + if (from.isReady()) { + last.setReady(from.getParticipatingShardNames()); + } last.setTotalMessagesSent(newModifications.size()); return newModifications; } private void handleCanCommit(final CohortEntry cohortEntry) { - cohortEntry.canCommit(new FutureCallback() { + cohortEntry.canCommit(new FutureCallback<>() { @Override - public void onSuccess(final Void result) { + public void onSuccess(final Empty result) { log.debug("{}: canCommit for {}: success", name, cohortEntry.getTransactionId()); if (cohortEntry.isDoImmediateCommit()) { @@ -249,8 +253,8 @@ final class ShardCommitCoordinator { @Override public void onFailure(final Throwable failure) { - log.debug("{}: An exception occurred during canCommit for {}: {}", name, - cohortEntry.getTransactionId(), failure); + log.debug("{}: An exception occurred during canCommit for {}", name, cohortEntry.getTransactionId(), + failure); cohortCache.remove(cohortEntry.getTransactionId()); cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self()); @@ -274,7 +278,7 @@ final class ShardCommitCoordinator { // between canCommit and ready and the entry was expired from the cache or it was aborted. IllegalStateException ex = new IllegalStateException( String.format("%s: Cannot canCommit transaction %s - no cohort entry found", name, transactionID)); - log.error(ex.getMessage()); + log.error("{}: Inconsistency during transaction {} canCommit", name, transactionID, ex); sender.tell(new Failure(ex), shard.self()); return; } @@ -309,7 +313,7 @@ final class ShardCommitCoordinator { }); } - void finishCommit(@Nonnull final ActorRef sender, @Nonnull final CohortEntry cohortEntry) { + void finishCommit(final @NonNull ActorRef sender, final @NonNull CohortEntry cohortEntry) { log.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionId()); cohortEntry.commit(new FutureCallback() { @@ -318,7 +322,6 @@ final class ShardCommitCoordinator { final TransactionIdentifier txId = cohortEntry.getTransactionId(); log.debug("{}: Transaction {} committed as {}, sending response to {}", persistenceId(), txId, result, sender); - cohortEntry.getShard().getDataStore().purgeTransaction(txId, null); cohortCache.remove(cohortEntry.getTransactionId()); sender.tell(CommitTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), @@ -329,7 +332,6 @@ final class ShardCommitCoordinator { public void onFailure(final Throwable failure) { final TransactionIdentifier txId = cohortEntry.getTransactionId(); log.error("{}, An exception occurred while committing transaction {}", persistenceId(), txId, failure); - cohortEntry.getShard().getDataStore().purgeTransaction(txId, null); cohortCache.remove(cohortEntry.getTransactionId()); sender.tell(new Failure(failure), cohortEntry.getShard().self()); @@ -351,7 +353,7 @@ final class ShardCommitCoordinator { // or it was aborted. IllegalStateException ex = new IllegalStateException( String.format("%s: Cannot commit transaction %s - no cohort entry found", name, transactionID)); - log.error(ex.getMessage()); + log.error("{}: Inconsistency during transaction {} commit", name, transactionID, ex); sender.tell(new Failure(ex), shard.self()); return; } @@ -370,11 +372,9 @@ final class ShardCommitCoordinator { log.debug("{}: Aborting transaction {}", name, transactionID); final ActorRef self = shard.getSelf(); - cohortEntry.abort(new FutureCallback() { + cohortEntry.abort(new FutureCallback<>() { @Override - public void onSuccess(final Void result) { - shard.getDataStore().purgeTransaction(cohortEntry.getTransactionId(), null); - + public void onSuccess(final Empty result) { if (sender != null) { sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self); } @@ -383,7 +383,6 @@ final class ShardCommitCoordinator { @Override public void onFailure(final Throwable failure) { log.error("{}: An exception happened during abort", name, failure); - shard.getDataStore().purgeTransaction(cohortEntry.getTransactionId(), null); if (sender != null) { sender.tell(new Failure(failure), self); @@ -395,13 +394,7 @@ final class ShardCommitCoordinator { } void checkForExpiredTransactions(final long timeout, final Shard shard) { - Iterator iter = cohortCache.values().iterator(); - while (iter.hasNext()) { - CohortEntry cohortEntry = iter.next(); - if (cohortEntry.isFailed()) { - iter.remove(); - } - } + cohortCache.values().removeIf(CohortEntry::isFailed); } void abortPendingTransactions(final String reason, final Shard shard) { @@ -454,7 +447,7 @@ final class ShardCommitCoordinator { if (last != null) { final boolean immediate = cohortEntry.isDoImmediateCommit(); last.setDoCommitOnReady(immediate); - last.setReady(true); + last.setReady(cohortEntry.getParticipatingShardNames()); last.setTotalMessagesSent(newMessages.size()); messages.addAll(newMessages);