You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by an...@apache.org on 2014/09/18 10:53:51 UTC
[2/5] git commit: introduce WeaklyPresent in maybe using weak
references, and use that for streams so that large streams attached to tasks
don't cause leaks;
also use a strongly typed tag for effector tasks, including entity information
introduce WeaklyPresent in maybe using weak references, and use that for streams so that large streams attached to tasks don't cause leaks;
also use a strongly typed tag for effector tasks, including entity information
Project: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/commit/189f426e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/tree/189f426e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/diff/189f426e
Branch: refs/heads/master
Commit: 189f426e60601b64f4c04411121d095b3713e4ff
Parents: 1f4a338
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Tue Sep 16 21:12:39 2014 +0100
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Sep 17 14:43:56 2014 +0100
----------------------------------------------------------------------
.../brooklyn/entity/basic/BrooklynTaskTags.java | 92 ++++++++++++++++----
.../management/internal/EffectorUtils.java | 2 +-
.../util/task/system/ProcessTaskWrapper.java | 4 +-
.../entity/basic/lifecycle/ScriptHelper.java | 6 +-
.../brooklynnode/BrooklynEntityMirrorImpl.java | 2 +-
.../main/java/brooklyn/util/guava/Maybe.java | 41 +++++++++
.../main/java/brooklyn/util/stream/Streams.java | 10 +++
7 files changed, 131 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java b/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
index cbc4ca6..fe573a7 100644
--- a/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
+++ b/core/src/main/java/brooklyn/entity/basic/BrooklynTaskTags.java
@@ -34,12 +34,14 @@ import brooklyn.entity.Effector;
import brooklyn.entity.Entity;
import brooklyn.management.ExecutionManager;
import brooklyn.management.Task;
+import brooklyn.util.guava.Maybe;
import brooklyn.util.stream.Streams;
import brooklyn.util.task.TaskTags;
import brooklyn.util.task.Tasks;
import brooklyn.util.text.StringEscapes.BashStringEscapes;
import brooklyn.util.text.Strings;
+import com.google.common.base.Functions;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
@@ -86,6 +88,7 @@ public class BrooklynTaskTags extends TaskTags {
}
@Override
public boolean equals(Object obj) {
+ if (this==obj) return true;
if (!(obj instanceof WrappedEntity)) return false;
return
Objects.equal(entity, ((WrappedEntity)obj).entity) &&
@@ -202,6 +205,14 @@ public class BrooklynTaskTags extends TaskTags {
public static WrappedStream tagForStream(String streamType, ByteArrayOutputStream stream) {
return new WrappedStream(streamType, stream);
}
+ /** creates a tag suitable for marking a stream available on a task, but which might be GC'd */
+ public static WrappedStream tagForStreamWeak(String streamType, ByteArrayOutputStream stream) {
+ Maybe<ByteArrayOutputStream> weakStream = Maybe.weakThen(stream,
+ Maybe.of(Streams.byteArrayOfString("<contents-garbage-collected>")));
+ return new WrappedStream(streamType,
+ Suppliers.compose(Functions.toStringFunction(), weakStream),
+ Suppliers.compose(Streams.sizeFunction(), weakStream));
+ }
/** creates a tag suitable for marking a stream available on a task */
public static WrappedStream tagForStream(String streamType, Supplier<String> contents, Supplier<Integer> size) {
@@ -241,18 +252,47 @@ public class BrooklynTaskTags extends TaskTags {
// ------ misc
- public static void setTransient(Task<?> task) {
- addTagDynamically(task, TRANSIENT_TASK_TAG);
- }
+ public static void setInessential(Task<?> task) { addTagDynamically(task, INESSENTIAL_TASK); }
+ public static void setTransient(Task<?> task) { addTagDynamically(task, TRANSIENT_TASK_TAG); }
+ public static boolean isTransient(Task<?> task) { return hasTag(task, TRANSIENT_TASK_TAG); }
+ public static boolean isSubTask(Task<?> task) { return hasTag(task, SUB_TASK_TAG); }
+ public static boolean isEffectorTask(Task<?> task) { return hasTag(task, EFFECTOR_TAG); }
- public static void setInessential(Task<?> task) {
- addTagDynamically(task, INESSENTIAL_TASK);
- }
-
// ------ effector tags
- public static String tagForEffectorName(String name) {
- return EFFECTOR_TAG+":"+name;
+ public static class EffectorCallTag {
+ protected final String entityId;
+ protected final String effectorName;
+ protected EffectorCallTag(String entityId, String effectorName) {
+ this.entityId = entityId;
+ this.effectorName = effectorName;
+ }
+ public String toString() {
+ return EFFECTOR_TAG+"@"+entityId+":"+effectorName;
+ }
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(entityId, effectorName);
+ }
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (!(obj instanceof EffectorCallTag)) return false;
+ EffectorCallTag other = (EffectorCallTag) obj;
+ return
+ Objects.equal(entityId, other.entityId) &&
+ Objects.equal(effectorName, other.effectorName);
+ }
+ public String getEntityId() {
+ return entityId;
+ }
+ public String getEffectorName() {
+ return effectorName;
+ }
+ }
+
+ public static EffectorCallTag tagForEffectorCall(Entity entity, String effectorName) {
+ return new EffectorCallTag(entity.getId(), effectorName);
}
/**
@@ -269,29 +309,43 @@ public class BrooklynTaskTags extends TaskTags {
while (t!=null) {
Set<Object> tags = t.getTags();
if (tags.contains(EFFECTOR_TAG)) {
- boolean match = true;
- if (entity!=null && !entity.equals(getTargetOrContextEntity(t)))
- match = false;
- if (effector!=null && !tags.contains(tagForEffectorName(effector.getName())))
- match = false;
- if (match) return true;
+ for (Object tag: tags) {
+ if (tag instanceof EffectorCallTag) {
+ EffectorCallTag et = (EffectorCallTag)tag;
+ if (entity!=null && !et.entityId.equals(entity.getId()))
+ continue;
+ if (effector!=null && !et.effectorName.equals(effector.getName()))
+ continue;
+ return true;
+ }
+ }
if (!allowNestedEffectorCalls) return false;
}
t = t.getSubmittedByTask();
}
return false;
}
-
- public static String getEffectorName(Task<?> task) {
+
+ /** finds the first {@link EffectorCallTag} tag on this tag, or optionally on submitters, or null */
+ public static EffectorCallTag getEffectorCallTag(Task<?> task, boolean recurse) {
Task<?> t = task;
while (t!=null) {
for (Object tag: task.getTags()) {
- if (tag instanceof String && ((String)tag).startsWith(EFFECTOR_TAG+":"))
- return Strings.removeFromStart((String)tag, EFFECTOR_TAG+":");
+ if (tag instanceof EffectorCallTag)
+ return (EffectorCallTag)tag;
}
+ if (!recurse)
+ return null;
t = t.getSubmittedByTask();
}
return null;
}
+ /** finds the first {@link EffectorCallTag} tag on this tag or a submitter, and returns the effector name */
+ public static String getEffectorName(Task<?> task) {
+ EffectorCallTag result = getEffectorCallTag(task, true);
+ if (result==null) return null;
+ return result.effectorName;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/core/src/main/java/brooklyn/management/internal/EffectorUtils.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/internal/EffectorUtils.java b/core/src/main/java/brooklyn/management/internal/EffectorUtils.java
index ac813db..39bbcd1 100644
--- a/core/src/main/java/brooklyn/management/internal/EffectorUtils.java
+++ b/core/src/main/java/brooklyn/management/internal/EffectorUtils.java
@@ -322,7 +322,7 @@ public class EffectorUtils {
.put("displayName", effector.getName())
.put("tags", MutableList.of(
BrooklynTaskTags.EFFECTOR_TAG,
- BrooklynTaskTags.tagForEffectorName(effector.getName()),
+ BrooklynTaskTags.tagForEffectorCall(entity, effector.getName()),
BrooklynTaskTags.tagForTargetEntity(entity)))
.build();
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/core/src/main/java/brooklyn/util/task/system/ProcessTaskWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/util/task/system/ProcessTaskWrapper.java b/core/src/main/java/brooklyn/util/task/system/ProcessTaskWrapper.java
index db4f080..9910bad 100644
--- a/core/src/main/java/brooklyn/util/task/system/ProcessTaskWrapper.java
+++ b/core/src/main/java/brooklyn/util/task/system/ProcessTaskWrapper.java
@@ -55,8 +55,8 @@ public abstract class ProcessTaskWrapper<RET> extends ProcessTaskStub implements
protected ProcessTaskWrapper(AbstractProcessTaskFactory<?,RET> constructor) {
super(constructor);
TaskBuilder<Object> tb = constructor.constructCustomizedTaskBuilder();
- if (stdout!=null) tb.tag(BrooklynTaskTags.tagForStream(BrooklynTaskTags.STREAM_STDOUT, stdout));
- if (stderr!=null) tb.tag(BrooklynTaskTags.tagForStream(BrooklynTaskTags.STREAM_STDERR, stderr));
+ if (stdout!=null) tb.tag(BrooklynTaskTags.tagForStreamWeak(BrooklynTaskTags.STREAM_STDOUT, stdout));
+ if (stderr!=null) tb.tag(BrooklynTaskTags.tagForStreamWeak(BrooklynTaskTags.STREAM_STDERR, stderr));
task = (Task<RET>) tb.body(new ProcessTaskInternalJob()).build();
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/software/base/src/main/java/brooklyn/entity/basic/lifecycle/ScriptHelper.java
----------------------------------------------------------------------
diff --git a/software/base/src/main/java/brooklyn/entity/basic/lifecycle/ScriptHelper.java b/software/base/src/main/java/brooklyn/entity/basic/lifecycle/ScriptHelper.java
index e60e1d7..a0dafe8 100644
--- a/software/base/src/main/java/brooklyn/entity/basic/lifecycle/ScriptHelper.java
+++ b/software/base/src/main/java/brooklyn/entity/basic/lifecycle/ScriptHelper.java
@@ -269,7 +269,7 @@ public class ScriptHelper {
stdin.write(line.getBytes());
stdin.write("\n".getBytes());
}
- tb.tag(BrooklynTaskTags.tagForStream(BrooklynTaskTags.STREAM_STDIN, stdin));
+ tb.tag(BrooklynTaskTags.tagForStreamWeak(BrooklynTaskTags.STREAM_STDIN, stdin));
} catch (IOException e) {
log.warn("Error registering stream "+BrooklynTaskTags.STREAM_STDIN+" on "+tb+": "+e, e);
}
@@ -283,9 +283,9 @@ public class ScriptHelper {
if (gatherOutput) {
stdout = new ByteArrayOutputStream();
- tb.tag(BrooklynTaskTags.tagForStream(BrooklynTaskTags.STREAM_STDOUT, stdout));
+ tb.tag(BrooklynTaskTags.tagForStreamWeak(BrooklynTaskTags.STREAM_STDOUT, stdout));
stderr = new ByteArrayOutputStream();
- tb.tag(BrooklynTaskTags.tagForStream(BrooklynTaskTags.STREAM_STDERR, stderr));
+ tb.tag(BrooklynTaskTags.tagForStreamWeak(BrooklynTaskTags.STREAM_STDERR, stderr));
}
task = tb.build();
if (isTransient) BrooklynTaskTags.setTransient(task);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/software/base/src/main/java/brooklyn/entity/brooklynnode/BrooklynEntityMirrorImpl.java
----------------------------------------------------------------------
diff --git a/software/base/src/main/java/brooklyn/entity/brooklynnode/BrooklynEntityMirrorImpl.java b/software/base/src/main/java/brooklyn/entity/brooklynnode/BrooklynEntityMirrorImpl.java
index fb87f39..254ac9b 100644
--- a/software/base/src/main/java/brooklyn/entity/brooklynnode/BrooklynEntityMirrorImpl.java
+++ b/software/base/src/main/java/brooklyn/entity/brooklynnode/BrooklynEntityMirrorImpl.java
@@ -176,7 +176,7 @@ public class BrooklynEntityMirrorImpl extends AbstractEntity implements Brooklyn
Exceptions.propagateIfFatal(e);
throw new IllegalStateException("Invalid response invoking "+uri+": "+e, e);
}
- Tasks.addTagDynamically(BrooklynTaskTags.tagForStream("http_response", Streams.byteArray(content)));
+ Tasks.addTagDynamically(BrooklynTaskTags.tagForStreamWeak("http_response", Streams.byteArray(content)));
if (!HttpTool.isStatusCodeHealthy(result.getResponseCode())) {
log.warn("Invalid response invoking "+uri+": response code "+result.getResponseCode()+"\n"+result+": "+new String(content));
throw new IllegalStateException("Invalid response invoking "+uri+": response code "+result.getResponseCode());
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/utils/common/src/main/java/brooklyn/util/guava/Maybe.java
----------------------------------------------------------------------
diff --git a/utils/common/src/main/java/brooklyn/util/guava/Maybe.java b/utils/common/src/main/java/brooklyn/util/guava/Maybe.java
index fb60bf8..f4f8d13 100644
--- a/utils/common/src/main/java/brooklyn/util/guava/Maybe.java
+++ b/utils/common/src/main/java/brooklyn/util/guava/Maybe.java
@@ -19,10 +19,12 @@
package brooklyn.util.guava;
import java.io.Serializable;
+import java.lang.ref.WeakReference;
import java.util.Collections;
import java.util.Iterator;
import java.util.Set;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import brooklyn.util.javalang.JavaClassNames;
@@ -69,6 +71,16 @@ public abstract class Maybe<T> implements Serializable, Supplier<T> {
return new Present<T>(value);
}
+ /** creates an instance wrapping a {@link WeakReference}, so it might go absent later on */
+ public static <T> Maybe<T> weak(T value) {
+ return weakThen(value, null);
+ }
+ /** creates an instance wrapping a {@link WeakReference}, using the second item given if lost */
+ public static <T> Maybe<T> weakThen(T value, Maybe<T> ifEmpty) {
+ if (value==null) return of((T)null);
+ return new WeaklyPresent<T>(value).usingAfterExpiry(ifEmpty);
+ }
+
/** like {@link Optional#fromNullable(Object)}, returns absent if the argument is null */
public static <T> Maybe<T> fromNullable(@Nullable T value) {
if (value==null) return absent();
@@ -193,6 +205,35 @@ public abstract class Maybe<T> implements Serializable, Supplier<T> {
}
}
+ public static class WeaklyPresent<T> extends Maybe<T> {
+ private static final long serialVersionUID = 436799990500336015L;
+ private final WeakReference<T> value;
+ private Maybe<T> defaultValue;
+ protected WeaklyPresent(@Nonnull T value) {
+ this.value = new WeakReference<T>(value);
+ }
+ @Override
+ public T get() {
+ T result = value.get();
+ if (result==null) {
+ if (defaultValue==null) throw new IllegalStateException("Weakly present item has been GC'd");
+ return defaultValue.get();
+ }
+ return result;
+ }
+ @Override
+ public boolean isPresent() {
+ return value.get()!=null || (defaultValue!=null && defaultValue.isPresent());
+ }
+ public Maybe<T> solidify() {
+ return Maybe.fromNullable(value.get());
+ }
+ WeaklyPresent<T> usingAfterExpiry(Maybe<T> defaultValue) {
+ this.defaultValue = defaultValue;
+ return this;
+ }
+ }
+
@Override
public String toString() {
return JavaClassNames.simpleClassName(this)+"["+(isPresent()?"value="+get():"")+"]";
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/189f426e/utils/common/src/main/java/brooklyn/util/stream/Streams.java
----------------------------------------------------------------------
diff --git a/utils/common/src/main/java/brooklyn/util/stream/Streams.java b/utils/common/src/main/java/brooklyn/util/stream/Streams.java
index 3e62e6b..d7f0a7d 100644
--- a/utils/common/src/main/java/brooklyn/util/stream/Streams.java
+++ b/utils/common/src/main/java/brooklyn/util/stream/Streams.java
@@ -38,6 +38,7 @@ import brooklyn.util.exceptions.Exceptions;
import com.google.common.annotations.Beta;
import com.google.common.base.Charsets;
+import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
import com.google.common.io.ByteStreams;
@@ -139,6 +140,15 @@ public class Streams {
};
}
+ public static Function<ByteArrayOutputStream,Integer> sizeFunction() {
+ return new Function<ByteArrayOutputStream,Integer>() {
+ @Override
+ public Integer apply(ByteArrayOutputStream input) {
+ return input.size();
+ }
+ };
+ }
+
public static ByteArrayOutputStream byteArrayOfString(String in) {
return byteArray(in.getBytes(Charsets.UTF_8));
}