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 92fe284b21c74a10efe6b84be454639acbd59348..ebeccd1c4b592bf55cedd1074a0c13f9874bbbff 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/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..7a1d071d80c5740ae9653fb04b34b2b0157d99df 100644
--- a/src/main/java/teetime/framework/RunnableConsumerStage.java
+++ b/src/main/java/teetime/framework/RunnableConsumerStage.java
@@ -43,14 +43,14 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
 	@SuppressWarnings("PMD.GuardLogStatement")
 	@Override
 	protected void beforeStageExecution(final Stage stage) throws InterruptedException {
-		logger.trace("Waiting for start signals..." + inputPorts);
+		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
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/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() {}