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