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%2FShardTransaction.java;h=dfd60afa21b84c7462a5459bc8752d4548276e57;hb=b5cb353e3553a39f576c284119af75ffa5ea66a9;hp=2421cce26c0afa6f7a7c7cc3cf6b5da5830f6fd0;hpb=4d1709660b7af992d4c382a2a38debb5c7d64fb9;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java index 2421cce26c..dfd60afa21 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java @@ -15,6 +15,7 @@ import akka.actor.ReceiveTimeout; import akka.japi.Creator; import com.google.common.base.Optional; import com.google.common.base.Preconditions; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering; import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; @@ -22,6 +23,7 @@ import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.DataExists; import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply; +import org.opendaylight.controller.cluster.datastore.messages.PersistAbortTransactionPayload; import org.opendaylight.controller.cluster.datastore.messages.ReadData; import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; @@ -29,31 +31,23 @@ import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; /** - * The ShardTransaction Actor represents a remote transaction - *

- * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction - *

- *

- * Handles Messages
- * ----------------
- *

  • {@link org.opendaylight.controller.cluster.datastore.messages.ReadData} - *
  • {@link org.opendaylight.controller.cluster.datastore.messages.CloseTransaction} - *

    + * The ShardTransaction Actor represents a remote transaction that delegates all actions to DOMDataReadWriteTransaction. */ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering { private final ActorRef shardActor; private final ShardStats shardStats; - private final TransactionIdentifier transactionID; + private final TransactionIdentifier transactionId; - protected ShardTransaction(ActorRef shardActor, ShardStats shardStats, TransactionIdentifier transactionID) { + protected ShardTransaction(final ActorRef shardActor, final ShardStats shardStats, + final TransactionIdentifier transactionId) { super("shard-tx"); //actor name override used for metering. This does not change the "real" actor name this.shardActor = shardActor; this.shardStats = shardStats; - this.transactionID = Preconditions.checkNotNull(transactionID); + this.transactionId = Preconditions.checkNotNull(transactionId); } - public static Props props(TransactionType type, AbstractShardDataTreeTransaction transaction, ActorRef shardActor, - DatastoreContext datastoreContext, ShardStats shardStats) { + public static Props props(final TransactionType type, final AbstractShardDataTreeTransaction transaction, + final ActorRef shardActor, final DatastoreContext datastoreContext, final ShardStats shardStats) { return Props.create(new ShardTransactionCreator(type, transaction, shardActor, datastoreContext, shardStats)); } @@ -63,16 +57,16 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering return shardActor; } - protected final TransactionIdentifier getTransactionID() { - return transactionID; + protected final TransactionIdentifier getTransactionId() { + return transactionId; } @Override - public void handleReceive(Object message) { + public void handleReceive(final Object message) { if (CloseTransaction.isSerializedType(message)) { closeTransaction(true); } else if (message instanceof ReceiveTimeout) { - LOG.debug("Got ReceiveTimeout for inactivity - closing transaction {}", transactionID); + LOG.debug("Got ReceiveTimeout for inactivity - closing transaction {}", transactionId); closeTransaction(false); } else { unknownMessage(message); @@ -83,26 +77,28 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering return true; } - private void closeTransaction(boolean sendReply) { - getDOMStoreTransaction().abort(); + private void closeTransaction(final boolean sendReply) { + getDOMStoreTransaction().abortFromTransactionActor(); + shardActor.tell(new PersistAbortTransactionPayload(transactionId), ActorRef.noSender()); - if(sendReply && returnCloseTransactionReply()) { + if (sendReply && returnCloseTransactionReply()) { getSender().tell(new CloseTransactionReply(), getSelf()); } getSelf().tell(PoisonPill.getInstance(), getSelf()); } - private boolean checkClosed(AbstractShardDataTreeTransaction transaction) { + private boolean checkClosed(final AbstractShardDataTreeTransaction transaction) { final boolean ret = transaction.isClosed(); if (ret) { shardStats.incrementFailedReadTransactionsCount(); - getSender().tell(new akka.actor.Status.Failure(new ReadFailedException("Transaction is closed")), getSelf()); + getSender().tell(new akka.actor.Status.Failure(new ReadFailedException("Transaction is closed")), + getSelf()); } return ret; } - protected void readData(AbstractShardDataTreeTransaction transaction, ReadData message) { + protected void readData(final AbstractShardDataTreeTransaction transaction, final ReadData message) { if (checkClosed(transaction)) { return; } @@ -113,7 +109,7 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering sender().tell(readDataReply.toSerializable(), self()); } - protected void dataExists(AbstractShardDataTreeTransaction transaction, DataExists message) { + protected void dataExists(final AbstractShardDataTreeTransaction transaction, final DataExists message) { if (checkClosed(transaction)) { return; } @@ -123,6 +119,8 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering getSender().tell(new DataExistsReply(exists, message.getVersion()).toSerializable(), getSelf()); } + @SuppressFBWarnings(value = "SE_BAD_FIELD", justification = "Some fields are not Serializable but we don't " + + "create remote instances of this actor and thus don't need it to be Serializable.") private static class ShardTransactionCreator implements Creator { private static final long serialVersionUID = 1L; @@ -133,8 +131,8 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering final ShardStats shardStats; final TransactionType type; - ShardTransactionCreator(TransactionType type, AbstractShardDataTreeTransaction transaction, ActorRef shardActor, - DatastoreContext datastoreContext, ShardStats shardStats) { + ShardTransactionCreator(final TransactionType type, final AbstractShardDataTreeTransaction transaction, + final ActorRef shardActor, final DatastoreContext datastoreContext, final ShardStats shardStats) { this.transaction = Preconditions.checkNotNull(transaction); this.shardActor = shardActor; this.shardStats = shardStats; @@ -146,17 +144,19 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering public ShardTransaction create() throws Exception { final ShardTransaction tx; switch (type) { - case READ_ONLY: - tx = new ShardReadTransaction(transaction, shardActor, shardStats); - break; - case READ_WRITE: - tx = new ShardReadWriteTransaction((ReadWriteShardDataTreeTransaction)transaction, shardActor, shardStats); - break; - case WRITE_ONLY: - tx = new ShardWriteTransaction((ReadWriteShardDataTreeTransaction)transaction, shardActor, shardStats); - break; - default: - throw new IllegalArgumentException("Unhandled transaction type " + type); + case READ_ONLY: + tx = new ShardReadTransaction(transaction, shardActor, shardStats); + break; + case READ_WRITE: + tx = new ShardReadWriteTransaction((ReadWriteShardDataTreeTransaction)transaction, shardActor, + shardStats); + break; + case WRITE_ONLY: + tx = new ShardWriteTransaction((ReadWriteShardDataTreeTransaction)transaction, shardActor, + shardStats); + break; + default: + throw new IllegalArgumentException("Unhandled transaction type " + type); } tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());