Merge "BUG-2288: implement DOMNotificationRouter"
authorTony Tkacik <ttkacik@cisco.com>
Tue, 10 Feb 2015 21:01:02 +0000 (21:01 +0000)
committerGerrit Code Review <gerrit@opendaylight.org>
Tue, 10 Feb 2015 21:01:03 +0000 (21:01 +0000)
features/mdsal/src/main/resources/features.xml
opendaylight/commons/opendaylight/pom.xml
opendaylight/md-sal/sal-binding-it/src/main/java/org/opendaylight/controller/test/sal/binding/it/TestHelper.java
opendaylight/md-sal/sal-dom-broker/pom.xml
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouter.java [new file with mode: 0644]
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouterEvent.java [new file with mode: 0644]

index 1582f45..8c166e6 100644 (file)
@@ -28,6 +28,7 @@
         <feature version='${mdsal.version}'>odl-mdsal-common</feature>
         <feature version='${config.version}'>odl-config-startup</feature>
         <feature version='${config.version}'>odl-config-netty</feature>
+        <bundle>mvn:com.lmax/disruptor/${lmax.version}</bundle>
         <bundle>mvn:org.opendaylight.controller/sal-core-api/${project.version}</bundle>
         <bundle>mvn:org.opendaylight.controller/sal-core-spi/${project.version}</bundle>
         <bundle>mvn:org.opendaylight.controller/sal-broker-impl/${project.version}</bundle>
index 5310db3..d6140f1 100644 (file)
     <yangtools.version>0.7.0-SNAPSHOT</yangtools.version>
     <sshd-core.version>0.12.0</sshd-core.version>
     <jmh.version>0.9.7</jmh.version>
+    <lmax.version>3.3.0</lmax.version>
   </properties>
 
   <dependencyManagement>
         <artifactId>guava</artifactId>
         <version>${guava.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.lmax</groupId>
+        <artifactId>disruptor</artifactId>
+        <version>${lmax.version}</version>
+      </dependency>
+
       <!-- 3rd party dependencies needed by config-->
       <dependency>
         <groupId>com.jcabi</groupId>
index 9b6d583..a64e360 100644 (file)
@@ -10,8 +10,8 @@ package org.opendaylight.controller.test.sal.binding.it;
 import static org.ops4j.pax.exam.CoreOptions.frameworkProperty;
 import static org.ops4j.pax.exam.CoreOptions.junitBundles;
 import static org.ops4j.pax.exam.CoreOptions.mavenBundle;
+import static org.ops4j.pax.exam.CoreOptions.systemPackages;
 import static org.ops4j.pax.exam.CoreOptions.systemProperty;
-
 import org.ops4j.pax.exam.Option;
 import org.ops4j.pax.exam.options.DefaultCompositeOption;
 import org.ops4j.pax.exam.util.PathUtils;
@@ -47,7 +47,7 @@ public class TestHelper {
                 bindingAwareSalBundles(),
                 mavenBundle("commons-codec", "commons-codec").versionAsInProject(),
 
-                systemProperty("org.osgi.framework.system.packages.extra").value("sun.nio.ch"),
+                systemPackages("sun.nio.ch", "sun.misc"),
                 mavenBundle("io.netty", "netty-common").versionAsInProject(), //
                 mavenBundle("io.netty", "netty-buffer").versionAsInProject(), //
                 mavenBundle("io.netty", "netty-handler").versionAsInProject(), //
@@ -123,7 +123,8 @@ public class TestHelper {
                 mavenBundle(CONTROLLER, "sal-common-util").versionAsInProject(), // //
 
 
-                mavenBundle(CONTROLLER, "sal-inmemory-datastore").versionAsInProject(), // /
+                mavenBundle("com.lmax", "disruptor").versionAsInProject(),
+                mavenBundle(CONTROLLER, "sal-inmemory-datastore").versionAsInProject(), //
                 mavenBundle(CONTROLLER, "sal-broker-impl").versionAsInProject(), // //
                 mavenBundle(CONTROLLER, "sal-core-spi").versionAsInProject().update(), //
 
index a824792..477ddea 100644 (file)
@@ -15,8 +15,8 @@
       <artifactId>guava</artifactId>
     </dependency>
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
+      <groupId>com.lmax</groupId>
+      <artifactId>disruptor</artifactId>
     </dependency>
     <dependency>
       <groupId>org.opendaylight.controller</groupId>
         <artifactId>ietf-yang-types</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouter.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouter.java
new file mode 100644 (file)
index 0000000..aac425b
--- /dev/null
@@ -0,0 +1,190 @@
+/*
+ * 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.broker.impl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableMultimap;
+import com.google.common.collect.ImmutableMultimap.Builder;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Multimaps;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.InsufficientCapacityException;
+import com.lmax.disruptor.SleepingWaitStrategy;
+import com.lmax.disruptor.WaitStrategy;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
+import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+
+/**
+ * Joint implementation of {@link DOMNotificationPublishService} and {@link DOMNotificationService}. Provides
+ * routing of notifications from publishers to subscribers.
+ *
+ * Internal implementation works by allocating a two-handler Disruptor. The first handler delivers notifications
+ * to subscribed listeners and the second one notifies whoever may be listening on the returned future. Registration
+ * state tracking is performed by a simple immutable multimap -- when a registration or unregistration occurs we
+ * re-generate the entire map from scratch and set it atomically. While registrations/unregistrations synchronize
+ * on this instance, notifications do not take any locks here.
+ *
+ * The fully-blocking {@link #publish(long, DOMNotification, Collection)} and non-blocking {@link #offerNotification(DOMNotification)}
+ * are realized using the Disruptor's native operations. The bounded-blocking {@link #offerNotification(DOMNotification, long, TimeUnit)}
+ * is realized by arming a background wakeup interrupt.
+ */
+public final class DOMNotificationRouter implements AutoCloseable, DOMNotificationPublishService, DOMNotificationService {
+    private static final ListenableFuture<Void> NO_LISTENERS = Futures.immediateFuture(null);
+    private static final WaitStrategy DEFAULT_STRATEGY = new SleepingWaitStrategy();
+    private static final EventHandler<DOMNotificationRouterEvent> DISPATCH_NOTIFICATIONS = new EventHandler<DOMNotificationRouterEvent>() {
+        @Override
+        public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) throws Exception {
+            event.deliverNotification();
+
+        }
+    };
+    private static final EventHandler<DOMNotificationRouterEvent> NOTIFY_FUTURE = new EventHandler<DOMNotificationRouterEvent>() {
+        @Override
+        public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) {
+            event.setFuture();
+        }
+    };
+
+    private final Disruptor<DOMNotificationRouterEvent> disruptor;
+    private final ExecutorService executor;
+    private volatile Multimap<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> listeners = ImmutableMultimap.of();
+
+    private DOMNotificationRouter(final ExecutorService executor, final Disruptor<DOMNotificationRouterEvent> disruptor) {
+        this.executor = Preconditions.checkNotNull(executor);
+        this.disruptor = Preconditions.checkNotNull(disruptor);
+    }
+
+    @SuppressWarnings("unchecked")
+    public static DOMNotificationRouter create(final int queueDepth) {
+        final ExecutorService executor = Executors.newCachedThreadPool();
+        final Disruptor<DOMNotificationRouterEvent> disruptor = new Disruptor<>(DOMNotificationRouterEvent.FACTORY, queueDepth, executor, ProducerType.MULTI, DEFAULT_STRATEGY);
+
+        disruptor.after(DISPATCH_NOTIFICATIONS).handleEventsWith(NOTIFY_FUTURE);
+        disruptor.start();
+
+        return new DOMNotificationRouter(executor, disruptor);
+    }
+
+    @Override
+    public synchronized <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener, final Collection<SchemaPath> types) {
+        final ListenerRegistration<T> reg = new AbstractListenerRegistration<T>(listener) {
+            @Override
+            protected void removeRegistration() {
+                final ListenerRegistration<T> me = this;
+
+                synchronized (DOMNotificationRouter.this) {
+                    listeners = ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, new Predicate<ListenerRegistration<? extends DOMNotificationListener>>() {
+                        @Override
+                        public boolean apply(final ListenerRegistration<? extends DOMNotificationListener> input) {
+                            return input != me;
+                        }
+                    }));
+                }
+            }
+        };
+
+        if (!types.isEmpty()) {
+            final Builder<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> b = ImmutableMultimap.builder();
+            b.putAll(listeners);
+
+            for (SchemaPath t : types) {
+                b.put(t, reg);
+            }
+
+            listeners = b.build();
+        }
+
+        return reg;
+    }
+
+    @Override
+    public <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener, final SchemaPath... types) {
+        return registerNotificationListener(listener, Arrays.asList(types));
+    }
+
+    private ListenableFuture<Void> publish(final long seq, final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) {
+        final DOMNotificationRouterEvent event = disruptor.get(seq);
+        final ListenableFuture<Void> future = event.initialize(notification, subscribers);
+        disruptor.getRingBuffer().publish(seq);
+        return future;
+    }
+
+    @Override
+    public ListenableFuture<? extends Object> putNotification(final DOMNotification notification) throws InterruptedException {
+        final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners.get(notification.getType());
+        if (subscribers.isEmpty()) {
+            return NO_LISTENERS;
+        }
+
+        final long seq = disruptor.getRingBuffer().next();
+        return publish(seq, notification, subscribers);
+    }
+
+    private ListenableFuture<? extends Object> tryPublish(final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) {
+        final long seq;
+        try {
+             seq = disruptor.getRingBuffer().tryNext();
+        } catch (InsufficientCapacityException e) {
+            return DOMNotificationPublishService.REJECTED;
+        }
+
+        return publish(seq, notification, subscribers);
+    }
+
+    @Override
+    public ListenableFuture<? extends Object> offerNotification(final DOMNotification notification) {
+        final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners.get(notification.getType());
+        if (subscribers.isEmpty()) {
+            return NO_LISTENERS;
+        }
+
+        return tryPublish(notification, subscribers);
+    }
+
+    @Override
+    public ListenableFuture<? extends Object> offerNotification(final DOMNotification notification, final long timeout,
+            final TimeUnit unit) throws InterruptedException {
+        final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners.get(notification.getType());
+        if (subscribers.isEmpty()) {
+            return NO_LISTENERS;
+        }
+
+        // Attempt to perform a non-blocking publish first
+        final ListenableFuture<? extends Object> noBlock = tryPublish(notification, subscribers);
+        if (!DOMNotificationPublishService.REJECTED.equals(noBlock)) {
+            return noBlock;
+        }
+
+        /*
+         * FIXME: we need a background thread, which will watch out for blocking too long. Here
+         *        we will arm a tasklet for it and synchronize delivery of interrupt properly.
+         */
+        throw new UnsupportedOperationException("Not implemented yet");
+    }
+
+    @Override
+    public void close() {
+        disruptor.shutdown();
+        executor.shutdown();
+    }
+}
diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouterEvent.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouterEvent.java
new file mode 100644 (file)
index 0000000..65c7166
--- /dev/null
@@ -0,0 +1,59 @@
+/*
+ * 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.broker.impl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import com.lmax.disruptor.EventFactory;
+import java.util.Collection;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+
+/**
+ * A single notification event in the disruptor ringbuffer. These objects are reused,
+ * so they do have mutable state.
+ */
+final class DOMNotificationRouterEvent {
+    public static final EventFactory<DOMNotificationRouterEvent> FACTORY = new EventFactory<DOMNotificationRouterEvent>() {
+        @Override
+        public DOMNotificationRouterEvent newInstance() {
+            return new DOMNotificationRouterEvent();
+        }
+    };
+
+    private Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers;
+    private DOMNotification notification;
+    private SettableFuture<Void> future;
+
+    private DOMNotificationRouterEvent() {
+        // Hidden on purpose, initialized in initialize()
+    }
+
+    ListenableFuture<Void> initialize(final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) {
+        this.notification = Preconditions.checkNotNull(notification);
+        this.subscribers = Preconditions.checkNotNull(subscribers);
+        this.future = SettableFuture.create();
+        return this.future;
+    }
+
+    void deliverNotification() {
+        for (ListenerRegistration<? extends DOMNotificationListener> r : subscribers) {
+            final DOMNotificationListener l = r.getInstance();
+            if (l != null) {
+                l.onNotification(notification);
+            }
+        }
+    }
+
+    void setFuture() {
+        future.set(null);
+    }
+
+}
\ No newline at end of file

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.