You are viewing a plain text version of this content. The canonical link for it is here.
Posted to s4-commits@incubator.apache.org by mm...@apache.org on 2012/01/03 14:03:28 UTC
[3/50] [abbrv] git commit: Changed initialization code in PE. Moved
threadsafe check to init proto (needs more debugging). Found bug in init
trigger,
should copy trigger map for each new PE instance. Added close() method to
receiverbut comm layer does no
Changed initialization code in PE. Moved threadsafe check to init proto (needs more debugging). Found bug in init trigger, should copy trigger map for each new PE instance. Added close() method to receiverbut comm layer does nott shutdown gracefully. Needs more work.
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/5a173e75
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/5a173e75
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/5a173e75
Branch: refs/heads/piper
Commit: 5a173e7578278167b935ea5c3797a08c781f47ba
Parents: 1804c47
Author: Leo Neumeyer <le...@s4.io>
Authored: Sun Dec 18 23:25:59 2011 -0800
Committer: Leo Neumeyer <le...@s4.io>
Committed: Sun Dec 18 23:25:59 2011 -0800
----------------------------------------------------------------------
.../java/org/apache/s4/core/ProcessingElement.java | 39 ++++-----------
.../src/main/java/org/apache/s4/core/Receiver.java | 9 +++-
subprojects/s4-core/src/main/resources/logback.xml | 2 +-
.../java/org/apache/s4/example/counter/MyApp.java | 3 +-
.../org/apache/s4/example/counter/PrintPE.java | 4 +-
5 files changed, 22 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/5a173e75/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
index 43d5a7b..befe94f 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
@@ -454,9 +454,8 @@ abstract public class ProcessingElement implements Cloneable {
private ProcessingElement createPE(String id) {
ProcessingElement pe = (ProcessingElement) this.clone();
pe.isPrototype = false;
- if (isFirst) // TODO: THIS DOESNT MAKE SENSE. WE ONLY CREATE INSTANCE ONCE. REVIEW.
- initPEInstance(pe);
pe.id = id;
+ pe.triggers = Maps.newHashMap(triggers);
pe.onCreate();
logger.trace("Num PE instances: {}.", getNumPEInstances());
return pe;
@@ -481,6 +480,15 @@ abstract public class ProcessingElement implements Cloneable {
logger.info("Started timer for PE [{}] with ID [{}].", this.getClass().getName(), id);
}
+ /* Check if this PE is annotated as thread safe. */
+ if (getClass().isAnnotationPresent(ThreadSafe.class) == true) {
+
+ // TODO: this doesn't seem to be working. isannotationpresent always returns false.
+
+ isThreadSafe = true;
+ logger.trace("Annotated with @ThreadSafe");
+ }
+
}
/**
@@ -547,33 +555,6 @@ abstract public class ProcessingElement implements Cloneable {
return ImmutableMap.copyOf(peInstances.asMap());
}
- /*
- * Called when we create the first PE instance. TODO: Would be better to do this as part of the PE lifecycle after
- * PE construction.
- */
- private void initPEInstance(ProcessingElement pe) {
-
- isFirst = false;
-
- logger.trace("OnCreateInternal");
-
- /*
- * If PE class has the @ThreadSafe annotation, then we set isThtreadSafe to true in the prototype so all future
- * PE instances inherit the setting.
- */
- if (pe.getClass().isAnnotationPresent(ThreadSafe.class) == true) {
- pe.isThreadSafe = true;
- isThreadSafe = true;
-
- logger.trace("Annotated with @ThreadSafe");
- }
-
- /*
- * Each PE instance needs its own triggers map to keep track time lapsed and event count.
- */
- pe.triggers = Maps.newHashMap(triggers);
- }
-
/**
* Unique ID for a PE instance.
*
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/5a173e75/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
index 7b8f364..90ae0bf 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Receiver.java
@@ -32,12 +32,15 @@ public class Receiver implements Runnable {
final private Listener listener;
final private SerializerDeserializer serDeser;
private Map<Integer, Map<Integer, Stream<? extends Event>>> streams;
+ private Thread thread;
@Inject
public Receiver(Listener listener, SerializerDeserializer serDeser) {
this.listener = listener;
this.serDeser = serDeser;
- new Thread(this).start();
+
+ thread = new Thread(this, "Receiver");
+ thread.start();
streams = new MapMaker().makeMap();
}
@@ -69,6 +72,8 @@ public class Receiver implements Runnable {
}
public void run() {
+ // TODO: this thread never seems to get interrupted. SHould we catch an interrupted exception from listener
+ // here?
while (!Thread.interrupted()) {
byte[] message = listener.recv();
Event event = (Event) serDeser.deserialize(message);
@@ -89,6 +94,6 @@ public class Receiver implements Runnable {
}
public void close() {
- Thread.currentThread().interrupt();
+ thread.interrupt();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/5a173e75/subprojects/s4-core/src/main/resources/logback.xml
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/resources/logback.xml b/subprojects/s4-core/src/main/resources/logback.xml
index ea8c85a..6dac308 100644
--- a/subprojects/s4-core/src/main/resources/logback.xml
+++ b/subprojects/s4-core/src/main/resources/logback.xml
@@ -8,7 +8,7 @@
</encoder>
</appender>
- <root level="debug">
+ <root level="trace">
<appender-ref ref="STDOUT" />
</root>
</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/5a173e75/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
index baf7c85..51c7261 100644
--- a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
+++ b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
@@ -120,8 +120,6 @@ final public class MyApp extends App {
@Override
protected void onStart() {
- System.out.println("Done. Wait until the main app closes.");
- // close();
}
@Override
@@ -147,5 +145,6 @@ final public class MyApp extends App {
e.printStackTrace();
}
myApp.close();
+ receiver.close();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/5a173e75/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/PrintPE.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/PrintPE.java b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/PrintPE.java
index f87003f..29837d4 100644
--- a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/PrintPE.java
+++ b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/PrintPE.java
@@ -15,11 +15,13 @@
*/
package org.apache.s4.example.counter;
+import javax.annotation.concurrent.ThreadSafe;
+
import org.apache.s4.base.Event;
import org.apache.s4.core.App;
import org.apache.s4.core.ProcessingElement;
-
+@ThreadSafe
public class PrintPE extends ProcessingElement {
public PrintPE(App app) {