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=7e4056a1900b99a777096910b32e6b37d0012574;hp=deddd6938ae477eeebbd9fdeb50c79b6d3694eab;hb=978152c5de3bf78ab6da5da4c2db391eec063429;hpb=3591817114661bb7971d6d355186ff1b39636fcd 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 deddd6938a..7e4056a190 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 @@ -19,14 +19,19 @@ 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.store.impl.SnapshotBackedWriteTransaction.TransactionReadyPrototype; +import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener; 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.controller.sal.core.spi.data.SnapshotBackedTransactions; +import org.opendaylight.controller.sal.core.spi.data.SnapshotBackedWriteTransaction; +import org.opendaylight.controller.sal.core.spi.data.SnapshotBackedWriteTransaction.TransactionReadyPrototype; import org.opendaylight.yangtools.concepts.AbstractListenerRegistration; import org.opendaylight.yangtools.concepts.Identifiable; import org.opendaylight.yangtools.concepts.ListenerRegistration; @@ -52,11 +57,11 @@ import org.slf4j.LoggerFactory; * * Implementation of {@link DOMStore} which uses {@link DataTree} and other * classes such as {@link SnapshotBackedWriteTransaction}. - * {@link SnapshotBackedReadTransaction} and {@link ResolveDataChangeEventsTask} + * {@link org.opendaylight.controller.sal.core.spi.data.SnapshotBackedReadTransaction} and {@link ResolveDataChangeEventsTask} * 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); @@ -78,6 +83,7 @@ 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 boolean debugTransactions; private final String name; @@ -98,6 +104,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) { @@ -115,17 +122,17 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D @Override public DOMStoreReadTransaction newReadOnlyTransaction() { - return new SnapshotBackedReadTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot()); + return SnapshotBackedTransactions.newReadTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot()); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new SnapshotBackedReadWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); + return SnapshotBackedTransactions.newReadWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new SnapshotBackedWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); + return SnapshotBackedTransactions.newWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); } @Override @@ -200,28 +207,44 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D } @Override - protected void transactionAborted(final SnapshotBackedWriteTransaction tx) { + public synchronized ListenerRegistration registerTreeChangeListener(final YangInstanceIdentifier treeId, final L listener) { + /* + * Make sure commit is not occurring right now. Listener has to be + * registered and its state capture enqueued at a consistent point. + */ + return changePublisher.registerTreeChangeListener(treeId, listener, dataTree.takeSnapshot()); + } + + @Override + protected void transactionAborted(final SnapshotBackedWriteTransaction tx) { LOG.debug("Tx: {} is closed.", tx.getIdentifier()); } @Override - protected DOMStoreThreePhaseCommitCohort transactionReady(final SnapshotBackedWriteTransaction tx, final DataTreeModification tree) { + protected DOMStoreThreePhaseCommitCohort transactionReady(final SnapshotBackedWriteTransaction tx, final DataTreeModification tree) { LOG.debug("Tx: {} is submitted. Modifications: {}", tx.getIdentifier(), tree); return new ThreePhaseCommitImpl(tx, tree); } - Object nextIdentifier() { + String nextIdentifier() { return name + "-" + txCounter.getAndIncrement(); } + private static void warnDebugContext(final 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 SnapshotBackedWriteTransaction transaction; private final DataTreeModification modification; private ResolveDataChangeEventsTask listenerResolver; private DataTreeCandidate candidate; - public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction, final DataTreeModification modification) { + public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction, final DataTreeModification modification) { this.transaction = writeTransaction; this.modification = modification; } @@ -235,12 +258,12 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D } catch (ConflictingModificationAppliedException e) { LOG.warn("Store Tx: {} Conflicting modification for {}.", transaction.getIdentifier(), e.getPath()); - transaction.warnDebugContext(LOG); + 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); - transaction.warnDebugContext(LOG); + 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. @@ -281,6 +304,7 @@ public class InMemoryDOMDataStore extends TransactionReadyPrototype implements D */ synchronized (InMemoryDOMDataStore.this) { dataTree.commit(candidate); + changePublisher.publishChange(candidate); listenerResolver.resolve(dataChangeListenerNotificationManager); }