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=b48e5946da46ecdc00b0a4c82dc5263ffcd13efc;hb=340a2d4c979ac6f8d5adff8bd9e1c9f724e7a164;hpb=412db94945c5db5d2da918f5e23bd3abcecc4d10 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 b48e5946da..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,19 +9,19 @@ 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 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; @@ -51,15 +51,9 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { /** * 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,7 +63,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { } @Override - public CheckedFuture submit(DOMDataWriteTransaction transaction, + protected CheckedFuture submit(DOMDataWriteTransaction transaction, Collection cohorts) { Preconditions.checkArgument(transaction != null, "Transaction must not be null."); @@ -112,7 +106,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { doPreCommit(startTime, clientSubmitFuture, transaction, cohorts); } else { ListenableFuture canCommitFuture = cohortIterator.next().canCommit(); - Futures.addCallback(canCommitFuture, this, internalFutureCallbackExecutor); + Futures.addCallback(canCommitFuture, this, MoreExecutors.directExecutor()); } } } @@ -125,7 +119,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { }; ListenableFuture canCommitFuture = cohortIterator.next().canCommit(); - Futures.addCallback(canCommitFuture, futureCallback, internalFutureCallbackExecutor); + Futures.addCallback(canCommitFuture, futureCallback, MoreExecutors.directExecutor()); } private void doPreCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture, @@ -143,7 +137,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { doCommit(startTime, clientSubmitFuture, transaction, cohorts); } else { ListenableFuture preCommitFuture = cohortIterator.next().preCommit(); - Futures.addCallback(preCommitFuture, this, internalFutureCallbackExecutor); + Futures.addCallback(preCommitFuture, this, MoreExecutors.directExecutor()); } } @@ -155,7 +149,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { }; ListenableFuture preCommitFuture = cohortIterator.next().preCommit(); - Futures.addCallback(preCommitFuture, futureCallback, internalFutureCallbackExecutor); + Futures.addCallback(preCommitFuture, futureCallback, MoreExecutors.directExecutor()); } private void doCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture, @@ -175,7 +169,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { clientSubmitFuture.set(); } else { ListenableFuture commitFuture = cohortIterator.next().commit(); - Futures.addCallback(commitFuture, this, internalFutureCallbackExecutor); + Futures.addCallback(commitFuture, this, MoreExecutors.directExecutor()); } } @@ -187,23 +181,25 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { }; ListenableFuture commitFuture = cohortIterator.next().commit(); - Futures.addCallback(commitFuture, futureCallback, internalFutureCallbackExecutor); + 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); @@ -216,7 +212,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { @SuppressWarnings("unchecked") ListenableFuture[] canCommitFutures = new ListenableFuture[cohorts.size()]; int i = 0; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { + for (DOMStoreThreePhaseCommitCohort cohort : cohorts) { canCommitFutures[i++] = cohort.abort(); } @@ -236,7 +232,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { // what's interesting to the client. clientSubmitFuture.setException(clientException); } - }, internalFutureCallbackExecutor); + }, MoreExecutors.directExecutor()); } /** @@ -257,10 +253,10 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { */ 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 @@ -321,41 +317,4 @@ public class ConcurrentDOMDataBroker extends AbstractDOMBroker { } } } - - /** - * 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; - } - } }