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.ListenableFuture;
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.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.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;
* shards will be executed.
* </p>
*/
-public class TransactionProxy implements DOMStoreReadWriteTransaction {
+public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIdentifier> 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);
+ }
+ }
}
static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
remoteTransactionActors = referent.remoteTransactionActors;
remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
actorContext = referent.actorContext;
- identifier = referent.identifier;
+ identifier = referent.getIdentifier();
}
@Override
* PhantomReference.
*/
private List<ActorSelection> remoteTransactionActors;
- private AtomicBoolean remoteTransactionActorsMB;
+ private volatile AtomicBoolean remoteTransactionActorsMB;
/**
* Stores the create transaction results per shard.
private final TransactionType transactionType;
private final ActorContext actorContext;
- private final TransactionIdentifier identifier;
private final String transactionChainId;
private final SchemaContext schemaContext;
private boolean inReadyState;
- private final Semaphore operationLimiter;
- private final OperationCompleter operationCompleter;
+
+ private volatile boolean initialized;
+ private Semaphore operationLimiter;
+ private OperationCompleter operationCompleter;
public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
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,
"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 = 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();
-
- TransactionProxyCleanupPhantomReference cleanup =
- new TransactionProxyCleanupPhantomReference(this);
- phantomReferenceCache.put(cleanup, cleanup);
- }
-
- // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
- this.operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
- this.operationCompleter = new OperationCompleter(operationLimiter);
-
- LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
+ return new TransactionIdentifier(memberName, counter.getAndIncrement());
}
@VisibleForTesting
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();
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();
}
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)){
+ if(!operationLimiter.tryAcquire(acquirePermits,
+ actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
}
} catch (InterruptedException e) {
checkModificationState();
- LOG.debug("Tx {} write {}", identifier, path);
+ LOG.debug("Tx {} write {}", getIdentifier(), path);
throttleOperation();
checkModificationState();
- LOG.debug("Tx {} merge {}", identifier, path);
+ LOG.debug("Tx {} merge {}", getIdentifier(), path);
throttleOperation();
checkModificationState();
- LOG.debug("Tx {} delete {}", identifier, path);
+ LOG.debug("Tx {} delete {}", getIdentifier(), path);
throttleOperation();
checkModificationState();
- throttleOperation(txFutureCallbackMap.size());
-
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.<Future<ActorSelection>>emptyList());
+ TransactionRateLimitingCallback.adjustRateLimitForUnusedTransaction(actorContext);
+ return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
+ }
+
+ throttleOperation(txFutureCallbackMap.size());
+
List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
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();
onTransactionReady(cohortFutures);
return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
- identifier.toString());
+ getIdentifier().toString());
}
/**
protected void onTransactionReady(List<Future<ActorSelection>> 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<Object> sendCreateTransaction(ActorSelection shard,
- Object serializedCreateMessage) {
- return actorContext.executeOperationAsync(shard, serializedCreateMessage);
- }
-
- @Override
- public Object getIdentifier() {
- return this.identifier;
- }
-
@Override
public void close() {
for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
txFutureCallbackMap.clear();
- if(transactionType == TransactionType.READ_ONLY) {
+ if(remoteTransactionActorsMB != null) {
remoteTransactionActors.clear();
remoteTransactionActorsMB.set(true);
}
return ShardStrategyFactory.getStrategy(path).findShard(path);
}
+ protected Future<ActorSelection> 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<ActorSelection> findPrimaryFuture = actorContext.findPrimaryShardAsync(shardName);
+ Future<ActorSelection> findPrimaryFuture = sendFindPrimaryShardAsync(shardName);
- final TransactionFutureCallback newTxFutureCallback =
- new TransactionFutureCallback(shardName);
+ final TransactionFutureCallback newTxFutureCallback = new TransactionFutureCallback(shardName);
txFutureCallback = newTxFutureCallback;
txFutureCallbackMap.put(shardName, txFutureCallback);
@Override
public void onComplete(Throwable failure, ActorSelection primaryShard) {
if(failure != null) {
- newTxFutureCallback.onComplete(failure, null);
+ newTxFutureCallback.createTransactionContext(failure, null);
} else {
newTxFutureCallback.setPrimaryShard(primaryShard);
}
* 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();
+ }
}
/**
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);
* Performs a CreateTransaction try async.
*/
private void tryCreateTransaction() {
- Future<Object> createTxFuture = sendCreateTransaction(primaryShard,
- new CreateTransaction(identifier.toString(),
- TransactionProxy.this.transactionType.ordinal(),
- getTransactionChainId()).toSerializable());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShard);
+ }
+
+ Object serializedCreateMessage = new CreateTransaction(getIdentifier().toString(),
+ TransactionProxy.this.transactionType.ordinal(),
+ getTransactionChainId()).toSerializable();
+
+ Future<Object> createTxFuture = actorContext.executeOperationAsync(primaryShard, serializedCreateMessage);
createTxFuture.onComplete(this, actorContext.getClientDispatcher());
}
// 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() {
}
}
+ createTransactionContext(failure, response);
+ }
+
+ private void createTransactionContext(Throwable failure, Object response) {
+ // Mainly checking for state violation here to perform a volatile read of "initialized" to
+ // ensure updates to operationLimter et al are visible to this thread (ie we're doing
+ // "piggy-back" synchronization here).
+ Preconditions.checkState(initialized, "Tx was not propertly initialized.");
+
// Create the TransactionContext from the response or failure. Store the new
// TransactionContext locally until we've completed invoking the
// TransactionOperations. This avoids thread timing issues which could cause
// 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", 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));
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);
}
private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
- String transactionPath = reply.getTransactionPath();
+ LOG.debug("Tx {} Received {}", getIdentifier(), reply);
- LOG.debug("Tx {} Received transaction actor path {}", identifier, transactionPath);
+ return createValidTransactionContext(actorContext.actorSelection(reply.getTransactionPath()),
+ reply.getTransactionPath(), reply.getVersion());
+ }
- ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
+ 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
+ // to close the remote Tx's when this instance is no longer in use and is garbage
+ // collected.
+
+ if(remoteTransactionActorsMB == null) {
+ remoteTransactionActors = Lists.newArrayList();
+ remoteTransactionActorsMB = new AtomicBoolean();
+
+ TransactionProxyCleanupPhantomReference cleanup =
+ new TransactionProxyCleanupPhantomReference(TransactionProxy.this);
+ phantomReferenceCache.put(cleanup, cleanup);
+ }
+
// Add the actor to the remoteTransactionActors list for access by the
// cleanup PhantonReference.
remoteTransactionActors.add(transactionActor);
// Check if TxActor is created in the same node
boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
- return new TransactionContextImpl(transactionPath, transactionActor, identifier,
- actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter);
+ if(remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) {
+ 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, getIdentifier(), transactionChainId,
+ actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
+ } else {
+ return new TransactionContextImpl(transactionActor, getIdentifier(), transactionChainId,
+ actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
+ }
+ }
+ }
+
+ private static class NoOpDOMStoreThreePhaseCommitCohort implements DOMStoreThreePhaseCommitCohort {
+ static NoOpDOMStoreThreePhaseCommitCohort INSTANCE = new NoOpDOMStoreThreePhaseCommitCohort();
+
+ private static final ListenableFuture<Void> IMMEDIATE_VOID_SUCCESS =
+ com.google.common.util.concurrent.Futures.immediateFuture(null);
+ private static final ListenableFuture<Boolean> IMMEDIATE_BOOLEAN_SUCCESS =
+ com.google.common.util.concurrent.Futures.immediateFuture(Boolean.TRUE);
+
+ private NoOpDOMStoreThreePhaseCommitCohort() {
+ }
+
+ @Override
+ public ListenableFuture<Boolean> canCommit() {
+ return IMMEDIATE_BOOLEAN_SUCCESS;
+ }
+
+ @Override
+ public ListenableFuture<Void> preCommit() {
+ return IMMEDIATE_VOID_SUCCESS;
+ }
+
+ @Override
+ public ListenableFuture<Void> abort() {
+ return IMMEDIATE_VOID_SUCCESS;
+ }
+
+ @Override
+ public ListenableFuture<Void> commit() {
+ return IMMEDIATE_VOID_SUCCESS;
}
}
}