added distribution concept

src/main/java/teetime/framework/core/Analysis.java

@@ -31,7 +31,7 @@ public class Analysis {
 
 	}
 
-	public void terminate() {
+	public void onTerminate() {
 
 	}
 }

src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis16Test.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.LinkedList;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import teetime.examples.throughput.methodcall.MethodCallThroughputAnalysis16;
+import teetime.util.StatisticsUtil;
+import teetime.util.StopWatch;
+
+import kieker.common.logging.LogFactory;
+
+/**
+ * @author Christian Wulf
+ * 
+ * @since 1.10
+ */
+public class MethodCallThoughputTimestampAnalysis16Test {
+
+	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 MethodCallThroughputAnalysis16 analysis = new MethodCallThroughputAnalysis16();
+		analysis.setNumNoopFilters(NUM_NOOP_FILTERS);
+		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();
+			analysis.onTerminate();
+		}
+
+		// merge
+		List<TimestampObject> timestampObjects = new LinkedList<TimestampObject>();
+		for (List<TimestampObject> timestampObjectList : analysis.getTimestampObjectsList()) {
+			timestampObjects.addAll(timestampObjectList);
+		}
+		StatisticsUtil.printStatistics(stopWatch.getDurationInNs(), timestampObjects);
+	}
+}

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

@@ -62,22 +62,26 @@ public class MethodCallThroughputAnalysis11 extends Analysis {
 		final StopTimestampFilter stopTimestampFilter = new StopTimestampFilter();
 		final CollectorSink<TimestampObject> collectorSink = new CollectorSink<TimestampObject>(this.timestampObjects);
 
+		// Relay<TimestampObject> relay = new Relay<TimestampObject>();
+
 		final Pipeline<Void, Object> pipeline = new Pipeline<Void, Object>();
 		pipeline.setFirstStage(objectProducer);
+		// pipeline.addIntermediateStage(relay);
 		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());
+		UnorderedGrowablePipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
+		// UnorderedGrowablePipe.connect(objectProducer.getOutputPort(), relay.getInputPort());
+		// UnorderedGrowablePipe.connect(relay.getOutputPort(), startTimestampFilter.getInputPort());
+
+		UnorderedGrowablePipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
 		for (int i = 0; i < noopFilters.length - 1; i++) {
-			FixedSizedPipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
+			UnorderedGrowablePipe.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();
+		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());
@@ -87,6 +91,7 @@ public class MethodCallThroughputAnalysis11 extends Analysis {
 		final Runnable runnable = new Runnable() {
 			@Override
 			public void run() {
+				pipeline.onStart();
 				do {
 					pipeline.executeWithPorts();
 				} while (pipeline.getSchedulingInformation().isActive() && pipeline.isReschedulable());

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

@@ -105,15 +105,15 @@ public class MethodCallThroughputAnalysis15 extends Analysis {
 
 		SpScPipe.connect(clock.getOutputPort(), delay.getTimestampTriggerInputPort());
 
-		FixedSizedPipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
-		FixedSizedPipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
+		UnorderedGrowablePipe.connect(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
+		UnorderedGrowablePipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
 		for (int i = 0; i < noopFilters.length - 1; i++) {
-			FixedSizedPipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
+			UnorderedGrowablePipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
 		}
-		FixedSizedPipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
-		FixedSizedPipe.connect(stopTimestampFilter.getOutputPort(), delay.getInputPort());
+		UnorderedGrowablePipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
+		UnorderedGrowablePipe.connect(stopTimestampFilter.getOutputPort(), delay.getInputPort());
 
-		FixedSizedPipe.connect(delay.getOutputPort(), collectorSink.getInputPort());
+		UnorderedGrowablePipe.connect(delay.getOutputPort(), collectorSink.getInputPort());
 
 		pipeline.onStart();
 

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis16.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.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;
+import teetime.examples.throughput.methodcall.stage.Distributor;
+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.Relay;
+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 MethodCallThroughputAnalysis16 extends Analysis {
+
+	private static final int NUM_WORKER_THREADS = Runtime.getRuntime().availableProcessors();
+
+	private int numInputObjects;
+	private Callable<TimestampObject> inputObjectCreator;
+	private int numNoopFilters;
+
+	private final List<List<TimestampObject>> timestampObjectsList = new LinkedList<List<TimestampObject>>();
+
+	private Distributor<TimestampObject> distributor;
+	private Thread producerThread;
+
+	private Thread[] workerThreads;
+
+	@Override
+	public void init() {
+		super.init();
+		Runnable producerRunnable = this.buildProducerPipeline();
+		this.producerThread = new Thread(producerRunnable);
+
+		int numWorkerThreads = Math.min(NUM_WORKER_THREADS, 1); // only for testing purpose
+
+		this.workerThreads = new Thread[numWorkerThreads];
+		for (int i = 0; i < this.workerThreads.length; i++) {
+			List<TimestampObject> resultList = new ArrayList<TimestampObject>(this.numInputObjects);
+			this.timestampObjectsList.add(resultList);
+
+			Runnable workerRunnable = this.buildPipeline(this.distributor, resultList);
+			this.workerThreads[i] = new Thread(workerRunnable);
+		}
+
+		this.producerThread.start();
+
+		try {
+			this.producerThread.join();
+		} catch (InterruptedException e1) {
+			// TODO Auto-generated catch block
+			e1.printStackTrace();
+		}
+	}
+
+	private Runnable buildProducerPipeline() {
+		final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(this.numInputObjects, this.inputObjectCreator);
+		this.distributor = new Distributor<TimestampObject>();
+
+		final Pipeline<Void, TimestampObject> pipeline = new Pipeline<Void, TimestampObject>();
+		pipeline.setFirstStage(objectProducer);
+		pipeline.setLastStage(this.distributor);
+
+		UnorderedGrowablePipe.connect(objectProducer.getOutputPort(), this.distributor.getInputPort());
+
+		final Runnable runnable = new Runnable() {
+			@Override
+			public void run() {
+				pipeline.onStart();
+				do {
+					pipeline.executeWithPorts();
+				} while (pipeline.isReschedulable());
+				// System.out.println("buildProducerPipeline finished");
+			}
+		};
+
+		return runnable;
+	}
+
+	/**
+	 * @param numNoopFilters
+	 * @since 1.10
+	 */
+	private Runnable buildPipeline(final Distributor<TimestampObject> distributor, final List<TimestampObject> timestampObjects) {
+		Relay<TimestampObject> relay = new Relay<TimestampObject>();
+		@SuppressWarnings("unchecked")
+		final NoopFilter<TimestampObject>[] noopFilters = new NoopFilter[this.numNoopFilters];
+		// create stages
+		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>(timestampObjects);
+
+		final Pipeline<TimestampObject, Object> pipeline = new Pipeline<TimestampObject, Object>();
+		pipeline.setFirstStage(relay);
+		pipeline.addIntermediateStage(startTimestampFilter);
+		pipeline.addIntermediateStages(noopFilters);
+		pipeline.addIntermediateStage(stopTimestampFilter);
+		pipeline.setLastStage(collectorSink);
+
+		OrderedGrowableArrayPipe.connect(distributor.getNewOutputPort(), relay.getInputPort());
+
+		UnorderedGrowablePipe.connect(relay.getOutputPort(), startTimestampFilter.getInputPort());
+
+		UnorderedGrowablePipe.connect(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
+		for (int i = 0; i < noopFilters.length - 1; i++) {
+			UnorderedGrowablePipe.connect(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
+		}
+		UnorderedGrowablePipe.connect(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
+		UnorderedGrowablePipe.connect(stopTimestampFilter.getOutputPort(), collectorSink.getInputPort());
+
+		final Runnable runnable = new Runnable() {
+			@Override
+			public void run() {
+				pipeline.onStart();
+				do {
+					pipeline.executeWithPorts();
+				} while (pipeline.isReschedulable());
+				// System.out.println("buildPipeline finished");
+			}
+		};
+
+		return runnable;
+	}
+
+	@Override
+	public void start() {
+		super.start();
+
+		for (Thread workerThread : this.workerThreads) {
+			workerThread.start();
+		}
+
+		try {
+			for (Thread workerThread : this.workerThreads) {
+				workerThread.join();
+			}
+		} catch (InterruptedException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	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<List<TimestampObject>> getTimestampObjectsList() {
+		return this.timestampObjectsList;
+	}
+
+}

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

+package teetime.examples.throughput.methodcall;
+
+import teetime.util.concurrent.workstealing.CircularArray;
+
+public class OrderedGrowableArrayPipe<T> implements IPipe<T> {
+
+	private CircularArray<T> elements;
+	private int head;
+	private int tail;
+
+	public OrderedGrowableArrayPipe() {
+		this(17);
+	}
+
+	public OrderedGrowableArrayPipe(final int initialCapacity) {
+		this.elements = new CircularArray<T>(initialCapacity);
+	}
+
+	public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
+		IPipe<T> pipe = new OrderedGrowableArrayPipe<T>();
+		sourcePort.pipe = pipe;
+		targetPort.pipe = pipe;
+	}
+
+	@Override
+	public void add(final T element) {
+		this.elements.put(this.tail++, element);
+	}
+
+	@Override
+	public T removeLast() {
+		if (this.head < this.tail) {
+			return this.elements.get(this.head++);
+		} else {
+			return null;
+		}
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return this.size() == 0;
+	}
+
+	@Override
+	public T readLast() {
+		return this.elements.get(this.head);
+	}
+
+	@Override
+	public int size() {
+		return this.tail - this.head;
+	}
+
+}

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

+package teetime.examples.throughput.methodcall;
+
+import java.util.LinkedList;
+
+public class OrderedGrowablePipe<T> implements IPipe<T> {
+
+	private LinkedList<T> elements;
+
+	public OrderedGrowablePipe() {
+		this(100000);
+	}
+
+	public OrderedGrowablePipe(final int initialCapacity) {
+		this.elements = new LinkedList<T>();
+	}
+
+	public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
+		IPipe<T> pipe = new OrderedGrowablePipe<T>();
+		sourcePort.pipe = pipe;
+		targetPort.pipe = pipe;
+	}
+
+	@Override
+	public void add(final T element) {
+		this.elements.add(element);
+	}
+
+	@Override
+	public T removeLast() {
+		return this.elements.removeFirst();
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return this.elements.isEmpty();
+	}
+
+	@Override
+	public T readLast() {
+		return this.elements.getFirst();
+	}
+
+	@Override
+	public int size() {
+		return this.elements.size();
+	}
+
+}

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

@@ -4,7 +4,7 @@ import teetime.util.concurrent.spsc.FFBufferOrdered3;
 
 public class SpScPipe<T> implements IPipe<T> {
 
-	private final FFBufferOrdered3<T> queue = new FFBufferOrdered3<T>(4);
+	private final FFBufferOrdered3<T> queue = new FFBufferOrdered3<T>(100010);
 
 	public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
 		IPipe<T> pipe = new SpScPipe<T>();

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

@@ -30,4 +30,6 @@ public interface Stage<I, O> {
 
 	void onIsPipelineHead();
 
+	void onStart();
+
 }

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

 package teetime.examples.throughput.methodcall;
 
-public class FixedSizedPipe<T> implements IPipe<T> {
+public class UnorderedGrowablePipe<T> implements IPipe<T> {
 
 	// private static final class ArrayWrapper2<T> {
 	//
@@ -34,13 +34,13 @@ public class FixedSizedPipe<T> implements IPipe<T> {
 	private int lastFreeIndex;
 
 	@SuppressWarnings("unchecked")
-	public FixedSizedPipe() {
+	public UnorderedGrowablePipe() {
 		this.MIN_CAPACITY = 4;
 		this.elements = (T[]) new Object[this.MIN_CAPACITY];
 	}
 
 	public static <T> void connect(final OutputPort<T> sourcePort, final InputPort<T> targetPort) {
-		IPipe<T> pipe = new FixedSizedPipe<T>();
+		IPipe<T> pipe = new UnorderedGrowablePipe<T>();
 		sourcePort.pipe = pipe;
 		targetPort.pipe = pipe;
 	}
@@ -80,6 +80,7 @@ public class FixedSizedPipe<T> implements IPipe<T> {
 
 	private T[] grow() {
 		int newSize = this.elements.length * 2;
+		// System.out.println("growing to " + newSize);
 		return this.newArray(newSize);
 	}
 

src/test/java/teetime/examples/throughput/methodcall/stage/AbstractStage.java

@@ -126,6 +126,11 @@ public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
 		}
 	}
 
+	@Override
+	public void onStart() {
+		// empty default implementation
+	}
+
 	@Override
 	public Stage getParentStage() {
 		return this.parentStage;

src/test/java/teetime/examples/throughput/methodcall/stage/CollectorSink.java

@@ -45,6 +45,11 @@ public class CollectorSink<T> extends ConsumerStage<T, Object> {
 		return continueSignal;
 	}
 
+	@Override
+	public void onIsPipelineHead() {
+		System.out.println("size: " + this.elements.size());
+	}
+
 	// @Override
 	// public void execute3() {
 	// T element = this.getInputPort().receive();

src/test/java/teetime/examples/throughput/methodcall/stage/Delay.java

@@ -26,7 +26,8 @@ public class Delay<I> extends AbstractStage<I, I> {
 			this.send(element);
 		}
 
-		this.setReschedulable(this.getInputPort().pipe.size() > 0);
+		// this.setReschedulable(this.getInputPort().pipe.size() > 0);
+		this.setReschedulable(false);
 		// System.out.println("delay: " + this.getInputPort().pipe.size());
 	}
 

src/test/java/teetime/examples/throughput/methodcall/stage/Distributor.java

+package teetime.examples.throughput.methodcall.stage;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import teetime.examples.throughput.methodcall.ConsumerStage;
+import teetime.examples.throughput.methodcall.OutputPort;
+import teetime.util.concurrent.spsc.Pow2;
+import teetime.util.list.CommittableQueue;
+
+public class Distributor<T> extends ConsumerStage<T, T> {
+
+	private final List<OutputPort<T>> outputPortList = new ArrayList<OutputPort<T>>();
+
+	private OutputPort<T>[] outputPorts;
+	private int nextOutputPortIndex;
+
+	private int size;
+
+	private int mask;
+
+	@Override
+	public T execute(final Object element) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	protected void execute4(final CommittableQueue<T> elements) {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	protected void execute5(final T element) {
+		OutputPort<T> outputPort = this.outputPorts[this.nextOutputPortIndex % this.size];
+		this.nextOutputPortIndex++;
+		outputPort.send(element);
+	}
+
+	@Override
+	public void onIsPipelineHead() {
+		for (OutputPort op : this.outputPorts) {
+			op.send(END_SIGNAL);
+			System.out.println("End signal sent, size: " + op.pipe.size() + ", end signal:" + (op.pipe.readLast() == END_SIGNAL));
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void onStart() {
+		this.size = this.outputPortList.size();
+		// this.mask = this.size - 1;
+
+		int sizeInPow2 = Pow2.findNextPositivePowerOfTwo(this.size); // is not necessary so far
+		this.outputPorts = this.outputPortList.toArray(new OutputPort[sizeInPow2]);
+		System.out.println("outputPorts: " + this.outputPorts);
+	}
+
+	public OutputPort<T> getNewOutputPort() {
+		OutputPort<T> outputPort = new OutputPort<T>();
+		this.outputPortList.add(outputPort);
+		return outputPort;
+	}
+
+}

src/test/java/teetime/examples/throughput/methodcall/stage/EndStage.java

@@ -97,4 +97,10 @@ public class EndStage<T> implements StageWithPort<T, T> {
 
 	}
 
+	@Override
+	public void onStart() {
+		// TODO Auto-generated method stub
+
+	}
+
 }

src/test/java/teetime/examples/throughput/methodcall/stage/Pipeline.java

@@ -99,6 +99,7 @@ public class Pipeline<I, O> implements StageWithPort<I, O>, OnDisableListener {
 		return this.stages[0].executeRecursively(element);
 	}
 
+	@Override
 	public void onStart() {
 		// Pipe pipe = new Pipe();
 		// this.outputPort.pipe = pipe;
@@ -141,6 +142,10 @@ public class Pipeline<I, O> implements StageWithPort<I, O>, OnDisableListener {
 			stage.setSuccessor(this.stages[i + 1]);
 		}
 		this.stages[this.stages.length - 1].setSuccessor(new EndStage<Object>());
+
+		for (StageWithPort<?, ?> stage : this.stages) {
+			stage.onStart();
+		}
 	}
 
 	//

src/test/java/teetime/examples/throughput/methodcall/stage/Relay.java

+package teetime.examples.throughput.methodcall.stage;
+
+import teetime.util.list.CommittableQueue;
+
+public class Relay<T> extends AbstractStage<T, T> {
+
+	public Relay() {
+		this.setReschedulable(true);
+	}
+
+	@Override
+	public void executeWithPorts() {
+		T element = this.getInputPort().receive();
+		if (null == element) {
+			return;
+		} else if (END_SIGNAL == element) {
+			this.setReschedulable(false);
+			System.out.println("got end signal; pipe.size: " + this.getInputPort().pipe.size());
+			return;
+		}
+
+		this.send(element);
+	}
+
+	@Override
+	public T execute(final Object element) {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public void onIsPipelineHead() {
+		// do nothing
+	}
+
+	@Override
+	protected void execute4(final CommittableQueue<T> elements) {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	protected void execute5(final T element) {
+		// TODO Auto-generated method stub
+
+	}
+
+}