diff --git a/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis8Test.java b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis8Test.java
index 2f147fe8756befefb7fe295fe94b492ebb97c00f..8a9b34ff0ca39af0f87dd5978b9634c37d489cab 100644
--- a/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis8Test.java
+++ b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis8Test.java
@@ -43,9 +43,6 @@ public class MethodCallThoughputTimestampAnalysis8Test {
System.setProperty(LogFactory.CUSTOM_LOGGER_JVM, "NONE");
}
- // 500 times faster than our new framework
- // TODO check why
-
@Test
public void testWithManyObjects() {
System.out.println("Testing teetime (mc) with NUM_OBJECTS_TO_CREATE=" + NUM_OBJECTS_TO_CREATE + ", NUM_NOOP_FILTERS="
diff --git a/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis9Test.java b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis9Test.java
new file mode 100644
index 0000000000000000000000000000000000000000..4c41d19959cc379efb3588628ab0c9fe8862999b
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis9Test.java
@@ -0,0 +1,74 @@
+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package teetime.examples.throughput;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import teetime.examples.throughput.methodcall.MethodCallThroughputAnalysis9;
+import teetime.util.StatisticsUtil;
+import teetime.util.StopWatch;
+
+import kieker.common.logging.LogFactory;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class MethodCallThoughputTimestampAnalysis9Test {
+
+ private static final int NUM_OBJECTS_TO_CREATE = 100000;
+ private static final int NUM_NOOP_FILTERS = 800;
+
+ @Before
+ public void before() {
+ System.setProperty(LogFactory.CUSTOM_LOGGER_JVM, "NONE");
+ }
+
+ @Test
+ public void testWithManyObjects() {
+ System.out.println("Testing teetime (mc) with NUM_OBJECTS_TO_CREATE=" + NUM_OBJECTS_TO_CREATE + ", NUM_NOOP_FILTERS="
+ + NUM_NOOP_FILTERS + "...");
+ final StopWatch stopWatch = new StopWatch();
+ final List<TimestampObject> timestampObjects = new ArrayList<TimestampObject>(NUM_OBJECTS_TO_CREATE);
+
+ final MethodCallThroughputAnalysis9 analysis = new MethodCallThroughputAnalysis9();
+ analysis.setNumNoopFilters(NUM_NOOP_FILTERS);
+ analysis.setTimestampObjects(timestampObjects);
+ analysis.setInput(NUM_OBJECTS_TO_CREATE, new Callable<TimestampObject>() {
+ @Override
+ public TimestampObject call() throws Exception {
+ return new TimestampObject();
+ }
+ });
+ analysis.init();
+
+ stopWatch.start();
+ try {
+ analysis.start();
+ } finally {
+ stopWatch.end();
+ }
+
+ StatisticsUtil.printStatistics(stopWatch.getDurationInNs(), timestampObjects);
+ }
+
+}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/AbstractStage.java b/src/test/java/teetime/examples/throughput/methodcall/AbstractStage.java
index acd437ab5afe783fddecfb611ee0d1bc697fcf4b..949dbab32b1e5d660f42e207442393e7e4fe8946 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/AbstractStage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/AbstractStage.java
@@ -3,10 +3,10 @@ package teetime.examples.throughput.methodcall;
import teetime.util.list.CommittableQueue;
import teetime.util.list.CommittableResizableArrayQueue;
-abstract class AbstractStage<I, O> implements Stage<I, O> {
+abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
- // private final InputPort<I> inputPort = new InputPort<I>();
- // private final OutputPort<O> outputPort = new OutputPort<O>();
+ private final InputPort<I> inputPort = new InputPort<I>();
+ private final OutputPort<O> outputPort = new OutputPort<O>();
protected final CommittableQueue<O> outputElements = new CommittableResizableArrayQueue<O>(null, 4);
@@ -22,15 +22,15 @@ abstract class AbstractStage<I, O> implements Stage<I, O> {
private boolean reschedulable;
- // @Override
- // public InputPort<I> getInputPort() {
- // return this.inputPort;
- // }
+ @Override
+ public InputPort<I> getInputPort() {
+ return this.inputPort;
+ }
- // @Override
- // public OutputPort<O> getOutputPort() {
- // return this.outputPort;
- // }
+ @Override
+ public OutputPort<O> getOutputPort() {
+ return this.outputPort;
+ }
@Override
public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
@@ -57,10 +57,10 @@ abstract class AbstractStage<I, O> implements Stage<I, O> {
return this.outputElements;
}
- // protected abstract void execute3();
-
protected abstract void execute4(CommittableQueue<I> elements);
+ protected abstract void execute5(I element);
+
protected final void send(final O element) {
this.outputElements.addToTailUncommitted(element);
diff --git a/src/test/java/teetime/examples/throughput/methodcall/CollectorSink.java b/src/test/java/teetime/examples/throughput/methodcall/CollectorSink.java
index 785c6b6fc7c41c114bad65097f7875f87b52a2a0..90fefc3a21292e4509a5495dacb0283d7e2fffb9 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/CollectorSink.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/CollectorSink.java
@@ -60,6 +60,11 @@ public class CollectorSink<T> extends ConsumerStage<T, Object> {
@Override
protected void execute4(final CommittableQueue<T> elements) {
T element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final T element) {
this.elements.add(element);
if ((this.elements.size() % THRESHOLD) == 0) {
System.out.println("size: " + this.elements.size());
diff --git a/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java b/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java
index 51ecdc300b50f0547dd2d9bd09580954c4072747..4720e0750c10f401fcae0d47453446658796eff1 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java
@@ -17,4 +17,10 @@ public abstract class ConsumerStage<I, O> extends AbstractStage<I, O> {
return output;
}
+ @Override
+ public void executeWithPorts() {
+ I element = this.getInputPort().receive();
+ this.execute5(element);
+ }
+
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis9.java b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis9.java
new file mode 100644
index 0000000000000000000000000000000000000000..3bb16df843bd4cee0797375f7c4f95cdc60d817a
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis9.java
@@ -0,0 +1,119 @@
+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package teetime.examples.throughput.methodcall;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import teetime.examples.throughput.TimestampObject;
+import teetime.framework.core.Analysis;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class MethodCallThroughputAnalysis9 extends Analysis {
+
+ private long numInputObjects;
+ private Callable<TimestampObject> inputObjectCreator;
+ private int numNoopFilters;
+ private List<TimestampObject> timestampObjects;
+ private Runnable runnable;
+
+ @Override
+ public void init() {
+ super.init();
+ this.runnable = this.buildPipeline();
+ }
+
+ /**
+ * @param numNoopFilters
+ * @since 1.10
+ */
+ private Runnable buildPipeline() {
+ @SuppressWarnings("unchecked")
+ final NoopFilter<TimestampObject>[] noopFilters = new NoopFilter[this.numNoopFilters];
+ // create stages
+ final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(this.numInputObjects, this.inputObjectCreator);
+ final StartTimestampFilter startTimestampFilter = new StartTimestampFilter();
+ for (int i = 0; i < noopFilters.length; i++) {
+ noopFilters[i] = new NoopFilter<TimestampObject>();
+ }
+ final StopTimestampFilter stopTimestampFilter = new StopTimestampFilter();
+ final CollectorSink<TimestampObject> collectorSink = new CollectorSink<TimestampObject>(this.timestampObjects);
+
+ final Pipeline<Void, Object> pipeline = new Pipeline<Void, Object>();
+ pipeline.setFirstStage(objectProducer);
+ pipeline.addIntermediateStage(startTimestampFilter);
+ pipeline.addIntermediateStages(noopFilters);
+ pipeline.addIntermediateStage(stopTimestampFilter);
+ pipeline.setLastStage(collectorSink);
+
+ Pipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
+ Pipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
+ for (int i = 0; i < noopFilters.length - 1; i++) {
+ Pipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
+ }
+ Pipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
+ Pipe.connect(stopTimestampFilter.getOutputPort(), collectorSink.getInputPort());
+
+ pipeline.onStart();
+
+ // pipeline.getInputPort().pipe = new Pipe<Void>();
+ // pipeline.getInputPort().pipe.add(new Object());
+
+ // pipeline.getOutputPort().pipe = new Pipe<Void>();
+
+ final Runnable runnable = new Runnable() {
+ @Override
+ public void run() {
+ do {
+ pipeline.executeWithPorts();
+ } while (pipeline.getSchedulingInformation().isActive() && pipeline.isReschedulable());
+ }
+ };
+
+ return runnable;
+ }
+
+ @Override
+ public void start() {
+ super.start();
+ this.runnable.run();
+ }
+
+ public void setInput(final int numInputObjects, final Callable<TimestampObject> inputObjectCreator) {
+ this.numInputObjects = numInputObjects;
+ this.inputObjectCreator = inputObjectCreator;
+ }
+
+ public int getNumNoopFilters() {
+ return this.numNoopFilters;
+ }
+
+ public void setNumNoopFilters(final int numNoopFilters) {
+ this.numNoopFilters = numNoopFilters;
+ }
+
+ public List<TimestampObject> getTimestampObjects() {
+ return this.timestampObjects;
+ }
+
+ public void setTimestampObjects(final List<TimestampObject> timestampObjects) {
+ this.timestampObjects = timestampObjects;
+ }
+}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/NoopFilter.java b/src/test/java/teetime/examples/throughput/methodcall/NoopFilter.java
index f1d1367e852559cdfa00a4de7b77c6dc85b3bc38..17d1abba174ea3ea9282c978364b3d111115a787 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/NoopFilter.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/NoopFilter.java
@@ -39,6 +39,11 @@ public class NoopFilter<T> extends ConsumerStage<T, T> {
@Override
protected void execute4(final CommittableQueue<T> elements) {
T element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final T element) {
this.send(element);
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/ObjectProducer.java b/src/test/java/teetime/examples/throughput/methodcall/ObjectProducer.java
index db409576ca95fdf336195304c8eb4818e3bcfb8f..deeb0df7155cb07f6785bffe0741a4e021239cd5 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/ObjectProducer.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/ObjectProducer.java
@@ -89,6 +89,11 @@ public class ObjectProducer<T> extends ProducerStage<Void, T> {
@Override
protected void execute4(final CommittableQueue<Void> elements) {
+ this.execute5(null);
+ }
+
+ @Override
+ protected void execute5(final Void element) {
if (this.numInputObjects == 0) {
this.setReschedulable(false);
return;
diff --git a/src/test/java/teetime/examples/throughput/methodcall/Pipe.java b/src/test/java/teetime/examples/throughput/methodcall/Pipe.java
index 18afaa07d2e6a1cbf642a6db752d91bf8c49c206..814e6a7425f8007418269dd218ec007138d00d18 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/Pipe.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/Pipe.java
@@ -6,6 +6,12 @@ public class Pipe<T> {
private final CommittableResizableArrayQueue<T> elements = new CommittableResizableArrayQueue<T>(null, 4);
+ public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
+ Pipe<T> pipe = new Pipe<T>();
+ sourcePort.pipe = pipe;
+ targetPort.pipe = pipe;
+ }
+
public void add(final T element) {
this.elements.addToTailUncommitted(element);
this.elements.commit();
diff --git a/src/test/java/teetime/examples/throughput/methodcall/Pipeline.java b/src/test/java/teetime/examples/throughput/methodcall/Pipeline.java
index 8be8637cae36318f037234577d2ba3f8ef0e2d53..82dfe6bf7fc6efe65703298e06b727dedb7179c2 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/Pipeline.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/Pipeline.java
@@ -6,15 +6,15 @@ import java.util.List;
import teetime.util.list.CommittableQueue;
-public class Pipeline<I, O> implements Stage<I, O>, OnDisableListener {
+public class Pipeline<I, O> implements StageWithPort<I, O>, OnDisableListener {
- private Stage firstStage;
- private final List<Stage> intermediateStages = new LinkedList<Stage>();
- private Stage lastStage;
+ private StageWithPort firstStage;
+ private final List<StageWithPort> intermediateStages = new LinkedList<StageWithPort>();
+ private StageWithPort lastStage;
private final SchedulingInformation schedulingInformation = new SchedulingInformation();
- private Stage[] stages;
+ private StageWithPort[] stages;
private Stage parentStage;
private int index;
private int startIndex;
@@ -22,19 +22,19 @@ public class Pipeline<I, O> implements Stage<I, O>, OnDisableListener {
private boolean reschedulable;
- void setFirstStage(final Stage<I, ?> stage) {
+ void setFirstStage(final StageWithPort<I, ?> stage) {
this.firstStage = stage;
}
- void addIntermediateStages(final Stage... stages) {
+ void addIntermediateStages(final StageWithPort... stages) {
this.intermediateStages.addAll(Arrays.asList(stages));
}
- void addIntermediateStage(final Stage stage) {
+ void addIntermediateStage(final StageWithPort stage) {
this.intermediateStages.add(stage);
}
- void setLastStage(final Stage<?, O> stage) {
+ void setLastStage(final StageWithPort<?, O> stage) {
this.lastStage = stage;
}
@@ -59,6 +59,13 @@ public class Pipeline<I, O> implements Stage<I, O>, OnDisableListener {
return queue;
}
+ @Override
+ public void executeWithPorts() {
+ this.stages[0].executeWithPorts();
+
+ this.setReschedulable(this.stages[0].isReschedulable());
+ }
+
void onStart() {
// Pipe pipe = new Pipe();
// this.outputPort.pipe = pipe;
@@ -82,10 +89,10 @@ public class Pipeline<I, O> implements Stage<I, O>, OnDisableListener {
// this.lastStage.getInputPort().pipe = pipe;
int size = 1 + this.intermediateStages.size() + 1;
- this.stages = new Stage[size];
+ this.stages = new StageWithPort[size];
this.stages[0] = this.firstStage;
for (int i = 0; i < this.intermediateStages.size(); i++) {
- Stage<?, ?> stage = this.intermediateStages.get(i);
+ StageWithPort<?, ?> stage = this.intermediateStages.get(i);
this.stages[1 + i] = stage;
}
this.stages[this.stages.length - 1] = this.lastStage;
@@ -176,6 +183,16 @@ public class Pipeline<I, O> implements Stage<I, O>, OnDisableListener {
this.reschedulable = reschedulable;
}
+ @Override
+ public InputPort<I> getInputPort() {
+ return this.firstStage.getInputPort();
+ }
+
+ @Override
+ public OutputPort<O> getOutputPort() {
+ return this.lastStage.getOutputPort();
+ }
+
// @Override
// public OutputPort getOutputPort() {
// return this.lastStage.getOutputPort();
diff --git a/src/test/java/teetime/examples/throughput/methodcall/ProducerStage.java b/src/test/java/teetime/examples/throughput/methodcall/ProducerStage.java
index 262119dc8bfd9eecc35d7b99c0be1ad3bb5116b9..835b78a90f83203975e1afeca1564b5b07df330b 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/ProducerStage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/ProducerStage.java
@@ -20,4 +20,9 @@ public abstract class ProducerStage<I, O> extends AbstractStage<I, O> {
return outputElements;
}
+ @Override
+ public void executeWithPorts() {
+ this.execute5(null);
+ }
+
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/Stage.java b/src/test/java/teetime/examples/throughput/methodcall/Stage.java
index 10eefc4593599d0d9b94b6cae1794a1da8928e42..f30063792029d1d4b62cbf93dd87178c628b0530 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/Stage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/Stage.java
@@ -10,14 +10,10 @@ public interface Stage<I, O> {
// CommittableQueue<O> execute2();
- // InputPort<I> getInputPort();
-
CommittableQueue<O> execute2(CommittableQueue<I> elements);
SchedulingInformation getSchedulingInformation();
- // OutputPort<O> getOutputPort();
-
Stage getParentStage();
void setParentStage(Stage parentStage, int index);
diff --git a/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java b/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java
new file mode 100644
index 0000000000000000000000000000000000000000..46ed73f2ec4dae3dedc91d1a7448f76948fa75e7
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java
@@ -0,0 +1,10 @@
+package teetime.examples.throughput.methodcall;
+
+public interface StageWithPort<I, O> extends Stage<I, O> {
+
+ InputPort<I> getInputPort();
+
+ OutputPort<O> getOutputPort();
+
+ void executeWithPorts();
+}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/StartTimestampFilter.java b/src/test/java/teetime/examples/throughput/methodcall/StartTimestampFilter.java
index 672af104bcab928334390ea0542796f36884b40d..d4d6d5a7b45e6bdfaaf406305cdcdb843d992367 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/StartTimestampFilter.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/StartTimestampFilter.java
@@ -42,6 +42,11 @@ public class StartTimestampFilter extends ConsumerStage<TimestampObject, Timesta
@Override
protected void execute4(final CommittableQueue<TimestampObject> elements) {
TimestampObject element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final TimestampObject element) {
element.setStartTimestamp(System.nanoTime());
this.send(element);
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/StopTimestampFilter.java b/src/test/java/teetime/examples/throughput/methodcall/StopTimestampFilter.java
index d1338b401e9a70c597a6edda62ee2110048a9265..78c333cdf78087eaa824cb7daeda56ea5162d40f 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/StopTimestampFilter.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/StopTimestampFilter.java
@@ -42,6 +42,11 @@ public class StopTimestampFilter extends ConsumerStage<TimestampObject, Timestam
@Override
protected void execute4(final CommittableQueue<TimestampObject> elements) {
TimestampObject element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final TimestampObject element) {
element.setStopTimestamp(System.nanoTime());
this.send(element);
}