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%2FLocalProxyTransaction.java;h=99806fd81f8bbaa27b97c9fc37563abe66bd0ee2;hp=747fd144ba72f30af9056c3c604b911c68f7481f;hb=18ddbfdc55a1faddf7aeb2df6b25481d34c820ab;hpb=caa423625725a7937205f003421e7a08a734760b diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java index 747fd144ba..99806fd81f 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/LocalProxyTransaction.java @@ -12,12 +12,15 @@ import com.google.common.base.Preconditions; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.Futures; import java.util.function.Consumer; +import javax.annotation.Nonnull; import javax.annotation.Nullable; import javax.annotation.concurrent.NotThreadSafe; import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest; +import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest; import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest; import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest; import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess; +import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest; import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest; import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest; import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess; @@ -56,8 +59,8 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { private final TransactionIdentifier identifier; - LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier) { - super(parent); + LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier, final boolean isDone) { + super(parent, isDone); this.identifier = Preconditions.checkNotNull(identifier); } @@ -66,11 +69,14 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { return identifier; } - abstract DataTreeSnapshot readOnlyView(); + abstract @Nonnull DataTreeSnapshot readOnlyView(); - abstract void applyModifyTransactionRequest(ModifyTransactionRequest request, + abstract void applyForwardedModifyTransactionRequest(ModifyTransactionRequest request, @Nullable Consumer> callback); + abstract void replayModifyTransactionRequest(ModifyTransactionRequest request, + @Nullable Consumer> callback, long enqueuedTicks); + @Override final CheckedFuture doExists(final YangInstanceIdentifier path) { return Futures.immediateCheckedFuture(readOnlyView().readNode(path).isPresent()); @@ -82,34 +88,89 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { } @Override - final void doAbort() { - sendAbort(new AbortLocalTransactionRequest(identifier, localActor()), response -> { - LOG.debug("Transaction {} abort completed with {}", identifier, response); - }); + final AbortLocalTransactionRequest abortRequest() { + return new AbortLocalTransactionRequest(identifier, localActor()); } @Override - void handleForwardedRemoteRequest(final TransactionRequest request, + void handleReplayedLocalRequest(final AbstractLocalTransactionRequest request, + final Consumer> callback, final long enqueuedTicks) { + if (request instanceof AbortLocalTransactionRequest) { + enqueueAbort(request, callback, enqueuedTicks); + } else { + throw new IllegalArgumentException("Unhandled request" + request); + } + } + + private boolean handleReadRequest(final TransactionRequest request, final @Nullable Consumer> callback) { - if (request instanceof ModifyTransactionRequest) { - applyModifyTransactionRequest((ModifyTransactionRequest) request, callback); - } else if (request instanceof ReadTransactionRequest) { + // Note we delay completion of read requests to limit the scope at which the client can run, as they have + // listeners, which we do not want to execute while we are reconnecting. + if (request instanceof ReadTransactionRequest) { final YangInstanceIdentifier path = ((ReadTransactionRequest) request).getPath(); final Optional> result = readOnlyView().readNode(path); - callback.accept(new ReadTransactionSuccess(request.getTarget(), request.getSequence(), result)); + if (callback != null) { + // XXX: FB does not see that callback is final, on stack and has be check for non-null. + final Consumer> fbIsStupid = Preconditions.checkNotNull(callback); + executeInActor(() -> fbIsStupid.accept(new ReadTransactionSuccess(request.getTarget(), + request.getSequence(), result))); + } + return true; } else if (request instanceof ExistsTransactionRequest) { final YangInstanceIdentifier path = ((ExistsTransactionRequest) request).getPath(); final boolean result = readOnlyView().readNode(path).isPresent(); - callback.accept(new ExistsTransactionSuccess(request.getTarget(), request.getSequence(), result)); + if (callback != null) { + // XXX: FB does not see that callback is final, on stack and has be check for non-null. + final Consumer> fbIsStupid = Preconditions.checkNotNull(callback); + executeInActor(() -> fbIsStupid.accept(new ExistsTransactionSuccess(request.getTarget(), + request.getSequence(), result))); + } + return true; + } else { + return false; + } + } + + @Override + void handleReplayedRemoteRequest(final TransactionRequest request, + final @Nullable Consumer> callback, final long enqueuedTicks) { + if (request instanceof ModifyTransactionRequest) { + replayModifyTransactionRequest((ModifyTransactionRequest) request, callback, enqueuedTicks); + } else if (handleReadRequest(request, callback)) { + // No-op + } else if (request instanceof TransactionPurgeRequest) { + enqueuePurge(callback, enqueuedTicks); + } else if (request instanceof IncrementTransactionSequenceRequest) { + // Local transactions do not have non-replayable requests which would be visible to the backend, + // hence we can skip sequence increments. + LOG.debug("Not replaying {}", request); + } else { + throw new IllegalArgumentException("Unhandled request " + request); + } + } + + /** + * Remote-to-local equivalent of {@link #handleReplayedRemoteRequest(TransactionRequest, Consumer, long)}, + * except it is invoked in the forwarding path from + * {@link RemoteProxyTransaction#forwardToLocal(LocalProxyTransaction, TransactionRequest, Consumer)}. + * + * @param request Forwarded request + * @param callback Callback to be invoked once the request completes + */ + void handleForwardedRemoteRequest(final TransactionRequest request, final Consumer> callback) { + if (request instanceof ModifyTransactionRequest) { + applyForwardedModifyTransactionRequest((ModifyTransactionRequest) request, callback); + } else if (handleReadRequest(request, callback)) { + // No-op } else if (request instanceof TransactionPurgeRequest) { - purge(); + enqueuePurge(callback); } else { throw new IllegalArgumentException("Unhandled request " + request); } } @Override - void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest request, + final void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest request, final Consumer> callback) { if (request instanceof CommitLocalTransactionRequest) { final CommitLocalTransactionRequest req = (CommitLocalTransactionRequest) request; @@ -133,8 +194,7 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { } }); - successor.ensureSealed(); - + successor.sealOnly(); final ModifyTransactionRequest successorReq = successor.commitRequest(req.isCoordinated()); successor.sendRequest(successorReq, callback); } else if (request instanceof AbortLocalTransactionRequest) { @@ -142,7 +202,7 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { successor.abort(); } else if (request instanceof TransactionPurgeRequest) { LOG.debug("Forwarding purge {} to successor {}", request, successor); - successor.purge(); + successor.enqueuePurge(callback); } else { throw new IllegalArgumentException("Unhandled request" + request); } @@ -154,7 +214,7 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { if (request instanceof AbortLocalTransactionRequest) { successor.sendAbort(request, callback); } else if (request instanceof TransactionPurgeRequest) { - successor.purge(); + successor.enqueuePurge(callback); } else { throw new IllegalArgumentException("Unhandled request" + request); } @@ -165,4 +225,9 @@ abstract class LocalProxyTransaction extends AbstractProxyTransaction { void sendAbort(final TransactionRequest request, final Consumer> callback) { sendRequest(request, callback); } + + void enqueueAbort(final TransactionRequest request, final Consumer> callback, + final long enqueuedTicks) { + enqueueRequest(request, callback, enqueuedTicks); + } }