X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FRemoteTransactionContextSupport.java;h=4276f3be52960e0ff9c68cd50adc5d9e591c4db6;hp=1e0d1279e3e749b9728a3d8cdfc1d61ec59facbc;hb=14c92df74247c884a43c5aaea2f154992b0ec798;hpb=925cb4a228d0fda99c7bfeb432eb25285a223887 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java index 1e0d1279e3..4276f3be52 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java @@ -8,11 +8,13 @@ */ package org.opendaylight.controller.cluster.datastore; +import static java.util.Objects.requireNonNull; + import akka.actor.ActorSelection; import akka.dispatch.OnComplete; import akka.pattern.AskTimeoutException; import akka.util.Timeout; -import com.google.common.base.Preconditions; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; @@ -20,7 +22,7 @@ import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRe import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; -import org.opendaylight.controller.cluster.datastore.utils.ActorContext; +import org.opendaylight.controller.cluster.datastore.utils.ActorUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.Future; @@ -61,19 +63,19 @@ final class RemoteTransactionContextSupport { RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextWrapper, final TransactionProxy parent, final String shardName) { - this.parent = Preconditions.checkNotNull(parent); + this.parent = requireNonNull(parent); this.shardName = shardName; this.transactionContextWrapper = transactionContextWrapper; // For the total create tx timeout, use 2 times the election timeout. This should be enough time for // a leader re-election to occur if we happen to hit it in transition. - totalCreateTxTimeout = parent.getActorContext().getDatastoreContext().getShardRaftConfig() + totalCreateTxTimeout = parent.getActorUtils().getDatastoreContext().getShardRaftConfig() .getElectionTimeOutInterval().toMillis() * 2; // We'll use the operationTimeout for the the create Tx message timeout so it can be set appropriately // for unit tests but cap it at MAX_CREATE_TX_MSG_TIMEOUT_IN_MS. The operationTimeout could be set // larger than the totalCreateTxTimeout in production which we don't want. - long operationTimeout = parent.getActorContext().getOperationTimeout().duration().toMillis(); + long operationTimeout = parent.getActorUtils().getOperationTimeout().duration().toMillis(); createTxMessageTimeout = new Timeout(Math.min(operationTimeout, MAX_CREATE_TX_MSG_TIMEOUT_IN_MS), TimeUnit.MILLISECONDS); } @@ -86,8 +88,8 @@ final class RemoteTransactionContextSupport { return parent.getType(); } - private ActorContext getActorContext() { - return parent.getActorContext(); + private ActorUtils getActorUtils() { + return parent.getActorUtils(); } private TransactionIdentifier getIdentifier() { @@ -97,12 +99,12 @@ final class RemoteTransactionContextSupport { /** * Sets the target primary shard and initiates a CreateTransaction try. */ - void setPrimaryShard(PrimaryShardInfo primaryShardInfo) { - this.primaryShardInfo = primaryShardInfo; + void setPrimaryShard(final PrimaryShardInfo newPrimaryShardInfo) { + this.primaryShardInfo = newPrimaryShardInfo; if (getTransactionType() == TransactionType.WRITE_ONLY - && getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) { - ActorSelection primaryShard = primaryShardInfo.getPrimaryShardActor(); + && getActorUtils().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) { + ActorSelection primaryShard = newPrimaryShardInfo.getPrimaryShardActor(); LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context", getIdentifier(), primaryShard); @@ -110,7 +112,7 @@ final class RemoteTransactionContextSupport { // For write-only Tx's we prepare the transaction modifications directly on the shard actor // to avoid the overhead of creating a separate transaction actor. transactionContextWrapper.executePriorTransactionOperations(createValidTransactionContext( - primaryShard, String.valueOf(primaryShard.path()), primaryShardInfo.getPrimaryShardVersion())); + primaryShard, String.valueOf(primaryShard.path()), newPrimaryShardInfo.getPrimaryShardVersion())); } else { tryCreateTransaction(); } @@ -126,30 +128,32 @@ final class RemoteTransactionContextSupport { Object serializedCreateMessage = new CreateTransaction(getIdentifier(), getTransactionType().ordinal(), primaryShardInfo.getPrimaryShardVersion()).toSerializable(); - Future createTxFuture = getActorContext().executeOperationAsync( + Future createTxFuture = getActorUtils().executeOperationAsync( primaryShardInfo.getPrimaryShardActor(), serializedCreateMessage, createTxMessageTimeout); createTxFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { onCreateTransactionComplete(failure, response); } - }, getActorContext().getClientDispatcher()); + }, getActorUtils().getClientDispatcher()); } private void tryFindPrimaryShard() { LOG.debug("Tx {} Retrying findPrimaryShardAsync for shard {}", getIdentifier(), shardName); this.primaryShardInfo = null; - Future findPrimaryFuture = getActorContext().findPrimaryShardAsync(shardName); + Future findPrimaryFuture = getActorUtils().findPrimaryShardAsync(shardName); findPrimaryFuture.onComplete(new OnComplete() { @Override public void onComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) { onFindPrimaryShardComplete(failure, newPrimaryShardInfo); } - }, getActorContext().getClientDispatcher()); + }, getActorUtils().getClientDispatcher()); } + @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", + justification = "https://github.com/spotbugs/spotbugs/issues/811") private void onFindPrimaryShardComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) { if (failure == null) { this.primaryShardInfo = newPrimaryShardInfo; @@ -161,41 +165,40 @@ final class RemoteTransactionContextSupport { } } - private void onCreateTransactionComplete(Throwable failure, Object response) { + private void onCreateTransactionComplete(final Throwable failure, final Object response) { // An AskTimeoutException will occur if the local shard forwards to an unavailable remote leader or // the cached remote leader actor is no longer available. boolean retryCreateTransaction = primaryShardInfo != null && (failure instanceof NoShardLeaderException || failure instanceof AskTimeoutException); - if (retryCreateTransaction) { - // Schedule a retry unless we're out of retries. Note: totalCreateTxTimeout is volatile as it may - // be written by different threads however not concurrently, therefore decrementing it - // non-atomically here is ok. - if (totalCreateTxTimeout > 0) { - long scheduleInterval = CREATE_TX_TRY_INTERVAL_IN_MS; - if (failure instanceof AskTimeoutException) { - // Since we use the createTxMessageTimeout for the CreateTransaction request and it timed - // out, subtract it from the total timeout. Also since the createTxMessageTimeout period - // has already elapsed, we can immediately schedule the retry (10 ms is virtually immediate). - totalCreateTxTimeout -= createTxMessageTimeout.duration().toMillis(); - scheduleInterval = 10; - } - - totalCreateTxTimeout -= scheduleInterval; - - LOG.debug("Tx {}: create tx on shard {} failed with exception \"{}\" - scheduling retry in {} ms", - getIdentifier(), shardName, failure, scheduleInterval); - - getActorContext().getActorSystem().scheduler().scheduleOnce( - FiniteDuration.create(scheduleInterval, TimeUnit.MILLISECONDS), - () -> tryFindPrimaryShard(), getActorContext().getClientDispatcher()); - return; + + // Schedule a retry unless we're out of retries. Note: totalCreateTxTimeout is volatile as it may + // be written by different threads however not concurrently, therefore decrementing it + // non-atomically here is ok. + if (retryCreateTransaction && totalCreateTxTimeout > 0) { + long scheduleInterval = CREATE_TX_TRY_INTERVAL_IN_MS; + if (failure instanceof AskTimeoutException) { + // Since we use the createTxMessageTimeout for the CreateTransaction request and it timed + // out, subtract it from the total timeout. Also since the createTxMessageTimeout period + // has already elapsed, we can immediately schedule the retry (10 ms is virtually immediate). + totalCreateTxTimeout -= createTxMessageTimeout.duration().toMillis(); + scheduleInterval = 10; } + + totalCreateTxTimeout -= scheduleInterval; + + LOG.debug("Tx {}: create tx on shard {} failed with exception \"{}\" - scheduling retry in {} ms", + getIdentifier(), shardName, failure, scheduleInterval); + + getActorUtils().getActorSystem().scheduler().scheduleOnce( + FiniteDuration.create(scheduleInterval, TimeUnit.MILLISECONDS), + this::tryFindPrimaryShard, getActorUtils().getClientDispatcher()); + return; } createTransactionContext(failure, response); } - private void createTransactionContext(Throwable failure, Object response) { + private void createTransactionContext(final Throwable failure, final Object response) { // Create the TransactionContext from the response or failure. Store the new // TransactionContext locally until we've completed invoking the // TransactionOperations. This avoids thread timing issues which could cause @@ -232,17 +235,17 @@ final class RemoteTransactionContextSupport { transactionContextWrapper.executePriorTransactionOperations(localTransactionContext); } - private TransactionContext createValidTransactionContext(CreateTransactionReply reply) { + private TransactionContext createValidTransactionContext(final CreateTransactionReply reply) { LOG.debug("Tx {} Received {}", getIdentifier(), reply); - return createValidTransactionContext(getActorContext().actorSelection(reply.getTransactionPath()), + return createValidTransactionContext(getActorUtils().actorSelection(reply.getTransactionPath()), reply.getTransactionPath(), primaryShardInfo.getPrimaryShardVersion()); } - private TransactionContext createValidTransactionContext(ActorSelection transactionActor, String transactionPath, - short remoteTransactionVersion) { + private TransactionContext createValidTransactionContext(final ActorSelection transactionActor, + final String transactionPath, final short remoteTransactionVersion) { final TransactionContext ret = new RemoteTransactionContext(transactionContextWrapper.getIdentifier(), - transactionActor, getActorContext(), remoteTransactionVersion, transactionContextWrapper.getLimiter()); + transactionActor, getActorUtils(), remoteTransactionVersion, transactionContextWrapper.getLimiter()); if (parent.getType() == TransactionType.READ_ONLY) { TransactionContextCleanup.track(parent, ret);