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=81125a7152bf562baede534f7b84b2560f4f9bae;hb=1f91b8457d56392e0c634c03dbf5fdf4d6105e3f;hp=678b7815693382179328a8c13adb567d80d61b13;hpb=85a4118a6fe51d8571a1e7bb398d954e4ceddd33;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 678b781569..81125a7152 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 @@ -31,7 +31,6 @@ import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; /** * The ShardTransaction Actor represents a remote transaction @@ -39,12 +38,6 @@ import org.opendaylight.yangtools.yang.model.api.SchemaContext; * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction *

*

- * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions - * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this - * time there are no known advantages for creating a read-only or write-only transaction which may change over time - * at which point we can optimize things in the distributed store as well. - *

- *

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

  • {@link org.opendaylight.controller.cluster.datastore.messages.ReadData} @@ -60,25 +53,22 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering protected static final boolean SERIALIZED_REPLY = true; private final ActorRef shardActor; - private final SchemaContext schemaContext; private final ShardStats shardStats; private final String transactionID; private final short clientTxVersion; - protected ShardTransaction(ActorRef shardActor, SchemaContext schemaContext, - ShardStats shardStats, String transactionID, short clientTxVersion) { + protected ShardTransaction(ActorRef shardActor, ShardStats shardStats, String transactionID, + short clientTxVersion) { super("shard-tx"); //actor name override used for metering. This does not change the "real" actor name this.shardActor = shardActor; - this.schemaContext = schemaContext; this.shardStats = shardStats; this.transactionID = transactionID; this.clientTxVersion = clientTxVersion; } public static Props props(DOMStoreTransaction transaction, ActorRef shardActor, - SchemaContext schemaContext,DatastoreContext datastoreContext, ShardStats shardStats, - String transactionID, short txnClientVersion) { - return Props.create(new ShardTransactionCreator(transaction, shardActor, schemaContext, + DatastoreContext datastoreContext, ShardStats shardStats, String transactionID, short txnClientVersion) { + return Props.create(new ShardTransactionCreator(transaction, shardActor, datastoreContext, shardStats, transactionID, txnClientVersion)); } @@ -92,10 +82,6 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering return transactionID; } - protected SchemaContext getSchemaContext() { - return schemaContext; - } - protected short getClientTxVersion() { return clientTxVersion; } @@ -114,10 +100,14 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering } } + protected boolean returnCloseTransactionReply() { + return true; + } + private void closeTransaction(boolean sendReply) { getDOMStoreTransaction().close(); - if(sendReply) { + if(sendReply && returnCloseTransactionReply()) { getSender().tell(CloseTransactionReply.INSTANCE.toSerializable(), getSelf()); } @@ -126,29 +116,20 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering protected void readData(DOMStoreReadTransaction transaction, ReadData message, final boolean returnSerialized) { - final ActorRef sender = getSender(); - final ActorRef self = getSelf(); - final YangInstanceIdentifier path = message.getPath(); - final CheckedFuture>, ReadFailedException> future = - transaction.read(path); - future.addListener(new Runnable() { - @Override - public void run() { - try { - Optional> optional = future.checkedGet(); - ReadDataReply readDataReply = new ReadDataReply(optional.orNull()); - - sender.tell((returnSerialized ? readDataReply.toSerializable(clientTxVersion): - readDataReply), self); + final YangInstanceIdentifier path = message.getPath(); + try { + final CheckedFuture>, ReadFailedException> future = transaction.read(path); + Optional> optional = future.checkedGet(); + ReadDataReply readDataReply = new ReadDataReply(optional.orNull(), clientTxVersion); - } catch (Exception e) { - shardStats.incrementFailedReadTransactionsCount(); - sender.tell(new akka.actor.Status.Failure(e), self); - } + sender().tell((returnSerialized ? readDataReply.toSerializable(): readDataReply), self()); - } - }, getContext().dispatcher()); + } catch (Exception e) { + LOG.debug(String.format("Unexpected error reading path %s", path), e); + shardStats.incrementFailedReadTransactionsCount(); + sender().tell(new akka.actor.Status.Failure(e), self()); + } } protected void dataExists(DOMStoreReadTransaction transaction, DataExists message, @@ -156,14 +137,13 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering final YangInstanceIdentifier path = message.getPath(); try { - Boolean exists = transaction.exists(path).checkedGet(); - DataExistsReply dataExistsReply = new DataExistsReply(exists); + boolean exists = transaction.exists(path).checkedGet(); + DataExistsReply dataExistsReply = DataExistsReply.create(exists); getSender().tell(returnSerialized ? dataExistsReply.toSerializable() : dataExistsReply, getSelf()); } catch (ReadFailedException e) { getSender().tell(new akka.actor.Status.Failure(e),getSelf()); } - } private static class ShardTransactionCreator implements Creator { @@ -172,19 +152,16 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering final DOMStoreTransaction transaction; final ActorRef shardActor; - final SchemaContext schemaContext; final DatastoreContext datastoreContext; final ShardStats shardStats; final String transactionID; final short txnClientVersion; ShardTransactionCreator(DOMStoreTransaction transaction, ActorRef shardActor, - SchemaContext schemaContext, DatastoreContext datastoreContext, - ShardStats shardStats, String transactionID, short txnClientVersion) { + DatastoreContext datastoreContext, ShardStats shardStats, String transactionID, short txnClientVersion) { this.transaction = transaction; this.shardActor = shardActor; this.shardStats = shardStats; - this.schemaContext = schemaContext; this.datastoreContext = datastoreContext; this.transactionID = transactionID; this.txnClientVersion = txnClientVersion; @@ -195,13 +172,13 @@ public abstract class ShardTransaction extends AbstractUntypedActorWithMetering ShardTransaction tx; if(transaction instanceof DOMStoreReadWriteTransaction) { tx = new ShardReadWriteTransaction((DOMStoreReadWriteTransaction)transaction, - shardActor, schemaContext, shardStats, transactionID, txnClientVersion); + shardActor, shardStats, transactionID, txnClientVersion); } else if(transaction instanceof DOMStoreReadTransaction) { tx = new ShardReadTransaction((DOMStoreReadTransaction)transaction, shardActor, - schemaContext, shardStats, transactionID, txnClientVersion); + shardStats, transactionID, txnClientVersion); } else { tx = new ShardWriteTransaction((DOMStoreWriteTransaction)transaction, - shardActor, schemaContext, shardStats, transactionID, txnClientVersion); + shardActor, shardStats, transactionID, txnClientVersion); } tx.getContext().setReceiveTimeout(datastoreContext.getShardTransactionIdleTimeout());