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%2Fdatabroker%2Factors%2Fdds%2FAbstractProxyTransaction.java;h=025e51684c056ba57360c8b554ac5d5a3e210e42;hb=1d7e8fd9d781f630dee9dfb1b509067dd7fb9caa;hp=36f9a4bccbe07e762dd8f4778ae4809fc4a2487a;hpb=b4d95acff78952020e9fbde4372d13b461fd7469;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java index 36f9a4bccb..025e51684c 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java @@ -8,6 +8,7 @@ package org.opendaylight.controller.cluster.databroker.actors.dds; import akka.actor.ActorRef; +import com.google.common.base.MoreObjects; import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Throwables; @@ -34,6 +35,7 @@ import org.opendaylight.controller.cluster.access.commands.TransactionCommitSucc import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest; import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest; import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess; +import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest; import org.opendaylight.controller.cluster.access.commands.TransactionRequest; import org.opendaylight.controller.cluster.access.concepts.Request; import org.opendaylight.controller.cluster.access.concepts.RequestFailure; @@ -199,16 +201,19 @@ abstract class AbstractProxyTransaction implements Identifiable data) { + checkReadWrite(); checkNotSealed(); doMerge(path, data); } final void write(final YangInstanceIdentifier path, final NormalizedNode data) { + checkReadWrite(); checkNotSealed(); doWrite(path, data); } @@ -235,6 +240,16 @@ abstract class AbstractProxyTransaction implements Identifiable req) { successfulRequests.add(Verify.verifyNotNull(req)); } @@ -302,7 +323,7 @@ abstract class AbstractProxyTransaction implements Identifiable directCommit() { + checkReadWrite(); checkSealed(); // Precludes startReconnect() from interfering with the fast path @@ -334,7 +356,7 @@ abstract class AbstractProxyTransaction implements Identifiable ret) { + checkReadWrite(); checkSealed(); // Precludes startReconnect() from interfering with the fast path @@ -379,6 +402,7 @@ abstract class AbstractProxyTransaction implements Identifiable ret) { + checkReadWrite(); checkSealed(); final TransactionRequest req = new TransactionPreCommitRequest(getIdentifier(), nextSequence(), @@ -392,12 +416,28 @@ abstract class AbstractProxyTransaction implements Identifiable req) { + /* + * The backend has agreed that the transaction has entered PRE_COMMIT phase, meaning it will be committed + * to storage after the timeout completes. + * + * All state has been replicated to the backend, hence we do not need to keep it around. Retain only + * the precommit request, so we know which request to use for resync. + */ + LOG.debug("Transaction {} preCommit completed, clearing successfulRequests", this); + successfulRequests.clear(); + + // TODO: this works, but can contain some useless state (like batched operations). Create an empty + // equivalent of this request and store that. + recordSuccessfulRequest(req); + } + final void doCommit(final VotingFuture ret) { + checkReadWrite(); checkSealed(); sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> { @@ -410,6 +450,16 @@ abstract class AbstractProxyTransaction implements Identifiable req = new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor()); + sendRequest(req, t -> { + LOG.debug("Transaction {} purge completed", this); parent.completeTransaction(this); }); } @@ -473,7 +523,7 @@ abstract class AbstractProxyTransaction implements Identifiable data); + abstract void doDelete(YangInstanceIdentifier path); - abstract void doWrite(final YangInstanceIdentifier path, final NormalizedNode data); + abstract void doMerge(YangInstanceIdentifier path, NormalizedNode data); - abstract CheckedFuture doExists(final YangInstanceIdentifier path); + abstract void doWrite(YangInstanceIdentifier path, NormalizedNode data); - abstract CheckedFuture>, ReadFailedException> doRead( - final YangInstanceIdentifier path); + abstract CheckedFuture doExists(YangInstanceIdentifier path); + + abstract CheckedFuture>, ReadFailedException> doRead(YangInstanceIdentifier path); abstract void doSeal(); @@ -550,4 +601,9 @@ abstract class AbstractProxyTransaction implements Identifiable request, Consumer> callback); + + @Override + public final String toString() { + return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString(); + } }