diff --git a/src/changes/changes.xml b/src/changes/changes.xml
index deabc8b156f59178ea8915388479515b5aa4ab0f..577f93a57a44804e0d6c0903e2e66e3553e8e560 100644
--- a/src/changes/changes.xml
+++ b/src/changes/changes.xml
@@ -6,7 +6,9 @@
</properties>
<body>
<release version="Snapshot" date="Daily basis" description="Unstable preview of oncoming versions">
-
+ <action dev="ntd" type="add" issue="154">
+ All stages will be initialized before starting the analysis.
+ </action>
</release>
diff --git a/src/main/java/teetime/framework/AbstractInterThreadPipe.java b/src/main/java/teetime/framework/AbstractInterThreadPipe.java
index 4deee9e7b64f5eb96fbf393e55cc879b2ca0c023..acb1752ef08afe1795d079ee994326d5c91ff555 100644
--- a/src/main/java/teetime/framework/AbstractInterThreadPipe.java
+++ b/src/main/java/teetime/framework/AbstractInterThreadPipe.java
@@ -24,6 +24,7 @@ import org.jctools.queues.spec.Ordering;
import org.jctools.queues.spec.Preference;
import teetime.framework.signal.ISignal;
+import teetime.framework.signal.InitializingSignal;
import teetime.util.concurrent.queue.PCBlockingQueue;
import teetime.util.concurrent.queue.putstrategy.PutStrategy;
import teetime.util.concurrent.queue.putstrategy.YieldPutStrategy;
@@ -69,6 +70,15 @@ public abstract class AbstractInterThreadPipe extends AbstractPipe {
cachedTargetStage.onSignal(signal, getTargetPort());
}
+ @Override
+ public final void waitForInitializingSignal() throws InterruptedException {
+ final ISignal signal = signalQueue.take();
+ if (!(signal instanceof InitializingSignal)) {
+ throw new IllegalStateException("Expected InitializingSignal, but was not the first arriving signal");
+ }
+ cachedTargetStage.onSignal(signal, getTargetPort());
+ }
+
@Override
public final boolean isClosed() {
return isClosed;
diff --git a/src/main/java/teetime/framework/AbstractIntraThreadPipe.java b/src/main/java/teetime/framework/AbstractIntraThreadPipe.java
index b3e47b9cf8e333a5f26cc55d70d7b7234b310e88..ccfce741b2945ac94d707b34d6a48a0eaf399f3b 100644
--- a/src/main/java/teetime/framework/AbstractIntraThreadPipe.java
+++ b/src/main/java/teetime/framework/AbstractIntraThreadPipe.java
@@ -56,4 +56,10 @@ public abstract class AbstractIntraThreadPipe extends AbstractPipe {
public void waitForStartSignal() throws InterruptedException {
// do nothing
}
+
+ @SuppressWarnings("PMD.EmptyMethodInAbstractClassShouldBeAbstract")
+ @Override
+ public void waitForInitializingSignal() throws InterruptedException {
+ // do nothing
+ }
}
diff --git a/src/main/java/teetime/framework/AbstractRunnableStage.java b/src/main/java/teetime/framework/AbstractRunnableStage.java
index a5d537c2c89629975cfbdb6d62801bda249e010b..535503166b3b564d2a1fa106f51e9daf992993a9 100644
--- a/src/main/java/teetime/framework/AbstractRunnableStage.java
+++ b/src/main/java/teetime/framework/AbstractRunnableStage.java
@@ -25,7 +25,7 @@ abstract class AbstractRunnableStage implements Runnable {
private static final String TERMINATING_THREAD_DUE_TO_THE_FOLLOWING_EXCEPTION = "Terminating thread due to the following exception: ";
- private final Stage stage;
+ protected final Stage stage;
@SuppressWarnings("PMD.LoggerIsNotStaticFinal")
protected final Logger logger;
@@ -39,16 +39,16 @@ abstract class AbstractRunnableStage implements Runnable {
this.logger.debug("Executing runnable stage...");
boolean failed = false;
try {
- beforeStageExecution(stage);
+ beforeStageExecution();
try {
do {
- executeStage(stage);
+ executeStage();
} while (!stage.shouldBeTerminated());
} catch (StageException e) {
this.stage.terminate();
failed = true;
}
- afterStageExecution(stage);
+ afterStageExecution();
} catch (RuntimeException e) {
this.logger.error(TERMINATING_THREAD_DUE_TO_THE_FOLLOWING_EXCEPTION, e);
@@ -72,10 +72,10 @@ abstract class AbstractRunnableStage implements Runnable {
}
- protected abstract void beforeStageExecution(Stage stage) throws InterruptedException;
+ protected abstract void beforeStageExecution() throws InterruptedException;
- protected abstract void executeStage(Stage stage);
+ protected abstract void executeStage();
- protected abstract void afterStageExecution(Stage stage);
+ protected abstract void afterStageExecution();
}
diff --git a/src/main/java/teetime/framework/Analysis.java b/src/main/java/teetime/framework/Analysis.java
index 78375efac944c52a8983dac250f9ea3dd524154c..c65916e440d9b045128d2ac9923a9571535048d1 100644
--- a/src/main/java/teetime/framework/Analysis.java
+++ b/src/main/java/teetime/framework/Analysis.java
@@ -59,7 +59,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
private final Collection<Pair<Thread, Throwable>> exceptions = new ConcurrentLinkedQueue<Pair<Thread, Throwable>>();
- private boolean initialized;
+ private final List<RunnableProducerStage> producerRunnables = new LinkedList<RunnableProducerStage>();
/**
* Creates a new {@link Analysis} that skips validating the port connections and uses the default listener.
@@ -116,10 +116,6 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
*
*/
private final void init() {
- if (initialized) {
- return;
- }
- initialized = true;
final List<Stage> threadableStageJobs = this.configuration.getThreadableStageJobs();
if (threadableStageJobs.isEmpty()) {
@@ -139,6 +135,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
}
case BY_SELF_DECISION: {
final RunnableProducerStage runnable = new RunnableProducerStage(stage);
+ producerRunnables.add(runnable);
thread = createThread(runnable, stage.getId());
this.finiteProducerThreads.add(thread);
InitializingSignal initializingSignal = new InitializingSignal();
@@ -147,6 +144,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
}
case BY_INTERRUPT: {
final RunnableProducerStage runnable = new RunnableProducerStage(stage);
+ producerRunnables.add(runnable);
thread = createThread(runnable, stage.getId());
InitializingSignal initializingSignal = new InitializingSignal();
stage.onSignal(initializingSignal, null);
@@ -161,6 +159,12 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
initializeIntraStages(intraStages, thread, newListener);
}
+ startThreads(this.consumerThreads);
+ startThreads(this.finiteProducerThreads);
+ startThreads(this.infiniteProducerThreads);
+
+ sendInitializingSignal();
+
}
private Thread createThread(final AbstractRunnableStage runnable, final String name) {
@@ -249,9 +253,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
* @since 1.1
*/
public void executeNonBlocking() {
- startThreads(this.consumerThreads);
- startThreads(this.finiteProducerThreads);
- startThreads(this.infiniteProducerThreads);
+ sendStartingSignal();
}
private void startThreads(final Iterable<Thread> threads) {
@@ -260,6 +262,18 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
}
}
+ private void sendInitializingSignal() {
+ for (RunnableProducerStage runnable : producerRunnables) {
+ runnable.triggerInitializingSignal();
+ }
+ }
+
+ private void sendStartingSignal() {
+ for (RunnableProducerStage runnable : producerRunnables) {
+ runnable.triggerStartingSignal();
+ }
+ }
+
/**
* Retrieves the Configuration which was used to add and arrange all stages needed for the Analysis
*
diff --git a/src/main/java/teetime/framework/InputPort.java b/src/main/java/teetime/framework/InputPort.java
index 2c2fee6f833fd6fc8476962a9ea3cf371845366b..f6e2a30d170607b2f237ec57a4145810ce6f2024 100644
--- a/src/main/java/teetime/framework/InputPort.java
+++ b/src/main/java/teetime/framework/InputPort.java
@@ -38,4 +38,8 @@ public final class InputPort<T> extends AbstractPort<T> {
pipe.waitForStartSignal();
}
+ public void waitForInitializingSignal() throws InterruptedException {
+ pipe.waitForInitializingSignal();
+ };
+
}
diff --git a/src/main/java/teetime/framework/RunnableConsumerStage.java b/src/main/java/teetime/framework/RunnableConsumerStage.java
index ded489ebed6581caff6da9b0cc4ae39c82fb20cb..c1a78e20917d21a7052dd47f66fd3ed18e3f023b 100644
--- a/src/main/java/teetime/framework/RunnableConsumerStage.java
+++ b/src/main/java/teetime/framework/RunnableConsumerStage.java
@@ -42,19 +42,19 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
@SuppressWarnings("PMD.GuardLogStatement")
@Override
- protected void beforeStageExecution(final Stage stage) throws InterruptedException {
- logger.trace("Waiting for start signals..." + inputPorts);
+ protected void beforeStageExecution() throws InterruptedException {
+ logger.trace("Waiting for start signals... " + stage);
for (InputPort<?> inputPort : inputPorts) {
- inputPort.waitForStartSignal();
+ inputPort.waitForInitializingSignal();
}
for (InputPort<?> inputPort : inputPorts) {
inputPort.waitForStartSignal();
}
- logger.trace("Starting..." + stage);
+ logger.trace("Starting... " + stage);
}
@Override
- protected void executeStage(final Stage stage) {
+ protected void executeStage() {
try {
stage.executeStage();
} catch (NotEnoughInputException e) {
@@ -73,7 +73,7 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
}
@Override
- protected void afterStageExecution(final Stage stage) {
+ protected void afterStageExecution() {
final ISignal signal = new TerminatingSignal();
for (InputPort<?> inputPort : inputPorts) {
stage.onSignal(signal, inputPort);
diff --git a/src/main/java/teetime/framework/RunnableProducerStage.java b/src/main/java/teetime/framework/RunnableProducerStage.java
index 2b7b556af36be9f5dd00b4c827c48dea1b831dc5..f8e9eb482df8956166be0ed421d1b51472d4ab9d 100644
--- a/src/main/java/teetime/framework/RunnableProducerStage.java
+++ b/src/main/java/teetime/framework/RunnableProducerStage.java
@@ -15,30 +15,53 @@
*/
package teetime.framework;
+import java.util.concurrent.Semaphore;
+
+import teetime.framework.signal.InitializingSignal;
import teetime.framework.signal.StartingSignal;
import teetime.framework.signal.TerminatingSignal;
-public final class RunnableProducerStage extends AbstractRunnableStage {
+final class RunnableProducerStage extends AbstractRunnableStage {
+
+ private final Semaphore startSemaphore = new Semaphore(0);
+ private final Semaphore initSemaphore = new Semaphore(0);
public RunnableProducerStage(final Stage stage) {
super(stage);
}
@Override
- protected void beforeStageExecution(final Stage stage) {
- final StartingSignal startingSignal = new StartingSignal();
- stage.onSignal(startingSignal, null);
+ protected void beforeStageExecution() throws InterruptedException {
+ waitForInitializingSignal();
+ this.stage.onSignal(new InitializingSignal(), null);
+ waitForStartingSignal();
+ this.stage.onSignal(new StartingSignal(), null);
}
@Override
- protected void executeStage(final Stage stage) {
- stage.executeStage();
+ protected void executeStage() {
+ this.stage.executeStage();
}
@Override
- protected void afterStageExecution(final Stage stage) {
+ protected void afterStageExecution() {
final TerminatingSignal terminatingSignal = new TerminatingSignal();
- stage.onSignal(terminatingSignal, null);
+ this.stage.onSignal(terminatingSignal, null);
}
+ public void triggerInitializingSignal() {
+ initSemaphore.release();
+ }
+
+ public void triggerStartingSignal() {
+ startSemaphore.release();
+ }
+
+ public void waitForInitializingSignal() throws InterruptedException {
+ initSemaphore.acquire();
+ }
+
+ public void waitForStartingSignal() throws InterruptedException {
+ startSemaphore.acquire();
+ }
}
diff --git a/src/main/java/teetime/framework/pipe/DummyPipe.java b/src/main/java/teetime/framework/pipe/DummyPipe.java
index 23031fe477fcca371ecca1ba43dbfc73a9527c62..2fe5d8833e237931be44dc2aa69ab3f3b4d8606b 100644
--- a/src/main/java/teetime/framework/pipe/DummyPipe.java
+++ b/src/main/java/teetime/framework/pipe/DummyPipe.java
@@ -86,6 +86,11 @@ public final class DummyPipe implements IPipe {
}
+ @Override
+ public void waitForInitializingSignal() throws InterruptedException {
+
+ }
+
@Override
public void close() {
diff --git a/src/main/java/teetime/framework/pipe/IPipe.java b/src/main/java/teetime/framework/pipe/IPipe.java
index f2fd15ecd4adf2a8f72f8d53e3f3fec9a9bf9a37..183aac72c8b50343ca9939dd1daf7ae7b4aa6bf7 100644
--- a/src/main/java/teetime/framework/pipe/IPipe.java
+++ b/src/main/java/teetime/framework/pipe/IPipe.java
@@ -92,6 +92,8 @@ public interface IPipe {
void waitForStartSignal() throws InterruptedException;
+ void waitForInitializingSignal() throws InterruptedException;
+
void close();
}
diff --git a/src/test/java/teetime/framework/RunnableProducerStageTest.java b/src/test/java/teetime/framework/RunnableProducerStageTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..387dd622513260e9ea08a0df5b21e7b06064513c
--- /dev/null
+++ b/src/test/java/teetime/framework/RunnableProducerStageTest.java
@@ -0,0 +1,28 @@
+package teetime.framework;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class RunnableProducerStageTest {
+
+ @Test
+ public void testInit() {
+ RunnableTestStage testStage = new RunnableTestStage();
+ RunnableProducerStage runnable = new RunnableProducerStage(testStage);
+ Thread thread = new Thread(runnable);
+ thread.start();
+ // Not running and not initialized
+ assertFalse(testStage.executed && testStage.initialized);
+ runnable.triggerInitializingSignal();
+ // Not running, but initialized
+ assertFalse(testStage.executed && !testStage.initialized);
+ runnable.triggerStartingSignal();
+
+ while (!testStage.shouldBeTerminated()) {
+ Thread.yield();
+ }
+ assertTrue(testStage.executed);
+ }
+}
diff --git a/src/test/java/teetime/framework/RunnableTestStage.java b/src/test/java/teetime/framework/RunnableTestStage.java
new file mode 100644
index 0000000000000000000000000000000000000000..034474a3a1ef81503f49f20d9d19f5a1aed7110c
--- /dev/null
+++ b/src/test/java/teetime/framework/RunnableTestStage.java
@@ -0,0 +1,25 @@
+package teetime.framework;
+
+
+class RunnableTestStage extends AbstractProducerStage<Object> {
+
+ boolean executed, initialized;
+
+ @Override
+ protected void executeStage() {
+ executed = true;
+ this.terminate();
+ }
+
+ @Override
+ protected void execute() {
+
+ }
+
+ @Override
+ public void onInitializing() throws Exception {
+ super.onInitializing();
+ initialized = true;
+ }
+
+}
diff --git a/src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java b/src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java
index 42696bf77420db1cbc4693771311b38ef5d47152..51c82d813ae7f7f70674160f60e9cc998fa124c8 100644
--- a/src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java
+++ b/src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java
@@ -102,6 +102,9 @@ class MergerTestingPipe implements IPipe {
@Override
public void waitForStartSignal() throws InterruptedException {}
+ @Override
+ public void waitForInitializingSignal() throws InterruptedException {}
+
@Override
public void close() {}