Bug 1378: Make sure config extender does not block bundle loading. 62/9262/2
authorTony Tkacik <ttkacik@cisco.com>
Wed, 23 Jul 2014 14:38:09 +0000 (16:38 +0200)
committerTony Tkacik <ttkacik@cisco.com>
Wed, 23 Jul 2014 15:11:22 +0000 (17:11 +0200)
Change-Id: Ic706ad9dd41ad951e0517e02b7f1f3b8b25ea9ea
Signed-off-by: Tony Tkacik <ttkacik@cisco.com>
opendaylight/config/config-manager/src/main/java/org/opendaylight/controller/config/manager/impl/osgi/ExtensibleBundleTracker.java

index c1ebba7..eff267a 100644 (file)
@@ -7,6 +7,13 @@
  */
 package org.opendaylight.controller.config.manager.impl.osgi;
 
  */
 package org.opendaylight.controller.config.manager.impl.osgi;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
 import org.osgi.framework.Bundle;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.BundleEvent;
 import org.osgi.framework.Bundle;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.BundleEvent;
@@ -15,74 +22,114 @@ import org.osgi.util.tracker.BundleTrackerCustomizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  *
 /**
  *
- * Extensible bundle tracker. Takes several BundleTrackerCustomizers and propagates bundle events to all of them.
- * Primary customizer
+ * Extensible bundle tracker. Takes several BundleTrackerCustomizers and
+ * propagates bundle events to all of them.
+ *
+ * Primary customizer may return tracking object,
+ * which will be passed to it during invocation of
+ * {@link BundleTrackerCustomizer#removedBundle(Bundle, BundleEvent, Future)}
+ *
+ *
+ * This extender modifies behaviour to not leak platform thread
+ * in {@link BundleTrackerCustomizer#addingBundle(Bundle, BundleEvent)}
+ * but deliver this event from its own single threaded executor.
+ *
+ * If bundle is removed before event for adding bundle was executed,
+ * that event is cancelled. If addingBundle event is currently in progress
+ * or was already executed, platform thread is block untill addingBundle
+ * finishes so bundle could be removed correctly in platform thread.
+ *
+ *
+ * Method {@link BundleTrackerCustomizer#removedBundle(Bundle, BundleEvent, Object)}
+ * is never invoked on registered trackers.
  *
  * @param <T>
  */
  *
  * @param <T>
  */
-public final class ExtensibleBundleTracker<T> extends BundleTracker<T> {
+public final class ExtensibleBundleTracker<T> extends BundleTracker<Future<T>> {
 
 
+    private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder()
+        .setNameFormat("config-bundle-tracker-%d")
+        .build();
+    private final ExecutorService eventExecutor;
     private final BundleTrackerCustomizer<T> primaryTracker;
     private final BundleTrackerCustomizer<?>[] additionalTrackers;
 
     private final BundleTrackerCustomizer<T> primaryTracker;
     private final BundleTrackerCustomizer<?>[] additionalTrackers;
 
-    private static final Logger logger = LoggerFactory.getLogger(ExtensibleBundleTracker.class);
+    private static final Logger LOG = LoggerFactory.getLogger(ExtensibleBundleTracker.class);
 
 
-    public ExtensibleBundleTracker(BundleContext context, BundleTrackerCustomizer<T> primaryBundleTrackerCustomizer,
-                                   BundleTrackerCustomizer<?>... additionalBundleTrackerCustomizers) {
+    public ExtensibleBundleTracker(final BundleContext context, final BundleTrackerCustomizer<T> primaryBundleTrackerCustomizer,
+                                   final BundleTrackerCustomizer<?>... additionalBundleTrackerCustomizers) {
         this(context, Bundle.ACTIVE, primaryBundleTrackerCustomizer, additionalBundleTrackerCustomizers);
     }
 
         this(context, Bundle.ACTIVE, primaryBundleTrackerCustomizer, additionalBundleTrackerCustomizers);
     }
 
-    public ExtensibleBundleTracker(BundleContext context, int bundleState,
-                                   BundleTrackerCustomizer<T> primaryBundleTrackerCustomizer,
-                                   BundleTrackerCustomizer<?>... additionalBundleTrackerCustomizers) {
+    public ExtensibleBundleTracker(final BundleContext context, final int bundleState,
+                                   final BundleTrackerCustomizer<T> primaryBundleTrackerCustomizer,
+                                   final BundleTrackerCustomizer<?>... additionalBundleTrackerCustomizers) {
         super(context, bundleState, null);
         this.primaryTracker = primaryBundleTrackerCustomizer;
         this.additionalTrackers = additionalBundleTrackerCustomizers;
         super(context, bundleState, null);
         this.primaryTracker = primaryBundleTrackerCustomizer;
         this.additionalTrackers = additionalBundleTrackerCustomizers;
-        logger.trace("Registered as extender with context {} and bundle state {}", context, bundleState);
+        eventExecutor = Executors.newSingleThreadExecutor(THREAD_FACTORY);
+        LOG.trace("Registered as extender with context {} and bundle state {}", context, bundleState);
     }
 
     @Override
     }
 
     @Override
-    public T addingBundle(final Bundle bundle, final BundleEvent event) {
-        T primaryTrackerRetVal = primaryTracker.addingBundle(bundle, event);
-
-        forEachAdditionalBundle(new BundleStrategy() {
+    public Future<T> addingBundle(final Bundle bundle, final BundleEvent event) {
+        LOG.trace("Submiting AddingBundle for bundle {} and event {} to be processed asynchronously",bundle,event);
+        Future<T> future = eventExecutor.submit(new Callable<T>() {
             @Override
             @Override
-            public void execute(BundleTrackerCustomizer<?> tracker) {
-                tracker.addingBundle(bundle, event);
+            public T call() throws Exception {
+                try {
+                    T primaryTrackerRetVal = primaryTracker.addingBundle(bundle, event);
+
+                    forEachAdditionalBundle(new BundleStrategy() {
+                        @Override
+                        public void execute(final BundleTrackerCustomizer<?> tracker) {
+                            tracker.addingBundle(bundle, event);
+                        }
+                    });
+                    LOG.trace("AddingBundle for {} and event {} finished successfully",bundle,event);
+                    return primaryTrackerRetVal;
+                } catch (Exception e) {
+                    LOG.error("Failed to add bundle {}",e);
+                    throw e;
+                }
             }
         });
             }
         });
-
-        return primaryTrackerRetVal;
+        return future;
     }
 
     @Override
     }
 
     @Override
-    public void modifiedBundle(final Bundle bundle, final BundleEvent event, final T object) {
-        primaryTracker.modifiedBundle(bundle, event, object);
-
-        forEachAdditionalBundle(new BundleStrategy() {
-            @Override
-            public void execute(BundleTrackerCustomizer<?> tracker) {
-                tracker.modifiedBundle(bundle, event, null);
-            }
-        });
+    public void modifiedBundle(final Bundle bundle, final BundleEvent event, final Future<T> object) {
+        // Intentionally NOOP
 
     }
 
     @Override
 
     }
 
     @Override
-    public void removedBundle(final Bundle bundle, final BundleEvent event, final T object) {
-        primaryTracker.removedBundle(bundle, event, object);
-
-        forEachAdditionalBundle(new BundleStrategy() {
-            @Override
-            public void execute(BundleTrackerCustomizer<?> tracker) {
-                tracker.removedBundle(bundle, event, null);
-            }
-        });
+    public void removedBundle(final Bundle bundle, final BundleEvent event, final Future<T> object) {
+        if(!object.isDone() && object.cancel(false)) {
+            // We canceled adding event before it was processed
+            // so it is safe to return
+            LOG.trace("Adding Bundle event for {} was cancelled. No additional work required.",bundle);
+            return;
+        }
+        try {
+            LOG.trace("Invoking removedBundle event for {}",bundle);
+            primaryTracker.removedBundle(bundle, event, object.get());
+            forEachAdditionalBundle(new BundleStrategy() {
+                @Override
+                public void execute(final BundleTrackerCustomizer<?> tracker) {
+                    tracker.removedBundle(bundle, event, null);
+                }
+            });
+            LOG.trace("Removed bundle event for {} finished successfully.",bundle);
+        } catch (InterruptedException | ExecutionException e) {
+            LOG.error("Addition of bundle failed, ", e);
+        }
     }
 
     }
 
-    private void forEachAdditionalBundle(BundleStrategy lambda) {
+    private void forEachAdditionalBundle(final BundleStrategy lambda) {
         for (BundleTrackerCustomizer<?> trac : additionalTrackers) {
             lambda.execute(trac);
         }
         for (BundleTrackerCustomizer<?> trac : additionalTrackers) {
             lambda.execute(trac);
         }

©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.