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=a2a7a12044a74e23bc181f65e450e14548059bec;hb=e2d9f9c57e124d46e117f17c44b77c89222fdb99;hp=ffb1ab7c55064ecf7683e1857800b256c9c82a1d;hpb=bc9665899219b2218abd8a9bee314cd07f0f84d0;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 ffb1ab7c55..a2a7a12044 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
@@ -18,26 +18,29 @@ import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Semaphore;
+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;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
-import org.opendaylight.controller.cluster.datastore.messages.DataExists;
-import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
-import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
-import org.opendaylight.controller.cluster.datastore.messages.MergeData;
-import org.opendaylight.controller.cluster.datastore.messages.ReadData;
-import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
-import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
-import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
-import org.opendaylight.controller.cluster.datastore.messages.WriteData;
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;
@@ -50,15 +53,6 @@ import scala.concurrent.Future;
import scala.concurrent.Promise;
import scala.concurrent.duration.FiniteDuration;
-import javax.annotation.concurrent.GuardedBy;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
/**
* TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
*
@@ -71,12 +65,29 @@ import java.util.concurrent.atomic.AtomicLong;
* 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
+ READ_WRITE;
+
+ // 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);
+ }
+ }
+ }
+
+ private static enum TransactionState {
+ OPEN,
+ READY,
+ CLOSED,
}
static final Mapper SAME_FAILURE_TRANSFORMER =
@@ -141,7 +152,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
remoteTransactionActors = referent.remoteTransactionActors;
remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
actorContext = referent.actorContext;
- identifier = referent.identifier;
+ identifier = referent.getIdentifier();
}
@Override
@@ -157,8 +168,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
if(remoteTransactionActorsMB.get()) {
for(ActorSelection actor : remoteTransactionActors) {
LOG.trace("Sending CloseTransaction to {}", actor);
- actorContext.sendOperationAsync(actor,
- new CloseTransaction().toSerializable());
+ actorContext.sendOperationAsync(actor, CloseTransaction.INSTANCE.toSerializable());
}
}
}
@@ -172,7 +182,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
* PhantomReference.
*/
private List remoteTransactionActors;
- private AtomicBoolean remoteTransactionActorsMB;
+ private volatile AtomicBoolean remoteTransactionActorsMB;
/**
* Stores the create transaction results per shard.
@@ -181,103 +191,86 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
private final TransactionType transactionType;
private final ActorContext actorContext;
- private final TransactionIdentifier identifier;
- private final TransactionChainProxy transactionChainProxy;
+ private final String transactionChainId;
private final SchemaContext schemaContext;
- private boolean inReadyState;
+ private TransactionState state = TransactionState.OPEN;
+
+ private volatile boolean initialized;
+ private Semaphore operationLimiter;
+ private OperationCompleter operationCompleter;
public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
- this(actorContext, transactionType, null);
+ this(actorContext, transactionType, "");
}
- public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
- TransactionChainProxy transactionChainProxy) {
+ 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,
"transactionType should not be null");
this.schemaContext = Preconditions.checkNotNull(actorContext.getSchemaContext(),
"schemaContext should not be null");
- this.transactionChainProxy = transactionChainProxy;
+ 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 = TransactionIdentifier.builder().memberName(memberName).counter(
- counter.getAndIncrement()).build();
-
- if(transactionType == TransactionType.READ_ONLY) {
- // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
- // to close the remote Tx's when this instance is no longer in use and is garbage
- // collected.
-
- remoteTransactionActors = Lists.newArrayList();
- remoteTransactionActorsMB = new AtomicBoolean();
+ return new TransactionIdentifier(memberName, counter.getAndIncrement());
+ }
- TransactionProxyCleanupPhantomReference cleanup =
- new TransactionProxyCleanupPhantomReference(this);
- phantomReferenceCache.put(cleanup, cleanup);
+ @VisibleForTesting
+ List> getRecordedOperationFutures() {
+ List> recordedOperationFutures = Lists.newArrayList();
+ for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
+ TransactionContext transactionContext = txFutureCallback.getTransactionContext();
+ if (transactionContext != null) {
+ transactionContext.copyRecordedOperationFutures(recordedOperationFutures);
+ }
}
- LOG.debug("Created txn {} of type {}", identifier, transactionType);
+ return recordedOperationFutures;
}
@VisibleForTesting
- List> getRecordedOperationFutures() {
- List> recordedOperationFutures = Lists.newArrayList();
+ boolean hasTransactionContext() {
for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
TransactionContext transactionContext = txFutureCallback.getTransactionContext();
if(transactionContext != null) {
- recordedOperationFutures.addAll(transactionContext.getRecordedOperationFutures());
+ return true;
}
}
- return recordedOperationFutures;
+ return false;
}
@Override
- public CheckedFuture>, ReadFailedException> read(
- final YangInstanceIdentifier path) {
+ public CheckedFuture>, ReadFailedException> read(final YangInstanceIdentifier path) {
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);
- TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
+ throttleOperation();
- CheckedFuture>, ReadFailedException> future;
- if(transactionContext != null) {
- future = transactionContext.readData(path);
- } else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- final SettableFuture>> proxyFuture = SettableFuture.create();
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- Futures.addCallback(transactionContext.readData(path),
- new FutureCallback>>() {
- @Override
- public void onSuccess(Optional> data) {
- proxyFuture.set(data);
- }
-
- @Override
- public void onFailure(Throwable t) {
- proxyFuture.setException(t);
- }
- });
- }
- });
+ final SettableFuture>> proxyFuture = SettableFuture.create();
- future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
- }
+ TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.readData(path, proxyFuture);
+ }
+ });
- return future;
+ return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
}
@Override
@@ -286,70 +279,75 @@ 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);
- TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
+ throttleOperation();
- CheckedFuture future;
- if(transactionContext != null) {
- future = transactionContext.dataExists(path);
- } else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- final SettableFuture proxyFuture = SettableFuture.create();
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- Futures.addCallback(transactionContext.dataExists(path),
- new FutureCallback() {
- @Override
- public void onSuccess(Boolean exists) {
- proxyFuture.set(exists);
- }
-
- @Override
- public void onFailure(Throwable t) {
- proxyFuture.setException(t);
- }
- });
- }
- });
+ final SettableFuture proxyFuture = SettableFuture.create();
- future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
- }
+ TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.dataExists(path, proxyFuture);
+ }
+ });
- return future;
+ return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
}
private void checkModificationState() {
Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
"Modification operation on read-only transaction is not allowed");
- Preconditions.checkState(!inReadyState,
+ Preconditions.checkState(state == TransactionState.OPEN,
"Transaction is sealed - further modifications are not allowed");
}
+ private void throttleOperation() {
+ throttleOperation(1);
+ }
+
+ private void throttleOperation(int acquirePermits) {
+ if(!initialized) {
+ // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
+ operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
+ operationCompleter = new OperationCompleter(operationLimiter);
+
+ // Make sure we write this last because it's volatile and will also publish the non-volatile writes
+ // above as well so they'll be visible to other threads.
+ initialized = true;
+ }
+
+ try {
+ if(!operationLimiter.tryAcquire(acquirePermits,
+ actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
+ LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
+ }
+ } catch (InterruptedException e) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Interrupted when trying to acquire operation permit for transaction " + getIdentifier().toString(), e);
+ } else {
+ LOG.warn("Interrupted when trying to acquire operation permit for transaction {}", getIdentifier());
+ }
+ }
+ }
+
@Override
public void write(final YangInstanceIdentifier path, final NormalizedNode, ?> data) {
checkModificationState();
- LOG.debug("Tx {} write {}", identifier, path);
+ LOG.debug("Tx {} write {}", getIdentifier(), path);
+
+ throttleOperation();
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
- if(transactionContext != null) {
- transactionContext.writeData(path, data);
- } else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- transactionContext.writeData(path, data);
- }
- });
- }
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.writeData(path, data);
+ }
+ });
}
@Override
@@ -357,22 +355,17 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} merge {}", identifier, path);
+ LOG.debug("Tx {} merge {}", getIdentifier(), path);
+
+ throttleOperation();
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
- if(transactionContext != null) {
- transactionContext.mergeData(path, data);
- } else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- transactionContext.mergeData(path, data);
- }
- });
- }
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.mergeData(path, data);
+ }
+ });
}
@Override
@@ -380,91 +373,111 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
checkModificationState();
- LOG.debug("Tx {} delete {}", identifier, path);
+ LOG.debug("Tx {} delete {}", getIdentifier(), path);
+
+ throttleOperation();
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
- if(transactionContext != null) {
- transactionContext.deleteData(path);
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.deleteData(path);
+ }
+ });
+ }
+
+ private boolean seal(final TransactionState newState) {
+ if (state == TransactionState.OPEN) {
+ state = newState;
+ return true;
} else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- transactionContext.deleteData(path);
- }
- });
+ return false;
}
}
@Override
public DOMStoreThreePhaseCommitCohort ready() {
+ Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
+ "Read-only transactions cannot be readied");
- checkModificationState();
-
- inReadyState = true;
+ final boolean success = seal(TransactionState.READY);
+ Preconditions.checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
- LOG.debug("Tx {} Readying {} transactions for commit", identifier,
+ LOG.debug("Tx {} Readying {} transactions for commit", getIdentifier(),
txFutureCallbackMap.size());
+ if (txFutureCallbackMap.isEmpty()) {
+ onTransactionReady(Collections.>emptyList());
+ TransactionRateLimitingCallback.adjustRateLimitForUnusedTransaction(actorContext);
+ return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
+ }
+
+ throttleOperation(txFutureCallbackMap.size());
+
List> cohortFutures = Lists.newArrayList();
for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
- LOG.debug("Tx {} Readying transaction for shard {}", identifier,
- txFutureCallback.getShardName());
+ LOG.debug("Tx {} Readying transaction for shard {} chain {}", getIdentifier(),
+ txFutureCallback.getShardName(), transactionChainId);
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
- if(transactionContext != null) {
- cohortFutures.add(transactionContext.readyTransaction());
+ final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
+ final Future future;
+ if (transactionContext != null) {
+ // avoid the creation of a promise and a TransactionOperation
+ future = transactionContext.readyTransaction();
} else {
- // The shard Tx hasn't been created yet so create a promise to ready the Tx later
- // after it's created.
- final Promise cohortPromise = akka.dispatch.Futures.promise();
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
+ final Promise promise = akka.dispatch.Futures.promise();
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
public void invoke(TransactionContext transactionContext) {
- cohortPromise.completeWith(transactionContext.readyTransaction());
+ promise.completeWith(transactionContext.readyTransaction());
}
});
-
- cohortFutures.add(cohortPromise.future());
+ future = promise.future();
}
- }
- if(transactionChainProxy != null){
- transactionChainProxy.onTransactionReady(cohortFutures);
+ cohortFutures.add(future);
}
+ onTransactionReady(cohortFutures);
+
return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
- identifier.toString());
+ getIdentifier().toString());
}
- @Override
- public Object getIdentifier() {
- return this.identifier;
+ /**
+ * Method for derived classes to be notified when the transaction has been readied.
+ *
+ * @param cohortFutures the cohort Futures for each shard transaction.
+ */
+ protected void onTransactionReady(List> cohortFutures) {
}
@Override
public void close() {
- for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
- TransactionContext transactionContext = txFutureCallback.getTransactionContext();
- if(transactionContext != null) {
- transactionContext.closeTransaction();
- } else {
- txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- transactionContext.closeTransaction();
- }
- });
+ if (!seal(TransactionState.CLOSED)) {
+ if (state == TransactionState.CLOSED) {
+ // Idempotent no-op as per AutoCloseable recommendation
+ return;
}
+
+ throw new IllegalStateException(String.format("Transaction %s is ready, it cannot be closed",
+ getIdentifier()));
+ }
+
+ for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.closeTransaction();
+ }
+ });
}
txFutureCallbackMap.clear();
- if(transactionType == TransactionType.READ_ONLY) {
+ if(remoteTransactionActorsMB != null) {
remoteTransactionActors.clear();
remoteTransactionActorsMB.set(true);
}
@@ -474,14 +487,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);
@@ -490,29 +506,23 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
@Override
public void onComplete(Throwable failure, ActorSelection primaryShard) {
if(failure != null) {
- newTxFutureCallback.onComplete(failure, null);
+ newTxFutureCallback.createTransactionContext(failure, null);
} else {
newTxFutureCallback.setPrimaryShard(primaryShard);
}
}
- }, actorContext.getActorSystem().dispatcher());
+ }, actorContext.getClientDispatcher());
}
return txFutureCallback;
}
public String getTransactionChainId() {
- if(transactionChainProxy == null){
- return "";
- }
- return transactionChainProxy.getTransactionChainId();
+ return transactionChainId;
}
- /**
- * Interface for a transaction operation to be invoked later.
- */
- private static interface TransactionOperation {
- void invoke(TransactionContext transactionContext);
+ protected ActorContext getActorContext() {
+ return actorContext;
}
/**
@@ -566,37 +576,68 @@ 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",
+ 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.
+ executeTxOperatonsOnComplete(createValidTransactionContext(this.primaryShard,
+ this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION));
+ } else {
+ tryCreateTransaction();
+ }
}
/**
* Adds a TransactionOperation to be executed after the CreateTransaction completes.
*/
void addTxOperationOnComplete(TransactionOperation operation) {
+ 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);
- } else {
- operation.invoke(transactionContext);
}
}
+
+ if(invokeOperation) {
+ operation.invoke(transactionContext);
+ }
+ }
+
+ void enqueueTransactionOperation(final TransactionOperation op) {
+
+ if (transactionContext != null) {
+ op.invoke(transactionContext);
+ } else {
+ // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
+ // callback to be executed after the Tx is created.
+ addTxOperationOnComplete(op);
+ }
}
/**
* Performs a CreateTransaction try async.
*/
private void tryCreateTransaction() {
- Future