Reads and writes with a zero timeout need to be non-blocking.

ext-src/swoole_runtime.cc

@@ -133,6 +133,7 @@ struct NetStream {
     php_netstream_data_t stream;
     std::shared_ptr<SocketImpl> socket;
     bool blocking;
+    bool zero_timer = false;
 };
 
 static struct {
@@ -495,7 +496,7 @@ static php_stream_size_t socket_write(php_stream *stream, const char *buf, size_
         return sock->get_socket()->send_sync(buf, count, 0);
     }
 
-    if (abstract->blocking) {
+    if (abstract->blocking && !abstract->zero_timer) {
         didwrite = sock->send_all(buf, count);
     } else {
         didwrite = sock->send_once(buf, count);
@@ -504,19 +505,34 @@ static php_stream_size_t socket_write(php_stream *stream, const char *buf, size_
 
     // Preserve positive partial counts so PHP can retry the unwritten remainder.
     if (didwrite < 0) {
-        abstract->stream.timeout_event = (sock->errCode == ETIMEDOUT);
-
         if (sock->errCode != ETIMEDOUT) {
+            // SW_WAIT is not an error for a non-blocking stream.
+            bool report_errors = false;
+
+            // Only non-blocking mode, or blocking mode with a zero timeout, can possibly take this branch.
             if (sock->get_socket()->catch_write_error(sock->errCode) == SW_WAIT) {
                 didwrite = 0;
+                if (abstract->blocking && abstract->zero_timer) {
+                    /**
+                     * Sending after `send` returns `EAGAIN` and immediately using a zero-timeout `poll`
+                     * to check writability is unreasonable, because the buffer state will not change in an instant.
+                     */
+                    report_errors = true;
+                    sock->set_err(ETIMEDOUT);
+                    didwrite = -1;
+                }
             } else {
+                report_errors = true;
+                stream->eof = 1;
+            }
+
+            if (report_errors && !(stream->flags & PHP_STREAM_FLAG_SUPPRESS_ERRORS)) {
                 php_error_docref(nullptr,
                                  E_NOTICE,
                                  "Send of " ZEND_LONG_FMT " bytes failed with errno=%d %s",
                                  (zend_long) count,
                                  sock->errCode,
                                  sock->errMsg);
-                stream->eof = 1;
             }
         }
     } else {
@@ -534,6 +550,7 @@ static php_stream_size_t socket_write(php_stream *stream, const char *buf, size_
 static php_stream_size_t socket_read(php_stream *stream, char *buf, size_t count) {
     std::shared_ptr<SocketImpl> sock;
     ssize_t nr_bytes = -1;
+    bool dont_wait = false;
 
     auto *abstract = static_cast<NetStream *>(stream->abstract);
     if (UNEXPECTED(!abstract || !abstract->socket)) {
@@ -545,11 +562,13 @@ static php_stream_size_t socket_read(php_stream *stream, char *buf, size_t count
         return sock->get_socket()->recv_sync(buf, count, 0);
     }
 
-    if (abstract->blocking
 #if PHP_VERSION_ID >= 80300
-        && !stream->has_buffered_data
+    dont_wait = stream->has_buffered_data || abstract->zero_timer;
+#else
+    dont_wait = abstract->zero_timer;
 #endif
-    ) {
+
+    if (abstract->blocking && !dont_wait) {
         nr_bytes = sock->recv(buf, count);
     } else {
         nr_bytes = sock->recv_once(buf, count);
@@ -1134,9 +1153,14 @@ static int socket_set_option(php_stream *stream, int option, int value, void *pt
         break;
     }
     case PHP_STREAM_OPTION_READ_TIMEOUT: {
+        auto timeout = static_cast<timeval *>(ptrparam);
+
+        // When the timeout is 0, all reads and writes are fully asynchronous.
+        abstract->zero_timer = timeout->tv_sec == 0 && timeout->tv_usec == 0;
+
         // Despite its name, PHP socket streams use this option as the shared timeout for both blocking reads and
         // writes (see php_sockop_read() and php_sockop_write()).
-        abstract->socket->set_timeout(static_cast<timeval *>(ptrparam), SW_TIMEOUT_RDWR);
+        abstract->socket->set_timeout(timeout, SW_TIMEOUT_RDWR);
         break;
     }
     case PHP_STREAM_OPTION_CRYPTO_API: {

tests/swoole_runtime/stream_timed_out_reset.phpt

@@ -0,0 +1,104 @@
+--TEST--
+swoole_runtime: stream_get_meta_data timed_out is reset by a later successful operation
+--SKIPIF--
+<?php
+require __DIR__ . '/../include/skipif.inc';
+?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+use Swoole\Runtime;
+use Swoole\Coroutine\Channel;
+
+use function Swoole\Coroutine\run;
+
+Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
+
+run(function () {
+    $port = get_one_free_port();
+
+    $ready = new Channel(1);
+    $ask = new Channel(1);
+    $sent = new Channel(1);
+    $release = new Channel(1);
+
+    // Server side: never reads, so the client send buffer fills up. It sends one byte only when the
+    // test body asks for it and acknowledges the queued byte, so the recovery read finds data ready.
+    go(function () use ($port, &$server, &$connection, $ready, $ask, $sent, $release) {
+        $server = stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr);
+        Assert::true($server !== false, "server create failed: {$errstr}");
+
+        $connection = stream_socket_accept($server, 5);
+        Assert::true($connection !== false, "accept failed");
+        $ready->push(true);
+
+        foreach (['r', 's'] as $byte) {
+            $ask->pop(30);
+            Assert::same(fwrite($connection, $byte), 1);
+            $sent->push(true);
+        }
+
+        // Wait for the test body to finish instead of sleeping for a fixed duration.
+        $release->pop(30);
+    });
+
+    $stream = stream_socket_client("tcp://127.0.0.1:{$port}", $errno, $errstr, 5);
+    Assert::true($stream !== false, "connect failed: {$errstr}");
+    Assert::true($ready->pop(30) !== false, "the server did not accept the connection");
+
+    // Fill the send buffer, then let a blocking zero-timeout write fail: this is the timeout to recover from.
+    stream_set_blocking($stream, false);
+    $payload = str_repeat('A', 8192);
+    $deadline = microtime(true) + 10;
+    do {
+        $written = @fwrite($stream, $payload);
+    } while ($written > 0 && microtime(true) < $deadline);
+    Assert::true(microtime(true) < $deadline, "failed to fill the send buffer");
+
+    stream_set_blocking($stream, true);
+    stream_set_timeout($stream, 0, 0);
+    Assert::same(@fwrite($stream, 'x'), false);
+
+    $meta = stream_get_meta_data($stream);
+    Assert::true($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    // A successful blocking read must clear the timeout state left by the failed write.
+    stream_set_timeout($stream, 0, 200000);
+    $ask->push(true);
+    Assert::true($sent->pop(30) !== false, "the server did not send 'r'");
+    Assert::same(fread($stream, 1), 'r');
+
+    $meta = stream_get_meta_data($stream);
+    Assert::false($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    // A read that does time out reports it again, still without setting eof.
+    $start = microtime(true);
+    Assert::false(fread($stream, 1));
+    $elapsed = microtime(true) - $start;
+    Assert::greaterThanEq($elapsed, 0.1, "read should wait for the timeout, took {$elapsed}s");
+    Assert::lessThan($elapsed, 0.5, "read timeout should be 200ms, took {$elapsed}s");
+
+    $meta = stream_get_meta_data($stream);
+    Assert::true($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    // and the next successful read clears it once more.
+    $ask->push(true);
+    Assert::true($sent->pop(30) !== false, "the server did not send 's'");
+    Assert::same(fread($stream, 1), 's');
+
+    $meta = stream_get_meta_data($stream);
+    Assert::false($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    fclose($stream);
+    $release->push(true);
+});
+
+echo "OK\n";
+?>
+--EXPECT--
+OK

tests/swoole_runtime/zero_timeout_read.phpt

@@ -0,0 +1,74 @@
+--TEST--
+swoole_runtime: zero timeout stream read without waiting
+--SKIPIF--
+<?php
+require __DIR__ . '/../include/skipif.inc';
+?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+use Swoole\Runtime;
+use Swoole\Coroutine;
+use Swoole\Coroutine\Channel;
+
+use function Swoole\Coroutine\run;
+
+Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
+
+run(function () {
+    $port = get_one_free_port();
+
+    $chan = new Channel(1);
+
+    $server = null;
+    $connection = null;
+
+    go(function () use ($port, &$server, &$connection, $chan) {
+        $server = stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr);
+        Assert::true($server !== false, "server create failed: {$errstr}");
+
+        $connection = stream_socket_accept($server, 5);
+        Assert::true($connection !== false, "accept failed");
+
+        // Send data only after the client has done its zero-timeout read.
+        Assert::true($chan->pop(30) !== false, "server was not released");
+        fwrite($connection, "line\nrecord1\n");
+    });
+
+    $stream = stream_socket_client("tcp://127.0.0.1:{$port}", $errno, $errstr, 5);
+    Assert::true($stream !== false, "connect failed: {$errstr}");
+
+    stream_set_timeout($stream, 0, 0);
+
+    $start = microtime(true);
+    Assert::same(fread($stream, 8192), "");
+    $elapsed = microtime(true) - $start;
+    Assert::true($elapsed < 0.1, "zero-timeout read should return immediately, took {$elapsed}s");
+
+    $meta = stream_get_meta_data($stream);
+    Assert::false($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    // Release the server and poll with a bounded deadline so a regression cannot hang the test.
+    $chan->push(true);
+
+    $data = "";
+    $deadline = microtime(true) + 5;
+    while ($data === "" && microtime(true) < $deadline) {
+        $data = fread($stream, 8192);
+        if ($data === "") {
+            Coroutine::sleep(0.05);
+        }
+    }
+    Assert::same($data, "line\nrecord1\n");
+
+    fclose($stream);
+    fclose($connection);
+    fclose($server);
+});
+
+echo "OK\n";
+?>
+--EXPECT--
+OK

tests/swoole_runtime/zero_timeout_write.phpt

@@ -0,0 +1,83 @@
+--TEST--
+swoole_runtime: zero timeout stream write without waiting
+--SKIPIF--
+<?php
+require __DIR__ . '/../include/skipif.inc';
+?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+use Swoole\Runtime;
+use Swoole\Coroutine\Channel;
+
+use function Swoole\Coroutine\run;
+
+Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
+
+run(function () {
+    $port = get_one_free_port();
+
+    $chan = new Channel(1);
+
+    $server = null;
+    $connection = null;
+
+    // Server side: only accept, never read, so the client send buffer fills up.
+    go(function () use ($port, &$server, &$connection, $chan) {
+        $server = stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr);
+        Assert::true($server !== false, "server create failed: {$errstr}");
+
+        $connection = stream_socket_accept($server, 5);
+        Assert::true($connection !== false, "accept failed");
+
+        // Wait for the test body to finish instead of sleeping for a fixed duration.
+        $chan->pop(30);
+    });
+
+    $stream = stream_socket_client("tcp://127.0.0.1:{$port}", $errno, $errstr, 5);
+    Assert::true($stream !== false, "connect failed: {$errstr}");
+
+    // Fill the send buffer in non-blocking mode until no further bytes are accepted.
+    stream_set_blocking($stream, false);
+    $payload = str_repeat('A', 8192);
+    $deadline = microtime(true) + 10;
+    do {
+        $written = @fwrite($stream, $payload);
+    } while ($written > 0 && microtime(true) < $deadline);
+    Assert::true(microtime(true) < $deadline, "failed to fill the send buffer");
+
+    // Explicitly non-blocking stream: a full send buffer is reported as a zero-byte write,
+    // without setting timed_out or eof (same as native PHP).
+    $start = microtime(true);
+    $written = @fwrite($stream, 'x');
+    Assert::same($written, 0);
+    Assert::true(microtime(true) - $start < 0.1, "non-blocking write should return immediately");
+
+    $meta = stream_get_meta_data($stream);
+    Assert::false($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    // Blocking stream with zero timeout: native PHP fails the write and sets timed_out
+    // when the send buffer is full, instead of reporting a zero-byte write.
+    stream_set_blocking($stream, true);
+    stream_set_timeout($stream, 0, 0);
+
+    $start = microtime(true);
+    $written = @fwrite($stream, 'x');
+    $elapsed = microtime(true) - $start;
+    Assert::same($written, false);
+    Assert::true($elapsed < 0.1, "zero-timeout write should return immediately, took {$elapsed}s");
+
+    $meta = stream_get_meta_data($stream);
+    Assert::true($meta['timed_out']);
+    Assert::false($meta['eof']);
+
+    fclose($stream);
+    $chan->push(true);
+});
+
+echo "OK\n";
+?>
+--EXPECT--
+OK