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=3e748618169889cdfcf0b9f8eb66c73e65cb4338;hp=129013378ea57c713304793248e1cdc2f2c0a6ce;hb=7292faba613ab556babd7bbcdd78984f5668bf9b;hpb=5952570885060553b7f747632c657659b9417aa4 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 129013378e..3582ff6ef1 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 @@ -7,111 +7,114 @@ */ package org.opendaylight.controller.md.sal.dom.store.impl; -import com.google.common.base.Optional; import com.google.common.base.Preconditions; -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.ListeningExecutorService; - +import java.util.Optional; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; -import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent; 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.yangtools.util.ExecutorServiceUtil; -import org.opendaylight.yangtools.util.concurrent.NotificationManager; -import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager; -import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; +import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; +import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener; import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree; -import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory; 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; +import org.opendaylight.yangtools.util.ExecutorServiceUtil; +import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager; +import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager.Invoker; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; +import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.opendaylight.yangtools.yang.model.api.SchemaContextListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.concurrent.GuardedBy; - -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; - -import static com.google.common.base.Preconditions.checkState; - /** * In-memory DOM Data Store * * 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 implements DOMStore, Identifiable, SchemaContextListener, - TransactionReadyPrototype,AutoCloseable { +public class InMemoryDOMDataStore extends TransactionReadyPrototype implements DOMStore, Identifiable, SchemaContextListener, AutoCloseable, DOMStoreTreeChangePublisher { private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class); - @SuppressWarnings("rawtypes") - private static final QueuedNotificationManager.Invoker DCL_NOTIFICATION_MGR_INVOKER = - new QueuedNotificationManager.Invoker() { - - @SuppressWarnings("unchecked") - @Override - public void invokeListener( AsyncDataChangeListener listener, - AsyncDataChangeEvent notification ) { - listener.onDataChanged(notification); + private static final Invoker, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER = + (listener, notification) -> { + final AsyncDataChangeListener> inst = listener.getInstance(); + if (inst != null) { + inst.onDataChanged(notification); } }; - private final DataTree dataTree = InMemoryDataTreeFactory.getInstance().create(); + private final DataTree dataTree; private final ListenerTree listenerTree = ListenerTree.create(); private final AtomicLong txCounter = new AtomicLong(0); - private final ListeningExecutorService listeningExecutor; - @SuppressWarnings("rawtypes") - private final NotificationManager - dataChangeListenerNotificationManager; + private final QueuedNotificationManager, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager; + private final InMemoryDOMStoreTreeChangePublisher changePublisher; private final ExecutorService dataChangeListenerExecutor; - + private final boolean debugTransactions; private final String name; - public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor, - final ExecutorService dataChangeListenerExecutor) { - this(name, listeningExecutor, dataChangeListenerExecutor, - InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE); + private volatile AutoCloseable closeable; + + public InMemoryDOMDataStore(final String name, final ExecutorService dataChangeListenerExecutor) { + this(name, LogicalDatastoreType.OPERATIONAL, dataChangeListenerExecutor, + InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE, false); } - public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor, - final ExecutorService dataChangeListenerExecutor, int maxDataChangeListenerQueueSize) { + public InMemoryDOMDataStore(final String name, final LogicalDatastoreType type, + final ExecutorService dataChangeListenerExecutor, + final int maxDataChangeListenerQueueSize, final boolean debugTransactions) { this.name = Preconditions.checkNotNull(name); - this.listeningExecutor = Preconditions.checkNotNull(listeningExecutor); - this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor); + this.debugTransactions = debugTransactions; dataChangeListenerNotificationManager = new QueuedNotificationManager<>(this.dataChangeListenerExecutor, DCL_NOTIFICATION_MGR_INVOKER, maxDataChangeListenerQueueSize, "DataChangeListenerQueueMgr"); + changePublisher = new InMemoryDOMStoreTreeChangePublisher(this.dataChangeListenerExecutor, maxDataChangeListenerQueueSize); + + switch (type) { + case CONFIGURATION: + dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_CONFIGURATION); + break; + case OPERATIONAL: + dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL); + break; + default: + throw new IllegalArgumentException("Data store " + type + " not supported"); + } + } + + public void setCloseable(final AutoCloseable closeable) { + this.closeable = closeable; + } + + public QueuedNotificationManager getDataChangeListenerNotificationManager() { + return dataChangeListenerNotificationManager; } @Override @@ -121,22 +124,22 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public DOMStoreReadTransaction newReadOnlyTransaction() { - return new SnapshotBackedReadTransaction(nextIdentifier(), dataTree.takeSnapshot()); + return SnapshotBackedTransactions.newReadTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot()); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new SnapshotBackedReadWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this); + return SnapshotBackedTransactions.newReadWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new SnapshotBackedWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this); + return SnapshotBackedTransactions.newWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); } @Override public DOMStoreTransactionChain createTransactionChain() { - return new DOMStoreTransactionChainImpl(); + return new DOMStoreTransactionChainImpl(this); } @Override @@ -146,9 +149,25 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public void close() { - ExecutorServiceUtil.tryGracefulShutdown(listeningExecutor, 30, TimeUnit.SECONDS); ExecutorServiceUtil.tryGracefulShutdown(dataChangeListenerExecutor, 30, TimeUnit.SECONDS); + + if(closeable != null) { + try { + closeable.close(); + } catch(Exception e) { + LOG.debug("Error closing instance", e); + } + } + } + + public final boolean getDebugTransactions() { + return debugTransactions; } + + final DataTreeSnapshot takeSnapshot() { + return dataTree.takeSnapshot(); + } + @Override public >> ListenerRegistration registerChangeListener( final YangInstanceIdentifier path, final L listener, final DataChangeScope scope) { @@ -175,8 +194,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch .addCreated(path, data) // .build(); - new ChangeListenerNotifyTask(reg, event, - dataChangeListenerNotificationManager).run(); + dataChangeListenerNotificationManager.submitNotification(reg, event); } } @@ -191,225 +209,42 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch } @Override - public synchronized DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction writeTx) { - LOG.debug("Tx: {} is submitted. Modifications: {}", writeTx.getIdentifier(), writeTx.getMutatedView()); - return new ThreePhaseCommitImpl(writeTx); + 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()); } - private Object nextIdentifier() { - return name + "-" + txCounter.getAndIncrement(); + @Override + protected void transactionAborted(final SnapshotBackedWriteTransaction tx) { + LOG.debug("Tx: {} is closed.", tx.getIdentifier()); } - private class DOMStoreTransactionChainImpl implements DOMStoreTransactionChain, TransactionReadyPrototype { - - @GuardedBy("this") - private SnapshotBackedWriteTransaction latestOutstandingTx; - - private boolean chainFailed = false; - - private void checkFailed() { - Preconditions.checkState(!chainFailed, "Transaction chain is failed."); - } - - @Override - public synchronized DOMStoreReadTransaction newReadOnlyTransaction() { - final DataTreeSnapshot snapshot; - checkFailed(); - if (latestOutstandingTx != null) { - checkState(latestOutstandingTx.isReady(), "Previous transaction in chain must be ready."); - snapshot = latestOutstandingTx.getMutatedView(); - } else { - snapshot = dataTree.takeSnapshot(); - } - return new SnapshotBackedReadTransaction(nextIdentifier(), snapshot); - } - - @Override - public synchronized DOMStoreReadWriteTransaction newReadWriteTransaction() { - final DataTreeSnapshot snapshot; - checkFailed(); - if (latestOutstandingTx != null) { - checkState(latestOutstandingTx.isReady(), "Previous transaction in chain must be ready."); - snapshot = latestOutstandingTx.getMutatedView(); - } else { - snapshot = dataTree.takeSnapshot(); - } - final SnapshotBackedReadWriteTransaction ret = new SnapshotBackedReadWriteTransaction(nextIdentifier(), - snapshot, this); - latestOutstandingTx = ret; - return ret; - } - - @Override - public synchronized DOMStoreWriteTransaction newWriteOnlyTransaction() { - final DataTreeSnapshot snapshot; - checkFailed(); - if (latestOutstandingTx != null) { - checkState(latestOutstandingTx.isReady(), "Previous transaction in chain must be ready."); - snapshot = latestOutstandingTx.getMutatedView(); - } else { - snapshot = dataTree.takeSnapshot(); - } - final SnapshotBackedWriteTransaction ret = new SnapshotBackedWriteTransaction(nextIdentifier(), snapshot, - this); - latestOutstandingTx = ret; - return ret; - } - - @Override - public DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction tx) { - DOMStoreThreePhaseCommitCohort storeCohort = InMemoryDOMDataStore.this.ready(tx); - return new ChainedTransactionCommitImpl(tx, storeCohort, this); - } - - @Override - public void close() { - - // FIXME: this call doesn't look right here - listeningExecutor is shared and owned - // by the outer class. - //listeningExecutor.shutdownNow(); - } - - protected synchronized void onTransactionFailed(final SnapshotBackedWriteTransaction transaction, - final Throwable t) { - chainFailed = true; - - } - - public synchronized void onTransactionCommited(final SnapshotBackedWriteTransaction transaction) { - // If commited transaction is latestOutstandingTx we clear - // latestOutstandingTx - // field in order to base new transactions on Datastore Data Tree - // directly. - if (transaction.equals(latestOutstandingTx)) { - latestOutstandingTx = null; - } - } - + @Override + protected DOMStoreThreePhaseCommitCohort transactionReady(final SnapshotBackedWriteTransaction tx, + final DataTreeModification modification, + final Exception readyError) { + LOG.debug("Tx: {} is submitted. Modifications: {}", tx.getIdentifier(), modification); + return new InMemoryDOMStoreThreePhaseCommitCohort(this, tx, modification, readyError); } - private static class ChainedTransactionCommitImpl implements DOMStoreThreePhaseCommitCohort { - - private final SnapshotBackedWriteTransaction transaction; - private final DOMStoreThreePhaseCommitCohort delegate; - - private final DOMStoreTransactionChainImpl txChain; - - protected ChainedTransactionCommitImpl(final SnapshotBackedWriteTransaction transaction, - final DOMStoreThreePhaseCommitCohort delegate, final DOMStoreTransactionChainImpl txChain) { - super(); - this.transaction = transaction; - this.delegate = delegate; - this.txChain = txChain; - } - - @Override - public ListenableFuture canCommit() { - return delegate.canCommit(); - } - - @Override - public ListenableFuture preCommit() { - return delegate.preCommit(); - } - - @Override - public ListenableFuture abort() { - return delegate.abort(); - } - - @Override - public ListenableFuture commit() { - ListenableFuture commitFuture = delegate.commit(); - Futures.addCallback(commitFuture, new FutureCallback() { - @Override - public void onFailure(final Throwable t) { - txChain.onTransactionFailed(transaction, t); - } - - @Override - public void onSuccess(final Void result) { - txChain.onTransactionCommited(transaction); - } - - }); - return commitFuture; - } - + String nextIdentifier() { + return name + "-" + txCounter.getAndIncrement(); } - private class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort { - - private final SnapshotBackedWriteTransaction transaction; - private final DataTreeModification modification; - - private ResolveDataChangeEventsTask listenerResolver; - private DataTreeCandidate candidate; - - public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction) { - this.transaction = writeTransaction; - this.modification = transaction.getMutatedView(); - } - - @Override - public ListenableFuture canCommit() { - return listeningExecutor.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()); - throw new OptimisticLockFailedException("Optimistic lock failed.",e); - } catch (DataValidationFailedException e) { - LOG.warn("Store Tx: {} Data Precondition failed for {}.", transaction.getIdentifier(), - e.getPath(), e); - throw new TransactionCommitFailedException("Data did not pass validation.",e); - } - } - }); - } - - @Override - public ListenableFuture preCommit() { - return listeningExecutor.submit(new Callable() { - @Override - public Void call() { - candidate = dataTree.prepare(modification); - listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree, - dataChangeListenerNotificationManager); - return null; - } - }); - } - - @Override - public ListenableFuture abort() { - candidate = null; - return Futures.immediateFuture(null); - } - - @Override - public ListenableFuture commit() { - checkState(candidate != null, "Proposed subtree must be computed"); - - /* - * The commit has to occur atomically with regard to listener - * registrations. - */ - synchronized (this) { - dataTree.commit(candidate); + void validate(final DataTreeModification modification) throws DataValidationFailedException { + dataTree.validate(modification); + } - for (ChangeListenerNotifyTask task : listenerResolver.call()) { - LOG.trace("Scheduling invocation of listeners: {}", task); - task.run(); - } - } + DataTreeCandidate prepare(final DataTreeModification modification) { + return dataTree.prepare(modification); + } - return Futures.immediateFuture(null); - } + synchronized void commit(final DataTreeCandidate candidate) { + dataTree.commit(candidate); + changePublisher.publishChange(candidate); + ResolveDataChangeEventsTask.create(candidate, listenerTree).resolve(dataChangeListenerNotificationManager); } }