Datastore: lower in-memory change tracking overhead 04/10604/5
authorRobert Varga <rovarga@cisco.com>
Mon, 1 Sep 2014 22:09:09 +0000 (00:09 +0200)
committerRobert Varga <rovarga@cisco.com>
Tue, 2 Sep 2014 09:03:35 +0000 (11:03 +0200)
We needlessly track the reference to notification manager inside the
even tasks. Instead simplify the return to a Map.Entry and pipe the
resulted data into the manager. This reduces the temporary size of a
tracker, lowering the rate at which we produce garbage.

Change-Id: Ib390d0e65ce1583371a2e887525d9b7ee0392396
Signed-off-by: Robert Varga <rovarga@cisco.com>
opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/ChangeListenerNotifyTask.java [deleted file]
opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/InMemoryDOMDataStore.java
opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/ResolveDataChangeEventsTask.java

diff --git a/opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/ChangeListenerNotifyTask.java b/opendaylight/md-sal/sal-inmemory-datastore/src/main/java/org/opendaylight/controller/md/sal/dom/store/impl/ChangeListenerNotifyTask.java
deleted file mode 100644 (file)
index 536cfa0..0000000
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.md.sal.dom.store.impl;
-
-import com.google.common.base.Preconditions;
-
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
-import org.opendaylight.yangtools.util.concurrent.NotificationManager;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-class ChangeListenerNotifyTask implements Runnable {
-    private static final Logger LOG = LoggerFactory.getLogger(ChangeListenerNotifyTask.class);
-
-    @SuppressWarnings("rawtypes")
-    private final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent> notificationMgr;
-    private final AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> event;
-    private final DataChangeListenerRegistration<?> listener;
-
-    @SuppressWarnings("rawtypes")
-    public ChangeListenerNotifyTask(final DataChangeListenerRegistration<?> listener,
-            final AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> event,
-            final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent> notificationMgr) {
-        this.notificationMgr = Preconditions.checkNotNull(notificationMgr);
-        this.listener = Preconditions.checkNotNull(listener);
-        this.event = Preconditions.checkNotNull(event);
-    }
-
-    @Override
-    public void run() {
-        final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> l = listener.getInstance();
-        if (l == null) {
-            LOG.trace("Skipping event delivery to unregistered listener {}", l);
-            return;
-        }
-        LOG.trace("Listener {} event {}", l, event);
-
-        // FIXME: Yo dawg I heard you like queues, so this was queued to be queued
-        notificationMgr.submitNotification(l, event);
-    }
-
-    @Override
-    public String toString() {
-        return "ChangeListenerNotifyTask [listener=" + listener + ", event=" + event + "]";
-    }
-}
index 7ad3b82..238e32e 100644 (file)
@@ -7,6 +7,8 @@
  */
 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;
@@ -14,23 +16,19 @@ 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.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 +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.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.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
  *
@@ -69,17 +69,15 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, Sch
         TransactionReadyPrototype,AutoCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataStore.class);
 
-    @SuppressWarnings("rawtypes")
-    private static final QueuedNotificationManager.Invoker<AsyncDataChangeListener,
-                                       AsyncDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER =
-            new QueuedNotificationManager.Invoker<AsyncDataChangeListener,
-                                                  AsyncDataChangeEvent>() {
-
-                @SuppressWarnings("unchecked")
+    private static final Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> DCL_NOTIFICATION_MGR_INVOKER =
+            new Invoker<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent>() {
                 @Override
-                public void invokeListener( AsyncDataChangeListener listener,
-                                            AsyncDataChangeEvent notification ) {
-                    listener.onDataChanged(notification);
+                public void invokeListener(final DataChangeListenerRegistration<?> listener,
+                                           final DOMImmutableDataChangeEvent notification ) {
+                    final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> inst = listener.getInstance();
+                    if (inst != null) {
+                        inst.onDataChanged(notification);
+                    }
                 }
             };
 
@@ -88,9 +86,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, Sch
     private final AtomicLong txCounter = new AtomicLong(0);
     private final ListeningExecutorService listeningExecutor;
 
-    @SuppressWarnings("rawtypes")
-    private final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent>
-                                                              dataChangeListenerNotificationManager;
+    private final NotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> dataChangeListenerNotificationManager;
     private final ExecutorService dataChangeListenerExecutor;
 
     private final String name;
@@ -102,10 +98,9 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, Sch
     }
 
     public InMemoryDOMDataStore(final String name, final ListeningExecutorService listeningExecutor,
-            final ExecutorService dataChangeListenerExecutor, int maxDataChangeListenerQueueSize) {
+            final ExecutorService dataChangeListenerExecutor, final int maxDataChangeListenerQueueSize) {
         this.name = Preconditions.checkNotNull(name);
         this.listeningExecutor = Preconditions.checkNotNull(listeningExecutor);
-
         this.dataChangeListenerExecutor = Preconditions.checkNotNull(dataChangeListenerExecutor);
 
         dataChangeListenerNotificationManager =
@@ -175,8 +170,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, Sch
                         .addCreated(path, data) //
                         .build();
 
-                new ChangeListenerNotifyTask(reg, event,
-                        dataChangeListenerNotificationManager).run();
+                dataChangeListenerNotificationManager.submitNotification(reg, event);
             }
         }
 
@@ -277,7 +271,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, 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 +373,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, Sch
                 @Override
                 public Void call() {
                     candidate = dataTree.prepare(modification);
-                    listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree,
-                            dataChangeListenerNotificationManager);
+                    listenerResolver = ResolveDataChangeEventsTask.create(candidate, listenerTree);
                     return null;
                 }
             });
@@ -402,11 +395,7 @@ public class InMemoryDOMDataStore implements DOMStore, Identifiable<String>, 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);
index a4e8c86..5fe9866 100644 (file)
@@ -12,14 +12,10 @@ import com.google.common.base.Preconditions;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.Multimap;
 
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Map.Entry;
-import java.util.concurrent.Callable;
 
 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.dom.store.impl.DOMImmutableDataChangeEvent.Builder;
 import org.opendaylight.controller.md.sal.dom.store.impl.DOMImmutableDataChangeEvent.SimpleEventFactory;
 import org.opendaylight.controller.md.sal.dom.store.impl.tree.ListenerTree;
@@ -40,35 +36,23 @@ import org.slf4j.LoggerFactory;
  * Computes data change events for all affected registered listeners in data
  * tree.
  */
-final class ResolveDataChangeEventsTask implements Callable<Iterable<ChangeListenerNotifyTask>> {
+final class ResolveDataChangeEventsTask {
     private static final Logger LOG = LoggerFactory.getLogger(ResolveDataChangeEventsTask.class);
 
-    @SuppressWarnings("rawtypes")
-    private final NotificationManager<AsyncDataChangeListener, AsyncDataChangeEvent> notificationMgr;
     private final DataTreeCandidate candidate;
     private final ListenerTree listenerRoot;
 
     private Multimap<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> collectedEvents;
 
-    @SuppressWarnings("rawtypes")
-    public ResolveDataChangeEventsTask(final DataTreeCandidate candidate, final ListenerTree listenerTree,
-            final NotificationManager<AsyncDataChangeListener, AsyncDataChangeEvent> notificationMgr) {
+    public ResolveDataChangeEventsTask(final DataTreeCandidate candidate, final ListenerTree listenerTree) {
         this.candidate = Preconditions.checkNotNull(candidate);
         this.listenerRoot = Preconditions.checkNotNull(listenerTree);
-        this.notificationMgr = Preconditions.checkNotNull(notificationMgr);
     }
 
     /**
-     * Resolves and creates Notification Tasks
-     *
-     * Implementation of done as Map-Reduce with two steps: 1. resolving events
-     * and their mapping to listeners 2. merging events affecting same listener
-     *
-     * @return An {@link Iterable} of Notification Tasks which needs to be executed in
-     *         order to delivery data change events.
+     * Resolves and submits notification tasks to the specified manager.
      */
-    @Override
-    public synchronized Iterable<ChangeListenerNotifyTask> call() {
+    public synchronized void resolve(final NotificationManager<DataChangeListenerRegistration<?>, DOMImmutableDataChangeEvent> manager) {
         try (final Walker w = listenerRoot.getWalker()) {
             // Defensive: reset internal state
             collectedEvents = ArrayListMultimap.create();
@@ -81,7 +65,6 @@ final class ResolveDataChangeEventsTask implements Callable<Iterable<ChangeListe
              * Convert to tasks, but be mindful of multiple values -- those indicate multiple
              * wildcard matches, which need to be merged.
              */
-            final Collection<ChangeListenerNotifyTask> ret = new ArrayList<>();
             for (Entry<DataChangeListenerRegistration<?>, Collection<DOMImmutableDataChangeEvent>> e : collectedEvents.asMap().entrySet()) {
                 final Collection<DOMImmutableDataChangeEvent> col = e.getValue();
                 final DOMImmutableDataChangeEvent event;
@@ -98,12 +81,8 @@ final class ResolveDataChangeEventsTask implements Callable<Iterable<ChangeListe
                     event = col.iterator().next();
                 }
 
-                ret.add(new ChangeListenerNotifyTask(e.getKey(), event, notificationMgr));
+                manager.submitNotification(e.getKey(), event);
             }
-
-            // FIXME: so now we have tasks to submit tasks... Inception-style!
-            LOG.debug("Created tasks {}", ret);
-            return ret;
         }
     }
 
@@ -333,10 +312,7 @@ final class ResolveDataChangeEventsTask implements Callable<Iterable<ChangeListe
         return scope != null;
     }
 
-    @SuppressWarnings("rawtypes")
-    public static ResolveDataChangeEventsTask create(final DataTreeCandidate candidate,
-            final ListenerTree listenerTree,
-            final NotificationManager<AsyncDataChangeListener,AsyncDataChangeEvent> notificationMgr) {
-        return new ResolveDataChangeEventsTask(candidate, listenerTree, notificationMgr);
+    public static ResolveDataChangeEventsTask create(final DataTreeCandidate candidate, final ListenerTree listenerTree) {
+        return new ResolveDataChangeEventsTask(candidate, listenerTree);
     }
 }