/* * 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 com.google.common.base.Preconditions; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.datastore.compat.PreLithiumTransactionContextImpl; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; 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); /** * Time interval in between transaction create retries. */ private static final FiniteDuration CREATE_TX_TRY_INTERVAL = FiniteDuration.create(1, TimeUnit.SECONDS); private final TransactionProxy parent; private final String shardName; /** * The target primary shard. */ private volatile ActorSelection primaryShard; private volatile int createTxTries; private final TransactionContextWrapper transactionContextAdapter; RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextAdapter, final TransactionProxy parent, final String shardName) { this.parent = Preconditions.checkNotNull(parent); this.shardName = shardName; this.transactionContextAdapter = transactionContextAdapter; createTxTries = (int) (parent.getActorContext().getDatastoreContext(). getShardLeaderElectionTimeout().duration().toMillis() / CREATE_TX_TRY_INTERVAL.toMillis()); } String getShardName() { return shardName; } private TransactionType getTransactionType() { return parent.getType(); } private ActorContext getActorContext() { return parent.getActorContext(); } private OperationLimiter getOperationLimiter() { return parent.getLimiter(); } private TransactionIdentifier getIdentifier() { return parent.getIdentifier(); } /** * Sets the target primary shard and initiates a CreateTransaction try. */ void setPrimaryShard(ActorSelection primaryShard) { this.primaryShard = primaryShard; if (getTransactionType() == TransactionType.WRITE_ONLY && getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) { 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. // FIXME: can't assume the shard version is LITHIUM_VERSION - need to obtain it somehow. transactionContextAdapter.executePriorTransactionOperations(createValidTransactionContext(this.primaryShard, this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION)); } else { tryCreateTransaction(); } } /** * Performs a CreateTransaction try async. */ private void tryCreateTransaction() { if(LOG.isDebugEnabled()) { LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShard); } Object serializedCreateMessage = new CreateTransaction(getIdentifier().toString(), getTransactionType().ordinal(), getIdentifier().getChainId()).toSerializable(); Future