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%2FAbstractTransactionContextFactory.java;h=6d573dedf53223b34c8c2ca525bea3435d9bc812;hb=55a9b9f42a14c56060f74b38f84d444c0fbfecc4;hp=a8a076f98fcee0c73919d8187ddb64ec2b88718b;hpb=95538aa1baf80a90f03e4ae6d8268c9db34b3bfa;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java index a8a076f98f..6d573dedf5 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java @@ -7,20 +7,25 @@ */ package org.opendaylight.controller.cluster.datastore; +import static java.util.Objects.requireNonNull; + import akka.actor.ActorSelection; import akka.dispatch.OnComplete; -import com.google.common.base.Optional; -import com.google.common.base.Preconditions; import java.util.Collection; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.atomic.AtomicLong; -import javax.annotation.Nonnull; -import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import org.eclipse.jdt.annotation.NonNull; +import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier; +import org.opendaylight.controller.cluster.access.concepts.MemberName; +import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; -import org.opendaylight.controller.cluster.datastore.utils.ActorContext; -import org.opendaylight.controller.cluster.datastore.utils.ShardInfoListener; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; +import org.opendaylight.controller.cluster.datastore.utils.ActorUtils; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction; +import org.opendaylight.yangtools.yang.data.api.schema.tree.ReadOnlyDataTree; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.Future; @@ -30,131 +35,131 @@ import scala.util.Try; * Factory for creating local and remote TransactionContext instances. Maintains a cache of known local * transaction factories. */ -abstract class AbstractTransactionContextFactory - implements ShardInfoListener, AutoCloseable { +abstract class AbstractTransactionContextFactory implements AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(AbstractTransactionContextFactory.class); - - protected static final AtomicLong TX_COUNTER = new AtomicLong(); + @SuppressWarnings("rawtypes") + private static final AtomicLongFieldUpdater TX_COUNTER_UPDATER = + AtomicLongFieldUpdater.newUpdater(AbstractTransactionContextFactory.class, "nextTx"); private final ConcurrentMap knownLocal = new ConcurrentHashMap<>(); - private final ActorContext actorContext; + private final LocalHistoryIdentifier historyId; + private final ActorUtils actorUtils; + + // Used via TX_COUNTER_UPDATER + @SuppressWarnings("unused") + private volatile long nextTx; - protected AbstractTransactionContextFactory(final ActorContext actorContext) { - this.actorContext = Preconditions.checkNotNull(actorContext); + protected AbstractTransactionContextFactory(final ActorUtils actorUtils, final LocalHistoryIdentifier historyId) { + this.actorUtils = requireNonNull(actorUtils); + this.historyId = requireNonNull(historyId); } - final ActorContext getActorContext() { - return actorContext; + final ActorUtils getActorUtils() { + return actorUtils; } - private TransactionContext maybeCreateLocalTransactionContext(final TransactionProxy parent, final String shardName) { + final LocalHistoryIdentifier getHistoryId() { + return historyId; + } + + @SuppressWarnings("checkstyle:IllegalCatch") + private TransactionContext maybeCreateLocalTransactionContext(final TransactionProxy parent, + final String shardName) { final LocalTransactionFactory local = knownLocal.get(shardName); if (local != null) { - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} - Creating local component for shard {} using factory {}", - parent.getIdentifier(), shardName, local); + LOG.debug("Tx {} - Creating local component for shard {} using factory {}", parent.getIdentifier(), + shardName, local); + + try { + return createLocalTransactionContext(local, parent); + } catch (Exception e) { + return new NoOpTransactionContext(e, parent.getIdentifier()); } - return createLocalTransactionContext(local, parent); } return null; } private void onFindPrimaryShardSuccess(PrimaryShardInfo primaryShardInfo, TransactionProxy parent, - String shardName, TransactionContextWrapper transactionContextAdapter) { - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {}: Found primary {} for shard {}", parent.getIdentifier(), - primaryShardInfo.getPrimaryShardActor(), shardName); - } + String shardName, TransactionContextWrapper transactionContextWrapper) { + LOG.debug("Tx {}: Found primary {} for shard {}", parent.getIdentifier(), + primaryShardInfo.getPrimaryShardActor(), shardName); updateShardInfo(shardName, primaryShardInfo); - TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName); - if(localContext != null) { - transactionContextAdapter.executePriorTransactionOperations(localContext); - } else { - RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(transactionContextAdapter, - parent, shardName); - remote.setPrimaryShard(primaryShardInfo.getPrimaryShardActor()); + try { + TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName); + if (localContext != null) { + transactionContextWrapper.executePriorTransactionOperations(localContext); + } else { + RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(transactionContextWrapper, + parent, shardName); + remote.setPrimaryShard(primaryShardInfo); + } + } finally { + onTransactionContextCreated(parent.getIdentifier()); } } private void onFindPrimaryShardFailure(Throwable failure, TransactionProxy parent, - String shardName, TransactionContextWrapper transactionContextAdapter) { + String shardName, TransactionContextWrapper transactionContextWrapper) { LOG.debug("Tx {}: Find primary for shard {} failed", parent.getIdentifier(), shardName, failure); - transactionContextAdapter.executePriorTransactionOperations(new NoOpTransactionContext(failure, - parent.getIdentifier(), parent.getLimiter())); + try { + transactionContextWrapper.executePriorTransactionOperations(new NoOpTransactionContext(failure, + parent.getIdentifier())); + } finally { + onTransactionContextCreated(parent.getIdentifier()); + } } - final TransactionContextWrapper newTransactionAdapter(final TransactionProxy parent, final String shardName) { - final TransactionContextWrapper transactionContextAdapter = new TransactionContextWrapper(parent.getIdentifier()); + final TransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent, + final String shardName) { + final TransactionContextWrapper transactionContextWrapper = + new TransactionContextWrapper(parent.getIdentifier(), actorUtils, shardName); - Future findPrimaryFuture = findPrimaryShard(shardName); - if(findPrimaryFuture.isCompleted()) { + Future findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier()); + if (findPrimaryFuture.isCompleted()) { Try maybe = findPrimaryFuture.value().get(); - if(maybe.isSuccess()) { - onFindPrimaryShardSuccess(maybe.get(), parent, shardName, transactionContextAdapter); + if (maybe.isSuccess()) { + onFindPrimaryShardSuccess(maybe.get(), parent, shardName, transactionContextWrapper); } else { - onFindPrimaryShardFailure(maybe.failed().get(), parent, shardName, transactionContextAdapter); + onFindPrimaryShardFailure(maybe.failed().get(), parent, shardName, transactionContextWrapper); } } else { findPrimaryFuture.onComplete(new OnComplete() { @Override public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) { if (failure == null) { - onFindPrimaryShardSuccess(primaryShardInfo, parent, shardName, transactionContextAdapter); + onFindPrimaryShardSuccess(primaryShardInfo, parent, shardName, transactionContextWrapper); } else { - onFindPrimaryShardFailure(failure, parent, shardName, transactionContextAdapter); + onFindPrimaryShardFailure(failure, parent, shardName, transactionContextWrapper); } } - }, actorContext.getClientDispatcher()); + }, actorUtils.getClientDispatcher()); } - return transactionContextAdapter; + return transactionContextWrapper; } private void updateShardInfo(final String shardName, final PrimaryShardInfo primaryShardInfo) { - final Optional maybeDataTree = primaryShardInfo.getLocalShardDataTree(); + final Optional maybeDataTree = primaryShardInfo.getLocalShardDataTree(); if (maybeDataTree.isPresent()) { - knownLocal.put(shardName, factoryForShard(shardName, primaryShardInfo.getPrimaryShardActor(), maybeDataTree.get())); - LOG.debug("Shard {} resolved to local data tree", shardName); - } - } + if (!knownLocal.containsKey(shardName)) { + LOG.debug("Shard {} resolved to local data tree - adding local factory", shardName); - @Override - public void onShardInfoUpdated(final String shardName, final PrimaryShardInfo primaryShardInfo) { - final F existing = knownLocal.get(shardName); - if (existing != null) { - if (primaryShardInfo != null) { - final Optional maybeDataTree = primaryShardInfo.getLocalShardDataTree(); - if (maybeDataTree.isPresent()) { - final DataTree newDataTree = maybeDataTree.get(); - final DataTree oldDataTree = dataTreeForFactory(existing); - if (!oldDataTree.equals(newDataTree)) { - final F newChain = factoryForShard(shardName, primaryShardInfo.getPrimaryShardActor(), newDataTree); - knownLocal.replace(shardName, existing, newChain); - LOG.debug("Replaced shard {} local data tree to {}", shardName, newDataTree); - } - - return; - } + F factory = factoryForShard(shardName, primaryShardInfo.getPrimaryShardActor(), maybeDataTree.get()); + knownLocal.putIfAbsent(shardName, factory); } - if (knownLocal.remove(shardName, existing)) { - LOG.debug("Shard {} invalidated data tree {}", shardName, existing); - } else { - LOG.debug("Shard {} failed to invalidate data tree {} ... strange", shardName, existing); - } - } - } + } else if (knownLocal.containsKey(shardName)) { + LOG.debug("Shard {} invalidating local data tree", shardName); - protected String getMemberName() { - String memberName = getActorContext().getCurrentMemberName(); - if (memberName == null) { - memberName = "UNKNOWN-MEMBER"; + knownLocal.remove(shardName); } + } - return memberName; + protected final MemberName getMemberName() { + return historyId.getClientId().getFrontendId().getMemberName(); } /** @@ -162,7 +167,9 @@ abstract class AbstractTransactionContextFactory findPrimaryShard(String shardName); + protected abstract Future findPrimaryShard(@NonNull String shardName, + @NonNull TransactionIdentifier txId); /** * Create local transaction factory for specified shard, backed by specified shard leader * and data tree instance. * - * @param shardName - * @param shardLeader + * @param shardName the shard name + * @param shardLeader the shard leader * @param dataTree Backing data tree instance. The data tree may only be accessed in * read-only manner. * @return Transaction factory for local use. */ - protected abstract F factoryForShard(String shardName, ActorSelection shardLeader, DataTree dataTree); - - /** - * Extract the backing data tree from a particular factory. - * - * @param factory Transaction factory - * @return Backing data tree - */ - protected abstract DataTree dataTreeForFactory(F factory); + protected abstract F factoryForShard(String shardName, ActorSelection shardLeader, ReadOnlyDataTree dataTree); /** * Callback invoked from child transactions to push any futures, which need to * be waited for before the next transaction is allocated. * @param cohortFutures Collection of futures */ - protected abstract void onTransactionReady(@Nonnull TransactionIdentifier transaction, @Nonnull Collection> cohortFutures); + protected abstract void onTransactionReady(@NonNull TransactionIdentifier transaction, + @NonNull Collection> cohortFutures); + + /** + * Callback invoked when the internal TransactionContext has been created for a transaction. + * + * @param transactionId the ID of the transaction. + */ + protected abstract void onTransactionContextCreated(@NonNull TransactionIdentifier transactionId); + + private static TransactionContext createLocalTransactionContext(final LocalTransactionFactory factory, + final TransactionProxy parent) { - private static TransactionContext createLocalTransactionContext(final LocalTransactionFactory factory, final TransactionProxy parent) { - return new LocalTransactionContext(parent.getIdentifier(), factory.newReadWriteTransaction(parent.getIdentifier()), parent.getCompleter()); + switch (parent.getType()) { + case READ_ONLY: + final DOMStoreReadTransaction readOnly = factory.newReadOnlyTransaction(parent.getIdentifier()); + return new LocalTransactionContext(readOnly, parent.getIdentifier(), factory) { + @Override + protected DOMStoreWriteTransaction getWriteDelegate() { + throw new UnsupportedOperationException(); + } + + @Override + protected DOMStoreReadTransaction getReadDelegate() { + return readOnly; + } + }; + case READ_WRITE: + final DOMStoreReadWriteTransaction readWrite = factory.newReadWriteTransaction(parent.getIdentifier()); + return new LocalTransactionContext(readWrite, parent.getIdentifier(), factory) { + @Override + protected DOMStoreWriteTransaction getWriteDelegate() { + return readWrite; + } + + @Override + protected DOMStoreReadTransaction getReadDelegate() { + return readWrite; + } + }; + case WRITE_ONLY: + final DOMStoreWriteTransaction writeOnly = factory.newWriteOnlyTransaction(parent.getIdentifier()); + return new LocalTransactionContext(writeOnly, parent.getIdentifier(), factory) { + @Override + protected DOMStoreWriteTransaction getWriteDelegate() { + return writeOnly; + } + + @Override + protected DOMStoreReadTransaction getReadDelegate() { + throw new UnsupportedOperationException(); + } + }; + default: + throw new IllegalArgumentException("Invalid transaction type: " + parent.getType()); + } } }