Skip to content

Commit a96a61f

Browse files
bukkabukka
committed
Report Io\Poll level watchers with buffered stream data
A stream can hold unread data in its buffer while its descriptor is idle. stream_select() treats such a stream as readable but a level watcher looked at the descriptor only, so a loop waiting for readability never woke up (GH-23948). Context::wait() now checks level watchers with Read interest for buffered data first and reports them with Read triggered. The backend is then asked without waiting so descriptor events are merged in and a watcher readable both ways is returned once. Edge watchers are not affected. A one-shot reported from its buffer is disarmed like one fired by the backend and modifyEvents() re-arms it. The check reads the stream buffer positions only and does no IO. The check is a handle operation so other handle types can provide it. It is registered separately as php_poll_handle_ext_ops because the handle ops struct cannot grow without an ABI change. This is temporary and the operation moves to the ops struct in master.
1 parent f5c6677 commit a96a61f

9 files changed

Lines changed: 395 additions & 15 deletions

‎NEWS‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,10 @@ PHP NEWS
9393
. Fixed reconstructing a SimpleXMLElement freeing a child element that
9494
another variable still references. (Ilia Alshanetsky)
9595

96+
- Standard:
97+
. Fixed bug GH-23948 (Io\Poll: data buffered by the stream layer is not
98+
reported). (Jakub Zelenka)
99+
96100
- Tidy:
97101
. Fixed a use-after-free when a tidyNode is used after its document is
98102
reparsed. (Ilia Alshanetsky)

‎ext/standard/io_poll.c‎

Lines changed: 114 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,9 @@ typedef struct php_io_poll_watcher_object {
5656
zval data;
5757
bool active;
5858
bool closed; /* Deactivated because its stream was closed */
59+
bool disarmed; /* Fired one-shot, until modifyEvents() */
60+
uint32_t wait_gen; /* Last wait() that reported it */
61+
const php_poll_handle_ext_ops *ext_ops;
5962
php_io_poll_context_object *context; /* Back reference to Context object */
6063
php_socket_t fd; /* Registered fd, SOCK_ERR when inactive */
6164
php_stream *stream; /* Watched stream, NULL when not registered in its watcher list */
@@ -66,6 +69,8 @@ typedef struct php_io_poll_watcher_object {
6669
struct php_io_poll_context_object {
6770
php_poll_ctx *ctx;
6871
HashTable *watchers; /* Maps fd -> watcher object */
72+
uint32_t wait_gen; /* Incremented per wait() */
73+
bool buffered_readers; /* Some watcher can report buffered data */
6974
zend_object std;
7075
};
7176

@@ -239,6 +244,17 @@ static php_poll_handle_ops php_stream_poll_handle_ops = {
239244
.cleanup = php_stream_poll_handle_cleanup
240245
};
241246

247+
/* The stream buffer only, without any IO */
248+
static bool php_stream_poll_handle_has_buffered_data(php_poll_handle_object *handle)
249+
{
250+
php_stream *stream = php_stream_poll_handle_get_stream(handle);
251+
return stream && stream->writepos > stream->readpos;
252+
}
253+
254+
static const php_poll_handle_ext_ops php_stream_poll_handle_ext_ops = {
255+
.has_buffered_data = php_stream_poll_handle_has_buffered_data
256+
};
257+
242258
/* Handle interface internal only */
243259
static int php_stream_poll_handle_implement_interface(zend_class_entry *interface, zend_class_entry *implementor)
244260
{
@@ -271,6 +287,9 @@ static zend_object *php_io_poll_watcher_create_object(zend_class_entry *ce)
271287
intern->triggered_events = 0;
272288
intern->active = false;
273289
intern->closed = false;
290+
intern->disarmed = false;
291+
intern->wait_gen = 0;
292+
intern->ext_ops = NULL;
274293
intern->context = NULL;
275294
intern->fd = SOCK_ERR;
276295
intern->stream = NULL;
@@ -288,6 +307,8 @@ static zend_object *php_io_poll_context_create_object(zend_class_entry *ce)
288307

289308
intern->ctx = NULL;
290309
intern->watchers = NULL;
310+
intern->wait_gen = 0;
311+
intern->buffered_readers = false;
291312

292313
return &intern->std;
293314
}
@@ -300,6 +321,13 @@ static zend_always_inline zend_ulong php_io_poll_compute_ptr_key(void *ptr)
300321
return (key >> 3) | (key << ((sizeof(key) * 8) - 3));
301322
}
302323

324+
/* Level Read watcher whose handle can have buffered data */
325+
static zend_always_inline bool php_io_poll_watcher_reads_buffered(php_io_poll_watcher_object *watcher)
326+
{
327+
return watcher->ext_ops && watcher->ext_ops->has_buffered_data
328+
&& (watcher->watched_events & (PHP_POLL_READ | PHP_POLL_ET)) == PHP_POLL_READ;
329+
}
330+
303331
static zend_always_inline void php_io_poll_watcher_deactivate(php_io_poll_watcher_object *watcher)
304332
{
305333
watcher->active = false;
@@ -461,6 +489,10 @@ static zend_result php_io_poll_watcher_modify_events(
461489

462490
/* Update watcher state */
463491
watcher->watched_events = events;
492+
watcher->disarmed = false;
493+
if (php_io_poll_watcher_reads_buffered(watcher)) {
494+
watcher->context->buffered_readers = true;
495+
}
464496

465497
return SUCCESS;
466498
}
@@ -841,6 +873,10 @@ PHP_METHOD(Io_Poll_Context, add)
841873
watcher->active = true;
842874
watcher->context = intern;
843875
watcher->fd = fd;
876+
watcher->ext_ops = php_poll_handle_get_ext_ops(handle->ops);
877+
if (php_io_poll_watcher_reads_buffered(watcher)) {
878+
intern->buffered_readers = true;
879+
}
844880

845881
if (handle->ops == &php_stream_poll_handle_ops) {
846882
php_stream *stream = php_stream_poll_handle_get_stream(handle);
@@ -889,33 +925,95 @@ PHP_METHOD(Io_Poll_Context, wait)
889925
RETURN_THROWS();
890926
}
891927

892-
php_poll_event *events = safe_emalloc((size_t) max_events, sizeof(*events), 0);
893-
int num_events = php_poll_wait(intern->ctx, events, (int) max_events, timeout ? &timeout_ts : NULL);
928+
/* Buffered watchers go first, then a non-blocking backend check */
929+
php_io_poll_watcher_object **buffered = NULL;
930+
int n_buffered = 0;
931+
if (intern->buffered_readers) {
932+
ZEND_HASH_FOREACH_VAL(intern->watchers, zval *zv) {
933+
php_io_poll_watcher_object *watcher = PHP_POLL_WATCHER_OBJ_FROM_ZOBJ(Z_OBJ_P(zv));
934+
if (watcher->disarmed || !php_io_poll_watcher_reads_buffered(watcher)
935+
|| !watcher->ext_ops->has_buffered_data(watcher->handle)) {
936+
continue;
937+
}
938+
if (!buffered) {
939+
buffered = safe_emalloc((size_t) max_events, sizeof(*buffered), 0);
940+
}
941+
buffered[n_buffered++] = watcher;
942+
if (n_buffered == max_events) {
943+
break;
944+
}
945+
} ZEND_HASH_FOREACH_END();
946+
}
894947

895-
if (num_events < 0) {
896-
php_poll_error err = php_poll_get_error(intern->ctx);
897-
efree(events);
898-
php_io_poll_throw_failed_operation(
899-
php_io_poll_failed_wait_class_entry, "Poll wait failed", err);
900-
RETURN_THROWS();
948+
php_poll_event *events = safe_emalloc((size_t) max_events, sizeof(*events), 0);
949+
int num_events = 0;
950+
if (n_buffered < max_events) {
951+
struct timespec zero_ts = {0, 0};
952+
num_events = php_poll_wait(intern->ctx, events, (int) max_events - n_buffered,
953+
n_buffered ? &zero_ts : (timeout ? &timeout_ts : NULL));
954+
if (num_events < 0) {
955+
php_poll_error err = php_poll_get_error(intern->ctx);
956+
if (n_buffered == 0 || err != PHP_POLL_ERR_INTERRUPTED) {
957+
efree(events);
958+
if (buffered) {
959+
efree(buffered);
960+
}
961+
php_io_poll_throw_failed_operation(
962+
php_io_poll_failed_wait_class_entry, "Poll wait failed", err);
963+
RETURN_THROWS();
964+
}
965+
/* The buffered data is the result */
966+
num_events = 0;
967+
}
901968
}
902969

903-
array_init(return_value);
970+
uint32_t gen = ++intern->wait_gen;
971+
array_init_size(return_value, (uint32_t) (n_buffered + num_events));
972+
973+
for (int i = 0; i < n_buffered; i++) {
974+
php_io_poll_watcher_object *watcher = buffered[i];
975+
watcher->triggered_events = PHP_POLL_READ;
976+
watcher->wait_gen = gen;
977+
978+
zval watcher_zv;
979+
ZVAL_OBJ_COPY(&watcher_zv, &watcher->std);
980+
add_next_index_zval(return_value, &watcher_zv);
981+
}
904982

905983
for (int i = 0; i < num_events; i++) {
906984
php_io_poll_watcher_object *watcher = (php_io_poll_watcher_object *) events[i].data;
907-
if (watcher) {
908-
watcher->triggered_events = events[i].revents;
985+
if (!watcher) {
986+
continue;
987+
}
988+
if (watcher->watched_events & PHP_POLL_ONESHOT) {
989+
watcher->disarmed = true;
990+
}
991+
if (watcher->wait_gen == gen) {
992+
/* Already reported from its buffer */
993+
watcher->triggered_events |= events[i].revents;
994+
continue;
995+
}
996+
watcher->triggered_events = events[i].revents;
997+
watcher->wait_gen = gen;
909998

910-
zval watcher_zv;
911-
ZVAL_OBJ(&watcher_zv, &watcher->std);
912-
GC_ADDREF(&watcher->std);
999+
zval watcher_zv;
1000+
ZVAL_OBJ_COPY(&watcher_zv, &watcher->std);
1001+
add_next_index_zval(return_value, &watcher_zv);
1002+
}
9131003

914-
add_next_index_zval(return_value, &watcher_zv);
1004+
/* Buffer-only one-shot is still armed in the backend */
1005+
for (int i = 0; i < n_buffered; i++) {
1006+
php_io_poll_watcher_object *watcher = buffered[i];
1007+
if ((watcher->watched_events & PHP_POLL_ONESHOT) && !watcher->disarmed) {
1008+
watcher->disarmed = true;
1009+
php_poll_remove(intern->ctx, (int) watcher->fd);
9151010
}
9161011
}
9171012

9181013
efree(events);
1014+
if (buffered) {
1015+
efree(buffered);
1016+
}
9191017
}
9201018

9211019
PHP_METHOD(Io_Poll_Context, getBackend)
@@ -1015,6 +1113,7 @@ PHP_MINIT_FUNCTION(poll)
10151113

10161114
/* Initialize polling backends */
10171115
php_poll_register_backends();
1116+
php_poll_handle_register_ext_ops(&php_stream_poll_handle_ops, &php_stream_poll_handle_ext_ops);
10181117

10191118
return SUCCESS;
10201119
}

‎ext/standard/tests/poll/poll_stream_buffered_edge.phpt‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
--TEST--
2+
Io\Poll: buffered data does not fire an edge-triggered watcher
3+
--SKIPIF--
4+
<?php
5+
require_once __DIR__ . '/poll.inc';
6+
pt_skip_for_backend(['Poll', 'WSAPoll', 'EventPorts'], 'does not support edge triggering')
7+
?>
8+
--FILE--
9+
<?php
10+
require_once __DIR__ . '/poll.inc';
11+
12+
[$a, $b] = pt_new_socket_pair();
13+
fwrite($b, "first\nsecond\n");
14+
var_dump(fgets($a));
15+
16+
$ctx = pt_new_stream_poll();
17+
$w = pt_stream_poll_add($ctx, $a, [Io\Poll\Event::Read, Io\Poll\Event::EdgeTriggered], 'a');
18+
19+
// Buffered data is state, an edge watcher reports transitions only
20+
var_dump(count($ctx->wait(Time\Duration::fromMilliseconds(100))));
21+
22+
// A new edge on the descriptor is reported as usual
23+
fwrite($b, "third\n");
24+
$events = $ctx->wait(Time\Duration::fromMilliseconds(100));
25+
var_dump(count($events), $events[0] === $w);
26+
?>
27+
--EXPECT--
28+
string(6) "first
29+
"
30+
int(0)
31+
int(1)
32+
bool(true)

‎ext/standard/tests/poll/poll_stream_buffered_level.phpt‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
--TEST--
2+
Io\Poll: a level watcher reports Read while the stream layer holds buffered data
3+
--FILE--
4+
<?php
5+
require_once __DIR__ . '/poll.inc';
6+
7+
[$a, $b] = pt_new_socket_pair();
8+
fwrite($b, "first\nsecond\n");
9+
var_dump(fgets($a)); // "second\n" is now in $a's read buffer, the socket is empty
10+
11+
$ctx = pt_new_stream_poll();
12+
$w = pt_stream_poll_add($ctx, $a, [Io\Poll\Event::Read], 'a');
13+
14+
// Returned at once although the descriptor is idle
15+
$start = hrtime(true);
16+
$events = $ctx->wait(Time\Duration::fromSeconds(5));
17+
var_dump(count($events), $events[0] === $w, $events[0]->hasTriggered(Io\Poll\Event::Read));
18+
var_dump(hrtime(true) - $start < 2_000_000_000);
19+
var_dump(stream_get_meta_data($a)['unread_bytes']);
20+
21+
// Level: reported on every wait while the data stays buffered
22+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0))));
23+
24+
var_dump(fgets($a));
25+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0))));
26+
27+
// Write interest is not concerned with read data
28+
fwrite($b, "third\nfourth\n");
29+
var_dump(fgets($a));
30+
$w->modifyEvents([Io\Poll\Event::Write]);
31+
$events = $ctx->wait(Time\Duration::fromSeconds(0));
32+
var_dump(count($events), $events[0]->hasTriggered(Io\Poll\Event::Write), $events[0]->hasTriggered(Io\Poll\Event::Read));
33+
34+
// The buffer counts as soon as Read interest is back
35+
$w->modifyEvents([Io\Poll\Event::Read]);
36+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0))));
37+
?>
38+
--EXPECT--
39+
string(6) "first
40+
"
41+
int(1)
42+
bool(true)
43+
bool(true)
44+
bool(true)
45+
int(7)
46+
int(1)
47+
string(7) "second
48+
"
49+
int(0)
50+
string(6) "third
51+
"
52+
int(1)
53+
bool(true)
54+
bool(false)
55+
int(1)

‎ext/standard/tests/poll/poll_stream_buffered_max_events.phpt‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
--TEST--
2+
Io\Poll: buffered data reports count against maxEvents
3+
--FILE--
4+
<?php
5+
require_once __DIR__ . '/poll.inc';
6+
7+
[$a, $b] = pt_new_socket_pair();
8+
[$c, $d] = pt_new_socket_pair();
9+
[$e, $f] = pt_new_socket_pair();
10+
$ctx = pt_new_stream_poll();
11+
pt_stream_poll_add($ctx, $a, [Io\Poll\Event::Read], 'a');
12+
pt_stream_poll_add($ctx, $c, [Io\Poll\Event::Read], 'c');
13+
pt_stream_poll_add($ctx, $e, [Io\Poll\Event::Read], 'e');
14+
15+
fwrite($b, "first\nsecond\n");
16+
fwrite($d, "first\nsecond\n");
17+
var_dump(fgets($a), fgets($c));
18+
fwrite($f, "x"); // on the socket, not buffered
19+
20+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0), 1)));
21+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0), 2)));
22+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0))));
23+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0), 3)));
24+
25+
var_dump(fgets($a), fgets($c), fread($e, 10));
26+
var_dump(count($ctx->wait(Time\Duration::fromSeconds(0))));
27+
?>
28+
--EXPECT--
29+
string(6) "first
30+
"
31+
string(6) "first
32+
"
33+
int(1)
34+
int(2)
35+
int(3)
36+
int(3)
37+
string(7) "second
38+
"
39+
string(7) "second
40+
"
41+
string(1) "x"
42+
int(0)

0 commit comments

Comments
 (0)