diff --git a/src/main/java/teetime/variant/methodcall/framework/core/AbstractStage.java b/src/main/java/teetime/variant/methodcall/framework/core/AbstractStage.java
index 947f68b505ff2f8e6cc3e9ea76fa5a6005d1432b..3fa3175e408bebe82ed7bfb024b1350c2f241d9d 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/AbstractStage.java
+++ b/src/main/java/teetime/variant/methodcall/framework/core/AbstractStage.java
@@ -1,20 +1,17 @@
package teetime.variant.methodcall.framework.core;
import teetime.util.list.CommittableQueue;
+import teetime.util.list.CommittableResizableArrayQueue;
-public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
+public abstract class AbstractStage<I, O> implements Stage<I, O> {
- private final InputPort<I> inputPort = new InputPort<I>();
- private final OutputPort<O> outputPort = new OutputPort<O>();
-
- // protected final CommittableQueue<O> outputElements = new CommittableResizableArrayQueue<O>(null, 4);
- // private final CommittableQueue<O> outputElements = null;
+ protected final CommittableQueue<O> outputElements = new CommittableResizableArrayQueue<O>(null, 4);
private Stage<?, ?> parentStage;
private int index;
- private StageWithPort<?, ?> successor;
+ private Stage<?, ?> successor;
private boolean reschedulable;
@@ -24,7 +21,7 @@ public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
if (result == null) {
return null;
}
- StageWithPort<?, ?> next = this.next();
+ Stage<?, ?> next = this.next();
// if (next != null) {
// return next.executeRecursively(result);
// } else {
@@ -33,16 +30,6 @@ public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
return next.executeRecursively(result);
}
- @Override
- public InputPort<I> getInputPort() {
- return this.inputPort;
- }
-
- @Override
- public OutputPort<O> getOutputPort() {
- return this.outputPort;
- }
-
@Override
public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
// pass through the end signal
@@ -63,61 +50,17 @@ public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
this.execute4(elements);
- // this.outputElements.commit();
+ this.outputElements.commit();
- // return this.outputElements;
- return null;
+ return this.outputElements;
}
- // @Override
- // public void executeWithPorts() {
- // CommittableQueue execute;
- // do {
- // // execute = this.next().execute2(this.outputElements);
- // // execute = this.next().execute2(this.getOutputPort().pipe.getElements());
- // this.next().executeWithPorts();
- // } while (this.next().isReschedulable());
- // }
-
protected abstract void execute4(CommittableQueue<I> elements);
- protected abstract void execute5(I element);
-
- protected final void send(final O element) {
- // this.outputElements.addToTailUncommitted(element);
- // this.outputElements.commit();
-
- this.getOutputPort().send(element);
-
- // CommittableQueue execute;
-
- StageWithPort<?, ?> next = this.next();
- do {
- // execute = this.next().execute2(this.outputElements);
- // execute = this.next().execute2(this.getOutputPort().pipe.getElements());
- next.executeWithPorts();
- // System.out.println("Executed " + this.next().getClass().getSimpleName());
- } while (next.isReschedulable());
- // } while (this.next().getInputPort().pipe.size() > 0);
- // } while (execute.size() > 0);
+ protected void send(final O element) {
+ this.outputElements.addToTailUncommitted(element);
}
- // @Override
- // public SchedulingInformation getSchedulingInformation() {
- // return this.schedulingInformation;
- // }
-
- // public void disable() {
- // this.schedulingInformation.setActive(false);
- // this.fireOnDisable();
- // }
-
- // private void fireOnDisable() {
- // if (this.listener != null) {
- // this.listener.onDisable(this, this.index);
- // }
- // }
-
@Override
public void onStart() {
// empty default implementation
@@ -135,20 +78,15 @@ public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
}
@Override
- public StageWithPort<?, ?> next() {
+ public Stage<?, ?> next() {
return this.successor;
}
@Override
- public void setSuccessor(final StageWithPort<?, ?> successor) {
+ public void setSuccessor(final Stage<? super O, ?> successor) {
this.successor = successor;
}
- @Override
- public void setSuccessor(final Stage<?, ?> successor) {
- throw new IllegalStateException();
- }
-
@Override
public boolean isReschedulable() {
return this.reschedulable;
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/ConsumerStage.java b/src/main/java/teetime/variant/methodcall/framework/core/ConsumerStage.java
index ea72ab4732ba77d5233ea40dd8ec3dec5c45049c..2ea5b659ea079ac7fd240bcd0952358596f44e20 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/ConsumerStage.java
+++ b/src/main/java/teetime/variant/methodcall/framework/core/ConsumerStage.java
@@ -18,21 +18,6 @@ public abstract class ConsumerStage<I, O> extends AbstractStage<I, O> {
return output;
}
- @Override
- public void executeWithPorts() {
- I element = this.getInputPort().receive();
-
- this.setReschedulable(this.getInputPort().getPipe().size() > 0);
-
- this.execute5(element);
-
- // this.send(result);
-
- // if (!this.getOutputPort().pipe.isEmpty()) {
- // super.executeWithPorts();
- // }
- }
-
@Override
public void onIsPipelineHead() {
// do nothing
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/Pipeline.java b/src/main/java/teetime/variant/methodcall/framework/core/Pipeline.java
index 265673788ff96710cf02e8a23cfbdbafd732ba4f..0309da6eea3ef6de3722c1171739c6a1bfa49236 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/Pipeline.java
+++ b/src/main/java/teetime/variant/methodcall/framework/core/Pipeline.java
@@ -7,33 +7,31 @@ import java.util.List;
import teetime.util.list.CommittableQueue;
import teetime.variant.methodcall.stage.EndStage;
-public class Pipeline<I, O> implements StageWithPort<I, O> {
+public class Pipeline<I, O> implements Stage<I, O> {
- private StageWithPort<I, ?> firstStage;
- private final List<StageWithPort<?, ?>> intermediateStages = new LinkedList<StageWithPort<?, ?>>();
- private StageWithPort<?, O> lastStage;
+ private Stage<I, ?> firstStage;
+ private final List<Stage<?, ?>> intermediateStages = new LinkedList<Stage<?, ?>>();
+ private Stage<?, O> lastStage;
- private StageWithPort<?, ?>[] stages;
+ private Stage<?, ?>[] stages;
private Stage<?, ?> parentStage;
private int index;
- private int startIndex;
private boolean reschedulable;
- private int firstStageIndex;
- public void setFirstStage(final StageWithPort<I, ?> stage) {
+ public void setFirstStage(final Stage<I, ?> stage) {
this.firstStage = stage;
}
- public void addIntermediateStages(final StageWithPort<?, ?>... stages) {
+ public void addIntermediateStages(final Stage<?, ?>... stages) {
this.intermediateStages.addAll(Arrays.asList(stages));
}
- public void addIntermediateStage(final StageWithPort stage) {
+ public void addIntermediateStage(final Stage<?, ?> stage) {
this.intermediateStages.add(stage);
}
- public void setLastStage(final StageWithPort<?, O> stage) {
+ public void setLastStage(final Stage<?, O> stage) {
this.lastStage = stage;
}
@@ -53,34 +51,10 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
// queue = stage.execute2(queue);
// }
- this.firstStage.execute2(elements);
+ queue = this.firstStage.execute2(elements);
this.setReschedulable(this.firstStage.isReschedulable());
- return queue;
- }
-
- @Override
- public void executeWithPorts() {
- StageWithPort<?, ?> firstStage = this.stages[this.firstStageIndex];
- firstStage.executeWithPorts();
- this.updateRescheduable(firstStage);
- // this.setReschedulable(stage.isReschedulable());
- }
-
- private final void updateRescheduable(final Stage<?, ?> stage) {
- Stage<?, ?> currentStage = stage;
- while (!currentStage.isReschedulable()) {
- this.firstStageIndex++;
- currentStage = currentStage.next();
- if (currentStage == null) { // loop reaches the last stage
- this.setReschedulable(false);
- this.cleanUp();
- return;
- }
- currentStage.onIsPipelineHead();
- // System.out.println("firstStageIndex: " + this.firstStageIndex + ", class:" + stage.getClass().getSimpleName());
- }
- this.setReschedulable(true);
+ return queue;
}
@Override
@@ -90,7 +64,7 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
@Override
public Object executeRecursively(final Object element) {
- return this.stages[0].executeRecursively(element);
+ return this.firstStage.executeRecursively(element);
}
@Override
@@ -117,44 +91,38 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
// this.lastStage.getInputPort().pipe = pipe;
int size = 1 + this.intermediateStages.size() + 1;
- this.stages = new StageWithPort[size];
+ this.stages = new Stage[size];
this.stages[0] = this.firstStage;
for (int i = 0; i < this.intermediateStages.size(); i++) {
- StageWithPort<?, ?> stage = this.intermediateStages.get(i);
+ Stage<?, ?> stage = this.intermediateStages.get(i);
this.stages[1 + i] = stage;
}
this.stages[this.stages.length - 1] = this.lastStage;
for (int i = 0; i < this.stages.length; i++) {
- StageWithPort<?, ?> stage = this.stages[i];
+ // Stage<?, ?> stage = this.stages[i];
// stage.setParentStage(this, i);
// stage.setListener(this);
}
for (int i = 0; i < this.stages.length - 1; i++) {
- StageWithPort<?, ?> stage = this.stages[i];
+ Stage stage = this.stages[i];
stage.setSuccessor(this.stages[i + 1]);
}
this.stages[this.stages.length - 1].setSuccessor(new EndStage<Object>());
- for (StageWithPort<?, ?> stage : this.stages) {
+ for (Stage<?, ?> stage : this.stages) {
stage.onStart();
}
}
- //
- // @Override
- // public InputPort getInputPort() {
- // return this.firstStage.getInputPort();
- // }
-
@Override
- public Stage getParentStage() {
+ public Stage<?, ?> getParentStage() {
return this.parentStage;
}
@Override
- public void setParentStage(final Stage parentStage, final int index) {
+ public void setParentStage(final Stage<?, ?> parentStage, final int index) {
this.index = index;
this.parentStage = parentStage;
}
@@ -170,12 +138,7 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
}
@Override
- public void setSuccessor(final Stage<?, ?> successor) {
- throw new IllegalStateException();
- }
-
- @Override
- public void setSuccessor(final StageWithPort<?, ?> successor) {
+ public void setSuccessor(final Stage<? super O, ?> successor) {
throw new IllegalStateException();
}
@@ -188,30 +151,4 @@ public class Pipeline<I, O> implements StageWithPort<I, O> {
this.reschedulable = reschedulable;
}
- @Override
- public InputPort<I> getInputPort() {
- return this.firstStage.getInputPort();
- }
-
- @Override
- public OutputPort<O> getOutputPort() {
- return this.lastStage.getOutputPort();
- }
-
- // TODO remove since it does not increase performances
- private void cleanUp() {
- for (int i = 0; i < this.stages.length; i++) {
- StageWithPort<?, ?> stage = this.stages[i];
- stage.setParentStage(null, i);
- // stage.setListener(null);
- stage.setSuccessor(null);
- }
-
- this.firstStage = null;
- this.intermediateStages.clear();
- this.lastStage = null;
-
- System.out.println("cleaned up");
- }
-
}
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/ProducerStage.java b/src/main/java/teetime/variant/methodcall/framework/core/ProducerStage.java
index b032b0e3af47e61c462be9b2785a9ef36fe3bc0f..bd7707bee1b50089cc5b03f2838bfc86d4cf97b4 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/ProducerStage.java
+++ b/src/main/java/teetime/variant/methodcall/framework/core/ProducerStage.java
@@ -14,21 +14,12 @@ public abstract class ProducerStage<I, O> extends AbstractStage<I, O> {
boolean outputIsEmpty = outputElements.isEmpty();
if (outputIsEmpty) {
- this.getOutputPort().getPipe().close();
+ // this.getOutputPort().getPipe().close();
}
return outputElements;
}
- @Override
- public void executeWithPorts() {
- this.execute5(null);
-
- // if (!this.getOutputPort().pipe.isEmpty()) {
- // super.executeWithPorts();
- // }
- }
-
@Override
public void onIsPipelineHead() {
// do nothing
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/Stage.java b/src/main/java/teetime/variant/methodcall/framework/core/Stage.java
index 95ac0b028d207eb14f58f5b3188cfa94fcf170d9..c891435007609996456a2c6aa35500b6cfa2f99a 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/Stage.java
+++ b/src/main/java/teetime/variant/methodcall/framework/core/Stage.java
@@ -22,7 +22,7 @@ public interface Stage<I, O> {
Stage<?, ?> next();
- void setSuccessor(Stage<?, ?> successor);
+ void setSuccessor(Stage<? super O, ?> successor);
boolean isReschedulable();
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/StageWithPort.java b/src/main/java/teetime/variant/methodcall/framework/core/StageWithPort.java
deleted file mode 100644
index 9246317e402f465b6fc6c7277392ee0293c5f41a..0000000000000000000000000000000000000000
--- a/src/main/java/teetime/variant/methodcall/framework/core/StageWithPort.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package teetime.variant.methodcall.framework.core;
-
-public interface StageWithPort<I, O> extends Stage<I, O> {
-
- InputPort<I> getInputPort();
-
- OutputPort<O> getOutputPort();
-
- void executeWithPorts();
-
- // void executeWithPorts(Object element);
-
- void setSuccessor(StageWithPort<?, ?> successor);
-}
diff --git a/src/main/java/teetime/variant/methodcall/stage/CollectorSink.java b/src/main/java/teetime/variant/methodcall/stage/CollectorSink.java
index 4bc890e0b372dd50f888e67af7dcb930e0ab8b4f..49ea2928527b536a95a6037d5782d8a8182a19c3 100644
--- a/src/main/java/teetime/variant/methodcall/stage/CollectorSink.java
+++ b/src/main/java/teetime/variant/methodcall/stage/CollectorSink.java
@@ -50,24 +50,9 @@ public class CollectorSink<T> extends ConsumerStage<T, Object> {
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());
- // }
- // }
-
@Override
protected void execute4(final CommittableQueue<T> elements) {
T element = elements.removeFromHead();
- this.execute5(element);
- }
-
- @Override
- protected void execute5(final T element) {
this.elements.add(element);
if ((this.elements.size() % THRESHOLD) == 0) {
System.out.println("size: " + this.elements.size());
diff --git a/src/main/java/teetime/variant/methodcall/stage/EndStage.java b/src/main/java/teetime/variant/methodcall/stage/EndStage.java
index f3c12625a089efd2a89a799163318a00fd175396..dc791a4465e29b11176e53962aab354482c68347 100644
--- a/src/main/java/teetime/variant/methodcall/stage/EndStage.java
+++ b/src/main/java/teetime/variant/methodcall/stage/EndStage.java
@@ -5,17 +5,9 @@ import java.util.List;
import teetime.util.ConstructorClosure;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
import teetime.variant.methodcall.framework.core.Stage;
-import teetime.variant.methodcall.framework.core.StageWithPort;
-public class EndStage<T> implements StageWithPort<T, T> {
-
- @Override
- public Object executeRecursively(final Object element) {
- return this.execute(element);
- }
+public class EndStage<T> implements Stage<T, T> {
public int count;
public ConstructorClosure<?> closure;
@@ -23,7 +15,7 @@ public class EndStage<T> implements StageWithPort<T, T> {
@Override
public T execute(final Object element) {
- return (T) element;
+ return null;
}
@Override
@@ -44,7 +36,7 @@ public class EndStage<T> implements StageWithPort<T, T> {
}
@Override
- public void setParentStage(final Stage parentStage, final int index) {
+ public void setParentStage(final Stage<?, ?> parentStage, final int index) {
// TODO Auto-generated method stub
}
@@ -56,48 +48,26 @@ public class EndStage<T> implements StageWithPort<T, T> {
}
@Override
- public void setSuccessor(final Stage<?, ?> successor) {
+ public void setSuccessor(final Stage<? super T, ?> successor) {
// TODO Auto-generated method stub
}
@Override
public boolean isReschedulable() {
- // TODO Auto-generated method stub
return false;
}
@Override
- public InputPort<T> getInputPort() {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public OutputPort<T> getOutputPort() {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public void executeWithPorts() {
- // this.getInputPort().receive(); // just consume
- // do nothing
- // this.count++;
- // Object r = this.closure.execute(null);
- // this.list.add(r);
- }
-
- @Override
- public void setSuccessor(final StageWithPort<?, ?> successor) {
+ public void onStart() {
// TODO Auto-generated method stub
}
@Override
- public void onStart() {
+ public Object executeRecursively(final Object element) {
// TODO Auto-generated method stub
-
+ return null;
}
}
diff --git a/src/main/java/teetime/variant/methodcall/stage/NoopFilter.java b/src/main/java/teetime/variant/methodcall/stage/NoopFilter.java
index ace51b02dd2b5984ba025352ee18de4954ce8c39..cdd4423708fcc71949351b161f0285b8d2bae983 100644
--- a/src/main/java/teetime/variant/methodcall/stage/NoopFilter.java
+++ b/src/main/java/teetime/variant/methodcall/stage/NoopFilter.java
@@ -31,21 +31,11 @@ public class NoopFilter<T> extends ConsumerStage<T, T> {
return (T) obj;
}
- // @Override
- // public void execute3() {
- // T element = this.getInputPort().receive();
- // // this.getOutputPort().send(element);
- // }
-
@Override
protected void execute4(final CommittableQueue<T> elements) {
T element = elements.removeFromHead();
- this.execute5(element);
- }
-
- @Override
- protected void execute5(final T element) {
- this.send(element); // "send" calls the next stage and so on
+ // this.send(element); // "send" calls the next stage and so on
+ throw new IllegalStateException();
}
}
diff --git a/src/main/java/teetime/variant/methodcall/stage/ObjectProducer.java b/src/main/java/teetime/variant/methodcall/stage/ObjectProducer.java
index efd8cbdc7b874f50e2bf4a78c5b5dbd8dcad708d..0a50a69d9a45057d2e57b89e08ade0dcf03d3d62 100644
--- a/src/main/java/teetime/variant/methodcall/stage/ObjectProducer.java
+++ b/src/main/java/teetime/variant/methodcall/stage/ObjectProducer.java
@@ -40,7 +40,7 @@ public class ObjectProducer<T> extends ProducerStage<Void, T> {
@Override
public T execute(final Object element) {
// if (this.numInputObjects == 0) {
- // this.setReschedulable(false);
+ // // this.setReschedulable(false);
// return null;
// }
@@ -75,30 +75,8 @@ public class ObjectProducer<T> extends ProducerStage<Void, T> {
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);
- // }
- // }
-
@Override
protected void execute4(final CommittableQueue<Void> elements) {
- this.execute5(null);
- }
-
- @Override
- protected void execute5(final Void element) {
T newObject = null;
newObject = this.inputObjectCreator.create();
this.numInputObjects--;
@@ -110,12 +88,7 @@ public class ObjectProducer<T> extends ProducerStage<Void, T> {
// System.out.println(this.getClass().getSimpleName() + ": sending " + this.numInputObjects);
this.send(newObject);
+ // throw new IllegalStateException();
}
- // @Override
- // public void onIsPipelineHead() {
- // // this.getOutputPort().pipe = null; // no performance increase
- // super.onIsPipelineHead();
- // }
-
}
diff --git a/src/main/java/teetime/variant/methodcall/stage/StartTimestampFilter.java b/src/main/java/teetime/variant/methodcall/stage/StartTimestampFilter.java
index 3619f24fbec264c488945fdd4cc5b302e3f7a74e..754de3dcdfee1e41815293b55b7c4e18b64827d1 100644
--- a/src/main/java/teetime/variant/methodcall/stage/StartTimestampFilter.java
+++ b/src/main/java/teetime/variant/methodcall/stage/StartTimestampFilter.java
@@ -33,22 +33,11 @@ public class StartTimestampFilter extends ConsumerStage<TimestampObject, Timesta
return timestampObject;
}
- // @Override
- // public void execute3() {
- // TimestampObject element = this.getInputPort().receive();
- // element.setStartTimestamp(System.nanoTime());
- // // this.getOutputPort().send(element);
- // }
-
@Override
protected void execute4(final CommittableQueue<TimestampObject> elements) {
TimestampObject element = elements.removeFromHead();
- this.execute5(element);
- }
-
- @Override
- protected void execute5(final TimestampObject element) {
element.setStartTimestamp(System.nanoTime());
- this.send(element);
+ // this.send(element);
+ throw new IllegalStateException();
}
}
diff --git a/src/main/java/teetime/variant/methodcall/stage/StopTimestampFilter.java b/src/main/java/teetime/variant/methodcall/stage/StopTimestampFilter.java
index 07ee037ac6d28e9d3567577cf8d40ffdc2dcecfd..b855088a68abd6ce1b9a5f2bb6648aa6f7c9fe65 100644
--- a/src/main/java/teetime/variant/methodcall/stage/StopTimestampFilter.java
+++ b/src/main/java/teetime/variant/methodcall/stage/StopTimestampFilter.java
@@ -43,12 +43,9 @@ public class StopTimestampFilter extends ConsumerStage<TimestampObject, Timestam
@Override
protected void execute4(final CommittableQueue<TimestampObject> elements) {
TimestampObject element = elements.removeFromHead();
- this.execute5(element);
- }
-
- @Override
- protected void execute5(final TimestampObject element) {
element.setStopTimestamp(System.nanoTime());
- this.send(element);
+ // this.send(element);
+ throw new IllegalStateException();
}
+
}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java
new file mode 100644
index 0000000000000000000000000000000000000000..ab6fc8943c7cec02751f08ee5ce66752cff30ea2
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java
@@ -0,0 +1,141 @@
+package teetime.variant.methodcallWithPorts.framework.core;
+
+import teetime.util.list.CommittableQueue;
+
+public abstract class AbstractStage<I, O> implements StageWithPort<I, O> {
+
+ private final InputPort<I> inputPort = new InputPort<I>();
+ private final OutputPort<O> outputPort = new OutputPort<O>();
+
+ // protected final CommittableQueue<O> outputElements = new CommittableResizableArrayQueue<O>(null, 4);
+ // private final CommittableQueue<O> outputElements = null;
+
+ private StageWithPort<?, ?> parentStage;
+
+ private int index;
+
+ private StageWithPort<?, ?> successor;
+
+ private boolean reschedulable;
+
+ @Override
+ public InputPort<I> getInputPort() {
+ return this.inputPort;
+ }
+
+ @Override
+ public OutputPort<O> getOutputPort() {
+ return this.outputPort;
+ }
+
+ @Override
+ public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
+ // pass through the end signal
+ // InputPort<I> port = this.getInputPort();
+ // if (elements != null) {
+ // // I element = port.read();
+ // // I element = elements.getTail();
+ // // if (element == END_SIGNAL) {
+ // // this.send((O) END_SIGNAL);
+ // // } else {
+ // // // elements = this.getInputPort().pipe.getElements();
+ // // }
+ //
+ // this.execute4(elements);
+ // } else {
+ // throw new IllegalStateException();
+ // }
+
+ this.execute4(elements);
+
+ // this.outputElements.commit();
+
+ // return this.outputElements;
+ return null;
+ }
+
+ // @Override
+ // public void executeWithPorts() {
+ // CommittableQueue execute;
+ // do {
+ // // execute = this.next().execute2(this.outputElements);
+ // // execute = this.next().execute2(this.getOutputPort().pipe.getElements());
+ // this.next().executeWithPorts();
+ // } while (this.next().isReschedulable());
+ // }
+
+ protected abstract void execute4(CommittableQueue<I> elements);
+
+ protected abstract void execute5(I element);
+
+ protected final void send(final O element) {
+ // this.outputElements.addToTailUncommitted(element);
+ // this.outputElements.commit();
+
+ this.getOutputPort().send(element);
+
+ // CommittableQueue execute;
+
+ StageWithPort<?, ?> next = this.next();
+ do {
+ // execute = this.next().execute2(this.outputElements);
+ // execute = this.next().execute2(this.getOutputPort().pipe.getElements());
+ next.executeWithPorts();
+ // System.out.println("Executed " + this.next().getClass().getSimpleName());
+ } while (next.isReschedulable());
+ // } while (this.next().getInputPort().pipe.size() > 0);
+ // } while (execute.size() > 0);
+ }
+
+ // @Override
+ // public SchedulingInformation getSchedulingInformation() {
+ // return this.schedulingInformation;
+ // }
+
+ // public void disable() {
+ // this.schedulingInformation.setActive(false);
+ // this.fireOnDisable();
+ // }
+
+ // private void fireOnDisable() {
+ // if (this.listener != null) {
+ // this.listener.onDisable(this, this.index);
+ // }
+ // }
+
+ @Override
+ public void onStart() {
+ // empty default implementation
+ }
+
+ @Override
+ public StageWithPort<?, ?> getParentStage() {
+ return this.parentStage;
+ }
+
+ @Override
+ public void setParentStage(final StageWithPort<?, ?> parentStage, final int index) {
+ this.index = index;
+ this.parentStage = parentStage;
+ }
+
+ @Override
+ public StageWithPort<?, ?> next() {
+ return this.successor;
+ }
+
+ @Override
+ public void setSuccessor(final StageWithPort<? super O, ?> successor) {
+ this.successor = successor;
+ }
+
+ @Override
+ public boolean isReschedulable() {
+ return this.reschedulable;
+ }
+
+ public void setReschedulable(final boolean reschedulable) {
+ this.reschedulable = reschedulable;
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/ConsumerStage.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/ConsumerStage.java
new file mode 100644
index 0000000000000000000000000000000000000000..1f4f6ab84f376e75ee699bae9d0241be09a5e022
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/ConsumerStage.java
@@ -0,0 +1,41 @@
+package teetime.variant.methodcallWithPorts.framework.core;
+
+import teetime.util.list.CommittableQueue;
+
+public abstract class ConsumerStage<I, O> extends AbstractStage<I, O> {
+
+ @Override
+ public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
+ // 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 (not reproducible)
+ return output;
+ }
+
+ @Override
+ public void executeWithPorts() {
+ I element = this.getInputPort().receive();
+
+ this.setReschedulable(this.getInputPort().getPipe().size() > 0);
+
+ this.execute5(element);
+
+ // this.send(result);
+
+ // if (!this.getOutputPort().pipe.isEmpty()) {
+ // super.executeWithPorts();
+ // }
+ }
+
+ @Override
+ public void onIsPipelineHead() {
+ // do nothing
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/InputPort.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/InputPort.java
similarity index 72%
rename from src/main/java/teetime/variant/methodcall/framework/core/InputPort.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/InputPort.java
index f74b2bfca09035a5f5d73a98c7359224b2287c50..95f674f2fe4302cd8dd97208b3599154883e2e91 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/InputPort.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/InputPort.java
@@ -1,6 +1,6 @@
-package teetime.variant.methodcall.framework.core;
+package teetime.variant.methodcallWithPorts.framework.core;
-import teetime.variant.methodcall.framework.core.pipe.IPipe;
+import teetime.variant.methodcallWithPorts.framework.core.pipe.IPipe;
public class InputPort<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/OutputPort.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java
similarity index 65%
rename from src/main/java/teetime/variant/methodcall/framework/core/OutputPort.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java
index b7612df5882e5469ba973792646a94c5d15e7089..b426506e35b586d6722b3e38b032b8eddf43ad2a 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/OutputPort.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java
@@ -1,6 +1,6 @@
-package teetime.variant.methodcall.framework.core;
+package teetime.variant.methodcallWithPorts.framework.core;
-import teetime.variant.methodcall.framework.core.pipe.IPipe;
+import teetime.variant.methodcallWithPorts.framework.core.pipe.IPipe;
public class OutputPort<T> {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java
new file mode 100644
index 0000000000000000000000000000000000000000..4ba07ebb8d77a99ac676d6c93d2708561e9e6e9c
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java
@@ -0,0 +1,196 @@
+package teetime.variant.methodcallWithPorts.framework.core;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+import teetime.util.list.CommittableQueue;
+import teetime.variant.methodcallWithPorts.stage.EndStage;
+
+public class Pipeline<I, O> implements StageWithPort<I, O> {
+
+ private StageWithPort<I, ?> firstStage;
+ private final List<StageWithPort<?, ?>> intermediateStages = new LinkedList<StageWithPort<?, ?>>();
+ private StageWithPort<?, O> lastStage;
+
+ private StageWithPort<?, ?>[] stages;
+ private StageWithPort<?, ?> parentStage;
+ private int index;
+ private int startIndex;
+
+ private boolean reschedulable;
+ private int firstStageIndex;
+
+ public void setFirstStage(final StageWithPort<I, ?> stage) {
+ this.firstStage = stage;
+ }
+
+ public void addIntermediateStages(final StageWithPort<?, ?>... stages) {
+ this.intermediateStages.addAll(Arrays.asList(stages));
+ }
+
+ public void addIntermediateStage(final StageWithPort stage) {
+ this.intermediateStages.add(stage);
+ }
+
+ public void setLastStage(final StageWithPort<?, O> stage) {
+ this.lastStage = stage;
+ }
+
+ @Override
+ public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
+ // CommittableQueue queue = this.firstStage.execute2(elements);
+ // for (Stage<?, ?> stage : this.intermediateStages) {
+ // queue = stage.execute2(queue);
+ // }
+ // return this.lastStage.execute2(queue);
+
+ // below is faster than above (probably because of the instantiation of a list iterator in each (!) execution)
+ CommittableQueue queue = elements;
+
+ // for (int i = this.startIndex; i < this.stages.length; i++) {
+ // Stage<?, ?> stage = this.stages[i];
+ // queue = stage.execute2(queue);
+ // }
+
+ this.firstStage.execute2(elements);
+ this.setReschedulable(this.firstStage.isReschedulable());
+ return queue;
+ }
+
+ @Override
+ public void executeWithPorts() {
+ StageWithPort<?, ?> firstStage = this.stages[this.firstStageIndex];
+ firstStage.executeWithPorts();
+
+ this.updateRescheduable(firstStage);
+ // this.setReschedulable(stage.isReschedulable());
+ }
+
+ private final void updateRescheduable(final StageWithPort<?, ?> stage) {
+ StageWithPort<?, ?> currentStage = stage;
+ while (!currentStage.isReschedulable()) {
+ this.firstStageIndex++;
+ currentStage = currentStage.next();
+ if (currentStage == null) { // loop reaches the last stage
+ this.setReschedulable(false);
+ this.cleanUp();
+ return;
+ }
+ currentStage.onIsPipelineHead();
+ // System.out.println("firstStageIndex: " + this.firstStageIndex + ", class:" + stage.getClass().getSimpleName());
+ }
+ this.setReschedulable(true);
+ }
+
+ @Override
+ public void onIsPipelineHead() {
+ // do nothing
+ }
+
+ @Override
+ public 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;
+
+ int size = 1 + this.intermediateStages.size() + 1;
+ this.stages = new StageWithPort[size];
+ this.stages[0] = this.firstStage;
+ for (int i = 0; i < this.intermediateStages.size(); i++) {
+ StageWithPort<?, ?> stage = this.intermediateStages.get(i);
+ this.stages[1 + i] = stage;
+ }
+ this.stages[this.stages.length - 1] = this.lastStage;
+
+ for (int i = 0; i < this.stages.length; i++) {
+ // StageWithPort<?, ?> stage = this.stages[i];
+ // stage.setParentStage(this, i);
+ // stage.setListener(this);
+ }
+
+ for (int i = 0; i < this.stages.length - 1; i++) {
+ StageWithPort stage = this.stages[i];
+ stage.setSuccessor(this.stages[i + 1]);
+ }
+ this.stages[this.stages.length - 1].setSuccessor(new EndStage<Object>());
+
+ for (StageWithPort<?, ?> stage : this.stages) {
+ stage.onStart();
+ }
+ }
+
+ @Override
+ public StageWithPort<?, ?> getParentStage() {
+ return this.parentStage;
+ }
+
+ @Override
+ public void setParentStage(final StageWithPort<?, ?> parentStage, final int index) {
+ this.index = index;
+ this.parentStage = parentStage;
+ }
+
+ @Override
+ public StageWithPort<?, ?> next() {
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public void setSuccessor(final StageWithPort<? super O, ?> successor) {
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public boolean isReschedulable() {
+ return this.reschedulable;
+ }
+
+ public void setReschedulable(final boolean reschedulable) {
+ this.reschedulable = reschedulable;
+ }
+
+ @Override
+ public InputPort<I> getInputPort() {
+ return this.firstStage.getInputPort();
+ }
+
+ @Override
+ public OutputPort<O> getOutputPort() {
+ return this.lastStage.getOutputPort();
+ }
+
+ // TODO remove since it does not increase performances
+ private void cleanUp() {
+ for (int i = 0; i < this.stages.length; i++) {
+ StageWithPort<?, ?> stage = this.stages[i];
+ stage.setParentStage(null, i);
+ // stage.setListener(null);
+ stage.setSuccessor(null);
+ }
+
+ this.firstStage = null;
+ this.intermediateStages.clear();
+ this.lastStage = null;
+
+ System.out.println("cleaned up");
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/ProducerStage.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/ProducerStage.java
new file mode 100644
index 0000000000000000000000000000000000000000..16dabc4cac9f9afcbbe24b0d1fdbe7b9c1b23339
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/ProducerStage.java
@@ -0,0 +1,37 @@
+package teetime.variant.methodcallWithPorts.framework.core;
+
+import teetime.util.list.CommittableQueue;
+
+public abstract class ProducerStage<I, O> extends AbstractStage<I, O> {
+
+ public ProducerStage() {
+ this.setReschedulable(true);
+ }
+
+ @Override
+ public CommittableQueue<O> execute2(final CommittableQueue<I> elements) {
+ CommittableQueue<O> outputElements = super.execute2(elements);
+
+ boolean outputIsEmpty = outputElements.isEmpty();
+ if (outputIsEmpty) {
+ this.getOutputPort().getPipe().close();
+ }
+
+ return outputElements;
+ }
+
+ @Override
+ public void executeWithPorts() {
+ this.execute5(null);
+
+ // if (!this.getOutputPort().pipe.isEmpty()) {
+ // super.executeWithPorts();
+ // }
+ }
+
+ @Override
+ public void onIsPipelineHead() {
+ // do nothing
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/RunnableStage.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java
similarity index 84%
rename from src/main/java/teetime/variant/methodcall/framework/core/RunnableStage.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java
index 61e9c6b4806ac4efe1f71045f49cd79d53e67ef4..c7435b030199b7a3ccfc03106c294780f1fa2412 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/RunnableStage.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java
@@ -1,4 +1,4 @@
-package teetime.variant.methodcall.framework.core;
+package teetime.variant.methodcallWithPorts.framework.core;
public class RunnableStage implements Runnable {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java
new file mode 100644
index 0000000000000000000000000000000000000000..d8ee283901b1464bf44f0f7b647b5acea74dabbd
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java
@@ -0,0 +1,38 @@
+package teetime.variant.methodcallWithPorts.framework.core;
+
+import teetime.util.list.CommittableQueue;
+
+public interface StageWithPort<I, O> {
+
+ InputPort<I> getInputPort();
+
+ OutputPort<O> getOutputPort();
+
+ void executeWithPorts();
+
+ // void executeWithPorts(Object element);
+
+ // O execute(Object element);
+
+ // CommittableQueue<O> execute2();
+
+ CommittableQueue<O> execute2(CommittableQueue<I> elements);
+
+ // SchedulingInformation getSchedulingInformation();
+
+ StageWithPort<?, ?> getParentStage();
+
+ void setParentStage(StageWithPort<?, ?> parentStage, int index);
+
+ // void setListener(OnDisableListener listener);
+
+ StageWithPort<?, ?> next();
+
+ void setSuccessor(StageWithPort<? super O, ?> successor);
+
+ boolean isReschedulable();
+
+ void onIsPipelineHead();
+
+ void onStart();
+}
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/AbstractPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/AbstractPipe.java
similarity index 85%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/AbstractPipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/AbstractPipe.java
index 390443afecde9f4cadbadd18228226c6923e7252..dc56e3c2aa02372b9ed51724cf709185b5fdb5e5 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/AbstractPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/AbstractPipe.java
@@ -1,4 +1,4 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
import java.util.concurrent.atomic.AtomicBoolean;
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/IPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java
similarity index 80%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/IPipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java
index 3551eb1fd0deb4e776eb4aa136063a2b8ac0a3e2..0cb86e454a016ad2d2456a7cbf0a05fdc416885b 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/IPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java
@@ -1,4 +1,4 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
public interface IPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/OrderedGrowableArrayPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/OrderedGrowableArrayPipe.java
similarity index 83%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/OrderedGrowableArrayPipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/OrderedGrowableArrayPipe.java
index 7989ba04563bc9523b97ffa037290e5e74055a45..26c8a884e64b7f641f543b25c6b1af8cdaa07c59 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/OrderedGrowableArrayPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/OrderedGrowableArrayPipe.java
@@ -1,8 +1,8 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
import teetime.util.concurrent.workstealing.CircularArray;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public class OrderedGrowableArrayPipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/OrderedGrowablePipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/OrderedGrowablePipe.java
similarity index 80%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/OrderedGrowablePipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/OrderedGrowablePipe.java
index 124d1c7ccf7bde6e5132fa072b6aed90f45a55ac..4c36c839d551d521447797d0b679afc4c551c813 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/OrderedGrowablePipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/OrderedGrowablePipe.java
@@ -1,9 +1,9 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
import java.util.LinkedList;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public class OrderedGrowablePipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/Pipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/Pipe.java
similarity index 86%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/Pipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/Pipe.java
index e710e6fa4944ca0a0ed1590a3170dc49bc7b0168..79321208b4cdfea93782d44f280276595904dba9 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/Pipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/Pipe.java
@@ -1,8 +1,8 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
import teetime.util.list.CommittableResizableArrayQueue;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public class Pipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/SingleElementPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SingleElementPipe.java
similarity index 81%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/SingleElementPipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SingleElementPipe.java
index c7f8053fec84f765e6ae6dd1e5154b409beff40a..f4dd5c6b0877b0bcadafdc506f8b00d36180199d 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/SingleElementPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SingleElementPipe.java
@@ -1,7 +1,7 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
//public class SingleElementPipe<T> implements IPipe<T> {
public class SingleElementPipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/SpScPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java
similarity index 78%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/SpScPipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java
index 2856c165a78e38b890a90f0bbcf0bdfda0b201d8..54869f0f5f87ddef489d44244976ef4c4ef88cda 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/SpScPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java
@@ -1,8 +1,8 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
import teetime.util.concurrent.spsc.FFBufferOrdered3;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public class SpScPipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/framework/core/pipe/UnorderedGrowablePipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/UnorderedGrowablePipe.java
similarity index 92%
rename from src/main/java/teetime/variant/methodcall/framework/core/pipe/UnorderedGrowablePipe.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/UnorderedGrowablePipe.java
index fe5359ab62555f188db8889040f6074c6f777b5b..48c2afa17062e810a323c01172fa7d41d344e8a1 100644
--- a/src/main/java/teetime/variant/methodcall/framework/core/pipe/UnorderedGrowablePipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/UnorderedGrowablePipe.java
@@ -1,7 +1,7 @@
-package teetime.variant.methodcall.framework.core.pipe;
+package teetime.variant.methodcallWithPorts.framework.core.pipe;
-import teetime.variant.methodcall.framework.core.InputPort;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public class UnorderedGrowablePipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcall/stage/Clock.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Clock.java
similarity index 85%
rename from src/main/java/teetime/variant/methodcall/stage/Clock.java
rename to src/main/java/teetime/variant/methodcallWithPorts/stage/Clock.java
index 7479e20555b120f445498b93ea135c35d5cda1e2..70ba4d7a8b41b920e6e70529e282672ba0ff3dc0 100644
--- a/src/main/java/teetime/variant/methodcall/stage/Clock.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Clock.java
@@ -1,7 +1,7 @@
-package teetime.variant.methodcall.stage;
+package teetime.variant.methodcallWithPorts.stage;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.ProducerStage;
+import teetime.variant.methodcallWithPorts.framework.core.ProducerStage;
public class Clock extends ProducerStage<Void, Long> {
@@ -10,12 +10,6 @@ public class Clock extends ProducerStage<Void, Long> {
private long initialDelayInMs;
private long intervalDelayInMs;
- @Override
- public Long execute(final Object element) {
- // TODO Auto-generated method stub
- return null;
- }
-
@Override
protected void execute4(final CommittableQueue<Void> elements) {
// TODO Auto-generated method stub
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/CollectorSink.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/CollectorSink.java
new file mode 100644
index 0000000000000000000000000000000000000000..c13fa8949b51faf21a26a1a022226911ca333317
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/CollectorSink.java
@@ -0,0 +1,61 @@
+/***************************************************************************
+ * 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.variant.methodcallWithPorts.stage;
+
+import java.util.List;
+
+import teetime.util.list.CommittableQueue;
+import teetime.variant.methodcallWithPorts.framework.core.ConsumerStage;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class CollectorSink<T> extends ConsumerStage<T, Object> {
+
+ private static final int THRESHOLD = 10000;
+
+ private final List<T> elements;
+
+ public CollectorSink(final List<T> list) {
+ this.elements = list;
+ }
+
+ @Override
+ public void onIsPipelineHead() {
+ System.out.println("size: " + this.elements.size());
+ }
+
+ @Override
+ protected void execute4(final CommittableQueue<T> elements) {
+ T element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final T element) {
+ this.elements.add(element);
+ if ((this.elements.size() % THRESHOLD) == 0) {
+ System.out.println("size: " + this.elements.size());
+ }
+
+ if (this.elements.size() > 90000) {
+ // System.out.println("size > 90000: " + this.elements.size());
+ }
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcall/stage/Delay.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Delay.java
similarity index 82%
rename from src/main/java/teetime/variant/methodcall/stage/Delay.java
rename to src/main/java/teetime/variant/methodcallWithPorts/stage/Delay.java
index f5103209fe969f15334000c5589c7e5d01ef9631..d27bc6b68ac4a3240df2849b464af9c4c470e774 100644
--- a/src/main/java/teetime/variant/methodcall/stage/Delay.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Delay.java
@@ -1,8 +1,8 @@
-package teetime.variant.methodcall.stage;
+package teetime.variant.methodcallWithPorts.stage;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.AbstractStage;
-import teetime.variant.methodcall.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
public class Delay<I> extends AbstractStage<I, I> {
@@ -37,12 +37,6 @@ public class Delay<I> extends AbstractStage<I, I> {
this.setReschedulable(true);
}
- @Override
- public I execute(final Object element) {
- // TODO Auto-generated method stub
- return null;
- }
-
@Override
protected void execute4(final CommittableQueue<I> elements) {
// TODO Auto-generated method stub
diff --git a/src/main/java/teetime/variant/methodcall/stage/Distributor.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Distributor.java
similarity index 87%
rename from src/main/java/teetime/variant/methodcall/stage/Distributor.java
rename to src/main/java/teetime/variant/methodcallWithPorts/stage/Distributor.java
index ea56a54b3763816a2fff4fcc5974467f4090ca75..5d8be6cd66d59704a1b25fe2a40ef7ea6f158909 100644
--- a/src/main/java/teetime/variant/methodcall/stage/Distributor.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Distributor.java
@@ -1,12 +1,12 @@
-package teetime.variant.methodcall.stage;
+package teetime.variant.methodcallWithPorts.stage;
import java.util.ArrayList;
import java.util.List;
import teetime.util.concurrent.spsc.Pow2;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.AbstractStage;
-import teetime.variant.methodcall.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public final class Distributor<T> extends AbstractStage<T, T> {
@@ -21,12 +21,6 @@ public final class Distributor<T> extends AbstractStage<T, T> {
// 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
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/EndStage.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/EndStage.java
new file mode 100644
index 0000000000000000000000000000000000000000..24d2907ce066b595b7db1894ccb6df0f0277370b
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/EndStage.java
@@ -0,0 +1,86 @@
+package teetime.variant.methodcallWithPorts.stage;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import teetime.util.ConstructorClosure;
+import teetime.util.list.CommittableQueue;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
+import teetime.variant.methodcallWithPorts.framework.core.StageWithPort;
+
+public class EndStage<T> implements StageWithPort<T, T> {
+
+ public int count;
+ public ConstructorClosure<?> closure;
+ public List<Object> list = new LinkedList<Object>();
+
+ @Override
+ public void onIsPipelineHead() {
+ // do nothing
+ }
+
+ @Override
+ public CommittableQueue<T> execute2(final CommittableQueue<T> elements) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public StageWithPort getParentStage() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void setParentStage(final StageWithPort parentStage, final int index) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public StageWithPort next() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void setSuccessor(final StageWithPort<? super T, ?> successor) {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public boolean isReschedulable() {
+ // TODO Auto-generated method stub
+ return false;
+ }
+
+ @Override
+ public InputPort<T> getInputPort() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public OutputPort<T> getOutputPort() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void executeWithPorts() {
+ // this.getInputPort().receive(); // just consume
+ // do nothing
+ // this.count++;
+ // Object r = this.closure.execute(null);
+ // this.list.add(r);
+ }
+
+ @Override
+ public void onStart() {
+ // TODO Auto-generated method stub
+
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcall/stage/Merger.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Merger.java
similarity index 86%
rename from src/main/java/teetime/variant/methodcall/stage/Merger.java
rename to src/main/java/teetime/variant/methodcallWithPorts/stage/Merger.java
index 0c5a9f3130d084345fface46793964437cac09c3..aae8c293fbe25861ae54c3bded374666664afd3f 100644
--- a/src/main/java/teetime/variant/methodcall/stage/Merger.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Merger.java
@@ -1,12 +1,12 @@
-package teetime.variant.methodcall.stage;
+package teetime.variant.methodcallWithPorts.stage;
import java.util.ArrayList;
import java.util.List;
import teetime.util.concurrent.spsc.Pow2;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.AbstractStage;
-import teetime.variant.methodcall.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
public class Merger<T> extends AbstractStage<T, T> {
@@ -17,12 +17,6 @@ public class Merger<T> extends AbstractStage<T, T> {
private int size;
private InputPort<T>[] inputPorts;
- @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
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/NoopFilter.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/NoopFilter.java
new file mode 100644
index 0000000000000000000000000000000000000000..69110cb6d6fcdc73a7eff2a9189be703dc22dddb
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/NoopFilter.java
@@ -0,0 +1,45 @@
+/***************************************************************************
+ * 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.variant.methodcallWithPorts.stage;
+
+import teetime.util.list.CommittableQueue;
+import teetime.variant.methodcallWithPorts.framework.core.ConsumerStage;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class NoopFilter<T> extends ConsumerStage<T, T> {
+
+ // @Override
+ // public void execute3() {
+ // T element = this.getInputPort().receive();
+ // // this.getOutputPort().send(element);
+ // }
+
+ @Override
+ protected void execute4(final CommittableQueue<T> elements) {
+ T element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final T element) {
+ this.send(element); // "send" calls the next stage and so on
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/ObjectProducer.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/ObjectProducer.java
new file mode 100644
index 0000000000000000000000000000000000000000..61d001cd2bf916c7803525d9df59ca38f7a99e2e
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/ObjectProducer.java
@@ -0,0 +1,99 @@
+/***************************************************************************
+ * 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.variant.methodcallWithPorts.stage;
+
+import teetime.util.ConstructorClosure;
+import teetime.util.list.CommittableQueue;
+import teetime.variant.methodcallWithPorts.framework.core.ProducerStage;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class ObjectProducer<T> extends ProducerStage<Void, T> {
+
+ private long numInputObjects;
+ private ConstructorClosure<T> inputObjectCreator;
+
+ /**
+ * @since 1.10
+ */
+ public ObjectProducer(final long numInputObjects, final ConstructorClosure<T> inputObjectCreator) {
+ this.numInputObjects = numInputObjects;
+ this.inputObjectCreator = inputObjectCreator;
+ }
+
+ public long getNumInputObjects() {
+ return this.numInputObjects;
+ }
+
+ public void setNumInputObjects(final long numInputObjects) {
+ this.numInputObjects = numInputObjects;
+ }
+
+ public ConstructorClosure<T> getInputObjectCreator() {
+ return this.inputObjectCreator;
+ }
+
+ public void setInputObjectCreator(final ConstructorClosure<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);
+ // }
+ // }
+
+ @Override
+ protected void execute4(final CommittableQueue<Void> elements) {
+ this.execute5(null);
+ }
+
+ @Override
+ protected void execute5(final Void element) {
+ T newObject = null;
+ newObject = this.inputObjectCreator.create();
+ this.numInputObjects--;
+
+ if (this.numInputObjects == 0) {
+ this.setReschedulable(false);
+ // this.getOutputPort().pipe.close();
+ }
+
+ // System.out.println(this.getClass().getSimpleName() + ": sending " + this.numInputObjects);
+ this.send(newObject);
+ }
+
+ // @Override
+ // public void onIsPipelineHead() {
+ // // this.getOutputPort().pipe = null; // no performance increase
+ // super.onIsPipelineHead();
+ // }
+
+}
diff --git a/src/main/java/teetime/variant/methodcall/stage/Relay.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java
similarity index 80%
rename from src/main/java/teetime/variant/methodcall/stage/Relay.java
rename to src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java
index f1a352092e1446607929d07396aaa0617c13dabe..c4e7a00628842cfaaf92488a0400e260281bf34a 100644
--- a/src/main/java/teetime/variant/methodcall/stage/Relay.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java
@@ -1,7 +1,7 @@
-package teetime.variant.methodcall.stage;
+package teetime.variant.methodcallWithPorts.stage;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.AbstractStage;
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
public class Relay<T> extends AbstractStage<T, T> {
@@ -22,12 +22,6 @@ public class Relay<T> extends AbstractStage<T, T> {
this.send(element);
}
- @Override
- public T execute(final Object element) {
- // TODO Auto-generated method stub
- return null;
- }
-
@Override
public void onIsPipelineHead() {
System.out.println("onIsPipelineHead");
diff --git a/src/main/java/teetime/variant/methodcall/stage/Sink.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Sink.java
similarity index 57%
rename from src/main/java/teetime/variant/methodcall/stage/Sink.java
rename to src/main/java/teetime/variant/methodcallWithPorts/stage/Sink.java
index 89ff7c754a9296944815db4634662efa671d8dc2..b4485b35fd5cda94f4aa71af8f4932897a818f2b 100644
--- a/src/main/java/teetime/variant/methodcall/stage/Sink.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Sink.java
@@ -1,16 +1,10 @@
-package teetime.variant.methodcall.stage;
+package teetime.variant.methodcallWithPorts.stage;
import teetime.util.list.CommittableQueue;
-import teetime.variant.methodcall.framework.core.ConsumerStage;
+import teetime.variant.methodcallWithPorts.framework.core.ConsumerStage;
public class Sink<T> extends ConsumerStage<T, T> {
- @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
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/StartTimestampFilter.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/StartTimestampFilter.java
new file mode 100644
index 0000000000000000000000000000000000000000..a35bf75f8bcb287526871274858eff9f23d32b0a
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/StartTimestampFilter.java
@@ -0,0 +1,47 @@
+/***************************************************************************
+ * 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.variant.methodcallWithPorts.stage;
+
+import teetime.util.list.CommittableQueue;
+import teetime.variant.explicitScheduling.examples.throughput.TimestampObject;
+import teetime.variant.methodcallWithPorts.framework.core.ConsumerStage;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class StartTimestampFilter extends ConsumerStage<TimestampObject, TimestampObject> {
+
+ // @Override
+ // public void execute3() {
+ // TimestampObject element = this.getInputPort().receive();
+ // element.setStartTimestamp(System.nanoTime());
+ // // this.getOutputPort().send(element);
+ // }
+
+ @Override
+ protected void execute4(final CommittableQueue<TimestampObject> elements) {
+ TimestampObject element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final TimestampObject element) {
+ element.setStartTimestamp(System.nanoTime());
+ this.send(element);
+ }
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/StopTimestampFilter.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/StopTimestampFilter.java
new file mode 100644
index 0000000000000000000000000000000000000000..a1601cb4c06e4aba99b07adbc0fde06ff6acfc78
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/StopTimestampFilter.java
@@ -0,0 +1,47 @@
+/***************************************************************************
+ * 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.variant.methodcallWithPorts.stage;
+
+import teetime.util.list.CommittableQueue;
+import teetime.variant.explicitScheduling.examples.throughput.TimestampObject;
+import teetime.variant.methodcallWithPorts.framework.core.ConsumerStage;
+
+/**
+ * @author Christian Wulf
+ *
+ * @since 1.10
+ */
+public class StopTimestampFilter extends ConsumerStage<TimestampObject, TimestampObject> {
+
+ // @Override
+ // public void execute3() {
+ // TimestampObject element = this.getInputPort().receive();
+ // element.setStopTimestamp(System.nanoTime());
+ // // this.getOutputPort().send(element);
+ // }
+
+ @Override
+ protected void execute4(final CommittableQueue<TimestampObject> elements) {
+ TimestampObject element = elements.removeFromHead();
+ this.execute5(element);
+ }
+
+ @Override
+ protected void execute5(final TimestampObject element) {
+ element.setStopTimestamp(System.nanoTime());
+ this.send(element);
+ }
+}
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment01/MethodCallThroughputAnalysis1.java b/src/test/java/teetime/variant/methodcall/examples/experiment01/MethodCallThroughputAnalysis1.java
index e53851df9e36ef816cdc16fa36c2011e0abfa851..f5a71bf9a6fbfd0ef7319a194ce755fc6677f88b 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment01/MethodCallThroughputAnalysis1.java
+++ b/src/test/java/teetime/variant/methodcall/examples/experiment01/MethodCallThroughputAnalysis1.java
@@ -65,10 +65,13 @@ public class MethodCallThroughputAnalysis1 extends Analysis {
@Override
public void run() {
while (true) {
- TimestampObject object = objectProducer.execute(null);
- if (object == null) {
+ if (!objectProducer.isReschedulable()) {
return;
}
+ TimestampObject object = objectProducer.execute(null);
+ // if (object == null)
+ // return;
+
object = startTimestampFilter.execute(object);
for (final NoopFilter<TimestampObject> noopFilter : noopFilters) {
object = noopFilter.execute(object);
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment02/MethodCallThroughputAnalysis2.java b/src/test/java/teetime/variant/methodcall/examples/experiment02/MethodCallThroughputAnalysis2.java
index 1155da2e6a40584bea439d5605eb6006063cf5ef..85eaebcad1f8457e3f6689f8e847b3b82cc5ec4c 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment02/MethodCallThroughputAnalysis2.java
+++ b/src/test/java/teetime/variant/methodcall/examples/experiment02/MethodCallThroughputAnalysis2.java
@@ -73,11 +73,6 @@ public class MethodCallThroughputAnalysis2 extends Analysis {
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() {
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment09/MethodCallThoughputTimestampAnalysis9Test.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment09/MethodCallThoughputTimestampAnalysis9Test.java
similarity index 97%
rename from src/test/java/teetime/variant/methodcall/examples/experiment09/MethodCallThoughputTimestampAnalysis9Test.java
rename to src/test/java/teetime/variant/methodcallWithPorts/examples/experiment09/MethodCallThoughputTimestampAnalysis9Test.java
index a09a084c9e75660a4c8cc7bdc012ce7ec705948d..a0aa656b5c18f0fe7afe7eb973928c6c693f2d49 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment09/MethodCallThoughputTimestampAnalysis9Test.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment09/MethodCallThoughputTimestampAnalysis9Test.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package teetime.variant.methodcall.examples.experiment09;
+package teetime.variant.methodcallWithPorts.examples.experiment09;
import java.util.ArrayList;
import java.util.List;
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment09/MethodCallThroughputAnalysis9.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment09/MethodCallThroughputAnalysis9.java
similarity index 98%
rename from src/test/java/teetime/variant/methodcall/examples/experiment09/MethodCallThroughputAnalysis9.java
rename to src/test/java/teetime/variant/methodcallWithPorts/examples/experiment09/MethodCallThroughputAnalysis9.java
index 2ace0eb42dd039a7b457a19907dac703c0a581f0..058b0b30e4197ee970e89ad4de165491ef808600 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment09/MethodCallThroughputAnalysis9.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment09/MethodCallThroughputAnalysis9.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package teetime.variant.methodcall.examples.experiment09;
+package teetime.variant.methodcallWithPorts.examples.experiment09;
import java.util.List;
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment10/MethodCallThoughputTimestampAnalysis10Test.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment10/MethodCallThoughputTimestampAnalysis10Test.java
similarity index 97%
rename from src/test/java/teetime/variant/methodcall/examples/experiment10/MethodCallThoughputTimestampAnalysis10Test.java
rename to src/test/java/teetime/variant/methodcallWithPorts/examples/experiment10/MethodCallThoughputTimestampAnalysis10Test.java
index 0c1e3e553c9e4e0b9fc4e35b77e5131fea96d4ba..16d44a2613a4b3a6661028442b5637c17dba4e13 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment10/MethodCallThoughputTimestampAnalysis10Test.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment10/MethodCallThoughputTimestampAnalysis10Test.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package teetime.variant.methodcall.examples.experiment10;
+package teetime.variant.methodcallWithPorts.examples.experiment10;
import java.util.ArrayList;
import java.util.List;
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment10/MethodCallThroughputAnalysis10.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment10/MethodCallThroughputAnalysis10.java
similarity index 98%
rename from src/test/java/teetime/variant/methodcall/examples/experiment10/MethodCallThroughputAnalysis10.java
rename to src/test/java/teetime/variant/methodcallWithPorts/examples/experiment10/MethodCallThroughputAnalysis10.java
index 35282b24b41803b6f33ad07eef9d200606cc6521..b5c714989329ba2844424bb429766e1d487a35c6 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment10/MethodCallThroughputAnalysis10.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment10/MethodCallThroughputAnalysis10.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package teetime.variant.methodcall.examples.experiment10;
+package teetime.variant.methodcallWithPorts.examples.experiment10;
import java.util.List;
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment11/MethodCallThoughputTimestampAnalysis11Test.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment11/MethodCallThoughputTimestampAnalysis11Test.java
similarity index 97%
rename from src/test/java/teetime/variant/methodcall/examples/experiment11/MethodCallThoughputTimestampAnalysis11Test.java
rename to src/test/java/teetime/variant/methodcallWithPorts/examples/experiment11/MethodCallThoughputTimestampAnalysis11Test.java
index 1f9d70991285f3d0251c18e20b743c31bfe10df8..a6f76d0071ef07c9399caf0869920e0b384d8211 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment11/MethodCallThoughputTimestampAnalysis11Test.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment11/MethodCallThoughputTimestampAnalysis11Test.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package teetime.variant.methodcall.examples.experiment11;
+package teetime.variant.methodcallWithPorts.examples.experiment11;
import java.util.ArrayList;
import java.util.List;
diff --git a/src/test/java/teetime/variant/methodcall/examples/experiment11/MethodCallThroughputAnalysis11.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment11/MethodCallThroughputAnalysis11.java
similarity index 87%
rename from src/test/java/teetime/variant/methodcall/examples/experiment11/MethodCallThroughputAnalysis11.java
rename to src/test/java/teetime/variant/methodcallWithPorts/examples/experiment11/MethodCallThroughputAnalysis11.java
index c544e75e3c18af8d3945e11a88a1e663fe26c10a..6651496397d30c6ae7439e467d1eec8114b623fd 100644
--- a/src/test/java/teetime/variant/methodcall/examples/experiment11/MethodCallThroughputAnalysis11.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment11/MethodCallThroughputAnalysis11.java
@@ -13,21 +13,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
***************************************************************************/
-package teetime.variant.methodcall.examples.experiment11;
+package teetime.variant.methodcallWithPorts.examples.experiment11;
import java.util.List;
import teetime.util.ConstructorClosure;
import teetime.variant.explicitScheduling.examples.throughput.TimestampObject;
import teetime.variant.explicitScheduling.framework.core.Analysis;
-import teetime.variant.methodcall.framework.core.Pipeline;
-import teetime.variant.methodcall.framework.core.RunnableStage;
-import teetime.variant.methodcall.framework.core.pipe.UnorderedGrowablePipe;
-import teetime.variant.methodcall.stage.CollectorSink;
-import teetime.variant.methodcall.stage.NoopFilter;
-import teetime.variant.methodcall.stage.ObjectProducer;
-import teetime.variant.methodcall.stage.StartTimestampFilter;
-import teetime.variant.methodcall.stage.StopTimestampFilter;
+import teetime.variant.methodcallWithPorts.framework.core.Pipeline;
+import teetime.variant.methodcallWithPorts.framework.core.RunnableStage;
+import teetime.variant.methodcallWithPorts.framework.core.pipe.UnorderedGrowablePipe;
+import teetime.variant.methodcallWithPorts.stage.CollectorSink;
+import teetime.variant.methodcallWithPorts.stage.NoopFilter;
+import teetime.variant.methodcallWithPorts.stage.ObjectProducer;
+import teetime.variant.methodcallWithPorts.stage.StartTimestampFilter;
+import teetime.variant.methodcallWithPorts.stage.StopTimestampFilter;
/**
* @author Christian Wulf
diff --git a/src/test/java/what tests work.txt b/src/test/java/what tests work.txt
new file mode 100644
index 0000000000000000000000000000000000000000..5a29be3e567de16bff8f4393c45d854f112c073c
--- /dev/null
+++ b/src/test/java/what tests work.txt
@@ -0,0 +1,17 @@
+ok
+no
+inf
+inf
+inf
+no
+no
+no
+
+
+ok
+no
+
+
+
+
+