X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fbroker%2Fimpl%2FDOMNotificationRouter.java;h=ebd671aebfa55075c685a8c2389fe6a0e4a293e4;hp=64c6b9e1fd1791a4e29e5868e64cf89975b9a2dd;hb=3ec97cd0a86ad1b79f6854dc6924eb7b06e359a3;hpb=dceb9db7853dabfbd4abdfb3d886a79871097831 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 index 64c6b9e1fd..ebd671aebf 100644 --- 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 @@ -7,27 +7,9 @@ */ 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.ImmutableList; -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.PhasedBackoffWaitStrategy; -import com.lmax.disruptor.WaitStrategy; -import com.lmax.disruptor.dsl.Disruptor; -import com.lmax.disruptor.dsl.ProducerType; -import java.util.Arrays; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.Collection; -import java.util.List; -import java.util.Set; -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; @@ -35,219 +17,94 @@ import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService; import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService; import org.opendaylight.controller.md.sal.dom.spi.DOMNotificationSubscriptionListener; import org.opendaylight.controller.md.sal.dom.spi.DOMNotificationSubscriptionListenerRegistry; -import org.opendaylight.yangtools.concepts.AbstractListenerRegistration; +import org.opendaylight.controller.sal.core.compat.LegacyDOMNotificationServiceAdapter; import org.opendaylight.yangtools.concepts.ListenerRegistration; -import org.opendaylight.yangtools.util.ListenerRegistry; import org.opendaylight.yangtools.yang.model.api.SchemaPath; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * 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)} + *

+ * The fully-blocking {@link #offerNotification(DOMNotification)} + * is 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, DOMNotificationSubscriptionListenerRegistry { - - private static final Logger LOG = LoggerFactory.getLogger(DOMNotificationRouter.class); - private static final ListenableFuture NO_LISTENERS = Futures.immediateFuture(null); - private static final WaitStrategy DEFAULT_STRATEGY = PhasedBackoffWaitStrategy.withLock(1L, 30L, TimeUnit.MILLISECONDS); - private static final EventHandler DISPATCH_NOTIFICATIONS = new EventHandler() { - @Override - public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) throws Exception { - event.deliverNotification(); - - } - }; - private static final EventHandler NOTIFY_FUTURE = new EventHandler() { - @Override - public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) { - event.setFuture(); - } - }; - - private final Disruptor disruptor; - private final ExecutorService executor; - private volatile Multimap> listeners = ImmutableMultimap.of(); - private final ListenerRegistry subscriptionListeners = ListenerRegistry.create(); - - @SuppressWarnings("unchecked") - private DOMNotificationRouter(final ExecutorService executor, final int queueDepth, final WaitStrategy strategy) { - this.executor = Preconditions.checkNotNull(executor); - - disruptor = new Disruptor<>(DOMNotificationRouterEvent.FACTORY, queueDepth, executor, ProducerType.MULTI, strategy); - disruptor.handleEventsWith(DISPATCH_NOTIFICATIONS); - disruptor.after(DISPATCH_NOTIFICATIONS).handleEventsWith(NOTIFY_FUTURE); - disruptor.start(); +@Deprecated +@SuppressFBWarnings(value = "NP_NONNULL_PARAM_VIOLATION", justification = "Void is the only allowed value") +public final class DOMNotificationRouter extends LegacyDOMNotificationServiceAdapter implements AutoCloseable, + DOMNotificationPublishService, DOMNotificationSubscriptionListenerRegistry { + + private final org.opendaylight.mdsal.dom.api.DOMNotificationPublishService delegateNotificationPublishService; + private final org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry delegateListenerRegistry; + + private DOMNotificationRouter( + final org.opendaylight.mdsal.dom.api.DOMNotificationService delegateNotificationService, + final org.opendaylight.mdsal.dom.api.DOMNotificationPublishService delegateNotificationPublishService, + final org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry delegateListenerRegistry) { + super(delegateNotificationService); + this.delegateNotificationPublishService = delegateNotificationPublishService; + this.delegateListenerRegistry = delegateListenerRegistry; } public static DOMNotificationRouter create(final int queueDepth) { - final ExecutorService executor = Executors.newCachedThreadPool(); - - return new DOMNotificationRouter(executor, queueDepth, DEFAULT_STRATEGY); + final org.opendaylight.mdsal.dom.broker.DOMNotificationRouter delegate = + org.opendaylight.mdsal.dom.broker.DOMNotificationRouter.create(queueDepth); + return create(delegate, delegate, delegate); } - public static DOMNotificationRouter create(final int queueDepth, final long spinTime, final long parkTime, final TimeUnit unit) { - Preconditions.checkArgument(Long.lowestOneBit(queueDepth) == Long.highestOneBit(queueDepth), - "Queue depth %s is not power-of-two", queueDepth); - final ExecutorService executor = Executors.newCachedThreadPool(); - final WaitStrategy strategy = PhasedBackoffWaitStrategy.withLock(spinTime, parkTime, unit); - - return new DOMNotificationRouter(executor, queueDepth, strategy); + public static DOMNotificationRouter create(final int queueDepth, final long spinTime, final long parkTime, + final TimeUnit unit) { + final org.opendaylight.mdsal.dom.broker.DOMNotificationRouter delegate = + org.opendaylight.mdsal.dom.broker.DOMNotificationRouter.create(queueDepth, spinTime, parkTime, unit); + return create(delegate, delegate, delegate); } - @Override - public synchronized ListenerRegistration registerNotificationListener(final T listener, final Collection types) { - final ListenerRegistration reg = new AbstractListenerRegistration(listener) { - @Override - protected void removeRegistration() { - final ListenerRegistration me = this; - - synchronized (DOMNotificationRouter.this) { - replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, new Predicate>() { - @Override - public boolean apply(final ListenerRegistration input) { - return input != me; - } - }))); - } - } - }; - - if (!types.isEmpty()) { - final Builder> b = ImmutableMultimap.builder(); - b.putAll(listeners); - - for (final SchemaPath t : types) { - b.put(t, reg); - } - - replaceListeners(b.build()); - } - - return reg; + public static DOMNotificationRouter create( + final org.opendaylight.mdsal.dom.api.DOMNotificationService delegateNotificationService, + final org.opendaylight.mdsal.dom.api.DOMNotificationPublishService delegateNotificationPublishService, + final org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry delegateListenerRegistry) { + return new DOMNotificationRouter(delegateNotificationService, delegateNotificationPublishService, + delegateListenerRegistry); } @Override - public ListenerRegistration registerNotificationListener(final T listener, final SchemaPath... types) { - return registerNotificationListener(listener, Arrays.asList(types)); - } - - /** - * Swaps registered listeners and triggers notification update - * - * @param newListeners - */ - private void replaceListeners( - final Multimap> newListeners) { - listeners = newListeners; - notifyListenerTypesChanged(newListeners.keySet()); - } - - private void notifyListenerTypesChanged(final Set typesAfter) { - final List> listenersAfter =ImmutableList.copyOf(subscriptionListeners.getListeners()); - executor.submit(new Runnable() { - - @Override - public void run() { - for (final ListenerRegistration subListener : listenersAfter) { - try { - subListener.getInstance().onSubscriptionChanged(typesAfter); - } catch (final Exception e) { - LOG.warn("Uncaught exception during invoking listener {}", subListener.getInstance(), e); - } - } - } - }); + public synchronized ListenerRegistration registerNotificationListener( + final T listener, final Collection types) { + return super.registerNotificationListener(listener, types); } @Override public ListenerRegistration registerSubscriptionListener( final L listener) { - final Set initialTypes = listeners.keySet(); - executor.submit(new Runnable() { - - @Override - public void run() { - listener.onSubscriptionChanged(initialTypes); - } - }); - return subscriptionListeners.registerWithType(listener); - } - - private ListenableFuture publish(final long seq, final DOMNotification notification, final Collection> subscribers) { - final DOMNotificationRouterEvent event = disruptor.get(seq); - final ListenableFuture future = event.initialize(notification, subscribers); - disruptor.getRingBuffer().publish(seq); - return future; + return delegateListenerRegistry.registerSubscriptionListener(listener); } @Override - public ListenableFuture putNotification(final DOMNotification notification) throws InterruptedException { - final Collection> subscribers = listeners.get(notification.getType()); - if (subscribers.isEmpty()) { - return NO_LISTENERS; - } - - final long seq = disruptor.getRingBuffer().next(); - return publish(seq, notification, subscribers); - } - - private ListenableFuture tryPublish(final DOMNotification notification, final Collection> subscribers) { - final long seq; - try { - seq = disruptor.getRingBuffer().tryNext(); - } catch (final InsufficientCapacityException e) { - return DOMNotificationPublishService.REJECTED; - } - - return publish(seq, notification, subscribers); + public ListenableFuture putNotification(final DOMNotification notification) throws InterruptedException { + return delegateNotificationPublishService.putNotification(notification); } @Override - public ListenableFuture offerNotification(final DOMNotification notification) { - final Collection> subscribers = listeners.get(notification.getType()); - if (subscribers.isEmpty()) { - return NO_LISTENERS; - } - - return tryPublish(notification, subscribers); + public ListenableFuture offerNotification(final DOMNotification notification) { + return delegateNotificationPublishService.offerNotification(notification); } @Override - public ListenableFuture offerNotification(final DOMNotification notification, final long timeout, - final TimeUnit unit) throws InterruptedException { - final Collection> subscribers = listeners.get(notification.getType()); - if (subscribers.isEmpty()) { - return NO_LISTENERS; - } - - // Attempt to perform a non-blocking publish first - final ListenableFuture 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"); + public ListenableFuture offerNotification(final DOMNotification notification, final long timeout, + final TimeUnit unit) throws InterruptedException { + return delegateNotificationPublishService.offerNotification(notification, timeout, unit); } @Override public void close() { - disruptor.shutdown(); - executor.shutdown(); } }