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

   1 /*
   2  * Copyright (c) 2014 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 com.google.common.base.Preconditions;
  11 import com.google.common.util.concurrent.AbstractFuture;
  12 import com.google.common.util.concurrent.AbstractListeningExecutorService;
  13 import com.google.common.util.concurrent.CheckedFuture;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import java.util.Collection;
  18 import java.util.List;
  19 import java.util.Map;
  20 import java.util.concurrent.Executor;
  21 import java.util.concurrent.ExecutorService;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.atomic.AtomicInteger;
  24 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  25 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  26 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  27 import org.opendaylight.controller.md.sal.dom.broker.impl.AbstractDOMDataBroker;
  28 import org.opendaylight.controller.md.sal.dom.broker.impl.TransactionCommitFailedExceptionMapper;
  29 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  30 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  31 import org.opendaylight.yangtools.util.DurationStatisticsTracker;
  32 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 /**
  37  * Implementation of DOMDataCommitExecutor that coordinates transaction commits concurrently. The 3
  38  * commit phases (canCommit, preCommit, and commit) are performed serially and non-blocking
  39  * (ie async) per transaction but multiple transaction commits can run concurrent.
  40  *
  41  * @author Thomas Pantelis
  42  */
  43 public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker {
  44     private static final Logger LOG = LoggerFactory.getLogger(ConcurrentDOMDataBroker.class);
  45     private static final String CAN_COMMIT = "CAN_COMMIT";
  46     private static final String PRE_COMMIT = "PRE_COMMIT";
  47     private static final String COMMIT = "COMMIT";
  48
  49     private final DurationStatisticsTracker commitStatsTracker = DurationStatisticsTracker.createConcurrent();
  50
  51     /**
  52      * This executor is used to execute Future listener callback Runnables async.
  53      */
  54     private final ExecutorService clientFutureCallbackExecutor;
  55
  56     /**
  57      * This executor is re-used internally in calls to Futures#addCallback to avoid the overhead
  58      * of Futures#addCallback creating a MoreExecutors#sameThreadExecutor for each call.
  59      */
  60     private final ExecutorService internalFutureCallbackExecutor = new SimpleSameThreadExecutor();
  61
  62     public ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores, ExecutorService listenableFutureExecutor) {
  63         super(datastores);
  64         this.clientFutureCallbackExecutor = Preconditions.checkNotNull(listenableFutureExecutor);
  65     }
  66
  67     public DurationStatisticsTracker getCommitStatsTracker() {
  68         return commitStatsTracker;
  69     }
  70
  71     @Override
  72     public CheckedFuture<Void, TransactionCommitFailedException> submit(DOMDataWriteTransaction transaction,
  73             Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
  74
  75         Preconditions.checkArgument(transaction != null, "Transaction must not be null.");
  76         Preconditions.checkArgument(cohorts != null, "Cohorts must not be null.");
  77         LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier());
  78
  79         final AsyncNotifyingSettableFuture clientSubmitFuture =
  80                 new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor);
  81
  82         doCanCommit(clientSubmitFuture, transaction, cohorts);
  83
  84         return MappingCheckedFuture.create(clientSubmitFuture,
  85                 TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER);
  86     }
  87
  88     private void doCanCommit(final AsyncNotifyingSettableFuture clientSubmitFuture,
  89             final DOMDataWriteTransaction transaction,
  90             final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
  91
  92         final long startTime = System.nanoTime();
  93
  94         // Not using Futures.allAsList here to avoid its internal overhead.
  95         final AtomicInteger remaining = new AtomicInteger(cohorts.size());
  96         FutureCallback<Boolean> futureCallback = new FutureCallback<Boolean>() {
  97             @Override
  98             public void onSuccess(Boolean result) {
  99                 if (result == null || !result) {
 100                     handleException(clientSubmitFuture, transaction, cohorts,
 101                             CAN_COMMIT, TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER,
 102                             new TransactionCommitFailedException(
 103                                             "Can Commit failed, no detailed cause available."));
 104                 } else {
 105                     if(remaining.decrementAndGet() == 0) {
 106                         // All cohorts completed successfully - we can move on to the preCommit phase
 107                         doPreCommit(startTime, clientSubmitFuture, transaction, cohorts);
 108                     }
 109                 }
 110             }
 111
 112             @Override
 113             public void onFailure(Throwable t) {
 114                 handleException(clientSubmitFuture, transaction, cohorts, CAN_COMMIT,
 115                         TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER, t);
 116             }
 117         };
 118
 119         for(DOMStoreThreePhaseCommitCohort cohort: cohorts) {
 120             ListenableFuture<Boolean> canCommitFuture = cohort.canCommit();
 121             Futures.addCallback(canCommitFuture, futureCallback, internalFutureCallbackExecutor);
 122         }
 123     }
 124
 125     private void doPreCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture,
 126             final DOMDataWriteTransaction transaction,
 127             final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
 128
 129         // Not using Futures.allAsList here to avoid its internal overhead.
 130         final AtomicInteger remaining = new AtomicInteger(cohorts.size());
 131         FutureCallback<Void> futureCallback = new FutureCallback<Void>() {
 132             @Override
 133             public void onSuccess(Void notUsed) {
 134                 if(remaining.decrementAndGet() == 0) {
 135                     // All cohorts completed successfully - we can move on to the commit phase
 136                     doCommit(startTime, clientSubmitFuture, transaction, cohorts);
 137                 }
 138             }
 139
 140             @Override
 141             public void onFailure(Throwable t) {
 142                 handleException(clientSubmitFuture, transaction, cohorts, PRE_COMMIT,
 143                         TransactionCommitFailedExceptionMapper.PRE_COMMIT_MAPPER, t);
 144             }
 145         };
 146
 147         for(DOMStoreThreePhaseCommitCohort cohort: cohorts) {
 148             ListenableFuture<Void> preCommitFuture = cohort.preCommit();
 149             Futures.addCallback(preCommitFuture, futureCallback, internalFutureCallbackExecutor);
 150         }
 151     }
 152
 153     private void doCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture,
 154             final DOMDataWriteTransaction transaction,
 155             final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
 156
 157         // Not using Futures.allAsList here to avoid its internal overhead.
 158         final AtomicInteger remaining = new AtomicInteger(cohorts.size());
 159         FutureCallback<Void> futureCallback = new FutureCallback<Void>() {
 160             @Override
 161             public void onSuccess(Void notUsed) {
 162                 if(remaining.decrementAndGet() == 0) {
 163                     // All cohorts completed successfully - we're done.
 164                     commitStatsTracker.addDuration(System.nanoTime() - startTime);
 165
 166                     clientSubmitFuture.set();
 167                 }
 168             }
 169
 170             @Override
 171             public void onFailure(Throwable t) {
 172                 handleException(clientSubmitFuture, transaction, cohorts, COMMIT,
 173                         TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER, t);
 174             }
 175         };
 176
 177         for(DOMStoreThreePhaseCommitCohort cohort: cohorts) {
 178             ListenableFuture<Void> commitFuture = cohort.commit();
 179             Futures.addCallback(commitFuture, futureCallback, internalFutureCallbackExecutor);
 180         }
 181     }
 182
 183     private void handleException(final AsyncNotifyingSettableFuture clientSubmitFuture,
 184             final DOMDataWriteTransaction transaction,
 185             final Collection<DOMStoreThreePhaseCommitCohort> cohorts,
 186             final String phase, final TransactionCommitFailedExceptionMapper exMapper,
 187             final Throwable t) {
 188
 189         if(clientSubmitFuture.isDone()) {
 190             // We must have had failures from multiple cohorts.
 191             return;
 192         }
 193
 194         LOG.warn("Tx: {} Error during phase {}, starting Abort", transaction.getIdentifier(), phase, t);
 195         Exception e;
 196         if(t instanceof Exception) {
 197             e = (Exception)t;
 198         } else {
 199             e = new RuntimeException("Unexpected error occurred", t);
 200         }
 201
 202         final TransactionCommitFailedException clientException = exMapper.apply(e);
 203
 204         // Transaction failed - tell all cohorts to abort.
 205
 206         @SuppressWarnings("unchecked")
 207         ListenableFuture<Void>[] canCommitFutures = new ListenableFuture[cohorts.size()];
 208         int i = 0;
 209         for(DOMStoreThreePhaseCommitCohort cohort: cohorts) {
 210             canCommitFutures[i++] = cohort.abort();
 211         }
 212
 213         ListenableFuture<List<Void>> combinedFuture = Futures.allAsList(canCommitFutures);
 214         Futures.addCallback(combinedFuture, new FutureCallback<List<Void>>() {
 215             @Override
 216             public void onSuccess(List<Void> notUsed) {
 217                 // Propagate the original exception to the client.
 218                 clientSubmitFuture.setException(clientException);
 219             }
 220
 221             @Override
 222             public void onFailure(Throwable t) {
 223                 LOG.error("Tx: {} Error during Abort.", transaction.getIdentifier(), t);
 224
 225                 // Propagate the original exception as that is what caused the Tx to fail and is
 226                 // what's interesting to the client.
 227                 clientSubmitFuture.setException(clientException);
 228             }
 229         }, internalFutureCallbackExecutor);
 230     }
 231
 232     /**
 233      * A settable future that uses an {@link Executor} to execute listener callback Runnables,
 234      * registered via {@link #addListener}, asynchronously when this future completes. This is
 235      * done to guarantee listener executions are off-loaded onto another thread to avoid blocking
 236      * the thread that completed this future, as a common use case is to pass an executor that runs
 237      * tasks in the same thread as the caller (ie MoreExecutors#sameThreadExecutor)
 238      * to {@link #addListener}.
 239      *
 240      * FIXME: This class should probably be moved to yangtools common utils for re-usability and
 241      * unified with AsyncNotifyingListenableFutureTask.
 242      */
 243     private static class AsyncNotifyingSettableFuture extends AbstractFuture<Void> {
 244
 245         /**
 246          * ThreadLocal used to detect if the task completion thread is running the future listener Runnables.
 247          */
 248         private static final ThreadLocal<Boolean> ON_TASK_COMPLETION_THREAD_TL = new ThreadLocal<Boolean>();
 249
 250         private final ExecutorService listenerExecutor;
 251
 252         AsyncNotifyingSettableFuture(ExecutorService listenerExecutor) {
 253             this.listenerExecutor = listenerExecutor;
 254         }
 255
 256         @Override
 257         public void addListener(final Runnable listener, final Executor executor) {
 258             // Wrap the listener Runnable in a DelegatingRunnable. If the specified executor is one
 259             // that runs tasks in the same thread as the caller submitting the task
 260             // (e.g. {@link com.google.common.util.concurrent.MoreExecutors#sameThreadExecutor}) and
 261             // the listener is executed from the #set methods, then the DelegatingRunnable will detect
 262             // this via the ThreadLocal and submit the listener Runnable to the listenerExecutor.
 263             //
 264             // On the other hand, if this task is already complete, the call to ExecutionList#add in
 265             // superclass will execute the listener Runnable immediately and, since the ThreadLocal
 266             // won't be set, the DelegatingRunnable will run the listener Runnable inline.
 267             super.addListener(new DelegatingRunnable(listener, listenerExecutor), executor);
 268         }
 269
 270         boolean set() {
 271             ON_TASK_COMPLETION_THREAD_TL.set(Boolean.TRUE);
 272             try {
 273                 return super.set(null);
 274             } finally {
 275                 ON_TASK_COMPLETION_THREAD_TL.set(null);
 276             }
 277         }
 278
 279         @Override
 280         protected boolean setException(Throwable throwable) {
 281             ON_TASK_COMPLETION_THREAD_TL.set(Boolean.TRUE);
 282             try {
 283                 return super.setException(throwable);
 284             } finally {
 285                 ON_TASK_COMPLETION_THREAD_TL.set(null);
 286             }
 287         }
 288
 289         private static final class DelegatingRunnable implements Runnable {
 290             private final Runnable delegate;
 291             private final Executor executor;
 292
 293             DelegatingRunnable(final Runnable delegate, final Executor executor) {
 294                 this.delegate = Preconditions.checkNotNull(delegate);
 295                 this.executor = Preconditions.checkNotNull(executor);
 296             }
 297
 298             @Override
 299             public void run() {
 300                 if (ON_TASK_COMPLETION_THREAD_TL.get() != null) {
 301                     // We're running on the task completion thread so off-load to the executor.
 302                     LOG.trace("Submitting ListenenableFuture Runnable from thread {} to executor {}",
 303                             Thread.currentThread().getName(), executor);
 304                     executor.execute(delegate);
 305                 } else {
 306                     // We're not running on the task completion thread so run the delegate inline.
 307                     LOG.trace("Executing ListenenableFuture Runnable on this thread: {}",
 308                             Thread.currentThread().getName());
 309                     delegate.run();
 310                 }
 311             }
 312         }
 313     }
 314
 315     /**
 316      * A simple same-thread executor without the internal locking overhead that
 317      * MoreExecutors#sameThreadExecutor has. The #execute method is the only one of concern - we
 318      * don't shutdown the executor so the other methods irrelevant.
 319      */
 320     private static class SimpleSameThreadExecutor extends AbstractListeningExecutorService {
 321
 322         @Override
 323         public void execute(Runnable command) {
 324             command.run();
 325         }
 326
 327         @Override
 328         public boolean awaitTermination(long arg0, TimeUnit arg1) throws InterruptedException {
 329             return true;
 330         }
 331
 332         @Override
 333         public boolean isShutdown() {
 334             return false;
 335         }
 336
 337         @Override
 338         public boolean isTerminated() {
 339             return false;
 340         }
 341
 342         @Override
 343         public void shutdown() {
 344         }
 345
 346         @Override
 347         public List<Runnable> shutdownNow() {
 348             return null;
 349         }
 350     }
 351 }