Config self reference fix

api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java

@@ -24,6 +24,9 @@
 import java.util.concurrent.Executor;
 
 import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.util.guava.Maybe;
+
+import com.google.common.annotations.Beta;
 
 /**
  * This is a Brooklyn extension to the Java {@link Executor}.
@@ -64,4 +67,18 @@ public interface ExecutionContext extends Executor {
 
     boolean isShutdown();
 
+    /**
+     * Gets the value promptly, or returns {@link Maybe#absent()} if the value is not yet available.
+     * It may throw an error if it cannot be determined whether a value is available immediately or not.
+     * <p>
+     * Implementations will typically attempt to execute in the current thread, with appropriate
+     * tricks to make it look like it is in a sub-thread, and will attempt to be non-blocking but
+     * if needed they may block.
+     * <p>
+     * Supports {@link Callable} and {@link Runnable} and some {@link Task} targets to be evaluated with "immediate" semantics.
+     */
+    // TODO reference ImmediateSupplier when that class is moved to utils project
+    @Beta
+    <T> Maybe<T> getImmediately(Object callableOrSupplierOrTask);
+
 }

camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java

@@ -48,6 +48,7 @@
 import org.apache.brooklyn.util.core.task.DeferredSupplier;
 import org.apache.brooklyn.util.core.task.ImmediateSupplier;
 import org.apache.brooklyn.util.core.task.TaskBuilder;
+import org.apache.brooklyn.util.core.task.TaskTags;
 import org.apache.brooklyn.util.core.task.Tasks;
 import org.apache.brooklyn.util.exceptions.Exceptions;
 import org.apache.brooklyn.util.groovy.GroovyJavaMethods;
@@ -206,6 +207,15 @@ public Maybe<Entity> getImmediately() {
             }
         }
 
+        @Override
+        public Entity get() {
+            try {
+                return call();
+            } catch (Exception e) {
+                throw Exceptions.propagate(e);
+            }
+        }
+
         @Override
         public Entity call() throws Exception {
             return callImpl(false).get();
@@ -219,7 +229,7 @@ protected Maybe<Entity> getEntity(boolean immediate) {
                     return Maybe.of(scopeComponent.get());
                 }
             } else {
-                return Maybe.<Entity>of(entity());
+                return Maybe.<Entity>ofDisallowingNull(entity()).or(Maybe.<Entity>absent("Context entity not available when trying to evaluate Brooklyn DSL"));
             }
         }
 
@@ -311,10 +321,11 @@ protected Maybe<Entity> callImpl(boolean immediate) throws Exception {
                 return Maybe.of(result.get());
             }
 
-            // TODO may want to block and repeat on new entities joining?
-            throw new NoSuchElementException("No entity matching id " + desiredComponentId+
+            // could be nice if DSL has an extra .block() method to allow it to wait for a matching entity.
+            // previously we threw if nothing existed; now we return an absent with a detailed error
+            return Maybe.absent(new NoSuchElementException("No entity matching id " + desiredComponentId+
                 (scope==Scope.GLOBAL ? "" : ", in scope "+scope+" wrt "+entity+
-                (scopeComponent!=null ? " ("+scopeComponent+" from "+entity()+")" : "")));
+                (scopeComponent!=null ? " ("+scopeComponent+" from "+entity()+")" : ""))));
         }
 
         private ExecutionContext getExecutionContext() {
@@ -538,14 +549,11 @@ protected String resolveKeyName(boolean immediately) {
 
         @Override
         public final Maybe<Object> getImmediately() {
-            Maybe<Entity> targetEntityMaybe = component.getImmediately();
-            if (targetEntityMaybe.isAbsent()) return Maybe.absent("Target entity not available");
-            EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
-
-            String keyNameS = resolveKeyName(true);
-            ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyNameS);
-            Maybe<?> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyNameS));
-            return Maybe.<Object>cast(result);
+            Maybe<Object> maybeWrappedMaybe = findExecutionContext(this).getImmediately(newCallable(true));
+            // the answer will be wrapped twice due to the callable semantics;
+            // the inner present/absent is important; it will only get an outer absent if interrupted
+            if (maybeWrappedMaybe.isAbsent()) return maybeWrappedMaybe;
+            return Maybe.<Object>cast( (Maybe<?>) maybeWrappedMaybe.get() );
         }
 
         @Override
@@ -554,15 +562,55 @@ public Task<Object> newTask() {
                     .displayName("retrieving config for "+keyName)
                     .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
                     .dynamic(false)
-                    .body(new Callable<Object>() {
-                        @Override
-                        public Object call() throws Exception {
-                            Entity targetEntity = component.get();
-                            String keyNameS = resolveKeyName(true);
-                            ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyNameS);
-                            return targetEntity.getConfig(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyNameS));
-                        }})
-                    .build();
+                    .body(newCallable(false)).build();
+        }
+
+        private Callable<Object> newCallable(final boolean immediate) {
+            return new Callable<Object>() {
+                @Override
+                public Object call() throws Exception {
+                    Entity targetEntity;
+                    if (immediate) { 
+                        Maybe<Entity> targetEntityMaybe = component.getImmediately();
+                        if (targetEntityMaybe.isAbsent()) return Maybe.<Object>cast(targetEntityMaybe);
+                        targetEntity = (EntityInternal) targetEntityMaybe.get();
+                    } else {
+                        targetEntity = component.get();
+                    }
+
+                    // this is always run in a new dedicated task (possibly a fake task if immediate), so no need to clear
+                    checkAndTagForRecursiveReference(targetEntity);
+
+                    String keyNameS = resolveKeyName(true);
+                    ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyNameS);
+                    if (key==null) key = ConfigKeys.newConfigKey(Object.class, keyNameS);
+                    if (immediate) {
+                        return ((EntityInternal)targetEntity).config().getNonBlocking(key);
+                    } else {
+                        return targetEntity.getConfig(key);
+                    }
+                }
+            };
+        }
+
+        private void checkAndTagForRecursiveReference(Entity targetEntity) {
+            String tag = "DSL:entity('"+targetEntity.getId()+"').config('"+keyName+"')";
+            Task<?> ancestor = Tasks.current();
+            if (ancestor!=null) {
+                // don't check on ourself; only look higher in hierarchy;
+                // this assumes impls always spawn new tasks (which they do, just maybe not always in new threads)
+                // but it means it does not rely on tag removal to prevent weird errors, 
+                // and more importantly it makes the strategy idempotent
+                ancestor = ancestor.getSubmittedByTask();
+            }
+            while (ancestor!=null) {
+                if (TaskTags.hasTag(ancestor, tag)) {
+                    throw new IllegalStateException("Recursive config reference "+tag); 
+                }
+                ancestor = ancestor.getSubmittedByTask();
+            }
+
+            Tasks.addTagDynamically(tag);
         }
 
         @Override

camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/ConfigYamlTest.java

@@ -28,8 +28,14 @@
 
 import org.apache.brooklyn.api.entity.Entity;
 import org.apache.brooklyn.core.config.ConfigKeys;
+import org.apache.brooklyn.core.entity.Entities;
+import org.apache.brooklyn.core.mgmt.internal.LocalManagementContext;
 import org.apache.brooklyn.core.sensor.Sensors;
 import org.apache.brooklyn.core.test.entity.TestEntity;
+import org.apache.brooklyn.test.Asserts;
+import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException;
+import org.apache.brooklyn.util.time.Duration;
+import org.apache.brooklyn.util.time.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.AfterMethod;
@@ -44,7 +50,6 @@
 
 public class ConfigYamlTest extends AbstractYamlTest {
 
-    @SuppressWarnings("unused")
     private static final Logger LOG = LoggerFactory.getLogger(ConfigYamlTest.class);
 
     private ExecutorService executor;
@@ -91,6 +96,61 @@ public void testConfigInConfigBlock() throws Exception {
         assertNull(entity.getMyField()); // field with @SetFromFlag
         assertNull(entity.getMyField2()); // field with @SetFromFlag("myField2Alias"), set using alias
     }
+
+
+    @Test
+    public void testRecursiveConfigFailsGracefully() throws Exception {
+        doTestRecursiveConfigFailsGracefully(false);
+    }
+
+    @Test
+    public void testRecursiveConfigImmediateFailsGracefully() throws Exception {
+        doTestRecursiveConfigFailsGracefully(true);
+    }
+
+    protected void doTestRecursiveConfigFailsGracefully(boolean immediate) throws Exception {
+        String yaml = Joiner.on("\n").join(
+                "services:",
+                "- type: org.apache.brooklyn.core.test.entity.TestEntity",
+                "  brooklyn.config:",
+                "    infinite_loop: $brooklyn:config(\"infinite_loop\")");
+
+        final Entity app = createStartWaitAndLogApplication(yaml);
+        TestEntity entity = (TestEntity) Iterables.getOnlyElement(app.getChildren());
+
+        Thread t = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    Time.sleep(Duration.FIVE_SECONDS);
+                    // error, loop wasn't interrupted or detected
+                    LOG.warn("Timeout elapsed, destroying items; usage: "+
+                            ((LocalManagementContext)mgmt()).getGarbageCollector().getUsageString());
+                    Entities.destroy(app);
+                } catch (RuntimeInterruptedException e) {
+                    // expected on normal execution; clear the interrupted flag to prevent ugly further warnings being logged
+                    Thread.interrupted();
+                }
+            }
+        });
+        t.start();
+        try {
+            String c;
+            if (immediate) {
+                // this should throw rather than return "absent", because the error is definitive (absent means couldn't resolve in time)
+                c = entity.config().getNonBlocking(ConfigKeys.newStringConfigKey("infinite_loop")).or("FAILED");
+            } else {
+                c = entity.config().get(ConfigKeys.newStringConfigKey("infinite_loop"));
+            }
+            Asserts.shouldHaveFailedPreviously("Expected recursive error, instead got: "+c);
+        } catch (Exception e) {
+            Asserts.expectedFailureContainsIgnoreCase(e, "infinite_loop", "recursive");
+        } finally {
+            if (!Entities.isManaged(app)) {
+                t.interrupt();
+            }
+        }
+    }
 
     @Test
     public void testConfigAtTopLevel() throws Exception {

camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java

@@ -48,6 +48,7 @@
 import org.apache.brooklyn.util.guava.Maybe;
 import org.apache.brooklyn.util.text.Identifiers;
 import org.apache.brooklyn.util.time.Duration;
+import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -296,14 +297,13 @@ public void testUrlEncode() throws Exception {
     @Test
     public void testEntityNotFound() throws Exception {
         BrooklynDslDeferredSupplier<?> dsl = BrooklynDslCommon.entity("myIdDoesNotExist");
+        Maybe<?> actualValue = execDslImmediately(dsl, Entity.class, app, true);
+        Assert.assertTrue(actualValue.isAbsent());
         try {
-            Maybe<?> actualValue = execDslImmediately(dsl, Entity.class, app, true);
+            actualValue.get();
             Asserts.shouldHaveFailedPreviously("actual="+actualValue);
         } catch (Exception e) {
-            NoSuchElementException nsee = Exceptions.getFirstThrowableOfType(e, NoSuchElementException.class);
-            if (nsee == null) {
-                throw e;
-            }
+            Asserts.expectedFailureOfType(e, NoSuchElementException.class);
         }
     }
 
@@ -365,7 +365,7 @@ public DslTestWorker satisfiedAsynchronously(boolean val) {
             return this;
         }
 
-        @SuppressWarnings("unused")  // included for completeness?
+        @SuppressWarnings("unused")  // kept in case useful for additional tests, for completeness
         public DslTestWorker wrapInTaskForImmediately(boolean val) {
             wrapInTaskForImmediately = val;
             return this;

core/src/main/java/org/apache/brooklyn/core/config/internal/AbstractConfigMapImpl.java

@@ -30,6 +30,7 @@
 import javax.annotation.Nullable;
 
 import org.apache.brooklyn.api.mgmt.ExecutionContext;
+import org.apache.brooklyn.api.mgmt.TaskFactory;
 import org.apache.brooklyn.api.objs.BrooklynObject;
 import org.apache.brooklyn.config.ConfigInheritance;
 import org.apache.brooklyn.config.ConfigInheritances;
@@ -231,7 +232,7 @@ public Maybe<Object> getConfigRaw(ConfigKey<?> key, boolean includeInherited) {
     }
 
     protected Object coerceConfigVal(ConfigKey<?> key, Object v) {
-        if ((v instanceof Future) || (v instanceof DeferredSupplier)) {
+        if ((v instanceof Future) || (v instanceof DeferredSupplier) || (v instanceof TaskFactory)) {
             // no coercion for these (coerce on exit)
             return v;
         } else if (key instanceof StructuredConfigKey) {

core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java

@@ -78,6 +78,8 @@ public class BrooklynTaskTags extends TaskTags {
      * and that it need not appear in some task lists;
      * often used for framework lifecycle events and sensor polling */
     public static final String TRANSIENT_TASK_TAG = "TRANSIENT";
+    /** marks that a task is meant to return immediately, without blocking (or if absolutely necessary blocking for a short while) */
+    public static final String IMMEDIATE_TASK_TAG = "IMMEDIATE";
 
     // ------------- entity tags -------------------------
 

core/src/main/java/org/apache/brooklyn/core/objs/AbstractConfigurationSupportInternal.java

@@ -146,7 +146,6 @@ protected <T> Maybe<T> getNonBlockingResolvingSimple(ConfigKey<T> key) {
                 .immediately(true)
                 .deep(true)
                 .context(getContext())
-                .swallowExceptions()
                 .get();
         return (resolved != marker)
                 ? TypeCoercions.tryCoerce(resolved, key.getTypeToken())

core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java

@@ -41,6 +41,8 @@
 import org.apache.brooklyn.core.mgmt.BrooklynTaskTags.WrappedEntity;
 import org.apache.brooklyn.core.mgmt.entitlement.Entitlements;
 import org.apache.brooklyn.util.collections.MutableMap;
+import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateUnsupportedException;
+import org.apache.brooklyn.util.guava.Maybe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -96,7 +98,50 @@ public ExecutionManager getExecutionManager() {
     /** returns tasks started by this context (or tasks which have all the tags on this object) */
     @Override
     public Set<Task<?>> getTasks() { return executionManager.getTasksWithAllTags(tags); }
-
+
+    /** performs execution without spawning a new task thread, though it does temporarily set a fake task for the purpose of getting context;
+     * currently supports suppliers or callables  */
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T> Maybe<T> getImmediately(Object callableOrSupplier) {
+        BasicTask<?> fakeTaskForContext;
+        if (callableOrSupplier instanceof BasicTask) {
+            fakeTaskForContext = (BasicTask<?>)callableOrSupplier;
+            if (fakeTaskForContext.isQueuedOrSubmitted()) {
+                if (fakeTaskForContext.isDone()) {
+                    return Maybe.of((T)fakeTaskForContext.getUnchecked());
+                } else {
+                    throw new ImmediateUnsupportedException("Task is in progress and incomplete: "+fakeTaskForContext);
+                }
+            }
+            callableOrSupplier = fakeTaskForContext.getJob();
+        } else {
+            fakeTaskForContext = new BasicTask<Object>(MutableMap.of("displayName", "immediate evaluation"));
+        }
+        fakeTaskForContext.tags.addAll(tags);
+        fakeTaskForContext.tags.add(BrooklynTaskTags.IMMEDIATE_TASK_TAG);
+        fakeTaskForContext.tags.add(BrooklynTaskTags.TRANSIENT_TASK_TAG);
+
+        Task<?> previousTask = BasicExecutionManager.getPerThreadCurrentTask().get();
+        BasicExecutionContext oldExecutionContext = getCurrentExecutionContext();
+        registerPerThreadExecutionContext();
+
+        if (previousTask!=null) fakeTaskForContext.setSubmittedByTask(previousTask);
+        fakeTaskForContext.cancel();
+        try {
+            BasicExecutionManager.getPerThreadCurrentTask().set(fakeTaskForContext);
+
+            if (!(callableOrSupplier instanceof ImmediateSupplier)) {
+                callableOrSupplier = InterruptingImmediateSupplier.of(callableOrSupplier);
+            }
+            return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately();
+
+        } finally {
+            BasicExecutionManager.getPerThreadCurrentTask().set(previousTask);
+            perThreadExecutionContext.set(oldExecutionContext);
+        }
+    }
+
     @SuppressWarnings({ "unchecked", "rawtypes" })
     @Override
     protected <T> Task<T> submitInternal(Map<?,?> propertiesQ, final Object task) {

core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java

@@ -591,7 +591,7 @@ public boolean cancel(TaskCancellationMode mode) {
             if (!task.isCancelled()) result |= ((TaskInternal<T>)task).cancel(mode);
             result |= delegate().cancel(mode.isAllowedToInterruptTask());
 
-            if (mode.isAllowedToInterruptAllSubmittedTasks() || mode.isAllowedToInterruptDependentSubmittedTasks()) {
+            if (mode.isAllowedToInterruptDependentSubmittedTasks()) {
                 int subtasksFound=0;
                 int subtasksReallyCancelled=0;
 
@@ -753,7 +753,10 @@ protected void beforeStartAtomicTask(Map<?,?> flags, Task<?> task) {
     /** invoked in a task's thread when a task is starting to run (may be some time after submitted), 
      * but before doing any of the task's work, so that we can update bookkeeping and notify callbacks */
     protected void internalBeforeStart(Map<?,?> flags, Task<?> task) {
-        activeTaskCount.incrementAndGet();
+        int count = activeTaskCount.incrementAndGet();
+        if (count % 1000==0) {
+            log.warn("High number of active tasks: task #"+count+" is "+task);
+        }
 
         //set thread _before_ start time, so we won't get a null thread when there is a start-time
         if (log.isTraceEnabled()) log.trace(""+this+" beforeStart, task: "+task + " running on thread " + Thread.currentThread().getName());