/* * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved. * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; import akka.dispatch.OnComplete; import akka.pattern.AskTimeoutException; import akka.util.Timeout; import com.google.common.base.Preconditions; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRespondingException; 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.TransactionIdentifierUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.Future; import scala.concurrent.duration.FiniteDuration; /** * Handles creation of TransactionContext instances for remote transactions. This class creates * remote transactions, if necessary, by sending CreateTransaction messages with retries, up to a limit, * if the shard doesn't have a leader yet. This is done by scheduling a retry task after a short delay. *
* The end result from a completed CreateTransaction message is a TransactionContext that is * used to perform transaction operations. Transaction operations that occur before the * CreateTransaction completes are cache via a TransactionContextWrapper and executed once the * CreateTransaction completes, successfully or not. */ final class RemoteTransactionContextSupport { private static final Logger LOG = LoggerFactory.getLogger(RemoteTransactionContextSupport.class); private static final long CREATE_TX_TRY_INTERVAL_IN_MS = 1000; private static final long MAX_CREATE_TX_MSG_TIMEOUT_IN_MS = 5000; private final TransactionProxy parent; private final String shardName; /** * The target primary shard. */ private volatile PrimaryShardInfo primaryShardInfo; /** * The total timeout for creating a tx on the primary shard. */ private volatile long totalCreateTxTimeout; private final Timeout createTxMessageTimeout; private final TransactionContextWrapper transactionContextWrapper; RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextWrapper, final TransactionProxy parent, final String shardName) { this.parent = Preconditions.checkNotNull(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() .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(); createTxMessageTimeout = new Timeout(Math.min(operationTimeout, MAX_CREATE_TX_MSG_TIMEOUT_IN_MS), TimeUnit.MILLISECONDS); } String getShardName() { return shardName; } private TransactionType getTransactionType() { return parent.getType(); } private ActorContext getActorContext() { return parent.getActorContext(); } private TransactionIdentifier getIdentifier() { return parent.getIdentifier(); } /** * Sets the target primary shard and initiates a CreateTransaction try. */ void setPrimaryShard(PrimaryShardInfo primaryShardInfo) { this.primaryShardInfo = primaryShardInfo; if (getTransactionType() == TransactionType.WRITE_ONLY && getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) { ActorSelection primaryShard = primaryShardInfo.getPrimaryShardActor(); LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context", getIdentifier(), primaryShard); // 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())); } else { tryCreateTransaction(); } } /** * @deprecated Temporary utility for extracting transaction chain ID from a {@link TransactionIdentifier} */ @Deprecated static String compatTransactionChainId(final TransactionIdentifier txId) { final long historyId = txId.getHistoryId().getHistoryId(); return historyId == 0 ? "" : Long.toUnsignedString(historyId); } /** * Performs a CreateTransaction try async. */ private void tryCreateTransaction() { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShardInfo.getPrimaryShardActor()); } Object serializedCreateMessage = new CreateTransaction(TransactionIdentifierUtils.actorNameFor(getIdentifier()), getTransactionType().ordinal(), compatTransactionChainId(getIdentifier()), primaryShardInfo.getPrimaryShardVersion()).toSerializable(); Future