diff --git a/config.m4 b/config.m4 index ccc2c258..d984829c 100644 --- 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 index 7c26e6b9..18eb7b00 100644 --- 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 index a49008d0..8fbc2e1f 100644 --- 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,6 +206,7 @@ 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); @@ -196,10 +214,20 @@ static zend_always_inline bool php_parallel_link_send_unbuffered(php_parallel_li 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,6 +269,24 @@ 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); @@ -247,8 +294,10 @@ static zend_always_inline bool php_parallel_link_recv_unbuffered(php_parallel_li 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 index d99af3cd..dd7151b3 100644 --- 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 index 573a7a4e..0eeb6865 100644 --- 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 index 01a8b1b9..6c6e528c 100644 --- a/src/monitor.h +++ b/src/monitor.h @@ -21,10 +21,13 @@ #include #include +#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 diff --git a/src/notify.c b/src/notify.c new file mode 100644 index 00000000..4cfed355 --- /dev/null +++ b/src/notify.c @@ -0,0 +1,116 @@ +/* + +----------------------------------------------------------------------+ + | parallel | + +----------------------------------------------------------------------+ + | Copyright (c) Florian Engelhardt 2026 | + +----------------------------------------------------------------------+ + | This source file is subject to version 3.01 of the PHP license, | + | that is bundled with this package in the file LICENSE, and is | + | available through the world-wide-web at the following url: | + | http://www.php.net/license/3_01.txt | + | If you did not receive a copy of the PHP license and are unable to | + | obtain it through the world-wide-web, please send a note to | + | license@php.net so we can mail you a copy immediately. | + +----------------------------------------------------------------------+ + | Author: Florian Engelhardt | + +----------------------------------------------------------------------+ + */ +#ifndef HAVE_PARALLEL_NOTIFY +#define HAVE_PARALLEL_NOTIFY + +#include "notify.h" + +#ifndef _WIN32 +#include +#include +#include +#endif + +void php_parallel_notify_init(php_parallel_notify_t *notify) +{ + notify->read = -1; + notify->write = -1; + notify->raised = false; +} + +#ifndef _WIN32 +static bool php_parallel_notify_flags(int descriptor) +{ + int flags = fcntl(descriptor, F_GETFL); + + if (flags == -1 || fcntl(descriptor, F_SETFL, flags | O_NONBLOCK) == -1) { + return false; + } + + flags = fcntl(descriptor, F_GETFD); + + return flags != -1 && fcntl(descriptor, F_SETFD, flags | FD_CLOEXEC) != -1; +} + +static bool php_parallel_notify_create(php_parallel_notify_t *notify) +{ + int descriptors[2]; + + if (pipe(descriptors) == -1) { + return false; + } + + if (!php_parallel_notify_flags(descriptors[0]) || !php_parallel_notify_flags(descriptors[1])) { + close(descriptors[0]); + close(descriptors[1]); + return false; + } + + notify->read = descriptors[0]; + notify->write = descriptors[1]; + + return true; +} + +#endif + +int php_parallel_notify_observe(php_parallel_notify_t *notify, bool ready) +{ +#ifdef _WIN32 + return -1; +#else + if (notify->read == -1 && !php_parallel_notify_create(notify)) { + return -1; + } + + php_parallel_notify_sync(notify, ready); + + return notify->read; +#endif +} + +void php_parallel_notify_sync(php_parallel_notify_t *notify, bool ready) +{ +#ifndef _WIN32 + char byte = 0; + ssize_t result; + + if (notify->read == -1 || notify->raised == ready) { + return; + } + + do { + result = ready ? write(notify->write, &byte, sizeof(byte)) : read(notify->read, &byte, sizeof(byte)); + } while (result == -1 && errno == EINTR); + + if (result == sizeof(byte) || (result == -1 && (errno == EAGAIN || errno == EWOULDBLOCK))) { + notify->raised = ready; + } +#endif +} + +void php_parallel_notify_destroy(php_parallel_notify_t *notify) +{ +#ifndef _WIN32 + if (notify->read != -1) { + close(notify->read); + close(notify->write); + } +#endif +} +#endif diff --git a/src/notify.h b/src/notify.h new file mode 100644 index 00000000..221a58cb --- /dev/null +++ b/src/notify.h @@ -0,0 +1,34 @@ +/* + +----------------------------------------------------------------------+ + | parallel | + +----------------------------------------------------------------------+ + | Copyright (c) Florian Engelhardt 2026 | + +----------------------------------------------------------------------+ + | This source file is subject to version 3.01 of the PHP license, | + | that is bundled with this package in the file LICENSE, and is | + | available through the world-wide-web at the following url: | + | http://www.php.net/license/3_01.txt | + | If you did not receive a copy of the PHP license and are unable to | + | obtain it through the world-wide-web, please send a note to | + | license@php.net so we can mail you a copy immediately. | + +----------------------------------------------------------------------+ + | Author: Florian Engelhardt | + +----------------------------------------------------------------------+ + */ +#ifndef HAVE_PARALLEL_NOTIFY_H +#define HAVE_PARALLEL_NOTIFY_H + +#include + +typedef struct _php_parallel_notify_t { + int read; + int write; + bool raised; +} php_parallel_notify_t; + +void php_parallel_notify_init(php_parallel_notify_t *notify); +int php_parallel_notify_observe(php_parallel_notify_t *notify, bool ready); +void php_parallel_notify_sync(php_parallel_notify_t *notify, bool ready); +void php_parallel_notify_destroy(php_parallel_notify_t *notify); + +#endif diff --git a/src/poll.c b/src/poll.c index 9042c269..b2687181 100644 --- a/src/poll.c +++ b/src/poll.c @@ -20,6 +20,23 @@ #include "parallel.h" +#if PHP_VERSION_ID >= 80300 +#include "Zend/zend_hrtime.h" +#endif + +#include +#include + +#ifdef _WIN32 +#if PHP_VERSION_ID < 80300 +#include +#endif +#else +#include +#endif + +#define PHP_PARALLEL_EVENTS_MICRO_IN_SEC 1000000ULL + #if PHP_VERSION_ID >= 80400 #include "ext/random/php_random.h" #else @@ -27,8 +44,9 @@ #endif typedef struct _php_parallel_events_poll_t { - uint32_t try; - struct timeval stop; + uint32_t try; + uint64_t stop; + bool native; struct { zend_fcall_info fci; zend_fcall_info_cache fcc; @@ -37,21 +55,42 @@ typedef struct _php_parallel_events_poll_t { php_parallel_events_state_t state; } php_parallel_events_poll_t; +static zend_always_inline uint64_t php_parallel_events_poll_now(void) +{ +#if PHP_VERSION_ID >= 80300 + return zend_hrtime() / 1000; +#elif defined(_WIN32) + LARGE_INTEGER now, frequency; + + QueryPerformanceCounter(&now); + QueryPerformanceFrequency(&frequency); + + return (now.QuadPart / frequency.QuadPart) * PHP_PARALLEL_EVENTS_MICRO_IN_SEC + + (now.QuadPart % frequency.QuadPart) * PHP_PARALLEL_EVENTS_MICRO_IN_SEC / frequency.QuadPart; +#else + struct timespec now; + + clock_gettime(CLOCK_MONOTONIC, &now); + + return (uint64_t)now.tv_sec * PHP_PARALLEL_EVENTS_MICRO_IN_SEC + now.tv_nsec / 1000; +#endif +} + static zend_always_inline php_parallel_events_poll_t *php_parallel_events_poll_init(php_parallel_events_t *events) { - if (events->targets.nNumUsed == 0) { + php_parallel_events_poll_t *poll; + + if (zend_hash_num_elements(&events->targets) == 0) { return NULL; } - php_parallel_events_poll_t *poll = (php_parallel_events_poll_t *)pecalloc(1, sizeof(php_parallel_events_poll_t), 1); + poll = (php_parallel_events_poll_t *)pecalloc(1, sizeof(php_parallel_events_poll_t), 1); +#ifndef _WIN32 + poll->native = true; +#endif if (events->timeout > -1) { - if (gettimeofday(&poll->stop, NULL) == SUCCESS) { - poll->stop.tv_sec += (events->timeout / 1000000L); - poll->stop.tv_sec += (poll->stop.tv_usec + (events->timeout % 1000000L)) / 1000000L; - poll->stop.tv_usec = (poll->stop.tv_usec + (events->timeout % 1000000L)) % 1000000L; - } - /* return 0 ? */ + poll->stop = php_parallel_events_poll_now() + (uint64_t)events->timeout; } if (!Z_ISUNDEF(events->blocker)) { @@ -85,41 +124,128 @@ static zend_always_inline void php_parallel_events_poll_end(php_parallel_events_ php_parallel_events_poll_free(poll); } -static zend_always_inline bool php_parallel_events_poll_timeout(php_parallel_events_poll_t *poll, +static zend_always_inline bool php_parallel_events_poll_expired(php_parallel_events_poll_t *poll, php_parallel_events_t *events) { - struct timeval now; - - if (events->timeout > -1 && gettimeofday(&now, NULL) == SUCCESS) { - if (now.tv_sec >= poll->stop.tv_sec && now.tv_usec >= poll->stop.tv_usec) { - php_parallel_exception_ex(php_parallel_events_error_timeout_ce, "timeout occured"); - return 1; - } + if (events->timeout > -1 && php_parallel_events_poll_now() >= poll->stop) { + php_parallel_exception_ex(php_parallel_events_error_timeout_ce, "timeout occured"); + return 1; } return 0; } -static zend_always_inline bool php_parallel_events_poll_random(php_parallel_events_t *events, zend_string **name, - zend_object **object) +static zend_always_inline bool php_parallel_events_poll_busy(php_parallel_events_poll_t *poll, + php_parallel_events_t *events) { - uint32_t size = events->targets.nNumUsed; - zend_long random = php_mt_rand_range(0, (zend_long)size - 1); + if ((poll->try++ % 10) == 0) { + usleep(1); + } - do { - Bucket *bucket = &events->targets.arData[random]; + return !php_parallel_events_poll_expired(poll, events); +} - if (!Z_ISUNDEF(bucket->val)) { - *name = bucket->key; - *object = Z_OBJ(bucket->val); +#ifndef _WIN32 +static zend_always_inline bool php_parallel_events_poll_descriptors(php_parallel_events_t *events, fd_set *readfds, + int *maximum) +{ + uint32_t index; - return 1; + FD_ZERO(readfds); + *maximum = -1; + + for (index = 0; index < events->targets.nNumUsed; index++) { + Bucket *bucket = &events->targets.arData[index]; + zend_object *object; + int descriptor; + + if (Z_ISUNDEF(bucket->val)) { + continue; } - random = php_mt_rand_range(0, (zend_long)size - 1); - } while (1); + object = Z_OBJ(bucket->val); - return 0; + if (instanceof_function(object->ce, php_parallel_channel_ce)) { + php_parallel_channel_t *channel = php_parallel_channel_fetch(object); + bool writable = php_parallel_events_input_exists(&events->input, bucket->key); + + php_parallel_link_lock(channel->link); + descriptor = php_parallel_link_notify(channel->link, writable); + php_parallel_link_unlock(channel->link); + } else { + php_parallel_future_t *future = php_parallel_future_fetch(object); + + php_parallel_future_lock(future); + descriptor = php_parallel_monitor_notify(future->monitor); + php_parallel_future_unlock(future); + } + + if (descriptor < 0 || descriptor >= FD_SETSIZE) { + return false; + } + + FD_SET(descriptor, readfds); + if (descriptor > *maximum) { + *maximum = descriptor; + } + } + + return *maximum >= 0; +} + +static zend_always_inline bool php_parallel_events_poll_native(php_parallel_events_poll_t *poll, + php_parallel_events_t *events) +{ + fd_set readfds; + int maximum; + int result; + struct timeval timeout; + struct timeval *timeout_pointer = NULL; + + if (!php_parallel_events_poll_descriptors(events, &readfds, &maximum)) { + poll->native = false; + return php_parallel_events_poll_busy(poll, events); + } + + if (events->timeout > -1) { + uint64_t now = php_parallel_events_poll_now(); + uint64_t remaining; + + if (now >= poll->stop) { + return !php_parallel_events_poll_expired(poll, events); + } + + remaining = poll->stop - now; + timeout.tv_sec = (long)(remaining / PHP_PARALLEL_EVENTS_MICRO_IN_SEC); + timeout.tv_usec = (long)(remaining % PHP_PARALLEL_EVENTS_MICRO_IN_SEC); + timeout_pointer = &timeout; + } + + result = select(maximum + 1, &readfds, NULL, NULL, timeout_pointer); + + if (result >= 0) { + return result > 0 || !php_parallel_events_poll_expired(poll, events); + } + + if (errno == EINTR) { + return true; + } + + poll->native = false; + return php_parallel_events_poll_busy(poll, events); +} +#endif + +static zend_always_inline bool php_parallel_events_poll_wait(php_parallel_events_poll_t *poll, + php_parallel_events_t *events) +{ +#ifndef _WIN32 + if (poll->native) { + return php_parallel_events_poll_native(poll, events); + } +#endif + + return php_parallel_events_poll_busy(poll, events); } static zend_always_inline bool php_parallel_events_poll_begin_link(php_parallel_events_t *events, @@ -177,19 +303,32 @@ static zend_always_inline bool php_parallel_events_poll_begin_future(php_paralle static zend_always_inline bool php_parallel_events_poll_begin(php_parallel_events_t *events, php_parallel_events_state_t *state) { - zend_string *name; - zend_object *object; + uint32_t size = events->targets.nNumUsed; + uint32_t index = (uint32_t)php_mt_rand_range(0, (zend_long)size - 1); + uint32_t scanned; - if (!php_parallel_events_poll_random(events, &name, &object)) { - return 0; - } + for (scanned = 0; scanned < size; scanned++) { + Bucket *bucket = &events->targets.arData[index]; + zend_object *object; - memset(state, 0, sizeof(php_parallel_events_state_t)); + if (++index == size) { + index = 0; + } - if (instanceof_function(object->ce, php_parallel_channel_ce)) { - return php_parallel_events_poll_begin_link(events, state, name, object); - } else { - return php_parallel_events_poll_begin_future(events, state, name, object); + if (Z_ISUNDEF(bucket->val)) { + continue; + } + + memset(state, 0, sizeof(php_parallel_events_state_t)); + object = Z_OBJ(bucket->val); + + if (instanceof_function(object->ce, php_parallel_channel_ce)) { + if (php_parallel_events_poll_begin_link(events, state, bucket->key, object)) { + return 1; + } + } else if (php_parallel_events_poll_begin_future(events, state, bucket->key, object)) { + return 1; + } } return 0; @@ -211,7 +350,7 @@ static zend_always_inline bool php_parallel_events_poll_link(php_parallel_events if (state->writable) { - if (php_parallel_link_send(channel->link, input)) { + if (php_parallel_link_send_event(channel->link, input)) { php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_WRITE, state->name, state->object, NULL, retval); @@ -293,13 +432,16 @@ void php_parallel_events_poll(php_parallel_events_t *events, zval *retval) } zval_ptr_dtor(&poll->block.ival); - } else { - if ((poll->try++ % 10) == 0) { - usleep(1); + + if (php_parallel_events_poll_expired(poll, events)) { + php_parallel_events_poll_free(poll); + return; } + + continue; } - if (php_parallel_events_poll_timeout(poll, events)) { + if (!php_parallel_events_poll_wait(poll, events)) { php_parallel_events_poll_free(poll); return; } diff --git a/tests/events/018.phpt b/tests/events/018.phpt new file mode 100644 index 00000000..89cfb747 --- /dev/null +++ b/tests/events/018.phpt @@ -0,0 +1,135 @@ +--TEST-- +Check Events notification-driven wakeups +--DESCRIPTION-- +Exercises repeated unbuffered reads, bounded buffered writes, unbuffered +writes, delayed Future completion among idle Channels, multiple observers of +a completed Future, and delayed Future errors. +--SKIPIF-- + +--FILE-- +run(static function (Channel $channel): void { + for ($i = 0; $i < 100; $i++) { + $channel->send($i); + } +}, [$channel]); +$events = new Events(); + +for ($i = 0; $i < 100; $i++) { + $events->addChannel($channel); + $event = $events->poll(); + + if ($event->type !== Type::Read || $event->value !== $i) { + echo "FAIL read\n"; + return; + } +} + +$future->value(); + +$channel = Channel::make('notify-buffered-write', 1); +$channel->send(-1); +$future = $runtime->run(static function (Channel $channel): int { + $sum = $channel->recv(); + + for ($i = 0; $i < 100; $i++) { + $sum += $channel->recv(); + } + + return $sum; +}, [$channel]); +$input = new Events\Input(); +$events = new Events(); +$events->setInput($input); + +for ($i = 0; $i < 100; $i++) { + $input->add('notify-buffered-write', $i); + $events->addChannel($channel); + + if ($events->poll()->type !== Type::Write) { + echo "FAIL buffered write\n"; + return; + } +} + +if ($future->value() !== 4949) { + echo "FAIL sum\n"; + return; +} + +$channel = Channel::make('notify-unbuffered-write'); +$future = $runtime->run(static function (Channel $channel): int { + usleep(20000); + return $channel->recv(); +}, [$channel]); +$input = new Events\Input(); +$input->add('notify-unbuffered-write', 42); +$events = new Events(); +$events->setInput($input); +$events->addChannel($channel); + +if ($events->poll()->type !== Type::Write || $future->value() !== 42) { + echo "FAIL unbuffered write\n"; + return; +} + +$start = Channel::make('notify-future'); +$future = $runtime->run(static function (Channel $start): int { + $start->recv(); + usleep(20000); + + return 42; +}, [$start]); +$events = new Events(); + +for ($i = 0; $i < 64; $i++) { + $events->addChannel(new Channel()); +} + +$events->addFuture('future', $future); +$start->send(true); +$event = $events->poll(); + +if ($event->source !== 'future' || $event->value !== 42) { + echo "FAIL future\n"; + return; +} + +$first = new Events(); +$second = new Events(); +$first->addFuture('first', $future); +$second->addFuture('second', $future); + +if ($first->poll()->value !== 42 || $second->poll()->value !== 42) { + echo "FAIL observers\n"; + return; +} + +$future = $runtime->run(static function (): void { + usleep(20000); + throw new RuntimeException('failed'); +}); +$events = new Events(); +$events->addFuture('error', $future); +$event = $events->poll(); + +if ($event->type !== Type::Error || !$event->value instanceof RuntimeException) { + echo "FAIL error\n"; + return; +} + +echo "OK\n"; +?> +--EXPECT-- +OK diff --git a/tests/events/019.phpt b/tests/events/019.phpt new file mode 100644 index 00000000..4aa6322f --- /dev/null +++ b/tests/events/019.phpt @@ -0,0 +1,78 @@ +--TEST-- +Check Events channel readiness levels +--DESCRIPTION-- +Exercises a lowered read notification timing out, the same notification being +raised by a delayed send and lowered when consumed, and Channel close raising +previously armed read and write notifications. +--SKIPIF-- + +--FILE-- +addChannel($channel); +$events->setTimeout(1000); + +try { + $events->poll(); + echo "FAIL lowered\n"; + return; +} catch (Events\Error\Timeout $error) { +} + +$events->setTimeout(-1); +$runtime = new Runtime(); +$future = $runtime->run(static function (Channel $channel): void { + usleep(20000); + $channel->send(2); +}, [$channel]); + +if ($events->poll()->value !== 2) { + echo "FAIL reraised\n"; + return; +} + +$future->value(); + +$channel = Channel::make('notify-close'); +$read = new Events(); +$read->addChannel($channel); +$read->setTimeout(1000); + +try { + $read->poll(); +} catch (Events\Error\Timeout $error) { +} + +$input = new Events\Input(); +$input->add('notify-close', true); +$write = new Events(); +$write->setInput($input); +$write->addChannel($channel); +$write->setTimeout(1000); + +try { + $write->poll(); +} catch (Events\Error\Timeout $error) { +} + +$channel->close(); + +if ($read->poll()->type !== Type::Close || $write->poll()->type !== Type::Close) { + echo "FAIL close\n"; + return; +} + +echo "OK\n"; +?> +--EXPECT-- +OK diff --git a/tests/events/020.phpt b/tests/events/020.phpt new file mode 100644 index 00000000..7a1935e2 --- /dev/null +++ b/tests/events/020.phpt @@ -0,0 +1,47 @@ +--TEST-- +Check Events polling fallback with more than 1000 Channels +--DESCRIPTION-- +Registers 1100 idle Channels before a delayed Future. On typical POSIX systems +this exceeds select's FD_SETSIZE while collecting notification descriptors, so +Events must fall back to busy polling and still deliver the Future before the +timeout. Other platforms still exercise delivery through a large target set. +--SKIPIF-- + +--FILE-- +run(static function (Channel $start): int { + $start->recv(); + usleep(20000); + + return 42; +}, [$start]); +$events = new Events(); + +for ($i = 0; $i < 1100; $i++) { + $events->addChannel(new Channel()); +} + +$events->addFuture('future', $future); +$events->setTimeout(1000000); +$start->send(true); +$event = $events->poll(); + +if ($event->source !== 'future' || $event->value !== 42) { + echo "FAIL fallback\n"; + return; +} + +echo "OK\n"; +?> +--EXPECT-- +OK