diff task submisson paths share code about how to end, and tidy listeners fix bug where some where executed and GC'd before the child had actually run, by ensuring listeners don't run until the task recognizes the cancellation.
also tidy how end code is shared, and remove deprecation Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/934cf4cc Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/934cf4cc Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/934cf4cc Branch: refs/heads/master Commit: 934cf4ccbed5719f1b22ff032af77583c1cf6ad2 Parents: 8983bf2 Author: Alex Heneveld <[email protected]> Authored: Tue Sep 12 12:12:00 2017 +0100 Committer: Alex Heneveld <[email protected]> Committed: Fri Sep 15 10:29:09 2017 +0100 ---------------------------------------------------------------------- .../java/org/apache/brooklyn/api/mgmt/Task.java | 16 +- .../mgmt/ha/HighAvailabilityManagerImpl.java | 2 +- .../mgmt/internal/BrooklynGarbageCollector.java | 20 +- .../util/core/task/BasicExecutionContext.java | 6 +- .../util/core/task/BasicExecutionManager.java | 247 +++++++++++-------- .../brooklyn/util/core/task/BasicTask.java | 30 +-- .../brooklyn/util/core/task/CompoundTask.java | 4 - .../util/core/task/DynamicSequentialTask.java | 1 + .../brooklyn/util/core/task/ForwardingTask.java | 5 + .../brooklyn/util/core/task/ScheduledTask.java | 14 +- .../brooklyn/util/core/task/TaskInternal.java | 8 +- .../core/mgmt/rebind/RebindFeedTest.java | 4 +- 12 files changed, 213 insertions(+), 144 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java ---------------------------------------------------------------------- diff --git a/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java b/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java index 969c3a8..3c8aa45 100644 --- a/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java +++ b/api/src/main/java/org/apache/brooklyn/api/mgmt/Task.java @@ -91,7 +91,7 @@ public interface Task<T> extends ListenableFuture<T>, TaskAdaptable<T> { * {@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, + * To check whether cancelled threads for this task have completed, use {@link #isDone(boolean))}. * 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 @@ -99,6 +99,20 @@ public interface Task<T> extends ListenableFuture<T>, TaskAdaptable<T> { */ @Override public boolean isDone(); + + /** + * As {@link #isDone()}, identical if the argument is false, but by supplying {@code true} + * this will also check {@link #getEndTimeUtc()} if {@link #isBegun()} + * to guarantee that the task is no longer running. + * {@link #isDone()} will return true for cancelled tasks even if they are still running. + * <p> + * In a task hierarchy, the threads of tasks submitted by this may still be ongoing. + * To determine their completion, inspect the {@link ExecutionManager}. + * + * @param andTaskHasEnded + * @return + */ + public boolean isDone(boolean andTaskNotRunning); /** * Causes calling thread to block until the task is started. http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/core/mgmt/ha/HighAvailabilityManagerImpl.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/ha/HighAvailabilityManagerImpl.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/ha/HighAvailabilityManagerImpl.java index a4f8870..f6dc1a4 100644 --- a/core/src/main/java/org/apache/brooklyn/core/mgmt/ha/HighAvailabilityManagerImpl.java +++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/ha/HighAvailabilityManagerImpl.java @@ -589,7 +589,7 @@ public class HighAvailabilityManagerImpl implements HighAvailabilityManager { } else { if (pollingTask!=null) pollingTask.cancel(true); - ScheduledTask task = new ScheduledTask(MutableMap.of("period", pollPeriod, "displayName", "scheduled:[HA poller task]"), taskFactory); + ScheduledTask task = new ScheduledTask(MutableMap.of("period", pollPeriod, "displayName", "scheduled:[HA poller task]", "tag", BrooklynTaskTags.TRANSIENT_TASK_TAG), taskFactory); pollingTask = managementContext.getExecutionManager().submit(task); } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/BrooklynGarbageCollector.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/BrooklynGarbageCollector.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/BrooklynGarbageCollector.java index 203c3a5..7995768 100644 --- a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/BrooklynGarbageCollector.java +++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/BrooklynGarbageCollector.java @@ -292,15 +292,12 @@ public class BrooklynGarbageCollector { } } - /** @deprecated since 0.7.0, method moved internal until semantics are clarified; see also {@link #shouldDeleteTaskImmediately(Task)} */ - @Deprecated - public boolean shouldDeleteTask(Task<?> task) { - return shouldDeleteTaskImmediately(task); - } /** whether this task should be deleted on completion, * because it is transient, or because it is submitted background without much context information */ protected boolean shouldDeleteTaskImmediately(Task<?> task) { - if (!task.isDone()) return false; + if (!task.isDone(true)) { + return false; + } Set<Object> tags = BrooklynTaskTags.getTagsFast(task); if (tags.contains(ManagementContextInternal.TRANSIENT_TASK_TAG)) @@ -430,7 +427,10 @@ public class BrooklynGarbageCollector { while (ei.hasNext()) { Entry<Entity, Task<?>> ee = ei.next(); if (Entities.isManaged(ee.getKey())) continue; - if (ee.getValue()!=null && !ee.getValue().isDone()) continue; + if (ee.getValue()!=null && !ee.getValue().isDone(true)) { + // wait for the unmanagement task to complete + continue; + } deleteTasksForEntity(ee.getKey()); synchronized (unmanagedEntitiesNeedingGc) { unmanagedEntitiesNeedingGc.remove(ee.getKey()); @@ -446,7 +446,7 @@ public class BrooklynGarbageCollector { try { for (Task<?> task: allTasks) { - if (!task.isDone()) continue; + if (!task.isDone(true)) continue; if (BrooklynTaskTags.isSubTask(task)) continue; if (maxTaskAge.isShorterThan(Duration.sinceUtc(task.getEndTimeUtc()))) @@ -466,7 +466,7 @@ public class BrooklynGarbageCollector { protected void expireTransientTasks() { Set<Task<?>> transientTasks = executionManager.getTasksWithTag(BrooklynTaskTags.TRANSIENT_TASK_TAG); for (Task<?> t: transientTasks) { - if (!t.isDone()) continue; + if (!t.isDone(true)) continue; executionManager.deleteTask(t); } } @@ -532,7 +532,7 @@ public class BrooklynGarbageCollector { List<Task<?>> tasksToConsiderDeleting = MutableList.of(); try { for (Task<?> task: tasks) { - if (!task.isDone()) continue; + if (!task.isDone(true)) continue; Set<Object> tags = TaskTags.getTagsFast(task); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/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 8166dc0..547ae24 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 @@ -183,8 +183,9 @@ public class BasicExecutionContext extends AbstractExecutionContext { BasicExecutionContext oldExecutionContext = getCurrentExecutionContext(); registerPerThreadExecutionContext(); ((BasicExecutionManager)executionManager).beforeSubmitInSameThreadTask(null, task); - + SimpleFuture<T> future = new SimpleFuture<>(); + Throwable error = null; try { ((BasicExecutionManager)executionManager).afterSubmitRecordFuture(task, future); ((BasicExecutionManager)executionManager).beforeStartInSameThreadTask(null, task); @@ -194,12 +195,13 @@ public class BasicExecutionContext extends AbstractExecutionContext { return future.set(job.call()); } catch (Exception e) { + error = e; future.set(Maybe.absent(e)); throw Exceptions.propagate(e); } finally { try { - ((BasicExecutionManager)executionManager).afterEndInSameThreadTask(null, task); + ((BasicExecutionManager)executionManager).afterEndInSameThreadTask(null, task, error); } finally { BasicExecutionManager.getPerThreadCurrentTask().set(previousTask); perThreadExecutionContext.set(oldExecutionContext); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/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 22d5b23..0f666ba 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 @@ -46,12 +46,14 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import org.apache.brooklyn.api.entity.Entity; import org.apache.brooklyn.api.mgmt.ExecutionManager; import org.apache.brooklyn.api.mgmt.HasTaskChildren; 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.entity.Entities; import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; import org.apache.brooklyn.util.collections.MutableList; import org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode; @@ -256,8 +258,13 @@ public class BasicExecutionManager implements ExecutionManager { } Task<?> removed = tasksById.remove(task.getId()); incompleteTaskIds.remove(task.getId()); - if (removed!=null && removed.isSubmitted() && !removed.isDone()) { - 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?"); + if (removed!=null && removed.isSubmitted() && !removed.isDone(true)) { + Entity context = BrooklynTaskTags.getContextEntity(removed); + if (!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?"); + } } return removed != null; } @@ -406,17 +413,20 @@ public class BasicExecutionManager implements ExecutionManager { protected Task<?> submitNewScheduledTask(final Map<?,?> flags, final ScheduledTask task) { beforeSubmitScheduledTaskAllIterations(flags, task); - return submitSubsequentScheduledTask(flags, task); + if (!submitSubsequentScheduledTask(flags, task)) { + afterEndScheduledTaskAllIterations(flags, task, null); + } + return task; } - protected Task<?> submitSubsequentScheduledTask(final Map<?,?> flags, final ScheduledTask task) { + private boolean submitSubsequentScheduledTask(final Map<?,?> flags, final ScheduledTask task) { if (!task.isDone()) { task.internalFuture = delayedRunner.schedule(new ScheduledTaskCallable(task, flags), task.delay.toNanoseconds(), TimeUnit.NANOSECONDS); + return true; } else { - afterEndScheduledTaskAllIterations(flags, task); + return false; } - return task; } protected class ScheduledTaskCallable implements Callable<Object> { @@ -431,15 +441,24 @@ public class BasicExecutionManager implements ExecutionManager { @Override @SuppressWarnings({ "rawtypes", "unchecked" }) public Object call() { - if (task.startTimeUtc==-1) task.startTimeUtc = System.currentTimeMillis(); + if (task.startTimeUtc==-1) { + // this is overwritten on each run; not sure if that's best or not + task.startTimeUtc = System.currentTimeMillis(); + } TaskInternal<?> taskScheduled = null; + Throwable error = null; try { - beforeStartScheduledTaskSubmissionIteration(flags, task); taskScheduled = (TaskInternal<?>) task.newTask(); taskScheduled.setSubmittedByTask(task); + beforeStartScheduledTaskSubmissionIteration(flags, task, taskScheduled); final Callable<?> oldJob = taskScheduled.getJob(); final TaskInternal<?> taskScheduledF = taskScheduled; taskScheduled.setJob(new Callable() { @Override public Object call() { + if (task.isCancelled()) { + afterEndScheduledTaskAllIterations(flags, task, new CancellationException("cancel detected")); + throw new CancellationException("cancel detected"); // above throws, but for good measure + } + Throwable lastError = null; boolean shouldResubmit = true; task.recentRun = taskScheduledF; try { @@ -448,19 +467,22 @@ public class BasicExecutionManager implements ExecutionManager { } Object result; try { + lastError = null; result = oldJob.call(); task.lastThrownType = null; } catch (Exception e) { + lastError = e; shouldResubmit = shouldResubmitOnException(oldJob, e); throw Exceptions.propagate(e); } return result; } finally { // do in finally block in case we were interrupted - if (shouldResubmit) { - resubmit(); + if (shouldResubmit && resubmit()) { + // resubmitted fine, no-op } else { - afterEndScheduledTaskAllIterations(flags, task); + // not resubmitted, note ending + afterEndScheduledTaskAllIterations(flags, task, lastError); } } }}); @@ -468,16 +490,23 @@ public class BasicExecutionManager implements ExecutionManager { BasicExecutionContext ec = BasicExecutionContext.getCurrentExecutionContext(); if (ec!=null) return ec.submit(taskScheduled); else return submit(taskScheduled); + + } catch (Exception e) { + error = e; + throw Exceptions.propagate(e); + } finally { - afterEndScheduledTaskSubmissionIteration(flags, task, taskScheduled); + afterEndScheduledTaskSubmissionIteration(flags, task, taskScheduled, error); } } - private void resubmit() { + private boolean resubmit() { task.runCount++; if (task.period!=null && !task.isCancelled()) { task.delay = task.period; - submitSubsequentScheduledTask(flags, task); + return submitSubsequentScheduledTask(flags, task); + } else { + return false; } } @@ -520,47 +549,19 @@ public class BasicExecutionManager implements ExecutionManager { @Override public T call() { + T result = null; + Throwable error = null; try { - T result = null; - Throwable error = null; - try { - beforeStartAtomicTask(flags, task); - if (!task.isCancelled()) { - result = ((TaskInternal<T>)task).getJob().call(); - } else throw new CancellationException(); - } catch(Throwable e) { - error = e; - } finally { - afterEndAtomicTask(flags, task); - } - if (error!=null) { - /* we throw, after logging debug. - * the throw means the error is available for task submitters to monitor. - * however it is possible no one is monitoring it, in which case we will have debug logging only for errors. - * (the alternative, of warn-level logging in lots of places where we don't want it, seems worse!) - */ - if (log.isDebugEnabled()) { - // debug only here, because most submitters will handle failures - if (error instanceof InterruptedException || error instanceof RuntimeInterruptedException) { - log.debug("Detected interruption on task "+task+" (rethrowing)" + - (Strings.isNonBlank(error.getMessage()) ? ": "+error.getMessage() : "")); - } else if (error instanceof NullPointerException || - error instanceof IndexOutOfBoundsException || - error instanceof ClassCastException) { - log.debug("Exception running task "+task+" (rethrowing): "+error, error); - } 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); - } - return result; + beforeStartAtomicTask(flags, task); + if (!task.isCancelled()) { + result = ((TaskInternal<T>)task).getJob().call(); + } else throw new CancellationException(); + } catch(Throwable e) { + error = e; } finally { - ((TaskInternal<?>)task).runListeners(); + afterEndAtomicTask(flags, task, error); } + return result; } @Override @@ -634,26 +635,30 @@ public class BasicExecutionManager implements ExecutionManager { log.trace("On cancel of "+task+", applicable subtask count "+subtasksFound+", of which "+subtasksReallyCancelled+" were actively cancelled"); } } - - ((TaskInternal<?>)task).runListeners(); + + // no longer run listeners when we say to cancel, they get run when the task really ends + // TODO confirm no problems (added 2017-09) +// ((TaskInternal<?>)task).runListeners(); return result; } } - private final class SubmissionListenerToCallOtherListeners<T> implements Runnable { + private final class SubmissionListenerToCallManagerListeners<T> implements Runnable { private final Task<T> task; - private SubmissionListenerToCallOtherListeners(Task<T> task) { + private SubmissionListenerToCallManagerListeners(Task<T> task) { this.task = task; } @Override public void run() { - try { - ((TaskInternal<?>)task).runListeners(); - } catch (Exception e) { - log.warn("Error running task listeners for task "+task+" done", e); - } + // TODO remove after confirmation this is fine; this listener runs as one of the task's listeners + // so the task.runListeners will always be no-op +// try { +// ((TaskInternal<?>)task).runListeners(); +// } catch (Exception e) { +// log.warn("Error running task listeners for task "+task+" done", e); +// } for (ExecutionListener listener : listeners) { try { @@ -715,7 +720,7 @@ public class BasicExecutionManager implements ExecutionManager { // 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); + ((TaskInternal<T>)task).addListener(new SubmissionListenerToCallManagerListeners<T>(task), runner); // NB: can the above mean multiple callbacks to TaskInternal#runListeners? // finally expose the future to callers @@ -735,10 +740,12 @@ public class BasicExecutionManager implements ExecutionManager { protected void internalBeforeSubmit(Map<?,?> flags, Task<?> task) { incompleteTaskIds.add(task.getId()); - Task<?> currentTask = Tasks.current(); - if (currentTask!=null) ((TaskInternal<?>)task).setSubmittedByTask( - // do this instead of soft reference (2017-09) as soft refs impact GC - Maybe.of(new TaskLookup(this, currentTask)) ); + if (task.getSubmittedByTaskId()==null) { + Task<?> currentTask = Tasks.current(); + if (currentTask!=null) ((TaskInternal<?>)task).setSubmittedByTask( + // do this instead of soft reference (2017-09) as soft refs impact GC + Maybe.of(new TaskLookup(this, currentTask)) ); + } ((TaskInternal<?>)task).setSubmitTimeUtc(System.currentTimeMillis()); if (flags!=null && flags.get("tag")!=null) ((TaskInternal<?>)task).getMutableTags().add(flags.remove("tag")); @@ -761,6 +768,9 @@ public class BasicExecutionManager implements ExecutionManager { public TaskLookup(BasicExecutionManager mgr, Task<?> t) { this.mgr = mgr; id = t.getId(); + if (mgr.getTask(id)==null) { + log.warn("Created task lookup for task which is not registered: "+t); + } displayName = t.getDisplayName(); } @Override @@ -771,16 +781,16 @@ public class BasicExecutionManager implements ExecutionManager { return gone(); } private <T> Task<T> gone() { - Task<T> t = Tasks.<T>builder().dynamic(false).displayName(displayName) - .description("Details of the original task "+id+" have been forgotten.") + Task<T> t = Tasks.<T>builder().dynamic(false).displayName(displayName+" (placeholder for "+id+")") + .description("Details of the original task have been forgotten.") .body(Callables.returning((T)null)).build(); ((BasicTask<T>)t).ignoreIfNotRun(); return t; } } - protected void beforeStartScheduledTaskSubmissionIteration(Map<?,?> flags, Task<?> task) { - internalBeforeStart(flags, task, true); + protected void beforeStartScheduledTaskSubmissionIteration(Map<?,?> flags, Task<?> taskRepeatedlyScheduling, Task<?> taskIteration) { + internalBeforeStart(flags, taskRepeatedlyScheduling, true); } protected void beforeStartAtomicTask(Map<?,?> flags, Task<?> task) { internalBeforeStart(flags, task, true); @@ -856,54 +866,87 @@ public class BasicExecutionManager implements ExecutionManager { private static boolean loggedClosureDeprecatedInInvokeCallback; /** 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); + protected void afterEndScheduledTaskAllIterations(Map<?,?> flags, Task<?> taskRepeatedlyScheduling, Throwable error) { + internalAfterEnd(flags, taskRepeatedlyScheduling, false, true, error); } /** called once for each call to {@link #beforeStartScheduledTaskSubmissionIteration(Map, Task)}, * with a per-iteration task generated by the surrounding scheduled task */ - protected void afterEndScheduledTaskSubmissionIteration(Map<?,?> flags, Task<?> scheduledTask, Task<?> taskIteration) { - internalAfterEnd(flags, scheduledTask, true, false); + protected void afterEndScheduledTaskSubmissionIteration(Map<?,?> flags, Task<?> taskRepeatedlyScheduling, Task<?> taskIteration, Throwable error) { + internalAfterEnd(flags, taskRepeatedlyScheduling, true, false, error); } /** called once for each task on which {@link #beforeStartAtomicTask(Map, Task)} is invoked, * and normally (if not interrupted prior to start) * called once for each task on which {@link #beforeSubmitAtomicTask(Map, Task)} */ - protected void afterEndAtomicTask(Map<?,?> flags, Task<?> task) { - internalAfterEnd(flags, task, true, true); + protected void afterEndAtomicTask(Map<?,?> flags, Task<?> task, Throwable error) { + internalAfterEnd(flags, task, true, true, error); } - protected void afterEndInSameThreadTask(Map<?,?> flags, Task<?> task) { - internalAfterEnd(flags, task, true, true); + protected void afterEndInSameThreadTask(Map<?,?> flags, Task<?> task, Throwable error) { + internalAfterEnd(flags, task, true, true, error); } /** 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 */ - protected void internalAfterEnd(Map<?,?> flags, Task<?> task, boolean startedInThisThread, boolean isEndingAllIterations) { - if (log.isTraceEnabled()) log.trace(this+" afterEnd, task: "+task); - if (startedInThisThread) { - activeTaskCount.decrementAndGet(); - } - if (isEndingAllIterations) { - incompleteTaskIds.remove(task.getId()); - if (flags!=null) { - invokeCallback(flags.get("newTaskEndCallback"), task); + protected void internalAfterEnd(Map<?,?> flags, Task<?> task, boolean startedInThisThread, boolean isEndingAllIterations, Throwable error) { + try { + if (log.isTraceEnabled()) log.trace(this+" afterEnd, task: "+task); + if (startedInThisThread) { + activeTaskCount.decrementAndGet(); } - ((TaskInternal<?>)task).setEndTimeUtc(System.currentTimeMillis()); - } - - if (startedInThisThread) { - PerThreadCurrentTaskHolder.perThreadCurrentTask.remove(); - //clear thread _after_ endTime set, so we won't get a null thread when there is no end-time - if (RENAME_THREADS) { - Thread thread = task.getThread(); - if (thread==null) { - log.warn("BasicTask.afterEnd invoked without corresponding beforeStart"); - } else { - thread.setName(threadOriginalName.get()); - threadOriginalName.remove(); + if (isEndingAllIterations) { + incompleteTaskIds.remove(task.getId()); + if (flags!=null) { + invokeCallback(flags.get("newTaskEndCallback"), task); + } + ((TaskInternal<?>)task).setEndTimeUtc(System.currentTimeMillis()); + } + + if (startedInThisThread) { + PerThreadCurrentTaskHolder.perThreadCurrentTask.remove(); + //clear thread _after_ endTime set, so we won't get a null thread when there is no end-time + if (RENAME_THREADS) { + Thread thread = task.getThread(); + if (thread==null) { + log.warn("BasicTask.afterEnd invoked without corresponding beforeStart"); + } else { + thread.setName(threadOriginalName.get()); + threadOriginalName.remove(); + } + } + ((TaskInternal<?>)task).setThread(null); + } + } finally { + try { + if (error!=null) { + /* we throw, after logging debug. + * the throw means the error is available for task submitters to monitor. + * however it is possible no one is monitoring it, in which case we will have debug logging only for errors. + * (the alternative, of warn-level logging in lots of places where we don't want it, seems worse!) + */ + if (log.isDebugEnabled()) { + // debug only here, because most submitters will handle failures + if (error instanceof InterruptedException || error instanceof RuntimeInterruptedException) { + log.debug("Detected interruption on task "+task+" (rethrowing)" + + (Strings.isNonBlank(error.getMessage()) ? ": "+error.getMessage() : "")); + } else if (error instanceof NullPointerException || + error instanceof IndexOutOfBoundsException || + error instanceof ClassCastException) { + log.debug("Exception running task "+task+" (rethrowing): "+error, error); + } 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); + } + } finally { + synchronized (task) { task.notifyAll(); } + if (isEndingAllIterations) { + ((TaskInternal<?>)task).runListeners(); } } - ((TaskInternal<?>)task).setThread(null); } - synchronized (task) { task.notifyAll(); } } public TaskScheduler getTaskSchedulerForTag(Object tag) { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java index f908ad4..3c9b163 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/BasicTask.java @@ -46,7 +46,6 @@ import org.apache.brooklyn.api.mgmt.Task; import org.apache.brooklyn.api.objs.Identifiable; import org.apache.brooklyn.util.JavaGroovyEquivalents; import org.apache.brooklyn.util.exceptions.Exceptions; -import org.apache.brooklyn.util.groovy.GroovyJavaMethods; import org.apache.brooklyn.util.guava.Maybe; import org.apache.brooklyn.util.guava.Maybe.MaybeSupplier; import org.apache.brooklyn.util.text.Identifiers; @@ -142,18 +141,6 @@ public class BasicTask<T> implements TaskInternal<T> { this(flags, JavaGroovyEquivalents.toCallable(job)); } - /** - * @deprecated since 0.11.0; explicit groovy utilities/support will be deleted. - */ - @Deprecated - public BasicTask(Closure<T> job) { this(GroovyJavaMethods.callableFromClosure(job)); } - - /** - * @deprecated since 0.11.0; explicit groovy utilities/support will be deleted. - */ - @Deprecated - public BasicTask(Map<?,?> flags, Closure<T> job) { this(flags, GroovyJavaMethods.callableFromClosure(job)); } - @Override public String getId() { return id; @@ -358,11 +345,19 @@ public class BasicTask<T> implements TaskInternal<T> { } @Override + public boolean isDone(boolean andTaskNotRunning) { + if (!cancelled && !(internalFuture!=null && internalFuture.isDone()) && endTimeUtc<=0) { + return false; + } + if (andTaskNotRunning && cancelled && isBegun() && endTimeUtc<=0) { + return false; + } + return true; + } + + @Override public boolean isDone() { - // if endTime is set, result might not be completed yet, but it will be set very soon - // (the two values are set close in time, result right after the endTime; - // but callback hooks might not see the result yet) - return cancelled || (internalFuture!=null && internalFuture.isDone()) || endTimeUtc>0; + return isDone(false); } /** @@ -415,7 +410,6 @@ public class BasicTask<T> implements TaskInternal<T> { @Override public synchronized boolean blockUntilStarted(Duration timeout) { Long endTime = timeout==null ? null : System.currentTimeMillis() + timeout.toMillisecondsRoundingUp(); - int count = 0; while (true) { if (cancelled) throw new CancellationException(); if (startTimeUtc>0) return true; http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java index 3cbb548..c7f2d69 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/CompoundTask.java @@ -89,10 +89,6 @@ public abstract class CompoundTask<T> extends BasicTask<List<T>> implements HasT for (Object job : jobs) { Task subtask; if (job instanceof TaskAdaptable) { subtask = ((TaskAdaptable)job).asTask(); } - else if (job instanceof Closure) { - log.warn("Use of groovy.lang.Closure is deprecated, in CompoundTask jobs"); - subtask = new BasicTask<T>((Closure) job); - } else if (job instanceof Callable) { subtask = new BasicTask<T>((Callable) job); } else if (job instanceof Runnable) { subtask = new BasicTask<T>((Runnable) job); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java index 5fb8ee8..362fd5d 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/DynamicSequentialTask.java @@ -173,6 +173,7 @@ public class DynamicSequentialTask<T> extends BasicTask<T> implements HasTaskChi @Override public boolean cancel(TaskCancellationMode mode) { + // super is invoked in the method below return cancel(mode, null); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/util/core/task/ForwardingTask.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/ForwardingTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/ForwardingTask.java index 83f2128..81bd972 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/ForwardingTask.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/ForwardingTask.java @@ -72,6 +72,11 @@ public abstract class ForwardingTask<T> extends ForwardingObject implements Task public boolean isDone() { return delegate().isDone(); } + + @Override + public boolean isDone(boolean andTaskNotRunning) { + return delegate().isDone(andTaskNotRunning); + } @Override public Task<T> asTask() { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java b/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java index c253916..89eb6ab 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/ScheduledTask.java @@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit; import org.apache.brooklyn.api.mgmt.Task; import org.apache.brooklyn.util.collections.MutableMap; +import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.time.Duration; import com.google.common.annotations.Beta; @@ -178,7 +179,10 @@ public class ScheduledTask extends BasicTask<Object> { } @Override - public boolean isDone() { + public boolean isDone(boolean andTaskNoLongerRunning) { + if (andTaskNoLongerRunning) { + return super.isDone(true); + } return isCancelled() || (maxIterations!=null && maxIterations <= runCount) || (period==null && nextRun!=null && nextRun.isDone()); } @@ -214,6 +218,14 @@ public class ScheduledTask extends BasicTask<Object> { protected boolean doCancel(org.apache.brooklyn.util.core.task.TaskInternal.TaskCancellationMode mode) { if (nextRun!=null) { ((TaskInternal<?>)nextRun).cancel(mode); + try { + ((TaskInternal<?>)nextRun).getJob().call(); + nextRun = null; + } catch (CancellationException e) { + // expected, ignore + } catch (Exception e) { + throw Exceptions.propagateAnnotated("Error cancelling scheduled task "+this, e); + } } return super.doCancel(mode); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java b/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java index 6b3365b..c49544a 100644 --- a/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java +++ b/core/src/main/java/org/apache/brooklyn/util/core/task/TaskInternal.java @@ -97,8 +97,10 @@ 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)}. */ + /** On task completion runs the listeners which have been registered using + * {@link #addListener(Runnable, java.util.concurrent.Executor)}. + * <p> + * Doeas not run immediately on cancellation, it waits for the task to recognize it is cancelled. */ void runListeners(); void setEndTimeUtc(long val); @@ -126,7 +128,7 @@ public interface TaskInternal<T> extends Task<T> { * this returns the "real" task represented by this one */ Task<?> getProxyTarget(); - /** clearer semantics around cancellation; may be promoted to {@link Task} if we */ + /** clearer semantics around cancellation; may be promoted to {@link Task} */ @Beta public boolean cancel(TaskCancellationMode mode); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/934cf4cc/core/src/test/java/org/apache/brooklyn/core/mgmt/rebind/RebindFeedTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/core/mgmt/rebind/RebindFeedTest.java b/core/src/test/java/org/apache/brooklyn/core/mgmt/rebind/RebindFeedTest.java index 1503bfa..68e6f78 100644 --- a/core/src/test/java/org/apache/brooklyn/core/mgmt/rebind/RebindFeedTest.java +++ b/core/src/test/java/org/apache/brooklyn/core/mgmt/rebind/RebindFeedTest.java @@ -44,6 +44,8 @@ import org.apache.brooklyn.feed.http.HttpValueFunctions; import org.apache.brooklyn.feed.ssh.SshFeed; import org.apache.brooklyn.feed.ssh.SshPollConfig; import org.apache.brooklyn.feed.ssh.SshValueFunctions; +import org.apache.brooklyn.location.localhost.LocalhostMachineProvisioningLocation; +import org.apache.brooklyn.location.ssh.SshMachineLocation; import org.apache.brooklyn.test.Asserts; import org.apache.brooklyn.util.collections.MutableList; import org.apache.brooklyn.util.core.http.BetterMockWebServer; @@ -58,8 +60,6 @@ import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; -import org.apache.brooklyn.location.localhost.LocalhostMachineProvisioningLocation; -import org.apache.brooklyn.location.ssh.SshMachineLocation; import com.google.common.base.Function; import com.google.common.base.Predicates;
