You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by he...@apache.org on 2016/01/20 22:04:35 UTC
[1/6] incubator-brooklyn git commit: for chained references,
prefer outer catalog item id, and outer tags
Repository: incubator-brooklyn
Updated Branches:
refs/heads/master 27cd0920d -> 8303fea86
for chained references, prefer outer catalog item id, and outer tags
as described in changes to release-notes.md
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/1affd4d6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/1affd4d6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/1affd4d6
Branch: refs/heads/master
Commit: 1affd4d606d17136d3e334d08f401398f1ff3c96
Parents: 7d2ae06
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Jan 20 12:08:31 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Jan 20 12:09:56 2016 +0000
----------------------------------------------------------------------
brooklyn-docs/guide/misc/release-notes.md | 4 +-
.../internal/AbstractBrooklynObjectSpec.java | 25 +++++++++++-
.../brooklyn/spi/creation/CampResolver.java | 2 +-
.../catalog/CatalogYamlTemplateTest.java | 40 ++++++++++++++++++++
.../core/mgmt/EntityManagementUtils.java | 18 ++++-----
.../typereg/AbstractTypePlanTransformer.java | 3 +-
.../core/typereg/BasicRegisteredType.java | 1 +
7 files changed, 80 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-docs/guide/misc/release-notes.md
----------------------------------------------------------------------
diff --git a/brooklyn-docs/guide/misc/release-notes.md b/brooklyn-docs/guide/misc/release-notes.md
index 03ba4a3..f65e77d 100644
--- a/brooklyn-docs/guide/misc/release-notes.md
+++ b/brooklyn-docs/guide/misc/release-notes.md
@@ -45,7 +45,9 @@ can be used to traverse the hierarchy. It also means that when a type in the re
and a caller references it, that location will now take priority over a location defined in a parent.
Additionally, any locations specified in YAML extending the registered type will now *replace* locations on the referenced type;
this means in many cases an explicit `locations: []` when extending a type will cause locations to be taken from the
-parent or application root in YAML.
+parent or application root in YAML. Related to this, tags from referencing specs now preceed tags in the referenced types,
+and the referencing catalog item ID also takes priority; this has no effect in most cases, but if you have a chain of
+referenced types blueprint plan source code and the catalog item ID are now set correctly.
For changes in prior versions, please refer to the release notes for
[0.8.0](/v/0.8.0-incubating/misc/release-notes.html).
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java b/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
index 560c499..c4701c6 100644
--- a/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
+++ b/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
@@ -83,7 +83,7 @@ public abstract class AbstractBrooklynObjectSpec<T,SpecT extends AbstractBrookly
@Override
public String toString() {
- return Objects.toStringHelper(this).add("type", getType()).toString();
+ return Objects.toStringHelper(this).add("type", getType()).toString()+"@"+Integer.toHexString(System.identityHashCode(this));
}
protected abstract void checkValidType(Class<? extends T> type);
@@ -97,6 +97,19 @@ public abstract class AbstractBrooklynObjectSpec<T,SpecT extends AbstractBrookly
catalogItemId = val;
return self();
}
+ // TODO in many places (callers to this method) we prefer a wrapper item ID;
+ // that is right, because the wrapper's defn will refer to the wrapped,
+ // but we might need also to collect the item ID's so that *all* can be searched.
+ // e.g. if R3 references R2 which references R1 any one of these might supply config keys
+ // referencing resources or types in their local bundles.
+ @Beta
+ public SpecT catalogItemIdIfNotNull(String val) {
+ if (val!=null) {
+ catalogItemId = val;
+ }
+ return self();
+ }
+
public SpecT tag(Object tag) {
tags.add(tag);
@@ -105,9 +118,19 @@ public abstract class AbstractBrooklynObjectSpec<T,SpecT extends AbstractBrookly
/** adds the given tags */
public SpecT tags(Iterable<Object> tagsToAdd) {
+ return tagsAdd(tagsToAdd);
+ }
+ /** adds the given tags */
+ public SpecT tagsAdd(Iterable<Object> tagsToAdd) {
Iterables.addAll(this.tags, tagsToAdd);
return self();
}
+ /** replaces the given tags */
+ public SpecT tagsReplace(Iterable<Object> tagsToReplace) {
+ this.tags.clear();
+ Iterables.addAll(this.tags, tagsToReplace);
+ return self();
+ }
// TODO which semantics are correct? replace has been the behaviour;
// add breaks tests and adds unwanted parameters,
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/creation/CampResolver.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/creation/CampResolver.java b/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/creation/CampResolver.java
index 7523343..5639945 100644
--- a/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/creation/CampResolver.java
+++ b/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/creation/CampResolver.java
@@ -114,7 +114,7 @@ class CampResolver {
throw new IllegalStateException("Creating spec from "+item+", got "+spec.getType()+" which is incompatible with expected "+expectedType);
}
- ((AbstractBrooklynObjectSpec<?, ?>)spec).catalogItemId(item.getId());
+ ((AbstractBrooklynObjectSpec<?, ?>)spec).catalogItemIdIfNotNull(item.getId());
if (Strings.isBlank( ((AbstractBrooklynObjectSpec<?, ?>)spec).getDisplayName() ))
((AbstractBrooklynObjectSpec<?, ?>)spec).displayName(item.getDisplayName());
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java b/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
index 165fd49..78c1bf7 100644
--- a/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
+++ b/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
@@ -40,6 +40,7 @@ import org.apache.brooklyn.entity.group.DynamicCluster;
import org.apache.brooklyn.entity.stock.BasicApplication;
import org.apache.brooklyn.test.Asserts;
import org.apache.brooklyn.test.support.TestResourceUnavailableException;
+import org.apache.brooklyn.util.core.config.ConfigBag;
import org.apache.brooklyn.util.osgi.OsgiTestResources;
import org.python.google.common.collect.Iterables;
import org.testng.Assert;
@@ -209,6 +210,45 @@ public class CatalogYamlTemplateTest extends AbstractYamlTest {
Assert.assertEquals(child.getCatalogItemId(), "t1:"+TEST_VERSION);
}
+ @Test
+ public void testMetadataOnSpecCreatedFromItemReferencingAnApp() throws Exception {
+ // this nested ref to an app caused nested plan contents also to be recorded,
+ // due to how tags are merged. the *first* one is the most important, however,
+ // and ordering of tags should guarantee that.
+ // similarly ensure we get the right outermost non-null catalog item id.
+ addCatalogItems(
+ "brooklyn.catalog:",
+ " version: '1'",
+ " items:",
+ " - id: app1",
+ " name: myApp1",
+ " item:",
+ " type: org.apache.brooklyn.entity.stock.BasicApplication",
+ " brooklyn.config: { foo: bar }",
+ " - id: app1r",
+ " item_type: template",
+ " item:",
+ " services:",
+ " - type: app1",
+ " brooklyn.config:",
+ " foo: boo"
+ );
+
+ EntitySpec<? extends Application> spec = EntityManagementUtils.createEntitySpecForApplication(mgmt(),
+ "services: [ { type: app1r } ]\n" +
+ "location: localhost");
+
+ List<NamedStringTag> yamls = BrooklynTags.findAll(BrooklynTags.YAML_SPEC_KIND, spec.getTags());
+ Assert.assertTrue(yamls.size() >= 1, "Expected at least 1 yaml tag; instead had: "+yamls);
+ String yaml = yamls.iterator().next().getContents();
+ Asserts.assertStringContains(yaml, "services:", "type: app1r", "localhost");
+
+ Assert.assertEquals(spec.getChildren().size(), 0);
+ Assert.assertEquals(spec.getType(), BasicApplication.class);
+ Assert.assertEquals(ConfigBag.newInstance(spec.getConfig()).getStringKey("foo"), "boo");
+ Assert.assertEquals(spec.getCatalogItemId(), "app1r:1");
+ }
+
private RegisteredType makeItem() {
TestResourceUnavailableException.throwIfResourceUnavailable(getClass(), OsgiStandaloneTest.BROOKLYN_TEST_OSGI_ENTITIES_PATH);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/mgmt/EntityManagementUtils.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/mgmt/EntityManagementUtils.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/mgmt/EntityManagementUtils.java
index e75e2de..2f37e7c 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/mgmt/EntityManagementUtils.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/mgmt/EntityManagementUtils.java
@@ -253,10 +253,9 @@ public class EntityManagementUtils {
if (!wrapperParent.getParameters().isEmpty())
wrappedChild.parametersReplace(wrapperParent.getParameters());
-
- if (wrappedChild.getCatalogItemId()==null) {
- wrappedChild.catalogItemId(wrapperParent.getCatalogItemId());
- }
+
+ // prefer the wrapper ID (change in 2016-01); see notes on the catalogItemIdIfNotNull method
+ wrappedChild.catalogItemIdIfNotNull(wrapperParent.getCatalogItemId());
// NB: this clobber's child config wherever they conflict; might prefer to deeply merge maps etc
// (or maybe even prevent the merge in these cases;
@@ -266,11 +265,12 @@ public class EntityManagementUtils {
wrappedChild.configure(wrapperParent.getFlags());
// copying tags to all entities may be something the caller wants to control,
- // e.g. if we're creating a list of entities which will be added,
- // ignoring the parent Application holder;
- // in that case each child's BrooklynTags.YAML_SPEC tag will show all entities;
- // but in the normal case where we're unwrapping one, it's probably right.
- wrappedChild.tags(wrapperParent.getTags());
+ // e.g. if we're adding multiple, the caller might not want to copy the parent
+ // (the BrooklynTags.YAML_SPEC tag will include the parents source including siblings),
+ // but OTOH they might because otherwise the parent's tags might get lost.
+ // also if we are unwrapping multiple registry references we will get the YAML_SPEC for each;
+ // putting the parent's tags first however causes the preferred (outer) one to be retrieved first.
+ wrappedChild.tagsReplace(MutableList.copyOf(wrapperParent.getTags()).appendAll(wrappedChild.getTags()));
}
public static EntitySpec<? extends Application> newWrapperApp() {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
index 8f671f2..cfa51e3 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
@@ -103,7 +103,8 @@ public abstract class AbstractTypePlanTransformer implements BrooklynTypePlanTra
@Override protected Object visitSpec() {
try {
AbstractBrooklynObjectSpec<?, ?> result = createSpec(type, context);
- result.catalogItemId(type.getId());
+ // see notes in CampResolver setting catalogItemId; we might need to keep both
+ result.catalogItemIdIfNotNull(type.getId());
return result;
} catch (Exception e) { throw Exceptions.propagate(e); }
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/1affd4d6/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/BasicRegisteredType.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/BasicRegisteredType.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/BasicRegisteredType.java
index 05f0773..e0e2305 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/BasicRegisteredType.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/BasicRegisteredType.java
@@ -64,6 +64,7 @@ public class BasicRegisteredType implements RegisteredType {
@Override
public String getId() {
+ if (symbolicName==null) return null;
return symbolicName + (version!=null ? ":"+version : "");
}
[6/6] incubator-brooklyn git commit: This closes #1160
Posted by he...@apache.org.
This closes #1160
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/8303fea8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/8303fea8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/8303fea8
Branch: refs/heads/master
Commit: 8303fea866a9d39c58dc7174f73543477b6e73bc
Parents: 2a164a8 54cba8a
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Jan 20 21:04:20 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Jan 20 21:04:20 2016 +0000
----------------------------------------------------------------------
brooklyn-docs/guide/misc/release-notes.md | 5 +
.../java/org/apache/brooklyn/api/mgmt/Task.java | 18 +++
.../spi/dsl/BrooklynDslDeferredSupplier.java | 2 +-
.../core/objs/proxy/InternalEntityFactory.java | 6 +-
.../core/sensor/DependentConfiguration.java | 3 +-
.../util/core/task/BasicExecutionContext.java | 12 +-
.../util/core/task/BasicExecutionManager.java | 122 ++++++++++++++++---
.../brooklyn/util/core/task/BasicTask.java | 33 +++--
.../util/core/task/DynamicSequentialTask.java | 41 +++++--
.../brooklyn/util/core/task/ExecutionUtils.java | 49 --------
.../core/task/ListenableForwardingFuture.java | 28 ++++-
.../brooklyn/util/core/task/ScheduledTask.java | 14 +--
.../brooklyn/util/core/task/TaskInternal.java | 39 ++++++
.../brooklyn/util/core/task/TaskPredicates.java | 16 +++
.../task/BasicTaskExecutionPerformanceTest.java | 3 -
.../core/task/DynamicSequentialTaskTest.java | 96 ++++++++++++++-
.../core/task/NonBasicTaskExecutionTest.java | 5 +
.../util/core/task/ScheduledExecutionTest.java | 2 +-
.../util/core/task/TaskPredicatesTest.java | 4 +-
.../rest/resources/EntityConfigResource.java | 34 ++++--
20 files changed, 411 insertions(+), 121 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/8303fea8/brooklyn-docs/guide/misc/release-notes.md
----------------------------------------------------------------------
[3/6] incubator-brooklyn git commit: address code review for #1162
Posted by he...@apache.org.
address code review for #1162
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/47e2aec2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/47e2aec2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/47e2aec2
Branch: refs/heads/master
Commit: 47e2aec26e10ab2abb24fbc14eeb8895e13c116f
Parents: 1affd4d
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Jan 20 20:53:04 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Jan 20 20:53:30 2016 +0000
----------------------------------------------------------------------
.../apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java | 2 +-
.../apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/47e2aec2/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java b/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
index c4701c6..789d282 100644
--- a/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
+++ b/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/internal/AbstractBrooklynObjectSpec.java
@@ -125,7 +125,7 @@ public abstract class AbstractBrooklynObjectSpec<T,SpecT extends AbstractBrookly
Iterables.addAll(this.tags, tagsToAdd);
return self();
}
- /** replaces the given tags */
+ /** replaces tags with the given */
public SpecT tagsReplace(Iterable<Object> tagsToReplace) {
this.tags.clear();
Iterables.addAll(this.tags, tagsToReplace);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/47e2aec2/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
index cfa51e3..5ba36b3 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/typereg/AbstractTypePlanTransformer.java
@@ -103,7 +103,7 @@ public abstract class AbstractTypePlanTransformer implements BrooklynTypePlanTra
@Override protected Object visitSpec() {
try {
AbstractBrooklynObjectSpec<?, ?> result = createSpec(type, context);
- // see notes in CampResolver setting catalogItemId; we might need to keep both
+ // see notes on catalogItemIdIfNotNull
result.catalogItemIdIfNotNull(type.getId());
return result;
} catch (Exception e) { throw Exceptions.propagate(e); }
[4/6] incubator-brooklyn git commit: code review for
task-cancellation mode
Posted by he...@apache.org.
code review for task-cancellation mode
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/54cba8a6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/54cba8a6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/54cba8a6
Branch: refs/heads/master
Commit: 54cba8a657f151f93eb3cb504ccbb94c6d1fc9c7
Parents: aed0786
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Jan 20 21:01:20 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Jan 20 21:01:38 2016 +0000
----------------------------------------------------------------------
.../apache/brooklyn/util/core/task/BasicExecutionManager.java | 3 ++-
.../main/java/org/apache/brooklyn/util/core/task/BasicTask.java | 1 +
.../brooklyn/util/core/task/DynamicSequentialTaskTest.java | 4 ++--
3 files changed, 5 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/54cba8a6/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
index 2010613..0aab7d5 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
@@ -576,6 +576,7 @@ public class BasicExecutionManager implements ExecutionManager {
}
}
}
+ // TODO this is inefficient; might want to keep an index on submitted-by
for (Task<?> t: execMgmt.getAllTasks()) {
if (task.equals(t.getSubmittedByTask())) {
if (mode.isAllowedToInterruptAllSubmittedTasks() || BrooklynTaskTags.isTransient(t)) {
@@ -740,7 +741,7 @@ public class BasicExecutionManager implements ExecutionManager {
try {
return ((Callable<?>)callable).call();
} catch (Throwable t) {
- throw Throwables.propagate(t);
+ throw Exceptions.propagate(t);
}
}
if (callable instanceof Runnable) { ((Runnable)callable).run(); return null; }
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/54cba8a6/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
index c727d10..7c29bba 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
@@ -294,6 +294,7 @@ public class BasicTask<T> implements TaskInternal<T> {
: TaskCancellationMode.DO_NOT_INTERRUPT);
}
+ @Override @Beta
public synchronized boolean cancel(TaskCancellationMode mode) {
if (isDone()) return false;
if (log.isTraceEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/54cba8a6/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
index 95f0952..364870a 100644
--- a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
+++ b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
@@ -267,14 +267,14 @@ public class DynamicSequentialTaskTest {
waitForMessages(Predicates.compose(MathPredicates.greaterThanOrEqual(2), CollectionFunctionals.sizeFunction()), TIMEOUT);
Asserts.assertEquals(MutableSet.copyOf(messages), MutableSet.of("1-wait", "2-wait"));
- Time.sleep(Duration.millis(400));
-
if (cancellationMode==null) {
((TaskInternal<?>)t).cancel();
} else if (cancellationMode instanceof Boolean) {
t.cancel((Boolean)cancellationMode);
} else if (cancellationMode instanceof TaskCancellationMode) {
((TaskInternal<?>)t).cancel((TaskCancellationMode)cancellationMode);
+ } else {
+ throw new IllegalStateException("Invalid cancellationMode: "+cancellationMode);
}
// the cancelled task always reports cancelled and done
[5/6] incubator-brooklyn git commit: This closes #1162
Posted by he...@apache.org.
This closes #1162
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/2a164a82
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/2a164a82
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/2a164a82
Branch: refs/heads/master
Commit: 2a164a82f388e659f2c66dd8402c47ab6dfc0f90
Parents: 27cd092 47e2aec
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Jan 20 21:04:15 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Jan 20 21:04:15 2016 +0000
----------------------------------------------------------------------
brooklyn-docs/guide/misc/release-notes.md | 4 +-
.../internal/AbstractBrooklynObjectSpec.java | 25 +++++++++++-
.../brooklyn/spi/creation/CampResolver.java | 2 +-
.../catalog/CatalogYamlTemplateTest.java | 40 ++++++++++++++++++++
.../core/mgmt/EntityManagementUtils.java | 18 ++++-----
.../typereg/AbstractTypePlanTransformer.java | 3 +-
.../core/typereg/BasicRegisteredType.java | 1 +
7 files changed, 80 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/2a164a82/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
----------------------------------------------------------------------
diff --cc brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
index a759fed,78c1bf7..d83711c
--- a/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
+++ b/brooklyn-server/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/catalog/CatalogYamlTemplateTest.java
@@@ -40,7 -40,9 +40,8 @@@ import org.apache.brooklyn.entity.group
import org.apache.brooklyn.entity.stock.BasicApplication;
import org.apache.brooklyn.test.Asserts;
import org.apache.brooklyn.test.support.TestResourceUnavailableException;
+ import org.apache.brooklyn.util.core.config.ConfigBag;
import org.apache.brooklyn.util.osgi.OsgiTestResources;
-import org.python.google.common.collect.Iterables;
import org.testng.Assert;
import org.testng.TestListenerAdapter;
import org.testng.TestNG;
[2/6] incubator-brooklyn git commit: clearer semantics for cancel,
and cancel dependent submitted tasks by default
Posted by he...@apache.org.
clearer semantics for cancel, and cancel dependent submitted tasks by default
prevents leaks where dependent tasks (eg resolveValue) are submitted in the background,
then the caller is cancelled; the interruption was not propagated. now by default it is,
to children tasks and to submitted transients, with options for other (weaker and stronger) cancellations.
see TaskInternal.cancel(TaskCancellationOptions), and new tests for cancelling children in DynamicSequentialTaskTest.
also remove deprecated loose-typing ExecutionUtils.invoke,
more TRACE logging for activities,
and wrap batch config in a task so more resolutions are nested
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/aed07863
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/aed07863
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/aed07863
Branch: refs/heads/master
Commit: aed078633aaf40a4a9da907469700808a245cc78
Parents: 2f3e465
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Tue Jan 19 12:48:15 2016 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Jan 20 16:19:53 2016 +0000
----------------------------------------------------------------------
brooklyn-docs/guide/misc/release-notes.md | 5 +
.../java/org/apache/brooklyn/api/mgmt/Task.java | 18 +++
.../spi/dsl/BrooklynDslDeferredSupplier.java | 2 +-
.../core/objs/proxy/InternalEntityFactory.java | 6 +-
.../core/sensor/DependentConfiguration.java | 3 +-
.../util/core/task/BasicExecutionContext.java | 12 +-
.../util/core/task/BasicExecutionManager.java | 121 ++++++++++++++++---
.../brooklyn/util/core/task/BasicTask.java | 32 +++--
.../util/core/task/DynamicSequentialTask.java | 41 +++++--
.../brooklyn/util/core/task/ExecutionUtils.java | 49 --------
.../core/task/ListenableForwardingFuture.java | 28 ++++-
.../brooklyn/util/core/task/ScheduledTask.java | 14 +--
.../brooklyn/util/core/task/TaskInternal.java | 39 ++++++
.../brooklyn/util/core/task/TaskPredicates.java | 16 +++
.../task/BasicTaskExecutionPerformanceTest.java | 3 -
.../core/task/DynamicSequentialTaskTest.java | 96 ++++++++++++++-
.../core/task/NonBasicTaskExecutionTest.java | 5 +
.../util/core/task/ScheduledExecutionTest.java | 2 +-
.../util/core/task/TaskPredicatesTest.java | 4 +-
.../rest/resources/EntityConfigResource.java | 34 ++++--
20 files changed, 409 insertions(+), 121 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-docs/guide/misc/release-notes.md
----------------------------------------------------------------------
diff --git a/brooklyn-docs/guide/misc/release-notes.md b/brooklyn-docs/guide/misc/release-notes.md
index 03ba4a3..721655d 100644
--- a/brooklyn-docs/guide/misc/release-notes.md
+++ b/brooklyn-docs/guide/misc/release-notes.md
@@ -49,3 +49,8 @@ parent or application root in YAML.
For changes in prior versions, please refer to the release notes for
[0.8.0](/v/0.8.0-incubating/misc/release-notes.html).
+
+3. Task cancellation is now propagated to dependent submitted tasks, including backgrounded tasks if they are transient.
+Previously when a task was cancelled the API did not guarantee semantics but the behaviour was to cancel sub-tasks only
+in very limited cases. Now the semantics are more precise and controllable, and more sub-tasks are cancelled.
+This can prevent some leaked waits on `attributeWhenReady`.
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java b/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java
index c8f1c00..42147c5 100644
--- a/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java
+++ b/brooklyn-server/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java
@@ -81,6 +81,24 @@ public interface Task<T> extends ListenableFuture<T>, TaskAdaptable<T> {
public boolean isError();
/**
+ * As {@link Future#isDone()}. In particular if cancelled, this will return true
+ * as soon as it is cancelled. The thread for this task may still be running,
+ * if the cancellation (often an interruption, but may be weaker) has not applied,
+ * and submitted threads may also be running depending on cancellation parameters.
+ * <p>
+ * {@link #get()} is guaranteed to return immediately, throwing in the case of cancellation
+ * prior to completion (and including the case above where a thread may still be running).
+ * <p>
+ * To check whether cancelled threads for this task have completed,
+ * inspect {@link #getEndTimeUtc()}, which is guaranteed to be set when threads complete
+ * if the thread is started (as determinable by whether {@link #getStartTimeUtc()} is set).
+ * (The threads of submitted/child tasks will usually be independent; to determine their
+ * completion requires inspecting the {@link ExecutionManager}.)
+ */
+ @Override
+ public boolean isDone();
+
+ /**
* Causes calling thread to block until the task is started.
*/
public void blockUntilStarted();
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java b/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
index a417e32..48a0283 100644
--- a/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
+++ b/brooklyn-server/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java
@@ -115,7 +115,7 @@ public abstract class BrooklynDslDeferredSupplier<T> implements DeferredSupplier
try {
if (log.isDebugEnabled())
- log.debug("Queuing task to resolve "+dsl);
+ log.debug("Queuing task to resolve "+dsl+", called by "+Tasks.current());
EntityInternal entity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current());
ExecutionContext exec =
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/objs/proxy/InternalEntityFactory.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/objs/proxy/InternalEntityFactory.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/objs/proxy/InternalEntityFactory.java
index 8914ca4..eb4ff10 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/objs/proxy/InternalEntityFactory.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/objs/proxy/InternalEntityFactory.java
@@ -307,12 +307,16 @@ public class InternalEntityFactory extends InternalFactory {
/* Marked transient so that the task is not needlessly kept around at the highest level.
* Note that the task is not normally visible in the GUI, because
- * (a) while it is running, the entity is parentless (and so not in the tree);
+ * (a) while it is running, the entity is often parentless (and so not in the tree);
* and (b) when it is completed it is GC'd, as it is transient.
* However task info is available via the API if you know its ID,
* and if better subtask querying is available it will be picked up as a background task
* of the parent entity creating this child entity
* (note however such subtasks are currently filtered based on parent entity so is excluded).
+ * <p>
+ * Some of these (initializers and enrichers) submit background scheduled tasks,
+ * which currently show up at the top level once the initializer task completes.
+ * TODO It would be nice if these schedule tasks were grouped in a bucket!
*/
((EntityInternal)entity).getExecutionContext().submit(Tasks.builder().dynamic(false).displayName("Entity initialization")
.tag(BrooklynTaskTags.tagForContextEntity(entity))
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java
index ac4bef5..0c622c3 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/core/sensor/DependentConfiguration.java
@@ -248,7 +248,7 @@ public class DependentConfiguration {
// return immediately if either the ready predicate or the abort conditions hold
if (ready(value)) return postProcess(value);
-
+
final List<Exception> abortionExceptions = Lists.newCopyOnWriteArrayList();
long start = System.currentTimeMillis();
@@ -790,6 +790,7 @@ public class DependentConfiguration {
.displayName("waiting on "+sensor.getName())
.description("Waiting on sensor "+sensor.getName()+" from "+source)
.tag("attributeWhenReady")
+ .tag(BrooklynTaskTags.TRANSIENT_TASK_TAG)
.body(new WaitInTaskForAttributeReady<T,V>(this))
.build();
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
index 13dda46..74e0ddd 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionContext.java
@@ -175,18 +175,18 @@ public class BasicExecutionContext extends AbstractExecutionContext {
}
final Object startCallback = properties.get("newTaskStartCallback");
- properties.put("newTaskStartCallback", new Function<Object,Void>() {
- public Void apply(Object it) {
+ properties.put("newTaskStartCallback", new Function<Task<?>,Void>() {
+ public Void apply(Task<?> it) {
registerPerThreadExecutionContext();
- if (startCallback!=null) ExecutionUtils.invoke(startCallback, it);
+ if (startCallback!=null) BasicExecutionManager.invokeCallback(startCallback, it);
return null;
}});
final Object endCallback = properties.get("newTaskEndCallback");
- properties.put("newTaskEndCallback", new Function<Object,Void>() {
- public Void apply(Object it) {
+ properties.put("newTaskEndCallback", new Function<Task<?>,Void>() {
+ public Void apply(Task<?> it) {
try {
- if (endCallback!=null) ExecutionUtils.invoke(endCallback, it);
+ if (endCallback!=null) BasicExecutionManager.invokeCallback(endCallback, it);
} finally {
clearPerThreadExecutionContext();
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
index d90b1a1..2010613 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicExecutionManager.java
@@ -19,6 +19,7 @@
package org.apache.brooklyn.util.core.task;
import static com.google.common.base.Preconditions.checkNotNull;
+import groovy.lang.Closure;
import java.util.Collection;
import java.util.Collections;
@@ -51,16 +52,22 @@ import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.mgmt.TaskAdaptable;
import org.apache.brooklyn.core.BrooklynFeatureEnablement;
import org.apache.brooklyn.core.config.Sanitizer;
+import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.util.collections.MutableList;
+import org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode;
import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.apache.brooklyn.util.exceptions.RuntimeInterruptedException;
import org.apache.brooklyn.util.text.Identifiers;
+import org.apache.brooklyn.util.text.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.annotations.Beta;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.CaseFormat;
+import com.google.common.base.Function;
import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ExecutionList;
@@ -368,7 +375,6 @@ public class BasicExecutionManager implements ExecutionManager {
return submitSubsequentScheduledTask(flags, task);
}
- @SuppressWarnings("unchecked")
protected Task<?> submitSubsequentScheduledTask(final Map<?,?> flags, final ScheduledTask task) {
if (!task.isDone()) {
task.internalFuture = delayedRunner.schedule(new ScheduledTaskCallable(task, flags),
@@ -508,9 +514,15 @@ public class BasicExecutionManager implements ExecutionManager {
*/
if (log.isDebugEnabled()) {
// debug only here, because most submitters will handle failures
- log.debug("Exception running task "+task+" (rethrowing): "+error.getMessage(), error);
- if (log.isTraceEnabled())
- log.trace("Trace for exception running task "+task+" (rethrowing): "+error.getMessage(), error);
+ if (error instanceof InterruptedException || error instanceof RuntimeInterruptedException) {
+ log.debug("Detected interruption on task "+task+" (rethrowing)" +
+ (Strings.isNonBlank(error.getMessage()) ? ": "+error.getMessage() : ""));
+ } else {
+ log.debug("Exception running task "+task+" (rethrowing): "+error);
+ }
+ if (log.isTraceEnabled()) {
+ log.trace("Trace for exception running task "+task+" (rethrowing): "+error, error);
+ }
}
throw Exceptions.propagate(error);
}
@@ -526,19 +538,63 @@ public class BasicExecutionManager implements ExecutionManager {
}
}
- private final static class ListenableForwardingFutureForTask<T> extends ListenableForwardingFuture<T> {
+ @SuppressWarnings("deprecation")
+ // TODO do we even need a listenable future here? possibly if someone wants to interrogate the future it might
+ // be interesting, so possibly it is useful that we implement ListenableFuture...
+ private final static class CancellingListenableForwardingFutureForTask<T> extends ListenableForwardingFuture<T> {
private final Task<T> task;
+ private BasicExecutionManager execMgmt;
- private ListenableForwardingFutureForTask(Future<T> delegate, ExecutionList list, Task<T> task) {
+ private CancellingListenableForwardingFutureForTask(BasicExecutionManager execMgmt, Future<T> delegate, ExecutionList list, Task<T> task) {
super(delegate, list);
+ this.execMgmt = execMgmt;
this.task = task;
}
@Override
- public boolean cancel(boolean mayInterruptIfRunning) {
+ public boolean cancel(TaskCancellationMode mode) {
boolean result = false;
- if (!task.isCancelled()) result |= task.cancel(mayInterruptIfRunning);
- result |= super.cancel(mayInterruptIfRunning);
+ if (log.isTraceEnabled()) {
+ log.trace("CLFFT cancelling "+task+" mode "+mode);
+ }
+ if (!task.isCancelled()) result |= ((TaskInternal<T>)task).cancel(mode);
+ result |= delegate().cancel(mode.isAllowedToInterruptTask());
+
+ if (mode.isAllowedToInterruptAllSubmittedTasks() || mode.isAllowedToInterruptDependentSubmittedTasks()) {
+ int subtasksFound=0;
+ int subtasksReallyCancelled=0;
+
+ if (task instanceof HasTaskChildren) {
+ for (Task<?> child: ((HasTaskChildren)task).getChildren()) {
+ if (log.isTraceEnabled()) {
+ log.trace("Cancelling "+child+" on recursive cancellation of "+task);
+ }
+ subtasksFound++;
+ if (((TaskInternal<?>)child).cancel(mode)) {
+ result = true;
+ subtasksReallyCancelled++;
+ }
+ }
+ }
+ for (Task<?> t: execMgmt.getAllTasks()) {
+ if (task.equals(t.getSubmittedByTask())) {
+ if (mode.isAllowedToInterruptAllSubmittedTasks() || BrooklynTaskTags.isTransient(t)) {
+ if (log.isTraceEnabled()) {
+ log.trace("Cancelling "+t+" on recursive cancellation of "+task);
+ }
+ subtasksFound++;
+ if (((TaskInternal<?>)t).cancel(mode)) {
+ result = true;
+ subtasksReallyCancelled++;
+ }
+ }
+ }
+ }
+ if (log.isTraceEnabled()) {
+ log.trace("On cancel of "+task+", applicable subtask count "+subtasksFound+", of which "+subtasksReallyCancelled+" were actively cancelled");
+ }
+ }
+
((TaskInternal<?>)task).runListeners();
return result;
}
@@ -571,9 +627,15 @@ public class BasicExecutionManager implements ExecutionManager {
@SuppressWarnings("unchecked")
protected <T> Task<T> submitNewTask(final Map<?,?> flags, final Task<T> task) {
- if (log.isTraceEnabled()) log.trace("Submitting task {} ({}), with flags {}, and tags {}, job {}",
+ if (log.isTraceEnabled()) {
+ log.trace("Submitting task {} ({}), with flags {}, and tags {}, job {}; caller {}",
new Object[] {task.getId(), task, Sanitizer.sanitize(flags), task.getTags(),
- (task instanceof TaskInternal ? ((TaskInternal<T>)task).getJob() : "<unavailable>")});
+ (task instanceof TaskInternal ? ((TaskInternal<T>)task).getJob() : "<unavailable>"),
+ Tasks.current() });
+ if (Tasks.current()==null && BrooklynTaskTags.isTransient(task)) {
+ log.trace("Stack trace for unparented submission of transient "+task, new Throwable("trace only (not an error)"));
+ }
+ }
if (task instanceof ScheduledTask)
return (Task<T>) submitNewScheduledTask(flags, (ScheduledTask)task);
@@ -604,15 +666,16 @@ public class BasicExecutionManager implements ExecutionManager {
} else {
future = runner.submit(job);
}
- // on completion, listeners get triggered above; here, below we ensure they get triggered on cancel
- // (and we make sure the same ExecutionList is used in the future as in the task)
- ListenableFuture<T> listenableFuture = new ListenableForwardingFutureForTask<T>(future, ((TaskInternal<T>)task).getListeners(), task);
- // doesn't matter whether the listener is added to the listenableFuture or the task,
- // except that for the task we can more easily wrap it so that it only logs debug if the executor is shutdown
- // (avoid a bunch of ugly warnings in tests which start and stop things a lot!)
- // [probably even nicer to run this in the same thread, it doesn't do much; but that is messier to implement]
+ // SubmissionCallable (above) invokes the listeners on completion;
+ // this future allows a caller to add custom listeners
+ // (it does not notify the listeners; that's our job);
+ // except on cancel we want to listen
+ ListenableFuture<T> listenableFuture = new CancellingListenableForwardingFutureForTask<T>(this, future, ((TaskInternal<T>)task).getListeners(), task);
+ // and we want to make sure *our* (manager) listeners are given suitable callback
((TaskInternal<T>)task).addListener(new SubmissionListenerToCallOtherListeners<T>(task), runner);
+ // NB: can the above mean multiple callbacks to TaskInternal#runListeners?
+ // finally expose the future to callers
((TaskInternal<T>)task).initInternalFuture(listenableFuture);
return task;
@@ -665,9 +728,27 @@ public class BasicExecutionManager implements ExecutionManager {
PerThreadCurrentTaskHolder.perThreadCurrentTask.set(task);
((TaskInternal<?>)task).setStartTimeUtc(System.currentTimeMillis());
}
- ExecutionUtils.invoke(flags.get("newTaskStartCallback"), task);
+ invokeCallback(flags.get("newTaskStartCallback"), task);
}
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ // not ideal, such loose typing on the callback -- should prefer Function<Task,Object>
+ // but at least it's package-private
+ static Object invokeCallback(Object callable, Task<?> task) {
+ if (callable instanceof Closure) return ((Closure<?>)callable).call(task);
+ if (callable instanceof Callable) {
+ try {
+ return ((Callable<?>)callable).call();
+ } catch (Throwable t) {
+ throw Throwables.propagate(t);
+ }
+ }
+ if (callable instanceof Runnable) { ((Runnable)callable).run(); return null; }
+ if (callable instanceof Function) { return ((Function)callable).apply(task); }
+ if (callable==null) return null;
+ throw new IllegalArgumentException("Cannot invoke unexpected callback object "+callable+" of type "+callable.getClass()+" on "+task);
+ }
+
/** normally (if not interrupted) called once for each call to {@link #beforeSubmitScheduledTaskAllIterations(Map, Task)} */
protected void afterEndScheduledTaskAllIterations(Map<?,?> flags, Task<?> task) {
internalAfterEnd(flags, task, false, true);
@@ -693,7 +774,7 @@ public class BasicExecutionManager implements ExecutionManager {
}
if (isEndingAllIterations) {
incompleteTaskIds.remove(task.getId());
- ExecutionUtils.invoke(flags.get("newTaskEndCallback"), task);
+ invokeCallback(flags.get("newTaskEndCallback"), task);
((TaskInternal<?>)task).setEndTimeUtc(System.currentTimeMillis());
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
index 0c26dd1..c727d10 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java
@@ -155,7 +155,7 @@ public class BasicTask<T> implements TaskInternal<T> {
(Strings.isNonEmpty(displayName) ?
displayName :
(job + (tags!=null && !tags.isEmpty() ? ";"+tags : "")) ) +
- ":"+getId()+"]";
+ "]@"+getId();
}
@Override
@@ -196,7 +196,7 @@ public class BasicTask<T> implements TaskInternal<T> {
protected Maybe<Task<?>> submittedByTask;
protected volatile Thread thread = null;
- private volatile boolean cancelled = false;
+ protected volatile boolean cancelled = false;
/** normally a {@link ListenableFuture}, except for scheduled tasks when it may be a {@link ScheduledFuture} */
protected volatile Future<T> internalFuture = null;
@@ -288,15 +288,33 @@ public class BasicTask<T> implements TaskInternal<T> {
}
@Override
- public synchronized boolean cancel(boolean mayInterruptIfRunning) {
+ public final synchronized boolean cancel(boolean mayInterruptIfRunning) {
+ // semantics changed in 2016-01, previously "true" was INTERRUPT_TASK_BUT_NOT_SUBMITTED_TASKS
+ return cancel(mayInterruptIfRunning ? TaskCancellationMode.INTERRUPT_TASK_AND_DEPENDENT_SUBMITTED_TASKS
+ : TaskCancellationMode.DO_NOT_INTERRUPT);
+ }
+
+ public synchronized boolean cancel(TaskCancellationMode mode) {
if (isDone()) return false;
- boolean cancel = true;
+ if (log.isTraceEnabled()) {
+ log.trace("BT cancelling "+this+" mode "+mode);
+ }
cancelled = true;
+ doCancel(mode);
+ notifyAll();
+ return true;
+ }
+
+ @SuppressWarnings("deprecation")
+ protected boolean doCancel(TaskCancellationMode mode) {
if (internalFuture!=null) {
- cancel = internalFuture.cancel(mayInterruptIfRunning);
+ if (internalFuture instanceof ListenableForwardingFuture) {
+ return ((ListenableForwardingFuture<?>)internalFuture).cancel(mode);
+ } else {
+ return internalFuture.cancel(mode.isAllowedToInterruptTask());
+ }
}
- notifyAll();
- return cancel;
+ return true;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
index b7985c8..51a4e34 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java
@@ -158,27 +158,44 @@ public class DynamicSequentialTask<T> extends BasicTask<T> implements HasTaskChi
}
@Override
- public boolean cancel(boolean mayInterruptIfRunning) {
- return cancel(mayInterruptIfRunning, mayInterruptIfRunning, true);
+ protected boolean doCancel(TaskCancellationMode mode) {
+ boolean result = false;
+ if (mode.isAllowedToInterruptDependentSubmittedTasks() || mode.isAllowedToInterruptAllSubmittedTasks()) {
+ for (Task<?> t: secondaryJobsAll)
+ result = ((TaskInternal<?>)t).cancel(mode) || result;
+ }
+ return super.doCancel(mode) || result;
+ // returns true if anything is successfully cancelled
+ }
+
+ public boolean cancel(TaskCancellationMode mode) {
+ return cancel(mode, null);
}
- public boolean cancel(boolean mayInterruptTask, boolean interruptPrimaryThread, boolean alsoCancelChildren) {
+
+ protected boolean cancel(TaskCancellationMode mode, Boolean interruptPrimaryThreadOverride) {
if (isDone()) return false;
- if (log.isTraceEnabled()) log.trace("cancelling {}", this);
- boolean cancel = super.cancel(mayInterruptTask);
- if (alsoCancelChildren) {
- for (Task<?> t: secondaryJobsAll)
- cancel |= t.cancel(mayInterruptTask);
+ if (log.isTraceEnabled()) log.trace("cancelling DST {}", this);
+
+ // first do the super's cancel, setting cancelled, and calling doCancel to cancel children
+ boolean result = super.cancel(mode);
+ // then come back and ensure our primary thread is cancelled if needed
+
+ if (interruptPrimaryThreadOverride==null) interruptPrimaryThreadOverride = mode.isAllowedToInterruptTask();
+ if (log.isTraceEnabled()) {
+ log.trace("DST cancelling "+this+" mode "+mode+", interruptPrimary "+interruptPrimaryThreadOverride);
}
+
synchronized (jobTransitionLock) {
if (primaryThread!=null) {
- if (interruptPrimaryThread) {
+ if (interruptPrimaryThreadOverride) {
if (log.isTraceEnabled()) log.trace("cancelling {} - interrupting", this);
primaryThread.interrupt();
}
- cancel = true;
+ result = true;
}
}
- return cancel;
+
+ return result;
}
@Override
@@ -309,7 +326,7 @@ public class DynamicSequentialTask<T> extends BasicTask<T> implements HasTaskChi
}
if (!primaryFinished && failureHandlingConfig.cancelPrimaryOnSecondaryFailure) {
- cancel(true, false, false);
+ cancel(TaskCancellationMode.INTERRUPT_TASK_BUT_NOT_SUBMITTED_TASKS, false);
}
result.add(Tasks.getError(secondaryJob));
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ExecutionUtils.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ExecutionUtils.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ExecutionUtils.java
deleted file mode 100644
index 72a5ae4..0000000
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ExecutionUtils.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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 groovy.lang.Closure;
-
-import java.util.concurrent.Callable;
-
-import com.google.common.base.Function;
-import com.google.common.base.Throwables;
-
-public class ExecutionUtils {
- /**
- * Attempts to run/call the given object, with the given arguments if possible, preserving the return value if there is one (null otherwise);
- * throws exception if the callable is a non-null object which cannot be invoked (not a callable or runnable)
- * @deprecated since 0.7.0 ; this super-loose typing should be avoided; if it is needed, let's move it to one of the Groovy compatibility classes
- */
- @SuppressWarnings({ "unchecked", "rawtypes" })
- public static Object invoke(Object callable, Object ...args) {
- if (callable instanceof Closure) return ((Closure<?>)callable).call(args);
- if (callable instanceof Callable) {
- try {
- return ((Callable<?>)callable).call();
- } catch (Throwable t) {
- throw Throwables.propagate(t);
- }
- }
- if (callable instanceof Runnable) { ((Runnable)callable).run(); return null; }
- if (callable instanceof Function && args.length == 1) { return ((Function)callable).apply(args[0]); }
- if (callable==null) return null;
- throw new IllegalArgumentException("Cannot invoke unexpected object "+callable+" of type "+callable.getClass()+", with "+args.length+" args");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ListenableForwardingFuture.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ListenableForwardingFuture.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ListenableForwardingFuture.java
index 4ce56d1..cbc474c 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ListenableForwardingFuture.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ListenableForwardingFuture.java
@@ -21,15 +21,26 @@ package org.apache.brooklyn.util.core.task;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
+import org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import com.google.common.util.concurrent.ExecutionList;
import com.google.common.util.concurrent.ForwardingFuture.SimpleForwardingFuture;
import com.google.common.util.concurrent.ListenableFuture;
-/** Wraps a Future, making it a ListenableForwardingFuture, but with the caller having the resposibility to:
+/** Wraps a Future, making it a ListenableForwardingFuture, but with the caller having the responsibility to:
* <li> invoke the listeners on job completion (success or error)
- * <li> invoke the listeners on cancel */
+ * <li> invoke the listeners on cancel
+ *
+ * @deprecated since 0.9.0 likely to leave the public API */
+@Deprecated // TODO just one subclass, it can hold the behaviour we need from this,
+// and the methods here are surprising as they expect the caller to notify the list
public abstract class ListenableForwardingFuture<T> extends SimpleForwardingFuture<T> implements ListenableFuture<T> {
+ private static final Logger log = LoggerFactory.getLogger(ListenableForwardingFuture.class);
+
+ // TODO these are never accessed or used
final ExecutionList listeners;
protected ListenableForwardingFuture(Future<T> delegate) {
@@ -42,9 +53,22 @@ public abstract class ListenableForwardingFuture<T> extends SimpleForwardingFutu
this.listeners = list;
}
+ private static boolean warned = false;
+
@Override
public void addListener(Runnable listener, Executor executor) {
+ if (!warned) {
+ log.warn("Use of deprecated ListenableForwardingFuture.addListener at "+this+" (future calls will not be logged)", new Throwable("stack trace"));
+ warned = true;
+ }
+
listeners.add(listener, executor);
}
+ public abstract boolean cancel(TaskCancellationMode mode);
+
+ public final boolean cancel(boolean mayInterrupt) {
+ return cancel(TaskCancellationMode.INTERRUPT_TASK_AND_DEPENDENT_SUBMITTED_TASKS);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java
index c1ad4f8..219f4f8 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java
@@ -43,7 +43,7 @@ import com.google.common.base.Throwables;
*/
// TODO ScheduledTask is a very pragmatic implementation; would be nice to tighten,
// reduce external assumptions about internal structure, and clarify "done" semantics
-public class ScheduledTask extends BasicTask {
+public class ScheduledTask extends BasicTask<Object> {
final Callable<Task<?>> taskFactory;
@@ -84,7 +84,7 @@ public class ScheduledTask extends BasicTask {
this(MutableMap.of(), task);
}
- public ScheduledTask(Map flags, final Task<?> task){
+ public ScheduledTask(Map<?,?> flags, final Task<?> task){
this(flags, new Callable<Task<?>>(){
@Override
public Task<?> call() throws Exception {
@@ -92,7 +92,7 @@ public class ScheduledTask extends BasicTask {
}});
}
- public ScheduledTask(Map flags, Callable<Task<?>> taskFactory) {
+ public ScheduledTask(Map<?,?> flags, Callable<Task<?>> taskFactory) {
super(flags);
this.taskFactory = taskFactory;
@@ -194,13 +194,11 @@ public class ScheduledTask extends BasicTask {
}
@Override
- public synchronized boolean cancel(boolean mayInterrupt) {
- boolean result = super.cancel(mayInterrupt);
+ protected boolean doCancel(org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode mode) {
if (nextRun!=null) {
- nextRun.cancel(mayInterrupt);
- notifyAll();
+ ((TaskInternal<?>)nextRun).cancel(mode);
}
- return result;
+ return super.doCancel(mode);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
index 2bf0fec..99c2773 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java
@@ -29,6 +29,7 @@ import org.apache.brooklyn.util.time.Duration;
import com.google.common.annotations.Beta;
import com.google.common.base.Function;
+import com.google.common.base.Objects;
import com.google.common.util.concurrent.ExecutionList;
import com.google.common.util.concurrent.ListenableFuture;
@@ -95,6 +96,8 @@ public interface TaskInternal<T> extends Task<T> {
Object getExtraStatusText();
+ /** On task completion (or cancellation) runs the listeners which have been registered using
+ * {@link #addListener(Runnable, java.util.concurrent.Executor)}. */
void runListeners();
void setEndTimeUtc(long val);
@@ -120,5 +123,41 @@ public interface TaskInternal<T> extends Task<T> {
/** if a task is a proxy for another one (used mainly for internal tasks),
* this returns the "real" task represented by this one */
Task<?> getProxyTarget();
+
+ /** clearer semantics around cancellation; may be promoted to {@link Task} if we */
+ @Beta
+ public boolean cancel(TaskCancellationMode mode);
+
+ @Beta
+ public static class TaskCancellationMode {
+ public static final TaskCancellationMode DO_NOT_INTERRUPT = new TaskCancellationMode(false, false, false);
+ public static final TaskCancellationMode INTERRUPT_TASK_BUT_NOT_SUBMITTED_TASKS = new TaskCancellationMode(true, false, false);
+ public static final TaskCancellationMode INTERRUPT_TASK_AND_DEPENDENT_SUBMITTED_TASKS = new TaskCancellationMode(true, true, false);
+ public static final TaskCancellationMode INTERRUPT_TASK_AND_ALL_SUBMITTED_TASKS = new TaskCancellationMode(true, true, true);
+
+ private final boolean allowedToInterruptTask,
+ allowedToInterruptDependentSubmittedTasks,
+ allowedToInterruptAllSubmittedTasks;
+
+ private TaskCancellationMode(boolean mayInterruptIfRunning, boolean interruptSubmittedTransients, boolean interruptAllSubmitted) {
+ this.allowedToInterruptTask = mayInterruptIfRunning;
+ this.allowedToInterruptDependentSubmittedTasks = interruptSubmittedTransients;
+ this.allowedToInterruptAllSubmittedTasks = interruptAllSubmitted;
+ }
+
+ public boolean isAllowedToInterruptTask() { return allowedToInterruptTask; }
+ /** Implementation-dependent what "dependent" means in this context,
+ * e.g. may be linked to a "transient" tag (that's what Brooklyn does) */
+ public boolean isAllowedToInterruptDependentSubmittedTasks() { return allowedToInterruptDependentSubmittedTasks; }
+ public boolean isAllowedToInterruptAllSubmittedTasks() { return allowedToInterruptAllSubmittedTasks; }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this).add("interruptTask", allowedToInterruptTask)
+ .add("interruptDependentSubmitted", allowedToInterruptDependentSubmittedTasks)
+ .add("interruptAllSubmitted", allowedToInterruptAllSubmittedTasks)
+ .toString();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskPredicates.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskPredicates.java b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskPredicates.java
index 8e46002..d8d3764 100644
--- a/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskPredicates.java
+++ b/brooklyn-server/core/src/main/java/org/apache/brooklyn/util/core/task/TaskPredicates.java
@@ -60,4 +60,20 @@ public class TaskPredicates {
return "displayNameMatches("+matcher+")";
}
}
+
+ public static Predicate<Task<?>> isDone() {
+ return new IsDone();
+ }
+
+ private static class IsDone implements Predicate<Task<?>> {
+ @Override
+ public boolean apply(Task<?> input) {
+ return input.isDone();
+ }
+ @Override
+ public String toString() {
+ return "isDone()";
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTaskExecutionPerformanceTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTaskExecutionPerformanceTest.java b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTaskExecutionPerformanceTest.java
index a291c53..e8e7890 100644
--- a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTaskExecutionPerformanceTest.java
+++ b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTaskExecutionPerformanceTest.java
@@ -70,7 +70,6 @@ public class BasicTaskExecutionPerformanceTest {
if (em != null) em.shutdownNow();
}
- @SuppressWarnings("unchecked")
@Test
public void testScheduledTaskExecutedAfterDelay() throws Exception {
int delay = 100;
@@ -95,7 +94,6 @@ public class BasicTaskExecutionPerformanceTest {
assertTrue(actualDelay < (delay+MAX_OVERHEAD_MS), "actualDelay="+actualDelay+"; delay="+delay);
}
- @SuppressWarnings("unchecked")
@Test
public void testScheduledTaskExecutedAtRegularPeriod() throws Exception {
final int period = 100;
@@ -127,7 +125,6 @@ public class BasicTaskExecutionPerformanceTest {
}
}
- @SuppressWarnings("unchecked")
@Test
public void testCanCancelScheduledTask() throws Exception {
final int period = 1;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
index ceff29f..95f0952 100644
--- a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
+++ b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/DynamicSequentialTaskTest.java
@@ -30,10 +30,13 @@ import java.util.concurrent.TimeoutException;
import org.apache.brooklyn.api.mgmt.HasTaskChildren;
import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.test.Asserts;
import org.apache.brooklyn.util.collections.CollectionFunctionals;
import org.apache.brooklyn.util.collections.MutableList;
import org.apache.brooklyn.util.collections.MutableMap;
+import org.apache.brooklyn.util.collections.MutableSet;
+import org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.math.MathPredicates;
import org.apache.brooklyn.util.time.CountdownTimer;
@@ -49,6 +52,8 @@ import org.testng.annotations.Test;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.base.Stopwatch;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
@@ -154,7 +159,17 @@ public class DynamicSequentialTaskTest {
}
public Task<String> sayTask(String message, Duration duration, String message2) {
- return Tasks.<String>builder().body(sayCallable(message, duration, message2)).build();
+ return Tasks.<String>builder().displayName("say:"+message).body(sayCallable(message, duration, message2)).build();
+ }
+
+ public <T> Task<T> submitting(final Task<T> task) {
+ return Tasks.<T>builder().displayName("submitting:"+task.getId()).body(new Callable<T>() {
+ @Override
+ public T call() throws Exception {
+ ec.submit(task);
+ return task.get();
+ }
+ }).build();
}
@Test
@@ -207,6 +222,85 @@ public class DynamicSequentialTaskTest {
// but we do _not_ get a mutex from task3 as it does not run (is not interrupted)
Assert.assertEquals(cancellations.availablePermits(), 0);
}
+
+ @Test
+ public void testCancellationModeAndSubmitted() throws Exception {
+ doTestCancellationModeAndSubmitted(true, TaskCancellationMode.DO_NOT_INTERRUPT, false, false);
+
+ doTestCancellationModeAndSubmitted(true, TaskCancellationMode.INTERRUPT_TASK_AND_ALL_SUBMITTED_TASKS, true, true);
+ doTestCancellationModeAndSubmitted(true, TaskCancellationMode.INTERRUPT_TASK_AND_DEPENDENT_SUBMITTED_TASKS, true, true);
+ doTestCancellationModeAndSubmitted(true, TaskCancellationMode.INTERRUPT_TASK_BUT_NOT_SUBMITTED_TASKS, true, false);
+
+ // if it's not transient, it should only be cancelled on "all submitted"
+ doTestCancellationModeAndSubmitted(false, TaskCancellationMode.INTERRUPT_TASK_AND_DEPENDENT_SUBMITTED_TASKS, true, false);
+ doTestCancellationModeAndSubmitted(false, TaskCancellationMode.INTERRUPT_TASK_AND_ALL_SUBMITTED_TASKS, true, true);
+
+ // cancellation mode left off should be the same as TASK_AND_DEPENDENT, i.e. don't cancel non-transient bg submitted
+ doTestCancellationModeAndSubmitted(true, null, true, true);
+ doTestCancellationModeAndSubmitted(false, null, true, false);
+ // and 'true' should be the same
+ doTestCancellationModeAndSubmitted(true, true, true, true);
+ doTestCancellationModeAndSubmitted(false, true, true, false);
+
+ // cancellation mode false should be the same as DO_NOT_INTERRUPT
+ doTestCancellationModeAndSubmitted(true, false, false, false);
+ }
+
+ public void doTestCancellationModeAndSubmitted(
+ boolean isSubtaskTransient,
+ Object cancellationMode,
+ boolean expectedTaskInterrupted,
+ boolean expectedSubtaskCancelled
+ ) throws Exception {
+ tearDown(); setUp();
+
+ final Task<String> t1 = sayTask("1-wait", Duration.minutes(10), "1-done");
+ if (isSubtaskTransient) {
+ BrooklynTaskTags.addTagDynamically(t1, BrooklynTaskTags.TRANSIENT_TASK_TAG);
+ }
+
+ final Task<List<?>> t = Tasks.parallel(
+ submitting(t1),
+ sayTask("2-wait", Duration.minutes(10), "2-done"));
+ ec.submit(t);
+
+ waitForMessages(Predicates.compose(MathPredicates.greaterThanOrEqual(2), CollectionFunctionals.sizeFunction()), TIMEOUT);
+ Asserts.assertEquals(MutableSet.copyOf(messages), MutableSet.of("1-wait", "2-wait"));
+
+ Time.sleep(Duration.millis(400));
+
+ if (cancellationMode==null) {
+ ((TaskInternal<?>)t).cancel();
+ } else if (cancellationMode instanceof Boolean) {
+ t.cancel((Boolean)cancellationMode);
+ } else if (cancellationMode instanceof TaskCancellationMode) {
+ ((TaskInternal<?>)t).cancel((TaskCancellationMode)cancellationMode);
+ }
+
+ // the cancelled task always reports cancelled and done
+ Assert.assertEquals(t.isDone(), true);
+ Assert.assertEquals(t.isCancelled(), true);
+ // end time might not be set for another fraction of a second
+ if (expectedTaskInterrupted) {
+ Asserts.eventually(new Supplier<Number>() {
+ @Override public Number get() { return t.getEndTimeUtc(); }},
+ MathPredicates.<Number>greaterThanOrEqual(0));
+ } else {
+ Assert.assertTrue(t.getEndTimeUtc() < 0, "Wrong end time: "+t.getEndTimeUtc());
+ }
+
+ if (expectedSubtaskCancelled) {
+ Asserts.eventually(Suppliers.ofInstance(t1), TaskPredicates.isDone());
+ Assert.assertTrue(t1.isCancelled());
+ Asserts.eventually(new Supplier<Number>() {
+ @Override public Number get() { return t1.getEndTimeUtc(); }},
+ MathPredicates.<Number>greaterThanOrEqual(0));
+ } else {
+ Time.sleep(Duration.millis(5));
+ Assert.assertFalse(t1.isCancelled());
+ Assert.assertFalse(t1.isDone());
+ }
+ }
protected void waitForMessages(Predicate<? super List<String>> predicate, Duration timeout) throws Exception {
long endtime = System.currentTimeMillis() + timeout.toMilliseconds();
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java
index 1d1c3af..6fd3021 100644
--- a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java
+++ b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java
@@ -64,6 +64,11 @@ public class NonBasicTaskExecutionTest {
protected TaskInternal<T> delegate() {
return delegate;
}
+
+ @Override
+ public boolean cancel(TaskCancellationMode mode) {
+ return delegate.cancel(mode);
+ }
}
private BasicExecutionManager em;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/ScheduledExecutionTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/ScheduledExecutionTest.java b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/ScheduledExecutionTest.java
index 1d551e8..5c11355 100644
--- a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/ScheduledExecutionTest.java
+++ b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/ScheduledExecutionTest.java
@@ -45,7 +45,7 @@ import org.testng.annotations.Test;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
-@SuppressWarnings({"unchecked","rawtypes"})
+@SuppressWarnings({"rawtypes"})
public class ScheduledExecutionTest {
public static final Logger log = LoggerFactory.getLogger(ScheduledExecutionTest.class);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/TaskPredicatesTest.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/TaskPredicatesTest.java b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/TaskPredicatesTest.java
index fce6f0f..8a25361 100644
--- a/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/TaskPredicatesTest.java
+++ b/brooklyn-server/core/src/test/java/org/apache/brooklyn/util/core/task/TaskPredicatesTest.java
@@ -57,8 +57,8 @@ public class TaskPredicatesTest extends BrooklynAppUnitTestSupport {
.body(Callables.<Object>returning("val"))
.displayName("myname")
.build());
- assertTrue(TaskPredicates.displayNameMatches(Predicates.equalTo("myname")).apply(task));
- assertFalse(TaskPredicates.displayNameMatches(Predicates.equalTo("wrong")).apply(task));
+ assertTrue(TaskPredicates.displayNameSatisfies(Predicates.equalTo("myname")).apply(task));
+ assertFalse(TaskPredicates.displayNameSatisfies(Predicates.equalTo("wrong")).apply(task));
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/aed07863/brooklyn-server/rest/rest-server/src/main/java/org/apache/brooklyn/rest/resources/EntityConfigResource.java
----------------------------------------------------------------------
diff --git a/brooklyn-server/rest/rest-server/src/main/java/org/apache/brooklyn/rest/resources/EntityConfigResource.java b/brooklyn-server/rest/rest-server/src/main/java/org/apache/brooklyn/rest/resources/EntityConfigResource.java
index e483d0b..a6d3b8e 100644
--- a/brooklyn-server/rest/rest-server/src/main/java/org/apache/brooklyn/rest/resources/EntityConfigResource.java
+++ b/brooklyn-server/rest/rest-server/src/main/java/org/apache/brooklyn/rest/resources/EntityConfigResource.java
@@ -22,6 +22,7 @@ import static com.google.common.collect.Iterables.transform;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.Callable;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.config.ConfigKey;
@@ -35,6 +36,7 @@ import org.apache.brooklyn.rest.filter.HaHotStateRequired;
import org.apache.brooklyn.rest.transform.EntityTransformer;
import org.apache.brooklyn.rest.util.WebResourceUtils;
import org.apache.brooklyn.util.core.flags.TypeCoercions;
+import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.core.task.ValueResolver;
import org.apache.brooklyn.util.text.Strings;
import org.apache.brooklyn.util.time.Duration;
@@ -71,14 +73,32 @@ public class EntityConfigResource extends AbstractBrooklynRestResource implement
public Map<String, Object> batchConfigRead(String application, String entityToken, Boolean raw) {
// TODO: add test
Entity entity = brooklyn().getEntity(application, entityToken);
- Map<ConfigKey<?>, ?> source = ((EntityInternal) entity).config().getBag().getAllConfigAsConfigKeyMap();
- Map<String, Object> result = Maps.newLinkedHashMap();
- for (Map.Entry<ConfigKey<?>, ?> ek : source.entrySet()) {
- Object value = ek.getValue();
- result.put(ek.getKey().getName(),
- resolving(value).preferJson(true).asJerseyOutermostReturnValue(false).raw(raw).context(entity).timeout(Duration.ZERO).renderAs(ek.getKey()).resolve());
+ // wrap in a task for better runtime view
+ return Entities.submit(entity, Tasks.<Map<String,Object>>builder().displayName("REST API batch config read").body(new BatchConfigRead(this, entity, raw)).build()).getUnchecked();
+ }
+
+ private static class BatchConfigRead implements Callable<Map<String,Object>> {
+ private EntityConfigResource resource;
+ private Entity entity;
+ private Boolean raw;
+
+ public BatchConfigRead(EntityConfigResource resource, Entity entity, Boolean raw) {
+ this.resource = resource;
+ this.entity = entity;
+ this.raw = raw;
+ }
+
+ @Override
+ public Map<String, Object> call() throws Exception {
+ Map<ConfigKey<?>, ?> source = ((EntityInternal) entity).config().getBag().getAllConfigAsConfigKeyMap();
+ Map<String, Object> result = Maps.newLinkedHashMap();
+ for (Map.Entry<ConfigKey<?>, ?> ek : source.entrySet()) {
+ Object value = ek.getValue();
+ result.put(ek.getKey().getName(),
+ resource.resolving(value).preferJson(true).asJerseyOutermostReturnValue(false).raw(raw).context(entity).timeout(Duration.ZERO).renderAs(ek.getKey()).resolve());
+ }
+ return result;
}
- return result;
}
@Override