opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/RemoteProxyTransaction.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.databroker.actors.dds;
   9
  10 import com.google.common.util.concurrent.FluentFuture;
  11 import com.google.common.util.concurrent.ListenableFuture;
  12 import com.google.common.util.concurrent.SettableFuture;
  13 import java.util.Optional;
  14 import java.util.function.Consumer;
  15 import javax.annotation.Nullable;
  16 import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
  17 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
  18 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
  19 import org.opendaylight.controller.cluster.access.commands.AbstractReadTransactionRequest;
  20 import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
  21 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
  22 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
  23 import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest;
  24 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
  25 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequestBuilder;
  26 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
  27 import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
  28 import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess;
  29 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
  30 import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
  31 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
  32 import org.opendaylight.controller.cluster.access.commands.TransactionMerge;
  33 import org.opendaylight.controller.cluster.access.commands.TransactionModification;
  34 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
  35 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
  36 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
  37 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
  38 import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
  39 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  40 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
  41 import org.opendaylight.controller.cluster.access.concepts.Response;
  42 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  43 import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor;
  44 import org.opendaylight.mdsal.common.api.DataStoreUnavailableException;
  45 import org.opendaylight.mdsal.common.api.ReadFailedException;
  46 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
  47 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  48 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  49 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  50 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  51 import org.slf4j.Logger;
  52 import org.slf4j.LoggerFactory;
  53
  54 /**
  55  * An {@link AbstractProxyTransaction} for dispatching a transaction towards a shard leader whose location is currently
  56  * not known or is known to be not co-located with the client.
  57  *
  58  * <p>
  59  * It packages operations and sends them via the client actor queue to the shard leader. That queue is responsible for
  60  * maintaining any submitted operations until the leader is discovered.
  61  *
  62  * <p>
  63  * This class is not safe to access from multiple application threads, as is usual for transactions. Its internal state
  64  * transitions based on backend responses are thread-safe.
  65  *
  66  * @author Robert Varga
  67  */
  68 final class RemoteProxyTransaction extends AbstractProxyTransaction {
  69     private static final Logger LOG = LoggerFactory.getLogger(RemoteProxyTransaction.class);
  70
  71     // FIXME: make this tuneable
  72     private static final int REQUEST_MAX_MODIFICATIONS = 1000;
  73
  74     private final ModifyTransactionRequestBuilder builder;
  75     private final boolean sendReadyOnSeal;
  76     private final boolean snapshotOnly;
  77
  78     private boolean builderBusy;
  79
  80     private volatile Exception operationFailure;
  81
  82     RemoteProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier,
  83             final boolean snapshotOnly, final boolean sendReadyOnSeal, final boolean isDone) {
  84         super(parent, isDone);
  85         this.snapshotOnly = snapshotOnly;
  86         this.sendReadyOnSeal = sendReadyOnSeal;
  87         builder = new ModifyTransactionRequestBuilder(identifier, localActor());
  88     }
  89
  90     @Override
  91     boolean isSnapshotOnly() {
  92         return snapshotOnly;
  93     }
  94
  95     @Override
  96     public TransactionIdentifier getIdentifier() {
  97         return builder.getIdentifier();
  98     }
  99
 100     @Override
 101     void doDelete(final YangInstanceIdentifier path) {
 102         appendModification(new TransactionDelete(path), Optional.empty());
 103     }
 104
 105     @Override
 106     void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 107         appendModification(new TransactionMerge(path, data), Optional.empty());
 108     }
 109
 110     @Override
 111     void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 112         appendModification(new TransactionWrite(path, data), Optional.empty());
 113     }
 114
 115     private <T> FluentFuture<T> sendReadRequest(final AbstractReadTransactionRequest<?> request,
 116             final Consumer<Response<?, ?>> completer, final ListenableFuture<T> future) {
 117         // Check if a previous operation failed. If it has, do not bother sending anything and report a failure
 118         final Exception local = operationFailure;
 119         if (local != null) {
 120             return FluentFutures.immediateFailedFluentFuture(
 121                     new ReadFailedException("Previous operation failed", local));
 122         }
 123
 124         // Make sure we send any modifications before issuing a read
 125         ensureFlushedBuider();
 126         sendRequest(request, completer);
 127         return FluentFuture.from(future);
 128     }
 129
 130     @Override
 131     FluentFuture<Boolean> doExists(final YangInstanceIdentifier path) {
 132         final SettableFuture<Boolean> future = SettableFuture.create();
 133         return sendReadRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(), path,
 134             isSnapshotOnly()), t -> completeExists(path, future, t), future);
 135     }
 136
 137     @Override
 138     FluentFuture<Optional<NormalizedNode<?, ?>>> doRead(final YangInstanceIdentifier path) {
 139         final SettableFuture<Optional<NormalizedNode<?, ?>>> future = SettableFuture.create();
 140         return sendReadRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), path,
 141             isSnapshotOnly()), t -> completeRead(path, future, t), future);
 142     }
 143
 144     private void ensureInitializedBuilder() {
 145         if (!builderBusy) {
 146             builder.setSequence(nextSequence());
 147             builderBusy = true;
 148         }
 149     }
 150
 151     private void ensureFlushedBuider() {
 152         ensureFlushedBuider(Optional.empty());
 153     }
 154
 155     private void ensureFlushedBuider(final Optional<Long> enqueuedTicks) {
 156         if (builderBusy) {
 157             flushBuilder(enqueuedTicks);
 158         }
 159     }
 160
 161     private void flushBuilder(final Optional<Long> enqueuedTicks) {
 162         final ModifyTransactionRequest request = builder.build();
 163         builderBusy = false;
 164
 165         sendModification(request, enqueuedTicks);
 166     }
 167
 168     private void sendModification(final TransactionRequest<?> request, final Optional<Long> enqueuedTicks) {
 169         if (enqueuedTicks.isPresent()) {
 170             enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.get().longValue());
 171         } else {
 172             sendRequest(request, response -> completeModify(request, response));
 173         }
 174     }
 175
 176     private void appendModification(final TransactionModification modification) {
 177         appendModification(modification, Optional.empty());
 178     }
 179
 180     private void appendModification(final TransactionModification modification, final Optional<Long> enqueuedTicks) {
 181         if (operationFailure == null) {
 182             ensureInitializedBuilder();
 183
 184             builder.addModification(modification);
 185             if (builder.size() >= REQUEST_MAX_MODIFICATIONS) {
 186                 flushBuilder(enqueuedTicks);
 187             }
 188         } else {
 189             LOG.debug("Transaction {} failed, not attempting further transactions", getIdentifier());
 190         }
 191     }
 192
 193     private void completeModify(final TransactionRequest<?> request, final Response<?, ?> response) {
 194         LOG.debug("Modification request {} completed with {}", request, response);
 195
 196         if (response instanceof TransactionSuccess) {
 197             // Happy path
 198             recordSuccessfulRequest(request);
 199         } else {
 200             recordFailedResponse(response);
 201         }
 202     }
 203
 204     private Exception recordFailedResponse(final Response<?, ?> response) {
 205         final Exception failure;
 206         if (response instanceof RequestFailure) {
 207             final RequestException cause = ((RequestFailure<?, ?>) response).getCause();
 208             failure = cause instanceof RequestTimeoutException
 209                     ? new DataStoreUnavailableException(cause.getMessage(), cause) : cause;
 210         } else {
 211             LOG.warn("Unhandled response {}", response);
 212             failure = new IllegalArgumentException("Unhandled response " + response.getClass());
 213         }
 214
 215         if (operationFailure == null) {
 216             LOG.debug("Transaction {} failed", getIdentifier(), failure);
 217             operationFailure = failure;
 218         }
 219         return failure;
 220     }
 221
 222     private void failReadFuture(final SettableFuture<?> future, final String message,
 223             final Response<?, ?> response) {
 224         future.setException(new ReadFailedException(message, recordFailedResponse(response)));
 225     }
 226
 227     private void completeExists(final YangInstanceIdentifier path, final SettableFuture<Boolean> future,
 228             final Response<?, ?> response) {
 229         LOG.debug("Exists request for {} completed with {}", path, response);
 230
 231         if (response instanceof ExistsTransactionSuccess) {
 232             future.set(((ExistsTransactionSuccess) response).getExists());
 233         } else {
 234             failReadFuture(future, "Error executing exists request for path " + path, response);
 235         }
 236
 237         recordFinishedRequest(response);
 238     }
 239
 240     private void completeRead(final YangInstanceIdentifier path,
 241             final SettableFuture<Optional<NormalizedNode<?, ?>>> future, final Response<?, ?> response) {
 242         LOG.debug("Read request for {} completed with {}", path, response);
 243
 244         if (response instanceof ReadTransactionSuccess) {
 245             future.set(((ReadTransactionSuccess) response).getData());
 246         } else {
 247             failReadFuture(future, "Error reading data for path " + path, response);
 248         }
 249
 250         recordFinishedRequest(response);
 251     }
 252
 253     @Override
 254     ModifyTransactionRequest abortRequest() {
 255         ensureInitializedBuilder();
 256         builder.setAbort();
 257         builderBusy = false;
 258         return builder.build();
 259     }
 260
 261     @Override
 262     ModifyTransactionRequest commitRequest(final boolean coordinated) {
 263         ensureInitializedBuilder();
 264         builder.setCommit(coordinated);
 265         builderBusy = false;
 266         return builder.build();
 267     }
 268
 269     private ModifyTransactionRequest readyRequest() {
 270         ensureInitializedBuilder();
 271         builder.setReady();
 272         builderBusy = false;
 273         return builder.build();
 274     }
 275
 276     @Override
 277     boolean sealAndSend(final Optional<Long> enqueuedTicks) {
 278         if (sendReadyOnSeal) {
 279             ensureInitializedBuilder();
 280             builder.setReady();
 281             flushBuilder(enqueuedTicks);
 282         }
 283         return super.sealAndSend(enqueuedTicks);
 284     }
 285
 286     @Override
 287     java.util.Optional<ModifyTransactionRequest> flushState() {
 288         if (!builderBusy) {
 289             return java.util.Optional.empty();
 290         }
 291
 292         final ModifyTransactionRequest request = builder.build();
 293         builderBusy = false;
 294         return java.util.Optional.of(request);
 295     }
 296
 297     @Override
 298     void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest<?> request,
 299             final Consumer<Response<?, ?>> callback) {
 300         successor.handleForwardedRequest(request, callback);
 301     }
 302
 303     void handleForwardedRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
 304         if (request instanceof ModifyTransactionRequest) {
 305             handleForwardedModifyTransactionRequest(callback, (ModifyTransactionRequest) request);
 306         } else if (request instanceof ReadTransactionRequest) {
 307             ensureFlushedBuider();
 308             sendRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(),
 309                 ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
 310                     recordFinishedRequest(resp);
 311                     callback.accept(resp);
 312                 });
 313         } else if (request instanceof ExistsTransactionRequest) {
 314             ensureFlushedBuider();
 315             sendRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(),
 316                 ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
 317                     recordFinishedRequest(resp);
 318                     callback.accept(resp);
 319                 });
 320         } else if (request instanceof TransactionPreCommitRequest) {
 321             ensureFlushedBuider();
 322             final TransactionRequest<?> tmp = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
 323                 localActor());
 324             sendRequest(tmp, resp -> {
 325                 recordSuccessfulRequest(tmp);
 326                 callback.accept(resp);
 327             });
 328         } else if (request instanceof TransactionDoCommitRequest) {
 329             ensureFlushedBuider();
 330             sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
 331         } else if (request instanceof TransactionAbortRequest) {
 332             ensureFlushedBuider();
 333             sendDoAbort(callback);
 334         } else if (request instanceof TransactionPurgeRequest) {
 335             enqueuePurge(callback);
 336         } else {
 337             throw new IllegalArgumentException("Unhandled request {}" + request);
 338         }
 339     }
 340
 341     private void handleForwardedModifyTransactionRequest(final Consumer<Response<?, ?>> callback,
 342             final ModifyTransactionRequest req) {
 343         req.getModifications().forEach(this::appendModification);
 344
 345         final java.util.Optional<PersistenceProtocol> maybeProto = req.getPersistenceProtocol();
 346         if (maybeProto.isPresent()) {
 347             // Persistence protocol implies we are sealed, propagate the marker, but hold off doing other actions
 348             // until we know what we are going to do.
 349             if (markSealed()) {
 350                 sealOnly();
 351             }
 352
 353             final TransactionRequest<?> tmp;
 354             switch (maybeProto.get()) {
 355                 case ABORT:
 356                     tmp = abortRequest();
 357                     sendRequest(tmp, resp -> {
 358                         completeModify(tmp, resp);
 359                         callback.accept(resp);
 360                     });
 361                     break;
 362                 case SIMPLE:
 363                     tmp = commitRequest(false);
 364                     sendRequest(tmp, resp -> {
 365                         completeModify(tmp, resp);
 366                         callback.accept(resp);
 367                     });
 368                     break;
 369                 case THREE_PHASE:
 370                     tmp = commitRequest(true);
 371                     sendRequest(tmp, resp -> {
 372                         recordSuccessfulRequest(tmp);
 373                         callback.accept(resp);
 374                     });
 375                     break;
 376                 case READY:
 377                     tmp = readyRequest();
 378                     sendRequest(tmp, resp -> {
 379                         recordSuccessfulRequest(tmp);
 380                         callback.accept(resp);
 381                     });
 382                     break;
 383                 default:
 384                     throw new IllegalArgumentException("Unhandled protocol " + maybeProto.get());
 385             }
 386         }
 387     }
 388
 389     @Override
 390     void forwardToLocal(final LocalProxyTransaction successor, final TransactionRequest<?> request,
 391             final Consumer<Response<?, ?>> callback) {
 392         successor.handleForwardedRemoteRequest(request, callback);
 393     }
 394
 395     @Override
 396     void handleReplayedLocalRequest(final AbstractLocalTransactionRequest<?> request,
 397             final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
 398         if (request instanceof CommitLocalTransactionRequest) {
 399             replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback, enqueuedTicks);
 400         } else if (request instanceof AbortLocalTransactionRequest) {
 401             enqueueRequest(abortRequest(), callback, enqueuedTicks);
 402         } else {
 403             throw new IllegalStateException("Unhandled request " + request);
 404         }
 405     }
 406
 407     private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
 408             final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
 409         final DataTreeModification mod = request.getModification();
 410         final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
 411
 412         mod.applyToCursor(new AbstractDataTreeModificationCursor() {
 413             @Override
 414             public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
 415                 appendModification(new TransactionWrite(current().node(child), data), optTicks);
 416             }
 417
 418             @Override
 419             public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
 420                 appendModification(new TransactionMerge(current().node(child), data), optTicks);
 421             }
 422
 423             @Override
 424             public void delete(final PathArgument child) {
 425                 appendModification(new TransactionDelete(current().node(child)), optTicks);
 426             }
 427         });
 428
 429         enqueueRequest(commitRequest(request.isCoordinated()), callback, enqueuedTicks);
 430     }
 431
 432     @Override
 433     void handleReplayedRemoteRequest(final TransactionRequest<?> request,
 434             @Nullable final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
 435         final Consumer<Response<?, ?>> cb = callback != null ? callback : resp -> { /* NOOP */ };
 436         final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
 437
 438         if (request instanceof ModifyTransactionRequest) {
 439             handleReplayedModifyTransactionRequest(enqueuedTicks, cb, (ModifyTransactionRequest) request);
 440         } else if (request instanceof ReadTransactionRequest) {
 441             ensureFlushedBuider(optTicks);
 442             enqueueRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(),
 443                 ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
 444                     recordFinishedRequest(resp);
 445                     cb.accept(resp);
 446                 }, enqueuedTicks);
 447         } else if (request instanceof ExistsTransactionRequest) {
 448             ensureFlushedBuider(optTicks);
 449             enqueueRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(),
 450                 ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
 451                     recordFinishedRequest(resp);
 452                     cb.accept(resp);
 453                 }, enqueuedTicks);
 454         } else if (request instanceof TransactionPreCommitRequest) {
 455             ensureFlushedBuider(optTicks);
 456             final TransactionRequest<?> tmp = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
 457                 localActor());
 458             enqueueRequest(tmp, resp -> {
 459                 recordSuccessfulRequest(tmp);
 460                 cb.accept(resp);
 461             }, enqueuedTicks);
 462         } else if (request instanceof TransactionDoCommitRequest) {
 463             ensureFlushedBuider(optTicks);
 464             enqueueRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback,
 465                 enqueuedTicks);
 466         } else if (request instanceof TransactionAbortRequest) {
 467             ensureFlushedBuider(optTicks);
 468             enqueueDoAbort(callback, enqueuedTicks);
 469         } else if (request instanceof TransactionPurgeRequest) {
 470             enqueuePurge(callback, enqueuedTicks);
 471         } else if (request instanceof IncrementTransactionSequenceRequest) {
 472             final IncrementTransactionSequenceRequest req = (IncrementTransactionSequenceRequest) request;
 473             ensureFlushedBuider(optTicks);
 474             enqueueRequest(new IncrementTransactionSequenceRequest(getIdentifier(), nextSequence(), localActor(),
 475                 snapshotOnly, req.getIncrement()), callback, enqueuedTicks);
 476             incrementSequence(req.getIncrement());
 477         } else {
 478             throw new IllegalArgumentException("Unhandled request {}" + request);
 479         }
 480     }
 481
 482     private void handleReplayedModifyTransactionRequest(final long enqueuedTicks, final Consumer<Response<?, ?>> cb,
 483             final ModifyTransactionRequest req) {
 484         req.getModifications().forEach(this::appendModification);
 485
 486         final java.util.Optional<PersistenceProtocol> maybeProto = req.getPersistenceProtocol();
 487         if (maybeProto.isPresent()) {
 488             // Persistence protocol implies we are sealed, propagate the marker, but hold off doing other actions
 489             // until we know what we are going to do.
 490             if (markSealed()) {
 491                 sealOnly();
 492             }
 493
 494             final TransactionRequest<?> tmp;
 495             switch (maybeProto.get()) {
 496                 case ABORT:
 497                     tmp = abortRequest();
 498                     enqueueRequest(tmp, resp -> {
 499                         completeModify(tmp, resp);
 500                         cb.accept(resp);
 501                     }, enqueuedTicks);
 502                     break;
 503                 case SIMPLE:
 504                     tmp = commitRequest(false);
 505                     enqueueRequest(tmp, resp -> {
 506                         completeModify(tmp, resp);
 507                         cb.accept(resp);
 508                     }, enqueuedTicks);
 509                     break;
 510                 case THREE_PHASE:
 511                     tmp = commitRequest(true);
 512                     enqueueRequest(tmp, resp -> {
 513                         recordSuccessfulRequest(tmp);
 514                         cb.accept(resp);
 515                     }, enqueuedTicks);
 516                     break;
 517                 case READY:
 518                     tmp = readyRequest();
 519                     enqueueRequest(tmp, resp -> {
 520                         recordSuccessfulRequest(tmp);
 521                         cb.accept(resp);
 522                     }, enqueuedTicks);
 523                     break;
 524                 default:
 525                     throw new IllegalArgumentException("Unhandled protocol " + maybeProto.get());
 526             }
 527         }
 528     }
 529 }