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=74fa73afb92f869f7cb2e945a625d489b71e71c2;hp=bef37980e5cb6b13c569fe8fb45eb3609609775e;hb=510f3bd2d25c0e667a60ff5ef18cfbd40534cec0;hpb=80aa861b74f7b0b3574f0962cdb45740ff71946c 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 bef37980e5..74fa73afb9 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,25 +7,24 @@ */ package org.opendaylight.controller.md.sal.dom.store.impl; +import static com.google.common.base.Preconditions.checkState; 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.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import javax.annotation.concurrent.GuardedBy; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope; 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.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.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; @@ -35,20 +34,23 @@ import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; import org.opendaylight.yangtools.concepts.AbstractListenerRegistration; import org.opendaylight.yangtools.concepts.Identifiable; import org.opendaylight.yangtools.concepts.ListenerRegistration; -import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier; +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.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.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.Collections; -import java.util.concurrent.Callable; -import java.util.concurrent.atomic.AtomicLong; - -import static com.google.common.base.Preconditions.checkState; - /** * In-memory DOM Data Store * @@ -58,19 +60,64 @@ import static com.google.common.base.Preconditions.checkState; * to implement {@link DOMStore} contract. * */ -public class InMemoryDOMDataStore implements DOMStore, Identifiable, SchemaContextListener, - TransactionReadyPrototype,AutoCloseable { +public class InMemoryDOMDataStore extends TransactionReadyPrototype implements DOMStore, Identifiable, SchemaContextListener, AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class); + private static final ListenableFuture SUCCESSFUL_FUTURE = Futures.immediateFuture(null); + + private static final Invoker, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER = + new Invoker, DOMImmutableDataChangeEvent>() { + @Override + public void invokeListener(final DataChangeListenerRegistration listener, + final DOMImmutableDataChangeEvent notification ) { + final AsyncDataChangeListener> inst = listener.getInstance(); + if (inst != null) { + inst.onDataChanged(notification); + } + } + }; + private final DataTree dataTree = InMemoryDataTreeFactory.getInstance().create(); private final ListenerTree listenerTree = ListenerTree.create(); private final AtomicLong txCounter = new AtomicLong(0); - private final ListeningExecutorService executor; + private final QueuedNotificationManager, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager; + private final ExecutorService dataChangeListenerExecutor; + private final ListeningExecutorService commitExecutor; + private final boolean debugTransactions; private final String name; - public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) { + 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 ListeningExecutorService commitExecutor, + final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize, + final boolean debugTransactions) { this.name = Preconditions.checkNotNull(name); - this.executor = Preconditions.checkNotNull(executor); + this.commitExecutor = Preconditions.checkNotNull(commitExecutor); + this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor); + this.debugTransactions = debugTransactions; + + dataChangeListenerNotificationManager = + new QueuedNotificationManager<>(this.dataChangeListenerExecutor, + DCL_NOTIFICATION_MGR_INVOKER, maxDataChangeListenerQueueSize, + "DataChangeListenerQueueMgr"); + } + + public void setCloseable(final AutoCloseable closeable) { + this.closeable = closeable; + } + + public QueuedNotificationManager getDataChangeListenerNotificationManager() { + return dataChangeListenerNotificationManager; + } + + public ExecutorService getDomStoreExecutor() { + return commitExecutor; } @Override @@ -80,17 +127,17 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public DOMStoreReadTransaction newReadOnlyTransaction() { - return new SnapshotBackedReadTransaction(nextIdentifier(), dataTree.takeSnapshot()); + return new SnapshotBackedReadTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot()); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new SnapshotBackedReadWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this); + return new SnapshotBackedReadWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new SnapshotBackedWriteTransaction(nextIdentifier(), dataTree.takeSnapshot(), this); + return new SnapshotBackedWriteTransaction(nextIdentifier(), debugTransactions, dataTree.takeSnapshot(), this); } @Override @@ -104,12 +151,26 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch } @Override - public void close(){ - executor.shutdownNow(); + public void close() { + ExecutorServiceUtil.tryGracefulShutdown(commitExecutor, 30, TimeUnit.SECONDS); + ExecutorServiceUtil.tryGracefulShutdown(dataChangeListenerExecutor, 30, TimeUnit.SECONDS); + + if(closeable != null) { + try { + closeable.close(); + } catch(Exception e) { + LOG.debug("Error closing instance", e); + } + } } + + boolean getDebugTransactions() { + return debugTransactions; + } + @Override - public >> ListenerRegistration registerChangeListener( - final InstanceIdentifier path, final L listener, final DataChangeScope scope) { + public >> ListenerRegistration registerChangeListener( + final YangInstanceIdentifier path, final L listener, final DataChangeScope scope) { /* * Make sure commit is not occurring right now. Listener has to be @@ -132,7 +193,8 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch .setAfter(data) // .addCreated(path, data) // .build(); - executor.submit(new ChangeListenerNotifyTask(Collections.singletonList(reg), event)); + + dataChangeListenerNotificationManager.submitNotification(reg, event); } } @@ -147,112 +209,127 @@ 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); + protected void transactionAborted(final SnapshotBackedWriteTransaction tx) { + LOG.debug("Tx: {} is closed.", tx.getIdentifier()); + } + + @Override + protected DOMStoreThreePhaseCommitCohort transactionReady(final SnapshotBackedWriteTransaction tx, final DataTreeModification tree) { + LOG.debug("Tx: {} is submitted. Modifications: {}", tx.getIdentifier(), tree); + return new ThreePhaseCommitImpl(tx, tree); } private Object nextIdentifier() { return name + "-" + txCounter.getAndIncrement(); } - private class DOMStoreTransactionChainImpl implements DOMStoreTransactionChain, TransactionReadyPrototype { - + private class DOMStoreTransactionChainImpl extends TransactionReadyPrototype implements DOMStoreTransactionChain { + @GuardedBy("this") + private SnapshotBackedWriteTransaction allocatedTransaction; + @GuardedBy("this") + private DataTreeSnapshot readySnapshot; @GuardedBy("this") - private SnapshotBackedWriteTransaction latestOutstandingTx; - private boolean chainFailed = false; + @GuardedBy("this") private void checkFailed() { Preconditions.checkState(!chainFailed, "Transaction chain is failed."); } - @Override - public synchronized DOMStoreReadTransaction newReadOnlyTransaction() { - final DataTreeSnapshot snapshot; + @GuardedBy("this") + private DataTreeSnapshot getSnapshot() { checkFailed(); - if (latestOutstandingTx != null) { - checkState(latestOutstandingTx.isReady(), "Previous transaction in chain must be ready."); - snapshot = latestOutstandingTx.getMutatedView(); + + if (allocatedTransaction != null) { + Preconditions.checkState(readySnapshot != null, "Previous transaction %s is not ready yet", allocatedTransaction.getIdentifier()); + return readySnapshot; } else { - snapshot = dataTree.takeSnapshot(); + return dataTree.takeSnapshot(); } - return new SnapshotBackedReadTransaction(nextIdentifier(), snapshot); + } + + @GuardedBy("this") + private T recordTransaction(final T transaction) { + allocatedTransaction = transaction; + readySnapshot = null; + return transaction; + } + + @Override + public synchronized DOMStoreReadTransaction newReadOnlyTransaction() { + final DataTreeSnapshot snapshot = getSnapshot(); + return new SnapshotBackedReadTransaction(nextIdentifier(), getDebugTransactions(), 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; + final DataTreeSnapshot snapshot = getSnapshot(); + return recordTransaction(new SnapshotBackedReadWriteTransaction(nextIdentifier(), + getDebugTransactions(), snapshot, this)); } @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; + final DataTreeSnapshot snapshot = getSnapshot(); + return recordTransaction(new SnapshotBackedWriteTransaction(nextIdentifier(), + getDebugTransactions(), snapshot, this)); } @Override - public DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction tx) { - DOMStoreThreePhaseCommitCohort storeCohort = InMemoryDOMDataStore.this.ready(tx); - return new ChainedTransactionCommitImpl(tx, storeCohort, this); + protected synchronized void transactionAborted(final SnapshotBackedWriteTransaction tx) { + if (tx.equals(allocatedTransaction)) { + Preconditions.checkState(readySnapshot == null, "Unexpected abort of transaction %s with ready snapshot %s", tx, readySnapshot); + allocatedTransaction = null; + } } @Override - public void close() { + protected synchronized DOMStoreThreePhaseCommitCohort transactionReady(final SnapshotBackedWriteTransaction tx, final DataTreeModification tree) { + Preconditions.checkState(tx.equals(allocatedTransaction), "Mis-ordered ready transaction %s last allocated was %s", tx, allocatedTransaction); + if (readySnapshot != null) { + // The snapshot should have been cleared + LOG.warn("Uncleared snapshot {} encountered, overwritten with transaction {} snapshot {}", readySnapshot, tx, tree); + } - executor.shutdownNow(); + final DOMStoreThreePhaseCommitCohort cohort = InMemoryDOMDataStore.this.transactionReady(tx, tree); + readySnapshot = tree; + return new ChainedTransactionCommitImpl(tx, cohort, 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; + // If the committed transaction was the one we allocated last, + // we clear it and the ready snapshot, so the next transaction + // allocated refers to the data tree directly. + if (transaction.equals(allocatedTransaction)) { + if (readySnapshot == null) { + LOG.warn("Transaction {} committed while no ready snapshot present", transaction); + } + + allocatedTransaction = null; + readySnapshot = null; } } - } 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; @@ -286,29 +363,26 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch public void onSuccess(final Void result) { txChain.onTransactionCommited(transaction); } - }); return commitFuture; } - } private class ThreePhaseCommitImpl implements DOMStoreThreePhaseCommitCohort { - private final SnapshotBackedWriteTransaction transaction; private final DataTreeModification modification; private ResolveDataChangeEventsTask listenerResolver; private DataTreeCandidate candidate; - public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction) { + public ThreePhaseCommitImpl(final SnapshotBackedWriteTransaction writeTransaction, final DataTreeModification modification) { this.transaction = writeTransaction; - this.modification = transaction.getMutatedView(); + this.modification = modification; } @Override public ListenableFuture canCommit() { - return executor.submit(new Callable() { + return commitExecutor.submit(new Callable() { @Override public Boolean call() throws TransactionCommitFailedException { try { @@ -318,10 +392,12 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch } 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); } } @@ -330,7 +406,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public ListenableFuture preCommit() { - return executor.submit(new Callable() { + return commitExecutor.submit(new Callable() { @Override public Void call() { candidate = dataTree.prepare(modification); @@ -343,7 +419,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public ListenableFuture abort() { candidate = null; - return Futures.immediateFuture(null); + return SUCCESSFUL_FUTURE; } @Override @@ -354,16 +430,12 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch * The commit has to occur atomically with regard to listener * registrations. */ - synchronized (this) { + synchronized (InMemoryDOMDataStore.this) { dataTree.commit(candidate); - - for (ChangeListenerNotifyTask task : listenerResolver.call()) { - LOG.trace("Scheduling invocation of listeners: {}", task); - executor.submit(task); - } + listenerResolver.resolve(dataChangeListenerNotificationManager); } - return Futures.immediateFuture(null); + return SUCCESSFUL_FUTURE; } } }