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