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%2FRemoteProxyTransaction.java;h=946e3341fd8f7778fdbf940299deb72112a61284;hb=HEAD;hp=3120f6f4edad443624b4ec989f541312e44b7144;hpb=8232a626b43fdd2f5799da0fbcfb0f02d3c8f4fb;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java index 3120f6f4ed..946e3341fd 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java @@ -7,13 +7,15 @@ */ package org.opendaylight.controller.cluster.databroker.actors.dds; -import com.google.common.base.Optional; -import com.google.common.util.concurrent.CheckedFuture; -import com.google.common.util.concurrent.Futures; +import static com.google.common.base.Verify.verify; + +import com.google.common.util.concurrent.FluentFuture; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; +import java.util.Optional; +import java.util.OptionalLong; import java.util.function.Consumer; -import javax.annotation.Nullable; +import org.opendaylight.controller.cluster.access.client.RequestTimeoutException; import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest; import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest; import org.opendaylight.controller.cluster.access.commands.AbstractReadTransactionRequest; @@ -36,16 +38,18 @@ import org.opendaylight.controller.cluster.access.commands.TransactionPurgeReque import org.opendaylight.controller.cluster.access.commands.TransactionRequest; import org.opendaylight.controller.cluster.access.commands.TransactionSuccess; import org.opendaylight.controller.cluster.access.commands.TransactionWrite; +import org.opendaylight.controller.cluster.access.concepts.RequestException; import org.opendaylight.controller.cluster.access.concepts.RequestFailure; import org.opendaylight.controller.cluster.access.concepts.Response; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor; +import org.opendaylight.mdsal.common.api.DataStoreUnavailableException; import org.opendaylight.mdsal.common.api.ReadFailedException; -import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture; +import org.opendaylight.yangtools.util.concurrent.FluentFutures; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,18 +64,14 @@ import org.slf4j.LoggerFactory; *

* This class is not safe to access from multiple application threads, as is usual for transactions. Its internal state * transitions based on backend responses are thread-safe. - * - * @author Robert Varga */ final class RemoteProxyTransaction extends AbstractProxyTransaction { private static final Logger LOG = LoggerFactory.getLogger(RemoteProxyTransaction.class); - // FIXME: make this tuneable - private static final int REQUEST_MAX_MODIFICATIONS = 1000; - private final ModifyTransactionRequestBuilder builder; private final boolean sendReadyOnSeal; private final boolean snapshotOnly; + private final int maxModifications; private boolean builderBusy; @@ -83,6 +83,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { this.snapshotOnly = snapshotOnly; this.sendReadyOnSeal = sendReadyOnSeal; builder = new ModifyTransactionRequestBuilder(identifier, localActor()); + maxModifications = parent.parent().actorUtils().getDatastoreContext().getShardBatchedModificationCount(); } @Override @@ -97,45 +98,46 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { @Override void doDelete(final YangInstanceIdentifier path) { - appendModification(new TransactionDelete(path), Optional.absent()); + appendModification(new TransactionDelete(path), OptionalLong.empty()); } @Override - void doMerge(final YangInstanceIdentifier path, final NormalizedNode data) { - appendModification(new TransactionMerge(path, data), Optional.absent()); + void doMerge(final YangInstanceIdentifier path, final NormalizedNode data) { + appendModification(new TransactionMerge(path, data), OptionalLong.empty()); } @Override - void doWrite(final YangInstanceIdentifier path, final NormalizedNode data) { - appendModification(new TransactionWrite(path, data), Optional.absent()); + void doWrite(final YangInstanceIdentifier path, final NormalizedNode data) { + appendModification(new TransactionWrite(path, data), OptionalLong.empty()); } - private CheckedFuture sendReadRequest(final AbstractReadTransactionRequest request, + private FluentFuture sendReadRequest(final AbstractReadTransactionRequest request, final Consumer> completer, final ListenableFuture future) { // Check if a previous operation failed. If it has, do not bother sending anything and report a failure final Exception local = operationFailure; if (local != null) { - return Futures.immediateFailedCheckedFuture(new ReadFailedException("Previous operation failed", local)); + return FluentFutures.immediateFailedFluentFuture( + new ReadFailedException("Previous operation failed", local)); } // Make sure we send any modifications before issuing a read ensureFlushedBuider(); sendRequest(request, completer); - return MappingCheckedFuture.create(future, ReadFailedException.MAPPER); + return FluentFuture.from(future); } @Override - CheckedFuture doExists(final YangInstanceIdentifier path) { + FluentFuture doExists(final YangInstanceIdentifier path) { final SettableFuture future = SettableFuture.create(); return sendReadRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(), path, - isSnapshotOnly()), t -> completeExists(future, t), future); + isSnapshotOnly()), t -> completeExists(path, future, t), future); } @Override - CheckedFuture>, ReadFailedException> doRead(final YangInstanceIdentifier path) { - final SettableFuture>> future = SettableFuture.create(); + FluentFuture> doRead(final YangInstanceIdentifier path) { + final SettableFuture> future = SettableFuture.create(); return sendReadRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), path, - isSnapshotOnly()), t -> completeRead(future, t), future); + isSnapshotOnly()), t -> completeRead(path, future, t), future); } private void ensureInitializedBuilder() { @@ -146,40 +148,40 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { } private void ensureFlushedBuider() { - ensureFlushedBuider(Optional.absent()); + ensureFlushedBuider(OptionalLong.empty()); } - private void ensureFlushedBuider(final Optional enqueuedTicks) { + private void ensureFlushedBuider(final OptionalLong enqueuedTicks) { if (builderBusy) { flushBuilder(enqueuedTicks); } } - private void flushBuilder(final Optional enqueuedTicks) { + private void flushBuilder(final OptionalLong enqueuedTicks) { final ModifyTransactionRequest request = builder.build(); builderBusy = false; sendModification(request, enqueuedTicks); } - private void sendModification(final TransactionRequest request, final Optional enqueuedTicks) { + private void sendModification(final TransactionRequest request, final OptionalLong enqueuedTicks) { if (enqueuedTicks.isPresent()) { - enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.get().longValue()); + enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.orElseThrow()); } else { sendRequest(request, response -> completeModify(request, response)); } } private void appendModification(final TransactionModification modification) { - appendModification(modification, Optional.absent()); + appendModification(modification, OptionalLong.empty()); } - private void appendModification(final TransactionModification modification, final Optional enqueuedTicks) { + private void appendModification(final TransactionModification modification, final OptionalLong enqueuedTicks) { if (operationFailure == null) { ensureInitializedBuilder(); builder.addModification(modification); - if (builder.size() >= REQUEST_MAX_MODIFICATIONS) { + if (builder.size() >= maxModifications) { flushBuilder(enqueuedTicks); } } else { @@ -200,8 +202,10 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { private Exception recordFailedResponse(final Response response) { final Exception failure; - if (response instanceof RequestFailure) { - failure = ((RequestFailure) response).getCause(); + if (response instanceof RequestFailure requestFailure) { + final RequestException cause = requestFailure.getCause(); + failure = cause instanceof RequestTimeoutException + ? new DataStoreUnavailableException(cause.getMessage(), cause) : cause; } else { LOG.warn("Unhandled response {}", response); failure = new IllegalArgumentException("Unhandled response " + response.getClass()); @@ -214,30 +218,32 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { return failure; } - private void failFuture(final SettableFuture future, final Response response) { - future.setException(recordFailedResponse(response)); + private void failReadFuture(final SettableFuture future, final String message, + final Response response) { + future.setException(new ReadFailedException(message, recordFailedResponse(response))); } - private void completeExists(final SettableFuture future, final Response response) { - LOG.debug("Exists request completed with {}", response); + private void completeExists(final YangInstanceIdentifier path, final SettableFuture future, + final Response response) { + LOG.debug("Exists request for {} completed with {}", path, response); - if (response instanceof ExistsTransactionSuccess) { - future.set(((ExistsTransactionSuccess) response).getExists()); + if (response instanceof ExistsTransactionSuccess success) { + future.set(success.getExists()); } else { - failFuture(future, response); + failReadFuture(future, "Error executing exists request for path " + path, response); } recordFinishedRequest(response); } - private void completeRead(final SettableFuture>> future, + private void completeRead(final YangInstanceIdentifier path, final SettableFuture> future, final Response response) { - LOG.debug("Read request completed with {}", response); + LOG.debug("Read request for {} completed with {}", path, response); - if (response instanceof ReadTransactionSuccess) { - future.set(((ReadTransactionSuccess) response).getData()); + if (response instanceof ReadTransactionSuccess success) { + future.set(success.getData()); } else { - failFuture(future, response); + failReadFuture(future, "Error reading data for path " + path, response); } recordFinishedRequest(response); @@ -267,7 +273,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { } @Override - boolean sealAndSend(final Optional enqueuedTicks) { + boolean sealAndSend(final OptionalLong enqueuedTicks) { if (sendReadyOnSeal) { ensureInitializedBuilder(); builder.setReady(); @@ -277,14 +283,14 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { } @Override - java.util.Optional flushState() { + Optional flushState() { if (!builderBusy) { - return java.util.Optional.empty(); + return Optional.empty(); } final ModifyTransactionRequest request = builder.build(); builderBusy = false; - return java.util.Optional.of(request); + return Optional.of(request); } @Override @@ -294,19 +300,19 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { } void handleForwardedRequest(final TransactionRequest request, final Consumer> callback) { - if (request instanceof ModifyTransactionRequest) { - handleForwardedModifyTransactionRequest(callback, (ModifyTransactionRequest) request); - } else if (request instanceof ReadTransactionRequest) { + if (request instanceof ModifyTransactionRequest modifyRequest) { + handleForwardedModifyTransactionRequest(callback, modifyRequest); + } else if (request instanceof ReadTransactionRequest readRequest) { ensureFlushedBuider(); sendRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), - ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> { + readRequest.getPath(), isSnapshotOnly()), resp -> { recordFinishedRequest(resp); callback.accept(resp); }); - } else if (request instanceof ExistsTransactionRequest) { + } else if (request instanceof ExistsTransactionRequest existsRequest) { ensureFlushedBuider(); sendRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(), - ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> { + existsRequest.getPath(), isSnapshotOnly()), resp -> { recordFinishedRequest(resp); callback.accept(resp); }); @@ -327,7 +333,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { } else if (request instanceof TransactionPurgeRequest) { enqueuePurge(callback); } else { - throw new IllegalArgumentException("Unhandled request {}" + request); + throw unhandledRequest(request); } } @@ -335,16 +341,18 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { final ModifyTransactionRequest req) { req.getModifications().forEach(this::appendModification); - final java.util.Optional maybeProto = req.getPersistenceProtocol(); + final Optional maybeProto = req.getPersistenceProtocol(); if (maybeProto.isPresent()) { // Persistence protocol implies we are sealed, propagate the marker, but hold off doing other actions // until we know what we are going to do. if (markSealed()) { - sealOnly(); + if (!sealOnly()) { + LOG.debug("Proxy {} has a successor, which should receive seal through a separate request", this); + } } final TransactionRequest tmp; - switch (maybeProto.get()) { + switch (maybeProto.orElseThrow()) { case ABORT: tmp = abortRequest(); sendRequest(tmp, resp -> { @@ -374,7 +382,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { }); break; default: - throw new IllegalArgumentException("Unhandled protocol " + maybeProto.get()); + throw new IllegalArgumentException("Unhandled protocol " + maybeProto.orElseThrow()); } } } @@ -388,28 +396,28 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { @Override void handleReplayedLocalRequest(final AbstractLocalTransactionRequest request, final Consumer> callback, final long enqueuedTicks) { - if (request instanceof CommitLocalTransactionRequest) { - replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback, enqueuedTicks); + if (request instanceof CommitLocalTransactionRequest commitRequest) { + replayLocalCommitRequest(commitRequest, callback, enqueuedTicks); } else if (request instanceof AbortLocalTransactionRequest) { enqueueRequest(abortRequest(), callback, enqueuedTicks); } else { - throw new IllegalStateException("Unhandled request " + request); + throw unhandledRequest(request); } } private void replayLocalCommitRequest(final CommitLocalTransactionRequest request, final Consumer> callback, final long enqueuedTicks) { final DataTreeModification mod = request.getModification(); - final Optional optTicks = Optional.of(Long.valueOf(enqueuedTicks)); + final OptionalLong optTicks = OptionalLong.of(enqueuedTicks); mod.applyToCursor(new AbstractDataTreeModificationCursor() { @Override - public void write(final PathArgument child, final NormalizedNode data) { + public void write(final PathArgument child, final NormalizedNode data) { appendModification(new TransactionWrite(current().node(child), data), optTicks); } @Override - public void merge(final PathArgument child, final NormalizedNode data) { + public void merge(final PathArgument child, final NormalizedNode data) { appendModification(new TransactionMerge(current().node(child), data), optTicks); } @@ -423,24 +431,24 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { } @Override - void handleReplayedRemoteRequest(final TransactionRequest request, - @Nullable final Consumer> callback, final long enqueuedTicks) { + void handleReplayedRemoteRequest(final TransactionRequest request, final Consumer> callback, + final long enqueuedTicks) { final Consumer> cb = callback != null ? callback : resp -> { /* NOOP */ }; - final Optional optTicks = Optional.of(Long.valueOf(enqueuedTicks)); + final OptionalLong optTicks = OptionalLong.of(enqueuedTicks); - if (request instanceof ModifyTransactionRequest) { - handleReplayedModifyTransactionRequest(enqueuedTicks, cb, (ModifyTransactionRequest) request); - } else if (request instanceof ReadTransactionRequest) { + if (request instanceof ModifyTransactionRequest modifyRequest) { + handleReplayedModifyTransactionRequest(enqueuedTicks, cb, modifyRequest); + } else if (request instanceof ReadTransactionRequest readRequest) { ensureFlushedBuider(optTicks); enqueueRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), - ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> { + readRequest.getPath(), isSnapshotOnly()), resp -> { recordFinishedRequest(resp); cb.accept(resp); }, enqueuedTicks); - } else if (request instanceof ExistsTransactionRequest) { + } else if (request instanceof ExistsTransactionRequest existsRequest) { ensureFlushedBuider(optTicks); enqueueRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(), - ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> { + existsRequest.getPath(), isSnapshotOnly()), resp -> { recordFinishedRequest(resp); cb.accept(resp); }, enqueuedTicks); @@ -461,14 +469,13 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { enqueueDoAbort(callback, enqueuedTicks); } else if (request instanceof TransactionPurgeRequest) { enqueuePurge(callback, enqueuedTicks); - } else if (request instanceof IncrementTransactionSequenceRequest) { - final IncrementTransactionSequenceRequest req = (IncrementTransactionSequenceRequest) request; + } else if (request instanceof IncrementTransactionSequenceRequest req) { ensureFlushedBuider(optTicks); enqueueRequest(new IncrementTransactionSequenceRequest(getIdentifier(), nextSequence(), localActor(), snapshotOnly, req.getIncrement()), callback, enqueuedTicks); incrementSequence(req.getIncrement()); } else { - throw new IllegalArgumentException("Unhandled request {}" + request); + throw unhandledRequest(request); } } @@ -476,16 +483,16 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { final ModifyTransactionRequest req) { req.getModifications().forEach(this::appendModification); - final java.util.Optional maybeProto = req.getPersistenceProtocol(); + final Optional maybeProto = req.getPersistenceProtocol(); if (maybeProto.isPresent()) { // Persistence protocol implies we are sealed, propagate the marker, but hold off doing other actions // until we know what we are going to do. if (markSealed()) { - sealOnly(); + verify(sealOnly(), "Attempted to replay seal on %s", this); } final TransactionRequest tmp; - switch (maybeProto.get()) { + switch (maybeProto.orElseThrow()) { case ABORT: tmp = abortRequest(); enqueueRequest(tmp, resp -> { @@ -515,7 +522,7 @@ final class RemoteProxyTransaction extends AbstractProxyTransaction { }, enqueuedTicks); break; default: - throw new IllegalArgumentException("Unhandled protocol " + maybeProto.get()); + throw new IllegalArgumentException("Unhandled protocol " + maybeProto.orElseThrow()); } } }