package org.opendaylight.controller.cluster.datastore;
-import akka.actor.ActorPath;
import akka.actor.ActorSelection;
import akka.dispatch.OnComplete;
for(ActorSelection actor : remoteTransactionActors) {
LOG.trace("Sending CloseTransaction to {}", actor);
actorContext.sendOperationAsync(actor,
- new CloseTransaction().toSerializable());
+ new CloseTransaction().toSerializable());
}
}
}
LOG.debug("Tx {} Trying to get {} transactions ready for commit", identifier,
remoteTransactionPaths.size());
}
- List<Future<ActorPath>> cohortPathFutures = Lists.newArrayList();
+ List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
LOG.debug("Tx {} Readying transaction for shard {}", identifier,
transactionContext.getShardName());
}
- cohortPathFutures.add(transactionContext.readyTransaction());
+ cohortFutures.add(transactionContext.readyTransaction());
}
if(transactionChainProxy != null){
- transactionChainProxy.onTransactionReady(cohortPathFutures);
+ transactionChainProxy.onTransactionReady(cohortFutures);
}
- return new ThreePhaseCommitCohortProxy(actorContext, cohortPathFutures,
+ return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
identifier.toString());
}
}
Object response = actorContext.executeOperation(primaryShard.get(),
- new CreateTransaction(identifier.toString(), this.transactionType.ordinal(),
- getTransactionChainId()).toSerializable());
+ new CreateTransaction(identifier.toString(), this.transactionType.ordinal(),
+ getTransactionChainId()).toSerializable());
if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
CreateTransactionReply reply =
CreateTransactionReply.fromSerializable(response);
remoteTransactionActorsMB.set(true);
}
+ // TxActor is always created where the leader of the shard is.
+ // Check if TxActor is created in the same node
+ boolean isTxActorLocal = actorContext.isLocalPath(transactionPath);
+
transactionContext = new TransactionContextImpl(shardName, transactionPath,
- transactionActor, identifier, actorContext, schemaContext);
+ transactionActor, identifier, actorContext, schemaContext, isTxActorLocal);
remoteTransactionPaths.put(shardName, transactionContext);
} else {
void closeTransaction();
- Future<ActorPath> readyTransaction();
+ Future<ActorSelection> readyTransaction();
void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
private final SchemaContext schemaContext;
private final String actorPath;
private final ActorSelection actor;
+ private final boolean isTxActorLocal;
private TransactionContextImpl(String shardName, String actorPath,
ActorSelection actor, TransactionIdentifier identifier, ActorContext actorContext,
- SchemaContext schemaContext) {
+ SchemaContext schemaContext, boolean isTxActorLocal) {
super(shardName, identifier);
this.actorPath = actorPath;
this.actor = actor;
this.actorContext = actorContext;
this.schemaContext = schemaContext;
+ this.isTxActorLocal = isTxActorLocal;
}
private ActorSelection getActor() {
return actor;
}
- private String getResolvedCohortPath(String cohortPath) {
- return actorContext.resolvePath(actorPath, cohortPath);
- }
-
@Override
public void closeTransaction() {
if(LOG.isDebugEnabled()) {
}
@Override
- public Future<ActorPath> readyTransaction() {
+ public Future<ActorSelection> readyTransaction() {
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending",
identifier, recordedOperationFutures.size());
}
// Send the ReadyTransaction message to the Tx actor.
+ ReadyTransaction readyTransaction = new ReadyTransaction();
final Future<Object> replyFuture = actorContext.executeOperationAsync(getActor(),
- new ReadyTransaction().toSerializable());
+ isTxActorLocal ? readyTransaction : readyTransaction.toSerializable());
// Combine all the previously recorded put/merge/delete operation reply Futures and the
// ReadyTransactionReply Future into one Future. If any one fails then the combined
// Transform the combined Future into a Future that returns the cohort actor path from
// the ReadyTransactionReply. That's the end result of the ready operation.
- return combinedFutures.transform(new AbstractFunction1<Iterable<Object>, ActorPath>() {
+ return combinedFutures.transform(new AbstractFunction1<Iterable<Object>, ActorSelection>() {
@Override
- public ActorPath apply(Iterable<Object> notUsed) {
+ public ActorSelection apply(Iterable<Object> notUsed) {
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded",
identifier);
// Note the Future get call here won't block as it's complete.
Object serializedReadyReply = replyFuture.value().get().get();
- if(serializedReadyReply.getClass().equals(
- ReadyTransactionReply.SERIALIZABLE_CLASS)) {
- ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(
- actorContext.getActorSystem(), serializedReadyReply);
+ if (serializedReadyReply instanceof ReadyTransactionReply) {
+ return actorContext.actorSelection(((ReadyTransactionReply)serializedReadyReply).getCohortPath());
- String resolvedCohortPath = getResolvedCohortPath(
- reply.getCohortPath().toString());
+ } else if(serializedReadyReply.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)) {
+ ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(serializedReadyReply);
+ return actorContext.actorSelection(reply.getCohortPath());
- if(LOG.isDebugEnabled()) {
- LOG.debug("Tx {} readyTransaction: resolved cohort path {}",
- identifier, resolvedCohortPath);
- }
- return actorContext.actorFor(resolvedCohortPath);
} else {
// Throwing an exception here will fail the Future.
-
throw new IllegalArgumentException(String.format("Invalid reply type {}",
serializedReadyReply.getClass()));
}
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} deleteData called path = {}", identifier, path);
}
+
+ DeleteData deleteData = new DeleteData(path);
recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
- new DeleteData(path).toSerializable()));
+ isTxActorLocal ? deleteData : deleteData.toSerializable()));
}
@Override
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} mergeData called path = {}", identifier, path);
}
+
+ MergeData mergeData = new MergeData(path, data, schemaContext);
recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
- new MergeData(path, data, schemaContext).toSerializable()));
+ isTxActorLocal ? mergeData : mergeData.toSerializable()));
}
@Override
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} writeData called path = {}", identifier, path);
}
+
+ WriteData writeData = new WriteData(path, data, schemaContext);
recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
- new WriteData(path, data, schemaContext).toSerializable()));
+ isTxActorLocal ? writeData : writeData.toSerializable()));
}
@Override
Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(
Lists.newArrayList(recordedOperationFutures),
actorContext.getActorSystem().dispatcher());
+
OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
@Override
public void onComplete(Throwable failure, Iterable<Object> notUsed)
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} read operation succeeded", identifier, failure);
}
- if (readResponse.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
- ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext,
- path, readResponse);
- if (reply.getNormalizedNode() == null) {
- returnFuture.set(Optional.<NormalizedNode<?, ?>>absent());
- } else {
- returnFuture.set(Optional.<NormalizedNode<?, ?>>of(
- reply.getNormalizedNode()));
- }
+
+ if (readResponse instanceof ReadDataReply) {
+ ReadDataReply reply = (ReadDataReply) readResponse;
+ returnFuture.set(Optional.<NormalizedNode<?, ?>>fromNullable(reply.getNormalizedNode()));
+
+ } else if (readResponse.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
+ ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext, path, readResponse);
+ returnFuture.set(Optional.<NormalizedNode<?, ?>>fromNullable(reply.getNormalizedNode()));
+
} else {
returnFuture.setException(new ReadFailedException(
- "Invalid response reading data for path " + path));
+ "Invalid response reading data for path " + path));
}
}
}
};
+ ReadData readData = new ReadData(path);
Future<Object> readFuture = actorContext.executeOperationAsync(getActor(),
- new ReadData(path).toSerializable());
+ isTxActorLocal ? readData : readData.toSerializable());
+
readFuture.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
}
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} dataExists operation succeeded", identifier, failure);
}
- if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) {
- returnFuture.set(Boolean.valueOf(DataExistsReply.
- fromSerializable(response).exists()));
+
+ if (response instanceof DataExistsReply) {
+ returnFuture.set(Boolean.valueOf(((DataExistsReply) response).exists()));
+
+ } else if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) {
+ returnFuture.set(Boolean.valueOf(DataExistsReply.fromSerializable(response).exists()));
+
} else {
returnFuture.setException(new ReadFailedException(
"Invalid response checking exists for path " + path));
}
};
+ DataExists dataExists = new DataExists(path);
Future<Object> future = actorContext.executeOperationAsync(getActor(),
- new DataExists(path).toSerializable());
+ isTxActorLocal ? dataExists : dataExists.toSerializable());
+
future.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
}
}
}
@Override
- public Future<ActorPath> readyTransaction() {
+ public Future<ActorSelection> readyTransaction() {
if(LOG.isDebugEnabled()) {
LOG.debug("Tx {} readyTransaction called", identifier);
}
Code Review / controller.git / blobdiff