From 08cf885a94e1ee923141c6bfb8eee37989d2b195 Mon Sep 17 00:00:00 2001
From: Christian Wulf <chw@informatik.uni-kiel.de>
Date: Tue, 20 Jan 2015 09:39:40 +0100
Subject: [PATCH] refactored package structure; renamed some classes; added
TcpReaderAnalysisConfigurationTest
---
.classpath | 16 +++-
.../{Collector.java => CollectorActor.java} | 13 ++--
src/main/java/kiekerdays/CounterActor.java | 27 +++++++
.../{Reaper.java => ReaperActor.java} | 2 +-
.../kiekerdays/tcpreader/TcpReaderActor.java | 4 +
.../TcpReaderAnalysisConfiguration.java | 78 +++++++++++++++++++
...ceReconstructionAnalysisConfiguration.java | 9 ++-
.../teetime/stage/io/AbstractTcpReader.java | 28 +++++--
.../tcpreader => teetime/util}/TcpReader.java | 10 +--
.../TcpReaderAnalysisConfigurationTest.java | 22 ++++++
...ceReductionAnalysisConfigurationTest.java} | 12 +--
11 files changed, 189 insertions(+), 32 deletions(-)
rename src/main/java/kiekerdays/{Collector.java => CollectorActor.java} (69%)
create mode 100644 src/main/java/kiekerdays/CounterActor.java
rename src/main/java/kiekerdays/{Reaper.java => ReaperActor.java} (89%)
create mode 100644 src/main/java/kiekerdays/tcpreader/TcpReaderAnalysisConfiguration.java
rename src/main/java/{kiekerdays/tcpreader => teetime/util}/TcpReader.java (95%)
create mode 100644 src/test/java/kiekerdays/tcpreader/TcpReaderAnalysisConfigurationTest.java
rename src/test/java/kiekerdays/tcpreconstruction/{TraceReductionAnalysisTest.java => TraceReductionAnalysisConfigurationTest.java} (63%)
diff --git a/.classpath b/.classpath
index 72009e2..d3a75f6 100644
--- a/.classpath
+++ b/.classpath
@@ -1,8 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
- <classpathentry kind="src" path="src/main/java"/>
- <classpathentry kind="src" path="src/test/java"/>
- <classpathentry kind="src" path="src/test/resources"/>
+ <classpathentry kind="src" output="target/classes" path="src/main/java">
+ <attributes>
+ <attribute name="optional" value="true"/>
+ <attribute name="maven.pomderived" value="true"/>
+ </attributes>
+ </classpathentry>
+ <classpathentry kind="src" output="target/test-classes" path="src/test/java">
+ <attributes>
+ <attribute name="optional" value="true"/>
+ <attribute name="maven.pomderived" value="true"/>
+ </attributes>
+ </classpathentry>
+ <classpathentry including="**/*.java" kind="src" path="src/test/resources"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.7">
<attributes>
<attribute name="maven.pomderived" value="true"/>
diff --git a/src/main/java/kiekerdays/Collector.java b/src/main/java/kiekerdays/CollectorActor.java
similarity index 69%
rename from src/main/java/kiekerdays/Collector.java
rename to src/main/java/kiekerdays/CollectorActor.java
index 88f77f7..8ed9804 100644
--- a/src/main/java/kiekerdays/Collector.java
+++ b/src/main/java/kiekerdays/CollectorActor.java
@@ -8,30 +8,29 @@ import org.slf4j.LoggerFactory;
import akka.actor.UntypedActor;
-public class Collector extends UntypedActor {
+public class CollectorActor extends UntypedActor {
- private static final Logger LOGGER = LoggerFactory.getLogger(Collector.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(CollectorActor.class);
private final Collection<Object> elements;
- public Collector() {
+ public CollectorActor() {
this(new LinkedList<Object>());
}
- public Collector(final Collection<Object> elements) {
+ public CollectorActor(final Collection<Object> elements) {
super();
this.elements = elements;
}
@Override
public void onReceive(Object message) throws Exception {
- elements.add(message.getClass());
+ elements.add(message);
}
@Override
public void postStop() throws Exception {
- LOGGER.info("stopped: " + elements);
+ LOGGER.info("stopped: " + elements.size());
super.postStop();
}
-
}
diff --git a/src/main/java/kiekerdays/CounterActor.java b/src/main/java/kiekerdays/CounterActor.java
new file mode 100644
index 0000000..de04ffc
--- /dev/null
+++ b/src/main/java/kiekerdays/CounterActor.java
@@ -0,0 +1,27 @@
+package kiekerdays;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import akka.actor.UntypedActor;
+
+public class CounterActor extends UntypedActor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(CounterActor.class);
+ private int numMessages;
+
+ public CounterActor() {
+ super();
+ }
+
+ @Override
+ public void onReceive(Object message) throws Exception {
+ numMessages++;
+ }
+
+ @Override
+ public void postStop() throws Exception {
+ LOGGER.info("stopped: " + numMessages);
+ super.postStop();
+ }
+}
diff --git a/src/main/java/kiekerdays/Reaper.java b/src/main/java/kiekerdays/ReaperActor.java
similarity index 89%
rename from src/main/java/kiekerdays/Reaper.java
rename to src/main/java/kiekerdays/ReaperActor.java
index 49fa860..1e48ff6 100644
--- a/src/main/java/kiekerdays/Reaper.java
+++ b/src/main/java/kiekerdays/ReaperActor.java
@@ -3,7 +3,7 @@ package kiekerdays;
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
-public class Reaper extends UntypedActor {
+public class ReaperActor extends UntypedActor {
public static class WatchMe {
private final ActorRef actorRef;
diff --git a/src/main/java/kiekerdays/tcpreader/TcpReaderActor.java b/src/main/java/kiekerdays/tcpreader/TcpReaderActor.java
index 426e524..48a4d40 100644
--- a/src/main/java/kiekerdays/tcpreader/TcpReaderActor.java
+++ b/src/main/java/kiekerdays/tcpreader/TcpReaderActor.java
@@ -5,6 +5,7 @@ import kieker.common.record.IMonitoringRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import teetime.util.TcpReader;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
@@ -16,9 +17,12 @@ public class TcpReaderActor extends UntypedActor {
private ActorRef tcpTraceReconstructionActor;
private TcpReader tcpReader = new TcpReader() {
+ // private int numProcessedRecords;
+
@Override
protected final void send(IMonitoringRecord record) {
tcpTraceReconstructionActor.tell(record, getSelf());
+ // LOGGER.info("#processed records: " + numProcessedRecords++);
};
};
diff --git a/src/main/java/kiekerdays/tcpreader/TcpReaderAnalysisConfiguration.java b/src/main/java/kiekerdays/tcpreader/TcpReaderAnalysisConfiguration.java
new file mode 100644
index 0000000..fd984ca
--- /dev/null
+++ b/src/main/java/kiekerdays/tcpreader/TcpReaderAnalysisConfiguration.java
@@ -0,0 +1,78 @@
+package kiekerdays.tcpreader;
+
+import kiekerdays.CounterActor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSelection;
+import akka.actor.ActorSystem;
+import akka.actor.PoisonPill;
+import akka.actor.Props;
+
+public class TcpReaderAnalysisConfiguration {
+
+ private static final String START_MESSAGE = "start";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TcpReaderAnalysisConfiguration.class);
+
+ private ActorRef tcpReaderActor;
+ private ActorRef rootActor;
+
+ public TcpReaderAnalysisConfiguration() {
+ init();
+ }
+
+ private void init() {
+ ActorSystem system = ActorSystem.create("TraceReductionAnalysisConfiguration");
+
+ // specify actors
+ Props counterProps = Props.create(CounterActor.class);
+ Props tcpReaderProps = Props.create(TcpReaderActor.class, counterProps);
+
+ // create actors
+ final ActorRef tcpReaderActor = system.actorOf(tcpReaderProps, "TcpReaderActor");
+ // final ActorRef tcpTraceReconstructonActor = system.actorOf(tcpTraceReconstructionProps, "TcpTraceReconstructionActor");
+ // final ActorRef validTraceReceiver = system.actorOf(collectorValidProps, "CollectorValid");
+ // final ActorRef invalidTraceReceiver = system.actorOf(collectorInvalidProps, "CollectorInvalid");
+ // final ActorRef tcpReaderActor = system.actorOf(Props.create(TcpReaderActor.class,
+ // new Creator<TcpReaderActor>() {
+ // @Override
+ // public TcpReaderActor create() throws Exception {
+ // return new TcpReaderActor(tcpTraceReconstructonActor);
+ // }
+ // }));
+
+ // tcpReader = TypedActor.get(system).typedActorOf(new TypedProps<TcpReaderActor>(ITcpReader.class, new
+ // Creator<TcpReaderActor>() {
+ // @Override
+ // public TcpReaderActor create() throws Exception {
+ // return new TcpReaderActor(tcpTraceReconstructonActor);
+ // }
+ // }));
+
+ this.tcpReaderActor = tcpReaderActor;
+
+ ActorSelection rootActorSelection = system.actorSelection(tcpReaderActor.path().root());
+ rootActor = rootActorSelection.anchor();
+
+ LOGGER.info("Configuration initialized.");
+ }
+
+ public void start() throws Exception {
+ LOGGER.info("Starting analysis...");
+ // tcpReader.onStarting();
+ // tcpReader.execute();
+ // tcpReader.onTerminating();
+ tcpReaderActor.tell(START_MESSAGE, ActorRef.noSender());
+ LOGGER.info("Analysis started.");
+
+ rootActor.tell(PoisonPill.getInstance(), rootActor);
+ // tcpReaderActor.parent().tell(PoisonPill.getInstance(), getSelf());
+ // Future<Terminated> future = system.terminate();
+ // Await.ready(future, Duration.Inf());
+ LOGGER.info("Analysis stopped.");
+ }
+
+}
diff --git a/src/main/java/kiekerdays/tcpreconstruction/TraceReconstructionAnalysisConfiguration.java b/src/main/java/kiekerdays/tcpreconstruction/TraceReconstructionAnalysisConfiguration.java
index c2d5c8c..2c4d4c4 100644
--- a/src/main/java/kiekerdays/tcpreconstruction/TraceReconstructionAnalysisConfiguration.java
+++ b/src/main/java/kiekerdays/tcpreconstruction/TraceReconstructionAnalysisConfiguration.java
@@ -1,10 +1,10 @@
package kiekerdays.tcpreconstruction;
import java.util.Collection;
-import java.util.Collections;
import java.util.LinkedList;
-import kiekerdays.Collector;
+import kiekerdays.CollectorActor;
+import kiekerdays.CounterActor;
import kiekerdays.tcpreader.TcpReaderActor;
import org.slf4j.Logger;
@@ -39,8 +39,9 @@ public class TraceReconstructionAnalysisConfiguration {
ActorSystem system = ActorSystem.create("TraceReductionAnalysisConfiguration");
// specify actors
- Props collectorValidProps = Props.create(Collector.class, validTraces);
- Props collectorInvalidProps = Props.create(Collector.class, Collections.emptyList());
+ // Props collectorValidProps = Props.create(Collector.class, validTraces); // do not use a collector since it pollutes the heap
+ Props collectorValidProps = Props.create(CounterActor.class);
+ Props collectorInvalidProps = Props.create(CollectorActor.class);
Props tcpTraceReconstructionProps = Props.create(TcpTraceReconstructionActor.class, traceId2trace, collectorValidProps, collectorInvalidProps);
Props tcpReaderProps = Props.create(TcpReaderActor.class, tcpTraceReconstructionProps);
diff --git a/src/main/java/teetime/stage/io/AbstractTcpReader.java b/src/main/java/teetime/stage/io/AbstractTcpReader.java
index 0a43dc4..3c9c255 100644
--- a/src/main/java/teetime/stage/io/AbstractTcpReader.java
+++ b/src/main/java/teetime/stage/io/AbstractTcpReader.java
@@ -1,6 +1,5 @@
package teetime.stage.io;
-
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.BufferUnderflowException;
@@ -14,10 +13,12 @@ import org.slf4j.LoggerFactory;
public abstract class AbstractTcpReader<T> {
protected final Logger logger = LoggerFactory.getLogger(AbstractTcpReader.class.getName());
-
+
private final int port;
private final int bufferCapacity;
+ private int numBufferUnderflows;
+
public AbstractTcpReader(final int port, final int bufferCapacity) {
super();
this.port = port;
@@ -41,13 +42,13 @@ public abstract class AbstractTcpReader<T> {
socketChannel.close();
}
} catch (final IOException ex) {
- logger.error("Error while reading", ex);
+ logger.error("Error while reading.", ex);
} finally {
if (null != serversocket) {
try {
serversocket.close();
} catch (final IOException e) {
- logger.debug("Failed to close TCP connection!", e);
+ logger.debug("Failed to close TCP connection.", e);
}
}
}
@@ -58,15 +59,30 @@ public abstract class AbstractTcpReader<T> {
try {
while (buffer.hasRemaining()) {
buffer.mark();
- this.read(buffer);
+ boolean success = this.read(buffer);
+ if (!success) {
+ buffer.reset();
+ buffer.compact();
+ return;
+ }
}
buffer.clear();
} catch (final BufferUnderflowException ex) {
+ // logger.warn("Unexpected exception. Resetting and compacting buffer.", ex);
+ numBufferUnderflows++;
+ logger.warn("numBufferUnderflows: " + numBufferUnderflows);
buffer.reset();
buffer.compact();
}
}
- protected abstract void read(final ByteBuffer buffer);
+ /**
+ * @param buffer
+ * to be read from
+ * @return <ul>
+ * <li><code>true</code> when there were enough bytes to perform the read operation
+ * <li><code>false</code> otherwise. In this case, the buffer is reset, compacted, and filled with new content.
+ */
+ protected abstract boolean read(final ByteBuffer buffer);
}
diff --git a/src/main/java/kiekerdays/tcpreader/TcpReader.java b/src/main/java/teetime/util/TcpReader.java
similarity index 95%
rename from src/main/java/kiekerdays/tcpreader/TcpReader.java
rename to src/main/java/teetime/util/TcpReader.java
index 7d1fb0f..ad42117 100644
--- a/src/main/java/kiekerdays/tcpreader/TcpReader.java
+++ b/src/main/java/teetime/util/TcpReader.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package kiekerdays.tcpreader;
+package teetime.util;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -86,7 +86,7 @@ public class TcpReader extends AbstractTcpReader<IMonitoringRecord> {
}
@Override
- protected final void read(final ByteBuffer buffer) {
+ protected final boolean read(final ByteBuffer buffer) {
final int clazzId = buffer.getInt();
final long loggingTimestamp = buffer.getLong();
@@ -97,11 +97,11 @@ public class TcpReader extends AbstractTcpReader<IMonitoringRecord> {
record.setLoggingTimestamp(loggingTimestamp);
send(record);
- } catch (final BufferUnderflowException ex) {
- super.logger.error("Failed to create record.", ex);
} catch (final RecordInstantiationException ex) {
- super.logger.error("Failed to create record.", ex);
+ super.logger.error("Failed to create record:", ex);
}
+
+ return true;
}
protected void send(IMonitoringRecord record) {};
diff --git a/src/test/java/kiekerdays/tcpreader/TcpReaderAnalysisConfigurationTest.java b/src/test/java/kiekerdays/tcpreader/TcpReaderAnalysisConfigurationTest.java
new file mode 100644
index 0000000..a758de5
--- /dev/null
+++ b/src/test/java/kiekerdays/tcpreader/TcpReaderAnalysisConfigurationTest.java
@@ -0,0 +1,22 @@
+package kiekerdays.tcpreader;
+
+public class TcpReaderAnalysisConfigurationTest {
+
+ // @Test
+ public void testAnalysis() throws Exception {
+ TcpReaderAnalysisConfiguration configuration = new TcpReaderAnalysisConfiguration();
+
+ configuration.start();
+ // fail("Akka does not support being executed by JUnit.");
+ }
+
+ public static void main(String[] args) {
+ try {
+ TcpReaderAnalysisConfigurationTest test = new TcpReaderAnalysisConfigurationTest();
+ test.testAnalysis();
+ } catch (Exception e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+}
diff --git a/src/test/java/kiekerdays/tcpreconstruction/TraceReductionAnalysisTest.java b/src/test/java/kiekerdays/tcpreconstruction/TraceReductionAnalysisConfigurationTest.java
similarity index 63%
rename from src/test/java/kiekerdays/tcpreconstruction/TraceReductionAnalysisTest.java
rename to src/test/java/kiekerdays/tcpreconstruction/TraceReductionAnalysisConfigurationTest.java
index 505c335..d8c8f9e 100644
--- a/src/test/java/kiekerdays/tcpreconstruction/TraceReductionAnalysisTest.java
+++ b/src/test/java/kiekerdays/tcpreconstruction/TraceReductionAnalysisConfigurationTest.java
@@ -1,19 +1,19 @@
package kiekerdays.tcpreconstruction;
-import org.junit.Test;
-public class TraceReductionAnalysisTest {
+public class TraceReductionAnalysisConfigurationTest {
- @Test
+ // @Test
public void testAnalysis() throws Exception {
TraceReconstructionAnalysisConfiguration configuration = new TraceReconstructionAnalysisConfiguration();
-
+
configuration.start();
+ // fail("Akka does not support being executed by JUnit.");
}
-
+
public static void main(String[] args) {
try {
- TraceReductionAnalysisTest test = new TraceReductionAnalysisTest();
+ TraceReductionAnalysisConfigurationTest test = new TraceReductionAnalysisConfigurationTest();
test.testAnalysis();
} catch (Exception e) {
// TODO Auto-generated catch block
--
GitLab