Switch more of the "immediate" / "non-blocking" calls to be truly non-blocking.
Also updates tests. Mainly uses ImmediateSupplier and InterruptingImmediateSupplier for true non-blocking evaluation, with some other tricks used in other places. Some non-reliable calls may still fail, but most have been repaired, and the rest should be. (If the old semantics are _really_ needed you can resolve with a short wait.) Re-enables many of the tests disabled for https://issues.apache.org/jira/browse/BROOKLYN-272 Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/3821e02c Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/3821e02c Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/3821e02c Branch: refs/heads/master Commit: 3821e02c504382cb5a5a2411ddecda5b58b73136 Parents: 0d77dbc Author: Alex Heneveld <[email protected]> Authored: Mon Feb 20 15:48:46 2017 +0000 Committer: Alex Heneveld <[email protected]> Committed: Mon Feb 20 17:25:56 2017 +0000 ---------------------------------------------------------------------- .../spi/dsl/DslDeferredFunctionCall.java | 3 +- .../spi/dsl/methods/BrooklynDslCommon.java | 15 +--- .../brooklyn/spi/dsl/methods/DslComponent.java | 10 +-- .../brooklyn/camp/brooklyn/ConfigYamlTest.java | 32 +------- .../config/internal/AbstractConfigMapImpl.java | 3 + .../AbstractConfigurationSupportInternal.java | 57 +++++++------- .../core/objs/BrooklynObjectInternal.java | 13 +++- .../core/sensor/DependentConfiguration.java | 15 ++-- .../util/core/task/ImmediateSupplier.java | 43 ++++++++++- .../task/InterruptingImmediateSupplier.java | 7 +- .../brooklyn/util/core/task/ValueResolver.java | 78 ++++++++------------ .../brooklyn/core/entity/EntityConfigTest.java | 18 ++--- .../util/core/task/ValueResolverTest.java | 34 ++++++--- .../org/apache/brooklyn/util/guava/Maybe.java | 9 +++ 14 files changed, 182 insertions(+), 155 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslDeferredFunctionCall.java ---------------------------------------------------------------------- diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslDeferredFunctionCall.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslDeferredFunctionCall.java index b8387b3..1c1cef5 100644 --- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslDeferredFunctionCall.java +++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslDeferredFunctionCall.java @@ -27,6 +27,7 @@ import org.apache.brooklyn.api.mgmt.Task; import org.apache.brooklyn.camp.brooklyn.spi.dsl.methods.BrooklynDslCommon; import org.apache.brooklyn.camp.brooklyn.spi.dsl.methods.DslToStringHelpers; import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; +import org.apache.brooklyn.util.core.task.ImmediateSupplier; import org.apache.brooklyn.util.core.task.Tasks; import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.guava.Maybe; @@ -94,7 +95,7 @@ public class DslDeferredFunctionCall extends BrooklynDslDeferredSupplier<Object> return invokeOn(instance); } else { if (immediate) { - return Maybe.absent("Could not evaluate immediately: " + obj); + return Maybe.absent(new ImmediateSupplier.ImmediateValueNotAvailableException("Could not evaluate immediately: " + obj)); } else { return Maybe.absent(Maybe.getException(resolvedMaybe)); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/BrooklynDslCommon.java ---------------------------------------------------------------------- diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/BrooklynDslCommon.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/BrooklynDslCommon.java index 2836895..e0ba90f 100644 --- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/BrooklynDslCommon.java +++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/BrooklynDslCommon.java @@ -589,18 +589,11 @@ public class BrooklynDslCommon { final Class<?> clazz = getOrLoadType(); final ExecutionContext executionContext = entity().getExecutionContext(); - // Marker exception that one of our component-parts cannot yet be resolved - - // throwing and catching this allows us to abort fast. - // A bit messy to use exceptions in normal control flow, but this allows the Maps util methods to be used. - @SuppressWarnings("serial") - class UnavailableException extends RuntimeException { - } - final Function<Object, Object> resolver = new Function<Object, Object>() { @Override public Object apply(Object value) { Maybe<Object> result = Tasks.resolving(value, Object.class).context(executionContext).deep(true).immediately(true).getMaybe(); if (result.isAbsent()) { - throw new UnavailableException(); + throw new ImmediateValueNotAvailableException(); } else { return result.get(); } @@ -620,8 +613,8 @@ public class BrooklynDslCommon { result = create(clazz, factoryMethodName, resolvedFactoryMethodArgs, resolvedFields, resolvedConfig); } return Maybe.of(result); - } catch (UnavailableException e) { - return Maybe.absent(); + } catch (ImmediateValueNotAvailableException e) { + return ImmediateValueNotAvailableException.newAbsentWithExceptionSupplier(); } } @@ -873,7 +866,7 @@ public class BrooklynDslCommon { public Maybe<Entity> getImmediately() { EntityInternal entity = entity(); if (entity == null) { - return Maybe.absent(); + return Maybe.absent("No entity available"); } Entity targetEntity = entity.getManagementContext().getEntityManager().getEntity(entityId); return Maybe.of(targetEntity); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java ---------------------------------------------------------------------- diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java index 1cce90e..0d2213a 100644 --- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java +++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java @@ -288,7 +288,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements if (immediate) { if (maybeComponentId.isAbsent()) { - return Maybe.absent(Maybe.getException(maybeComponentId)); + return ImmediateValueNotAvailableException.newAbsentWrapping("Cannot find component ID", maybeComponentId); } } @@ -418,7 +418,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements @Override public Maybe<Object> getImmediately() { Maybe<Entity> targetEntityMaybe = component.getImmediately(); - if (targetEntityMaybe.isAbsent()) return Maybe.absent("Target entity not available"); + if (targetEntityMaybe.isAbsent()) return ImmediateValueNotAvailableException.newAbsentWrapping("Target entity is not available: "+component, targetEntityMaybe); Entity targetEntity = targetEntityMaybe.get(); return Maybe.<Object>of(targetEntity.getId()); @@ -477,7 +477,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements @Override public final Maybe<Object> getImmediately() { Maybe<Entity> targetEntityMaybe = component.getImmediately(); - if (targetEntityMaybe.isAbsent()) return Maybe.absent("Target entity not available"); + if (targetEntityMaybe.isAbsent()) return ImmediateValueNotAvailableException.newAbsentWrapping("Target entity not available: "+component, targetEntityMaybe); Entity targetEntity = targetEntityMaybe.get(); String sensorNameS = resolveSensorName(true); @@ -486,7 +486,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements targetSensor = Sensors.newSensor(Object.class, sensorNameS); } Object result = targetEntity.sensors().get(targetSensor); - return GroovyJavaMethods.truth(result) ? Maybe.of(result) : Maybe.absent(); + return GroovyJavaMethods.truth(result) ? Maybe.of(result) : ImmediateValueNotAvailableException.newAbsentWithExceptionSupplier(); } @SuppressWarnings("unchecked") @@ -660,7 +660,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements return Maybe.<Sensor<?>>of((Sensor<?>)si); } else if (si instanceof String) { Maybe<Entity> targetEntityMaybe = component.getImmediately(); - if (targetEntityMaybe.isAbsent()) return Maybe.absent("Target entity not available"); + if (targetEntityMaybe.isAbsent()) return ImmediateValueNotAvailableException.newAbsentWrapping("Target entity is not available: "+component, targetEntityMaybe); Entity targetEntity = targetEntityMaybe.get(); Sensor<?> result = null; http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/ConfigYamlTest.java ---------------------------------------------------------------------- diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/ConfigYamlTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/ConfigYamlTest.java index e1f1708..53a6be8 100644 --- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/ConfigYamlTest.java +++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/ConfigYamlTest.java @@ -307,16 +307,7 @@ public class ConfigYamlTest extends AbstractYamlTest { assertEquals(entity.config().get(TestEntity.CONF_SET_PLAIN), ImmutableSet.of("myOther")); } - /** - * TODO The {@code entity.config().getNonBlocking()} can return absent. When it's called with - * a deferred supplier value, it will kick off a task and then wait just a few millis for that - * task to execute deferredSupplier.get(). If it times out, then it returns Maybe.absent. - * However, on apache jenkins the machine is often slow so the task doesn't complete in the - * given number of millis (even though deferredSupplier.get() doesn't need to block for anything). - * Same for {@link #testDeferredSupplierToAttributeWhenReadyInSpecialTypes()}. - * See https://issues.apache.org/jira/browse/BROOKLYN-272. - */ - @Test(groups="Broken") + @Test public void testDeferredSupplierToAttributeWhenReady() throws Exception { String yaml = Joiner.on("\n").join( "services:", @@ -344,17 +335,9 @@ public class ConfigYamlTest extends AbstractYamlTest { /** * This tests config keys of type {@link org.apache.brooklyn.core.config.MapConfigKey}, etc. - * For plain maps, see {@link #testDeferredSupplierToAttributeWhenReadyInPlainCollections()}. - * - * TODO The {@code entity.config().getNonBlocking()} can return absent. When it's called with - * a deferred supplier value, it will kick off a task and then wait just a few millis for that - * task to execute deferredSupplier.get(). If it times out, then it returns Maybe.absent. - * However, on apache jenkins the machine is often slow so the task doesn't complete in the - * given number of millis (even though deferredSupplier.get() doesn't need to block for anything). - * Same for {@link #testDeferredSupplierToAttributeWhenReady()}. - * See https://issues.apache.org/jira/browse/BROOKLYN-272. + * For plain maps, see {@link #testDeferredSupplierToAttributeWhenReadyInPlainCollections()} */ - @Test(groups="Broken") + @Test public void testDeferredSupplierToAttributeWhenReadyInSpecialTypes() throws Exception { String yaml = Joiner.on("\n").join( "services:", @@ -399,15 +382,8 @@ public class ConfigYamlTest extends AbstractYamlTest { * This tests config keys of type {@link java.util.Map}, etc. * For special types (e.g. {@link org.apache.brooklyn.core.config.MapConfigKey}), see * {@link #testDeferredSupplierToAttributeWhenReadyInPlainCollections()}. - * - * TODO test doesn't work because getNonBlocking returns even when no value. - * For example, we get back: Present[value={mykey=attributeWhenReady("myOtherSensor")}]. - * However, the `config().get()` does behave as desired. - * - * Including the "WIP" group because this test would presumably have never worked! - * Added to demonstrate the short-coming. */ - @Test(groups={"Broken", "WIP"}) + @Test public void testDeferredSupplierToAttributeWhenReadyInPlainCollections() throws Exception { String yaml = Joiner.on("\n").join( "services:", http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java b/core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java index b736beb..091f874 100644 --- a/core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java +++ b/core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java @@ -313,6 +313,9 @@ public abstract class AbstractConfigMapImpl<TContainer extends BrooklynObject> i // wasteful to make a copy to look up; maybe try once opportunistically? ownCopy = MutableMap.copyOf(oc); } + // would be cleaner here to have an extractValueMaybe but semantics can get confusing whether absent + // means no value can be extracted (getRaw semantics) and immediate mode is on but blocking is needed (ImmediateSupplier semantics); + // simpler not to support maybe, in which case here null means the former, and the latter throws something (which the caller catches) Maybe<Object> result = Maybe.of((Object) ((ConfigKeySelfExtracting<?>) key).extractValue(ownCopy, getExecutionContext(container)) ); postLocalEvaluate(key, bo, value, result); return result; http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/core/objs/AbstractConfigurationSupportInternal.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/objs/AbstractConfigurationSupportInternal.java b/core/src/main/java/org/apache/brooklyn/core/objs/AbstractConfigurationSupportInternal.java index 796ab13..113daac 100644 --- a/core/src/main/java/org/apache/brooklyn/core/objs/AbstractConfigurationSupportInternal.java +++ b/core/src/main/java/org/apache/brooklyn/core/objs/AbstractConfigurationSupportInternal.java @@ -22,8 +22,6 @@ package org.apache.brooklyn.core.objs; import java.util.Map; import java.util.Set; import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeoutException; import javax.annotation.Nullable; @@ -41,8 +39,9 @@ import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; import org.apache.brooklyn.util.collections.MutableMap; import org.apache.brooklyn.util.core.config.ConfigBag; import org.apache.brooklyn.util.core.flags.TypeCoercions; +import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateUnsupportedException; +import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateValueNotAvailableException; import org.apache.brooklyn.util.core.task.Tasks; -import org.apache.brooklyn.util.core.task.ValueResolver; import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException; import org.apache.brooklyn.util.guava.Maybe; @@ -53,6 +52,7 @@ import com.google.common.base.Predicate; public abstract class AbstractConfigurationSupportInternal implements BrooklynObjectInternal.ConfigurationSupportInternal { + @SuppressWarnings("unused") private static final Logger LOG = LoggerFactory.getLogger(AbstractConfigurationSupportInternal.class); @Override @@ -77,10 +77,16 @@ public abstract class AbstractConfigurationSupportInternal implements BrooklynOb @Override public <T> Maybe<T> getNonBlocking(final ConfigKey<T> key) { - if (key instanceof StructuredConfigKey || key instanceof SubElementConfigKey) { - return getNonBlockingResolvingStructuredKey(key); - } else { - return getNonBlockingResolvingSimple(key); + try { + if (key instanceof StructuredConfigKey || key instanceof SubElementConfigKey) { + return getNonBlockingResolvingStructuredKey(key); + } else { + return getNonBlockingResolvingSimple(key); + } + } catch (ImmediateValueNotAvailableException e) { + return Maybe.absent(e); + } catch (ImmediateUnsupportedException e) { + return Maybe.absent(e); } } @@ -89,12 +95,6 @@ public abstract class AbstractConfigurationSupportInternal implements BrooklynOb * execute the custom logic, as is done by {@link #get(ConfigKey)}, but non-blocking! */ protected <T> Maybe<T> getNonBlockingResolvingStructuredKey(final ConfigKey<T> key) { - // TODO This is a poor implementation. We risk timing out when it's just doing its - // normal work (e.g. because job's thread was starved), rather than when it's truly - // blocked. Really we'd need to dig into the implementation of get(key), so that the - // underlying work can be configured with a timeout, for when it finally calls - // ValueResolver. - Callable<T> job = new Callable<T>() { @Override public T call() { @@ -106,22 +106,15 @@ public abstract class AbstractConfigurationSupportInternal implements BrooklynOb } }; - Task<T> t = getContext().submit(Tasks.<T>builder().body(job) + Task<T> t = Tasks.<T>builder().body(job) .displayName("Resolving dependent value") .description("Resolving "+key.getName()) .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG) - .build()); + .build(); try { - T result = t.get(ValueResolver.NON_BLOCKING_WAIT); - return Maybe.of(result); - } catch (TimeoutException e) { - t.cancel(true); - return Maybe.<T>absent(); - } catch (ExecutionException e) { - LOG.debug("Problem resolving "+key.getName()+", returning <absent>", e); - return Maybe.<T>absent(); - } catch (InterruptedException e) { - throw Exceptions.propagate(e); + return getContext().getImmediately(t); + } catch (ImmediateUnsupportedException e) { + return Maybe.absent(); } } @@ -139,17 +132,19 @@ public abstract class AbstractConfigurationSupportInternal implements BrooklynOb // or Absent if the config key was unset. Object unresolved = getRaw(key).or(key.getDefaultValue()); final Object marker = new Object(); - // Give tasks a short grace period to resolve. - Object resolved = Tasks.resolving(unresolved) + Maybe<Object> resolved = Tasks.resolving(unresolved) .as(Object.class) .defaultValue(marker) .immediately(true) .deep(true) .context(getContext()) - .get(); - return (resolved != marker) - ? TypeCoercions.tryCoerce(resolved, key.getTypeToken()) - : Maybe.<T>absent(); + .getMaybe(); + if (resolved.isAbsent()) return Maybe.Absent.<T>castAbsent(resolved); + if (resolved.get()==marker) { + // TODO changed Feb 2017, previously returned absent, in contrast to what the javadoc says + return Maybe.of((T)null); + } + return TypeCoercions.tryCoerce(resolved.get(), key.getTypeToken()); } @Override http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/core/objs/BrooklynObjectInternal.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/objs/BrooklynObjectInternal.java b/core/src/main/java/org/apache/brooklyn/core/objs/BrooklynObjectInternal.java index 86daf1f..3fc2c99 100644 --- a/core/src/main/java/org/apache/brooklyn/core/objs/BrooklynObjectInternal.java +++ b/core/src/main/java/org/apache/brooklyn/core/objs/BrooklynObjectInternal.java @@ -28,6 +28,8 @@ import org.apache.brooklyn.config.ConfigKey; import org.apache.brooklyn.config.ConfigKey.HasConfigKey; import org.apache.brooklyn.config.ConfigMap.ConfigMapWithInheritance; import org.apache.brooklyn.util.core.config.ConfigBag; +import org.apache.brooklyn.util.core.task.ImmediateSupplier; +import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateUnsupportedException; import org.apache.brooklyn.util.guava.Maybe; import com.google.common.annotations.Beta; @@ -113,13 +115,20 @@ public interface BrooklynObjectInternal extends BrooklynObject, Rebindable { /** * Attempts to coerce the value for this config key, if available, - * taking a default and {@link Maybe#absent absent} if the uncoerced - * cannot be resolved within a short timeframe. + * including returning a default if the config key is unset, + * returning a {@link Maybe#absent absent} if the uncoerced + * does not support immediate resolution. * <p> * Note: if no value for the key is available, not even as a default, * this returns a {@link Maybe#isPresent()} containing <code>null</code> * (following the semantics of {@link #get(ConfigKey)} * rather than {@link #getRaw(ConfigKey)}). + * Thus a {@link Maybe#absent()} definitively indicates that + * the absence is due to the request to evaluate immediately. + * <p> + * This will include catching {@link ImmediateUnsupportedException} + * and returning it as an absence, thus making the semantics here slightly + * "safer" than that of {@link ImmediateSupplier#getImmediately()}. */ @Beta <T> Maybe<T> getNonBlocking(ConfigKey<T> key); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java b/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java index ca79d73..32a409f 100644 --- a/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java +++ b/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java @@ -68,6 +68,7 @@ import org.apache.brooklyn.util.exceptions.RuntimeTimeoutException; import org.apache.brooklyn.util.groovy.GroovyJavaMethods; import org.apache.brooklyn.util.guava.Functionals; import org.apache.brooklyn.util.guava.Maybe; +import org.apache.brooklyn.util.guava.Maybe.Absent; import org.apache.brooklyn.util.net.Urls; import org.apache.brooklyn.util.text.StringFunctions; import org.apache.brooklyn.util.text.StringFunctions.RegexReplacer; @@ -499,7 +500,7 @@ public class DependentConfiguration { List<Object> resolvedArgs = Lists.newArrayList(); for (Object arg : args) { Maybe<?> argVal = resolveImmediately(arg); - if (argVal.isAbsent()) return Maybe.absent(); + if (argVal.isAbsent()) return Maybe.Absent.castAbsent(argVal); resolvedArgs.add(argVal.get()); } @@ -511,7 +512,7 @@ public class DependentConfiguration { */ public static Maybe<String> urlEncodeImmediately(Object arg) { Maybe<?> resolvedArg = resolveImmediately(arg); - if (resolvedArg.isAbsent()) return Maybe.absent(); + if (resolvedArg.isAbsent()) return Absent.castAbsent(resolvedArg); if (resolvedArg.isNull()) return Maybe.<String>of((String)null); String resolvedString = resolvedArg.get().toString(); @@ -564,15 +565,15 @@ public class DependentConfiguration { public static Maybe<String> regexReplacementImmediately(Object source, Object pattern, Object replacement) { Maybe<?> resolvedSource = resolveImmediately(source); - if (resolvedSource.isAbsent()) return Maybe.absent(); + if (resolvedSource.isAbsent()) return Absent.castAbsent(resolvedSource); String resolvedSourceStr = String.valueOf(resolvedSource.get()); Maybe<?> resolvedPattern = resolveImmediately(pattern); - if (resolvedPattern.isAbsent()) return Maybe.absent(); + if (resolvedPattern.isAbsent()) return Absent.castAbsent(resolvedPattern); String resolvedPatternStr = String.valueOf(resolvedPattern.get()); Maybe<?> resolvedReplacement = resolveImmediately(replacement); - if (resolvedReplacement.isAbsent()) return Maybe.absent(); + if (resolvedReplacement.isAbsent()) return Absent.castAbsent(resolvedReplacement); String resolvedReplacementStr = String.valueOf(resolvedReplacement.get()); String result = new StringFunctions.RegexReplacer(resolvedPatternStr, resolvedReplacementStr).apply(resolvedSourceStr); @@ -591,11 +592,11 @@ public class DependentConfiguration { public static Maybe<Function<String, String>> regexReplacementImmediately(Object pattern, Object replacement) { Maybe<?> resolvedPattern = resolveImmediately(pattern); - if (resolvedPattern.isAbsent()) return Maybe.absent(); + if (resolvedPattern.isAbsent()) return Absent.castAbsent(resolvedPattern); String resolvedPatternStr = String.valueOf(resolvedPattern.get()); Maybe<?> resolvedReplacement = resolveImmediately(replacement); - if (resolvedReplacement.isAbsent()) return Maybe.absent(); + if (resolvedReplacement.isAbsent()) return Absent.castAbsent(resolvedReplacement); String resolvedReplacementStr = String.valueOf(resolvedReplacement.get()); RegexReplacer result = new StringFunctions.RegexReplacer(resolvedPatternStr, resolvedReplacementStr); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/util/core/task/ImmediateSupplier.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/ImmediateSupplier.java b/core/src/main/java/org/apache/brooklyn/util/core/task/ImmediateSupplier.java index 5ec8d68..522c361 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/ImmediateSupplier.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/ImmediateSupplier.java @@ -45,9 +45,48 @@ public interface ImmediateSupplier<T> extends Supplier<T> { } /** - * Gets the value promptly, or returns {@link Maybe#absent()} if the value is not yet available. + * Indicates that an attempt was made to forcibly get a requested immediate value + * where blocking is required. See {@link ImmediateSupplier#getImmediately()}, which if + * it returns an absent result, that absent will throw this. + * <p> + * This is useful for passing between contexts that support immediate evaluation, + * through contexts that do not, to outer contexts which do, as the outer context + * will be able to use this exception to return a {@link Maybe#absent()} rather than throwing. + */ + public static class ImmediateValueNotAvailableException extends RuntimeException { + private static final long serialVersionUID = -5860437285154375232L; + + public ImmediateValueNotAvailableException() { } + public ImmediateValueNotAvailableException(String message) { + super(message); + } + public ImmediateValueNotAvailableException(String message, Throwable cause) { + super(message, cause); + } + public static <T> Maybe<T> newAbsentWithExceptionSupplier() { + return Maybe.Absent.changeExceptionSupplier(Maybe.<T>absent(), ImmediateValueNotAvailableException.class); + } + public static <T> Maybe<T> newAbsentWrapping(String message, Maybe<?> inner) { + return Maybe.absent(new ImmediateValueNotAvailableException(message, Maybe.getException(inner))); + } + } + + /** + * Gets the value promptly, or returns {@link Maybe#absent()} if the value requires blocking, + * or throws {@link ImmediateUnsupportedException} if it cannot be determined whether the value requires blocking or not. + * <p> + * The {@link Maybe#absent()} returned here indicates that a value definitively <i>is</i> pending, just it is not yet available, + * and an attempt to {@link Maybe#get()} it should throw an {@link ImmediateValueNotAvailableException}; + * it can be created with {@link ImmediateValueNotAvailableException#newAbsentWithExceptionSupplier()} to + * avoid creating traces (or simply with <code>Maybe.absent(new ImmediateValueNotAvailableException(...))</code>). + * This is in contrast with this method throwing a {@link ImmediateUnsupportedException} which should be done + * if the presence of an eventual value cannot even be determined in a non-blocking way. + * <p> + * Implementations of this method should typically catch the former exception if encountered and return a + * {@link Maybe#absent()} wrapping it, whereas {@link ImmediateUnsupportedException} instances should be propagated. * - * @throws ImmediateUnsupportedException if cannot determine whether a value is immediately available + * @throws ImmediateUnsupportedException as above, if cannot be determined whether a value is or might eventually be available */ Maybe<T> getImmediately(); + } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java b/core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java index 84b1bb4..0a34a18 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java @@ -58,9 +58,12 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D if (!interrupted) Thread.currentThread().interrupt(); return Maybe.ofAllowingNull(get()); } catch (Throwable t) { + if (Exceptions.getFirstThrowableOfType(t, ImmediateValueNotAvailableException.class)!=null) { + return Maybe.absent(Exceptions.getFirstThrowableOfType(t, ImmediateValueNotAvailableException.class)); + } if (Exceptions.getFirstThrowableOfType(t, InterruptedException.class)!=null || Exceptions.getFirstThrowableOfType(t, RuntimeInterruptedException.class)!=null) { - return Maybe.absent(new UnsupportedOperationException("Immediate value not available", t)); + return Maybe.absent(new ImmediateValueNotAvailableException("Immediate value not available, required non-blocking execution", t)); } throw Exceptions.propagate(t); } finally { @@ -105,7 +108,7 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D } } - public static class InterruptingImmediateSupplierNotSupportedForObject extends UnsupportedOperationException { + public static class InterruptingImmediateSupplierNotSupportedForObject extends ImmediateUnsupportedException { private static final long serialVersionUID = 307517409005386500L; public InterruptingImmediateSupplierNotSupportedForObject(Object o) { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/main/java/org/apache/brooklyn/util/core/task/ValueResolver.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/ValueResolver.java b/core/src/main/java/org/apache/brooklyn/util/core/task/ValueResolver.java index f8cb91b..4446d92 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/ValueResolver.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/ValueResolver.java @@ -33,6 +33,8 @@ import org.apache.brooklyn.api.mgmt.TaskFactory; import org.apache.brooklyn.core.entity.EntityInternal; import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; import org.apache.brooklyn.util.core.flags.TypeCoercions; +import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateUnsupportedException; +import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateValueNotAvailableException; import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.guava.Maybe; import org.apache.brooklyn.util.javalang.JavaClassNames; @@ -275,8 +277,10 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj } /** - * Whether the value should be resolved immediately (and if not available immediately, - * return absent). + * Whether the value should be resolved immediately + * (following {@link ImmediateSupplier#getImmediately()} semantics with regards to when errors may be thrown, + * except some cases where {@link ImmediateUnsupportedException} is thrown may be re-run with a {@link #NON_BLOCKING_WAIT} + * after which the more definitive {@link ImmediateValueNotAvailableException} will be thrown/wrapped. */ @Beta public ValueResolver<T> immediately(boolean val) { @@ -364,19 +368,15 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj boolean allowImmediateExecution = false; boolean bailOutAfterImmediateExecution = false; - if (v instanceof ImmediateSupplier) { + if (v instanceof ImmediateSupplier || v instanceof DeferredSupplier) { allowImmediateExecution = true; } else { - if ((v instanceof TaskFactory<?>) && !(v instanceof DeferredSupplier)) { + if (v instanceof TaskFactory<?>) { v = ((TaskFactory<?>)v).newTask(); allowImmediateExecution = true; bailOutAfterImmediateExecution = true; BrooklynTaskTags.setTransient(((TaskAdaptable<?>)v).asTask()); - if (isEvaluatingImmediately()) { - // not needed if executing immediately - BrooklynTaskTags.addTagDynamically( ((TaskAdaptable<?>)v).asTask(), BrooklynTaskTags.IMMEDIATE_TASK_TAG ); - } } //if it's a task or a future, we wait for the task to complete @@ -384,31 +384,32 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj v = ((TaskAdaptable<?>) v).asTask(); } } - + if (allowImmediateExecution && isEvaluatingImmediately()) { - // TODO could allow for everything, when evaluating immediately -- but if the target isn't safe to run again - // then we have to fail if immediate didn't work; to avoid breaking semantics we only do that for a few cases; - // might be nice to get to the point where we can break those semantics however, - // ie weakening what getImmediate supports and making it be non-blocking, so that bailOut=true is the default. - // if: v instanceof TaskFactory -- it is safe, it's a new API (but it is currently the only one supported); - // more than safe, we have to do it -- or add code here to cancel tasks -- because it spawns new tasks - // (other objects passed through here don't get cancelled, because other things might try again later; - // ie a task or future passed in here might naturally be long-running so cancelling is wrong, - // but with a task factory generated task it would leak if we submitted and didn't cancel!) - // if: v instanceof ImmediateSupplier -- it probably is safe to change to bailOut = true ? - // if: v instanceof Task or other things -- it currently isn't safe, there are places where - // we expect to getImmediate on things which don't support it nicely, - // and we rely on the blocking-short-wait behaviour, e.g. QuorumChecks in ConfigYamlTest + // Feb 2017 - many things now we try to run immediate; notable exceptions are: + // * where the target isn't safe to run again (such as a Task which someone else might need), + // * or where he can't be run in an "interrupting" mode even if non-blocking (eg Future.get(), some other tasks) + // (the latter could be tried here, with bailOut false, but in most cases it will just throw so we still need to + // have the timings as in SHORT_WAIT etc as a fallack) + + Maybe<T> result = null; try { - Maybe<T> result = execImmediate(exec, v); - if (result!=null) return result; + result = exec.getImmediately(v); + + return (result.isPresent()) + ? recursive + ? new ValueResolver<T>(result.get(), type, this).getMaybe() + : result + : result; + } catch (ImmediateSupplier.ImmediateUnsupportedException e) { if (bailOutAfterImmediateExecution) { throw new ImmediateSupplier.ImmediateUnsupportedException("Cannot get immediately: "+v); } - } catch (InterruptingImmediateSupplier.InterruptingImmediateSupplierNotSupportedForObject o) { - // ignore, continue below - log.debug("Unable to resolve-immediately for "+description+" ("+v+", wrong type "+v.getClass()+"); falling back to executing with timeout"); - } + // else proceed to below + } catch (ImmediateSupplier.ImmediateValueNotAvailableException e) { + // definitively not available + return ImmediateSupplier.ImmediateValueNotAvailableException.newAbsentWithExceptionSupplier(); + } } if (v instanceof Task) { @@ -472,8 +473,7 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj .description(description); if (isTransientTask) tb.tag(BrooklynTaskTags.TRANSIENT_TASK_TAG); - // Note that immediate resolution is handled by using ImmediateSupplier (using an instanceof check), - // so that it executes in the current thread instead of using task execution. + // immediate resolution is handled above Task<Object> vt = exec.submit(tb.build()); Maybe<Object> vm = Durations.get(vt, timer); vt.cancel(true); @@ -559,24 +559,6 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj } } - protected Maybe<T> execImmediate(ExecutionContext exec, Object immediateSupplierOrImmediateTask) { - Maybe<T> result; - try { - result = exec.getImmediately(immediateSupplierOrImmediateTask); - } catch (ImmediateSupplier.ImmediateUnsupportedException e) { - return null; - } - // let InterruptingImmediateSupplier.InterruptingImmediateSupplierNotSupportedForObject - // bet thrown, and caller who cares will catch that to know it can continue - - // Recurse: need to ensure returned value is cast, etc - return (result.isPresent()) - ? recursive - ? new ValueResolver<T>(result.get(), type, this).getMaybe() - : result - : result; - } - protected String getDescription() { return description!=null ? description : ""+value; } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java b/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java index 2f40fe9..90f8a12 100644 --- a/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java +++ b/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java @@ -374,7 +374,7 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport { assertAllOurConfigTasksCancelled(); } else { // TaskFactory tasks are cancelled, but others are not, - // things (ValueResolver?) are smart enough to know to leave it running + // things (ValueResolver.getMaybeInternal()) are smart enough to know to leave it running assertAllOurConfigTasksNotCancelled(); } @@ -411,30 +411,30 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport { } } - @Test(groups="Integration") // because takes 1s+ + @Test(groups="Integration") // still takes 1s+ -- because we don't want to interrupt the task, we have to run it in BG public void testGetTaskNonBlockingKey() throws Exception { new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInKey(); } - @Test(groups="Integration") // because takes 1s+ + @Test public void testGetTaskNonBlockingMap() throws Exception { new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInMap(); } - @Test(groups="Integration") // because takes 1s+ + @Test public void testGetTaskFactoryNonBlockingKey() throws Exception { new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey(); } - @Test(groups="Integration") // because takes 1s+ + @Test public void testGetTaskFactoryNonBlockingMap() throws Exception { new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInMap(); } - @Test(groups="Integration") // because takes 1s+ + @Test public void testGetSupplierNonBlockingKey() throws Exception { new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInKey(); } - @Test(groups="Integration") // because takes 1s+ - public void testGetSuppierNonBlockingMap() throws Exception { + @Test + public void testGetSupplierNonBlockingMap() throws Exception { new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInMap(); } @@ -442,7 +442,7 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport { public void testGetImmediateSupplierNonBlockingKey() throws Exception { new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInKey(); } - @Test(groups="Integration") // because takes 1s+ + @Test public void testGetImmediateSupplierNonBlockingMap() throws Exception { new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInMap(); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java b/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java index 550d475..c9c76fb 100644 --- a/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java +++ b/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java @@ -32,6 +32,7 @@ import org.apache.brooklyn.api.mgmt.TaskFactory; import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; import org.apache.brooklyn.core.test.BrooklynAppUnitTestSupport; import org.apache.brooklyn.test.Asserts; +import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateValueNotAvailableException; import org.apache.brooklyn.util.guava.Maybe; import org.apache.brooklyn.util.time.Duration; import org.apache.brooklyn.util.time.Time; @@ -169,11 +170,26 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport { assertImmediateFakeTaskFromMethod(callInfo, "myUniquelyNamedMethod"); } - public void testGetImmediatelyFallsBackToDeferredCallInTask() throws Exception { - final MyImmediateAndDeferredSupplier supplier = new MyImmediateAndDeferredSupplier(true); + public void testGetImmediatelyFallsBackToDeferredCallInTaskOnUnsupported() throws Exception { + final MyImmediateAndDeferredSupplier supplier = new MyImmediateAndDeferredSupplier(new ImmediateSupplier.ImmediateUnsupportedException("Simulate immediate unsupported")); CallInfo callInfo = Tasks.resolving(supplier).as(CallInfo.class).context(app).immediately(true).get(); - assertRealTaskNotFromMethod(callInfo, "testGetImmediatelyFallsBackToDeferredCallInTask"); + assertNotNull(callInfo.task); assertEquals(BrooklynTaskTags.getContextEntity(callInfo.task), app); + assertNotContainsCallingMethod(callInfo.stackTrace, "testGetImmediatelyFallsBackToDeferredCallInTask"); + } + + public void testGetImmediatelyDoesntFallBackToDeferredCallOnNotAvailable() throws Exception { + final MyImmediateAndDeferredSupplier supplier = new MyImmediateAndDeferredSupplier(new ImmediateSupplier.ImmediateValueNotAvailableException()); + Maybe<CallInfo> callInfo = Tasks.resolving(supplier).as(CallInfo.class).context(app).immediately(true).getMaybe(); + Asserts.assertNotPresent(callInfo); + + try { + callInfo.get(); + Asserts.shouldHaveFailedPreviously("resolution should have failed now the ImmediateSupplier is not expected to fallback to other evaluation forms; instead got "+callInfo); + + } catch (Exception e) { + Asserts.expectedFailureOfType(e, ImmediateValueNotAvailableException.class); + } } public void testNonRecursiveBlockingFailsOnNonObjectType() throws Exception { @@ -271,20 +287,20 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport { } private static class MyImmediateAndDeferredSupplier implements ImmediateSupplier<CallInfo>, DeferredSupplier<CallInfo> { - private final boolean failImmediately; + private final RuntimeException failImmediately; public MyImmediateAndDeferredSupplier() { - this(false); + this(null); } - public MyImmediateAndDeferredSupplier(boolean simulateImmediateUnsupported) { - this.failImmediately = simulateImmediateUnsupported; + public MyImmediateAndDeferredSupplier(RuntimeException failImmediately) { + this.failImmediately = failImmediately; } @Override public Maybe<CallInfo> getImmediately() { - if (failImmediately) { - throw new ImmediateSupplier.ImmediateUnsupportedException("Simulate immediate unsupported"); + if (failImmediately!=null) { + throw failImmediately; } else { return Maybe.of(CallInfo.newInstance()); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/3821e02c/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java ---------------------------------------------------------------------- diff --git a/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java b/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java index 2dfe627..b115acb 100644 --- a/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java +++ b/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java @@ -365,6 +365,15 @@ public abstract class Maybe<T> implements Serializable, Supplier<T> { return Maybe.absent(transform.apply((AnyExceptionSupplier<?>)supplier)); } + /** Like {@link #cast(Maybe)} but allows any casting because that is valid for absents. + * Enforces that the argument really is absent. */ + @SuppressWarnings("unchecked") + public static <T> Maybe<T> castAbsent(Maybe<?> absent) { + if (absent!=null && absent.isPresent()) { + throw new IllegalArgumentException("Expected an absent, but instead got: "+absent); + } + return (Maybe<T>)absent; + } } public static class AbsentNull<T> extends Absent<T> {
