added more performance tests

results/overhead-findings.txt

-[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

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

+/***************************************************************************
+ * 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);
+	}
+
+}

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

+/***************************************************************************
+ * 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);
+	}
+
+}

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;

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;
 	}
 

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

+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];
+	}
+
+}

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

+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();
+
+}

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();

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

+/***************************************************************************
+ * 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;
+	}
+}

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

+/***************************************************************************
+ * 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;
+	}
+}

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);

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();
 	}

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

+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;
+	}
+
+}

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);
 }

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;