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=667b64ce18839386f0c9ce44375bd7f88649588a;hb=f97618f25dfc073d1de5d883f1794eefdb3e5c16;hp=64f914b19fbebfa517afb2a7f23deb473a46a1a5;hpb=edcc020c8fda4b13f22a31d79c13feef0b53b0ee;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 64f914b19f..667b64ce18 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
@@ -41,6 +41,7 @@ import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionR
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.controller.sal.core.spi.data.AbstractDOMStoreTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
@@ -65,22 +66,21 @@ import scala.concurrent.duration.FiniteDuration;
* shards will be executed.
*
*/
-public class TransactionProxy implements DOMStoreReadWriteTransaction {
+public class TransactionProxy extends AbstractDOMStoreTransaction implements DOMStoreReadWriteTransaction {
public static enum TransactionType {
READ_ONLY,
WRITE_ONLY,
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);
+ // Cache all values
+ private static final TransactionType[] VALUES = values();
+
+ public static TransactionType fromInt(final int type) {
+ try {
+ return VALUES[type];
+ } catch (IndexOutOfBoundsException e) {
+ throw new IllegalArgumentException("In TransactionType enum value " + type, e);
}
}
}
@@ -147,7 +147,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
remoteTransactionActors = referent.remoteTransactionActors;
remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
actorContext = referent.actorContext;
- identifier = referent.identifier;
+ identifier = referent.getIdentifier();
}
@Override
@@ -186,7 +186,6 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
private final TransactionType transactionType;
private final ActorContext actorContext;
- private final TransactionIdentifier identifier;
private final String transactionChainId;
private final SchemaContext schemaContext;
private boolean inReadyState;
@@ -199,8 +198,8 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
this(actorContext, transactionType, "");
}
- public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
- String transactionChainId) {
+ public TransactionProxy(ActorContext actorContext, TransactionType transactionType, String transactionChainId) {
+ super(createIdentifier(actorContext));
this.actorContext = Preconditions.checkNotNull(actorContext,
"actorContext should not be null");
this.transactionType = Preconditions.checkNotNull(transactionType,
@@ -209,14 +208,16 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
"schemaContext should not be null");
this.transactionChainId = transactionChainId;
+ LOG.debug("Created txn {} of type {} on chain {}", getIdentifier(), transactionType, transactionChainId);
+ }
+
+ private static TransactionIdentifier createIdentifier(ActorContext actorContext) {
String memberName = actorContext.getCurrentMemberName();
- if(memberName == null){
+ if (memberName == null) {
memberName = "UNKNOWN-MEMBER";
}
- this.identifier = new TransactionIdentifier(memberName, counter.getAndIncrement());
-
- LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
+ return new TransactionIdentifier(memberName, counter.getAndIncrement());
}
@VisibleForTesting
@@ -250,7 +251,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
"Read operation on write-only transaction is not allowed");
- LOG.debug("Tx {} read {}", identifier, path);
+ LOG.debug("Tx {} read {}", getIdentifier(), path);
throttleOperation();
@@ -273,7 +274,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
"Exists operation on write-only transaction is not allowed");
- LOG.debug("Tx {} exists {}", identifier, path);
+ LOG.debug("Tx {} exists {}", getIdentifier(), path);
throttleOperation();
@@ -332,7 +333,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} write {}", identifier, path);
+ LOG.debug("Tx {} write {}", getIdentifier(), path);
throttleOperation();
@@ -350,7 +351,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} merge {}", identifier, path);
+ LOG.debug("Tx {} merge {}", getIdentifier(), path);
throttleOperation();
@@ -368,7 +369,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} delete {}", identifier, path);
+ LOG.debug("Tx {} delete {}", getIdentifier(), path);
throttleOperation();
@@ -388,11 +389,12 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
inReadyState = true;
- LOG.debug("Tx {} Readying {} transactions for commit", identifier,
+ LOG.debug("Tx {} Readying {} transactions for commit", getIdentifier(),
txFutureCallbackMap.size());
if(txFutureCallbackMap.size() == 0) {
onTransactionReady(Collections.>emptyList());
+ TransactionRateLimitingCallback.adjustRateLimitForUnusedTransaction(actorContext);
return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
}
@@ -402,7 +404,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
- LOG.debug("Tx {} Readying transaction for shard {} chain {}", identifier,
+ LOG.debug("Tx {} Readying transaction for shard {} chain {}", getIdentifier(),
txFutureCallback.getShardName(), transactionChainId);
final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
@@ -427,7 +429,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
onTransactionReady(cohortFutures);
return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
- identifier.toString());
+ getIdentifier().toString());
}
/**
@@ -438,11 +440,6 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
protected void onTransactionReady(List> cohortFutures) {
}
- @Override
- public Object getIdentifier() {
- return this.identifier;
- }
-
@Override
public void close() {
for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
@@ -567,7 +564,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
if(transactionType == TransactionType.WRITE_ONLY &&
actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
- identifier, primaryShard);
+ 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.
@@ -586,7 +583,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", getIdentifier());
invokeOperation = false;
txOperationsOnComplete.add(operation);
@@ -614,10 +611,10 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
*/
private void tryCreateTransaction() {
if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} Primary shard {} found - trying create transaction", identifier, primaryShard);
+ LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShard);
}
- Object serializedCreateMessage = new CreateTransaction(identifier.toString(),
+ Object serializedCreateMessage = new CreateTransaction(getIdentifier().toString(),
TransactionProxy.this.transactionType.ordinal(),
getTransactionChainId()).toSerializable();
@@ -635,7 +632,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
// is ok.
if(--createTxTries > 0) {
LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
- identifier, shardName);
+ getIdentifier(), shardName);
actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
new Runnable() {
@@ -667,9 +664,9 @@ 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);
+ LOG.debug("Tx {} Creating NoOpTransaction because of error", getIdentifier(), failure);
- localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
+ localTransactionContext = new NoOpTransactionContext(failure, getIdentifier(), operationLimiter);
} else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
localTransactionContext = createValidTransactionContext(
CreateTransactionReply.fromSerializable(response));
@@ -677,7 +674,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
IllegalArgumentException exception = new IllegalArgumentException(String.format(
"Invalid reply type %s for CreateTransaction", response.getClass()));
- localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
+ localTransactionContext = new NoOpTransactionContext(exception, getIdentifier(), operationLimiter);
}
executeTxOperatonsOnComplete(localTransactionContext);
@@ -717,7 +714,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
}
private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
- LOG.debug("Tx {} Received {}", identifier, reply);
+ LOG.debug("Tx {} Received {}", getIdentifier(), reply);
return createValidTransactionContext(actorContext.actorSelection(reply.getTransactionPath()),
reply.getTransactionPath(), reply.getVersion());
@@ -754,15 +751,15 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
if(remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) {
- return new PreLithiumTransactionContextImpl(transactionPath, transactionActor, identifier,
+ return new PreLithiumTransactionContextImpl(transactionPath, transactionActor, getIdentifier(),
transactionChainId, actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion,
operationCompleter);
} else if (transactionType == TransactionType.WRITE_ONLY &&
actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
- return new WriteOnlyTransactionContextImpl(transactionActor, identifier, transactionChainId,
+ return new WriteOnlyTransactionContextImpl(transactionActor, getIdentifier(), transactionChainId,
actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
} else {
- return new TransactionContextImpl(transactionActor, identifier, transactionChainId,
+ return new TransactionContextImpl(transactionActor, getIdentifier(), transactionChainId,
actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
}
}