X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FConcurrentDOMDataBroker.java;h=c16a84933572d436a14c0506858822039fdc8a77;hp=886c4730678208fdbf129be10463eddf252f5c0f;hb=340a2d4c979ac6f8d5adff8bd9e1c9f724e7a164;hpb=8cfbc45d2679de6d39f9bedde07c38218bc6955a diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ConcurrentDOMDataBroker.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ConcurrentDOMDataBroker.java index 886c473067..c16a849335 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ConcurrentDOMDataBroker.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ConcurrentDOMDataBroker.java @@ -9,22 +9,22 @@ package org.opendaylight.controller.cluster.datastore; import com.google.common.base.Preconditions; import com.google.common.util.concurrent.AbstractFuture; -import com.google.common.util.concurrent.AbstractListeningExecutorService; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; import java.util.Collection; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; +import org.opendaylight.controller.cluster.databroker.AbstractDOMBroker; +import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; +import org.opendaylight.controller.md.sal.common.api.data.DataStoreUnavailableException; import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction; -import org.opendaylight.controller.md.sal.dom.broker.impl.AbstractDOMDataBroker; import org.opendaylight.controller.md.sal.dom.broker.impl.TransactionCommitFailedExceptionMapper; import org.opendaylight.controller.sal.core.spi.data.DOMStore; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; @@ -34,13 +34,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Implementation of DOMDataCommitExecutor that coordinates transaction commits concurrently. The 3 + * ConcurrentDOMDataBroker commits transactions concurrently. The 3 * commit phases (canCommit, preCommit, and commit) are performed serially and non-blocking * (ie async) per transaction but multiple transaction commits can run concurrent. * * @author Thomas Pantelis */ -public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { +public class ConcurrentDOMDataBroker extends AbstractDOMBroker { private static final Logger LOG = LoggerFactory.getLogger(ConcurrentDOMDataBroker.class); private static final String CAN_COMMIT = "CAN_COMMIT"; private static final String PRE_COMMIT = "PRE_COMMIT"; @@ -51,15 +51,9 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { /** * This executor is used to execute Future listener callback Runnables async. */ - private final ExecutorService clientFutureCallbackExecutor; + private final Executor clientFutureCallbackExecutor; - /** - * This executor is re-used internally in calls to Futures#addCallback to avoid the overhead - * of Futures#addCallback creating a MoreExecutors#sameThreadExecutor for each call. - */ - private final ExecutorService internalFutureCallbackExecutor = new SimpleSameThreadExecutor(); - - public ConcurrentDOMDataBroker(final Map datastores, ExecutorService listenableFutureExecutor) { + public ConcurrentDOMDataBroker(final Map datastores, Executor listenableFutureExecutor) { super(datastores); this.clientFutureCallbackExecutor = Preconditions.checkNotNull(listenableFutureExecutor); } @@ -69,13 +63,17 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { } @Override - public CheckedFuture submit(DOMDataWriteTransaction transaction, + protected CheckedFuture submit(DOMDataWriteTransaction transaction, Collection cohorts) { Preconditions.checkArgument(transaction != null, "Transaction must not be null."); Preconditions.checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); + if(cohorts.isEmpty()){ + return Futures.immediateCheckedFuture(null); + } + final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); @@ -91,8 +89,9 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { final long startTime = System.nanoTime(); + final Iterator cohortIterator = cohorts.iterator(); + // Not using Futures.allAsList here to avoid its internal overhead. - final AtomicInteger remaining = new AtomicInteger(cohorts.size()); FutureCallback futureCallback = new FutureCallback() { @Override public void onSuccess(Boolean result) { @@ -102,9 +101,12 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { new TransactionCommitFailedException( "Can Commit failed, no detailed cause available.")); } else { - if(remaining.decrementAndGet() == 0) { + if(!cohortIterator.hasNext()) { // All cohorts completed successfully - we can move on to the preCommit phase doPreCommit(startTime, clientSubmitFuture, transaction, cohorts); + } else { + ListenableFuture canCommitFuture = cohortIterator.next().canCommit(); + Futures.addCallback(canCommitFuture, this, MoreExecutors.directExecutor()); } } } @@ -116,24 +118,26 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { } }; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { - ListenableFuture canCommitFuture = cohort.canCommit(); - Futures.addCallback(canCommitFuture, futureCallback, internalFutureCallbackExecutor); - } + ListenableFuture canCommitFuture = cohortIterator.next().canCommit(); + Futures.addCallback(canCommitFuture, futureCallback, MoreExecutors.directExecutor()); } private void doPreCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, final Collection cohorts) { + final Iterator cohortIterator = cohorts.iterator(); + // Not using Futures.allAsList here to avoid its internal overhead. - final AtomicInteger remaining = new AtomicInteger(cohorts.size()); FutureCallback futureCallback = new FutureCallback() { @Override public void onSuccess(Void notUsed) { - if(remaining.decrementAndGet() == 0) { + if(!cohortIterator.hasNext()) { // All cohorts completed successfully - we can move on to the commit phase doCommit(startTime, clientSubmitFuture, transaction, cohorts); + } else { + ListenableFuture preCommitFuture = cohortIterator.next().preCommit(); + Futures.addCallback(preCommitFuture, this, MoreExecutors.directExecutor()); } } @@ -144,26 +148,28 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { } }; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { - ListenableFuture preCommitFuture = cohort.preCommit(); - Futures.addCallback(preCommitFuture, futureCallback, internalFutureCallbackExecutor); - } + ListenableFuture preCommitFuture = cohortIterator.next().preCommit(); + Futures.addCallback(preCommitFuture, futureCallback, MoreExecutors.directExecutor()); } private void doCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, final Collection cohorts) { + final Iterator cohortIterator = cohorts.iterator(); + // Not using Futures.allAsList here to avoid its internal overhead. - final AtomicInteger remaining = new AtomicInteger(cohorts.size()); FutureCallback futureCallback = new FutureCallback() { @Override public void onSuccess(Void notUsed) { - if(remaining.decrementAndGet() == 0) { + if(!cohortIterator.hasNext()) { // All cohorts completed successfully - we're done. commitStatsTracker.addDuration(System.nanoTime() - startTime); clientSubmitFuture.set(); + } else { + ListenableFuture commitFuture = cohortIterator.next().commit(); + Futures.addCallback(commitFuture, this, MoreExecutors.directExecutor()); } } @@ -174,26 +180,26 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { } }; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { - ListenableFuture commitFuture = cohort.commit(); - Futures.addCallback(commitFuture, futureCallback, internalFutureCallbackExecutor); - } + ListenableFuture commitFuture = cohortIterator.next().commit(); + Futures.addCallback(commitFuture, futureCallback, MoreExecutors.directExecutor()); } - private void handleException(final AsyncNotifyingSettableFuture clientSubmitFuture, + private static void handleException(final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, final Collection cohorts, final String phase, final TransactionCommitFailedExceptionMapper exMapper, final Throwable t) { - if(clientSubmitFuture.isDone()) { + if (clientSubmitFuture.isDone()) { // We must have had failures from multiple cohorts. return; } LOG.warn("Tx: {} Error during phase {}, starting Abort", transaction.getIdentifier(), phase, t); - Exception e; - if(t instanceof Exception) { + final Exception e; + if(t instanceof NoShardLeaderException) { + e = new DataStoreUnavailableException(t.getMessage(), t); + } else if (t instanceof Exception) { e = (Exception)t; } else { e = new RuntimeException("Unexpected error occurred", t); @@ -206,7 +212,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { @SuppressWarnings("unchecked") ListenableFuture[] canCommitFutures = new ListenableFuture[cohorts.size()]; int i = 0; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { + for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { canCommitFutures[i++] = cohort.abort(); } @@ -226,7 +232,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { // what's interesting to the client. clientSubmitFuture.setException(clientException); } - }, internalFutureCallbackExecutor); + }, MoreExecutors.directExecutor()); } /** @@ -247,10 +253,10 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { */ private static final ThreadLocal ON_TASK_COMPLETION_THREAD_TL = new ThreadLocal(); - private final ExecutorService listenerExecutor; + private final Executor listenerExecutor; - AsyncNotifyingSettableFuture(ExecutorService listenerExecutor) { - this.listenerExecutor = listenerExecutor; + AsyncNotifyingSettableFuture(Executor listenerExecutor) { + this.listenerExecutor = Preconditions.checkNotNull(listenerExecutor); } @Override @@ -311,41 +317,4 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { } } } - - /** - * A simple same-thread executor without the internal locking overhead that - * MoreExecutors#sameThreadExecutor has. The #execute method is the only one of concern - we - * don't shutdown the executor so the other methods irrelevant. - */ - private static class SimpleSameThreadExecutor extends AbstractListeningExecutorService { - - @Override - public void execute(Runnable command) { - command.run(); - } - - @Override - public boolean awaitTermination(long arg0, TimeUnit arg1) throws InterruptedException { - return true; - } - - @Override - public boolean isShutdown() { - return false; - } - - @Override - public boolean isTerminated() { - return false; - } - - @Override - public void shutdown() { - } - - @Override - public List shutdownNow() { - return null; - } - } }