address many of the PR comments, bigger ones to follow
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/8077f16a Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/8077f16a Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/8077f16a Branch: refs/heads/master Commit: 8077f16a807ccd585a0f13310a5a44f278770864 Parents: 2bdcf1a Author: Alex Heneveld <[email protected]> Authored: Thu Sep 28 15:25:10 2017 +0100 Committer: Alex Heneveld <[email protected]> Committed: Thu Sep 28 15:34:16 2017 +0100 ---------------------------------------------------------------------- .../util/core/task/BasicExecutionContext.java | 16 +++++++++----- .../util/core/task/BasicExecutionManager.java | 3 +-- .../brooklyn/util/core/task/ValueResolver.java | 3 ++- .../core/effector/EffectorSayHiTest.java | 1 + .../util/core/task/BasicTasksFutureTest.java | 23 ++++++++++---------- .../core/task/NonBasicTaskExecutionTest.java | 8 +------ .../org/apache/brooklyn/util/guava/Maybe.java | 2 +- .../apache/brooklyn/util/repeat/Repeater.java | 7 ++++-- 8 files changed, 33 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/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 429cad7..5e645a9 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 @@ -98,7 +98,7 @@ public class BasicExecutionContext extends AbstractExecutionContext { * Creates an execution context which wraps {@link ExecutionManager} * adding the given tags to all tasks submitted through this context. */ - public BasicExecutionContext(ExecutionManager executionManager, Iterable<Object> tagsForThisContext) { + public BasicExecutionContext(ExecutionManager executionManager, Iterable<?> tagsForThisContext) { this.executionManager = executionManager; if (tagsForThisContext!=null) Iterables.addAll(tags, tagsForThisContext); @@ -128,11 +128,7 @@ public class BasicExecutionContext extends AbstractExecutionContext { final TaskInternal<T> t = (TaskInternal<T>) task.asTask(); if (t.isQueuedOrSubmitted()) { - if (t.isDone()) { - return t.getUnchecked(); - } else { - throw new ImmediateUnsupportedException("Task is in progress and incomplete: "+t); - } + return t.getUnchecked(); } ContextSwitchingInfo<T> switchContextWrapper = getContextSwitchingTask(t, Collections.emptyList(), false); @@ -151,6 +147,8 @@ public class BasicExecutionContext extends AbstractExecutionContext { } } + // could perhaps use Guava's SettableFuture -- though would have to take care re + // supporting set(Maybe<T>) private static class SimpleFuture<T> implements Future<T> { boolean cancelled = false; boolean done = false; @@ -206,6 +204,12 @@ public class BasicExecutionContext extends AbstractExecutionContext { } } + /** Internal utility method to avoid replication between + * implementations in {@link #get(Task)} and {@link #getImmediately(Object)}. + * The two submit different jobs but after doing a lot of the same setup and catch/finally. + * Logic re return type is a little fiddly given the differences but should be clearer + * seeing how the two work (as opposed to this method being designed as something + * more generally useful). */ private <T> Maybe<T> runInSameThread(final Task<T> task, Callable<Maybe<T>> job) throws Exception { ((TaskInternal<T>)task).getMutableTags().addAll(tags); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/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 4fcfadb..5818568 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 @@ -263,7 +263,7 @@ public class BasicExecutionManager implements ExecutionManager { incompleteTaskIds.remove(task.getId()); if (removed!=null && removed.isSubmitted() && !removed.isDone(true)) { Entity context = BrooklynTaskTags.getContextEntity(removed); - if (!Entities.isManaged(context)) { + if (context!=null && !Entities.isManaged(context)) { log.debug("Forgetting about active task on unmanagement of "+context+": "+removed); } else { log.warn("Deleting submitted task before completion: "+removed+"; this task will continue to run in the background outwith "+this+", but perhaps it should have been cancelled?"); @@ -470,7 +470,6 @@ public class BasicExecutionManager implements ExecutionManager { } Object result; try { - lastError = null; result = oldJob.call(); task.lastThrownType = null; } catch (Exception e) { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/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 d93dfa4..0e75d16 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 @@ -434,7 +434,8 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj // (should discourage this in favour of task factories which can be transiently interrupted?) BrooklynTaskTags.addTagDynamically(task, BrooklynTaskTags.NON_TRANSIENT_TASK_TAG); } - if (timer!=null || Thread.currentThread().isInterrupted()) { + // FIXME + if (timer!=null || Thread.currentThread().isInterrupted() || isEvaluatingImmediately()) { exec.submit(task); } else { exec.get(task); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/core/src/test/java/org/apache/brooklyn/core/effector/EffectorSayHiTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/core/effector/EffectorSayHiTest.java b/core/src/test/java/org/apache/brooklyn/core/effector/EffectorSayHiTest.java index 9039cf2..5110315 100644 --- a/core/src/test/java/org/apache/brooklyn/core/effector/EffectorSayHiTest.java +++ b/core/src/test/java/org/apache/brooklyn/core/effector/EffectorSayHiTest.java @@ -110,6 +110,7 @@ public class EffectorSayHiTest extends BrooklynAppUnitTestSupport { } @Test(groups="WIP") // see comments at BasicExecutionContext.getImmediately + // FIXME public void testInvocationGetImmediately() throws Exception { assertEquals(((EntityInternal)e).getExecutionContext() .getImmediately( Effectors.invocation(e, MyEntity.SAY_HI_1, ImmutableMap.of("name", "Bob", "greeting", "hi")) ), "hi Bob"); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTasksFutureTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTasksFutureTest.java b/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTasksFutureTest.java index a2bfd59..0db1158 100644 --- a/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTasksFutureTest.java +++ b/core/src/test/java/org/apache/brooklyn/util/core/task/BasicTasksFutureTest.java @@ -176,7 +176,7 @@ public class BasicTasksFutureTest { @Test public void testCancelAfterStartTriggersListenableFutureDynamic() throws Exception { - doTestCancelTriggersListenableFuture(Duration.millis(50), true); + doTestCancelTriggersListenableFuture(null, Duration.millis(50), true); } @Test public void testCancelImmediateTriggersListenableFutureDynamic() throws Exception { @@ -184,32 +184,32 @@ public class BasicTasksFutureTest { // that needs handling separately as it falls into an edge where the future is set and cancelled // so the executor won't run it, and our wrapper logic doesn't apply; // this test doesn't guarantee this code path, but makes it likely enough it happens once in a while. - doTestCancelTriggersListenableFuture(Duration.ZERO, true); + doTestCancelTriggersListenableFuture(null, Duration.ZERO, true); } @Test public void testCancelBeforeTriggersListenableFutureDynamic() throws Exception { - doTestCancelTriggersListenableFuture(Duration.millis(-50), true); + doTestCancelTriggersListenableFuture(Duration.millis(50), null, true); } @Test public void testCancelAfterStartTriggersListenableFutureSimple() throws Exception { - doTestCancelTriggersListenableFuture(Duration.millis(50), true); + doTestCancelTriggersListenableFuture(null, Duration.millis(50), true); } @Test public void testCancelImmediateTriggersListenableFutureSimple() throws Exception { - doTestCancelTriggersListenableFuture(Duration.ZERO, false); + doTestCancelTriggersListenableFuture(null, Duration.ZERO, false); } @Test public void testCancelBeforeTriggersListenableFutureSimple() throws Exception { - doTestCancelTriggersListenableFuture(Duration.millis(-50), false); + doTestCancelTriggersListenableFuture(Duration.millis(50), null, false); } - public void doTestCancelTriggersListenableFuture(Duration delay, boolean dynamic) throws Exception { + public void doTestCancelTriggersListenableFuture(Duration delayBeforeSubmit, Duration delayBeforeCancel, boolean dynamic) throws Exception { Task<String> t = waitForSemaphore(Duration.TEN_SECONDS, true, "x", dynamic); addFutureListener(t, "before"); Stopwatch watch = Stopwatch.createStarted(); - if (delay.isNegative()) { + if (delayBeforeSubmit!=null) { new Thread(() -> { - Time.sleep(delay.multiply(-1)); + Time.sleep(delayBeforeSubmit); ec.submit(t); }).start(); } else { @@ -218,10 +218,11 @@ public class BasicTasksFutureTest { addFutureListener(t, "during"); - log.info("test cancelling "+t+" ("+t.getClass()+") after "+delay); + log.info("test cancelling "+t+" ("+t.getClass()+") after "+delayBeforeCancel+ + ", submit delay "+delayBeforeSubmit); // NB: three different code paths (callers to this method) for notifying futures // depending whether task is started before, at, or after the cancel - if (!delay.isNegative()) Time.sleep(delay); + if (delayBeforeCancel!=null) Time.sleep(delayBeforeCancel); synchronized (data) { t.cancel(true); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java b/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java index 6fd3021..f31c7ab 100644 --- a/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java +++ b/core/src/test/java/org/apache/brooklyn/util/core/task/NonBasicTaskExecutionTest.java @@ -33,10 +33,6 @@ import java.util.concurrent.TimeUnit; import org.apache.brooklyn.api.mgmt.Task; import org.apache.brooklyn.test.Asserts; import org.apache.brooklyn.util.collections.MutableMap; -import org.apache.brooklyn.util.core.task.BasicExecutionManager; -import org.apache.brooklyn.util.core.task.BasicTask; -import org.apache.brooklyn.util.core.task.ForwardingTask; -import org.apache.brooklyn.util.core.task.TaskInternal; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testng.annotations.AfterMethod; @@ -44,9 +40,7 @@ import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; /** - * Test the operation of the {@link BasicTask} class. - * - * TODO clarify test purpose + * Test some internal aspects of {@link BasicTask} execution. */ public class NonBasicTaskExecutionTest { private static final Logger log = LoggerFactory.getLogger(NonBasicTaskExecutionTest.class); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java ---------------------------------------------------------------------- diff --git a/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java b/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java index a68b400..8b69cc0 100644 --- a/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java +++ b/utils/common/src/main/java/org/apache/brooklyn/util/guava/Maybe.java @@ -204,7 +204,7 @@ public abstract class Maybe<T> implements Serializable, Supplier<T> { public static class MaybeSupplier<T> extends AbstractPresent<T> { private static final long serialVersionUID = -823731500051341455L; - private Supplier<T> supplier; + private final Supplier<T> supplier; public MaybeSupplier(Supplier<T> value) { this.supplier = value; } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8077f16a/utils/common/src/main/java/org/apache/brooklyn/util/repeat/Repeater.java ---------------------------------------------------------------------- diff --git a/utils/common/src/main/java/org/apache/brooklyn/util/repeat/Repeater.java b/utils/common/src/main/java/org/apache/brooklyn/util/repeat/Repeater.java index 4c59c8d..bcd1c9c 100644 --- a/utils/common/src/main/java/org/apache/brooklyn/util/repeat/Repeater.java +++ b/utils/common/src/main/java/org/apache/brooklyn/util/repeat/Repeater.java @@ -30,6 +30,7 @@ import javax.annotation.Nullable; import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.exceptions.ReferenceWithError; +import org.apache.brooklyn.util.guava.Maybe; import org.apache.brooklyn.util.time.CountdownTimer; import org.apache.brooklyn.util.time.Duration; import org.apache.brooklyn.util.time.Time; @@ -278,13 +279,15 @@ public class Repeater implements Callable<Boolean> { Preconditions.checkNotNull(supplier, "supplier must not be null"); Preconditions.checkNotNull(exitCondition, "exitCondition must not be null"); return until(new Callable<Boolean>() { + private Maybe<T> lastValue = Maybe.absent(); @Override public Boolean call() throws Exception { - return exitCondition.apply(supplier.get()); + lastValue = Maybe.ofAllowingNull(supplier.get()); + return exitCondition.apply(lastValue.get()); } @Override public String toString() { - return supplier.get()+" "+exitCondition.toString(); + return ""+(lastValue.isPresent() ? lastValue.get() : supplier) + " " + exitCondition; } }); }
