X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-inmemory-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fstore%2Fimpl%2FInMemoryDOMDataStore.java;h=476356a19e79113b0b0121662b8842610757d12f;hb=f71a2c712690ecfd1260543ab58d8e16453f7918;hp=129013378ea57c713304793248e1cdc2f2c0a6ce;hpb=d5a00419b66920905e68ff857f44d3a0ab417db7;p=controller.git 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..476356a19e 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,30 +7,29 @@ */ 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 com.google.common.util.concurrent.MoreExecutors; + +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.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.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; @@ -40,22 +39,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.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.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 * @@ -68,18 +68,17 @@ 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 ListenableFuture SUCCESSFUL_FUTURE = Futures.immediateFuture(null); - @SuppressWarnings("rawtypes") - private static final QueuedNotificationManager.Invoker DCL_NOTIFICATION_MGR_INVOKER = - new QueuedNotificationManager.Invoker() { - - @SuppressWarnings("unchecked") + private static final Invoker, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER = + new Invoker, DOMImmutableDataChangeEvent>() { @Override - public void invokeListener( AsyncDataChangeListener listener, - AsyncDataChangeEvent notification ) { - listener.onDataChanged(notification); + public void invokeListener(final DataChangeListenerRegistration listener, + final DOMImmutableDataChangeEvent notification ) { + final AsyncDataChangeListener> inst = listener.getInstance(); + if (inst != null) { + inst.onDataChanged(notification); + } } }; @@ -88,24 +87,26 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch private final AtomicLong txCounter = new AtomicLong(0); private final ListeningExecutorService listeningExecutor; - @SuppressWarnings("rawtypes") - private final NotificationManager - dataChangeListenerNotificationManager; + private final QueuedNotificationManager, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager; private final ExecutorService dataChangeListenerExecutor; + private final ExecutorService domStoreExecutor; + private final String name; - public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor, + private volatile AutoCloseable closeable; + + public InMemoryDOMDataStore(final String name, final ExecutorService domStoreExecutor, final ExecutorService dataChangeListenerExecutor) { - this(name, listeningExecutor, dataChangeListenerExecutor, - InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE); + this(name, domStoreExecutor, dataChangeListenerExecutor, + InMemoryDOMDataStoreConfigProperties.DEFAULT_MAX_DATA_CHANGE_LISTENER_QUEUE_SIZE); } - public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor, - final ExecutorService dataChangeListenerExecutor, int maxDataChangeListenerQueueSize) { + public InMemoryDOMDataStore(final String name, final ExecutorService domStoreExecutor, + final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize) { this.name = Preconditions.checkNotNull(name); - this.listeningExecutor = Preconditions.checkNotNull(listeningExecutor); - + this.domStoreExecutor = Preconditions.checkNotNull(domStoreExecutor); + this.listeningExecutor = MoreExecutors.listeningDecorator(this.domStoreExecutor); this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor); dataChangeListenerNotificationManager = @@ -114,6 +115,18 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch "DataChangeListenerQueueMgr"); } + public void setCloseable(AutoCloseable closeable) { + this.closeable = closeable; + } + + public QueuedNotificationManager getDataChangeListenerNotificationManager() { + return dataChangeListenerNotificationManager; + } + + public ExecutorService getDomStoreExecutor() { + return domStoreExecutor; + } + @Override public final String getIdentifier() { return name; @@ -148,7 +161,16 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch 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); + } + } } + @Override public >> ListenerRegistration registerChangeListener( final YangInstanceIdentifier path, final L listener, final DataChangeScope scope) { @@ -175,8 +197,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch .addCreated(path, data) // .build(); - new ChangeListenerNotifyTask(reg, event, - dataChangeListenerNotificationManager).run(); + dataChangeListenerNotificationManager.submitNotification(reg, event); } } @@ -191,7 +212,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); } @@ -277,7 +298,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. @@ -379,8 +400,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public Void call() { candidate = dataTree.prepare(modification); - listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree, - dataChangeListenerNotificationManager); + listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree); return null; } }); @@ -389,7 +409,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch @Override public ListenableFuture abort() { candidate = null; - return Futures.immediateFuture(null); + return SUCCESSFUL_FUTURE; } @Override @@ -402,14 +422,10 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable, Sch */ synchronized (this) { dataTree.commit(candidate); - - for (ChangeListenerNotifyTask task : listenerResolver.call()) { - LOG.trace("Scheduling invocation of listeners: {}", task); - task.run(); - } + listenerResolver.resolve(dataChangeListenerNotificationManager); } - return Futures.immediateFuture(null); + return SUCCESSFUL_FUTURE; } } }