You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by al...@apache.org on 2017/03/02 17:01:45 UTC
[01/19] brooklyn-server git commit: handle recursive task errors incl
self-ref config
Repository: brooklyn-server
Updated Branches:
refs/heads/master b3397aaf0 -> 3a6a2ee83
handle recursive task errors incl self-ref config
address BROOKLYN-329 case (2), where a config key is defined as a function of itself
detect the endless loop that results and fail with a good message
also better handling in general of endless-loop task failures,
including:
* bail-out logic in Exceptions.collapse to prevent crazy long strings and traces
* warning whenever active tasks passes N*1000
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/d7c60364
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/d7c60364
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/d7c60364
Branch: refs/heads/master
Commit: d7c60364e2e83fa6b9255504c515170e497eb2ef
Parents: f1d4c59
Author: Alex Heneveld <al...@Alexs-MacBook-Pro.local>
Authored: Tue Dec 6 15:50:18 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:48:09 2017 +0000
----------------------------------------------------------------------
.../brooklyn/spi/dsl/methods/DslComponent.java | 19 ++++++--
.../brooklyn/camp/brooklyn/ConfigYamlTest.java | 47 +++++++++++++++++++-
.../util/core/task/BasicExecutionManager.java | 5 ++-
.../brooklyn/util/core/task/ValueResolver.java | 4 +-
.../brooklyn/util/exceptions/Exceptions.java | 36 ++++++++++++++-
5 files changed, 104 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/d7c60364/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 8c0fe3b..98466be 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
@@ -47,6 +47,7 @@ import org.apache.brooklyn.util.core.task.BasicExecutionContext;
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;
@@ -502,8 +503,8 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
if (targetEntityMaybe.isAbsent()) return Maybe.absent("Target entity not available");
EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
- ConfigKey<Object> key = (ConfigKey<Object>) targetEntity.getEntityType().getConfigKey(keyName);
- Maybe<Object> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
+ ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
+ Maybe<?> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
return Maybe.<Object>cast(result);
}
@@ -517,7 +518,19 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
@Override
public Object call() throws Exception {
Entity targetEntity = component.get();
- ConfigKey<Object> key = (ConfigKey<Object>) targetEntity.getEntityType().getConfigKey(keyName);
+
+ String tag = "DSL:entity('"+targetEntity.getId()+"').config('"+keyName+"')";
+ Task<?> ancestor = Tasks.current();
+ while (ancestor!=null) {
+ if (TaskTags.hasTag(ancestor, tag)) {
+ throw new IllegalStateException("Recursive config reference "+tag);
+ }
+ ancestor = ancestor.getSubmittedByTask();
+ }
+
+ Tasks.addTagDynamically(tag);
+
+ ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
return targetEntity.getConfig(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
}})
.build();
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/d7c60364/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 07fc36a..013a6f7 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
@@ -28,8 +28,14 @@ import java.util.concurrent.Executors;
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 @@ import com.google.common.collect.Iterables;
public class ConfigYamlTest extends AbstractYamlTest {
- @SuppressWarnings("unused")
private static final Logger LOG = LoggerFactory.getLogger(ConfigYamlTest.class);
private ExecutorService executor;
@@ -91,6 +96,46 @@ public class ConfigYamlTest extends AbstractYamlTest {
assertNull(entity.getMyField()); // field with @SetFromFlag
assertNull(entity.getMyField2()); // field with @SetFromFlag("myField2Alias"), set using alias
}
+
+
+ @Test
+ public void testRecursiveConfigFailsGracefully() 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
+ Thread.interrupted();
+ }
+ }
+ });
+ t.start();
+ try {
+ String 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 {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/d7c60364/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
index 51f1b67..b6ad041 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
@@ -753,7 +753,10 @@ public class BasicExecutionManager implements ExecutionManager {
/** 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());
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/d7c60364/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 f81594e..8c61c45 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
@@ -493,7 +493,9 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
} catch (Exception e) {
Exceptions.propagateIfFatal(e);
- IllegalArgumentException problem = new IllegalArgumentException("Error resolving "+(description!=null ? description+", " : "")+v+", in "+exec+": "+e, e);
+ String msg = "Error resolving "+(description!=null ? description+", " : "")+v+", in "+exec;
+ String eTxt = Exceptions.collapseText(e);
+ IllegalArgumentException problem = eTxt.startsWith(msg) ? new IllegalArgumentException(e) : new IllegalArgumentException(msg+": "+eTxt, e);
if (swallowExceptions) {
if (log.isDebugEnabled())
log.debug("Resolution of "+this+" failed, swallowing and returning: "+e);
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/d7c60364/utils/common/src/main/java/org/apache/brooklyn/util/exceptions/Exceptions.java
----------------------------------------------------------------------
diff --git a/utils/common/src/main/java/org/apache/brooklyn/util/exceptions/Exceptions.java b/utils/common/src/main/java/org/apache/brooklyn/util/exceptions/Exceptions.java
index ee49b4a..3b41cf6 100644
--- a/utils/common/src/main/java/org/apache/brooklyn/util/exceptions/Exceptions.java
+++ b/utils/common/src/main/java/org/apache/brooklyn/util/exceptions/Exceptions.java
@@ -20,7 +20,6 @@ package org.apache.brooklyn.util.exceptions;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Predicates.instanceOf;
-import static com.google.common.base.Throwables.getCausalChain;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.UndeclaredThrowableException;
@@ -55,6 +54,8 @@ public class Exceptions {
private static final List<Class<? extends Throwable>> BORING_IF_NO_MESSAGE_THROWABLE_SUPERTYPES = ImmutableList.<Class<? extends Throwable>>of(
PropagatedRuntimeException.class);
+ public static final int MAX_COLLAPSE_RECURSIVE_DEPTH = 100;
+
/** NB: might be useful for stack trace, e.g. {@link ExecutionException} */
private static boolean isBoringForMessage(Throwable t) {
for (Class<? extends Throwable> type: ALWAYS_BORING_MESSAGE_THROWABLE_SUPERTYPES)
@@ -263,8 +264,41 @@ public class Exceptions {
public static Throwable collapse(Throwable source, boolean collapseCausalChain) {
return collapse(source, collapseCausalChain, false, ImmutableSet.<Throwable>of(), new Object[0]);
}
+
+ /** As {@link Throwables#getCausalChain(Throwable)} but safe in the face of perverse classes which return themselves as their cause or otherwise have a recursive causal chain. */
+ public static List<Throwable> getCausalChain(Throwable t) {
+ Set<Throwable> result = MutableSet.of();
+ while (t!=null) {
+ if (!result.add(t)) break;
+ t = t.getCause();
+ }
+ return ImmutableList.copyOf(result);
+ }
+
+ private static boolean isCausalChainDepthExceeding(Throwable t, int size) {
+ if (size<0) return true;
+ if (t==null) return false;
+ return isCausalChainDepthExceeding(t.getCause(), size-1);
+ }
private static Throwable collapse(Throwable source, boolean collapseCausalChain, boolean includeAllCausalMessages, Set<Throwable> visited, Object contexts[]) {
+ if (visited.isEmpty()) {
+ if (isCausalChainDepthExceeding(source, MAX_COLLAPSE_RECURSIVE_DEPTH)) {
+ // do fast check above, then do deeper check which survives recursive causes
+ List<Throwable> chain = getCausalChain(source);
+ if (chain.size() > MAX_COLLAPSE_RECURSIVE_DEPTH) {
+ // if it's an OOME or other huge stack, shrink it so we don't spin huge cycles processing the trace and printing it
+ // (sometimes generating subsequent OOME's in logback that mask the first!)
+ // coarse heuristic for how to reduce it, but that's better than killing cpu, causing further errors, and suppressing the root cause altogether!
+ String msg = chain.get(0).getMessage();
+ if (msg.length() > 512) msg = msg.substring(0, 500)+"...";
+ return new PropagatedRuntimeException("Huge stack trace (size "+chain.size()+", removing all but last few), "
+ + "starting: "+chain.get(0).getClass().getName()+": "+msg+"; ultimately caused by: ",
+ chain.get(chain.size() - 10));
+ }
+ }
+ }
+
visited = MutableSet.copyOf(visited);
String message = "";
Throwable collapsed = source;
[14/19] brooklyn-server git commit: more assertions for task
cancellation
Posted by al...@apache.org.
more assertions for task cancellation
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/99ccc0f6
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/99ccc0f6
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/99ccc0f6
Branch: refs/heads/master
Commit: 99ccc0f6c8703e924c5fff5197bf1a0c6e39bc81
Parents: cd3d486
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Mon Feb 20 12:38:48 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Mon Feb 20 12:38:48 2017 +0000
----------------------------------------------------------------------
.../brooklyn/core/entity/EntityConfigTest.java | 49 ++++++++++++++++++--
1 file changed, 45 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/99ccc0f6/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 4882b7c..2f40fe9 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
@@ -22,6 +22,7 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
@@ -46,10 +47,12 @@ import org.apache.brooklyn.core.test.BrooklynAppUnitTestSupport;
import org.apache.brooklyn.core.test.entity.TestEntity;
import org.apache.brooklyn.entity.stock.BasicEntity;
import org.apache.brooklyn.test.Asserts;
+import org.apache.brooklyn.util.collections.MutableList;
import org.apache.brooklyn.util.core.flags.SetFromFlag;
import org.apache.brooklyn.util.core.task.BasicTask;
import org.apache.brooklyn.util.core.task.DeferredSupplier;
import org.apache.brooklyn.util.core.task.InterruptingImmediateSupplier;
+import org.apache.brooklyn.util.core.task.TaskBuilder;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.slf4j.Logger;
@@ -276,6 +279,7 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
final Semaphore latch = new Semaphore(0);
final String expectedVal = "myval";
Object blockingVal;
+ List<Task<String>> tasksMadeByFactory = MutableList.of();
protected ConfigNonBlockingFixture usingTask() {
blockingVal = taskFactory().newTask();
@@ -298,15 +302,22 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
}
private TaskFactory<Task<String>> taskFactory() {
- return Tasks.<String>builder().body(
+ final TaskBuilder<String> tb = Tasks.<String>builder().body(
new Callable<String>() {
@Override
public String call() throws Exception {
if (!latch.tryAcquire()) latch.acquire();
latch.release();
return "myval";
- }})
- .buildFactory();
+ }});
+ return new TaskFactory<Task<String>>() {
+ @Override
+ public Task<String> newTask() {
+ Task<String> t = tb.build();
+ tasksMadeByFactory.add(t);
+ return t;
+ }
+ };
}
private DeferredSupplier<String> deferredSupplier() {
@@ -358,6 +369,14 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
// Will initially return absent, because task is not done
assertTrue(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING).isAbsent());
assertTrue(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")).isAbsent());
+
+ if (blockingVal instanceof TaskFactory) {
+ assertAllOurConfigTasksCancelled();
+ } else {
+ // TaskFactory tasks are cancelled, but others are not,
+ // things (ValueResolver?) are smart enough to know to leave it running
+ assertAllOurConfigTasksNotCancelled();
+ }
latch.release();
@@ -367,6 +386,28 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
assertEquals(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING).get(), ImmutableMap.of("mysub", expectedVal));
assertEquals(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")).get(), expectedVal);
+
+ assertAllTasksDone();
+ }
+
+ private void assertAllOurConfigTasksNotCancelled() {
+ for (Task<?> t: tasksMadeByFactory) {
+ Assert.assertFalse( t.isCancelled(), "Task should not have been cancelled: "+t+" - "+t.getStatusDetail(false) );
+ }
+ }
+
+ private void assertAllOurConfigTasksCancelled() {
+ // TODO added Feb 2017 - but might need an "eventually" here, if cancel is happening in a BG thread
+ // (but I think it is always foreground)
+ for (Task<?> t: tasksMadeByFactory) {
+ Assert.assertTrue( t.isCancelled(), "Task should have been cancelled: "+t+" - "+t.getStatusDetail(false) );
+ }
+ }
+
+ private void assertAllTasksDone() {
+ for (Task<?> t: tasksMadeByFactory) {
+ Assert.assertTrue( t.isDone(), "Task should have been done: "+t+" - "+t.getStatusDetail(false) );
+ }
}
}
@@ -381,7 +422,7 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
@Test(groups="Integration") // because takes 1s+
public void testGetTaskFactoryNonBlockingKey() throws Exception {
- new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey();
+ new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey();
}
@Test(groups="Integration") // because takes 1s+
public void testGetTaskFactoryNonBlockingMap() throws Exception {
[10/19] brooklyn-server git commit: add many of the code review
comments
Posted by al...@apache.org.
add many of the code review comments
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/7476d3b5
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/7476d3b5
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/7476d3b5
Branch: refs/heads/master
Commit: 7476d3b5f830d80fdd872ccfd70c8f0f7bf98015
Parents: 3f3e3d6
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Sat Feb 18 01:30:55 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Sat Feb 18 01:31:32 2017 +0000
----------------------------------------------------------------------
.../brooklyn/api/mgmt/ExecutionContext.java | 14 ++
.../brooklyn/camp/brooklyn/ConfigYamlTest.java | 5 +-
.../task/InterruptingImmediateSupplier.java | 2 +-
.../brooklyn/core/entity/EntityConfigTest.java | 24 ++--
.../task/InterruptingImmediateSupplierTest.java | 133 +++++++++++++++++++
.../util/exceptions/ExceptionsTest.java | 20 +++
6 files changed, 186 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/7476d3b5/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
index d8e538c..f8a963a 100644
--- a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
+++ b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
@@ -26,6 +26,8 @@ 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}.
*
@@ -65,6 +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} targets to be evaluated with "immediate" semantics.
+ */
+ // TODO reference ImmediateSupplier when that class is moved to utils project
+ @Beta
<T> Maybe<T> getImmediately(Object callableOrSupplier);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/7476d3b5/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 64700de..1686f55 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
@@ -103,7 +103,6 @@ public class ConfigYamlTest extends AbstractYamlTest {
doTestRecursiveConfigFailsGracefully(false);
}
- // TODO this test fails because entities aren't available when evaluating immediately
@Test
public void testRecursiveConfigImmediateFailsGracefully() throws Exception {
doTestRecursiveConfigFailsGracefully(true);
@@ -127,9 +126,9 @@ public class ConfigYamlTest extends AbstractYamlTest {
// error, loop wasn't interrupted or detected
LOG.warn("Timeout elapsed, destroying items; usage: "+
((LocalManagementContext)mgmt()).getGarbageCollector().getUsageString());
- //Entities.destroy(app);
+ Entities.destroy(app);
} catch (RuntimeInterruptedException e) {
- // expected on normal execution
+ // expected on normal execution; clear the interrupted flag to prevent ugly further warnings being logged
Thread.interrupted();
}
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/7476d3b5/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 ced001e..a92a641 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
@@ -44,7 +44,7 @@ import com.google.common.base.Supplier;
@Beta
public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, DeferredSupplier<T> {
- final Supplier<T> nestedSupplier;
+ private final Supplier<T> nestedSupplier;
public InterruptingImmediateSupplier(Supplier<T> nestedSupplier) {
this.nestedSupplier = nestedSupplier;
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/7476d3b5/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 672924f..4882b7c 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
@@ -372,31 +372,39 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
@Test(groups="Integration") // because takes 1s+
public void testGetTaskNonBlockingKey() throws Exception {
- new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInKey(); }
+ new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInKey();
+ }
@Test(groups="Integration") // because takes 1s+
public void testGetTaskNonBlockingMap() throws Exception {
- new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInMap(); }
+ new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInMap();
+ }
@Test(groups="Integration") // because takes 1s+
public void testGetTaskFactoryNonBlockingKey() throws Exception {
- new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey(); }
+ new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey();
+ }
@Test(groups="Integration") // because takes 1s+
public void testGetTaskFactoryNonBlockingMap() throws Exception {
- new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInMap(); }
+ new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInMap();
+ }
@Test(groups="Integration") // because takes 1s+
public void testGetSupplierNonBlockingKey() throws Exception {
- new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInKey(); }
+ new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInKey();
+ }
@Test(groups="Integration") // because takes 1s+
public void testGetSuppierNonBlockingMap() throws Exception {
- new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInMap(); }
+ new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInMap();
+ }
@Test // fast
public void testGetImmediateSupplierNonBlockingKey() throws Exception {
- new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInKey(); }
+ new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInKey();
+ }
@Test(groups="Integration") // because takes 1s+
public void testGetImmediateSupplierNonBlockingMap() throws Exception {
- new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInMap(); }
+ new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInMap();
+ }
@Test
public void testGetConfigKeysReturnsFromSuperAndInterfacesAndSubClass() throws Exception {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/7476d3b5/core/src/test/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplierTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplierTest.java b/core/src/test/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplierTest.java
new file mode 100644
index 0000000..fe83225
--- /dev/null
+++ b/core/src/test/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplierTest.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.brooklyn.util.core.task;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import java.util.concurrent.Callable;
+
+import org.apache.brooklyn.test.Asserts;
+import org.apache.brooklyn.util.guava.Maybe;
+import org.apache.brooklyn.util.time.Duration;
+import org.apache.brooklyn.util.time.Time;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
+import com.google.common.util.concurrent.Callables;
+import com.google.common.util.concurrent.Runnables;
+
+public class InterruptingImmediateSupplierTest {
+
+ @Test(expectedExceptions=UnsupportedOperationException.class)
+ public void testOfInvalidType() throws Exception {
+ InterruptingImmediateSupplier.of("myval");
+ }
+
+ @Test
+ public void testRunnable() throws Exception {
+ assertImmediatelyPresent(Runnables.doNothing(), null);
+ assertImmediatelyAbsent(new SleepingRunnable());
+ assertImmediatelyFails(new FailingRunnable(), MarkerException.class);
+ }
+
+ @Test
+ public void testCallable() throws Exception {
+ assertImmediatelyPresent(Callables.returning("myval"), "myval");
+ assertImmediatelyAbsent(new SleepingCallable());
+ assertImmediatelyFails(new FailingCallable(), MarkerException.class);
+ }
+
+ @Test
+ public void testSupplier() throws Exception {
+ assertImmediatelyPresent(Suppliers.ofInstance("myval"), "myval");
+ assertImmediatelyAbsent(new SleepingSupplier());
+ assertImmediatelyFails(new FailingSupplier(), MarkerException.class);
+ }
+
+ private void assertImmediatelyPresent(Object orig, Object expected) {
+ Maybe<Object> result = getImmediately(orig);
+ assertEquals(result.get(), expected);
+ assertFalse(Thread.currentThread().isInterrupted());
+ }
+
+ private void assertImmediatelyAbsent(Object orig) {
+ Maybe<Object> result = getImmediately(orig);
+ assertTrue(result.isAbsent(), "result="+result);
+ assertFalse(Thread.currentThread().isInterrupted());
+ }
+
+ private void assertImmediatelyFails(Object orig, Class<? extends Exception> expected) {
+ try {
+ Maybe<Object> result = getImmediately(orig);
+ Asserts.shouldHaveFailedPreviously("result="+result);
+ } catch (Exception e) {
+ Asserts.expectedFailureOfType(e, expected);
+ }
+ assertFalse(Thread.currentThread().isInterrupted());
+ }
+
+ private Maybe<Object> getImmediately(Object val) {
+ InterruptingImmediateSupplier<Object> supplier = InterruptingImmediateSupplier.of(val);
+ return supplier.getImmediately();
+ }
+
+ public static class SleepingRunnable implements Runnable {
+ @Override public void run() {
+ Time.sleep(Duration.ONE_MINUTE);
+ }
+ }
+
+ public static class SleepingCallable implements Callable<Void> {
+ @Override public Void call() {
+ Time.sleep(Duration.ONE_MINUTE);
+ return null;
+ }
+ }
+
+ public static class SleepingSupplier implements Supplier<Void> {
+ @Override public Void get() {
+ Time.sleep(Duration.ONE_MINUTE);
+ return null;
+ }
+ }
+
+ public static class FailingRunnable implements Runnable {
+ @Override public void run() {
+ throw new MarkerException();
+ }
+ }
+
+ public static class FailingCallable implements Callable<Void> {
+ @Override public Void call() {
+ throw new MarkerException();
+ }
+ }
+
+ public static class FailingSupplier implements Supplier<Void> {
+ @Override public Void get() {
+ throw new MarkerException();
+ }
+ }
+
+ public static class MarkerException extends RuntimeException {
+ private static final long serialVersionUID = -3395361406478634652L;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/7476d3b5/utils/common/src/test/java/org/apache/brooklyn/util/exceptions/ExceptionsTest.java
----------------------------------------------------------------------
diff --git a/utils/common/src/test/java/org/apache/brooklyn/util/exceptions/ExceptionsTest.java b/utils/common/src/test/java/org/apache/brooklyn/util/exceptions/ExceptionsTest.java
index 65b5d91..272d3f6 100644
--- a/utils/common/src/test/java/org/apache/brooklyn/util/exceptions/ExceptionsTest.java
+++ b/utils/common/src/test/java/org/apache/brooklyn/util/exceptions/ExceptionsTest.java
@@ -34,6 +34,7 @@ import org.testng.Assert;
import org.testng.annotations.Test;
import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableList;
public class ExceptionsTest {
@@ -307,6 +308,25 @@ public class ExceptionsTest {
}
@Test
+ public void testGetCausalChain() throws Exception {
+ Exception e1 = new Exception("e1");
+ Exception e2 = new Exception("e2", e1);
+ assertEquals(Exceptions.getCausalChain(e2), ImmutableList.of(e2, e1));
+ }
+
+ @Test
+ public void testGetCausalChainRecursive() throws Exception {
+ Exception e1 = new Exception("e1") {
+ private static final long serialVersionUID = 1L;
+ public synchronized Throwable getCause() {
+ return this;
+ }
+ };
+ Exception e2 = new Exception("e2", e1);
+ assertEquals(Exceptions.getCausalChain(e2), ImmutableList.of(e2, e1));
+ }
+
+ @Test
public void testComplexJcloudsExample() {
Throwable t;
t = new IOException("POST https://ec2.us-east-1.amazonaws.com/ HTTP/1.1 -> HTTP/1.1 401 Unauthorized");
[12/19] brooklyn-server git commit: use a task so tag is always
applied in a dedicated context so it doesn't need to be removed
Posted by al...@apache.org.
use a task so tag is always applied in a dedicated context so it doesn't need to be removed
the previous attempt at idempotency could set the tag too broadly, e.g. when evaluating K1,
then subsequently looking at a K2 that refers to K1, the latter would think it's recursed inside the former
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/0aa29efc
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/0aa29efc
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/0aa29efc
Branch: refs/heads/master
Commit: 0aa29efc6750575427628b025dd84dc8aa8020e6
Parents: b073349
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Sat Feb 18 14:12:12 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Sat Feb 18 14:12:12 2017 +0000
----------------------------------------------------------------------
.../brooklyn/spi/dsl/methods/DslComponent.java | 53 +++++++++++++-------
1 file changed, 34 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0aa29efc/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 bc93c05..1cce90e 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
@@ -549,14 +549,11 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
@Override
public final Maybe<Object> getImmediately() {
- Maybe<Entity> targetEntityMaybe = component.getImmediately();
- if (targetEntityMaybe.isAbsent()) return Maybe.<Object>cast(targetEntityMaybe);
- EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
- checkAndTagForRecursiveReference(targetEntity);
- 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
@@ -565,17 +562,35 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
.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();
- checkAndTagForRecursiveReference(targetEntity);
-
- 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) {
[15/19] brooklyn-server git commit: wrap immediate executions in an
(entity) execution context
Posted by al...@apache.org.
wrap immediate executions in an (entity) execution context
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/2e6f11fa
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/2e6f11fa
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/2e6f11fa
Branch: refs/heads/master
Commit: 2e6f11fae83170f5f5dab8f39bf1a416412f685d
Parents: 99ccc0f
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Mon Feb 20 12:41:07 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Mon Feb 20 12:41:07 2017 +0000
----------------------------------------------------------------------
.../apache/brooklyn/util/core/task/BasicExecutionContext.java | 4 ++++
1 file changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/2e6f11fa/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index 0799607..a3ea321 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -123,6 +123,9 @@ public class BasicExecutionContext extends AbstractExecutionContext {
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 {
@@ -135,6 +138,7 @@ public class BasicExecutionContext extends AbstractExecutionContext {
} finally {
BasicExecutionManager.getPerThreadCurrentTask().set(previousTask);
+ perThreadExecutionContext.set(oldExecutionContext);
}
}
[11/19] brooklyn-server git commit: tweak self-ref check strategy to
make idempotent
Posted by al...@apache.org.
tweak self-ref check strategy to make idempotent
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/b0733494
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/b0733494
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/b0733494
Branch: refs/heads/master
Commit: b0733494ce0ef307235ff055bfb0f4815736e876
Parents: 7476d3b
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Sat Feb 18 11:53:11 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Sat Feb 18 11:53:11 2017 +0000
----------------------------------------------------------------------
.../brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java | 8 +++++++-
1 file changed, 7 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/b0733494/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 7d9a951..bc93c05 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
@@ -553,7 +553,6 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
if (targetEntityMaybe.isAbsent()) return Maybe.<Object>cast(targetEntityMaybe);
EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
checkAndTagForRecursiveReference(targetEntity);
-
String keyNameS = resolveKeyName(true);
ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyNameS);
Maybe<?> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyNameS));
@@ -582,6 +581,13 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
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);
[07/19] brooklyn-server git commit: cleanup,
and allow TaskFactory to be supplied as a config and other
ValueResolver input
Posted by al...@apache.org.
cleanup, and allow TaskFactory to be supplied as a config and other ValueResolver input
the TF will create a task which will then be used for evaluation.
much cleaner semantics than setting tasks as values:
tasks evaluate once and remember their result, whereas task factory spawns a new task each time.
furthermore, the former cannot be interrupted without making the value _never_ resolvable (which was the case prior to the previous commit)
so it is left running if immediate eval is done, whereas the latter can be safely cancelled.
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/49f0e225
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/49f0e225
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/49f0e225
Branch: refs/heads/master
Commit: 49f0e225f8196c9d2314afe52cffbc1839cdfcf6
Parents: f84d886
Author: Alex Heneveld <al...@Alexs-MacBook-Pro.local>
Authored: Tue Dec 6 22:45:14 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:51:46 2017 +0000
----------------------------------------------------------------------
.../brooklyn/spi/dsl/methods/DslComponent.java | 16 ++++++++--
.../brooklyn/camp/brooklyn/spi/dsl/DslTest.java | 11 ++++---
.../config/internal/AbstractConfigMapImpl.java | 3 +-
.../util/core/task/BasicExecutionContext.java | 16 +++++-----
.../util/core/task/ImmediateSupplier.java | 8 +++--
.../task/InterruptingImmediateSupplier.java | 33 ++++++++++++++++++--
.../brooklyn/util/core/task/TaskTags.java | 1 +
.../brooklyn/util/core/task/ValueResolver.java | 24 +++++++++++---
.../brooklyn/core/entity/EntityConfigTest.java | 20 ++++++------
.../util/core/task/ValueResolverTest.java | 29 +++++++++++------
10 files changed, 114 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 80e202d..e745794 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
@@ -204,6 +204,15 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
}
@Override
+ public Entity get() {
+ try {
+ return call();
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
+ }
+ }
+
+ @Override
public Entity call() throws Exception {
return callImpl(false).get();
}
@@ -304,10 +313,11 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
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() {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
----------------------------------------------------------------------
diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
index 514a788..170b799 100644
--- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
+++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
@@ -48,6 +48,7 @@ import org.apache.brooklyn.util.exceptions.Exceptions;
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 class DslTest extends BrooklynAppUnitTestSupport {
@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,6 +365,7 @@ public class DslTest extends BrooklynAppUnitTestSupport {
return this;
}
+ @SuppressWarnings("unused") // kept in case useful for additional tests
public DslTestWorker wrapInTaskForImmediately(boolean val) {
wrapInTaskForImmediately = val;
return this;
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 2d92617..b736beb 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
@@ -30,6 +30,7 @@ import javax.annotation.Nonnull;
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 abstract class AbstractConfigMapImpl<TContainer extends BrooklynObject> i
}
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) {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index f23053b..6c69509 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -46,7 +46,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Function;
-import com.google.common.base.Supplier;
import com.google.common.collect.Iterables;
/**
@@ -99,7 +98,8 @@ public class BasicExecutionContext extends AbstractExecutionContext {
@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 */
+ /** 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) {
@@ -110,17 +110,15 @@ public class BasicExecutionContext extends AbstractExecutionContext {
Task<?> previousTask = BasicExecutionManager.getPerThreadCurrentTask().get();
if (previousTask!=null) fakeTaskForContext.setSubmittedByTask(previousTask);
+ fakeTaskForContext.cancel();
try {
BasicExecutionManager.getPerThreadCurrentTask().set(fakeTaskForContext);
- if ((callableOrSupplier instanceof Supplier) && !(callableOrSupplier instanceof ImmediateSupplier)) {
- callableOrSupplier = new InterruptingImmediateSupplier<>((Supplier<Object>)callableOrSupplier);
+ if (!(callableOrSupplier instanceof ImmediateSupplier)) {
+ callableOrSupplier = InterruptingImmediateSupplier.of(callableOrSupplier);
}
- if (callableOrSupplier instanceof ImmediateSupplier) {
- return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately();
- }
- // TODO could add more types here
- throw new IllegalArgumentException("Type "+callableOrSupplier.getClass()+" not supported for getImmediately (instance "+callableOrSupplier+")");
+ return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately();
+
} finally {
BasicExecutionManager.getPerThreadCurrentTask().set(previousTask);
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 ef9d648..5ec8d68 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
@@ -20,11 +20,13 @@ package org.apache.brooklyn.util.core.task;
import org.apache.brooklyn.util.guava.Maybe;
+import com.google.common.base.Supplier;
+
/**
- * A class that supplies objects of a single type, without blocking for any significant length
- * of time.
+ * A {@link Supplier} that has an extra method capable of supplying a value immediately or an absent if definitely not available,
+ * or throwing an {@link ImmediateUnsupportedException} if it cannot determine whether a value is immediately available.
*/
-public interface ImmediateSupplier<T> {
+public interface ImmediateSupplier<T> extends Supplier<T> {
/**
* Indicates that a supplier does not support immediate evaluation,
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 c478f5e..ced001e 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
@@ -18,13 +18,14 @@
*/
package org.apache.brooklyn.util.core.task;
-import java.util.NoSuchElementException;
+import java.util.concurrent.Callable;
import java.util.concurrent.Semaphore;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException;
import org.apache.brooklyn.util.guava.Maybe;
+import com.google.common.annotations.Beta;
import com.google.common.base.Supplier;
/**
@@ -40,6 +41,7 @@ import com.google.common.base.Supplier;
* will throw if the thread is interrupted. Typically there are workarounds, for instance:
* <code>if (semaphore.tryAcquire()) semaphore.acquire();</code>.
*/
+@Beta
public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, DeferredSupplier<T> {
final Supplier<T> nestedSupplier;
@@ -69,6 +71,33 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D
public T get() {
return nestedSupplier.get();
}
-
+
+ @SuppressWarnings("unchecked")
+ public static <T> InterruptingImmediateSupplier<T> of(final Object o) {
+ if (o instanceof Supplier) {
+ return new InterruptingImmediateSupplier<T>((Supplier<T>)o);
+ } else if (o instanceof Callable) {
+ return new InterruptingImmediateSupplier<T>(new Supplier<T>() {
+ @Override
+ public T get() {
+ try {
+ return ((Callable<T>)o).call();
+ } catch (Exception e) {
+ throw Exceptions.propagate(e);
+ }
+ }
+ });
+ } else if (o instanceof Runnable) {
+ return new InterruptingImmediateSupplier<T>(new Supplier<T>() {
+ @Override
+ public T get() {
+ ((Runnable)o).run();
+ return null;
+ }
+ });
+ } else {
+ throw new UnsupportedOperationException("Type "+o.getClass()+" not supported as InterruptingImmediateSupplier (instance "+o+")");
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/core/src/main/java/org/apache/brooklyn/util/core/task/TaskTags.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/TaskTags.java b/core/src/main/java/org/apache/brooklyn/util/core/task/TaskTags.java
index 6b64a6b..4319796 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/TaskTags.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/TaskTags.java
@@ -62,6 +62,7 @@ public class TaskTags {
}
public static boolean hasTag(Task<?> task, Object tag) {
+ if (task==null) return false;
return task.getTags().contains(tag);
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 672fef4..6644a9a 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
@@ -29,6 +29,7 @@ import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.mgmt.ExecutionContext;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.mgmt.TaskAdaptable;
+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;
@@ -322,6 +323,10 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
return result;
}
+ protected boolean isEvaluatingImmediately() {
+ return immediately || BrooklynTaskTags.hasTag(Tasks.current(), BrooklynTaskTags.IMMEDIATE_TASK_TAG);
+ }
+
@SuppressWarnings({ "unchecked", "rawtypes" })
protected Maybe<T> getMaybeInternal() {
if (started.getAndSet(true))
@@ -352,11 +357,11 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
//if the expected type is a closure or map and that's what we have, we're done (or if it's null);
//but not allowed to return a future or DeferredSupplier as the resolved value
- if (v==null || (!forceDeep && type.isInstance(v) && !Future.class.isInstance(v) && !DeferredSupplier.class.isInstance(v)))
+ if (v==null || (!forceDeep && type.isInstance(v) && !Future.class.isInstance(v) && !DeferredSupplier.class.isInstance(v) && !TaskFactory.class.isInstance(v)))
return Maybe.of((T) v);
try {
- if (immediately && v instanceof ImmediateSupplier) {
+ if (isEvaluatingImmediately() && v instanceof ImmediateSupplier) {
final ImmediateSupplier<Object> supplier = (ImmediateSupplier<Object>) v;
try {
Maybe<Object> result = exec.getImmediately(supplier);
@@ -366,12 +371,23 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
? recursive
? new ValueResolver(result.get(), type, this).getMaybe()
: result
- : Maybe.<T>absent();
+ : result;
} catch (ImmediateSupplier.ImmediateUnsupportedException e) {
log.debug("Unable to resolve-immediately for "+description+" ("+v+"); falling back to executing with timeout", e);
}
}
+ // TODO if evaluating immediately should use a new ExecutionContext.submitImmediate(...)
+ // and sets a timeout but which wraps a task but does not spawn a new thread
+
+ if ((v instanceof TaskFactory<?>) && !(v instanceof DeferredSupplier)) {
+ v = ((TaskFactory<?>)v).newTask();
+ BrooklynTaskTags.setTransient(((TaskAdaptable<?>)v).asTask());
+ if (isEvaluatingImmediately()) {
+ BrooklynTaskTags.addTagDynamically( ((TaskAdaptable<?>)v).asTask(), BrooklynTaskTags.IMMEDIATE_TASK_TAG );
+ }
+ }
+
//if it's a task or a future, we wait for the task to complete
if (v instanceof TaskAdaptable<?>) {
//if it's a task, we make sure it is submitted
@@ -382,7 +398,7 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
}
if (!task.getTags().contains(BrooklynTaskTags.TRANSIENT_TASK_TAG)) {
// mark this non-transient, because this value is usually something set e.g. in config
- // (ideally we'd discourage this in favour of task factories which can be transiently interrupted)
+ // (should discourage this in favour of task factories which can be transiently interrupted?)
BrooklynTaskTags.addTagDynamically(task, BrooklynTaskTags.NON_TRANSIENT_TASK_TAG);
}
exec.submit(task);
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 820fc14..672924f 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
@@ -313,12 +313,12 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
return new DeferredSupplier<String>() {
@Override public String get() {
try {
- log.info("acquiring");
+ log.trace("acquiring");
if (!latch.tryAcquire()) latch.acquire();
latch.release();
- log.info("acquired and released");
+ log.trace("acquired and released");
} catch (InterruptedException e) {
- log.info("interrupted");
+ log.trace("interrupted");
throw Exceptions.propagate(e);
}
return "myval";
@@ -333,21 +333,21 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
TestEntity entity = (TestEntity) mgmt.getEntityManager().createEntity(EntitySpec.create(TestEntity.class)
.configure((ConfigKey<Object>)(ConfigKey<?>)TestEntity.CONF_NAME, blockingVal));
- log.info("get non-blocking");
+ log.trace("get non-blocking");
// Will initially return absent, because task is not done
assertTrue(entity.config().getNonBlocking(TestEntity.CONF_NAME).isAbsent());
- log.info("got absent");
+ log.trace("got absent");
latch.release();
// Can now finish task, so will return expectedVal
- log.info("get blocking");
+ log.trace("get blocking");
assertEquals(entity.config().get(TestEntity.CONF_NAME), expectedVal);
- log.info("got blocking");
+ log.trace("got blocking");
assertEquals(entity.config().getNonBlocking(TestEntity.CONF_NAME).get(), expectedVal);
latch.acquire();
- log.info("finished");
+ log.trace("finished");
}
protected void runGetConfigNonBlockingInMap() throws Exception {
@@ -526,7 +526,7 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
assertEquals(getConfigFuture.get(TIMEOUT_MS, TimeUnit.MILLISECONDS), "abc");
}
- @Test
+ @Test(groups="Integration") // takes 0.5s
public void testGetConfigWithExecutedTaskWaitsForResult() throws Exception {
LatchingCallable<String> work = new LatchingCallable<String>("abc");
Task<String> task = executionManager.submit(work);
@@ -548,7 +548,7 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
assertEquals(work.callCount.get(), 1);
}
- @Test
+ @Test(groups="Integration") // takes 0.5s
public void testGetConfigWithUnexecutedTaskIsExecutedAndWaitsForResult() throws Exception {
LatchingCallable<String> work = new LatchingCallable<String>("abc");
Task<String> task = new BasicTask<String>(work);
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/49f0e225/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 e47e4c9..358f39d 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
@@ -20,7 +20,6 @@ package org.apache.brooklyn.util.core.task;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
import static org.testng.Assert.fail;
import java.util.Arrays;
@@ -137,19 +136,17 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport {
assertMaybeIsAbsent(result);
Assert.assertEquals(result.get(), "foo");
}
-
+
public void testGetImmediately() {
MyImmediateAndDeferredSupplier supplier = new MyImmediateAndDeferredSupplier();
CallInfo callInfo = Tasks.resolving(supplier).as(CallInfo.class).context(app).immediately(true).get();
- assertNull(callInfo.task);
- assertContainsCallingMethod(callInfo.stackTrace, "testGetImmediately");
+ assertImmediateFakeTaskFromMethod(callInfo, "testGetImmediately");
}
public void testImmediateSupplierWithTimeoutUsesBlocking() {
MyImmediateAndDeferredSupplier supplier = new MyImmediateAndDeferredSupplier();
CallInfo callInfo = Tasks.resolving(supplier).as(CallInfo.class).context(app).timeout(Asserts.DEFAULT_LONG_TIMEOUT).get();
- assertNotNull(callInfo.task);
- assertNotContainsCallingMethod(callInfo.stackTrace, "testImmediateSupplierWithTimeoutUsesBlocking");
+ assertRealTaskNotFromMethod(callInfo, "testImmediateSupplierWithTimeoutUsesBlocking");
}
public void testGetImmediatelyInTask() throws Exception {
@@ -164,16 +161,14 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport {
}
});
CallInfo callInfo = task.get();
- assertEquals(callInfo.task, task);
- assertContainsCallingMethod(callInfo.stackTrace, "myUniquelyNamedMethod");
+ assertImmediateFakeTaskFromMethod(callInfo, "myUniquelyNamedMethod");
}
public void testGetImmediatelyFallsBackToDeferredCallInTask() throws Exception {
final MyImmediateAndDeferredSupplier supplier = new MyImmediateAndDeferredSupplier(true);
CallInfo callInfo = Tasks.resolving(supplier).as(CallInfo.class).context(app).immediately(true).get();
- assertNotNull(callInfo.task);
+ assertRealTaskNotFromMethod(callInfo, "testGetImmediatelyFallsBackToDeferredCallInTask");
assertEquals(BrooklynTaskTags.getContextEntity(callInfo.task), app);
- assertNotContainsCallingMethod(callInfo.stackTrace, "testGetImmediatelyFallsBackToDeferredCallInTask");
}
public void testNonRecursiveBlockingFailsOnNonObjectType() throws Exception {
@@ -359,4 +354,18 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport {
}
}
}
+
+ private void assertImmediateFakeTaskFromMethod(CallInfo callInfo, String method) {
+ // previously task was null, but now there is a "fake task"
+ assertNotNull(callInfo.task);
+ Assert.assertFalse(callInfo.task.isSubmitted());
+ assertContainsCallingMethod(callInfo.stackTrace, method);
+ }
+
+ private void assertRealTaskNotFromMethod(CallInfo callInfo, String method) {
+ assertNotNull(callInfo.task);
+ Assert.assertTrue(callInfo.task.isSubmitted());
+ assertNotContainsCallingMethod(callInfo.stackTrace, method);
+ }
+
}
[17/19] brooklyn-server git commit: Merge branch 'master' into
config-self-reference-fix
Posted by al...@apache.org.
Merge branch 'master' into config-self-reference-fix
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/a3f42d3b
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/a3f42d3b
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/a3f42d3b
Branch: refs/heads/master
Commit: a3f42d3b88553aad401ff158ee473ea0120df3a3
Parents: 4602114 fa7c38b
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Mar 1 12:08:59 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Mar 1 12:08:59 2017 +0000
----------------------------------------------------------------------
LICENSE | 8 +-
.../brooklyn/api/catalog/CatalogItem.java | 6 +
.../brooklyn/api/mgmt/ManagementContext.java | 15 +
.../apache/brooklyn/api/mgmt/Scratchpad.java | 27 +
.../camp/brooklyn/BrooklynCampPlatform.java | 6 +-
.../brooklyn/spi/creation/CampResolver.java | 3 +
.../methods/FormatStringIntegrationTest.java | 76 ++
.../camp/brooklyn/ExternalConfigYamlTest.java | 2 +-
.../brooklyn/ReloadBrooklynPropertiesTest.java | 6 +-
.../catalog/CatalogMakeOsgiBundleTest.java | 201 ++++
.../catalog/GetFileContentsEffector.java | 46 +
.../lite/CampPlatformWithJustBrooklynMgmt.java | 2 +
.../osgi/catalog-bundle-1/META-INF/MANIFEST.MF | 3 +
.../catalog/osgi/catalog-bundle-1/catalog.bom | 31 +
core/pom.xml | 10 +
.../core/BrooklynFeatureEnablement.java | 2 +-
.../catalog/internal/BasicBrooklynCatalog.java | 2 +-
.../catalog/internal/CatalogBomScanner.java | 72 +-
.../core/entity/AbstractApplication.java | 51 +-
.../location/cloud/CloudLocationConfig.java | 23 +-
.../brooklyn/core/mgmt/BrooklynTaskTags.java | 2 +
.../JavaBrooklynClassLoadingContext.java | 6 +-
.../internal/AbstractManagementContext.java | 8 +
.../core/mgmt/internal/BasicScratchpad.java | 43 +
.../internal/DeferredBrooklynProperties.java | 2 +-
.../mgmt/internal/LocalSubscriptionManager.java | 147 +--
.../NonDeploymentManagementContext.java | 7 +
.../brooklyn/core/mgmt/usage/LocationUsage.java | 8 +-
.../brooklyn/util/core/osgi/BundleMaker.java | 353 +++++++
.../brooklyn/util/core/task/BasicTask.java | 4 +-
.../entity/ApplicationLifecycleStateTest.java | 16 +
.../LocalManagementContextRebindTest.java | 38 +
.../internal/LocalManagementContextTest.java | 20 +-
.../util/core/osgi/BundleMakerTest.java | 270 ++++++
.../osgi/test/bundlemaker/nomanifest/myfile.txt | 1 +
.../bundlemaker/nomanifest/subdir/myfile2.txt | 1 +
.../withmanifest/META-INF/MANIFEST.MF | 2 +
.../test/bundlemaker/withmanifest/myfile.txt | 1 +
.../bundlemaker/withmanifest/subdir/myfile2.txt | 1 +
.../brooklyn/launcher/osgi/OsgiLauncher.java | 4 +
.../location/jclouds/ConnectivityResolver.java | 55 ++
.../jclouds/ConnectivityResolverOptions.java | 245 +++++
.../jclouds/DefaultConnectivityResolver.java | 490 ++++++++++
.../location/jclouds/JcloudsLocation.java | 389 ++++----
.../location/jclouds/JcloudsLocationConfig.java | 4 +
.../jclouds/JcloudsSshMachineLocation.java | 2 +-
.../brooklyn/location/jclouds/JcloudsUtil.java | 1 -
.../jclouds/ManagementAddressResolveResult.java | 54 ++
.../jclouds/AbstractJcloudsStubbedLiveTest.java | 3 +-
...BasicLocationNetworkInfoInitializerTest.java | 44 +
.../DefaultConnectivityResolverTest.java | 262 ++++++
...dsByonLocationResolverStubbedRebindTest.java | 2 +
.../JcloudsReachableAddressStubbedTest.java | 69 +-
.../rest/domain/CatalogEntitySummary.java | 3 +-
.../rest/domain/CatalogItemSummary.java | 12 +-
.../rest/domain/CatalogLocationSummary.java | 3 +-
.../rest/domain/CatalogPolicySummary.java | 3 +-
.../rest/security/jaas/BrooklynLoginModule.java | 2 +
.../provider/DelegatingSecurityProvider.java | 6 +-
.../rest/transform/CatalogTransformer.java | 8 +-
.../util/json/BrooklynJacksonJsonProvider.java | 4 +-
.../filter/EntitlementContextFilterTest.java | 2 +-
server-cli/src/main/license/files/LICENSE | 13 +
.../src/main/license/source-inclusions.yaml | 3 +
.../main/resources/brooklyn/default.catalog.bom | 4 +
.../main/resources/brooklyn/icons/cluster.svg | 1 +
.../main/resources/brooklyn/icons/server.svg | 933 +++++++++++++++++++
.../MachineLifecycleEffectorTasks.java | 16 +-
.../brooklyn/entity/AbstractEc2LiveTest.java | 17 -
.../util/net/ReachableSocketFinder.java | 107 +--
.../util/net/ReachableSocketFinderTest.java | 42 +-
71 files changed, 3823 insertions(+), 502 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/a3f42d3b/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
----------------------------------------------------------------------
[13/19] brooklyn-server git commit: ensure TaskFactory items
evaluated immediately don't leak long-running tasks
Posted by al...@apache.org.
ensure TaskFactory items evaluated immediately don't leak long-running tasks
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/cd3d4864
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/cd3d4864
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/cd3d4864
Branch: refs/heads/master
Commit: cd3d4864aa2a59a18f28997313ca07bc9185fd62
Parents: 0aa29ef
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Sat Feb 18 16:31:26 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Sat Feb 18 16:31:26 2017 +0000
----------------------------------------------------------------------
.../brooklyn/api/mgmt/ExecutionContext.java | 4 +-
.../util/core/task/BasicExecutionContext.java | 16 +++-
.../task/InterruptingImmediateSupplier.java | 26 ++++--
.../brooklyn/util/core/task/ValueResolver.java | 88 ++++++++++++++------
.../util/core/task/ValueResolverTest.java | 51 ++++++++++++
5 files changed, 151 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/cd3d4864/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
index f8a963a..344907a 100644
--- a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
+++ b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
@@ -75,10 +75,10 @@ public interface ExecutionContext extends Executor {
* 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} targets to be evaluated with "immediate" semantics.
+ * 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 callableOrSupplier);
+ <T> Maybe<T> getImmediately(Object callableOrSupplierOrTask);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/cd3d4864/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index 6c69509..0799607 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -41,6 +41,7 @@ import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
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;
@@ -103,7 +104,20 @@ public class BasicExecutionContext extends AbstractExecutionContext {
@SuppressWarnings("unchecked")
@Override
public <T> Maybe<T> getImmediately(Object callableOrSupplier) {
- BasicTask<?> fakeTaskForContext = new BasicTask<Object>(MutableMap.of("displayName", "immediate evaluation"));
+ 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);
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/cd3d4864/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 a92a641..84b1bb4 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
@@ -22,6 +22,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.Semaphore;
import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.apache.brooklyn.util.exceptions.ReferenceWithError;
import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException;
import org.apache.brooklyn.util.guava.Maybe;
@@ -72,12 +73,16 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D
return nestedSupplier.get();
}
- @SuppressWarnings("unchecked")
public static <T> InterruptingImmediateSupplier<T> of(final Object o) {
+ return InterruptingImmediateSupplier.<T>ofSafe(o).get();
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> ReferenceWithError<InterruptingImmediateSupplier<T>> ofSafe(final Object o) {
if (o instanceof Supplier) {
- return new InterruptingImmediateSupplier<T>((Supplier<T>)o);
+ return ReferenceWithError.newInstanceWithoutError(new InterruptingImmediateSupplier<T>((Supplier<T>)o));
} else if (o instanceof Callable) {
- return new InterruptingImmediateSupplier<T>(new Supplier<T>() {
+ return ReferenceWithError.newInstanceWithoutError(new InterruptingImmediateSupplier<T>(new Supplier<T>() {
@Override
public T get() {
try {
@@ -86,18 +91,25 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D
throw Exceptions.propagate(e);
}
}
- });
+ }));
} else if (o instanceof Runnable) {
- return new InterruptingImmediateSupplier<T>(new Supplier<T>() {
+ return ReferenceWithError.newInstanceWithoutError(new InterruptingImmediateSupplier<T>(new Supplier<T>() {
@Override
public T get() {
((Runnable)o).run();
return null;
}
- });
+ }));
} else {
- throw new UnsupportedOperationException("Type "+o.getClass()+" not supported as InterruptingImmediateSupplier (instance "+o+")");
+ return ReferenceWithError.newInstanceThrowingError(null, new InterruptingImmediateSupplierNotSupportedForObject(o));
}
}
+ public static class InterruptingImmediateSupplierNotSupportedForObject extends UnsupportedOperationException {
+ private static final long serialVersionUID = 307517409005386500L;
+
+ public InterruptingImmediateSupplierNotSupportedForObject(Object o) {
+ super("Type "+o.getClass()+" not supported as InterruptingImmediateSupplier (instance "+o+")");
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/cd3d4864/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 6644a9a..f8cb91b 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
@@ -361,37 +361,59 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
return Maybe.of((T) v);
try {
- if (isEvaluatingImmediately() && v instanceof ImmediateSupplier) {
- final ImmediateSupplier<Object> supplier = (ImmediateSupplier<Object>) v;
- try {
- Maybe<Object> result = exec.getImmediately(supplier);
-
- // Recurse: need to ensure returned value is cast, etc
- return (result.isPresent())
- ? recursive
- ? new ValueResolver(result.get(), type, this).getMaybe()
- : result
- : result;
- } catch (ImmediateSupplier.ImmediateUnsupportedException e) {
- log.debug("Unable to resolve-immediately for "+description+" ("+v+"); falling back to executing with timeout", e);
+ boolean allowImmediateExecution = false;
+ boolean bailOutAfterImmediateExecution = false;
+
+ if (v instanceof ImmediateSupplier) {
+ allowImmediateExecution = true;
+
+ } else {
+ if ((v instanceof TaskFactory<?>) && !(v instanceof DeferredSupplier)) {
+ 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
+ if (v instanceof TaskAdaptable<?>) {
+ v = ((TaskAdaptable<?>) v).asTask();
}
}
- // TODO if evaluating immediately should use a new ExecutionContext.submitImmediate(...)
- // and sets a timeout but which wraps a task but does not spawn a new thread
-
- if ((v instanceof TaskFactory<?>) && !(v instanceof DeferredSupplier)) {
- v = ((TaskFactory<?>)v).newTask();
- BrooklynTaskTags.setTransient(((TaskAdaptable<?>)v).asTask());
- if (isEvaluatingImmediately()) {
- BrooklynTaskTags.addTagDynamically( ((TaskAdaptable<?>)v).asTask(), BrooklynTaskTags.IMMEDIATE_TASK_TAG );
+ 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
+ try {
+ Maybe<T> result = execImmediate(exec, v);
+ if (result!=null) return result;
+ 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");
}
}
- //if it's a task or a future, we wait for the task to complete
- if (v instanceof TaskAdaptable<?>) {
+ if (v instanceof Task) {
//if it's a task, we make sure it is submitted
- Task<?> task = ((TaskAdaptable<?>) v).asTask();
+ Task<?> task = (Task<?>) v;
if (!task.isSubmitted()) {
if (exec==null) {
return Maybe.absent("Value for unsubmitted task '"+getDescription()+"' requested but no execution context available");
@@ -537,6 +559,24 @@ 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/cd3d4864/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 358f39d..550d475 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
@@ -24,8 +24,11 @@ import static org.testng.Assert.fail;
import java.util.Arrays;
import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.api.mgmt.TaskAdaptable;
+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;
@@ -36,6 +39,8 @@ import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import com.google.common.util.concurrent.Callables;
+
/**
* see also {@link TasksTest} for more tests
*/
@@ -219,6 +224,52 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport {
assertEquals(result.getClass(), FailingImmediateAndDeferredSupplier.class);
}
+ public void testTaskFactoryGet() {
+ TaskFactory<TaskAdaptable<String>> taskFactory = new TaskFactory<TaskAdaptable<String>>() {
+ @Override public TaskAdaptable<String> newTask() {
+ return new BasicTask<>(Callables.returning("myval"));
+ }
+ };
+ String result = Tasks.resolving(taskFactory).as(String.class).context(app).get();
+ assertEquals(result, "myval");
+ }
+
+ public void testTaskFactoryGetImmediately() {
+ TaskFactory<TaskAdaptable<String>> taskFactory = new TaskFactory<TaskAdaptable<String>>() {
+ @Override public TaskAdaptable<String> newTask() {
+ return new BasicTask<>(Callables.returning("myval"));
+ }
+ };
+ String result = Tasks.resolving(taskFactory).as(String.class).context(app).immediately(true).get();
+ assertEquals(result, "myval");
+ }
+
+ public void testTaskFactoryGetImmediatelyDoesNotBlock() {
+ final AtomicBoolean executing = new AtomicBoolean();
+ TaskFactory<TaskAdaptable<String>> taskFactory = new TaskFactory<TaskAdaptable<String>>() {
+ @Override public TaskAdaptable<String> newTask() {
+ return new BasicTask<>(new Callable<String>() {
+ public String call() {
+ executing.set(true);
+ try {
+ Time.sleep(Duration.ONE_MINUTE);
+ return "myval";
+ } finally {
+ executing.set(false);
+ }
+ }});
+ }
+ };
+ Maybe<String> result = Tasks.resolving(taskFactory).as(String.class).context(app).immediately(true).getMaybe();
+ Asserts.assertTrue(result.isAbsent(), "result="+result);
+ // the call below default times out after 30s while the task above is still running
+ Asserts.succeedsEventually(new Runnable() {
+ public void run() {
+ Asserts.assertFalse(executing.get());
+ }
+ });
+ }
+
private static class MyImmediateAndDeferredSupplier implements ImmediateSupplier<CallInfo>, DeferredSupplier<CallInfo> {
private final boolean failImmediately;
[09/19] brooklyn-server git commit: Merge branch 'master' into
config-self-reference-fix
Posted by al...@apache.org.
Merge branch 'master' into config-self-reference-fix
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/3f3e3d67
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/3f3e3d67
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/3f3e3d67
Branch: refs/heads/master
Commit: 3f3e3d67936e3059a48a0659d1e82c54b95ff7d7
Parents: 72eff85 1104828
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Feb 15 18:58:30 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Feb 15 18:58:30 2017 +0000
----------------------------------------------------------------------
.../camp/brooklyn/AbstractYamlRebindTest.java | 26 +-
.../CompositeEffectorYamlRebindTest.java | 102 +++++++
.../brooklyn/CompositeEffectorYamlTest.java | 79 ++++++
.../HttpCommandEffectorYamlRebindTest.java | 92 +++++++
.../brooklyn/HttpCommandEffectorYamlTest.java | 79 ++++++
.../brooklyn/core/effector/AddSensor.java | 2 +
.../core/effector/CompositeEffector.java | 135 +++++++++
.../brooklyn/core/effector/Effectors.java | 4 +
.../core/effector/http/HttpCommandEffector.java | 209 ++++++++++++++
.../core/entity/EntityInitializers.java | 32 ++-
.../core/sensor/http/HttpRequestSensor.java | 47 ++--
.../CompositeEffectorIntegrationTest.java | 91 +++++++
.../core/effector/CompositeEffectorTest.java | 262 ++++++++++++++++++
.../HttpCommandEffectorIntegrationTest.java | 125 +++++++++
.../effector/http/HttpCommandEffectorTest.java | 272 +++++++++++++++++++
.../core/effector/http/int-response.json | 16 ++
.../core/effector/http/list-response.json | 19 ++
.../brooklyn/core/effector/http/login.json | 16 ++
.../core/effector/http/map-response.json | 16 ++
.../org/apache/brooklyn/core/effector/test.json | 16 ++
.../apache/brooklyn/rest/api/EffectorApi.java | 28 +-
.../brooklyn/util/time/DurationPredicates.java | 162 +++++++++++
.../util/time/DurationPredicatesTest.java | 150 ++++++++++
23 files changed, 1931 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
[04/19] brooklyn-server git commit: add (failing) test re config loop
and immediate evaluation
Posted by al...@apache.org.
add (failing) test re config loop and immediate evaluation
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/4121554d
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/4121554d
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/4121554d
Branch: refs/heads/master
Commit: 4121554d8664ee11ed989428e0e84992267da634
Parents: d7c6036
Author: Alex Heneveld <al...@Alexs-MacBook-Pro.local>
Authored: Tue Dec 6 16:09:20 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:48:10 2017 +0000
----------------------------------------------------------------------
.../brooklyn/camp/brooklyn/ConfigYamlTest.java | 18 +++++++++++++++++-
1 file changed, 17 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/4121554d/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 013a6f7..64700de 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
@@ -100,6 +100,16 @@ public class ConfigYamlTest extends AbstractYamlTest {
@Test
public void testRecursiveConfigFailsGracefully() throws Exception {
+ doTestRecursiveConfigFailsGracefully(false);
+ }
+
+ // TODO this test fails because entities aren't available when evaluating immediately
+ @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",
@@ -126,7 +136,13 @@ public class ConfigYamlTest extends AbstractYamlTest {
});
t.start();
try {
- String c = entity.config().get(ConfigKeys.newStringConfigKey("infinite_loop"));
+ 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");
[03/19] brooklyn-server git commit: better logging and reporting if
no entity available for immediate eval
Posted by al...@apache.org.
better logging and reporting if no entity available for immediate eval
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/5324f821
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/5324f821
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/5324f821
Branch: refs/heads/master
Commit: 5324f8212aac0aee1deac9b47155e37b1120c78a
Parents: 4121554
Author: Alex Heneveld <al...@Alexs-MacBook-Pro.local>
Authored: Tue Dec 6 16:26:41 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:48:10 2017 +0000
----------------------------------------------------------------------
.../brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/5324f821/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 98466be..1f704b3 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
@@ -216,7 +216,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
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"));
}
}
@@ -500,9 +500,8 @@ 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 Maybe.<Object>cast(targetEntityMaybe);
EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
-
ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
Maybe<?> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
return Maybe.<Object>cast(result);
[08/19] brooklyn-server git commit: Merge branch 'master' into
config-self-reference-fix
Posted by al...@apache.org.
Merge branch 'master' into config-self-reference-fix
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/72eff857
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/72eff857
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/72eff857
Branch: refs/heads/master
Commit: 72eff85763db0a6c6b6d854af3834f5e1940f3d8
Parents: 49f0e22 39301e0
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Feb 15 18:33:30 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Feb 15 18:33:30 2017 +0000
----------------------------------------------------------------------
.../spi/dsl/BrooklynDslInterpreter.java | 2 +-
.../spi/dsl/DslDeferredFunctionCall.java | 109 +++++++----
.../spi/dsl/methods/BrooklynDslCommon.java | 56 +++---
.../brooklyn/spi/dsl/methods/DslComponent.java | 144 +++++++++-----
.../spi/dsl/methods/DslToStringHelpers.java | 79 ++++++++
.../brooklyn/AddChildrenEffectorYamlTest.java | 190 +++++++++++++++++++
.../camp/brooklyn/ConfigParametersYamlTest.java | 177 +++++++++++++++++
.../camp/brooklyn/CreatePasswordSensorTest.java | 64 +++++++
.../CreatePasswordSensorIntegrationTest.java | 67 -------
.../spi/dsl/DslParseComponentsTest.java | 162 ++++++++++++++++
.../brooklyn/camp/brooklyn/spi/dsl/DslTest.java | 2 +-
.../camp/brooklyn/spi/dsl/DslYamlTest.java | 13 +-
.../brooklyn/test/lite/CampYamlLiteTest.java | 1 +
.../EmptySoftwareProcessWithPassword.yaml | 36 ----
.../example-with-CreatePasswordSensor.yaml | 36 ++++
.../core/effector/AddChildrenEffector.java | 40 +++-
.../brooklyn/core/objs/BasicSpecParameter.java | 69 +++++--
.../brooklyn/core/objs/BrooklynDynamicType.java | 2 +-
.../apache/brooklyn/rest/api/ActivityApi.java | 4 +-
.../rest/resources/ActivityResource.java | 41 +++-
20 files changed, 1057 insertions(+), 237 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/72eff857/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java
----------------------------------------------------------------------
diff --cc camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java
index e745794,9de3340..7d9a951
--- 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
@@@ -510,11 -539,12 +550,13 @@@ public class DslComponent extends Brook
@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 Maybe.<Object>cast(targetEntityMaybe);
EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
- ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
+ checkAndTagForRecursiveReference(targetEntity);
- Maybe<?> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
+
+ 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);
}
@@@ -528,24 -558,11 +570,26 @@@
@Override
public Object call() throws Exception {
Entity targetEntity = component.get();
+ checkAndTagForRecursiveReference(targetEntity);
- ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
- return targetEntity.getConfig(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
++
+ String keyNameS = resolveKeyName(true);
+ ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyNameS);
+ return targetEntity.getConfig(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyNameS));
- }})
- .build();
+ }
+ }).build();
+ }
+
+ private void checkAndTagForRecursiveReference(Entity targetEntity) {
+ String tag = "DSL:entity('"+targetEntity.getId()+"').config('"+keyName+"')";
+ Task<?> ancestor = Tasks.current();
+ while (ancestor!=null) {
+ if (TaskTags.hasTag(ancestor, tag)) {
+ throw new IllegalStateException("Recursive config reference "+tag);
+ }
+ ancestor = ancestor.getSubmittedByTask();
+ }
+
+ Tasks.addTagDynamically(tag);
}
@Override
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/72eff857/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
----------------------------------------------------------------------
diff --cc camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
index 170b799,d387920..63aba8e
--- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
+++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/DslTest.java
@@@ -365,7 -365,7 +365,7 @@@ public class DslTest extends BrooklynAp
return this;
}
- @SuppressWarnings("unused") // kept in case useful for additional tests
- @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;
[05/19] brooklyn-server git commit: immediate execution runs in a
fake tag allowing context to be evaluated
Posted by al...@apache.org.
immediate execution runs in a fake tag allowing context to be evaluated
most new immediate tests now passing, including a new test which detects recursive config values for immediate;
except we still have:
* cancellations of immediate execution goes too far, and cancels tasks which are set as values
* task factories still not supported for evaluation
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/faeeb1bd
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/faeeb1bd
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/faeeb1bd
Branch: refs/heads/master
Commit: faeeb1bdfc065b7e44554f45cca6a3f87074efb1
Parents: a679682
Author: Alex Heneveld <al...@Alexs-MacBook-Pro.local>
Authored: Tue Dec 6 22:15:10 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:49:00 2017 +0000
----------------------------------------------------------------------
.../brooklyn/api/mgmt/ExecutionContext.java | 3 ++
.../brooklyn/spi/dsl/methods/DslComponent.java | 31 +++++++++++---------
.../brooklyn/core/mgmt/BrooklynTaskTags.java | 2 ++
.../AbstractConfigurationSupportInternal.java | 1 -
.../util/core/task/BasicExecutionContext.java | 31 +++++++++++++++++++-
.../task/InterruptingImmediateSupplier.java | 3 +-
.../brooklyn/util/core/task/ValueResolver.java | 4 +--
7 files changed, 56 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
index 4540240..d8e538c 100644
--- a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
+++ b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java
@@ -24,6 +24,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.util.guava.Maybe;
/**
* This is a Brooklyn extension to the Java {@link Executor}.
@@ -64,4 +65,6 @@ public interface ExecutionContext extends Executor {
boolean isShutdown();
+ <T> Maybe<T> getImmediately(Object callableOrSupplier);
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/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 1f704b3..80e202d 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
@@ -503,6 +503,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
if (targetEntityMaybe.isAbsent()) return Maybe.<Object>cast(targetEntityMaybe);
EntityInternal targetEntity = (EntityInternal) targetEntityMaybe.get();
ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
+ checkAndTagForRecursiveReference(targetEntity);
Maybe<?> result = targetEntity.config().getNonBlocking(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
return Maybe.<Object>cast(result);
}
@@ -517,22 +518,24 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
@Override
public Object call() throws Exception {
Entity targetEntity = component.get();
-
- String tag = "DSL:entity('"+targetEntity.getId()+"').config('"+keyName+"')";
- Task<?> ancestor = Tasks.current();
- while (ancestor!=null) {
- if (TaskTags.hasTag(ancestor, tag)) {
- throw new IllegalStateException("Recursive config reference "+tag);
- }
- ancestor = ancestor.getSubmittedByTask();
- }
-
- Tasks.addTagDynamically(tag);
-
+ checkAndTagForRecursiveReference(targetEntity);
ConfigKey<?> key = targetEntity.getEntityType().getConfigKey(keyName);
return targetEntity.getConfig(key != null ? key : ConfigKeys.newConfigKey(Object.class, keyName));
- }})
- .build();
+ }
+ }).build();
+ }
+
+ private void checkAndTagForRecursiveReference(Entity targetEntity) {
+ String tag = "DSL:entity('"+targetEntity.getId()+"').config('"+keyName+"')";
+ Task<?> ancestor = Tasks.current();
+ while (ancestor!=null) {
+ if (TaskTags.hasTag(ancestor, tag)) {
+ throw new IllegalStateException("Recursive config reference "+tag);
+ }
+ ancestor = ancestor.getSubmittedByTask();
+ }
+
+ Tasks.addTagDynamically(tag);
}
@Override
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
index 39b2f70..8f4fa14 100644
--- a/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java
+++ b/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 -------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/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 ce10c86..796ab13 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
@@ -146,7 +146,6 @@ public abstract class AbstractConfigurationSupportInternal implements BrooklynOb
.immediately(true)
.deep(true)
.context(getContext())
- .swallowExceptions()
.get();
return (resolved != marker)
? TypeCoercions.tryCoerce(resolved, key.getTypeToken())
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index 38f1b5a..f23053b 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -41,10 +41,12 @@ import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
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.guava.Maybe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Function;
+import com.google.common.base.Supplier;
import com.google.common.collect.Iterables;
/**
@@ -96,7 +98,34 @@ public class BasicExecutionContext extends AbstractExecutionContext {
/** 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 */
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> Maybe<T> getImmediately(Object callableOrSupplier) {
+ BasicTask<?> 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();
+ if (previousTask!=null) fakeTaskForContext.setSubmittedByTask(previousTask);
+ try {
+ BasicExecutionManager.getPerThreadCurrentTask().set(fakeTaskForContext);
+
+ if ((callableOrSupplier instanceof Supplier) && !(callableOrSupplier instanceof ImmediateSupplier)) {
+ callableOrSupplier = new InterruptingImmediateSupplier<>((Supplier<Object>)callableOrSupplier);
+ }
+ if (callableOrSupplier instanceof ImmediateSupplier) {
+ return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately();
+ }
+ // TODO could add more types here
+ throw new IllegalArgumentException("Type "+callableOrSupplier.getClass()+" not supported for getImmediately (instance "+callableOrSupplier+")");
+ } finally {
+ BasicExecutionManager.getPerThreadCurrentTask().set(previousTask);
+ }
+ }
+
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
protected <T> Task<T> submitInternal(Map<?,?> propertiesQ, final Object task) {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/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 c29b458..c478f5e 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
@@ -18,6 +18,7 @@
*/
package org.apache.brooklyn.util.core.task;
+import java.util.NoSuchElementException;
import java.util.concurrent.Semaphore;
import org.apache.brooklyn.util.exceptions.Exceptions;
@@ -56,7 +57,7 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D
} catch (Throwable t) {
if (Exceptions.getFirstThrowableOfType(t, InterruptedException.class)!=null ||
Exceptions.getFirstThrowableOfType(t, RuntimeInterruptedException.class)!=null) {
- return Maybe.absent("Immediate value not available");
+ return Maybe.absent(new UnsupportedOperationException("Immediate value not available", t));
}
throw Exceptions.propagate(t);
} finally {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/faeeb1bd/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 8c61c45..ec7eb01 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
@@ -357,9 +357,9 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
try {
if (immediately && v instanceof ImmediateSupplier) {
- final ImmediateSupplier<?> supplier = (ImmediateSupplier<?>) v;
+ final ImmediateSupplier<Object> supplier = (ImmediateSupplier<Object>) v;
try {
- Maybe<?> result = supplier.getImmediately();
+ Maybe<Object> result = exec.getImmediately(supplier);
// Recurse: need to ensure returned value is cast, etc
return (result.isPresent())
[06/19] brooklyn-server git commit: solve problem with map eval where
tasks are cancelled permanently
Posted by al...@apache.org.
solve problem with map eval where tasks are cancelled permanently
sets non-transient if a task is requested for a value-resolver
we might want to deprecate that altogether, instead use TaskFactory
so we can cancel things
don't think there will be much leaking because the ValueResolver isn't used for new tasks,
just for tasks which are set as values -- but we need to keep an eye on that.
such tasks should be cancelled when the entities are cleaned up.
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/f84d8863
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/f84d8863
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/f84d8863
Branch: refs/heads/master
Commit: f84d886337de242450097fec06ec62ccaf1fe807
Parents: faeeb1b
Author: Alex Heneveld <al...@Alexs-MacBook-Pro.local>
Authored: Tue Dec 6 22:34:46 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:49:01 2017 +0000
----------------------------------------------------------------------
.../util/core/task/BasicExecutionManager.java | 2 +-
.../util/core/task/DynamicSequentialTask.java | 6 +++--
.../brooklyn/util/core/task/TaskInternal.java | 3 +++
.../brooklyn/util/core/task/ValueResolver.java | 13 +++++++---
.../brooklyn/core/entity/EntityConfigTest.java | 25 +++++++++++++-------
5 files changed, 34 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/f84d8863/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
index b6ad041..8b59498 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
@@ -591,7 +591,7 @@ public class BasicExecutionManager implements ExecutionManager {
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;
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/f84d8863/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
index 1b421b0..2869ff9 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
@@ -160,9 +160,11 @@ public class DynamicSequentialTask<T> extends BasicTask<T> implements HasTaskChi
@Override
protected boolean doCancel(TaskCancellationMode mode) {
boolean result = false;
- if (mode.isAllowedToInterruptDependentSubmittedTasks() || mode.isAllowedToInterruptAllSubmittedTasks()) {
- for (Task<?> t: secondaryJobsAll)
+ if (mode.isAllowedToInterruptDependentSubmittedTasks()) {
+ for (Task<?> t: secondaryJobsAll) {
+ // secondary jobs are dependent
result = ((TaskInternal<?>)t).cancel(mode) || result;
+ }
}
return super.doCancel(mode) || result;
// returns true if anything is successfully cancelled
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/f84d8863/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java b/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
index 99c2773..f565aa0 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
@@ -143,6 +143,9 @@ public interface TaskInternal<T> extends Task<T> {
this.allowedToInterruptTask = mayInterruptIfRunning;
this.allowedToInterruptDependentSubmittedTasks = interruptSubmittedTransients;
this.allowedToInterruptAllSubmittedTasks = interruptAllSubmitted;
+
+ // if dependent isn't set, then all shouldn't be set
+ assert !(this.allowedToInterruptAllSubmittedTasks && !this.allowedToInterruptDependentSubmittedTasks);
}
public boolean isAllowedToInterruptTask() { return allowedToInterruptTask; }
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/f84d8863/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 ec7eb01..672fef4 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
@@ -375,10 +375,17 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
//if it's a task or a future, we wait for the task to complete
if (v instanceof TaskAdaptable<?>) {
//if it's a task, we make sure it is submitted
- if (!((TaskAdaptable<?>) v).asTask().isSubmitted() ) {
- if (exec==null)
+ Task<?> task = ((TaskAdaptable<?>) v).asTask();
+ if (!task.isSubmitted()) {
+ if (exec==null) {
return Maybe.absent("Value for unsubmitted task '"+getDescription()+"' requested but no execution context available");
- exec.submit(((TaskAdaptable<?>) v).asTask());
+ }
+ if (!task.getTags().contains(BrooklynTaskTags.TRANSIENT_TASK_TAG)) {
+ // mark this non-transient, because this value is usually something set e.g. in config
+ // (ideally we'd discourage this in favour of task factories which can be transiently interrupted)
+ BrooklynTaskTags.addTagDynamically(task, BrooklynTaskTags.NON_TRANSIENT_TASK_TAG);
+ }
+ exec.submit(task);
}
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/f84d8863/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 6367071..820fc14 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
@@ -352,7 +352,6 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
protected void runGetConfigNonBlockingInMap() throws Exception {
Preconditions.checkNotNull(blockingVal, "Fixture must set blocking val before running this");
-
TestEntity entity = (TestEntity) mgmt.getEntityManager().createEntity(EntitySpec.create(TestEntity.class)
.configure(TestEntity.CONF_MAP_OBJ_THING, ImmutableMap.<String, Object>of("mysub", blockingVal)));
@@ -371,24 +370,32 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
}
}
- @Test public void testGetTaskNonBlockingKey() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetTaskNonBlockingKey() throws Exception {
new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInKey(); }
- @Test public void testGetTaskNonBlockingMap() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetTaskNonBlockingMap() throws Exception {
new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInMap(); }
- @Test public void testGetTaskFactoryNonBlockingKey() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetTaskFactoryNonBlockingKey() throws Exception {
new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey(); }
- @Test public void testGetTaskFactoryNonBlockingMap() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetTaskFactoryNonBlockingMap() throws Exception {
new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInMap(); }
- @Test public void testGetSupplierNonBlockingKey() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetSupplierNonBlockingKey() throws Exception {
new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInKey(); }
- @Test public void testGetSuppierNonBlockingMap() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetSuppierNonBlockingMap() throws Exception {
new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInMap(); }
- @Test public void testGetImmediateSupplierNonBlockingKey() throws Exception {
+ @Test // fast
+ public void testGetImmediateSupplierNonBlockingKey() throws Exception {
new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInKey(); }
- @Test public void testGetImmediateSupplierNonBlockingMap() throws Exception {
+ @Test(groups="Integration") // because takes 1s+
+ public void testGetImmediateSupplierNonBlockingMap() throws Exception {
new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInMap(); }
@Test
[18/19] brooklyn-server git commit: fix import (wrong Supplier)
Posted by al...@apache.org.
fix import (wrong Supplier)
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/e6fd10c3
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/e6fd10c3
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/e6fd10c3
Branch: refs/heads/master
Commit: e6fd10c35ee34910c9a1a0fd8748ade66b5d32f5
Parents: a3f42d3
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Thu Mar 2 12:15:39 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Thu Mar 2 12:15:39 2017 +0000
----------------------------------------------------------------------
.../org/apache/brooklyn/util/core/task/BasicExecutionContext.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/e6fd10c3/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index 2ff4dc8..f35a68a 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -28,7 +28,6 @@ import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
-import java.util.function.Supplier;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.mgmt.ExecutionContext;
@@ -48,6 +47,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Function;
+import com.google.common.base.Supplier;
import com.google.common.collect.Iterables;
/**
[19/19] brooklyn-server git commit: This closes #480
Posted by al...@apache.org.
This closes #480
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/3a6a2ee8
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/3a6a2ee8
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/3a6a2ee8
Branch: refs/heads/master
Commit: 3a6a2ee83caccf200fe847c6e9ef8627b31a9a0a
Parents: b3397aa e6fd10c
Author: Aled Sage <al...@gmail.com>
Authored: Thu Mar 2 17:01:08 2017 +0000
Committer: Aled Sage <al...@gmail.com>
Committed: Thu Mar 2 17:01:08 2017 +0000
----------------------------------------------------------------------
.../brooklyn/api/mgmt/ExecutionContext.java | 17 ++
.../brooklyn/spi/dsl/methods/DslComponent.java | 90 +++++--
.../brooklyn/camp/brooklyn/ConfigYamlTest.java | 62 ++++-
.../brooklyn/camp/brooklyn/spi/dsl/DslTest.java | 12 +-
.../config/internal/AbstractConfigMapImpl.java | 3 +-
.../brooklyn/core/mgmt/BrooklynTaskTags.java | 2 +
.../AbstractConfigurationSupportInternal.java | 1 -
.../util/core/task/BasicExecutionContext.java | 53 ++++-
.../util/core/task/BasicExecutionManager.java | 7 +-
.../brooklyn/util/core/task/CompoundTask.java | 15 ++
.../util/core/task/DynamicSequentialTask.java | 6 +-
.../util/core/task/ImmediateSupplier.java | 13 +-
.../task/InterruptingImmediateSupplier.java | 117 ++++++++++
.../brooklyn/util/core/task/TaskInternal.java | 3 +
.../brooklyn/util/core/task/TaskTags.java | 1 +
.../brooklyn/util/core/task/ValueResolver.java | 103 ++++++--
.../brooklyn/core/entity/EntityConfigTest.java | 232 +++++++++++++++----
.../task/InterruptingImmediateSupplierTest.java | 133 +++++++++++
.../util/core/task/ValueResolverTest.java | 126 +++++++++-
.../brooklyn/util/exceptions/Exceptions.java | 36 ++-
.../util/exceptions/ExceptionsTest.java | 20 ++
21 files changed, 942 insertions(+), 110 deletions(-)
----------------------------------------------------------------------
[16/19] brooklyn-server git commit: fixes and comments from code
review
Posted by al...@apache.org.
fixes and comments from code review
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/46021142
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/46021142
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/46021142
Branch: refs/heads/master
Commit: 46021142e7b063ca8b6da9690cdb27280e6e491c
Parents: 2e6f11f
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Tue Feb 28 17:13:15 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 28 17:13:15 2017 +0000
----------------------------------------------------------------------
.../brooklyn/spi/dsl/methods/DslComponent.java | 6 +--
.../util/core/task/BasicExecutionContext.java | 8 +++-
.../brooklyn/util/core/task/CompoundTask.java | 15 +++++++
.../task/InterruptingImmediateSupplier.java | 4 +-
.../brooklyn/util/core/task/ValueResolver.java | 24 +++++-----
.../util/core/task/ValueResolverTest.java | 46 ++++++++++++++++++++
6 files changed, 87 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/46021142/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..867c108 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
@@ -549,7 +549,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
@Override
public final Maybe<Object> getImmediately() {
- Maybe<Object> maybeWrappedMaybe = findExecutionContext(this).getImmediately(newCallable(true));
+ Maybe<Object> maybeWrappedMaybe = findExecutionContext(this).getImmediately(newCallableReturningImmediateMaybeOrNonImmediateValue(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;
@@ -562,10 +562,10 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements
.displayName("retrieving config for "+keyName)
.tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
.dynamic(false)
- .body(newCallable(false)).build();
+ .body(newCallableReturningImmediateMaybeOrNonImmediateValue(false)).build();
}
- private Callable<Object> newCallable(final boolean immediate) {
+ private Callable<Object> newCallableReturningImmediateMaybeOrNonImmediateValue(final boolean immediate) {
return new Callable<Object>() {
@Override
public Object call() throws Exception {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/46021142/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index a3ea321..2ff4dc8 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -28,6 +28,7 @@ import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.mgmt.ExecutionContext;
@@ -100,7 +101,12 @@ public class BasicExecutionContext extends AbstractExecutionContext {
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 */
+ * currently supports {@link Supplier}, {@link Callable}, {@link Runnable}, or {@link Task} instances;
+ * with tasks if it is submitted or in progress,
+ * it fails if not completed; with unsubmitted, unqueued tasks, it gets the {@link Callable} job and
+ * uses that; with such a job, or any other callable/supplier/runnable, it runs that
+ * in an {@link InterruptingImmediateSupplier}, with as much metadata as possible (eg task name if
+ * given a task) set <i>temporarily</i> in the current thread context */
@SuppressWarnings("unchecked")
@Override
public <T> Maybe<T> getImmediately(Object callableOrSupplier) {
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/46021142/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java
index 72dfb44..db17229 100644
--- a/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java
@@ -128,4 +128,19 @@ public abstract class CompoundTask<T> extends BasicTask<List<T>> implements HasT
return (List) getChildrenTyped();
}
+ @Override
+ protected boolean doCancel(org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode mode) {
+ boolean result = false;
+ if (mode.isAllowedToInterruptDependentSubmittedTasks()) {
+ for (Task<?> t: getChildren()) {
+ if (!t.isDone()) {
+ result = ((TaskInternal<?>)t).cancel(mode) || result;
+ }
+ }
+ }
+ result = super.doCancel(mode) || result;
+ return result;
+ // returns true if anything is successfully cancelled
+ }
+
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/46021142/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..afbc285 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
@@ -19,6 +19,7 @@
package org.apache.brooklyn.util.core.task;
import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.Semaphore;
import org.apache.brooklyn.util.exceptions.Exceptions;
@@ -59,7 +60,8 @@ public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, D
return Maybe.ofAllowingNull(get());
} catch (Throwable t) {
if (Exceptions.getFirstThrowableOfType(t, InterruptedException.class)!=null ||
- Exceptions.getFirstThrowableOfType(t, RuntimeInterruptedException.class)!=null) {
+ Exceptions.getFirstThrowableOfType(t, RuntimeInterruptedException.class)!=null ||
+ Exceptions.getFirstThrowableOfType(t, CancellationException.class)!=null) {
return Maybe.absent(new UnsupportedOperationException("Immediate value not available", t));
}
throw Exceptions.propagate(t);
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/46021142/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..3c6d96b 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
@@ -405,9 +405,14 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
if (bailOutAfterImmediateExecution) {
throw new ImmediateSupplier.ImmediateUnsupportedException("Cannot get immediately: "+v);
}
- } catch (InterruptingImmediateSupplier.InterruptingImmediateSupplierNotSupportedForObject o) {
+ } catch (ImmediateSupplier.ImmediateUnsupportedException e) {
+ if (bailOutAfterImmediateExecution) {
+ throw new ImmediateSupplier.ImmediateUnsupportedException("Cannot get immediately: "+v, e);
+ }
+ log.debug("Unable to resolve-immediately for "+description+" ("+v+", unsupported, type "+v.getClass()+"); falling back to executing with timeout: "+e);
+ } catch (InterruptingImmediateSupplier.InterruptingImmediateSupplierNotSupportedForObject e) {
// ignore, continue below
- log.debug("Unable to resolve-immediately for "+description+" ("+v+", wrong type "+v.getClass()+"); falling back to executing with timeout");
+ log.debug("Unable to resolve-immediately for "+description+" ("+v+", not supported for type "+v.getClass()+"); falling back to executing with timeout: "+e);
}
}
@@ -559,17 +564,14 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj
}
}
+ /** tries to get immediately, then resolve recursively (including for casting) if {@link #recursive} is set
+ *
+ * @throws InterruptingImmediateSupplier.InterruptingImmediateSupplierNotSupportedForObject
+ * ImmediateSupplier.ImmediateUnsupportedException
+ * if underlying call to {@link ExecutionContext#getImmediately(Object)} does so */
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
+ Maybe<T> result = exec.getImmediately(immediateSupplierOrImmediateTask);
- // Recurse: need to ensure returned value is cast, etc
return (result.isPresent())
? recursive
? new ValueResolver<T>(result.get(), type, this).getMaybe()
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/46021142/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..64cb024 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
@@ -23,8 +23,10 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.fail;
import java.util.Arrays;
+import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.mgmt.TaskAdaptable;
@@ -39,6 +41,8 @@ import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Callables;
/**
@@ -270,6 +274,48 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport {
});
}
+ public void testTaskFactoryGetImmediatelyDoesNotBlockWithNestedTasks() {
+ final int NUM_CALLS = 3;
+ final AtomicInteger executingCount = new AtomicInteger();
+ final List<SequentialTask<?>> outerTasks = Lists.newArrayList();
+
+ TaskFactory<Task<?>> taskFactory = new TaskFactory<Task<?>>() {
+ @Override public Task<?> newTask() {
+ SequentialTask<?> result = new SequentialTask<>(ImmutableList.of(new Callable<String>() {
+ public String call() {
+ executingCount.incrementAndGet();
+ try {
+ Time.sleep(Duration.ONE_MINUTE);
+ return "myval";
+ } finally {
+ executingCount.decrementAndGet();
+ }
+ }}));
+ outerTasks.add(result);
+ return result;
+ }
+ };
+ for (int i = 0; i < NUM_CALLS; i++) {
+ Maybe<String> result = Tasks.resolving(taskFactory).as(String.class).context(app).immediately(true).getMaybe();
+ Asserts.assertTrue(result.isAbsent(), "result="+result);
+ }
+ // the call below default times out after 30s while the task above is still running
+ Asserts.succeedsEventually(new Runnable() {
+ public void run() {
+ Asserts.assertEquals(outerTasks.size(), NUM_CALLS);
+ for (Task<?> task : outerTasks) {
+ Asserts.assertTrue(task.isDone());
+ Asserts.assertTrue(task.isCancelled());
+ }
+ }
+ });
+ Asserts.succeedsEventually(new Runnable() {
+ public void run() {
+ Asserts.assertEquals(executingCount.get(), 0);
+ }
+ });
+ }
+
private static class MyImmediateAndDeferredSupplier implements ImmediateSupplier<CallInfo>, DeferredSupplier<CallInfo> {
private final boolean failImmediately;
[02/19] brooklyn-server git commit: flesh out test cases around
non-blocking evaluation
Posted by al...@apache.org.
flesh out test cases around non-blocking evaluation
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/a6796829
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/a6796829
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/a6796829
Branch: refs/heads/master
Commit: a6796829e89f4239a5480622abd1d166485c766c
Parents: 5324f82
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Thu Nov 17 15:11:14 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Tue Feb 14 16:48:10 2017 +0000
----------------------------------------------------------------------
.../util/core/task/ImmediateSupplier.java | 5 +-
.../task/InterruptingImmediateSupplier.java | 73 ++++++++
.../brooklyn/core/entity/EntityConfigTest.java | 174 ++++++++++++++-----
3 files changed, 210 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/a6796829/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 ac0aae4..ef9d648 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
@@ -27,7 +27,8 @@ import org.apache.brooklyn.util.guava.Maybe;
public interface ImmediateSupplier<T> {
/**
- * Indicates that we are unable to get the value immediately, because that is not supported
+ * Indicates that a supplier does not support immediate evaluation,
+ * i.e. it may need to block to evaluate even if there is a value available
* (e.g. because the supplier is composed of sub-tasks that do not support {@link ImmediateSupplier}.
*/
public static class ImmediateUnsupportedException extends UnsupportedOperationException {
@@ -44,7 +45,7 @@ public interface ImmediateSupplier<T> {
/**
* Gets the value promptly, or returns {@link Maybe#absent()} if the value is not yet available.
*
- * @throws ImmediateUnsupportedException if cannot determinte the value immediately
+ * @throws ImmediateUnsupportedException if cannot determine whether a value is immediately available
*/
Maybe<T> getImmediately();
}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/a6796829/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
new file mode 100644
index 0000000..c29b458
--- /dev/null
+++ b/core/src/main/java/org/apache/brooklyn/util/core/task/InterruptingImmediateSupplier.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.brooklyn.util.core.task;
+
+import java.util.concurrent.Semaphore;
+
+import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException;
+import org.apache.brooklyn.util.guava.Maybe;
+
+import com.google.common.base.Supplier;
+
+/**
+ * Wraps a {@link Supplier} as an {@link ImmediateSupplier} by interrupting the thread before calling {@link Supplier#get()}.
+ * If the call succeeds, the result is returned.
+ * If the call throws any trace including an {@link InterruptedException} or {@link RuntimeInterruptedException}
+ * (ie the call failed due to the interruption, typically because it tried to wait)
+ * then this class concludes that there is no value available immediately and returns {@link Maybe#absent()}.
+ * If the call throws any other error, that is returned.
+ * The interruption is cleared afterwards (unless the thread was interrupted when the method was entered).
+ * <p>
+ * Note that some "immediate" methods, such as {@link Semaphore#acquire()} when a semaphore is available,
+ * will throw if the thread is interrupted. Typically there are workarounds, for instance:
+ * <code>if (semaphore.tryAcquire()) semaphore.acquire();</code>.
+ */
+public class InterruptingImmediateSupplier<T> implements ImmediateSupplier<T>, DeferredSupplier<T> {
+
+ final Supplier<T> nestedSupplier;
+
+ public InterruptingImmediateSupplier(Supplier<T> nestedSupplier) {
+ this.nestedSupplier = nestedSupplier;
+ }
+
+ @Override
+ public Maybe<T> getImmediately() {
+ boolean interrupted = Thread.currentThread().isInterrupted();
+ try {
+ if (!interrupted) Thread.currentThread().interrupt();
+ return Maybe.ofAllowingNull(get());
+ } catch (Throwable t) {
+ if (Exceptions.getFirstThrowableOfType(t, InterruptedException.class)!=null ||
+ Exceptions.getFirstThrowableOfType(t, RuntimeInterruptedException.class)!=null) {
+ return Maybe.absent("Immediate value not available");
+ }
+ throw Exceptions.propagate(t);
+ } finally {
+ if (!interrupted) Thread.interrupted();
+ }
+ }
+
+ @Override
+ public T get() {
+ return nestedSupplier.get();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/a6796829/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 38f5f90..6367071 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
@@ -28,6 +28,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -36,6 +37,7 @@ import org.apache.brooklyn.api.entity.EntitySpec;
import org.apache.brooklyn.api.entity.ImplementedBy;
import org.apache.brooklyn.api.mgmt.ExecutionManager;
import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.api.mgmt.TaskFactory;
import org.apache.brooklyn.config.ConfigKey;
import org.apache.brooklyn.core.config.ConfigKeys;
import org.apache.brooklyn.core.config.ConfigPredicates;
@@ -47,13 +49,17 @@ import org.apache.brooklyn.test.Asserts;
import org.apache.brooklyn.util.core.flags.SetFromFlag;
import org.apache.brooklyn.util.core.task.BasicTask;
import org.apache.brooklyn.util.core.task.DeferredSupplier;
+import org.apache.brooklyn.util.core.task.InterruptingImmediateSupplier;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
@@ -63,6 +69,8 @@ import groovy.lang.Closure;
public class EntityConfigTest extends BrooklynAppUnitTestSupport {
+ private static final Logger log = LoggerFactory.getLogger(EntityConfigTest.class);
+
private static final int TIMEOUT_MS = 10*1000;
private ExecutorService executor;
@@ -244,59 +252,145 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport {
// of the previous "test.confMapThing.obj".
//
// Presumably an earlier call to task.get() timed out, causing it to cancel the task?
+ // Alex: yes, a task.cancel is performed for maps in
+ // AbstractEntity$BasicConfigurationSupport(AbstractConfigurationSupportInternal).getNonBlockingResolvingStructuredKey(ConfigKey<T>)
+
+ //
// I (Aled) question whether we want to support passing a task (rather than a
// DeferredSupplier or TaskFactory, for example). Our EntitySpec.configure is overloaded
// to take a Task, but that feels wrong!?
- @Test(groups="Broken")
- public void testGetTaskNonBlocking() throws Exception {
- final CountDownLatch latch = new CountDownLatch(1);
- Task<String> task = Tasks.<String>builder().body(
+ //
+ // If starting clean I (Alex) would agree, we should use TaskFactory. However the
+ // DependentConfiguration methods -- including the ubiquitous AttributeWhenReady --
+ // return Task instances so they should survive a getNonBlocking or get with a short timeout
+ // access, and if a value is subsequently available it should be returned
+ // (which this test asserts, but is currently failing). If TaskFactory is used the
+ // intended semantics are clear -- you create a new task on each access, and can interrupt it
+ // and discard it if needed. For a Task it's less clear: probably the semantics are that the
+ // first returned value is what the value is forevermore. Probably it should not be interrupted
+ // on a non-blocking / short-wait access, or possibly it should simply be re-run if a previous
+ // execution was interrupted (but take care if we have a simultaneous non-blocking and blocking
+ // access, if the first one interrupts the second one should still get a value).
+ // I tend to think ideally we should switch to using TaskFactory in DependentConfiguration.
+ class ConfigNonBlockingFixture {
+ final Semaphore latch = new Semaphore(0);
+ final String expectedVal = "myval";
+ Object blockingVal;
+
+ protected ConfigNonBlockingFixture usingTask() {
+ blockingVal = taskFactory().newTask();
+ return this;
+ }
+
+ protected ConfigNonBlockingFixture usingTaskFactory() {
+ blockingVal = taskFactory();
+ return this;
+ }
+
+ protected ConfigNonBlockingFixture usingDeferredSupplier() {
+ blockingVal = deferredSupplier();
+ return this;
+ }
+
+ protected ConfigNonBlockingFixture usingImmediateSupplier() {
+ blockingVal = new InterruptingImmediateSupplier<String>(deferredSupplier());
+ return this;
+ }
+
+ private TaskFactory<Task<String>> taskFactory() {
+ return Tasks.<String>builder().body(
new Callable<String>() {
@Override
public String call() throws Exception {
- latch.await();
+ if (!latch.tryAcquire()) latch.acquire();
+ latch.release();
return "myval";
}})
- .build();
- runGetConfigNonBlocking(latch, task, "myval");
- }
-
- @Test
- public void testGetDeferredSupplierNonBlocking() throws Exception {
- final CountDownLatch latch = new CountDownLatch(1);
- DeferredSupplier<String> task = new DeferredSupplier<String>() {
- @Override public String get() {
- try {
- latch.await();
- } catch (InterruptedException e) {
- throw Exceptions.propagate(e);
- }
- return "myval";
- }
- };
- runGetConfigNonBlocking(latch, task, "myval");
- }
-
- @SuppressWarnings({"unchecked", "rawtypes"})
- protected void runGetConfigNonBlocking(CountDownLatch latch, Object blockingVal, String expectedVal) throws Exception {
- TestEntity entity = (TestEntity) mgmt.getEntityManager().createEntity(EntitySpec.create(TestEntity.class)
- .configure(TestEntity.CONF_MAP_OBJ_THING, ImmutableMap.<String, Object>of("mysub", blockingVal))
- .configure((ConfigKey)TestEntity.CONF_NAME, blockingVal));
-
- // Will initially return absent, because task is not done
- assertTrue(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING).isAbsent());
- assertTrue(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")).isAbsent());
+ .buildFactory();
+ }
- latch.countDown();
+ private DeferredSupplier<String> deferredSupplier() {
+ return new DeferredSupplier<String>() {
+ @Override public String get() {
+ try {
+ log.info("acquiring");
+ if (!latch.tryAcquire()) latch.acquire();
+ latch.release();
+ log.info("acquired and released");
+ } catch (InterruptedException e) {
+ log.info("interrupted");
+ throw Exceptions.propagate(e);
+ }
+ return "myval";
+ }
+ };
+ }
- // Can now finish task, so will return expectedVal
- assertEquals(entity.config().get(TestEntity.CONF_MAP_OBJ_THING), ImmutableMap.of("mysub", expectedVal));
- assertEquals(entity.config().get(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")), expectedVal);
+ protected void runGetConfigNonBlockingInKey() throws Exception {
+ Preconditions.checkNotNull(blockingVal, "Fixture must set blocking val before running this");
+
+ @SuppressWarnings("unchecked")
+ TestEntity entity = (TestEntity) mgmt.getEntityManager().createEntity(EntitySpec.create(TestEntity.class)
+ .configure((ConfigKey<Object>)(ConfigKey<?>)TestEntity.CONF_NAME, blockingVal));
+
+ log.info("get non-blocking");
+ // Will initially return absent, because task is not done
+ assertTrue(entity.config().getNonBlocking(TestEntity.CONF_NAME).isAbsent());
+ log.info("got absent");
+
+ latch.release();
+
+ // Can now finish task, so will return expectedVal
+ log.info("get blocking");
+ assertEquals(entity.config().get(TestEntity.CONF_NAME), expectedVal);
+ log.info("got blocking");
+ assertEquals(entity.config().getNonBlocking(TestEntity.CONF_NAME).get(), expectedVal);
+
+ latch.acquire();
+ log.info("finished");
+ }
- assertEquals(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING).get(), ImmutableMap.of("mysub", expectedVal));
- assertEquals(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")).get(), expectedVal);
+ protected void runGetConfigNonBlockingInMap() throws Exception {
+ Preconditions.checkNotNull(blockingVal, "Fixture must set blocking val before running this");
+
+ TestEntity entity = (TestEntity) mgmt.getEntityManager().createEntity(EntitySpec.create(TestEntity.class)
+ .configure(TestEntity.CONF_MAP_OBJ_THING, ImmutableMap.<String, Object>of("mysub", blockingVal)));
+
+ // Will initially return absent, because task is not done
+ assertTrue(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING).isAbsent());
+ assertTrue(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")).isAbsent());
+
+ latch.release();
+
+ // Can now finish task, so will return expectedVal
+ assertEquals(entity.config().get(TestEntity.CONF_MAP_OBJ_THING), ImmutableMap.of("mysub", expectedVal));
+ assertEquals(entity.config().get(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")), expectedVal);
+
+ assertEquals(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING).get(), ImmutableMap.of("mysub", expectedVal));
+ assertEquals(entity.config().getNonBlocking(TestEntity.CONF_MAP_OBJ_THING.subKey("mysub")).get(), expectedVal);
+ }
}
+ @Test public void testGetTaskNonBlockingKey() throws Exception {
+ new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInKey(); }
+ @Test public void testGetTaskNonBlockingMap() throws Exception {
+ new ConfigNonBlockingFixture().usingTask().runGetConfigNonBlockingInMap(); }
+
+ @Test public void testGetTaskFactoryNonBlockingKey() throws Exception {
+ new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInKey(); }
+ @Test public void testGetTaskFactoryNonBlockingMap() throws Exception {
+ new ConfigNonBlockingFixture().usingTaskFactory().runGetConfigNonBlockingInMap(); }
+
+ @Test public void testGetSupplierNonBlockingKey() throws Exception {
+ new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInKey(); }
+ @Test public void testGetSuppierNonBlockingMap() throws Exception {
+ new ConfigNonBlockingFixture().usingDeferredSupplier().runGetConfigNonBlockingInMap(); }
+
+ @Test public void testGetImmediateSupplierNonBlockingKey() throws Exception {
+ new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInKey(); }
+ @Test public void testGetImmediateSupplierNonBlockingMap() throws Exception {
+ new ConfigNonBlockingFixture().usingImmediateSupplier().runGetConfigNonBlockingInMap(); }
+
@Test
public void testGetConfigKeysReturnsFromSuperAndInterfacesAndSubClass() throws Exception {
MySubEntity entity = app.addChild(EntitySpec.create(MySubEntity.class));