X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardWriteTransaction.java;h=69a696f2942ecd3035dc9f52ef2b9831188f8e9a;hp=424ab2052c13c6dd545d7aac1f53b4a5a054daae;hb=56c1339ee7dbd85bc567fc44f21ecfd322c9e803;hpb=81d25b3c5f4a040b5f11475752a26712b46f5c0b diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardWriteTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardWriteTransaction.java index 424ab2052c..69a696f294 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardWriteTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardWriteTransaction.java @@ -15,11 +15,13 @@ import akka.actor.PoisonPill; import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; +import org.opendaylight.controller.cluster.datastore.messages.DataExists; import org.opendaylight.controller.cluster.datastore.messages.DeleteData; import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply; import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction; import org.opendaylight.controller.cluster.datastore.messages.MergeData; import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply; +import org.opendaylight.controller.cluster.datastore.messages.ReadData; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; import org.opendaylight.controller.cluster.datastore.messages.WriteData; import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply; @@ -30,8 +32,6 @@ import org.opendaylight.controller.cluster.datastore.modification.Modification; import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; /** * @author: syedbahm @@ -42,16 +42,16 @@ public class ShardWriteTransaction extends ShardTransaction { private final MutableCompositeModification compositeModification = new MutableCompositeModification(); private int totalBatchedModificationsReceived; private Exception lastBatchedModificationsException; - private final DOMStoreWriteTransaction transaction; + private final ReadWriteShardDataTreeTransaction transaction; - public ShardWriteTransaction(DOMStoreWriteTransaction transaction, ActorRef shardActor, + public ShardWriteTransaction(ReadWriteShardDataTreeTransaction transaction, ActorRef shardActor, ShardStats shardStats, String transactionID, short clientTxVersion) { super(shardActor, shardStats, transactionID, clientTxVersion); this.transaction = transaction; } @Override - protected DOMStoreTransaction getDOMStoreTransaction() { + protected ReadWriteShardDataTreeTransaction getDOMStoreTransaction() { return transaction; } @@ -61,17 +61,17 @@ public class ShardWriteTransaction extends ShardTransaction { if (message instanceof BatchedModifications) { batchedModifications((BatchedModifications)message); } else if (message instanceof ReadyTransaction) { - readyTransaction(transaction, !SERIALIZED_REPLY, false); + readyTransaction(!SERIALIZED_REPLY, false); } else if(ReadyTransaction.SERIALIZABLE_CLASS.equals(message.getClass())) { - readyTransaction(transaction, SERIALIZED_REPLY, false); + readyTransaction(SERIALIZED_REPLY, false); } else if(WriteData.isSerializedType(message)) { - writeData(transaction, WriteData.fromSerializable(message), SERIALIZED_REPLY); + writeData(WriteData.fromSerializable(message), SERIALIZED_REPLY); } else if(MergeData.isSerializedType(message)) { - mergeData(transaction, MergeData.fromSerializable(message), SERIALIZED_REPLY); + mergeData(MergeData.fromSerializable(message), SERIALIZED_REPLY); } else if(DeleteData.isSerializedType(message)) { - deleteData(transaction, DeleteData.fromSerializable(message), SERIALIZED_REPLY); + deleteData(DeleteData.fromSerializable(message), SERIALIZED_REPLY); } else if (message instanceof GetCompositedModification) { // This is here for testing only @@ -82,10 +82,17 @@ public class ShardWriteTransaction extends ShardTransaction { } private void batchedModifications(BatchedModifications batched) { + if (checkClosed()) { + if (batched.isReady()) { + getSelf().tell(PoisonPill.getInstance(), getSelf()); + } + return; + } + try { for(Modification modification: batched.getModifications()) { compositeModification.addModification(modification); - modification.apply(transaction); + modification.apply(transaction.getSnapshot()); } totalBatchedModificationsReceived++; @@ -100,7 +107,7 @@ public class ShardWriteTransaction extends ShardTransaction { totalBatchedModificationsReceived, batched.getTotalMessagesSent())); } - readyTransaction(transaction, false, batched.isDoCommitOnReady()); + readyTransaction(false, batched.isDoCommitOnReady()); } else { getSender().tell(new BatchedModificationsReply(batched.getModifications().size()), getSelf()); } @@ -114,14 +121,33 @@ public class ShardWriteTransaction extends ShardTransaction { } } - private void writeData(DOMStoreWriteTransaction transaction, WriteData message, - boolean returnSerialized) { + protected final void dataExists(DataExists message, final boolean returnSerialized) { + super.dataExists(transaction, message, returnSerialized); + } + + protected final void readData(ReadData message, final boolean returnSerialized) { + super.readData(transaction, message, returnSerialized); + } + + private boolean checkClosed() { + if (transaction.isClosed()) { + getSender().tell(new akka.actor.Status.Failure(new IllegalStateException("Transaction is closed, no modifications allowed")), getSelf()); + return true; + } else { + return false; + } + } + + private void writeData(WriteData message, boolean returnSerialized) { LOG.debug("writeData at path : {}", message.getPath()); + if (checkClosed()) { + return; + } compositeModification.addModification( new WriteModification(message.getPath(), message.getData())); try { - transaction.write(message.getPath(), message.getData()); + transaction.getSnapshot().write(message.getPath(), message.getData()); WriteDataReply writeDataReply = WriteDataReply.INSTANCE; getSender().tell(returnSerialized ? writeDataReply.toSerializable(message.getVersion()) : writeDataReply, getSelf()); @@ -130,15 +156,17 @@ public class ShardWriteTransaction extends ShardTransaction { } } - private void mergeData(DOMStoreWriteTransaction transaction, MergeData message, - boolean returnSerialized) { + private void mergeData(MergeData message, boolean returnSerialized) { LOG.debug("mergeData at path : {}", message.getPath()); + if (checkClosed()) { + return; + } compositeModification.addModification( new MergeModification(message.getPath(), message.getData())); try { - transaction.merge(message.getPath(), message.getData()); + transaction.getSnapshot().merge(message.getPath(), message.getData()); MergeDataReply mergeDataReply = MergeDataReply.INSTANCE; getSender().tell(returnSerialized ? mergeDataReply.toSerializable(message.getVersion()) : mergeDataReply, getSelf()); @@ -147,23 +175,24 @@ public class ShardWriteTransaction extends ShardTransaction { } } - private void deleteData(DOMStoreWriteTransaction transaction, DeleteData message, - boolean returnSerialized) { + private void deleteData(DeleteData message, boolean returnSerialized) { LOG.debug("deleteData at path : {}", message.getPath()); + if (checkClosed()) { + return; + } compositeModification.addModification(new DeleteModification(message.getPath())); try { - transaction.delete(message.getPath()); + transaction.getSnapshot().delete(message.getPath()); DeleteDataReply deleteDataReply = DeleteDataReply.INSTANCE; getSender().tell(returnSerialized ? deleteDataReply.toSerializable(message.getVersion()) : deleteDataReply, getSelf()); - }catch(Exception e){ + } catch(Exception e) { getSender().tell(new akka.actor.Status.Failure(e), getSelf()); } } - private void readyTransaction(DOMStoreWriteTransaction transaction, boolean returnSerialized, - boolean doImmediateCommit) { + private void readyTransaction(boolean returnSerialized, boolean doImmediateCommit) { String transactionID = getTransactionID(); LOG.debug("readyTransaction : {}", transactionID);