share code for inferring execution context, and ensure EC is set in subscriptions
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/95c5b1e7 Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/95c5b1e7 Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/95c5b1e7 Branch: refs/heads/master Commit: 95c5b1e78fba24e37216e99c209568a5e3cf2fec Parents: e797cbe Author: Alex Heneveld <[email protected]> Authored: Thu Nov 2 15:20:17 2017 +0000 Committer: Alex Heneveld <[email protected]> Committed: Thu Nov 2 15:36:16 2017 +0000 ---------------------------------------------------------------------- .../spi/dsl/BrooklynDslDeferredSupplier.java | 5 +- .../brooklyn/spi/dsl/methods/DslComponent.java | 6 +-- .../brooklyn/core/mgmt/BrooklynTaskTags.java | 49 ++++++++++++++++++++ .../mgmt/internal/LocalSubscriptionManager.java | 7 +++ .../util/core/task/BasicExecutionContext.java | 12 ++++- .../customize/TemplateOptionsOption.java | 10 +--- 6 files changed, 70 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java ---------------------------------------------------------------------- diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java index 835f5d4..3b31637 100644 --- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java +++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/BrooklynDslDeferredSupplier.java @@ -100,10 +100,7 @@ public abstract class BrooklynDslDeferredSupplier<T> implements DeferredSupplier if (log.isDebugEnabled()) log.debug("Queuing task to resolve "+dsl+", called by "+Tasks.current()); - EntityInternal entity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current()); - ExecutionContext exec = - (entity != null) ? entity.getExecutionContext() - : BasicExecutionContext.getCurrentExecutionContext(); + ExecutionContext exec = BrooklynTaskTags.getCurrentExecutionContext(); if (exec == null) { throw new IllegalStateException("No execution context available to resolve " + dsl); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java ---------------------------------------------------------------------- diff --git a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java index 9ecad76..9dc5ed1 100644 --- a/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java +++ b/camp/camp-brooklyn/src/main/java/org/apache/brooklyn/camp/brooklyn/spi/dsl/methods/DslComponent.java @@ -46,7 +46,6 @@ import org.apache.brooklyn.core.sensor.DependentConfiguration; import org.apache.brooklyn.core.sensor.Sensors; import org.apache.brooklyn.util.JavaGroovyEquivalents; import org.apache.brooklyn.util.core.flags.TypeCoercions; -import org.apache.brooklyn.util.core.task.BasicExecutionContext; import org.apache.brooklyn.util.core.task.DeferredSupplier; import org.apache.brooklyn.util.core.task.ImmediateSupplier; import org.apache.brooklyn.util.core.task.TaskBuilder; @@ -339,10 +338,7 @@ public class DslComponent extends BrooklynDslDeferredSupplier<Entity> implements } static ExecutionContext findExecutionContext(Object callerContext) { - EntityInternal contextEntity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current()); - ExecutionContext execContext = - (contextEntity != null) ? contextEntity.getExecutionContext() - : BasicExecutionContext.getCurrentExecutionContext(); + ExecutionContext execContext = BrooklynTaskTags.getCurrentExecutionContext(); if (execContext == null) { throw new IllegalStateException("No execution context available to resolve " + callerContext); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java index f8a9d0b..b9dfbc1 100644 --- a/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java +++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/BrooklynTaskTags.java @@ -30,13 +30,17 @@ import javax.annotation.Nullable; import org.apache.brooklyn.api.effector.Effector; import org.apache.brooklyn.api.entity.Entity; +import org.apache.brooklyn.api.mgmt.ExecutionContext; import org.apache.brooklyn.api.mgmt.ExecutionManager; import org.apache.brooklyn.api.mgmt.ManagementContext; import org.apache.brooklyn.api.mgmt.Task; import org.apache.brooklyn.api.mgmt.entitlement.EntitlementContext; import org.apache.brooklyn.api.objs.EntityAdjunct; +import org.apache.brooklyn.core.entity.EntityInternal; import org.apache.brooklyn.core.mgmt.internal.AbstractManagementContext; +import org.apache.brooklyn.core.objs.AbstractEntityAdjunct; import org.apache.brooklyn.util.core.config.ConfigBag; +import org.apache.brooklyn.util.core.task.BasicExecutionContext; import org.apache.brooklyn.util.core.task.DynamicTasks; import org.apache.brooklyn.util.core.task.TaskTags; import org.apache.brooklyn.util.core.task.Tasks; @@ -183,11 +187,25 @@ public class BrooklynTaskTags extends TaskTags { return getWrappedEntityTagOfType( getTagsFast(t), wrappingType); } public static WrappedEntity getWrappedEntityTagOfType(Collection<?> tags, String wrappingType) { + if (tags==null) return null; for (Object x: tags) if ((x instanceof WrappedEntity) && ((WrappedEntity)x).wrappingType.equals(wrappingType)) return (WrappedEntity)x; return null; } + + @SuppressWarnings("unchecked") + public static <T> WrappedObject<T> getWrappedObjectTagOfType(Collection<?> tags, String wrappingType, Class<T> type) { + if (tags==null) return null; + for (Object x: tags) + if ((x instanceof WrappedObject) && ((WrappedObject<?>)x).wrappingType.equals(wrappingType) && type.isInstance( ((WrappedObject<?>)x).object )) + return (WrappedObject<T>)x; + return null; + } + public static <T> T getUnwrappedObjectTagOfType(Collection<?> tags, String wrappingType, Class<T> type) { + WrappedObject<T> result = getWrappedObjectTagOfType(tags, wrappingType, type); + return result!=null ? result.unwrap() : null; + } public static Entity getWrappedEntityOfType(Task<?> t, String wrappingType) { WrappedEntity wrapper = getWrappedEntityTagOfType(t, wrappingType); @@ -540,5 +558,36 @@ public class BrooklynTaskTags extends TaskTags { tag.entitlementContext = context; return tag; } + + public static ExecutionContext getExecutionContext(Collection<?> tags) { + EntityAdjunct ea = getUnwrappedObjectTagOfType(tags, CONTEXT_ADJUNCT, EntityAdjunct.class); + if (ea instanceof AbstractEntityAdjunct) { + return ((AbstractEntityAdjunct)ea).getExecutionContext(); + } + Entity e = getWrappedEntityOfType(tags, CONTEXT_ENTITY); + if (e==null) e= getWrappedEntityOfType(tags, TARGET_ENTITY); + if (e instanceof EntityInternal) { + return ((EntityInternal)e).getExecutionContext(); + } + + return null; + } + public static ExecutionContext getExecutionContext(Task<?> t) { + return getExecutionContext(getTagsFast(t)); + } + /** + * This should always be identical to {@link BasicExecutionContext#getCurrentExecutionContext()} as + * there are two ways to derive the {@link ExecutionContext}: one is the {@link ThreadLocal} accessed from the above method + * and set by {@link BasicExecutionContext} submissions; the other is by looking at {@link #getTargetOrContextEntity(Task)} + * and {@link #CONTEXT_ADJUNCT} tags on the current task. As {@link BasicExecutionContext} also sets those tags, + * the only time the two will be different is if {@link ExecutionManager#submit(org.apache.brooklyn.api.mgmt.TaskAdaptable)} + * is used supplying one or more of these context tags, in which case only this method will determine the execution context. + */ + public static ExecutionContext getCurrentExecutionContext() { + Task<?> t = Tasks.current(); + ExecutionContext result = t!=null ? getExecutionContext(t) : null; + if (result==null) result = BasicExecutionContext.getCurrentExecutionContext(); + return result; + } } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java index 623fb26..a5fed0e 100644 --- a/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java +++ b/core/src/main/java/org/apache/brooklyn/core/mgmt/internal/LocalSubscriptionManager.java @@ -34,6 +34,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; import org.apache.brooklyn.api.entity.Entity; +import org.apache.brooklyn.api.mgmt.ExecutionContext; import org.apache.brooklyn.api.mgmt.ExecutionManager; import org.apache.brooklyn.api.mgmt.SubscriptionHandle; import org.apache.brooklyn.api.mgmt.SubscriptionManager; @@ -48,6 +49,7 @@ import org.apache.brooklyn.core.sensor.AttributeMap; import org.apache.brooklyn.core.sensor.BasicSensorEvent; import org.apache.brooklyn.util.collections.MutableList; import org.apache.brooklyn.util.collections.MutableMap; +import org.apache.brooklyn.util.core.task.BasicExecutionContext; import org.apache.brooklyn.util.core.task.BasicExecutionManager; import org.apache.brooklyn.util.core.task.SingleThreadedScheduler; import org.apache.brooklyn.util.text.Identifiers; @@ -311,6 +313,7 @@ public class LocalSubscriptionManager extends AbstractSubscriptionManager { boolean isEntityStarting = s.subscriber instanceof Entity && isInitialPublicationOfOldValueInCorrectScheduledThread; // will have entity (and adjunct) execution context from tags, so can skip getting exec context + final ExecutionContext ec = BrooklynTaskTags.getExecutionContext(tags); Runnable deliverer = new Runnable() { @Override public String toString() { @@ -322,7 +325,9 @@ public class LocalSubscriptionManager extends AbstractSubscriptionManager { } @Override public void run() { + BasicExecutionContext oldEC = ec instanceof BasicExecutionContext ? BasicExecutionContext.setPerThreadExecutionContext((BasicExecutionContext)ec) : null; try { + if (isEntityStarting) { /* don't let sub deliveries start until this is completed; * this is a pragmatic way to ensure the publish events @@ -349,6 +354,8 @@ public class LocalSubscriptionManager extends AbstractSubscriptionManager { } else { LOG.warn("Error processing subscriptions to "+this+": "+t, t); } + } finally { + BasicExecutionContext.setPerThreadExecutionContext(oldEC); } }}; if (!isInitialPublicationOfOldValueInCorrectScheduledThread) { http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/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 f81ad35..8b37e2f 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 @@ -56,6 +56,7 @@ import org.apache.brooklyn.util.time.Duration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.annotations.Beta; import com.google.common.base.Function; import com.google.common.base.Supplier; import com.google.common.collect.Iterables; @@ -69,7 +70,6 @@ public class BasicExecutionContext extends AbstractExecutionContext { private static final Logger log = LoggerFactory.getLogger(BasicExecutionContext.class); static final ThreadLocal<BasicExecutionContext> perThreadExecutionContext = new ThreadLocal<BasicExecutionContext>(); - public static BasicExecutionContext getCurrentExecutionContext() { return perThreadExecutionContext.get(); } final ExecutionManager executionManager; @@ -448,6 +448,16 @@ public class BasicExecutionContext extends AbstractExecutionContext { } private void registerPerThreadExecutionContext() { perThreadExecutionContext.set(this); } + /** For use if external code wants to subsequently use an {@link ExecutionContext} but cannot submit via one. + * Caller should store the result and reset it back afterwards, in case a task may be running + * in the same thread as a synchronous submitter. */ + // only LocalSubscriptionManager needs to do that; and it could be refactored to take the execution context rather than tags. + @Beta + public static BasicExecutionContext setPerThreadExecutionContext(BasicExecutionContext ec) { + BasicExecutionContext old = perThreadExecutionContext.get(); + perThreadExecutionContext.set(ec); + return old; + } private void clearPerThreadExecutionContext() { perThreadExecutionContext.remove(); } http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/95c5b1e7/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java ---------------------------------------------------------------------- diff --git a/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java b/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java index c9b0e81..8174c22 100644 --- a/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java +++ b/locations/jclouds/src/main/java/org/apache/brooklyn/location/jclouds/templates/customize/TemplateOptionsOption.java @@ -23,14 +23,9 @@ import java.util.Map; import java.util.concurrent.ExecutionException; import org.apache.brooklyn.api.mgmt.ExecutionContext; -import org.apache.brooklyn.api.mgmt.ManagementContext; -import org.apache.brooklyn.api.mgmt.Task; -import org.apache.brooklyn.core.entity.EntityInternal; import org.apache.brooklyn.core.mgmt.BrooklynTaskTags; -import org.apache.brooklyn.core.mgmt.internal.LocalManagementContext; import org.apache.brooklyn.util.core.config.ConfigBag; import org.apache.brooklyn.util.core.flags.MethodCoercions; -import org.apache.brooklyn.util.core.task.BasicExecutionContext; import org.apache.brooklyn.util.core.task.Tasks; import org.apache.brooklyn.util.exceptions.Exceptions; import org.apache.brooklyn.util.guava.Maybe; @@ -53,10 +48,7 @@ public class TemplateOptionsOption implements TemplateOptionCustomizer { if (optionValue != null) { try { - final EntityInternal entity = (EntityInternal) BrooklynTaskTags.getTargetOrContextEntity(Tasks.current()); - final ExecutionContext exec = (null != entity - ? entity.getExecutionContext() - : BasicExecutionContext.getCurrentExecutionContext()); + final ExecutionContext exec = BrooklynTaskTags.getCurrentExecutionContext(); if (exec != null) { optionValue = Tasks.resolveDeepValue(optionValue, Object.class, exec); }
