From 989c6656b4fe1da61c929c1c565cd40f489dc970 Mon Sep 17 00:00:00 2001
From: Christian Wulf <chw@informatik.uni-kiel.de>
Date: Mon, 16 Jun 2014 14:02:08 +0200
Subject: [PATCH] added more performance tests
---
results/overhead-findings.txt | 19 ++-
...dCallThoughputTimestampAnalysis10Test.java | 74 +++++++++++
...dCallThoughputTimestampAnalysis11Test.java | 74 +++++++++++
.../throughput/methodcall/AbstractStage.java | 23 +++-
.../throughput/methodcall/ConsumerStage.java | 13 +-
.../throughput/methodcall/FixedSizedPipe.java | 34 +++++
.../examples/throughput/methodcall/IPipe.java | 13 ++
.../throughput/methodcall/InputPort.java | 2 +-
.../MethodCallThroughputAnalysis10.java | 125 ++++++++++++++++++
.../MethodCallThroughputAnalysis11.java | 125 ++++++++++++++++++
.../throughput/methodcall/OutputPort.java | 2 +-
.../examples/throughput/methodcall/Pipe.java | 28 +++-
.../methodcall/SingleElementPipe.java | 35 +++++
.../throughput/methodcall/StageWithPort.java | 4 +
.../throughput/methodcall/stage/Pipeline.java | 9 +-
15 files changed, 558 insertions(+), 22 deletions(-)
create mode 100644 src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis10Test.java
create mode 100644 src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis11Test.java
create mode 100644 src/test/java/teetime/examples/throughput/methodcall/FixedSizedPipe.java
create mode 100644 src/test/java/teetime/examples/throughput/methodcall/IPipe.java
create mode 100644 src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis10.java
create mode 100644 src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis11.java
create mode 100644 src/test/java/teetime/examples/throughput/methodcall/SingleElementPipe.java
diff --git a/results/overhead-findings.txt b/results/overhead-findings.txt
index 888c4d3e..64e7d3a4 100644
--- a/results/overhead-findings.txt
+++ b/results/overhead-findings.txt
@@ -1,9 +1,20 @@
-[increases overhead]
--for loop with list vs. array (due to new instantiation of iterator)
--for loop with super type vs. concrete type (due to less JIT optimization possibilities)
+[increases overhead (first is slower)]
+-for loop with list vs. array (reason: due to new instantiation of iterator)
+-for loop with super type vs. concrete type (reason: due to less JIT optimization possibilities)
+-passing by argument vs. instance variable (reason: unknown)
+-pipe with array vs. single element (reason: unknown)
-
[irrelevant w.r.t. overhead]
-foreach vs. index-based iteration
-iterative vs. recursive execution
--
\ No newline at end of file
+-
+
+
+[analysis performance results (50%)]
+
+2: 7400 ns
+9: 9400 ns
+10: 4900 ns (single element pipe)
+11: 7400 ns (fixed sized pipe)
+
\ No newline at end of file
diff --git a/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis10Test.java b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis10Test.java
new file mode 100644
index 00000000..1329d235
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis10Test.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.MethodCallThroughputAnalysis10;
+import teetime.util.StatisticsUtil;
+import teetime.util.StopWatch;
+
+import kieker.common.logging.LogFactory;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class MethodCallThoughputTimestampAnalysis10Test {
+
+ 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 MethodCallThroughputAnalysis10 analysis = new MethodCallThroughputAnalysis10();
+ 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/MethodCallThoughputTimestampAnalysis11Test.java b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis11Test.java
new file mode 100644
index 00000000..3886829f
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis11Test.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.MethodCallThroughputAnalysis11;
+import teetime.util.StatisticsUtil;
+import teetime.util.StopWatch;
+
+import kieker.common.logging.LogFactory;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class MethodCallThoughputTimestampAnalysis11Test {
+
+ 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 MethodCallThroughputAnalysis11 analysis = new MethodCallThroughputAnalysis11();
+ 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 949dbab3..03e99132 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/AbstractStage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/AbstractStage.java
@@ -18,7 +18,7 @@ abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
private int index;
- private Stage successor;
+ private StageWithPort successor;
private boolean reschedulable;
@@ -62,12 +62,18 @@ abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
protected abstract void execute5(I element);
protected final void send(final O element) {
- this.outputElements.addToTailUncommitted(element);
+ // this.outputElements.addToTailUncommitted(element);
+ // this.outputElements.commit();
- this.outputElements.commit();
+ this.getOutputPort().send(element);
+
+ CommittableQueue execute;
do {
- CommittableQueue execute = this.next().execute2(this.outputElements);
+ // execute = this.next().execute2(this.outputElements);
+ // execute = this.next().execute2(this.getOutputPort().pipe.getElements());
+ this.next().executeWithPorts();
} while (this.next().isReschedulable());
+ // } while (execute.size() > 0);
}
@Override
@@ -107,15 +113,20 @@ abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
}
@Override
- public Stage next() {
+ public StageWithPort next() {
return this.successor;
}
@Override
- public void setSuccessor(final Stage<?, ?> successor) {
+ public void setSuccessor(final StageWithPort<?, ?> successor) {
this.successor = successor;
}
+ @Override
+ public void setSuccessor(final Stage<?, ?> successor) {
+ throw new IllegalStateException();
+ }
+
@Override
public boolean isReschedulable() {
return this.reschedulable;
diff --git a/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java b/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java
index 4720e075..7429195a 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/ConsumerStage.java
@@ -6,14 +6,15 @@ public abstract class ConsumerStage<I, O> extends AbstractStage<I, O> {
@Override
public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
- boolean inputIsEmpty = elements.isEmpty();
- if (inputIsEmpty) {
- this.disable();
- return this.outputElements;
- }
+ // the following code block does not harm the performance
+ // boolean inputIsEmpty = elements.isEmpty();
+ // if (inputIsEmpty) {
+ // this.disable();
+ // return this.outputElements;
+ // }
CommittableQueue<O> output = super.execute2(elements);
- this.setReschedulable(!elements.isEmpty()); // costs ~1200 ns on chw-work
+ this.setReschedulable(!elements.isEmpty()); // costs ~1200 ns on chw-work (not reproducible)
return output;
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/FixedSizedPipe.java b/src/test/java/teetime/examples/throughput/methodcall/FixedSizedPipe.java
new file mode 100644
index 00000000..119add4c
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/FixedSizedPipe.java
@@ -0,0 +1,34 @@
+package teetime.examples.throughput.methodcall;
+
+public class FixedSizedPipe<T> implements IPipe<T> {
+
+ private final T[] elements = (T[]) new Object[4];
+ private int lastFreeIndex;
+
+ public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
+ IPipe<T> pipe = new FixedSizedPipe<T>();
+ sourcePort.pipe = pipe;
+ targetPort.pipe = pipe;
+ }
+
+ @Override
+ public void add(final T element) {
+ this.elements[this.lastFreeIndex++] = element;
+ }
+
+ @Override
+ public T removeLast() {
+ return this.elements[--this.lastFreeIndex];
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.lastFreeIndex == 0;
+ }
+
+ @Override
+ public T readLast() {
+ return this.elements[this.lastFreeIndex];
+ }
+
+}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/IPipe.java b/src/test/java/teetime/examples/throughput/methodcall/IPipe.java
new file mode 100644
index 00000000..431998a5
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/IPipe.java
@@ -0,0 +1,13 @@
+package teetime.examples.throughput.methodcall;
+
+public interface IPipe<T> {
+
+ public abstract void add(T element);
+
+ public abstract T removeLast();
+
+ public abstract boolean isEmpty();
+
+ public abstract T readLast();
+
+}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/InputPort.java b/src/test/java/teetime/examples/throughput/methodcall/InputPort.java
index 6f93eb3e..b41efcbd 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/InputPort.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/InputPort.java
@@ -2,7 +2,7 @@ package teetime.examples.throughput.methodcall;
public class InputPort<T> {
- public Pipe<T> pipe;
+ public IPipe<T> pipe;
public T receive() {
T element = this.pipe.removeLast();
diff --git a/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis10.java b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis10.java
new file mode 100644
index 00000000..8f16576d
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis10.java
@@ -0,0 +1,125 @@
+/***************************************************************************
+ * 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.examples.throughput.methodcall.stage.CollectorSink;
+import teetime.examples.throughput.methodcall.stage.NoopFilter;
+import teetime.examples.throughput.methodcall.stage.ObjectProducer;
+import teetime.examples.throughput.methodcall.stage.Pipeline;
+import teetime.examples.throughput.methodcall.stage.StartTimestampFilter;
+import teetime.examples.throughput.methodcall.stage.StopTimestampFilter;
+import teetime.framework.core.Analysis;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class MethodCallThroughputAnalysis10 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);
+
+ SingleElementPipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
+ SingleElementPipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
+ for (int i = 0; i < noopFilters.length - 1; i++) {
+ SingleElementPipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
+ }
+ SingleElementPipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
+ SingleElementPipe.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/MethodCallThroughputAnalysis11.java b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis11.java
new file mode 100644
index 00000000..aa726513
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis11.java
@@ -0,0 +1,125 @@
+/***************************************************************************
+ * 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.examples.throughput.methodcall.stage.CollectorSink;
+import teetime.examples.throughput.methodcall.stage.NoopFilter;
+import teetime.examples.throughput.methodcall.stage.ObjectProducer;
+import teetime.examples.throughput.methodcall.stage.Pipeline;
+import teetime.examples.throughput.methodcall.stage.StartTimestampFilter;
+import teetime.examples.throughput.methodcall.stage.StopTimestampFilter;
+import teetime.framework.core.Analysis;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class MethodCallThroughputAnalysis11 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);
+
+ FixedSizedPipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
+ FixedSizedPipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
+ for (int i = 0; i < noopFilters.length - 1; i++) {
+ FixedSizedPipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
+ }
+ FixedSizedPipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
+ FixedSizedPipe.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/OutputPort.java b/src/test/java/teetime/examples/throughput/methodcall/OutputPort.java
index 41c4b0d2..19f2bc5e 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/OutputPort.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/OutputPort.java
@@ -2,7 +2,7 @@ package teetime.examples.throughput.methodcall;
public class OutputPort<T> {
- public Pipe<T> pipe;
+ public IPipe<T> pipe;
public void send(final T element) {
this.pipe.add(element);
diff --git a/src/test/java/teetime/examples/throughput/methodcall/Pipe.java b/src/test/java/teetime/examples/throughput/methodcall/Pipe.java
index 814e6a74..3df16ac2 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/Pipe.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/Pipe.java
@@ -2,31 +2,55 @@ package teetime.examples.throughput.methodcall;
import teetime.util.list.CommittableResizableArrayQueue;
-public class Pipe<T> {
+public class Pipe<T> implements IPipe<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>();
+ IPipe<T> pipe = new Pipe<T>();
sourcePort.pipe = pipe;
targetPort.pipe = pipe;
}
+ /*
+ * (non-Javadoc)
+ *
+ * @see teetime.examples.throughput.methodcall.IPipe#add(T)
+ */
+ @Override
public void add(final T element) {
this.elements.addToTailUncommitted(element);
this.elements.commit();
}
+ /*
+ * (non-Javadoc)
+ *
+ * @see teetime.examples.throughput.methodcall.IPipe#removeLast()
+ */
+ @Override
public T removeLast() {
T element = this.elements.removeFromHeadUncommitted();
this.elements.commit();
return element;
}
+ /*
+ * (non-Javadoc)
+ *
+ * @see teetime.examples.throughput.methodcall.IPipe#isEmpty()
+ */
+ @Override
public boolean isEmpty() {
return this.elements.isEmpty();
}
+ /*
+ * (non-Javadoc)
+ *
+ * @see teetime.examples.throughput.methodcall.IPipe#readLast()
+ */
+ @Override
public T readLast() {
return this.elements.getTail();
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/SingleElementPipe.java b/src/test/java/teetime/examples/throughput/methodcall/SingleElementPipe.java
new file mode 100644
index 00000000..ac8a0e65
--- /dev/null
+++ b/src/test/java/teetime/examples/throughput/methodcall/SingleElementPipe.java
@@ -0,0 +1,35 @@
+package teetime.examples.throughput.methodcall;
+
+public class SingleElementPipe<T> implements IPipe<T> {
+
+ private T element;
+
+ public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
+ IPipe<T> pipe = new SingleElementPipe<T>();
+ sourcePort.pipe = pipe;
+ targetPort.pipe = pipe;
+ }
+
+ @Override
+ public void add(final T element) {
+ this.element = element;
+ }
+
+ @Override
+ public T removeLast() {
+ T temp = this.element;
+ this.element = null;
+ return temp;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return this.element != null;
+ }
+
+ @Override
+ public T readLast() {
+ return this.element;
+ }
+
+}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java b/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java
index 46ed73f2..d6953540 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/StageWithPort.java
@@ -7,4 +7,8 @@ public interface StageWithPort<I, O> extends Stage<I, O> {
OutputPort<O> getOutputPort();
void executeWithPorts();
+
+ // void executeWithPorts(Object element);
+
+ void setSuccessor(StageWithPort<?, ?> successor);
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java b/src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java
index 23c63395..10ca670c 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java
@@ -104,13 +104,13 @@ public class Pipeline<I, O> implements StageWithPort<I, O>, OnDisableListener {
this.stages[this.stages.length - 1] = this.lastStage;
for (int i = 0; i < this.stages.length; i++) {
- Stage<?, ?> stage = this.stages[i];
+ StageWithPort<?, ?> stage = this.stages[i];
stage.setParentStage(this, i);
stage.setListener(this);
}
for (int i = 0; i < this.stages.length - 1; i++) {
- Stage<?, ?> stage = this.stages[i];
+ StageWithPort<?, ?> stage = this.stages[i];
stage.setSuccessor(this.stages[i + 1]);
}
}
@@ -180,6 +180,11 @@ public class Pipeline<I, O> implements StageWithPort<I, O>, OnDisableListener {
throw new IllegalStateException();
}
+ @Override
+ public void setSuccessor(final StageWithPort<?, ?> successor) {
+ throw new IllegalStateException();
+ }
+
@Override
public boolean isReschedulable() {
return this.reschedulable;
--
GitLab