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=238e32e60fec6884ee961aaa0d9e2e568c1c9a2b;hp=bef37980e5cb6b13c569fe8fb45eb3609609775e;hb=5f2e2dc8b9162844bfe6cd4a7fca30baa386f1b4;hpb=cbe83ca3074fa0182d4f079f528bb710a997ced7 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..238e32e60f 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,28 @@ */ 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 +38,24 @@ 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.NotificationManager; +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 * @@ -61,16 +68,45 @@ import static com.google.common.base.Preconditions.checkState; public class InMemoryDOMDataStore implements DOMStore, Identifiable, SchemaContextListener, TransactionReadyPrototype,AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class); + + 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 ListeningExecutorService listeningExecutor; + + private final NotificationManager, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager; + private final ExecutorService dataChangeListenerExecutor; private final String name; - public InMemoryDOMDataStore(final String name, final ListeningExecutorService executor) { + public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor, + final ExecutorService dataChangeListenerExecutor) { + this(name, listeningExecutor, dataChangeListenerExecutor, + InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE); + } + + public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor, + final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize) { this.name = Preconditions.checkNotNull(name); - this.executor = Preconditions.checkNotNull(executor); + this.listeningExecutor = Preconditions.checkNotNull(listeningExecutor); + this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor); + + dataChangeListenerNotificationManager = + new QueuedNotificationManager<>(this.dataChangeListenerExecutor, + DCL_NOTIFICATION_MGR_INVOKER, maxDataChangeListenerQueueSize, + "DataChangeListenerQueueMgr"); } @Override @@ -104,12 +140,13 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch } @Override - public void close(){ - executor.shutdownNow(); + public void close() { + ExecutorServiceUtil.tryGracefulShutdown(listeningExecutor, 30, TimeUnit.SECONDS); + ExecutorServiceUtil.tryGracefulShutdown(dataChangeListenerExecutor, 30, TimeUnit.SECONDS); } @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 +169,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,7 +185,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch } @Override - public synchronized DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction writeTx) { + public DOMStoreThreePhaseCommitCohort ready(final SnapshotBackedWriteTransaction writeTx) { LOG.debug("Tx: {} is submitted. Modifications: {}", writeTx.getIdentifier(), writeTx.getMutatedView()); return new ThreePhaseCommitImpl(writeTx); } @@ -221,8 +259,9 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public void close() { - executor.shutdownNow(); - + // 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, @@ -232,7 +271,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch } public synchronized void onTransactionCommited(final SnapshotBackedWriteTransaction transaction) { - // If commited transaction is latestOutstandingTx we clear + // If committed transaction is latestOutstandingTx we clear // latestOutstandingTx // field in order to base new transactions on Datastore Data Tree // directly. @@ -308,7 +347,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public ListenableFuture canCommit() { - return executor.submit(new Callable() { + return listeningExecutor.submit(new Callable() { @Override public Boolean call() throws TransactionCommitFailedException { try { @@ -330,7 +369,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public ListenableFuture preCommit() { - return executor.submit(new Callable() { + return listeningExecutor.submit(new Callable() { @Override public Void call() { candidate = dataTree.prepare(modification); @@ -356,11 +395,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch */ synchronized (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);