pipe calls the target stage from now on if it is an intra pipe

src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java

@@ -48,11 +48,13 @@ public abstract class AbstractStage implements StageWithPort {
 			return false;
 		}
 
-		StageWithPort next = outputPort.getCachedTargetStage();
+		outputPort.reportNewElement();
 
-		do {
-			next.executeWithPorts(); // PERFORMANCE use the return value as indicator for re-schedulability instead
-		} while (next.isReschedulable());
+		// StageWithPort next = outputPort.getCachedTargetStage();
+		//
+		// do {
+		// next.executeWithPorts(); // PERFORMANCE use the return value as indicator for re-schedulability instead
+		// } while (next.isReschedulable());
 
 		return true;
 	}

src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java

@@ -2,7 +2,7 @@ package teetime.variant.methodcallWithPorts.framework.core;
 
 import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
 
-public class OutputPort<T> extends AbstractPort<T> {
+public final class OutputPort<T> extends AbstractPort<T> {
 
 	/**
 	 * Performance cache: Avoids the following method chain
@@ -11,7 +11,7 @@ public class OutputPort<T> extends AbstractPort<T> {
 	 * this.getPipe().getTargetPort().getOwningStage()
 	 * </pre>
 	 */
-	private StageWithPort cachedTargetStage;
+	// private StageWithPort cachedTargetStage;
 
 	OutputPort() {
 		super();
@@ -26,16 +26,21 @@ public class OutputPort<T> extends AbstractPort<T> {
 		return this.pipe.add(element);
 	}
 
-	public StageWithPort getCachedTargetStage() {
-		return this.cachedTargetStage;
-	}
+	// public StageWithPort getCachedTargetStage() {
+	// return this.cachedTargetStage;
+	// }
 
+	@Deprecated
 	public void setCachedTargetStage(final StageWithPort cachedTargetStage) {
-		this.cachedTargetStage = cachedTargetStage;
+		// this.cachedTargetStage = cachedTargetStage;
 	}
 
 	public void sendSignal(final Signal signal) {
 		this.pipe.setSignal(signal);
 	}
 
+	public void reportNewElement() {
+		this.pipe.reportNewElement();
+	}
+
 }

src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/AbstractPipe.java

 package teetime.variant.methodcallWithPorts.framework.core.pipe;
 
 import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.StageWithPort;
 
 public abstract class AbstractPipe<T> implements IPipe<T> {
 
-	// private final AtomicBoolean closed = new AtomicBoolean();
 	private InputPort<T> targetPort;
 
-	// @Override
-	// public boolean isClosed() {
-	// return this.closed.get();
-	// }
-	//
-	// @Override
-	// public void close() {
-	// this.closed.lazySet(true); // lazySet is legal due to our single-writer requirement
-	// }
+	/**
+	 * Performance cache: Avoids the following method chain
+	 *
+	 * <pre>
+	 * this.getPipe().getTargetPort().getOwningStage()
+	 * </pre>
+	 */
+	protected StageWithPort cachedTargetStage;
 
 	@Override
 	public InputPort<T> getTargetPort() {
@@ -25,6 +24,7 @@ public abstract class AbstractPipe<T> implements IPipe<T> {
 	@Override
 	public void setTargetPort(final InputPort<T> targetPort) {
 		this.targetPort = targetPort;
+		this.cachedTargetStage = targetPort.getOwningStage();
 	}
 
 }

src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/DummyPipe.java

@@ -52,4 +52,9 @@ public final class DummyPipe implements IPipe {
 	@Override
 	public void connectPorts(final OutputPort sourcePort, final InputPort targetPort) {}
 
+	@Override
+	public void reportNewElement() {
+		// do nothing
+	}
+
 }

src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java

@@ -29,4 +29,6 @@ public interface IPipe<T> {
 	// BETTER change signature to allow {OutputPort<T>, OutputPort<A0 extends T>, OutputPort<A1 extends T>, ...}
 	void connectPorts(OutputPort<T> sourcePort, InputPort<T> targetPort);
 
+	void reportNewElement();
+
 }

src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IntraThreadPipe.java

@@ -7,8 +7,13 @@ public abstract class IntraThreadPipe<T> extends AbstractPipe<T> {
 	@Override
 	public void setSignal(final Signal signal) {
 		if (this.getTargetPort() != null) {
-			this.getTargetPort().getOwningStage().onSignal(signal, this.getTargetPort());
+			this.cachedTargetStage.onSignal(signal, this.getTargetPort());
 		}
 	}
 
+	@Override
+	public final void reportNewElement() {
+		this.cachedTargetStage.executeWithPorts();
+	}
+
 }

src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java

@@ -83,4 +83,9 @@ public class SpScPipe<T> extends AbstractPipe<T> {
 		return this.signal.get();
 	}
 
+	@Override
+	public void reportNewElement() {
+		// do nothing
+	}
+
 }

src/test/java/teetime/variant/methodcallWithPorts/examples/ChwWorkComparisonMethodcallWithPorts.java

@@ -89,10 +89,10 @@ public class ChwWorkComparisonMethodcallWithPorts implements PerformanceCheckPro
 		// assertEquals(53, value17, 4.1); // +0
 
 		// since 27.08.2014 (incl.)
-		assertEquals(112, value14, 5.1); // +16
-		assertEquals(42, value10, 2.1); // +16
-		assertEquals(41, value11, 4.1); // -3
-		assertEquals(42, value9, 2.1); // +6
+		assertEquals(102, value14, 5.1); // +16
+		assertEquals(56, value10, 2.1); // +30
+		assertEquals(64, value11, 4.1); // +15
+		assertEquals(77, value9, 2.1); // +35
 		assertEquals(44, value15, 4.1); // +0
 		assertEquals(53, value17, 4.1); // +0
 

src/test/java/teetime/variant/methodcallWithPorts/examples/experiment14/MethodCallThroughputAnalysis14.java

@@ -23,7 +23,9 @@ import teetime.variant.explicitScheduling.framework.core.Analysis;
 import teetime.variant.methodcallWithPorts.framework.core.Pipeline;
 import teetime.variant.methodcallWithPorts.framework.core.RunnableStage;
 import teetime.variant.methodcallWithPorts.framework.core.StageWithPort;
-import teetime.variant.methodcallWithPorts.framework.core.pipe.SpScPipe;
+import teetime.variant.methodcallWithPorts.framework.core.pipe.IPipe;
+import teetime.variant.methodcallWithPorts.framework.core.pipe.PipeFactory;
+import teetime.variant.methodcallWithPorts.framework.core.pipe.PipeFactory.ThreadCommunication;
 import teetime.variant.methodcallWithPorts.stage.CollectorSink;
 import teetime.variant.methodcallWithPorts.stage.NoopFilter;
 import teetime.variant.methodcallWithPorts.stage.ObjectProducer;
@@ -37,8 +39,6 @@ import teetime.variant.methodcallWithPorts.stage.StopTimestampFilter;
  */
 public class MethodCallThroughputAnalysis14 extends Analysis {
 
-	private static final int SPSC_INITIAL_CAPACITY = 4;
-
 	private long numInputObjects;
 	private ConstructorClosure<TimestampObject> inputObjectCreator;
 	private int numNoopFilters;
@@ -75,13 +75,19 @@ public class MethodCallThroughputAnalysis14 extends Analysis {
 		pipeline.addIntermediateStage(stopTimestampFilter);
 		pipeline.setLastStage(collectorSink);
 
-		SpScPipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort(), SPSC_INITIAL_CAPACITY);
-		SpScPipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort(), SPSC_INITIAL_CAPACITY);
+		PipeFactory pipeFactory = new PipeFactory();
+		IPipe<TimestampObject> pipe = pipeFactory.create(ThreadCommunication.INTRA);
+		pipe.connectPorts(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
+		pipe = pipeFactory.create(ThreadCommunication.INTRA);
+		pipe.connectPorts(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
 		for (int i = 0; i < noopFilters.length - 1; i++) {
-			SpScPipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort(), SPSC_INITIAL_CAPACITY);
+			pipe = pipeFactory.create(ThreadCommunication.INTRA);
+			pipe.connectPorts(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
 		}
-		SpScPipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort(), SPSC_INITIAL_CAPACITY);
-		SpScPipe.connect(stopTimestampFilter.getOutputPort(), collectorSink.getInputPort(), SPSC_INITIAL_CAPACITY);
+		pipe = pipeFactory.create(ThreadCommunication.INTRA);
+		pipe.connectPorts(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
+		pipe = pipeFactory.create(ThreadCommunication.INTRA);
+		pipe.connectPorts(stopTimestampFilter.getOutputPort(), collectorSink.getInputPort());
 
 		return pipeline;
 	}