[php-src] master: main/poll: Fix kqueue event buffer overflow in grouped mode (#22606)
Jakub Zelenka via GitHub <[email protected]>
| Newsgroups | gmane.comp.php.cvs.general |
|---|---|
| Message-ID | <[email protected]> |
Author: Jakub Zelenka (bukka)
Committer: GitHub (web-flow)
Pusher: bukka
Date: 2026-07-12T20:20:53+02:00
Commit: https://github.com/php/php-src/commit/c6c8db5a0061c352a9b208d0c00124978afb96b5
Raw diff: https://github.com/php/php-src/commit/c6c8db5a0061c352a9b208d0c00124978afb96b5.diff
main/poll: Fix kqueue event buffer overflow in grouped mode (#22606)
Cap each kevent() request at the number of entries that can still be
delivered instead of requesting twice the buffer size. Events that are
not retrieved (including oneshot and edge-triggered ones) stay pending
in the kqueue and are delivered by the next wait, matching the epoll
behavior. Merged read+write events can under-fill the buffer, so top up
with zero timeout rounds, and base the suitable max events on the
filter count so a default sized wait retrieves everything in one call.
Changed paths:
A ext/standard/tests/poll/poll_stream_sock_rw_max_events.phpt
A ext/standard/tests/poll/poll_stream_tcp_read_max_events.phpt
A ext/standard/tests/poll/poll_stream_tcp_read_one_shot_max_events.phpt
M main/poll/poll_backend_kqueue.c
Diff:
diff --git a/ext/standard/tests/poll/poll_stream_sock_rw_max_events.phpt b/ext/standard/tests/poll/poll_stream_sock_rw_max_events.phpt
new file mode 100644
index 000000000000..d78d3f4fde35
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_sock_rw_max_events.phpt
@@ -0,0 +1,38 @@
+--TEST--
+Poll stream - socket read/write events with maxEvents equal to the FD count
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+$poll_ctx = pt_new_stream_poll();
+
+// Each FD is ready for both read and write. Backends that report read and
+// write separately (kqueue) see twice as many events as FDs and have to merge
+// them without dropping any FD, filling only half of the maxEvents entries.
+$writers = [];
+$expected = [];
+for ($i = 0; $i < 4; $i++) {
+ list($reader, $writer) = pt_new_socket_pair();
+ $writers[$i] = $writer;
+ pt_stream_poll_add($poll_ctx, $reader, [Io\Poll\Event::Read, Io\Poll\Event::Write], "sock$i");
+ pt_write_sleep($writer, "test $i data");
+ $expected[] = [
+ 'events' => [Io\Poll\Event::Read, Io\Poll\Event::Write],
+ 'data' => "sock$i",
+ 'read' => "test $i data",
+ ];
+}
+
+pt_expect_events($poll_ctx->wait(0, 100000, 8), $expected);
+
+// All read data was drained above, so only write events remain
+$expected = [];
+for ($i = 0; $i < 4; $i++) {
+ $expected[] = ['events' => [Io\Poll\Event::Write], 'data' => "sock$i"];
+}
+pt_expect_events($poll_ctx->wait(0, 100000, 8), $expected);
+
+?>
+--EXPECT--
+Events matched - count: 4
+Events matched - count: 4
diff --git a/ext/standard/tests/poll/poll_stream_tcp_read_max_events.phpt b/ext/standard/tests/poll/poll_stream_tcp_read_max_events.phpt
new file mode 100644
index 000000000000..f927e73ed4d3
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_tcp_read_max_events.phpt
@@ -0,0 +1,45 @@
+--TEST--
+Poll stream - TCP read with maxEvents smaller than the number of ready FDs
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+list($clients, $servers) = pt_new_tcp_socket_connections(8);
+$poll_ctx = pt_new_stream_poll();
+
+for ($i = 0; $i < count($servers); $i++) {
+ pt_stream_poll_add($poll_ctx, $servers[$i], [Io\Poll\Event::Read], "server$i");
+}
+
+for ($i = 0; $i < count($clients); $i++) {
+ pt_write_sleep($clients[$i], "test $i data");
+}
+
+// 8 FDs are ready but only 4 events fit into the buffer, so the remaining
+// events must be delivered by the following wait() without any duplicates.
+$seen = [];
+for ($round = 1; $round <= 2; $round++) {
+ $watchers = $poll_ctx->wait(0, 100000, 4);
+ echo "Round $round count: " . count($watchers) . "\n";
+ foreach ($watchers as $watcher) {
+ $data = $watcher->getData();
+ if (isset($seen[$data])) {
+ echo "Duplicate event for $data\n";
+ }
+ $seen[$data] = true;
+ // Drain so level triggering does not re-report the FD
+ fread($watcher->getHandle()->getStream(), 1024);
+ }
+}
+
+ksort($seen);
+echo "Seen: " . implode(',', array_keys($seen)) . "\n";
+
+pt_expect_events($poll_ctx->wait(0), []);
+
+?>
+--EXPECT--
+Round 1 count: 4
+Round 2 count: 4
+Seen: server0,server1,server2,server3,server4,server5,server6,server7
+Events matched - count: 0
diff --git a/ext/standard/tests/poll/poll_stream_tcp_read_one_shot_max_events.phpt b/ext/standard/tests/poll/poll_stream_tcp_read_one_shot_max_events.phpt
new file mode 100644
index 000000000000..d7a2e48be6f4
--- /dev/null
+++ b/ext/standard/tests/poll/poll_stream_tcp_read_one_shot_max_events.phpt
@@ -0,0 +1,46 @@
+--TEST--
+Poll stream - TCP read oneshot with maxEvents smaller than the number of ready FDs
+--FILE--
+<?php
+require_once __DIR__ . '/poll.inc';
+
+list($clients, $servers) = pt_new_tcp_socket_connections(8);
+$poll_ctx = pt_new_stream_poll();
+
+for ($i = 0; $i < count($servers); $i++) {
+ pt_stream_poll_add($poll_ctx, $servers[$i], [Io\Poll\Event::Read, Io\Poll\Event::OneShot], "server$i");
+}
+
+for ($i = 0; $i < count($clients); $i++) {
+ pt_write_sleep($clients[$i], "test $i data");
+}
+
+// 8 oneshot FDs are ready but only 4 events fit into the buffer. The events
+// that did not fit must not be consumed - the following wait() has to deliver
+// them, and each oneshot event must be delivered exactly once.
+$seen = [];
+for ($round = 1; $round <= 2; $round++) {
+ $watchers = $poll_ctx->wait(0, 100000, 4);
+ echo "Round $round count: " . count($watchers) . "\n";
+ foreach ($watchers as $watcher) {
+ $data = $watcher->getData();
+ if (isset($seen[$data])) {
+ echo "Duplicate event for $data\n";
+ }
+ $seen[$data] = true;
+ }
+}
+
+ksort($seen);
+echo "Seen: " . implode(',', array_keys($seen)) . "\n";
+
+// Every oneshot watcher fired once, so new data must not be reported
+pt_write_sleep($clients[0], "more data");
+pt_expect_events($poll_ctx->wait(0, 100000), []);
+
+?>
+--EXPECT--
+Round 1 count: 4
+Round 2 count: 4
+Seen: server0,server1,server2,server3,server4,server5,server6,server7
+Events matched - count: 0
diff --git a/main/poll/poll_backend_kqueue.c b/main/poll/poll_backend_kqueue.c
index 00ba7abc9ee6..19ff0ad6d228 100644
--- a/main/poll/poll_backend_kqueue.c
+++ b/main/poll/poll_backend_kqueue.c
@@ -303,149 +303,179 @@ static int kqueue_backend_wait(
{
kqueue_backend_data_t *backend_data = (kqueue_backend_data_t *) ctx->backend_data;
- /* For raw events, we need capacity for max_events.
- * For grouped events, kqueue can return up to 2 events per FD, so we need 2x capacity. */
- int required_capacity = ctx->raw_events ? max_events : (max_events * 2);
- if (required_capacity > backend_data->events_capacity) {
+ /* Cap every kevent() request at the entries we can still deliver, like epoll's maxevents.
+ * Unretrieved events stay pending in the kqueue (EV_ONESHOT deletes and EV_CLEAR resets only
+ * at retrieval), so nothing is lost and the caller's buffer cannot overflow. */
+ if (max_events > backend_data->events_capacity) {
struct kevent *new_events = php_poll_realloc(
- backend_data->events, required_capacity * sizeof(struct kevent), ctx->persistent);
+ backend_data->events, max_events * sizeof(struct kevent), ctx->persistent);
if (!new_events) {
php_poll_set_error(ctx, PHP_POLL_ERR_NOMEM);
return -1;
}
backend_data->events = new_events;
- backend_data->events_capacity = required_capacity;
+ backend_data->events_capacity = max_events;
}
- int nfds = kevent(
- backend_data->kqueue_fd, NULL, 0, backend_data->events, required_capacity, timeout);
+ if (ctx->raw_events) {
+ /* Raw events mode - direct 1:1 mapping, no grouping */
+ int nfds = kevent(
+ backend_data->kqueue_fd, NULL, 0, backend_data->events, max_events, timeout);
- if (nfds < 0) {
- php_poll_set_current_errno_error(ctx);
- return -1;
+ if (nfds < 0) {
+ php_poll_set_current_errno_error(ctx);
+ return -1;
+ }
+
+ for (int i = 0; i < nfds; i++) {
+ events[i].fd = (int) backend_data->events[i].ident;
+ events[i].events = 0;
+ events[i].revents = 0;
+ events[i].data = backend_data->events[i].udata;
+
+ /* Convert kqueue filter to poll event */
+ if (backend_data->events[i].filter == EVFILT_READ) {
+ events[i].revents |= PHP_POLL_READ;
+ } else if (backend_data->events[i].filter == EVFILT_WRITE) {
+ events[i].revents |= PHP_POLL_WRITE;
+ }
+
+ /* Convert kqueue flags to poll events */
+ if (backend_data->events[i].flags & EV_EOF) {
+ events[i].revents |= PHP_POLL_HUP;
+ }
+ if (backend_data->events[i].flags & EV_ERROR) {
+ events[i].revents |= PHP_POLL_ERROR;
+ }
+ }
+
+ return nfds;
}
- if (nfds > 0) {
- if (ctx->raw_events) {
- /* Raw events mode - direct 1:1 mapping, no grouping */
- for (int i = 0; i < nfds && i < max_events; i++) {
- events[i].fd = (int) backend_data->events[i].ident;
- events[i].events = 0;
- events[i].revents = 0;
- events[i].data = backend_data->events[i].udata;
-
- /* Convert kqueue filter to poll event */
- if (backend_data->events[i].filter == EVFILT_READ) {
- events[i].revents |= PHP_POLL_READ;
- } else if (backend_data->events[i].filter == EVFILT_WRITE) {
- events[i].revents |= PHP_POLL_WRITE;
- }
+ /* Grouped events mode with improved oneshot tracking.
+ *
+ * kqueue reports one event per (FD, filter) pair, so a READ+WRITE pair merges into a single
+ * entry and one round can fill fewer than max_events entries while more events are pending.
+ * Top up with zero-timeout rounds, each requesting only the remaining free entries. Both
+ * FreeBSD and XNU re-enqueue delivered level-triggered events at the tail of the active queue,
+ * so top-up rounds see not-yet-delivered events first, matching epoll's round-robin fill. */
+ int unique_events = 0, garbage_events = 0;
+ const struct timespec zero_timeout = {0, 0};
+ const struct timespec *round_timeout = timeout;
+
+ while (true) {
+ int slots = max_events - unique_events;
+ int nfds = kevent(
+ backend_data->kqueue_fd, NULL, 0, backend_data->events, slots, round_timeout);
+
+ if (nfds < 0) {
+ if (unique_events > 0) {
+ /* Deliver what was already gathered */
+ break;
+ }
+ php_poll_set_current_errno_error(ctx);
+ return -1;
+ }
- /* Convert kqueue flags to poll events */
- if (backend_data->events[i].flags & EV_EOF) {
- events[i].revents |= PHP_POLL_HUP;
- }
- if (backend_data->events[i].flags & EV_ERROR) {
- events[i].revents |= PHP_POLL_ERROR;
- }
+ int new_unique_events = 0;
+
+ for (int i = 0; i < nfds; i++) {
+ int fd = (int) backend_data->events[i].ident;
+ uint32_t revents = 0;
+ void *data = backend_data->events[i].udata;
+ bool is_oneshot = (backend_data->events[i].flags & EV_ONESHOT) != 0;
+
+ /* Convert this event */
+ if (backend_data->events[i].filter == EVFILT_READ) {
+ revents |= PHP_POLL_READ;
+ } else if (backend_data->events[i].filter == EVFILT_WRITE) {
+ revents |= PHP_POLL_WRITE;
}
- return nfds > max_events ? max_events : nfds;
- } else {
- /* Grouped events mode with improved oneshot tracking */
- int unique_events = 0, garbage_events = 0, fd;
-
- for (int i = 0; i < nfds; i++) {
- fd = (int) backend_data->events[i].ident;
- uint32_t revents = 0;
- void *data = backend_data->events[i].udata;
- bool is_oneshot = (backend_data->events[i].flags & EV_ONESHOT) != 0;
-
- /* Convert this event */
- if (backend_data->events[i].filter == EVFILT_READ) {
- revents |= PHP_POLL_READ;
- } else if (backend_data->events[i].filter == EVFILT_WRITE) {
- revents |= PHP_POLL_WRITE;
- }
- if (backend_data->events[i].flags & EV_EOF) {
- revents |= PHP_POLL_HUP;
- }
- if (backend_data->events[i].flags & EV_ERROR) {
- revents |= PHP_POLL_ERROR;
- }
+ if (backend_data->events[i].flags & EV_EOF) {
+ revents |= PHP_POLL_HUP;
+ }
+ if (backend_data->events[i].flags & EV_ERROR) {
+ revents |= PHP_POLL_ERROR;
+ }
- /* Look for existing event for this FD */
- bool found = false;
- for (int j = 0; j < unique_events; j++) {
- if (events[j].fd == fd) {
- /* Combine with existing event */
- events[j].revents |= revents;
- found = true;
- break;
- }
+ /* Look for existing event for this FD */
+ bool found = false;
+ for (int j = 0; j < unique_events; j++) {
+ if (events[j].fd == fd) {
+ /* Combine with existing event */
+ events[j].revents |= revents;
+ found = true;
+ break;
}
+ }
- if (!found) {
- /* New FD, create new event */
- ZEND_ASSERT(unique_events < max_events);
- events[unique_events].fd = fd;
- events[unique_events].events = 0;
- events[unique_events].revents = revents;
- events[unique_events].data = data;
- unique_events++;
-
- /* Handle oneshot tracking */
- if (is_oneshot) {
- zval *tracking = zend_hash_index_find(backend_data->fd_tracking, fd);
- if (tracking && (Z_LVAL_P(tracking) & KQUEUE_FD_ONESHOT_COMPLETE)) {
- /* Mark which filter fired for garbage collection */
- zend_long flags = Z_LVAL_P(tracking);
- flags &= ~KQUEUE_FD_ONESHOT_COMPLETE; /* Clear complete flag */
- if (revents & PHP_POLL_READ) {
- flags |= KQUEUE_FD_GARBAGE_READ; /* Need to clean write */
- }
- if (revents & PHP_POLL_WRITE) {
- flags |= KQUEUE_FD_GARBAGE_WRITE; /* Need to clean read */
- }
- ZVAL_LONG(tracking, flags);
- backend_data->fd_count--;
- garbage_events++;
- }
- }
- } else if (is_oneshot) {
- /* Second filter for same FD fired - clear garbage flags */
+ if (!found) {
+ /* New FD, create new event */
+ ZEND_ASSERT(unique_events < max_events);
+ events[unique_events].fd = fd;
+ events[unique_events].events = 0;
+ events[unique_events].revents = revents;
+ events[unique_events].data = data;
+ unique_events++;
+ new_unique_events++;
+
+ /* Handle oneshot tracking */
+ if (is_oneshot) {
zval *tracking = zend_hash_index_find(backend_data->fd_tracking, fd);
- if (tracking) {
- /* Remove FD from tracking as it gets deleted from kqueue as well */
- zend_hash_index_del(backend_data->fd_tracking, fd);
- garbage_events--;
+ if (tracking && (Z_LVAL_P(tracking) & KQUEUE_FD_ONESHOT_COMPLETE)) {
+ /* Mark which filter fired for garbage collection */
+ zend_long flags = Z_LVAL_P(tracking);
+ flags &= ~KQUEUE_FD_ONESHOT_COMPLETE; /* Clear complete flag */
+ if (revents & PHP_POLL_READ) {
+ flags |= KQUEUE_FD_GARBAGE_READ; /* Need to clean write */
+ }
+ if (revents & PHP_POLL_WRITE) {
+ flags |= KQUEUE_FD_GARBAGE_WRITE; /* Need to clean read */
+ }
+ ZVAL_LONG(tracking, flags);
+ backend_data->fd_count--;
+ garbage_events++;
}
}
- }
-
- if (garbage_events > 0) {
- /* Clean up orphaned filters from complete oneshot FDs */
- struct kevent cleanup_change;
- ZEND_HASH_FOREACH_NUM_KEY_VAL(backend_data->fd_tracking, zend_ulong fd_key, zval *tracking)
- {
- zend_long flags = Z_LVAL_P(tracking);
- if (flags & KQUEUE_FD_HAS_GARBAGE) {
- int filter = (flags & KQUEUE_FD_GARBAGE_READ) ? EVFILT_WRITE : EVFILT_READ;
- EV_SET(&cleanup_change, fd_key, filter, EV_DELETE, 0, 0, NULL);
- kevent(backend_data->kqueue_fd, &cleanup_change, 1, NULL, 0, NULL);
-
- /* Remove FD from tracking after cleanup */
- zend_hash_index_del(backend_data->fd_tracking, fd_key);
- }
+ } else if (is_oneshot) {
+ /* Second filter for same FD fired - clear garbage flags */
+ zval *tracking = zend_hash_index_find(backend_data->fd_tracking, fd);
+ if (tracking) {
+ /* Remove FD from tracking as it gets deleted from kqueue as well */
+ zend_hash_index_del(backend_data->fd_tracking, fd);
+ garbage_events--;
}
- ZEND_HASH_FOREACH_END();
}
+ }
- return unique_events;
+ /* Stop on a partial round (drained or timed out), a full buffer, or a full round with no
+ * new FDs (kernel re-reported level-triggered events already merged above - don't spin) */
+ if (nfds < slots || unique_events == max_events || new_unique_events == 0) {
+ break;
}
+ round_timeout = &zero_timeout;
}
- return nfds;
+ if (garbage_events > 0) {
+ /* Clean up orphaned filters from complete oneshot FDs */
+ struct kevent cleanup_change;
+ ZEND_HASH_FOREACH_NUM_KEY_VAL(backend_data->fd_tracking, zend_ulong fd_key, zval *tracking)
+ {
+ zend_long flags = Z_LVAL_P(tracking);
+ if (flags & KQUEUE_FD_HAS_GARBAGE) {
+ int filter = (flags & KQUEUE_FD_GARBAGE_READ) ? EVFILT_WRITE : EVFILT_READ;
+ EV_SET(&cleanup_change, fd_key, filter, EV_DELETE, 0, 0, NULL);
+ kevent(backend_data->kqueue_fd, &cleanup_change, 1, NULL, 0, NULL);
+
+ /* Remove FD from tracking after cleanup */
+ zend_hash_index_del(backend_data->fd_tracking, fd_key);
+ }
+ }
+ ZEND_HASH_FOREACH_END();
+ }
+
+ return unique_events;
}
static bool kqueue_backend_is_available(void)
@@ -466,15 +496,11 @@ static int kqueue_backend_get_suitable_max_events(php_poll_ctx *ctx)
return -1;
}
- if (ctx->raw_events) {
- /* For raw events, return the total number of filters */
- int active_filters = backend_data->filter_count;
- return active_filters == 0 ? 1 : active_filters;
- } else {
- /* For grouped events, return the number of unique FDs */
- int active_fds = backend_data->fd_count;
- return active_fds == 0 ? 1 : active_fds;
- }
+ /* kqueue reports one event per filter even in grouped mode (events are merged into
+ * per-FD entries afterwards) and the wait request is capped at max_events, so size
+ * for the total filter count to allow retrieving all pending events in one call. */
+ int active_filters = backend_data->filter_count;
+ return active_filters == 0 ? 1 : active_filters;
}
const php_poll_backend_ops php_poll_backend_kqueue_ops = {