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%2Fdatabroker%2Factors%2Fdds%2FLocalReadWriteProxyTransaction.java;h=ccce603652d9f1e52be7345fd80b093e6b3f78ce;hp=eee4fd0e137f714b86d24c3279d717c9c323a051;hb=7e62b4a59f9e43bcd0585845f1aeb55c44199f27;hpb=31316f39aecc6bad171de539292ff5d7f4743419

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalReadWriteProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalReadWriteProxyTransaction.java
index eee4fd0e13..ccce603652 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalReadWriteProxyTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalReadWriteProxyTransaction.java
@@ -20,6 +20,7 @@ import org.opendaylight.controller.cluster.access.commands.AbortLocalTransaction
 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequestBuilder;
 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
@@ -176,32 +177,48 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
         return ret;
     }
 
-    @Override
-    void doSeal() {
-        Preconditions.checkState(sealedModification == null, "Transaction %s is already sealed", getIdentifier());
+    private void sealModification() {
+        Preconditions.checkState(sealedModification == null, "Transaction %s is already sealed", this);
         final CursorAwareDataTreeModification mod = getModification();
         mod.ready();
         sealedModification = mod;
     }
 
     @Override
-    void flushState(final AbstractProxyTransaction successor) {
+    void sealOnly() {
+        sealModification();
+        super.sealOnly();
+    }
+
+    @Override
+    boolean sealAndSend(final Optional<Long> enqueuedTicks) {
+        sealModification();
+        return super.sealAndSend(enqueuedTicks);
+    }
+
+    @Override
+    Optional<ModifyTransactionRequest> flushState() {
+        final ModifyTransactionRequestBuilder b = new ModifyTransactionRequestBuilder(getIdentifier(), localActor());
+        b.setSequence(0);
+
         sealedModification.applyToCursor(new AbstractDataTreeModificationCursor() {
             @Override
             public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
-                successor.write(current().node(child), data);
+                b.addModification(new TransactionWrite(current().node(child), data));
             }
 
             @Override
             public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
-                successor.merge(current().node(child), data);
+                b.addModification(new TransactionMerge(current().node(child), data));
             }
 
             @Override
             public void delete(final PathArgument child) {
-                successor.delete(current().node(child));
+                b.addModification(new TransactionDelete(current().node(child)));
             }
         });
+
+        return Optional.of(b.build());
     }
 
     DataTreeSnapshot getSnapshot() {
@@ -211,18 +228,18 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
 
     @Override
     void applyForwardedModifyTransactionRequest(final ModifyTransactionRequest request,
-            final @Nullable Consumer<Response<?, ?>> callback) {
+            @Nullable final Consumer<Response<?, ?>> callback) {
         commonModifyTransactionRequest(request, callback, this::sendRequest);
     }
 
     @Override
     void replayModifyTransactionRequest(final ModifyTransactionRequest request,
-            final @Nullable Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+            @Nullable final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
         commonModifyTransactionRequest(request, callback, (req, cb) -> enqueueRequest(req, cb, enqueuedTicks));
     }
 
     private void commonModifyTransactionRequest(final ModifyTransactionRequest request,
-            final @Nullable Consumer<Response<?, ?>> callback,
+            @Nullable final Consumer<Response<?, ?>> callback,
             final BiConsumer<TransactionRequest<?>, Consumer<Response<?, ?>>> sendMethod) {
         for (final TransactionModification mod : request.getModifications()) {
             if (mod instanceof TransactionWrite) {
@@ -238,15 +255,17 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
 
         final Optional<PersistenceProtocol> maybeProtocol = request.getPersistenceProtocol();
         if (maybeProtocol.isPresent()) {
-            Verify.verify(callback != null, "Request {} has null callback", request);
-            ensureSealed();
+            Verify.verify(callback != null, "Request %s has null callback", request);
+            if (markSealed()) {
+                sealOnly();
+            }
 
             switch (maybeProtocol.get()) {
                 case ABORT:
                     sendMethod.accept(new AbortLocalTransactionRequest(getIdentifier(), localActor()), callback);
                     break;
                 case READY:
-                    // No-op, as we have already issued a seal()
+                    // No-op, as we have already issued a sealOnly() and we are not transmitting anything
                     break;
                 case SIMPLE:
                     sendMethod.accept(commitRequest(false), callback);
@@ -264,7 +283,7 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
     void handleReplayedLocalRequest(final AbstractLocalTransactionRequest<?> request,
             final Consumer<Response<?, ?>> callback, final long now) {
         if (request instanceof CommitLocalTransactionRequest) {
-            sendCommit((CommitLocalTransactionRequest) request, callback);
+            enqueueRequest(rebaseCommit((CommitLocalTransactionRequest)request), callback, now);
         } else {
             super.handleReplayedLocalRequest(request, callback, now);
         }
@@ -272,7 +291,7 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
 
     @Override
     void handleReplayedRemoteRequest(final TransactionRequest<?> request,
-            final @Nullable Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+            @Nullable final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
         LOG.debug("Applying replayed request {}", request);
 
         if (request instanceof TransactionPreCommitRequest) {
@@ -308,7 +327,7 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
             final Consumer<Response<?, ?>> callback) {
         if (request instanceof CommitLocalTransactionRequest) {
             Verify.verify(successor instanceof LocalReadWriteProxyTransaction);
-            ((LocalReadWriteProxyTransaction) successor).sendCommit((CommitLocalTransactionRequest)request, callback);
+            ((LocalReadWriteProxyTransaction) successor).sendRebased((CommitLocalTransactionRequest)request, callback);
             LOG.debug("Forwarded request {} to successor {}", request, successor);
         } else {
             super.forwardToLocal(successor, request, callback);
@@ -328,7 +347,8 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
         closedException = this::abortedException;
     }
 
-    private @Nonnull CursorAwareDataTreeModification getModification() {
+    @Nonnull
+    private CursorAwareDataTreeModification getModification() {
         if (closedException != null) {
             throw closedException.get();
         }
@@ -336,7 +356,11 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
         return Preconditions.checkNotNull(modification, "Transaction %s is DONE", getIdentifier());
     }
 
-    private void sendCommit(final CommitLocalTransactionRequest request, final Consumer<Response<?, ?>> callback) {
+    private void sendRebased(final CommitLocalTransactionRequest request, final Consumer<Response<?, ?>> callback) {
+        sendRequest(rebaseCommit(request), callback);
+    }
+
+    private CommitLocalTransactionRequest rebaseCommit(final CommitLocalTransactionRequest request) {
         // Rebase old modification on new data tree.
         final CursorAwareDataTreeModification mod = getModification();
 
@@ -344,7 +368,10 @@ final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
             request.getModification().applyToCursor(cursor);
         }
 
-        ensureSealed();
-        sendRequest(commitRequest(request.isCoordinated()), callback);
+        if (markSealed()) {
+            sealOnly();
+        }
+
+        return commitRequest(request.isCoordinated());
     }
 }