fixed analysis 11

results/overhead-findings.txt

@@ -26,7 +26,7 @@
 	11: 8200 ns (executeWithPorts: fixed sized pipe; with CircularArray(int) w/o mask)
 	11: 7800 ns (executeWithPorts: fixed sized pipe; with setReschedulable() after each read)
 	11: 8200 ns (executeWithPorts: fixed sized pipe; with setReschedulable() after each read; non-final elements)
-	11: 7000 ns (executeWithPorts: fixed sized pipe; with setReschedulable() after each read; non-final elements; pipeline searches for firstStageIndex)
+	11: 7800 ns (executeWithPorts: fixed sized pipe; with setReschedulable() after each read; non-final elements; pipeline searches for firstStageIndex)
 12: 3300 ns (recursive; argument/return w/o pipe)
 13: 3300 ns (recursive; argument/return w/o pipe; w/o pipeline class)
 14: 21,000 ns (spsc pipe)

src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis11Test.java

@@ -17,11 +17,11 @@ 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.Closure;
 import teetime.examples.throughput.methodcall.MethodCallThroughputAnalysis11;
 import teetime.util.StatisticsUtil;
 import teetime.util.StopWatch;
@@ -53,9 +53,9 @@ public class MethodCallThoughputTimestampAnalysis11Test {
 		final MethodCallThroughputAnalysis11 analysis = new MethodCallThroughputAnalysis11();
 		analysis.setNumNoopFilters(NUM_NOOP_FILTERS);
 		analysis.setTimestampObjects(timestampObjects);
-		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();
 			}
 		});

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis11.java

@@ -16,7 +16,6 @@
 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;
@@ -35,7 +34,7 @@ import teetime.framework.core.Analysis;
 public class MethodCallThroughputAnalysis11 extends Analysis {
 
 	private long numInputObjects;
-	private Callable<TimestampObject> inputObjectCreator;
+	private Closure<Void, TimestampObject> inputObjectCreator;
 	private int numNoopFilters;
 	private List<TimestampObject> timestampObjects;
 	private Runnable runnable;
@@ -43,18 +42,19 @@ public class MethodCallThroughputAnalysis11 extends Analysis {
 	@Override
 	public void init() {
 		super.init();
-		this.runnable = this.buildPipeline();
+		Pipeline<Void, Object> pipeline = this.buildPipeline(this.numInputObjects, this.inputObjectCreator);
+		this.runnable = new RunnableStage(pipeline);
 	}
 
 	/**
 	 * @param numNoopFilters
 	 * @since 1.10
 	 */
-	private Runnable buildPipeline() {
+	private Pipeline<Void, Object> buildPipeline(final long numInputObjects, final Closure<Void, TimestampObject> inputObjectCreator) {
 		@SuppressWarnings("unchecked")
 		final NoopFilter<TimestampObject>[] noopFilters = new NoopFilter[this.numNoopFilters];
 		// create stages
-		final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(this.numInputObjects, this.inputObjectCreator);
+		final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(numInputObjects, inputObjectCreator);
 		final StartTimestampFilter startTimestampFilter = new StartTimestampFilter();
 		for (int i = 0; i < noopFilters.length; i++) {
 			noopFilters[i] = new NoopFilter<TimestampObject>();
@@ -83,22 +83,7 @@ public class MethodCallThroughputAnalysis11 extends Analysis {
 		UnorderedGrowablePipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
 		UnorderedGrowablePipe.connect(stopTimestampFilter.getOutputPort(), collectorSink.getInputPort());
 
-		// 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() {
-				pipeline.onStart();
-				do {
-					pipeline.executeWithPorts();
-				} while (pipeline.getSchedulingInformation().isActive() && pipeline.isReschedulable());
-			}
-		};
-
-		return runnable;
+		return pipeline;
 	}
 
 	@Override
@@ -107,7 +92,7 @@ public class MethodCallThroughputAnalysis11 extends Analysis {
 		this.runnable.run();
 	}
 
-	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;
 	}

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis14.java

@@ -77,23 +77,7 @@ public class MethodCallThroughputAnalysis14 extends Analysis {
 		SpScPipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
 		SpScPipe.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;
+		return new RunnableStage(pipeline);
 	}
 
 	@Override

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis15.java

@@ -64,18 +64,7 @@ public class MethodCallThroughputAnalysis15 extends Analysis {
 		pipeline.setFirstStage(this.clock);
 		pipeline.setLastStage(new EndStage<Long>());
 
-		pipeline.onStart();
-
-		final Runnable runnable = new Runnable() {
-			@Override
-			public void run() {
-				do {
-					pipeline.executeWithPorts();
-				} while (pipeline.isReschedulable());
-			}
-		};
-
-		return runnable;
+		return new RunnableStage(pipeline);
 	}
 
 	/**

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis17.java

@@ -126,7 +126,7 @@ public class MethodCallThroughputAnalysis17 extends Analysis {
 
 		pipeline.setLastStage(distributor);
 		// pipeline.setLastStage(sink);
-		pipeline.setLastStage(new EndStage<TimestampObject>());
+		// pipeline.setLastStage(new EndStage<TimestampObject>());
 
 		// UnorderedGrowablePipe.connect(objectProducer.getOutputPort(), sink.getInputPort());
 		// objectProducer.getOutputPort().pipe = new UnorderedGrowablePipe<TimestampObject>();

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis2.java

@@ -86,7 +86,7 @@ public class MethodCallThroughputAnalysis2 extends Analysis {
 
 				do {
 					outputQueue = pipeline.execute2(inputQueue);
-				} while (pipeline.getSchedulingInformation().isActive() && pipeline.isReschedulable());
+				} while (pipeline.isReschedulable());
 			}
 		};
 

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis9.java

@@ -77,23 +77,7 @@ public class MethodCallThroughputAnalysis9 extends Analysis {
 		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;
+		return new RunnableStage(pipeline);
 	}
 
 	@Override