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%2FTransactionProxy.java;h=7f2f3281353bb689fdf5b74a12bb119d87ef8e59;hb=b03c5a20673792f0fb8df847fbaf9c359c7cce1b;hp=0bc82af3358c4747901fab005db5c19cf2d41288;hpb=e638203390314beb735c5d54f7936122e8d01d42;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java index 0bc82af335..7f2f328135 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java @@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import javax.annotation.concurrent.GuardedBy; +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.CloseTransaction; @@ -69,7 +70,19 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { public static enum TransactionType { READ_ONLY, WRITE_ONLY, - READ_WRITE + READ_WRITE; + + public static TransactionType fromInt(int type) { + if(type == WRITE_ONLY.ordinal()) { + return WRITE_ONLY; + } else if(type == READ_WRITE.ordinal()) { + return READ_WRITE; + } else if(type == READ_ONLY.ordinal()) { + return READ_ONLY; + } else { + throw new IllegalArgumentException("In TransactionType enum value" + type); + } + } } static final Mapper SAME_FAILURE_TRANSFORMER = @@ -425,18 +438,6 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { protected void onTransactionReady(List> cohortFutures) { } - /** - * Method called to send a CreateTransaction message to a shard. - * - * @param shard the shard actor to send to - * @param serializedCreateMessage the serialized message to send - * @return the response Future - */ - protected Future sendCreateTransaction(ActorSelection shard, - Object serializedCreateMessage) { - return actorContext.executeOperationAsync(shard, serializedCreateMessage); - } - @Override public Object getIdentifier() { return this.identifier; @@ -465,14 +466,17 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { return ShardStrategyFactory.getStrategy(path).findShard(path); } + protected Future sendFindPrimaryShardAsync(String shardName) { + return actorContext.findPrimaryShardAsync(shardName); + } + private TransactionFutureCallback getOrCreateTxFutureCallback(YangInstanceIdentifier path) { String shardName = shardNameFromIdentifier(path); TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName); if(txFutureCallback == null) { - Future findPrimaryFuture = actorContext.findPrimaryShardAsync(shardName); + Future findPrimaryFuture = sendFindPrimaryShardAsync(shardName); - final TransactionFutureCallback newTxFutureCallback = - new TransactionFutureCallback(shardName); + final TransactionFutureCallback newTxFutureCallback = new TransactionFutureCallback(shardName); txFutureCallback = newTxFutureCallback; txFutureCallbackMap.put(shardName, txFutureCallback); @@ -558,10 +562,20 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { * Sets the target primary shard and initiates a CreateTransaction try. */ void setPrimaryShard(ActorSelection primaryShard) { - LOG.debug("Tx {} Primary shard found - trying create transaction", identifier); - this.primaryShard = primaryShard; - tryCreateTransaction(); + + if(transactionType == TransactionType.WRITE_ONLY && + actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) { + LOG.debug("Tx {} Primary shard found - creating WRITE_ONLY transaction context", identifier); + + // 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. + executeTxOperatonsOnComplete(createValidTransactionContext(this.primaryShard, + this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION)); + } else { + tryCreateTransaction(); + } } /** @@ -571,7 +585,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { boolean invokeOperation = true; synchronized(txOperationsOnComplete) { if(transactionContext == null) { - LOG.debug("Tx {} Adding operation on complete {}", identifier); + LOG.debug("Tx {} Adding operation on complete", identifier); invokeOperation = false; txOperationsOnComplete.add(operation); @@ -598,10 +612,13 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { * Performs a CreateTransaction try async. */ private void tryCreateTransaction() { - Future createTxFuture = sendCreateTransaction(primaryShard, - new CreateTransaction(identifier.toString(), - TransactionProxy.this.transactionType.ordinal(), - getTransactionChainId()).toSerializable()); + LOG.debug("Tx {} Primary shard found - trying create transaction", identifier); + + Object serializedCreateMessage = new CreateTransaction(identifier.toString(), + TransactionProxy.this.transactionType.ordinal(), + getTransactionChainId()).toSerializable(); + + Future createTxFuture = actorContext.executeOperationAsync(primaryShard, serializedCreateMessage); createTxFuture.onComplete(this, actorContext.getClientDispatcher()); } @@ -643,8 +660,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { // TransactionContext until after we've executed all cached TransactionOperations. TransactionContext localTransactionContext; if(failure != null) { - LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier, - failure.getMessage()); + LOG.debug("Tx {} Creating NoOpTransaction because of error", identifier, failure); localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter); } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) { @@ -694,11 +710,14 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { } private TransactionContext createValidTransactionContext(CreateTransactionReply reply) { - String transactionPath = reply.getTransactionPath(); - LOG.debug("Tx {} Received {}", identifier, reply); - ActorSelection transactionActor = actorContext.actorSelection(transactionPath); + return createValidTransactionContext(actorContext.actorSelection(reply.getTransactionPath()), + reply.getTransactionPath(), reply.getVersion()); + } + + private TransactionContext createValidTransactionContext(ActorSelection transactionActor, + String transactionPath, short remoteTransactionVersion) { if (transactionType == TransactionType.READ_ONLY) { // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference @@ -727,12 +746,17 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction { // Check if TxActor is created in the same node boolean isTxActorLocal = actorContext.isPathLocal(transactionPath); - if(reply.getVersion() >= DataStoreVersions.LITHIUM_VERSION) { - return new TransactionContextImpl(transactionPath, transactionActor, identifier, - actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter); + if(remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) { + return new PreLithiumTransactionContextImpl(transactionPath, transactionActor, identifier, + transactionChainId, actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, + operationCompleter); + } else if (transactionType == TransactionType.WRITE_ONLY && + actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) { + return new WriteOnlyTransactionContextImpl(transactionActor, identifier, transactionChainId, + actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter); } else { - return new LegacyTransactionContextImpl(transactionPath, transactionActor, identifier, - actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter); + return new TransactionContextImpl(transactionActor, identifier, transactionChainId, + actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter); } } }