opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouter.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.md.sal.dom.broker.impl;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.base.Predicate;
  12 import com.google.common.collect.ImmutableMultimap;
  13 import com.google.common.collect.ImmutableMultimap.Builder;
  14 import com.google.common.collect.Multimap;
  15 import com.google.common.collect.Multimaps;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import com.lmax.disruptor.EventHandler;
  19 import com.lmax.disruptor.InsufficientCapacityException;
  20 import com.lmax.disruptor.SleepingWaitStrategy;
  21 import com.lmax.disruptor.WaitStrategy;
  22 import com.lmax.disruptor.dsl.Disruptor;
  23 import com.lmax.disruptor.dsl.ProducerType;
  24 import java.util.Arrays;
  25 import java.util.Collection;
  26 import java.util.concurrent.ExecutorService;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.TimeUnit;
  29 import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
  30 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
  31 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
  32 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationService;
  33 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
  34 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  35 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  36
  37 /**
  38  * Joint implementation of {@link DOMNotificationPublishService} and {@link DOMNotificationService}. Provides
  39  * routing of notifications from publishers to subscribers.
  40  *
  41  * Internal implementation works by allocating a two-handler Disruptor. The first handler delivers notifications
  42  * to subscribed listeners and the second one notifies whoever may be listening on the returned future. Registration
  43  * state tracking is performed by a simple immutable multimap -- when a registration or unregistration occurs we
  44  * re-generate the entire map from scratch and set it atomically. While registrations/unregistrations synchronize
  45  * on this instance, notifications do not take any locks here.
  46  *
  47  * The fully-blocking {@link #publish(long, DOMNotification, Collection)} and non-blocking {@link #offerNotification(DOMNotification)}
  48  * are realized using the Disruptor's native operations. The bounded-blocking {@link #offerNotification(DOMNotification, long, TimeUnit)}
  49  * is realized by arming a background wakeup interrupt.
  50  */
  51 public final class DOMNotificationRouter implements AutoCloseable, DOMNotificationPublishService, DOMNotificationService {
  52     private static final ListenableFuture<Void> NO_LISTENERS = Futures.immediateFuture(null);
  53     private static final WaitStrategy DEFAULT_STRATEGY = new SleepingWaitStrategy();
  54     private static final EventHandler<DOMNotificationRouterEvent> DISPATCH_NOTIFICATIONS = new EventHandler<DOMNotificationRouterEvent>() {
  55         @Override
  56         public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) throws Exception {
  57             event.deliverNotification();
  58
  59         }
  60     };
  61     private static final EventHandler<DOMNotificationRouterEvent> NOTIFY_FUTURE = new EventHandler<DOMNotificationRouterEvent>() {
  62         @Override
  63         public void onEvent(final DOMNotificationRouterEvent event, final long sequence, final boolean endOfBatch) {
  64             event.setFuture();
  65         }
  66     };
  67
  68     private final Disruptor<DOMNotificationRouterEvent> disruptor;
  69     private final ExecutorService executor;
  70     private volatile Multimap<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> listeners = ImmutableMultimap.of();
  71
  72     private DOMNotificationRouter(final ExecutorService executor, final Disruptor<DOMNotificationRouterEvent> disruptor) {
  73         this.executor = Preconditions.checkNotNull(executor);
  74         this.disruptor = Preconditions.checkNotNull(disruptor);
  75     }
  76
  77     @SuppressWarnings("unchecked")
  78     public static DOMNotificationRouter create(final int queueDepth) {
  79         final ExecutorService executor = Executors.newCachedThreadPool();
  80         final Disruptor<DOMNotificationRouterEvent> disruptor = new Disruptor<>(DOMNotificationRouterEvent.FACTORY, queueDepth, executor, ProducerType.MULTI, DEFAULT_STRATEGY);
  81
  82         disruptor.handleEventsWith(DISPATCH_NOTIFICATIONS);
  83         disruptor.after(DISPATCH_NOTIFICATIONS).handleEventsWith(NOTIFY_FUTURE);
  84         disruptor.start();
  85
  86         return new DOMNotificationRouter(executor, disruptor);
  87     }
  88
  89     @Override
  90     public synchronized <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener, final Collection<SchemaPath> types) {
  91         final ListenerRegistration<T> reg = new AbstractListenerRegistration<T>(listener) {
  92             @Override
  93             protected void removeRegistration() {
  94                 final ListenerRegistration<T> me = this;
  95
  96                 synchronized (DOMNotificationRouter.this) {
  97                     listeners = ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, new Predicate<ListenerRegistration<? extends DOMNotificationListener>>() {
  98                         @Override
  99                         public boolean apply(final ListenerRegistration<? extends DOMNotificationListener> input) {
 100                             return input != me;
 101                         }
 102                     }));
 103                 }
 104             }
 105         };
 106
 107         if (!types.isEmpty()) {
 108             final Builder<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> b = ImmutableMultimap.builder();
 109             b.putAll(listeners);
 110
 111             for (SchemaPath t : types) {
 112                 b.put(t, reg);
 113             }
 114
 115             listeners = b.build();
 116         }
 117
 118         return reg;
 119     }
 120
 121     @Override
 122     public <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener, final SchemaPath... types) {
 123         return registerNotificationListener(listener, Arrays.asList(types));
 124     }
 125
 126     private ListenableFuture<Void> publish(final long seq, final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) {
 127         final DOMNotificationRouterEvent event = disruptor.get(seq);
 128         final ListenableFuture<Void> future = event.initialize(notification, subscribers);
 129         disruptor.getRingBuffer().publish(seq);
 130         return future;
 131     }
 132
 133     @Override
 134     public ListenableFuture<? extends Object> putNotification(final DOMNotification notification) throws InterruptedException {
 135         final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners.get(notification.getType());
 136         if (subscribers.isEmpty()) {
 137             return NO_LISTENERS;
 138         }
 139
 140         final long seq = disruptor.getRingBuffer().next();
 141         return publish(seq, notification, subscribers);
 142     }
 143
 144     private ListenableFuture<? extends Object> tryPublish(final DOMNotification notification, final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers) {
 145         final long seq;
 146         try {
 147              seq = disruptor.getRingBuffer().tryNext();
 148         } catch (InsufficientCapacityException e) {
 149             return DOMNotificationPublishService.REJECTED;
 150         }
 151
 152         return publish(seq, notification, subscribers);
 153     }
 154
 155     @Override
 156     public ListenableFuture<? extends Object> offerNotification(final DOMNotification notification) {
 157         final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners.get(notification.getType());
 158         if (subscribers.isEmpty()) {
 159             return NO_LISTENERS;
 160         }
 161
 162         return tryPublish(notification, subscribers);
 163     }
 164
 165     @Override
 166     public ListenableFuture<? extends Object> offerNotification(final DOMNotification notification, final long timeout,
 167             final TimeUnit unit) throws InterruptedException {
 168         final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners.get(notification.getType());
 169         if (subscribers.isEmpty()) {
 170             return NO_LISTENERS;
 171         }
 172
 173         // Attempt to perform a non-blocking publish first
 174         final ListenableFuture<? extends Object> noBlock = tryPublish(notification, subscribers);
 175         if (!DOMNotificationPublishService.REJECTED.equals(noBlock)) {
 176             return noBlock;
 177         }
 178
 179         /*
 180          * FIXME: we need a background thread, which will watch out for blocking too long. Here
 181          *        we will arm a tasklet for it and synchronize delivery of interrupt properly.
 182          */
 183         throw new UnsupportedOperationException("Not implemented yet");
 184     }
 185
 186     @Override
 187     public void close() {
 188         disruptor.shutdown();
 189         executor.shutdown();
 190     }
 191 }