Do not report ECANCELED under a pending exception as a stream failure

ext/standard/tests/streams/hooks/data-ops-cancel-pipe-not-eof.phpt

@@ -0,0 +1,44 @@
+--TEST--
+IO hooks: a pipe read or write the provider abandons with an exception is not a stream error
+--SKIPIF--
+<?php if (PHP_OS_FAMILY === 'Windows') die('skip plain-wrapper pipes are not hooked on Windows'); ?>
+--FILE--
+<?php
+$throwing = new class implements Io\Hooks\Hooks {
+    public function getCapabilities(): array { return [Io\Hooks\Capability::DirectData]; }
+    public function add(Io\Registration $registration): void {}
+    public function remove(Io\Registration $registration): void {}
+    public function run(Io\Operation $op): Io\Completion {
+        throw new RuntimeException("cancelled " . $op::class);
+    }
+};
+
+function cancelled(Io\Hooks\Hooks $hooks, callable $fn): void
+{
+    Io\Hooks\set_hooks($hooks);
+    try {
+        $fn();
+    } catch (RuntimeException $e) {
+        echo $e->getMessage(), "\n";
+    } finally {
+        Io\Hooks\set_hooks(null);
+    }
+}
+
+$php = getenv('TEST_PHP_EXECUTABLE');
+$proc = proc_open([$php, '-n', '-r', 'echo fgets(STDIN);'], [['pipe', 'r'], ['pipe', 'w']], $pipes);
+cancelled($throwing, fn () => fread($pipes[1], 10));
+cancelled($throwing, fn () => fwrite($pipes[0], "lost\n"));
+var_dump(feof($pipes[1]));
+fwrite($pipes[0], "ping\n");
+var_dump(fgets($pipes[1]));
+fclose($pipes[0]);
+fclose($pipes[1]);
+proc_close($proc);
+?>
+--EXPECT--
+cancelled Io\Operation\Read
+cancelled Io\Operation\Write
+bool(false)
+string(5) "ping
+"

ext/standard/tests/streams/hooks/data-ops-cancel-socket-not-eof.phpt

@@ -0,0 +1,36 @@
+--TEST--
+IO hooks: a socket read the provider abandons with an exception does not set eof
+--FILE--
+<?php
+$throwing = new class implements Io\Hooks\Hooks {
+    public function getCapabilities(): array { return [Io\Hooks\Capability::DirectData]; }
+    public function add(Io\Registration $registration): void {}
+    public function remove(Io\Registration $registration): void {}
+    public function run(Io\Operation $op): Io\Completion {
+        throw new RuntimeException("cancelled " . $op::class);
+    }
+};
+
+function cancelled(Io\Hooks\Hooks $hooks, callable $fn): void
+{
+    Io\Hooks\set_hooks($hooks);
+    try {
+        $fn();
+    } catch (RuntimeException $e) {
+        echo $e->getMessage(), "\n";
+    } finally {
+        Io\Hooks\set_hooks(null);
+    }
+}
+
+[$r, $w] = stream_socket_pair(PHP_OS_FAMILY === 'Windows' ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+cancelled($throwing, fn () => fread($r, 10));
+var_dump(feof($r));
+fwrite($w, "x");
+var_dump(fread($r, 10));
+
+?>
+--EXPECT--
+cancelled Io\Operation\Recv
+bool(false)
+string(1) "x"

ext/standard/tests/streams/hooks/registrations-ring-accept-select.phpt

@@ -0,0 +1,83 @@
+--TEST--
+Io\Ring\Engine: a select on a listener sees the connections its multishot accept took
+--SKIPIF--
+<?php
+if (!class_exists(Io\Ring\Engine::class)) die("skip Io\\Ring\\Engine not available");
+$caps = (new Io\Ring\Engine())->getSupportedHookCapabilities();
+if (!in_array(Io\Hooks\Capability::DirectAccept, $caps, true)) {
+    die("skip the backend does not support DirectAccept");
+}
+if (!in_array(Io\Hooks\Capability::EdgeRegistrations, $caps, true)) {
+    die("skip the listener's pair needs EdgeRegistrations");
+}
+?>
+--FILE--
+<?php
+include __DIR__ . '/scheduler.inc';
+
+$scheduler = new Scheduler(new Io\Ring\Engine(),
+    [Io\Hooks\Capability::EdgeRegistrations, Io\Hooks\Capability::DirectAccept]);
+Io\Hooks\set_hooks($scheduler);
+
+function select_then_accept($server, int $timeout, bool $write = false): string
+{
+    $r = [$server];
+    $w = $write ? [$server] : null;
+    $e = null;
+    $start = hrtime(true);
+    $n = stream_select($r, $w, $e, $timeout);
+    $atOnce = (hrtime(true) - $start) < 500_000_000;
+    $conn = @stream_socket_accept($server, 0);
+    return sprintf("select: %d, at once: %s, accepted: %s",
+        $n, var_export($atOnce, true), var_export($conn !== false, true));
+}
+
+// A blocking accept arms the multishot; the two connections after it go to the ring's buffer,
+// and the usual loop of a select and a non-blocking accept takes them
+$server = stream_socket_server('tcp://127.0.0.1:0');
+$addr = 'tcp://' . stream_socket_get_name($server, false);
+$clients = [];
+$scheduler->spawn(function () use ($server, $addr, &$clients) {
+    $clients[] = stream_socket_client($addr);
+    var_dump(stream_socket_accept($server, 5) !== false);
+    $clients[] = stream_socket_client($addr);
+    $clients[] = stream_socket_client($addr);
+    usleep(50000);
+    stream_set_blocking($server, false);
+    echo select_then_accept($server, 2), "\n";
+    // A stream in both sets is one poll for reading and writing
+    echo select_then_accept($server, 2, true), "\n";
+});
+$scheduler->loop();
+
+// A select waiting while the multishot is armed sees the connection the multishot takes
+$scheduler->spawn(function () use ($server) {
+    echo select_then_accept($server, 2), "\n";
+});
+$scheduler->spawn(function () use ($addr, &$clients) {
+    $clients[] = stream_socket_client($addr);
+});
+$scheduler->loop();
+
+// A non-blocking accept with nothing pending arms no multishot: the next connection stays in the
+// kernel, where even a select with a zero timeout sees it
+$server = stream_socket_server('tcp://127.0.0.1:0');
+$addr = 'tcp://' . stream_socket_get_name($server, false);
+$scheduler->spawn(function () use ($server, $addr, &$clients) {
+    var_dump(@stream_socket_accept($server, 0));
+    $clients[] = stream_socket_client($addr);
+    usleep(50000);
+    stream_set_blocking($server, false);
+    echo select_then_accept($server, 0), "\n";
+});
+$scheduler->loop();
+
+Io\Hooks\set_hooks(null);
+?>
+--EXPECT--
+bool(true)
+select: 1, at once: true, accepted: true
+select: 1, at once: true, accepted: true
+select: 1, at once: true, accepted: true
+bool(false)
+select: 1, at once: true, accepted: true

main/io/php_io_ring.c

@@ -1174,6 +1174,21 @@ static int php_io_ring_reg_take(php_io_ring *ring, php_io_ring_reg *rec)
 	return fd;
 }
 
+/* The polls parked on a listener while its multishot accept takes the connections from the kernel:
+ * one buffered, or the multishot gone, makes the listener worth a look */
+static void php_io_ring_reg_wake_polls(php_io_ring *ring, php_io_ring_reg *rec)
+{
+	php_io_ring_req *w = ring->waiting;
+	while (w) {
+		php_io_ring_req *next = w->w_next;
+		if (w->waiting == rec && w->type == PHP_IO_OP_POLL) {
+			php_io_ring_waiter_settle(ring, w);
+			php_io_ring_req_main_cqe(ring, w, (int32_t) IOR_POLL_IN);
+		}
+		w = next;
+	}
+}
+
 /* A connection the multishot accepted: the oldest parked accept takes it, or it is buffered */
 static void php_io_ring_reg_connection(php_io_ring *ring, php_io_ring_reg *rec, int fd)
 {
@@ -1194,6 +1209,7 @@ static void php_io_ring_reg_connection(php_io_ring *ring, php_io_ring_reg *rec,
 		rec->fds = safe_erealloc(rec->fds, rec->fds_cap, sizeof(*rec->fds), 0);
 	}
 	rec->fds[rec->n_fds++] = fd;
+	php_io_ring_reg_wake_polls(ring, rec);
 	if (rec->accept && rec->n_fds >= PHP_IO_RING_ACCEPT_CAP) {
 		php_io_ring_reg_disarm_accept(ring, rec, true);
 	}
@@ -1215,6 +1231,16 @@ static bool php_io_ring_accept_wait(php_io_ring *ring, php_io_ring_req *req)
 		php_io_ring_accept_complete(ring, req, php_io_ring_reg_take(ring, rec));
 		return true;
 	}
+	if (!php_deadline_is_infinite(&req->deadline) && req->deadline.hrtime <= zend_hrtime()) {
+		/* A non-blocking accept arms no multishot, which would take the connections arriving after
+		 * it out of a select's sight: a direct one is left to the core's accept(), the wait after
+		 * it takes its own entry */
+		if (!(op->flags & PHP_IO_OP_F_AFTER_DRAIN)) {
+			php_io_ring_req_fail(ring, req, ENOTSUP);
+			return true;
+		}
+		return false;
+	}
 	if (!rec->accept && !php_io_ring_reg_arm_accept(ring, rec)) {
 		return false;
 	}
@@ -1230,6 +1256,19 @@ static bool php_io_ring_edge_wait(php_io_ring *ring, php_io_ring_req *req)
 	if (op->type == PHP_IO_OP_ACCEPT) {
 		return php_io_ring_accept_wait(ring, req);
 	}
+	if (op->type == PHP_IO_OP_POLL && (op->u.poll.events & PHP_POLL_READ)) {
+		/* A select's poll on a listener whose multishot accept takes the connections from the
+		 * kernel: answered from the buffer, or by the next connection the multishot takes */
+		php_io_ring_reg *listener = php_io_ring_reg_of(ring, op->registration, op->fd);
+		if (listener && listener->n_fds) {
+			php_io_ring_req_main_cqe(ring, req, (int32_t) IOR_POLL_IN);
+			return true;
+		}
+		if (listener && listener->accept) {
+			php_io_ring_waiter_park(ring, req, listener, IOR_POLL_IN);
+			return true;
+		}
+	}
 	if (op->type != PHP_IO_OP_POLL || !(op->flags & PHP_IO_OP_F_AFTER_DRAIN) || !op->registration
 			|| op->registration->trigger != PHP_IO_TRIGGER_EDGE) {
 		return false;
@@ -1735,6 +1774,7 @@ static void php_io_ring_multishot_cqe(php_io_ring *ring, php_io_ring_req *req, i
 				rec->accept = NULL;
 			}
 			req->reg = NULL;
+			php_io_ring_reg_wake_polls(ring, rec);
 			if (res < 0 && res != -ECANCELED) {
 				/* Refused (an old kernel, -EMFILE): the parked accepts get the error, later
 				 * ones take their own entry */

main/php_io_hooks.h

@@ -258,9 +258,11 @@ PHPAPI bool php_io_interrupt_pending(void);
 /* Without a provider it waits itself */
 PHPAPI zend_result php_io_run(php_io_op *op, php_io_op_result *result);
 
-/* The wrappers return like the syscall, with errno ETIMEDOUT or ECANCELED. stream may be NULL; a
- * stream is frozen for the call. On Windows php_socket_errno() is the Winsock code, see
- * PHP_IO_SOCK_*. */
+/* The wrappers return like the syscall, with errno ETIMEDOUT or ECANCELED. ECANCELED with
+ * EG(exception) set means the op was abandoned, or never started, because of that exception, and
+ * the exception reports it; without one, the provider completed the op as Cancelled, a failure like
+ * any other errno. stream may be NULL; a stream is frozen for the call. On Windows
+ * php_socket_errno() is the Winsock code, see PHP_IO_SOCK_*. */
 #ifdef PHP_WIN32
 # define PHP_IO_SOCK_ETIMEDOUT WSAETIMEDOUT
 # define PHP_IO_SOCK_ECANCELED WSAECANCELLED

main/streams/plain_wrapper.c

@@ -583,6 +583,10 @@ static ssize_t php_stdiop_write(php_stream *stream, const char *buf, size_t coun
 				/* TODO: Should this be treated as a proper error or not? */
 				return bytes_written;
 			}
+			if (errno == ECANCELED && EG(exception)) {
+				/* ECANCELED under a pending exception (php_io_hooks.h): the exception reports the failure */
+				return bytes_written;
+			}
 			if (!(stream->flags & PHP_STREAM_FLAG_SUPPRESS_ERRORS)) {
 				char errstr[256];
 				php_stream_notice(stream, WriteFailed,
@@ -675,6 +679,8 @@ static ssize_t php_stdiop_read(php_stream *stream, char *buf, size_t count)
 				ret = 0;
 			} else if (errno == EINTR) {
 				/* TODO: Should this be treated as a proper error or not? */
+			} else if (errno == ECANCELED && EG(exception)) {
+				/* ECANCELED under a pending exception (php_io_hooks.h): no notice, the stream is not at its end */
 			} else {
 				if (!(stream->flags & PHP_STREAM_FLAG_SUPPRESS_ERRORS)) {
 					char errstr[256];

main/streams/xp_socket.c

@@ -154,7 +154,8 @@ static ssize_t php_sockop_read(php_stream *stream, char *buf, size_t count)
 	if (nr_bytes < 0) {
 		if (PHP_IS_TRANSIENT_ERROR(err)) {
 			nr_bytes = 0;
-		} else if (err != PHP_IO_SOCK_EINTR) {
+		} else if (err != PHP_IO_SOCK_EINTR && !(err == PHP_IO_SOCK_ECANCELED && EG(exception))) {
+			/* ECANCELED under a pending exception (php_io_hooks.h) does not end the stream */
 			stream->eof = 1;
 		}
 	} else if (nr_bytes == 0) {