add support for same-thread execution submitting and blocking is unnecessary overhead in many places
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/8983bf2c Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/8983bf2c Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/8983bf2c Branch: refs/heads/master Commit: 8983bf2cdae3e1b3bbc00d12cfb4baf09fb732fe Parents: 4c2468d Author: Alex Heneveld <[email protected]> Authored: Mon Sep 11 23:15:50 2017 +0100 Committer: Alex Heneveld <[email protected]> Committed: Fri Sep 15 10:29:09 2017 +0100 ---------------------------------------------------------------------- .../brooklyn/api/mgmt/ExecutionContext.java | 24 ++- .../util/core/task/BasicExecutionContext.java | 156 +++++++++++++++---- .../util/core/task/BasicExecutionManager.java | 8 +- .../brooklyn/util/core/task/BasicTask.java | 9 +- .../brooklyn/util/core/task/ValueResolver.java | 8 +- .../brooklyn/core/entity/EntityConfigTest.java | 12 ++ .../internal/EntityExecutionManagerTest.java | 1 - 7 files changed, 174 insertions(+), 44 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java ---------------------------------------------------------------------- diff --git a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java index 344907a..142e664 100644 --- a/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java +++ b/api/src/main/java/org/apache/brooklyn/api/mgmt/ExecutionContext.java @@ -27,6 +27,7 @@ import org.apache.brooklyn.api.entity.Entity; import org.apache.brooklyn.util.guava.Maybe; import com.google.common.annotations.Beta; +import com.google.common.base.Supplier; /** * This is a Brooklyn extension to the Java {@link Executor}. @@ -71,14 +72,29 @@ public interface ExecutionContext extends Executor { * Gets the value promptly, or returns {@link Maybe#absent()} if the value is not yet available. * It may throw an error if it cannot be determined whether a value is available immediately or not. * <p> - * Implementations will typically attempt to execute in the current thread, with appropriate - * tricks to make it look like it is in a sub-thread, and will attempt to be non-blocking but - * if needed they may block. + * Implementations will typically act like {@link #get(TaskAdaptable)} with additional + * tricks to attempt to be non-blocking, such as recognizing some "immediate" markers. * <p> - * Supports {@link Callable} and {@link Runnable} and some {@link Task} targets to be evaluated with "immediate" semantics. + * Also supports {@link Callable}, {@link Runnable}, and {@link Supplier} argument types. */ // TODO reference ImmediateSupplier when that class is moved to utils project @Beta <T> Maybe<T> getImmediately(Object callableOrSupplierOrTask); + /** + * Efficient shortcut for {@link #submit(TaskAdaptable)} followed by an immediate {@link Task#get()}. + * <p> + * Implementations will typically attempt to execute in the current thread, with appropriate + * configuration to make it look like it is in a sub-thread, + * ie registering this as a task and allowing + * context methods on tasks to see the given sub-task. + * <p> + * If the argument has already been submitted it simply blocks on it. + * + * @param task + * @return result of the task execution + */ + @Beta + <T> T get(TaskAdaptable<T> task); + } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/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 e0dd8b0..8166dc0 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 @@ -26,8 +26,12 @@ import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.brooklyn.api.entity.Entity; import org.apache.brooklyn.api.mgmt.ExecutionContext; @@ -42,7 +46,10 @@ import org.apache.brooklyn.core.mgmt.BrooklynTaskTags.WrappedEntity; import org.apache.brooklyn.core.mgmt.entitlement.Entitlements; import org.apache.brooklyn.util.collections.MutableMap; import org.apache.brooklyn.util.core.task.ImmediateSupplier.ImmediateUnsupportedException; +import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.guava.Maybe; +import org.apache.brooklyn.util.time.CountdownTimer; +import org.apache.brooklyn.util.time.Duration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -100,6 +107,106 @@ public class BasicExecutionContext extends AbstractExecutionContext { @Override public Set<Task<?>> getTasks() { return executionManager.getTasksWithAllTags(tags); } + @Override + public <T> T get(TaskAdaptable<T> task) { + 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 runInSameThread(t, new Callable<Maybe<T>>() { + public Maybe<T> call() { + try { + return Maybe.of(t.getJob().call()); + } catch (Exception e) { + Exceptions.propagateIfFatal(e); + return Maybe.absent(e); + } + } + }).get(); + } + + private static class SimpleFuture<T> implements Future<T> { + boolean cancelled = false; + boolean done = false; + Maybe<T> result; + + public synchronized Maybe<T> set(Maybe<T> result) { + this.result = result; + done = true; + notifyAll(); + return result; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + cancelled = true; + return true; + } + + @Override + public boolean isCancelled() { + return cancelled; + } + + @Override + public boolean isDone() { + return done; + } + + @Override + public T get() throws InterruptedException, ExecutionException { + return result.get(); + } + + @Override + public synchronized T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + if (isDone()) return get(); + CountdownTimer time = CountdownTimer.newInstanceStarted( Duration.of(timeout, unit) ); + while (!time.isExpired()) { + wait(time.getDurationRemaining().lowerBound(Duration.ONE_MILLISECOND).toMilliseconds()); + if (isDone()) return get(); + } + throw new TimeoutException(); + } + } + + private <T> Maybe<T> runInSameThread(final Task<T> task, Callable<Maybe<T>> job) { + ((TaskInternal<T>)task).getMutableTags().addAll(tags); + + Task<?> previousTask = BasicExecutionManager.getPerThreadCurrentTask().get(); + BasicExecutionContext oldExecutionContext = getCurrentExecutionContext(); + registerPerThreadExecutionContext(); + ((BasicExecutionManager)executionManager).beforeSubmitInSameThreadTask(null, task); + + SimpleFuture<T> future = new SimpleFuture<>(); + try { + ((BasicExecutionManager)executionManager).afterSubmitRecordFuture(task, future); + ((BasicExecutionManager)executionManager).beforeStartInSameThreadTask(null, task); + + // TODO this does not apply the same context-switching logic as submit + + return future.set(job.call()); + + } catch (Exception e) { + future.set(Maybe.absent(e)); + throw Exceptions.propagate(e); + + } finally { + try { + ((BasicExecutionManager)executionManager).afterEndInSameThreadTask(null, task); + } finally { + BasicExecutionManager.getPerThreadCurrentTask().set(previousTask); + perThreadExecutionContext.set(oldExecutionContext); + } + } + } + /** performs execution without spawning a new task thread, though it does temporarily set a fake task for the purpose of getting context; * currently supports {@link Supplier}, {@link Callable}, {@link Runnable}, or {@link Task} instances; * with tasks if it is submitted or in progress, @@ -110,9 +217,9 @@ public class BasicExecutionContext extends AbstractExecutionContext { @SuppressWarnings("unchecked") @Override public <T> Maybe<T> getImmediately(Object callableOrSupplier) { - BasicTask<?> fakeTaskForContext; + BasicTask<T> fakeTaskForContext; if (callableOrSupplier instanceof BasicTask) { - fakeTaskForContext = (BasicTask<?>)callableOrSupplier; + fakeTaskForContext = (BasicTask<T>)callableOrSupplier; if (fakeTaskForContext.isQueuedOrSubmitted()) { if (fakeTaskForContext.isDone()) { return Maybe.of((T)fakeTaskForContext.getUnchecked()); @@ -122,41 +229,26 @@ public class BasicExecutionContext extends AbstractExecutionContext { } callableOrSupplier = fakeTaskForContext.getJob(); } else { - fakeTaskForContext = new BasicTask<Object>(MutableMap.of("displayName", "immediate evaluation")); + fakeTaskForContext = new BasicTask<T>(MutableMap.of("displayName", "immediate evaluation")); } - fakeTaskForContext.tags.addAll(tags); + final ImmediateSupplier<T> job = callableOrSupplier instanceof ImmediateSupplier ? (ImmediateSupplier<T>) callableOrSupplier + : InterruptingImmediateSupplier.<T>of(callableOrSupplier); fakeTaskForContext.tags.add(BrooklynTaskTags.IMMEDIATE_TASK_TAG); fakeTaskForContext.tags.add(BrooklynTaskTags.TRANSIENT_TASK_TAG); - - Task<?> previousTask = BasicExecutionManager.getPerThreadCurrentTask().get(); - BasicExecutionContext oldExecutionContext = getCurrentExecutionContext(); - registerPerThreadExecutionContext(); - ((BasicExecutionManager)executionManager).beforeSubmitInSameThreadTask(null, fakeTaskForContext); - try { - ((BasicExecutionManager)executionManager).beforeStartInSameThreadTask(null, fakeTaskForContext); - fakeTaskForContext.cancel(); - - if (!(callableOrSupplier instanceof ImmediateSupplier)) { - callableOrSupplier = InterruptingImmediateSupplier.of(callableOrSupplier); - } - boolean wasAlreadyInterrupted = Thread.interrupted(); - try { - return ((ImmediateSupplier<T>)callableOrSupplier).getImmediately(); - } finally { - if (wasAlreadyInterrupted) { - Thread.currentThread().interrupt(); + return runInSameThread(fakeTaskForContext, new Callable<Maybe<T>>() { + public Maybe<T> call() { + fakeTaskForContext.cancel(); + + boolean wasAlreadyInterrupted = Thread.interrupted(); + try { + return job.getImmediately(); + } finally { + if (wasAlreadyInterrupted) { + Thread.currentThread().interrupt(); + } } - } - - } finally { - try { - ((BasicExecutionManager)executionManager).afterEndInSameThreadTask(null, fakeTaskForContext); - } finally { - BasicExecutionManager.getPerThreadCurrentTask().set(previousTask); - perThreadExecutionContext.set(oldExecutionContext); - } - } + } }); } @SuppressWarnings({ "unchecked", "rawtypes" }) http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/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 afc8c99..22d5b23 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 @@ -703,6 +703,12 @@ public class BasicExecutionManager implements ExecutionManager { } else { future = runner.submit(job); } + afterSubmitRecordFuture(task, future); + + return task; + } + + protected <T> void afterSubmitRecordFuture(final Task<T> task, Future<T> future) { // 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); @@ -714,8 +720,6 @@ public class BasicExecutionManager implements ExecutionManager { // finally expose the future to callers ((TaskInternal<T>)task).initInternalFuture(listenableFuture); - - return task; } protected void beforeSubmitScheduledTaskAllIterations(Map<?,?> flags, Task<?> task) { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/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 f1a066f..f908ad4 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 @@ -412,17 +412,20 @@ public class BasicTask<T> implements TaskInternal<T> { blockUntilStarted(null); } - // TODO: This should log a message if timeout is null and the method blocks for an unreasonably long time - - // it probably means someone called .get() and forgot to submit the task. @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; if (internalFuture==null) try { if (timeout==null) { - wait(); + // 5s so that it will repeat in case something sets the future without notifying; + // can of course repeat here forever if someone calls get on an unsubmitted task, + // but that is not hard to discover with a thread dump, seeing it waiting here + wait(5*1000); } else { long remaining = endTime - System.currentTimeMillis(); if (remaining>0) http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/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 afd6109..d93dfa4 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,11 @@ 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); } - exec.submit(task); + if (timer!=null || Thread.currentThread().isInterrupted()) { + exec.submit(task); + } else { + exec.get(task); + } } } @@ -483,7 +487,7 @@ public class ValueResolver<T> implements DeferredSupplier<T>, Iterable<Maybe<Obj String description = getDescription(); TaskBuilder<Object> tb = Tasks.<Object>builder() .body(callable) - .displayName("Resolving dependent value") + .displayName("Resolving dependent value of deferred supplier") .description(description); if (isTransientTask) tb.tag(BrooklynTaskTags.TRANSIENT_TASK_TAG); http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java b/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java index b583b1e..b510178 100644 --- a/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java +++ b/core/src/test/java/org/apache/brooklyn/core/entity/EntityConfigTest.java @@ -697,6 +697,18 @@ public class EntityConfigTest extends BrooklynAppUnitTestSupport { assertEquals(getConfigFuture.get(TIMEOUT_MS, TimeUnit.MILLISECONDS), "abc"); assertEquals(work.callCount.get(), 1); } + + @Test + public void testGetConfigRunsInSameThread() throws Exception { + Callable<String> job = () -> ""+Thread.currentThread().getId(); + + final MyOtherEntity entity = app.addChild(EntitySpec.create(MyOtherEntity.class)); + assertEquals( ((EntityInternal)entity).getExecutionContext().get(new BasicTask<>(job)), job.call() ); + + // and config also runs in same thread (after validation) + entity.config().set(MyOtherEntity.STRING_KEY, new BasicTask<>(job)); + assertEquals(entity.config().get(MyOtherEntity.STRING_KEY), job.call()); + } @ImplementedBy(MyBaseEntityImpl.class) public interface MyBaseEntity extends EntityInternal { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/8983bf2c/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/EntityExecutionManagerTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/EntityExecutionManagerTest.java b/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/EntityExecutionManagerTest.java index c84e016..51f5bdc 100644 --- a/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/EntityExecutionManagerTest.java +++ b/core/src/test/java/org/apache/brooklyn/core/mgmt/internal/EntityExecutionManagerTest.java @@ -521,5 +521,4 @@ public class EntityExecutionManagerTest extends BrooklynAppUnitTestSupport { } return ((EntityInternal)e).getExecutionContext().submit(tb.build()); } - }
