opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/SingleCommitCohortProxy.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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.datastore;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import akka.dispatch.OnComplete;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.SettableFuture;
  15 import java.util.List;
  16 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  17 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  18 import org.opendaylight.mdsal.common.api.CommitInfo;
  19 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
  20 import org.opendaylight.yangtools.yang.common.Empty;
  21 import org.slf4j.Logger;
  22 import org.slf4j.LoggerFactory;
  23 import scala.concurrent.Future;
  24
  25 /**
  26  * A cohort proxy implementation for a single-shard transaction commit. If the transaction was a direct commit
  27  * to the shard, this implementation elides the CanCommitTransaction and CommitTransaction messages to the
  28  * shard as an optimization.
  29  *
  30  * @author Thomas Pantelis
  31  */
  32 class SingleCommitCohortProxy extends AbstractThreePhaseCommitCohort<Object> {
  33     private static final Logger LOG = LoggerFactory.getLogger(SingleCommitCohortProxy.class);
  34
  35     private final ActorUtils actorUtils;
  36     private final Future<Object> cohortFuture;
  37     private final TransactionIdentifier transactionId;
  38     private volatile DOMStoreThreePhaseCommitCohort delegateCohort = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
  39     private final OperationCallback.Reference operationCallbackRef;
  40
  41     SingleCommitCohortProxy(final ActorUtils actorUtils, final Future<Object> cohortFuture,
  42             final TransactionIdentifier transactionId, final OperationCallback.Reference operationCallbackRef) {
  43         this.actorUtils = actorUtils;
  44         this.cohortFuture = cohortFuture;
  45         this.transactionId = requireNonNull(transactionId);
  46         this.operationCallbackRef = operationCallbackRef;
  47     }
  48
  49     @Override
  50     public ListenableFuture<Boolean> canCommit() {
  51         LOG.debug("Tx {} canCommit", transactionId);
  52
  53         final SettableFuture<Boolean> returnFuture = SettableFuture.create();
  54
  55         cohortFuture.onComplete(new OnComplete<>() {
  56             @Override
  57             public void onComplete(final Throwable failure, final Object cohortResponse) {
  58                 if (failure != null) {
  59                     operationCallbackRef.get().failure();
  60                     returnFuture.setException(failure);
  61                     return;
  62                 }
  63
  64                 operationCallbackRef.get().success();
  65
  66                 LOG.debug("Tx {} successfully completed direct commit", transactionId);
  67
  68                 // The Future was the result of a direct commit to the shard, essentially eliding the
  69                 // front-end 3PC coordination. We don't really care about the specific Future
  70                 // response object, only that it completed successfully. At this point the Tx is complete
  71                 // so return true. The subsequent preCommit and commit phases will be no-ops, ie return
  72                 // immediate success, to complete the 3PC for the front-end.
  73                 returnFuture.set(Boolean.TRUE);
  74             }
  75         }, actorUtils.getClientDispatcher());
  76
  77         return returnFuture;
  78     }
  79
  80     @Override
  81     public ListenableFuture<Empty> preCommit() {
  82         return delegateCohort.preCommit();
  83     }
  84
  85     @Override
  86     public ListenableFuture<Empty> abort() {
  87         return delegateCohort.abort();
  88     }
  89
  90     @Override
  91     public ListenableFuture<? extends CommitInfo> commit() {
  92         return delegateCohort.commit();
  93     }
  94
  95     @Override
  96     List<Future<Object>> getCohortFutures() {
  97         return List.of(cohortFuture);
  98     }
  99 }