diff --git a/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java b/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java index 2aa75037c..b177e7074 100644 --- a/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java +++ b/core/camel-core-osgi/src/main/java/org/apache/camel/karaf/core/OsgiTypeConverter.java @@ -20,8 +20,12 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; import org.apache.camel.CamelContext; import org.apache.camel.Exchange; @@ -57,6 +61,32 @@ public class OsgiTypeConverter extends ServiceSupport implements TypeConverter, private CamelContext camelContext; private final Injector injector; private final ServiceTracker tracker; + /** + * The loaders the tracker has handed us, kept here rather than read back from the tracker: resolving them + * through the tracker inside {@link #createRegistry()} would mean calling into the ServiceTracker and the + * framework while holding this instance's monitor. + */ + private final Map, TypeConverterLoader> trackedLoaders + = new ConcurrentHashMap<>(); + /** + * Registrations made through this facade rather than by a {@link TypeConverterLoader}, so a rebuilt registry can + * be brought back to the same state. Discarding the delegate would otherwise drop them with no way to get them + * back. + *

+ * Keyed rather than a plain list, and insertion ordered so replay keeps the original order. The key is what the + * registration is about - a {@link TypeConvertible} for a converter, the contributed instance for the + * bulk and fallback forms, a sentinel for each setter - so re-registering replaces instead of appending, and + * removing deletes the entry instead of appending an inverse. Both matter in OSGi: each registration strongly + * references the converter it captured, and through it the classloader of the bundle that contributed it, so a + * collection that only ever grows pins bundles that have long since been uninstalled. + *

+ * Guarded by this instance's monitor, the same one {@link #getDelegate()} takes. + */ + private final Map> programmaticRegistrations = new LinkedHashMap<>(); + + private static final Object INJECTOR_KEY = new Object(); + private static final Object TYPE_CONVERTER_EXISTS_KEY = new Object(); + private static final Object TYPE_CONVERTER_EXISTS_LOGGING_LEVEL_KEY = new Object(); private volatile DefaultTypeConverter delegate; private volatile boolean trackerOpened; @@ -67,27 +97,36 @@ public OsgiTypeConverter(BundleContext bundleContext, CamelContext camelContext, this.tracker = new ServiceTracker<>(bundleContext, TypeConverterLoader.class.getName(), this); } - private void ensureTrackerOpen() { + private synchronized void ensureTrackerOpen() { if (!trackerOpened) { tracker.open(); trackerOpened = true; } } + // deliberately not synchronized: the tracker calls this from the framework's service event dispatch, and + // taking this instance's monitor here would put our lock on the far side of the framework's, which is the + // ordering that makes a lock inversion possible @Override public Object addingService(ServiceReference serviceReference) { LOG.trace("AddingService: {}, Bundle: {}", serviceReference, serviceReference.getBundle()); TypeConverterLoader loader = bundleContext.getService(serviceReference); if (loader != null) { + trackedLoaders.put(serviceReference, loader); try { LOG.debug("loading type converter from bundle: {}", serviceReference.getBundle().getSymbolicName()); - if (delegate != null) { + DefaultTypeConverter current = delegate; + if (current != null) { // load the converter directly into the existing delegate to preserve // any converters that were added programmatically (e.g. via Blueprint beans // implementing TypeConverters) - loader.load(delegate); + loader.load(current); } } catch (Throwable t) { + // the tracker treats a customizer that throws as "never tracked", so it will not call + // removedService for this reference and nothing else will release the use count taken above + trackedLoaders.remove(serviceReference); + ungetQuietly(serviceReference); throw new RuntimeCamelException("Error loading type converters from service: " + serviceReference + " due: " + t.getMessage(), t); } } @@ -99,9 +138,21 @@ public Object addingService(ServiceReference serviceReferen public void modifiedService(ServiceReference serviceReference, Object o) { } + // not synchronized, for the same reason as addingService @Override public void removedService(ServiceReference serviceReference, Object o) { LOG.trace("RemovedService: {}, Bundle: {}", serviceReference, serviceReference.getBundle()); + trackedLoaders.remove(serviceReference); + // we took the service in addingService, so releasing it is ours to do + ungetQuietly(serviceReference); + if (this.delegate != null && !isStopping() && !isStopped()) { + // worth saying out loud: one loader going away discards the whole registry, and the rebuild is a + // full reload of the core converters plus every remaining loader, not an incremental removal + LOG.warn("TypeConverterLoader from bundle {} was unregistered, discarding the type converter registry;" + + " it is rebuilt on next use from the remaining loaders, and the converters registered" + + " programmatically on this context are replayed onto the rebuilt registry.", + serviceReference.getBundle() != null ? serviceReference.getBundle().getSymbolicName() : serviceReference); + } try { ServiceHelper.stopService(this.delegate); } catch (Exception e) { @@ -112,6 +163,16 @@ public void removedService(ServiceReference serviceReferenc this.delegate = null; } + private void ungetQuietly(ServiceReference serviceReference) { + try { + bundleContext.ungetService(serviceReference); + } catch (Exception e) { + // the bundle or the framework may already be gone; releasing the use count is best effort + LOG.debug("Error ungetting service {} due: {}. This exception will be ignored.", serviceReference, + e.getMessage(), e); + } + } + @Override protected void doStart() throws Exception { ensureTrackerOpen(); @@ -121,6 +182,14 @@ protected void doStart() throws Exception { protected void doStop() throws Exception { this.tracker.close(); this.trackerOpened = false; + // close() calls removedService for everything still tracked, this only makes the end state explicit + this.trackedLoaders.clear(); + synchronized (this) { + // ServiceSupport allows stop then start again, and this is a context scoped service, so without this a + // restart would replay the previous lifecycle's registrations - including converters belonging to + // bundles that are gone by the time the context comes back up + this.programmaticRegistrations.clear(); + } ServiceHelper.stopService(this.delegate); this.delegate = null; } @@ -173,27 +242,33 @@ public T tryConvertTo(Class type, Object value) { @Override public void addTypeConverter(Class toType, Class fromType, TypeConverter typeConverter) { - getDelegate().addTypeConverter(toType, fromType, typeConverter); + register(new TypeConvertible<>(fromType, toType), + registry -> registry.addTypeConverter(toType, fromType, typeConverter)); } @Override public void addTypeConverters(Object typeConverters) { - getDelegate().addTypeConverters(typeConverters); + register(typeConverters, registry -> registry.addTypeConverters(typeConverters)); } @Override public void addBulkTypeConverters(BulkTypeConverters bulkTypeConverters) { - getDelegate().addBulkTypeConverters(bulkTypeConverters); + register(bulkTypeConverters, registry -> registry.addBulkTypeConverters(bulkTypeConverters)); } @Override - public boolean removeTypeConverter(Class toType, Class fromType) { - return getDelegate().removeTypeConverter(toType, fromType); + public synchronized boolean removeTypeConverter(Class toType, Class fromType) { + boolean removed = getDelegate().removeTypeConverter(toType, fromType); + // delete the matching registration rather than recording an inverse. An inverse would reproduce the same + // end state on replay, but it would also keep the removed converter - and its bundle's classloader - + // strongly reachable for the life of the context, and cost a no-op call on every future rebuild + programmaticRegistrations.remove(new TypeConvertible<>(fromType, toType)); + return removed; } @Override public void addFallbackTypeConverter(TypeConverter typeConverter, boolean canPromote) { - getDelegate().addFallbackTypeConverter(typeConverter, canPromote); + register(typeConverter, registry -> registry.addFallbackTypeConverter(typeConverter, canPromote)); } @Override @@ -203,7 +278,7 @@ public TypeConverter lookup(Class toType, Class fromType) { @Override public void setInjector(Injector injector) { - getDelegate().setInjector(injector); + register(INJECTOR_KEY, registry -> registry.setInjector(injector)); } @Override @@ -228,7 +303,8 @@ public LoggingLevel getTypeConverterExistsLoggingLevel() { @Override public void setTypeConverterExistsLoggingLevel(LoggingLevel loggingLevel) { - getDelegate().setTypeConverterExistsLoggingLevel(loggingLevel); + register(TYPE_CONVERTER_EXISTS_LOGGING_LEVEL_KEY, + registry -> registry.setTypeConverterExistsLoggingLevel(loggingLevel)); } @Override @@ -238,10 +314,13 @@ public TypeConverterExists getTypeConverterExists() { @Override public void setTypeConverterExists(TypeConverterExists typeConverterExists) { - getDelegate().setTypeConverterExists(typeConverterExists); + register(TYPE_CONVERTER_EXISTS_KEY, registry -> registry.setTypeConverterExists(typeConverterExists)); } - public DefaultTypeConverter getDelegate() { + // fully synchronized rather than double checked: the delegate is not immutable after publication - + // removedService stops and replaces it - so a lock free read of the field buys a race for no real gain, + // conversion work dwarfing an uncontended monitor either way + public synchronized DefaultTypeConverter getDelegate() { if (delegate == null) { // ensure the tracker is open so we can discover TypeConverterLoader services // before creating the registry - this is important because getDelegate() may be @@ -279,28 +358,73 @@ public Set getClassLoaders() { throw new RuntimeCamelException("Error loading CoreTypeConverter due: " + e.getMessage(), e); } - // Load the type converters the tracker has been tracking - // Here we need to use the ServiceReference to check the ranking - ServiceReference[] serviceReferences = this.tracker.getServiceReferences(); - if (serviceReferences != null) { - ArrayList> servicesList = - new ArrayList<>(Arrays.asList(serviceReferences)); - // Just make sure we install the high ranking fallback converter at last - Collections.sort(servicesList); - for (ServiceReference sr : servicesList) { - try { - LOG.debug("loading type converter from bundle: {}", sr.getBundle().getSymbolicName()); - ((TypeConverterLoader)this.tracker.getService(sr)).load(answer); - } catch (Throwable t) { - throw new RuntimeCamelException("Error loading type converters from service: " + sr + " due: " + t.getMessage(), t); - } + // Load the type converters the tracker has been tracking. These come from our own map rather than from + // tracker.getServiceReferences()/getService(): this runs while holding this instance's monitor, and + // calling back into the tracker from here is what would establish a lock ordering against the framework. + List> servicesList = new ArrayList<>(trackedLoaders.keySet()); + // Just make sure we install the high ranking fallback converter at last + Collections.sort(servicesList); + for (ServiceReference sr : servicesList) { + TypeConverterLoader loader = trackedLoaders.get(sr); + if (loader == null) { + // unregistered between the snapshot and here + continue; + } + try { + LOG.debug("loading type converter from bundle: {}", sr.getBundle().getSymbolicName()); + loader.load(answer); + } catch (Throwable t) { + throw new RuntimeCamelException("Error loading type converters from service: " + sr + " due: " + t.getMessage(), t); } } + replayProgrammaticRegistrations(answer); + LOG.trace("Created TypeConverter: {}", answer); return answer; } + /** + * Re-applies everything that was registered through this facade rather than by a + * {@link TypeConverterLoader}, in the order it was originally applied. + */ + private void replayProgrammaticRegistrations(DefaultTypeConverter registry) { + if (programmaticRegistrations.isEmpty()) { + return; + } + LOG.debug("Replaying {} programmatic registration(s) onto the rebuilt type converter registry", + programmaticRegistrations.size()); + for (Consumer registration : programmaticRegistrations.values()) { + registration.accept(registry); + } + } + + /** + * Applies a registration to the current delegate and remembers it under {@code key}, so that discarding the + * delegate does not discard the registration with it. + *

+ * Synchronized, and on the same monitor {@link #getDelegate()} takes: applying and recording have to be one + * step. A rebuild landing between them would replay a list this registration is not in yet and then hand back + * a registry it was never applied to, so the converter would be recorded but not actually present - the same + * loss this class is meant to prevent, one level down. The monitor is reentrant, so the nested + * {@link #getDelegate()} is free. + */ + /** + * How many programmatic registrations are currently retained for replay. Package private for the tests, which + * pin the invariant that this does not grow without bound - each entry retains the converter it captured, and + * with it the classloader of the contributing bundle. + */ + synchronized int programmaticRegistrationCount() { + return programmaticRegistrations.size(); + } + + private synchronized void register(Object key, Consumer registration) { + // apply first: a registration the delegate rejects is not one worth replaying. getDelegate() may build the + // registry here, replaying the map as it stands - this entry goes in after, so it cannot be applied twice + registration.accept(getDelegate()); + programmaticRegistrations.put(key, registration); + } + private class OsgiDefaultTypeConverter extends DefaultTypeConverter { public OsgiDefaultTypeConverter(PackageScanClassResolver resolver, Injector injector, boolean loadTypeConverters, @@ -332,7 +456,7 @@ public Map, TypeConverter> lookup(Class toType) { @Override public void addConverter(TypeConvertible typeConvertible, TypeConverter typeConverter) { - getDelegate().addConverter(typeConvertible, typeConverter); + register(typeConvertible, registry -> registry.addConverter(typeConvertible, typeConverter)); } } \ No newline at end of file diff --git a/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java b/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java index 838193400..92f9b89c4 100644 --- a/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java +++ b/core/camel-core-osgi/src/test/java/org/apache/camel/karaf/core/OsgiTypeConverterTest.java @@ -17,7 +17,20 @@ package org.apache.camel.karaf.core; import org.apache.camel.CamelContext; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.TypeConverter; import org.apache.camel.spi.Injector; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.impl.converter.DefaultTypeConverter; import org.apache.camel.spi.TypeConverterLoader; import org.apache.camel.spi.TypeConverterRegistry; import org.junit.jupiter.api.BeforeEach; @@ -46,6 +59,8 @@ public class OsgiTypeConverterTest { @Mock private TypeConverterLoader loader; @Mock + private TypeConverter typeConverter; + @Mock private Bundle bundle; private OsgiTypeConverter osgiTypeConverter; @@ -110,4 +125,243 @@ void removedServiceShouldInvalidateDelegate() throws Exception { var delegateAfter = osgiTypeConverter.getDelegate(); assertNotNull(delegateAfter); } + + @Test + void concurrentFirstAccessShouldBuildTheRegistryOnce() throws Exception { + int threads = 16; + AtomicInteger created = new AtomicInteger(); + CountDownLatch startLine = new CountDownLatch(1); + + OsgiTypeConverter counting = new OsgiTypeConverter(bundleContext, camelContext, injector) { + @Override + protected DefaultTypeConverter createRegistry() { + created.incrementAndGet(); + return super.createRegistry(); + } + }; + + ExecutorService pool = Executors.newFixedThreadPool(threads); + try { + List> futures = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + futures.add(pool.submit(() -> { + startLine.await(); + return counting.getDelegate(); + })); + } + // release them all at once so they race on the null check + startLine.countDown(); + + DefaultTypeConverter first = futures.get(0).get(30, TimeUnit.SECONDS); + assertNotNull(first); + for (Future f : futures) { + assertSame(first, f.get(30, TimeUnit.SECONDS), + "every caller must see the same registry instance"); + } + } finally { + pool.shutdownNow(); + } + + assertEquals(1, created.get(), + "the registry must be built exactly once, otherwise converters registered on a discarded" + + " instance are silently lost"); + } + + /** Marker source type, so the registered converter cannot collide with a core one. */ + interface Marker { + } + + @Test + void rebuiltRegistryReloadsTheTrackedLoadersWithoutAskingTheTracker() throws Exception { + // arrives before the registry exists, so addingService only records it + osgiTypeConverter.addingService(serviceReference); + verify(loader, never()).load(any()); + + DefaultTypeConverter first = osgiTypeConverter.getDelegate(); + + // createRegistry replayed it from the recorded loaders; it never called back into the ServiceTracker, + // which is what would put a framework call underneath this instance's monitor + verify(loader).load(first); + } + + @Test + void programmaticConverterSurvivesARegistryRebuild() throws Exception { + DefaultTypeConverter before = osgiTypeConverter.getDelegate(); + osgiTypeConverter.addTypeConverter(String.class, Marker.class, typeConverter); + assertNotNull(before.lookup(String.class, Marker.class), "precondition: the converter is registered"); + + // one loader going away discards the whole registry + osgiTypeConverter.removedService(serviceReference, loader); + DefaultTypeConverter after = osgiTypeConverter.getDelegate(); + + assertNotSame(before, after, "the registry should have been rebuilt"); + assertNotNull(after.lookup(String.class, Marker.class), + "a converter registered programmatically must be replayed onto the rebuilt registry, otherwise it" + + " disappears from a running context when any bundle unregisters a loader"); + } + + @Test + void removedTypeConverterIsNotResurrectedByARebuild() throws Exception { + osgiTypeConverter.addTypeConverter(String.class, Marker.class, typeConverter); + osgiTypeConverter.removeTypeConverter(String.class, Marker.class); + + osgiTypeConverter.removedService(serviceReference, loader); + + assertNull(osgiTypeConverter.getDelegate().lookup(String.class, Marker.class), + "the replay must reproduce the sequence, not just the additions"); + } + + @Test + void addingServiceReleasesTheServiceWhenLoadingFails() throws Exception { + osgiTypeConverter.getDelegate(); + doThrow(new RuntimeException("boom")).when(loader).load(any()); + + assertThrows(RuntimeCamelException.class, () -> osgiTypeConverter.addingService(serviceReference)); + + // a customizer that throws is treated as never tracked, so removedService will not run for this + // reference and nothing else would release the use count taken by addingService + verify(bundleContext).ungetService(serviceReference); + } + + @Test + void removedServiceReleasesTheService() { + osgiTypeConverter.addingService(serviceReference); + + osgiTypeConverter.removedService(serviceReference, loader); + + verify(bundleContext).ungetService(serviceReference); + } + + @Test + void addingServiceMustNotHoldTheInstanceMonitor() throws Exception { + // build first, so addingService takes the branch that calls into the loader + osgiTypeConverter.getDelegate(); + + CountDownLatch insideLoad = new CountDownLatch(1); + CountDownLatch releaseLoad = new CountDownLatch(1); + doAnswer(invocation -> { + insideLoad.countDown(); + releaseLoad.await(30, TimeUnit.SECONDS); + return null; + }).when(loader).load(any()); + + ExecutorService pool = Executors.newFixedThreadPool(2); + try { + Future adding = pool.submit(() -> osgiTypeConverter.addingService(serviceReference)); + assertTrue(insideLoad.await(30, TimeUnit.SECONDS), "addingService should have reached loader.load"); + + // the framework calls addingService while it is dispatching a service event; if it took this + // instance's monitor, every conversion in the container would block behind an arbitrary bundle's + // loader for as long as that loader takes + Future reader = pool.submit(osgiTypeConverter::getDelegate); + assertNotNull(reader.get(10, TimeUnit.SECONDS), + "getDelegate must not be blocked by an in-flight addingService"); + + releaseLoad.countDown(); + adding.get(30, TimeUnit.SECONDS); + } finally { + releaseLoad.countDown(); + pool.shutdownNow(); + } + } + + @Test + void addRemovePairsDoNotAccumulate() { + osgiTypeConverter.getDelegate(); + + for (int i = 0; i < 500; i++) { + osgiTypeConverter.addTypeConverter(String.class, Marker.class, typeConverter); + osgiTypeConverter.removeTypeConverter(String.class, Marker.class); + } + + // an inverse-appending list would sit at 1000 here, and every converter it captured - and the classloader + // of the bundle that contributed it - would stay strongly reachable for the life of the context + assertEquals(0, osgiTypeConverter.programmaticRegistrationCount(), + "add/remove pairs must prune, not accumulate"); + } + + @Test + void reRegisteringTheSameConversionReplaces() { + osgiTypeConverter.getDelegate(); + + for (int i = 0; i < 10; i++) { + // what a Blueprint container refresh looks like: the same conversion registered again + osgiTypeConverter.addTypeConverter(String.class, Marker.class, typeConverter); + } + + assertEquals(1, osgiTypeConverter.programmaticRegistrationCount(), + "re-registering the same conversion must replace rather than append"); + } + + @Test + void removingAConversionThatWasNeverRegisteredDoesNotAccumulate() { + osgiTypeConverter.getDelegate(); + + osgiTypeConverter.removeTypeConverter(String.class, Marker.class); + + assertEquals(0, osgiTypeConverter.programmaticRegistrationCount(), + "a removal for a pair that was never registered must not be retained"); + } + + @Test + void restartDoesNotReplayThePreviousLifecycle() throws Exception { + osgiTypeConverter.start(); + osgiTypeConverter.addTypeConverter(String.class, Marker.class, typeConverter); + assertNotNull(osgiTypeConverter.getDelegate().lookup(String.class, Marker.class)); + + osgiTypeConverter.stop(); + osgiTypeConverter.start(); + + // the converters captured before the stop belong to bundles that may be gone by now + assertEquals(0, osgiTypeConverter.programmaticRegistrationCount()); + assertNull(osgiTypeConverter.getDelegate().lookup(String.class, Marker.class), + "a stop/start cycle must not resurrect the previous lifecycle's registrations"); + } + + @Test + void registrationCannotInterleaveWithARebuild() throws Exception { + CountDownLatch insideBuild = new CountDownLatch(1); + CountDownLatch releaseBuild = new CountDownLatch(1); + + OsgiTypeConverter stalling = new OsgiTypeConverter(bundleContext, camelContext, injector) { + @Override + protected DefaultTypeConverter createRegistry() { + DefaultTypeConverter built = super.createRegistry(); + insideBuild.countDown(); + try { + releaseBuild.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return built; + } + }; + + ExecutorService pool = Executors.newFixedThreadPool(2); + try { + Future builder = pool.submit(stalling::getDelegate); + assertTrue(insideBuild.await(30, TimeUnit.SECONDS), "the rebuild should have started"); + + Future registrar = pool.submit(() -> { + stalling.addTypeConverter(String.class, Marker.class, typeConverter); + return null; + }); + + // apply-and-record has to be one step against the rebuild. If it were not, this registration would be + // applied to the registry being discarded and only recorded afterwards, so the rebuilt one would + // neither have it applied nor replay it + assertThrows(TimeoutException.class, () -> registrar.get(2, TimeUnit.SECONDS), + "a registration must not proceed while a rebuild holds the monitor"); + + releaseBuild.countDown(); + builder.get(30, TimeUnit.SECONDS); + registrar.get(30, TimeUnit.SECONDS); + + assertNotNull(stalling.getDelegate().lookup(String.class, Marker.class), + "the registration must land on the registry that is actually live"); + } finally { + releaseBuild.countDown(); + pool.shutdownNow(); + } + } }