more bookkeeping for immediate/same-thread tasks
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/9d2faf0e Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/9d2faf0e Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/9d2faf0e Branch: refs/heads/master Commit: 9d2faf0eddcd37ed0ec31cf1b9856a95b5702e14 Parents: a11af34 Author: Alex Heneveld <[email protected]> Authored: Mon Sep 11 08:40:31 2017 +0100 Committer: Alex Heneveld <[email protected]> Committed: Fri Sep 15 10:29:08 2017 +0100 ---------------------------------------------------------------------- .../util/core/task/BasicExecutionContext.java | 14 ++++++--- .../util/core/task/BasicExecutionManager.java | 32 ++++++++++++++------ .../util/core/task/ValueResolverTest.java | 3 +- 3 files changed, 34 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/9d2faf0e/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 6bb7679..e0dd8b0 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 @@ -131,11 +131,11 @@ public class BasicExecutionContext extends AbstractExecutionContext { Task<?> previousTask = BasicExecutionManager.getPerThreadCurrentTask().get(); BasicExecutionContext oldExecutionContext = getCurrentExecutionContext(); registerPerThreadExecutionContext(); + ((BasicExecutionManager)executionManager).beforeSubmitInSameThreadTask(null, fakeTaskForContext); - if (previousTask!=null) fakeTaskForContext.setSubmittedByTask(previousTask); - fakeTaskForContext.cancel(); try { - BasicExecutionManager.getPerThreadCurrentTask().set(fakeTaskForContext); + ((BasicExecutionManager)executionManager).beforeStartInSameThreadTask(null, fakeTaskForContext); + fakeTaskForContext.cancel(); if (!(callableOrSupplier instanceof ImmediateSupplier)) { callableOrSupplier = InterruptingImmediateSupplier.of(callableOrSupplier); @@ -150,8 +150,12 @@ public class BasicExecutionContext extends AbstractExecutionContext { } } finally { - BasicExecutionManager.getPerThreadCurrentTask().set(previousTask); - perThreadExecutionContext.set(oldExecutionContext); + try { + ((BasicExecutionManager)executionManager).afterEndInSameThreadTask(null, fakeTaskForContext); + } finally { + BasicExecutionManager.getPerThreadCurrentTask().set(previousTask); + perThreadExecutionContext.set(oldExecutionContext); + } } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/9d2faf0e/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 4f8a9ec..aa61a57 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 @@ -721,6 +721,9 @@ public class BasicExecutionManager implements ExecutionManager { protected void beforeSubmitAtomicTask(Map<?,?> flags, Task<?> task) { internalBeforeSubmit(flags, task); } + protected void beforeSubmitInSameThreadTask(Map<?,?> flags, Task<?> task) { + internalBeforeSubmit(flags, task); + } /** invoked when a task is submitted */ protected void internalBeforeSubmit(Map<?,?> flags, Task<?> task) { incompleteTaskIds.add(task.getId()); @@ -729,8 +732,8 @@ public class BasicExecutionManager implements ExecutionManager { if (currentTask!=null) ((TaskInternal<?>)task).setSubmittedByTask(currentTask); ((TaskInternal<?>)task).setSubmitTimeUtc(System.currentTimeMillis()); - if (flags.get("tag")!=null) ((TaskInternal<?>)task).getMutableTags().add(flags.remove("tag")); - if (flags.get("tags")!=null) ((TaskInternal<?>)task).getMutableTags().addAll((Collection<?>)flags.remove("tags")); + if (flags!=null && flags.get("tag")!=null) ((TaskInternal<?>)task).getMutableTags().add(flags.remove("tag")); + if (flags!=null && flags.get("tags")!=null) ((TaskInternal<?>)task).getMutableTags().addAll((Collection<?>)flags.remove("tags")); for (Object tag: BrooklynTaskTags.getTagsFast(task)) { tasksWithTagCreating(tag).add(task); @@ -741,15 +744,18 @@ public class BasicExecutionManager implements ExecutionManager { } protected void beforeStartScheduledTaskSubmissionIteration(Map<?,?> flags, Task<?> task) { - internalBeforeStart(flags, task); + internalBeforeStart(flags, task, true); } protected void beforeStartAtomicTask(Map<?,?> flags, Task<?> task) { - internalBeforeStart(flags, task); + internalBeforeStart(flags, task, true); + } + protected void beforeStartInSameThreadTask(Map<?,?> flags, Task<?> task) { + internalBeforeStart(flags, task, false); } /** invoked in a task's thread when a task is starting to run (may be some time after submitted), * but before doing any of the task's work, so that we can update bookkeeping and notify callbacks */ - protected void internalBeforeStart(Map<?,?> flags, Task<?> task) { + protected void internalBeforeStart(Map<?,?> flags, Task<?> task, boolean allowJitter) { int count = activeTaskCount.incrementAndGet(); if (count % 1000==0) { log.warn("High number of active tasks: task #"+count+" is "+task); @@ -769,9 +775,12 @@ public class BasicExecutionManager implements ExecutionManager { ((TaskInternal<?>)task).setStartTimeUtc(System.currentTimeMillis()); } - jitterThreadStart(task); - - invokeCallback(flags.get("newTaskStartCallback"), task); + if (allowJitter) { + jitterThreadStart(task); + } + if (flags!=null) { + invokeCallback(flags.get("newTaskStartCallback"), task); + } } private void jitterThreadStart(Task<?> task) { @@ -825,6 +834,9 @@ public class BasicExecutionManager implements ExecutionManager { protected void afterEndAtomicTask(Map<?,?> flags, Task<?> task) { internalAfterEnd(flags, task, true, true); } + protected void afterEndInSameThreadTask(Map<?,?> flags, Task<?> task) { + internalAfterEnd(flags, task, true, true); + } /** normally (if not interrupted) called once for each call to {@link #internalBeforeSubmit(Map, Task)}, * and, for atomic tasks and scheduled-task submission iterations where * always called once if {@link #internalBeforeStart(Map, Task)} is invoked and in the same thread as that method */ @@ -835,7 +847,9 @@ public class BasicExecutionManager implements ExecutionManager { } if (isEndingAllIterations) { incompleteTaskIds.remove(task.getId()); - invokeCallback(flags.get("newTaskEndCallback"), task); + if (flags!=null) { + invokeCallback(flags.get("newTaskEndCallback"), task); + } ((TaskInternal<?>)task).setEndTimeUtc(System.currentTimeMillis()); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/9d2faf0e/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java b/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java index 6e57c84..1f5c754 100644 --- a/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java +++ b/core/src/test/java/org/apache/brooklyn/util/core/task/ValueResolverTest.java @@ -471,7 +471,8 @@ public class ValueResolverTest extends BrooklynAppUnitTestSupport { private void assertImmediateFakeTaskFromMethod(CallInfo callInfo, String method) { // previously task was null, but now there is a "fake task" assertNotNull(callInfo.task); - Assert.assertFalse(callInfo.task.isSubmitted()); + // it is now submitted in same thread (2017-09) + Assert.assertTrue(callInfo.task.isSubmitted()); assertContainsCallingMethod(callInfo.stackTrace, method); }
