krakjoe · realFlowControl · Jul 15, 2026 · Jul 15, 2026 · Aug 4, 2026 · Aug 4, 2026
diff --git a/config.m4 b/config.m4
@@ -78,7 +78,7 @@ if test "$PHP_PARALLEL" != "no"; then
     AX_CHECK_COMPILE_FLAG(-ftest-coverage,          EXTRA_CFLAGS="$EXTRA_CFLAGS -ftest-coverage")
   fi
 
-  PHP_NEW_EXTENSION(parallel, php_parallel.c src/exceptions.c src/monitor.c src/parallel.c src/runtime.c src/scheduler.c src/future.c src/copy.c src/check.c src/dependencies.c src/cache.c src/channel.c src/link.c src/handlers.c src/events.c src/poll.c src/loop.c src/event.c src/input.c src/sync.c, $ext_shared,, "-Wall -DZEND_ENABLE_STATIC_TSRMLS_CACHE=1 $_MAINTAINER_CFLAGS")
+  PHP_NEW_EXTENSION(parallel, php_parallel.c src/exceptions.c src/notify.c src/monitor.c src/parallel.c src/runtime.c src/scheduler.c src/future.c src/copy.c src/check.c src/dependencies.c src/cache.c src/channel.c src/link.c src/handlers.c src/events.c src/poll.c src/loop.c src/event.c src/input.c src/sync.c, $ext_shared,, "-Wall -DZEND_ENABLE_STATIC_TSRMLS_CACHE=1 $_MAINTAINER_CFLAGS")
 
   PHP_ADD_BUILD_DIR($ext_builddir/src, 1)
   PHP_ADD_INCLUDE($ext_srcdir)

diff --git a/config.w32 b/config.w32
@@ -53,7 +53,7 @@ if (PHP_PARALLEL != 'no') {
 		EXTENSION("parallel", "php_parallel.c", PHP_PARALLEL_SHARED, "/DZEND_ENABLE_STATIC_TSRMLS_CACHE=1 /I" + configure_module_dirname);
 		ADD_SOURCES(
 			configure_module_dirname + "/src",
-			"exceptions.c copy.c check.c dependencies.c cache.c monitor.c parallel.c runtime.c scheduler.c future.c channel.c link.c handlers.c events.c poll.c loop.c event.c input.c sync.c",
+			"exceptions.c copy.c check.c dependencies.c cache.c notify.c monitor.c parallel.c runtime.c scheduler.c future.c channel.c link.c handlers.c events.c poll.c loop.c event.c input.c sync.c",
 			"parallel"
 		);
 	})();

diff --git a/src/link.c b/src/link.c
@@ -19,6 +19,7 @@
 #define HAVE_PARALLEL_LINK
 
 #include "link.h"
+#include "notify.h"
 #include "parallel.h"
 
 #define PHP_PARALLEL_LINK_CLOSURE_BUFFER GC_IMMUTABLE
@@ -42,18 +43,24 @@ typedef struct {
 	uint32_t w;
 } php_parallel_link_state_t;
 
+typedef struct {
+	php_parallel_notify_t read;
+	php_parallel_notify_t write;
+} php_parallel_link_notify_t;
+
 typedef struct _php_parallel_link_queue_t {
 	zend_llist l;
 	zend_long  c;
 } php_parallel_link_queue_t;
 
 struct _php_parallel_link_t {
-	php_parallel_link_type_t  type;
-	zend_string              *name;
+	php_parallel_link_type_t   type;
+	zend_string               *name;
 
-	php_parallel_link_mutex_t m;
-	php_parallel_link_cond_t  c;
-	php_parallel_link_state_t s;
+	php_parallel_link_mutex_t  m;
+	php_parallel_link_cond_t   c;
+	php_parallel_link_state_t  s;
+	php_parallel_link_notify_t notify;
 
 	union {
 		php_parallel_link_queue_t q;
@@ -139,6 +146,8 @@ php_parallel_link_t *php_parallel_link_init(zend_string *name, bool buffered, ze
 	} else {
 		link->type = PHP_PARALLEL_LINK_UNBUFFERED;
 	}
+	php_parallel_notify_init(&link->notify.read);
+	php_parallel_notify_init(&link->notify.write);
 	link->name = php_parallel_copy_string_interned(name);
 	link->refcount = 1;
 
@@ -148,6 +157,8 @@ php_parallel_link_t *php_parallel_link_init(zend_string *name, bool buffered, ze
 void php_parallel_link_destroy(php_parallel_link_t *link)
 {
 	if (php_parallel_atomic_delref(&link->refcount) == 0) {
+		php_parallel_notify_destroy(&link->notify.read);
+		php_parallel_notify_destroy(&link->notify.write);
 		php_parallel_link_mutex_destroy(&link->m);
 		php_parallel_link_cond_destroy(&link->c);
 
@@ -169,14 +180,20 @@ php_parallel_link_t *php_parallel_link_copy(php_parallel_link_t *link)
 	return link;
 }
 
-static zend_always_inline bool php_parallel_link_send_unbuffered(php_parallel_link_t *link, zval *value)
+static zend_always_inline void php_parallel_link_notify_sync(php_parallel_link_t *link)
 {
-	pthread_mutex_lock(&link->m.w);
-	pthread_mutex_lock(&link->m.m);
+	if (link->notify.read.read != -1) {
+		php_parallel_notify_sync(&link->notify.read, link->s.c || php_parallel_link_readable(link));
+	}
+
+	if (link->notify.write.read != -1) {
+		php_parallel_notify_sync(&link->notify.write, link->s.c || php_parallel_link_writable(link));
+	}
+}
 
+static zend_always_inline bool php_parallel_link_send_unbuffered_locked(php_parallel_link_t *link, zval *value)
+{
 	if (link->s.c) {
-		pthread_mutex_unlock(&link->m.m);
-		pthread_mutex_unlock(&link->m.w);
 		return 0;
 	}
 
@@ -189,17 +206,28 @@ static zend_always_inline bool php_parallel_link_send_unbuffered(php_parallel_li
 		ZEND_ASSERT(Z_TYPE_FLAGS(link->port.z) != PHP_PARALLEL_LINK_CLOSURE_BUFFER);
 	}
 	link->s.w++;
+	php_parallel_link_notify_sync(link);
 
 	if (link->s.r) {
 		pthread_cond_signal(&link->c.r);
 	}
 
 	pthread_cond_wait(&link->c.w, &link->m.m);
 
+	return 1;
+}
+
+static zend_always_inline bool php_parallel_link_send_unbuffered(php_parallel_link_t *link, zval *value)
+{
+	bool result;
+
+	pthread_mutex_lock(&link->m.w);
+	pthread_mutex_lock(&link->m.m);
+	result = php_parallel_link_send_unbuffered_locked(link, value);
 	pthread_mutex_unlock(&link->m.m);
 	pthread_mutex_unlock(&link->m.w);
 
-	return 1;
+	return result;
 }
 
 static zend_always_inline bool php_parallel_link_send_buffered(php_parallel_link_t *link, zval *value)
@@ -222,6 +250,7 @@ static zend_always_inline bool php_parallel_link_send_buffered(php_parallel_link
 	PARALLEL_ZVAL_COPY(&sent, value, 1);
 
 	zend_llist_add_element(&link->port.q.l, &sent);
+	php_parallel_link_notify_sync(link);
 
 	if (link->s.r) {
 		pthread_cond_signal(&link->c.r);
@@ -240,15 +269,35 @@ bool php_parallel_link_send(php_parallel_link_t *link, zval *value)
 	}
 }
 
+bool php_parallel_link_send_event(php_parallel_link_t *link, zval *value)
+{
+	bool result;
+
+	if (link->type == PHP_PARALLEL_LINK_BUFFERED) {
+		return php_parallel_link_send_buffered(link, value);
+	}
+
+	if (pthread_mutex_trylock(&link->m.w) != SUCCESS) {
+		return false;
+	}
+
+	result = php_parallel_link_send_unbuffered_locked(link, value);
+	pthread_mutex_unlock(&link->m.w);
+
+	return result;
+}
+
 static zend_always_inline bool php_parallel_link_recv_unbuffered(php_parallel_link_t *link, zval *value)
 {
 	pthread_mutex_lock(&link->m.r);
 	pthread_mutex_lock(&link->m.m);
 
 	while (!link->s.c && !link->s.w) {
 		link->s.r++;
+		php_parallel_link_notify_sync(link);
 		pthread_cond_wait(&link->c.r, &link->m.m);
 		link->s.r--;
+		php_parallel_link_notify_sync(link);
 	}
 
 	if (link->s.c) {
@@ -263,6 +312,7 @@ static zend_always_inline bool php_parallel_link_recv_unbuffered(php_parallel_li
 	}
 	ZVAL_UNDEF(&link->port.z);
 	link->s.w--;
+	php_parallel_link_notify_sync(link);
 
 	pthread_cond_signal(&link->c.w);
 	pthread_mutex_unlock(&link->m.m);
@@ -294,6 +344,7 @@ static zend_always_inline bool php_parallel_link_recv_buffered(php_parallel_link
 	PARALLEL_ZVAL_COPY(value, head, 0);
 
 	zend_llist_del_element(&link->port.q.l, head, php_parallel_link_queue_delete);
+	php_parallel_link_notify_sync(link);
 
 	if (link->s.w) {
 		pthread_cond_signal(&link->c.w);
@@ -323,6 +374,7 @@ bool php_parallel_link_close(php_parallel_link_t *link)
 	}
 
 	link->s.c = 1;
+	php_parallel_link_notify_sync(link);
 	pthread_cond_broadcast(&link->c.r);
 	pthread_cond_broadcast(&link->c.w);
 	pthread_mutex_unlock(&link->m.m);
@@ -362,6 +414,14 @@ bool php_parallel_link_readable(php_parallel_link_t *link)
 	return 0;
 }
 
+int php_parallel_link_notify(php_parallel_link_t *link, bool writable)
+{
+	php_parallel_notify_t *notify = writable ? &link->notify.write : &link->notify.read;
+	bool ready = link->s.c || (writable ? php_parallel_link_writable(link) : php_parallel_link_readable(link));
+
+	return php_parallel_notify_observe(notify, ready);
+}
+
 void php_parallel_link_debug(php_parallel_link_t *link, HashTable *debug)
 {
 	zval zdbg;

diff --git a/src/link.h b/src/link.h
@@ -26,6 +26,7 @@ php_parallel_link_t                *php_parallel_link_init(zend_string *name, bo
 zend_string                        *php_parallel_link_name(php_parallel_link_t *link);
 php_parallel_link_t                *php_parallel_link_copy(php_parallel_link_t *link);
 bool                                php_parallel_link_send(php_parallel_link_t *link, zval *value);
+bool                                php_parallel_link_send_event(php_parallel_link_t *link, zval *value);
 bool                                php_parallel_link_recv(php_parallel_link_t *link, zval *value);
 bool                                php_parallel_link_close(php_parallel_link_t *link);
 bool                                php_parallel_link_closed(php_parallel_link_t *link);
@@ -34,6 +35,7 @@ void                                php_parallel_link_destroy(php_parallel_link_
 bool                                php_parallel_link_lock(php_parallel_link_t *link);
 bool                                php_parallel_link_writable(php_parallel_link_t *link);
 bool                                php_parallel_link_readable(php_parallel_link_t *link);
+int                                 php_parallel_link_notify(php_parallel_link_t *link, bool writable);
 bool                                php_parallel_link_unlock(php_parallel_link_t *link);
 
 void                                php_parallel_link_debug(php_parallel_link_t *link, HashTable *debug);

diff --git a/src/monitor.c b/src/monitor.c
@@ -26,6 +26,7 @@ php_parallel_monitor_t *php_parallel_monitor_create(void)
 
 	php_parallel_mutex_init(&monitor->mutex, 1);
 	php_parallel_cond_init(&monitor->condition);
+	php_parallel_notify_init(&monitor->notify);
 
 	return monitor;
 }
@@ -64,6 +65,7 @@ int32_t php_parallel_monitor_wait(php_parallel_monitor_t *monitor, int32_t state
 	}
 
 	monitor->state ^= changed;
+	php_parallel_notify_sync(&monitor->notify, monitor->state & PHP_PARALLEL_READY);
 
 	if (pthread_mutex_unlock(&monitor->mutex) != SUCCESS) {
 		return FAILURE;
@@ -84,6 +86,7 @@ int32_t php_parallel_monitor_wait_locked(php_parallel_monitor_t *monitor, int32_
 	}
 
 	monitor->state ^= changed;
+	php_parallel_notify_sync(&monitor->notify, monitor->state & PHP_PARALLEL_READY);
 
 	return changed;
 }
@@ -93,6 +96,7 @@ void php_parallel_monitor_set(php_parallel_monitor_t *monitor, int32_t state)
 	pthread_mutex_lock(&monitor->mutex);
 
 	monitor->state |= state;
+	php_parallel_notify_sync(&monitor->notify, monitor->state & PHP_PARALLEL_READY);
 
 	pthread_cond_signal(&monitor->condition);
 
@@ -104,6 +108,7 @@ void php_parallel_monitor_add(php_parallel_monitor_t *monitor, int32_t state)
 	pthread_mutex_lock(&monitor->mutex);
 
 	monitor->state |= state;
+	php_parallel_notify_sync(&monitor->notify, monitor->state & PHP_PARALLEL_READY);
 
 	pthread_mutex_unlock(&monitor->mutex);
 }
@@ -113,12 +118,19 @@ void php_parallel_monitor_remove(php_parallel_monitor_t *monitor, int32_t state)
 	pthread_mutex_lock(&monitor->mutex);
 
 	monitor->state &= ~state;
+	php_parallel_notify_sync(&monitor->notify, monitor->state & PHP_PARALLEL_READY);
 
 	pthread_mutex_unlock(&monitor->mutex);
 }
 
+int php_parallel_monitor_notify(php_parallel_monitor_t *monitor)
+{
+	return php_parallel_notify_observe(&monitor->notify, monitor->state & PHP_PARALLEL_READY);
+}
+
 void php_parallel_monitor_destroy(php_parallel_monitor_t *monitor)
 {
+	php_parallel_notify_destroy(&monitor->notify);
 	php_parallel_mutex_destroy(&monitor->mutex);
 	php_parallel_cond_destroy(&monitor->condition);
 

diff --git a/src/monitor.h b/src/monitor.h
@@ -21,10 +21,13 @@
 #include <pthread.h>
 #include <stdint.h>
 
+#include "notify.h"
+
 typedef struct _php_parallel_monitor_t {
-	pthread_mutex_t mutex;
-	pthread_cond_t  condition;
-	int32_t         state;
+	pthread_mutex_t       mutex;
+	pthread_cond_t        condition;
+	int32_t               state;
+	php_parallel_notify_t notify;
 } php_parallel_monitor_t;
 
 #define PHP_PARALLEL_READY (1 << 0)
@@ -48,5 +51,6 @@ int32_t                 php_parallel_monitor_wait_locked(php_parallel_monitor_t
 void                    php_parallel_monitor_set(php_parallel_monitor_t *monitor, int32_t state);
 void                    php_parallel_monitor_add(php_parallel_monitor_t *monitor, int32_t state);
 void                    php_parallel_monitor_remove(php_parallel_monitor_t *monitor, int32_t state);
+int                     php_parallel_monitor_notify(php_parallel_monitor_t *monitor);
 void                    php_parallel_monitor_destroy(php_parallel_monitor_t *);
 #endif