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=538f2981daae891406934ffcf9605a9aecdd72c2;hp=13334c927236cdbeba31bf82ee10fb4f7d0cc952;hb=5f553b1657b97adbeb59b8e346d6eed8148b281b;hpb=78aa007feef2169a0280bd3d3fbbbea160b63d82 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 13334c9272..538f2981da 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 @@ -8,19 +8,19 @@ package org.opendaylight.controller.cluster.datastore; import com.google.common.base.Preconditions; -import com.google.common.collect.Iterables; 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 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.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; @@ -70,17 +70,16 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { @Override public CheckedFuture submit(DOMDataWriteTransaction transaction, - Iterable cohorts) { + 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()); - final int cohortSize = Iterables.size(cohorts); final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); - doCanCommit(clientSubmitFuture, transaction, cohorts, cohortSize); + doCanCommit(clientSubmitFuture, transaction, cohorts); return MappingCheckedFuture.create(clientSubmitFuture, TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER); @@ -88,102 +87,108 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { private void doCanCommit(final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, - final Iterable cohorts, final int cohortSize) { + final Collection cohorts) { 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(cohortSize); FutureCallback futureCallback = new FutureCallback() { @Override public void onSuccess(Boolean result) { if (result == null || !result) { - handleException(clientSubmitFuture, transaction, cohorts, cohortSize, + handleException(clientSubmitFuture, transaction, cohorts, CAN_COMMIT, TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER, 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, cohortSize); + doPreCommit(startTime, clientSubmitFuture, transaction, cohorts); + } else { + ListenableFuture canCommitFuture = cohortIterator.next().canCommit(); + Futures.addCallback(canCommitFuture, this, internalFutureCallbackExecutor); } } } @Override public void onFailure(Throwable t) { - handleException(clientSubmitFuture, transaction, cohorts, cohortSize, CAN_COMMIT, + handleException(clientSubmitFuture, transaction, cohorts, CAN_COMMIT, TransactionCommitFailedExceptionMapper.CAN_COMMIT_ERROR_MAPPER, t); } }; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { - ListenableFuture canCommitFuture = cohort.canCommit(); - Futures.addCallback(canCommitFuture, futureCallback, internalFutureCallbackExecutor); - } + ListenableFuture canCommitFuture = cohortIterator.next().canCommit(); + Futures.addCallback(canCommitFuture, futureCallback, internalFutureCallbackExecutor); } private void doPreCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, - final Iterable cohorts, final int cohortSize) { + final Collection cohorts) { + + final Iterator cohortIterator = cohorts.iterator(); // Not using Futures.allAsList here to avoid its internal overhead. - final AtomicInteger remaining = new AtomicInteger(cohortSize); 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, cohortSize); + doCommit(startTime, clientSubmitFuture, transaction, cohorts); + } else { + ListenableFuture preCommitFuture = cohortIterator.next().preCommit(); + Futures.addCallback(preCommitFuture, this, internalFutureCallbackExecutor); } } @Override public void onFailure(Throwable t) { - handleException(clientSubmitFuture, transaction, cohorts, cohortSize, PRE_COMMIT, + handleException(clientSubmitFuture, transaction, cohorts, PRE_COMMIT, TransactionCommitFailedExceptionMapper.PRE_COMMIT_MAPPER, t); } }; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { - ListenableFuture preCommitFuture = cohort.preCommit(); - Futures.addCallback(preCommitFuture, futureCallback, internalFutureCallbackExecutor); - } + ListenableFuture preCommitFuture = cohortIterator.next().preCommit(); + Futures.addCallback(preCommitFuture, futureCallback, internalFutureCallbackExecutor); } private void doCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, - final Iterable cohorts, final int cohortSize) { + final Collection cohorts) { + + final Iterator cohortIterator = cohorts.iterator(); // Not using Futures.allAsList here to avoid its internal overhead. - final AtomicInteger remaining = new AtomicInteger(cohortSize); 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, internalFutureCallbackExecutor); } } @Override public void onFailure(Throwable t) { - handleException(clientSubmitFuture, transaction, cohorts, cohortSize, COMMIT, + handleException(clientSubmitFuture, transaction, cohorts, COMMIT, TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER, t); } }; - for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { - ListenableFuture commitFuture = cohort.commit(); - Futures.addCallback(commitFuture, futureCallback, internalFutureCallbackExecutor); - } + ListenableFuture commitFuture = cohortIterator.next().commit(); + Futures.addCallback(commitFuture, futureCallback, internalFutureCallbackExecutor); } private void handleException(final AsyncNotifyingSettableFuture clientSubmitFuture, final DOMDataWriteTransaction transaction, - final Iterable cohorts, int cohortSize, + final Collection cohorts, final String phase, final TransactionCommitFailedExceptionMapper exMapper, final Throwable t) { @@ -205,7 +210,7 @@ public class ConcurrentDOMDataBroker extends AbstractDOMDataBroker { // Transaction failed - tell all cohorts to abort. @SuppressWarnings("unchecked") - ListenableFuture[] canCommitFutures = new ListenableFuture[cohortSize]; + ListenableFuture[] canCommitFutures = new ListenableFuture[cohorts.size()]; int i = 0; for(DOMStoreThreePhaseCommitCohort cohort: cohorts) { canCommitFutures[i++] = cohort.abort();