added performance test for method call with pipe

src/test/java/teetime/examples/throughput/MethodCallThoughputTimestampAnalysis2Test.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.MethodCallThroughputAnalysis2;
+import teetime.util.StatisticsUtil;
+import teetime.util.StopWatch;
+
+import kieker.common.logging.LogFactory;
+
+/**
+ * @author Christian Wulf
+ * 
+ * @since 1.10
+ */
+public class MethodCallThoughputTimestampAnalysis2Test {
+
+	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");
+	}
+
+	// 500 times faster than our new framework
+	// TODO check why
+
+	@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 MethodCallThroughputAnalysis2 analysis = new MethodCallThroughputAnalysis2();
+		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

+package teetime.examples.throughput.methodcall;
+
+public abstract class AbstractStage<I, O> implements Stage<I, O> {
+
+	Runnable inputPortIsUsed = new Runnable() {
+		@Override
+		public void run() {
+			// pass through the end signal
+			I element = AbstractStage.this.getInputPort().read();
+			if (element == END_SIGNAL) {
+				AbstractStage.this.getOutputPort().send((O) END_SIGNAL);
+				return;
+			}
+
+			AbstractStage.this.execute3();
+		}
+	};
+
+	Runnable inputPortIsNotUsed = new Runnable() {
+		@Override
+		public void run() {
+			// do not check
+
+			AbstractStage.this.execute3();
+		}
+	};
+
+	private final InputPort<I> inputPort = new InputPort<I>();
+	private final OutputPort<O> outputPort = new OutputPort<O>();
+	protected Runnable endSignalCheck = this.inputPortIsUsed;
+
+	@Override
+	public InputPort<I> getInputPort() {
+		return this.inputPort;
+	}
+
+	@Override
+	public OutputPort<O> getOutputPort() {
+		return this.outputPort;
+	}
+
+	@Override
+	public final void execute2() {
+		this.endSignalCheck.run();
+	}
+
+	protected abstract void execute3();
+}

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

@@ -22,7 +22,7 @@ import java.util.List;
  * 
  * @since 1.10
  */
-public class CollectorSink<T> {
+public class CollectorSink<T> extends AbstractStage<T, T> {
 
 	private static final int THRESHOLD = 10000;
 
@@ -41,4 +41,14 @@ public class CollectorSink<T> {
 			System.out.println("size: " + this.elements.size());
 		}
 	}
+
+	@Override
+	public void execute3() {
+		T element = this.getInputPort().receive();
+
+		this.elements.add(element);
+		if ((this.elements.size() % THRESHOLD) == 0) {
+			System.out.println("size: " + this.elements.size());
+		}
+	}
 }

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

+package teetime.examples.throughput.methodcall;
+
+public class InputPort<T> {
+
+	public Pipe<T> pipe;
+
+	public T receive() {
+		T element = this.pipe.removeLast();
+		return element;
+	}
+
+	public T read() {
+		T element = this.pipe.readLast();
+		return element;
+	}
+}

src/test/java/teetime/examples/throughput/methodcall/MethodCallThroughputAnalysis2.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.framework.core.Analysis;
+
+/**
+ * @author Christian Wulf
+ * 
+ * @since 1.10
+ */
+public class MethodCallThroughputAnalysis2 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 pipeline = new Pipeline();
+		pipeline.setFirstStage(objectProducer);
+		pipeline.addIntermediateStage(startTimestampFilter);
+		pipeline.addIntermediateStages(noopFilters);
+		pipeline.addIntermediateStage(stopTimestampFilter);
+		pipeline.setLastStage(collectorSink);
+
+		// 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();
+				while (!(pipeline.getOutputPort().pipe.readLast() == Stage.END_SIGNAL)) {
+					pipeline.execute2();
+				}
+			}
+		};
+
+		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/NoopFilter.java

@@ -20,9 +20,15 @@ package teetime.examples.throughput.methodcall;
  * 
  * @since 1.10
  */
-public class NoopFilter<T> {
+public class NoopFilter<T> extends AbstractStage<T, T> {
 
 	public T execute(final T obj) {
 		return obj;
 	}
+
+	@Override
+	public void execute3() {
+		T element = this.getInputPort().receive();
+		this.getOutputPort().send(element);
+	}
 }

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

@@ -22,7 +22,7 @@ import java.util.concurrent.Callable;
  * 
  * @since 1.10
  */
-public class ObjectProducer<T> {
+public class ObjectProducer<T> extends AbstractStage<Void, T> {
 
 	private long numInputObjects;
 	private Callable<T> inputObjectCreator;
@@ -33,6 +33,8 @@ public class ObjectProducer<T> {
 	public ObjectProducer(final long numInputObjects, final Callable<T> inputObjectCreator) {
 		this.numInputObjects = numInputObjects;
 		this.inputObjectCreator = inputObjectCreator;
+
+		this.endSignalCheck = this.inputPortIsNotUsed;
 	}
 
 	public T execute() {
@@ -46,7 +48,7 @@ public class ObjectProducer<T> {
 
 			return newObject;
 		} catch (final Exception e) {
-			throw new IllegalStateException();
+			throw new IllegalStateException(e);
 		}
 	}
 
@@ -65,4 +67,22 @@ public class ObjectProducer<T> {
 	public void setInputObjectCreator(final Callable<T> inputObjectCreator) {
 		this.inputObjectCreator = inputObjectCreator;
 	}
+
+	@Override
+	protected void execute3() {
+		if (this.numInputObjects == 0) {
+			this.getOutputPort().send((T) END_SIGNAL);
+			return;
+		}
+
+		try {
+			final T newObject = this.inputObjectCreator.call();
+			this.numInputObjects--;
+
+			this.getOutputPort().send(newObject);
+		} catch (final Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
 }

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

+package teetime.examples.throughput.methodcall;
+
+public class OutputPort<T> {
+
+	public Pipe<T> pipe;
+
+	public void send(final T element) {
+		this.pipe.add(element);
+	}
+}

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

+package teetime.examples.throughput.methodcall;
+
+import teetime.util.list.CommittableResizableArrayQueue;
+
+public class Pipe<T> {
+
+	private final CommittableResizableArrayQueue<T> elements = new CommittableResizableArrayQueue<T>(null, 4);
+
+	public void add(final T element) {
+		this.elements.addToTailUncommitted(element);
+		this.elements.commit();
+	}
+
+	public T removeLast() {
+		T element = this.elements.removeFromHeadUncommitted();
+		this.elements.commit();
+		return element;
+	}
+
+	public boolean isEmpty() {
+		return this.elements.isEmpty();
+	}
+
+	public T readLast() {
+		return this.elements.getTail();
+	}
+}

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

+package teetime.examples.throughput.methodcall;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+public class Pipeline implements Stage {
+
+	private Stage firstStage;
+	private final List<Stage> intermediateStages = new LinkedList<Stage>();
+	private Stage lastStage;
+
+	void setFirstStage(final Stage stage) {
+		this.firstStage = stage;
+	}
+
+	void addIntermediateStages(final Stage... stages) {
+		this.intermediateStages.addAll(Arrays.asList(stages));
+	}
+
+	void addIntermediateStage(final Stage stage) {
+		this.intermediateStages.add(stage);
+	}
+
+	void setLastStage(final Stage stage) {
+		this.lastStage = stage;
+	}
+
+	@Override
+	public void execute2() {
+		this.firstStage.execute2();
+		for (Stage<?, ?> stage : this.intermediateStages) {
+			stage.execute2();
+		}
+		this.lastStage.execute2();
+	}
+
+	void onStart() {
+		// Pipe pipe = new Pipe();
+		// this.outputPort.pipe = pipe;
+		// this.firstStage.getInputPort().pipe = pipe;
+
+		Pipe pipe = new Pipe();
+		this.firstStage.getOutputPort().pipe = pipe;
+		this.intermediateStages.get(0).getInputPort().pipe = pipe;
+
+		for (int i = 0; i < this.intermediateStages.size() - 1; i++) {
+			Stage left = this.intermediateStages.get(i);
+			Stage right = this.intermediateStages.get(i + 1);
+
+			pipe = new Pipe();
+			left.getOutputPort().pipe = pipe;
+			right.getInputPort().pipe = pipe;
+		}
+
+		pipe = new Pipe();
+		this.intermediateStages.get(this.intermediateStages.size() - 1).getOutputPort().pipe = pipe;
+		this.lastStage.getInputPort().pipe = pipe;
+	}
+
+	@Override
+	public InputPort getInputPort() {
+		return this.firstStage.getInputPort();
+	}
+
+	@Override
+	public OutputPort getOutputPort() {
+		return this.lastStage.getOutputPort();
+	}
+
+}

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

+package teetime.examples.throughput.methodcall;
+
+public interface Stage<I, O> {
+
+	public static final Object END_SIGNAL = new Object();
+
+	void execute2();
+
+	InputPort<I> getInputPort();
+
+	OutputPort<O> getOutputPort();
+}

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

@@ -19,13 +19,20 @@ import teetime.examples.throughput.TimestampObject;
 
 /**
  * @author Christian Wulf
- *
+ * 
  * @since 1.10
  */
-public class StartTimestampFilter {
+public class StartTimestampFilter extends AbstractStage<TimestampObject, TimestampObject> {
 
 	public TimestampObject execute(final TimestampObject obj) {
 		obj.setStartTimestamp(System.nanoTime());
 		return obj;
 	}
+
+	@Override
+	public void execute3() {
+		TimestampObject element = this.getInputPort().receive();
+		element.setStartTimestamp(System.nanoTime());
+		this.getOutputPort().send(element);
+	}
 }

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

@@ -19,13 +19,20 @@ import teetime.examples.throughput.TimestampObject;
 
 /**
  * @author Christian Wulf
- *
+ * 
  * @since 1.10
  */
-public class StopTimestampFilter {
+public class StopTimestampFilter extends AbstractStage<TimestampObject, TimestampObject> {
 
 	public TimestampObject execute(final TimestampObject obj) {
 		obj.setStopTimestamp(System.nanoTime());
 		return obj;
 	}
+
+	@Override
+	public void execute3() {
+		TimestampObject element = this.getInputPort().receive();
+		element.setStopTimestamp(System.nanoTime());
+		this.getOutputPort().send(element);
+	}
 }