X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-inmemory-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fstore%2Fimpl%2FInMemoryDOMDataStore.java;h=b617a8087fc771d6f413b1ec6c3dc4e5bf597aa3;hp=213f60e951cc41794864fcb1fd813a1958469728;hb=8edd92f999e5ff4588d284c3a1ae553549b84fb4;hpb=0eba94d9411ea40945ddc8c732640c0cc004599f diff --git a/opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java b/opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java index 213f60e951..b617a8087f 100644 --- a/opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java +++ b/opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java @@ -12,8 +12,6 @@ import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.ListeningExecutorService; -import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; @@ -21,13 +19,16 @@ import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataCh import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener; import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException; import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; +import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener; import org.opendaylight.controller.md.sal.dom.store.impl.SnapshotBackedWriteTransaction.TransactionReadyPrototype; import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree; +import org.opendaylight.controller.sal.core.spi.data.AbstractDOMStoreTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStore; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; +import org.opendaylight.controller.sal.core.spi.data.DOMStoreTreeChangePublisher; import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; import org.opendaylight.yangtools.concepts.AbstractListenerRegistration; import org.opendaylight.yangtools.concepts.Identifiable; @@ -58,9 +59,10 @@ import org.slf4j.LoggerFactory; * to implement {@link DOMStore} contract. * */ -public class InMemoryDOMDataStore extends TransactionReadyPrototype implements DOMStore, Identifiable, SchemaContextListener, AutoCloseable { +public class InMemoryDOMDataStore extends TransactionReadyPrototype implements DOMStore, Identifiable, SchemaContextListener, AutoCloseable, DOMStoreTreeChangePublisher { private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class); private static final ListenableFuture SUCCESSFUL_FUTURE = Futures.immediateFuture(null); + private static final ListenableFuture CAN_COMMIT_FUTURE = Futures.immediateFuture(Boolean.TRUE); private static final Invoker, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER = new Invoker, DOMImmutableDataChangeEvent>() { @@ -79,24 +81,20 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D private final AtomicLong txCounter = new AtomicLong(0); private final QueuedNotificationManager, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager; + private final InMemoryDOMStoreTreeChangePublisher changePublisher; private final ExecutorService dataChangeListenerExecutor; - private final ListeningExecutorService commitExecutor; private final boolean debugTransactions; private final String name; private volatile AutoCloseable closeable; - public InMemoryDOMDataStore(final String name, final ListeningExecutorService commitExecutor, - final ExecutorService dataChangeListenerExecutor) { - this(name, commitExecutor, dataChangeListenerExecutor, - InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE, false); + public InMemoryDOMDataStore(final String name, final ExecutorService dataChangeListenerExecutor) { + this(name, dataChangeListenerExecutor, InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE, false); } - public InMemoryDOMDataStore(final String name, final ListeningExecutorService commitExecutor, - final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize, - final boolean debugTransactions) { + public InMemoryDOMDataStore(final String name, final ExecutorService dataChangeListenerExecutor, + final int maxDataChangeListenerQueueSize, final boolean debugTransactions) { this.name = Preconditions.checkNotNull(name); - this.commitExecutor = Preconditions.checkNotNull(commitExecutor); this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor); this.debugTransactions = debugTransactions; @@ -104,6 +102,7 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D new QueuedNotificationManager<>(this.dataChangeListenerExecutor, DCL_NOTIFICATION_MGR_INVOKER, maxDataChangeListenerQueueSize, "DataChangeListenerQueueMgr"); + changePublisher = new InMemoryDOMStoreTreeChangePublisher(this.dataChangeListenerExecutor, maxDataChangeListenerQueueSize); } public void setCloseable(final AutoCloseable closeable) { @@ -114,10 +113,6 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D return dataChangeListenerNotificationManager; } - public ExecutorService getDomStoreExecutor() { - return commitExecutor; - } - @Override public final String getIdentifier() { return name; @@ -150,7 +145,6 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D @Override public void close() { - ExecutorServiceUtil.tryGracefulShutdown(commitExecutor, 30, TimeUnit.SECONDS); ExecutorServiceUtil.tryGracefulShutdown(dataChangeListenerExecutor, 30, TimeUnit.SECONDS); if(closeable != null) { @@ -210,6 +204,11 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D }; } + @Override + public ListenerRegistration registerTreeChangeListener(final YangInstanceIdentifier treeId, final L listener) { + return changePublisher.registerTreeChangeListener(treeId, listener); + } + @Override protected void transactionAborted(final SnapshotBackedWriteTransaction tx) { LOG.debug("Tx: {} is closed.", tx.getIdentifier()); @@ -225,6 +224,13 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D return name + "-" + txCounter.getAndIncrement(); } + private static void warnDebugContext(AbstractDOMStoreTransaction transaction) { + final Throwable ctx = transaction.getDebugContext(); + if (ctx != null) { + LOG.warn("Transaction {} has been allocated in the following context", transaction.getIdentifier(), ctx); + } + } + private final class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort { private final SnapshotBackedWriteTransaction transaction; private final DataTreeModification modification; @@ -239,38 +245,41 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D @Override public ListenableFuture canCommit() { - return commitExecutor.submit(new Callable() { - @Override - public Boolean call() throws TransactionCommitFailedException { - try { - dataTree.validate(modification); - LOG.debug("Store Transaction: {} can be committed", transaction.getIdentifier()); - return true; - } catch (ConflictingModificationAppliedException e) { - LOG.warn("Store Tx: {} Conflicting modification for {}.", transaction.getIdentifier(), - e.getPath()); - transaction.warnDebugContext(LOG); - throw new OptimisticLockFailedException("Optimistic lock failed.",e); - } catch (DataValidationFailedException e) { - LOG.warn("Store Tx: {} Data Precondition failed for {}.", transaction.getIdentifier(), - e.getPath(), e); - transaction.warnDebugContext(LOG); - throw new TransactionCommitFailedException("Data did not pass validation.",e); - } - } - }); + try { + dataTree.validate(modification); + LOG.debug("Store Transaction: {} can be committed", transaction.getIdentifier()); + return CAN_COMMIT_FUTURE; + } catch (ConflictingModificationAppliedException e) { + LOG.warn("Store Tx: {} Conflicting modification for {}.", transaction.getIdentifier(), + e.getPath()); + warnDebugContext(transaction); + return Futures.immediateFailedFuture(new OptimisticLockFailedException("Optimistic lock failed.", e)); + } catch (DataValidationFailedException e) { + LOG.warn("Store Tx: {} Data Precondition failed for {}.", transaction.getIdentifier(), + e.getPath(), e); + warnDebugContext(transaction); + + // For debugging purposes, allow dumping of the modification. Coupled with the above + // precondition log, it should allow us to understand what went on. + LOG.trace("Store Tx: {} modifications: {} tree: {}", modification, dataTree); + + return Futures.immediateFailedFuture(new TransactionCommitFailedException("Data did not pass validation.", e)); + } catch (Exception e) { + LOG.warn("Unexpected failure in validation phase", e); + return Futures.immediateFailedFuture(e); + } } @Override public ListenableFuture preCommit() { - return commitExecutor.submit(new Callable() { - @Override - public Void call() { - candidate = dataTree.prepare(modification); - listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree); - return null; - } - }); + try { + candidate = dataTree.prepare(modification); + listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree); + return SUCCESSFUL_FUTURE; + } catch (Exception e) { + LOG.warn("Unexpected failure in pre-commit phase", e); + return Futures.immediateFailedFuture(e); + } } @Override @@ -289,6 +298,7 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D */ synchronized (InMemoryDOMDataStore.this) { dataTree.commit(candidate); + changePublisher.publishChange(candidate); listenerResolver.resolve(dataChangeListenerNotificationManager); }