diff --git a/results/overhead-findings.txt b/results/overhead-findings.txt
index 8c6effa478cfb8bb8b2fe8ddffda743059456ba2..716448668817907baac6044e2bfc9b8b70f82c48 100644
--- a/results/overhead-findings.txt
+++ b/results/overhead-findings.txt
@@ -31,4 +31,4 @@
13: 3300 ns (recursive; argument/return w/o pipe; w/o pipeline class)
14: 21,000 ns (spsc pipe)
16: 14,500 ns (with distributor thread)
-17: 9800 ns (as 16, but with direct feeding of SpScPipe)
+17: 8600 ns (as 16, but with direct feeding of SpScPipe)
diff --git a/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis16Test.java b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis16Test.java
index d6c1aa27e26cbc42ab04946a6a3519a6c871ccc5..8332575461b4bb56b32ed8fbd09a5d6bee767c2c 100644
--- a/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis16Test.java
+++ b/src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis16Test.java
@@ -16,11 +16,11 @@
package teetime.examples.throughput;
import java.util.List;
-import java.util.concurrent.Callable;
import org.junit.Before;
import org.junit.Test;
+import teetime.examples.throughput.methodcall.Closure;
import teetime.examples.throughput.methodcall.MethodCallThroughputAnalysis16;
import teetime.util.ListUtil;
import teetime.util.StatisticsUtil;
@@ -51,9 +51,9 @@ public class MethodCallThoughputTimestampAnalysis16Test {
final MethodCallThroughputAnalysis16 analysis = new MethodCallThroughputAnalysis16();
analysis.setNumNoopFilters(NUM_NOOP_FILTERS);
- analysis.setInput(NUM_OBJECTS_TO_CREATE, new Callable<TimestampObject>() {
+ analysis.setInput(NUM_OBJECTS_TO_CREATE, new Closure<Void, TimestampObject>() {
@Override
- public TimestampObject call() throws Exception {
+ public TimestampObject execute(final Void element) {
return new TimestampObject();
}
});
@@ -70,4 +70,10 @@ public class MethodCallThoughputTimestampAnalysis16Test {
List<TimestampObject> timestampObjects = ListUtil.merge(analysis.getTimestampObjectsList());
StatisticsUtil.printStatistics(stopWatch.getDurationInNs(), timestampObjects);
}
+
+ public static void main(final String[] args) {
+ MethodCallThoughputTimestampAnalysis16Test test = new MethodCallThoughputTimestampAnalysis16Test();
+ test.before();
+ test.testWithManyObjects();
+ }
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/Closure.java b/src/test/java/teetime/examples/throughput/methodcall/Closure.java
index de9c4e64e158acf5dc2f6db6e9b415731f2872b6..f208c9a1a6e6b698dd150e109256a0616bc41df1 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/Closure.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/Closure.java
@@ -2,5 +2,5 @@ package teetime.examples.throughput.methodcall;
public interface Closure<I, O> {
- O execute(I element);
+ public abstract O execute(I element);
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis16.java b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis16.java
index 86cfc65f623018d47ebff2bfbb12e3a415ea7793..d3a2d10f45206d39d49286740c171ed40d395195 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis16.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis16.java
@@ -18,7 +18,6 @@ package teetime.examples.throughput.methodcall;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
-import java.util.concurrent.Callable;
import teetime.examples.throughput.TimestampObject;
import teetime.examples.throughput.methodcall.stage.CollectorSink;
@@ -41,7 +40,7 @@ public class MethodCallThroughputAnalysis16 extends Analysis {
private static final int NUM_WORKER_THREADS = Runtime.getRuntime().availableProcessors();
private int numInputObjects;
- private Callable<TimestampObject> inputObjectCreator;
+ private Closure<Void, TimestampObject> inputObjectCreator;
private int numNoopFilters;
private final List<List<TimestampObject>> timestampObjectsList = new LinkedList<List<TimestampObject>>();
@@ -54,7 +53,7 @@ public class MethodCallThroughputAnalysis16 extends Analysis {
@Override
public void init() {
super.init();
- Pipeline<Void, TimestampObject> producerPipeline = this.buildProducerPipeline();
+ Pipeline<Void, TimestampObject> producerPipeline = this.buildProducerPipeline(this.numInputObjects, this.inputObjectCreator);
this.producerThread = new Thread(new RunnableStage(producerPipeline));
int numWorkerThreads = Math.min(NUM_WORKER_THREADS, 1); // only for testing purpose
@@ -78,8 +77,8 @@ public class MethodCallThroughputAnalysis16 extends Analysis {
}
}
- private Pipeline<Void, TimestampObject> buildProducerPipeline() {
- final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(this.numInputObjects, this.inputObjectCreator);
+ private Pipeline<Void, TimestampObject> buildProducerPipeline(final int numInputObjects, final Closure<Void, TimestampObject> inputObjectCreator) {
+ final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(numInputObjects, inputObjectCreator);
this.distributor = new Distributor<TimestampObject>();
final Pipeline<Void, TimestampObject> pipeline = new Pipeline<Void, TimestampObject>();
@@ -146,7 +145,7 @@ public class MethodCallThroughputAnalysis16 extends Analysis {
}
}
- public void setInput(final int numInputObjects, final Callable<TimestampObject> inputObjectCreator) {
+ public void setInput(final int numInputObjects, final Closure<Void, TimestampObject> inputObjectCreator) {
this.numInputObjects = numInputObjects;
this.inputObjectCreator = inputObjectCreator;
}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis17.java b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis17.java
index 79ce596e309e322c0160d98e08f7af90811e20de..ec9b620c506da4500302e4633f3d219280383a1b 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis17.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis17.java
@@ -87,7 +87,7 @@ public class MethodCallThroughputAnalysis17 extends Analysis {
// this.producerThread.start();
// this.producerThread.run();
- new RunnableStage(producerPipeline).run();
+ // new RunnableStage(producerPipeline).run();
// Pipeline<Void, TimestampObject> stage = producerPipeline;
// stage.onStart();
diff --git a/src/test/java/teetime/examples/throughput/methodcall/SchedulingInformation.java b/src/test/java/teetime/examples/throughput/methodcall/SchedulingInformation.java
deleted file mode 100644
index 0286f43beeea7296d67cd0a25d894f45844cc35d..0000000000000000000000000000000000000000
--- a/src/test/java/teetime/examples/throughput/methodcall/SchedulingInformation.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package teetime.examples.throughput.methodcall;
-
-public class SchedulingInformation {
-
- private boolean active = true;
-
- public boolean isActive() {
- return this.active;
- }
-
- public void setActive(final boolean active) {
- this.active = active;
- }
-}
diff --git a/src/test/java/teetime/examples/throughput/methodcall/stage/AbstractStage.java b/src/test/java/teetime/examples/throughput/methodcall/stage/AbstractStage.java
index 4b5502f79562f5d5cad330009a8cdfec3c45d574..50bb9f9248ee85bb0f2cb65a400cb248fdadd1fc 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/stage/AbstractStage.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/stage/AbstractStage.java
@@ -12,7 +12,7 @@ public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
private final OutputPort<O> outputPort = new OutputPort<O>();
// protected final CommittableQueue<O> outputElements = new CommittableResizableArrayQueue<O>(null, 4);
- protected final CommittableQueue<O> outputElements = null;
+ private final CommittableQueue<O> outputElements = null;
private Stage<?, ?> parentStage;
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 d77aa89906af2ebd7c467e32964a1059cf828cfe..86e33fa985308a942033e07204a3a078666846f3 100644
--- a/src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java
+++ b/src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java
@@ -6,7 +6,6 @@ import java.util.List;
import teetime.examples.throughput.methodcall.InputPort;
import teetime.examples.throughput.methodcall.OutputPort;
-import teetime.examples.throughput.methodcall.SchedulingInformation;
import teetime.examples.throughput.methodcall.Stage;
import teetime.examples.throughput.methodcall.StageWithPort;
import teetime.util.list.CommittableQueue;
@@ -14,13 +13,11 @@ import teetime.util.list.CommittableQueue;
public class Pipeline<I, O> implements StageWithPort<I, O> {
private StageWithPort<I, ?> firstStage;
- private final List<StageWithPort> intermediateStages = new LinkedList<StageWithPort>();
+ private final List<StageWithPort<?, ?>> intermediateStages = new LinkedList<StageWithPort<?, ?>>();
private StageWithPort<?, O> lastStage;
- private final SchedulingInformation schedulingInformation = new SchedulingInformation();
-
- private StageWithPort[] stages;
- private Stage parentStage;
+ private StageWithPort<?, ?>[] stages;
+ private Stage<?, ?> parentStage;
private int index;
private int startIndex;
@@ -31,7 +28,7 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
this.firstStage = stage;
}
- public void addIntermediateStages(final StageWithPort... stages) {
+ public void addIntermediateStages(final StageWithPort<?, ?>... stages) {
this.intermediateStages.addAll(Arrays.asList(stages));
}
@@ -59,8 +56,8 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
// queue = stage.execute2(queue);
// }
- this.stages[0].execute2(elements);
- this.setReschedulable(this.stages[0].isReschedulable());
+ this.firstStage.execute2(elements);
+ this.setReschedulable(this.firstStage.isReschedulable());
return queue;
}
@@ -132,7 +129,7 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
for (int i = 0; i < this.stages.length; i++) {
StageWithPort<?, ?> stage = this.stages[i];
- stage.setParentStage(this, i);
+ // stage.setParentStage(this, i);
// stage.setListener(this);
}