From 02db2f5fd69c9ffe9924bb1b1ba96aad610bee84 Mon Sep 17 00:00:00 2001
From: MARiASoCute <1056159381@qq.com>
Date: Fri, 9 Oct 2026 14:59:15 +0800
Subject: [PATCH 1/5] Reads and writes with a zero timeout need to be
 non-blocking.

---
 ext-src/swoole_runtime.cc                    | 64 ++++++++++-----
 tests/swoole_runtime/zero_timeout_read.phpt  | 74 +++++++++++++++++
 tests/swoole_runtime/zero_timeout_write.phpt | 83 ++++++++++++++++++++
 3 files changed, 201 insertions(+), 20 deletions(-)
 create mode 100644 tests/swoole_runtime/zero_timeout_read.phpt
 create mode 100644 tests/swoole_runtime/zero_timeout_write.phpt

diff --git a/ext-src/swoole_runtime.cc b/ext-src/swoole_runtime.cc
index 88977d87ca..e3370b5d96 100644
--- a/ext-src/swoole_runtime.cc
+++ b/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,32 @@ 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);
+        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;
             if (sock->get_socket()->catch_write_error(sock->errCode) == SW_WAIT) {
                 didwrite = 0;
+                if (abstract->blocking) {
+                    report_errors = true;
+                    if (abstract->zero_timer) {
+                        abstract->stream.timeout_event = 1;
+                        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 +548,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 +560,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);
@@ -560,11 +577,13 @@ static php_stream_size_t socket_read(php_stream *stream, char *buf, size_t count
         php_stream_notify_progress_increment(PHP_STREAM_CONTEXT(stream), nr_bytes, 0);
     }
 
-    if (nr_bytes < 0 && sock->errCode != ETIMEDOUT) {
-        if (sock->get_socket()->catch_read_error(sock->errCode) == SW_WAIT) {
-            nr_bytes = 0;
-        } else {
-            stream->eof = 1;
+    if (nr_bytes < 0) {
+        if (sock->errCode != ETIMEDOUT) {
+            if (sock->get_socket()->catch_read_error(sock->errCode) == SW_WAIT) {
+                nr_bytes = 0;
+            } else {
+                stream->eof = 1;
+            }
         }
     } else if (nr_bytes == 0) {
         stream->eof = 1;
@@ -1128,15 +1147,20 @@ static int socket_set_option(php_stream *stream, int option, int value, void *pt
             add_assoc_string(&tmp, "cipher_version", SSL_CIPHER_get_version(cipher));
             add_assoc_zval((zval *) ptrparam, "crypto", &tmp);
         }
-        add_assoc_bool((zval *) ptrparam, "timed_out", sock->errCode == ETIMEDOUT);
+        add_assoc_bool((zval *) ptrparam, "timed_out", sock->errCode == ETIMEDOUT || abstract->stream.timeout_event);
         add_assoc_bool((zval *) ptrparam, "eof", stream->eof);
         add_assoc_bool((zval *) ptrparam, "blocked", true);
         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: {
@@ -1299,7 +1323,7 @@ static php_stream *socket_create(const char *proto,
 
     auto abstract = new NetStream();
     abstract->socket.reset(sock);
-    abstract->stream.socket = (int)sock->get_fd();
+    abstract->stream.socket = (int) sock->get_fd();
     abstract->blocking = true;
 
     stream = php_stream_alloc_rel(&socket_ops, abstract, persistent_id, "r+");
@@ -1601,7 +1625,7 @@ static void hook_all_func(uint32_t flags) {
             SW_UNHOOK_FUNC(proc_terminate);
         }
     }
-#endif // _WIN32
+#endif  // _WIN32
     // ext-sockets
     if (flags & PHPCoroutine::HOOK_SOCKETS) {
         if (!(runtime_hook_flags & PHPCoroutine::HOOK_SOCKETS)) {
@@ -1953,7 +1977,7 @@ static PHP_FUNCTION(swoole_time_nanosleep) {
 
 #ifdef _WIN32
         // Windows: use Sleep for millisecond resolution
-        Sleep((DWORD)(tv_sec * 1000 + tv_nsec / 1000000));
+        Sleep((DWORD) (tv_sec * 1000 + tv_nsec / 1000000));
         RETURN_TRUE;
 #else
         if (nanosleep(&php_req, &php_rem) == 0) {
@@ -2018,12 +2042,12 @@ static void stream_array_to_fd_set(zval *stream_array, std::unordered_map<swSock
     ZEND_HASH_FOREACH_KEY_VAL(Z_ARRVAL_P(stream_array), index, key, elem) {
         ZVAL_DEREF(elem);
         php_socket_t sock = php_swoole_convert_to_fd(elem);
-        if (sock < (php_socket_t)0) {
+        if (sock < (php_socket_t) 0) {
             continue;
         }
-        auto i = fds.find((swSocketFd)sock);
+        auto i = fds.find((swSocketFd) sock);
         if (i == fds.end()) {
-            fds.emplace((swSocketFd)sock, PollSocket(event, new zend::KeyValue(index, key, elem)));
+            fds.emplace((swSocketFd) sock, PollSocket(event, new zend::KeyValue(index, key, elem)));
         } else {
             i->second.events |= event;
         }
@@ -2265,7 +2289,7 @@ php_stream *php_swoole_create_stream_from_socket(swSocketFd _fd, int domain, int
         abstract->socket->set_timeout((double) FG(default_socket_timeout));
     }
     abstract->stream.timeout.tv_sec = FG(default_socket_timeout);
-    abstract->stream.socket = (int)abstract->socket->get_fd();
+    abstract->stream.socket = (int) abstract->socket->get_fd();
     abstract->blocking = true;
 
     php_stream *stream = php_stream_alloc_rel(&socket_ops, abstract, nullptr, "r+");
diff --git a/tests/swoole_runtime/zero_timeout_read.phpt b/tests/swoole_runtime/zero_timeout_read.phpt
new file mode 100644
index 0000000000..8edc708b27
--- /dev/null
+++ b/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
diff --git a/tests/swoole_runtime/zero_timeout_write.phpt b/tests/swoole_runtime/zero_timeout_write.phpt
new file mode 100644
index 0000000000..47d18069a2
--- /dev/null
+++ b/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

From 5471d6f1c64614de432bb360c8bd314b176f3c6b Mon Sep 17 00:00:00 2001
From: NathanFreeman <1056159381@qq.com>
Date: Thu, 15 Oct 2026 22:42:38 +0800
Subject: [PATCH 2/5] restore formatting

---
 ext-src/swoole_runtime.cc | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/ext-src/swoole_runtime.cc b/ext-src/swoole_runtime.cc
index 58df725e33..a412350ea4 100644
--- a/ext-src/swoole_runtime.cc
+++ b/ext-src/swoole_runtime.cc
@@ -1323,7 +1323,7 @@ static php_stream *socket_create(const char *proto,
 
     auto abstract = new NetStream();
     abstract->socket.reset(sock);
-    abstract->stream.socket = (int) sock->get_fd();
+    abstract->stream.socket = (int)sock->get_fd();
     abstract->blocking = true;
 
     stream = php_stream_alloc_rel(&socket_ops, abstract, persistent_id, "r+");
@@ -1640,7 +1640,7 @@ static void hook_all_func(uint32_t flags) {
             SW_UNHOOK_FUNC(proc_terminate);
         }
     }
-#endif  // _WIN32
+#endif // _WIN32
     // ext-sockets
     if (flags & PHPCoroutine::HOOK_SOCKETS) {
         if (!(runtime_hook_flags & PHPCoroutine::HOOK_SOCKETS)) {
@@ -1994,7 +1994,7 @@ static PHP_FUNCTION(swoole_time_nanosleep) {
 
 #ifdef _WIN32
         // Windows: use Sleep for millisecond resolution
-        Sleep((DWORD) (tv_sec * 1000 + tv_nsec / 1000000));
+        Sleep((DWORD)(tv_sec * 1000 + tv_nsec / 1000000));
         RETURN_TRUE;
 #else
         if (nanosleep(&php_req, &php_rem) == 0) {
@@ -2059,12 +2059,12 @@ static void stream_array_to_fd_set(zval *stream_array, std::unordered_map<swSock
     ZEND_HASH_FOREACH_KEY_VAL(Z_ARRVAL_P(stream_array), index, key, elem) {
         ZVAL_DEREF(elem);
         php_socket_t sock = php_swoole_convert_to_fd(elem);
-        if (sock < (php_socket_t) 0) {
+        if (sock < (php_socket_t)0) {
             continue;
         }
-        auto i = fds.find((swSocketFd) sock);
+        auto i = fds.find((swSocketFd)sock);
         if (i == fds.end()) {
-            fds.emplace((swSocketFd) sock, PollSocket(event, new zend::KeyValue(index, key, elem)));
+            fds.emplace((swSocketFd)sock, PollSocket(event, new zend::KeyValue(index, key, elem)));
         } else {
             i->second.events |= event;
         }

From 85196d99bb27d9387c65c1e44cc4b7d1dcbf9c22 Mon Sep 17 00:00:00 2001
From: NathanFreeman <1056159381@qq.com>
Date: Thu, 15 Oct 2026 22:44:32 +0800
Subject: [PATCH 3/5] optimize code

---
 ext-src/swoole_runtime.cc | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ext-src/swoole_runtime.cc b/ext-src/swoole_runtime.cc
index a412350ea4..df2ba0d0b1 100644
--- a/ext-src/swoole_runtime.cc
+++ b/ext-src/swoole_runtime.cc
@@ -1147,7 +1147,7 @@ static int socket_set_option(php_stream *stream, int option, int value, void *pt
             add_assoc_string(&tmp, "cipher_version", SSL_CIPHER_get_version(cipher));
             add_assoc_zval((zval *) ptrparam, "crypto", &tmp);
         }
-        add_assoc_bool((zval *) ptrparam, "timed_out", sock->errCode == ETIMEDOUT || abstract->stream.timeout_event);
+        add_assoc_bool((zval *) ptrparam, "timed_out", sock->errCode == ETIMEDOUT || abstract->stream.timeout_event == 1);
         add_assoc_bool((zval *) ptrparam, "eof", stream->eof);
         add_assoc_bool((zval *) ptrparam, "blocked", true);
         break;

From 1e2c95a5fe42de42f8847624b09a503a8d663f39 Mon Sep 17 00:00:00 2001
From: NathanFreeman <1056159381@qq.com>
Date: Thu, 15 Oct 2026 22:47:50 +0800
Subject: [PATCH 4/5] optimize code

---
 ext-src/swoole_runtime.cc | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/ext-src/swoole_runtime.cc b/ext-src/swoole_runtime.cc
index df2ba0d0b1..3421b95b25 100644
--- a/ext-src/swoole_runtime.cc
+++ b/ext-src/swoole_runtime.cc
@@ -577,13 +577,11 @@ static php_stream_size_t socket_read(php_stream *stream, char *buf, size_t count
         php_stream_notify_progress_increment(PHP_STREAM_CONTEXT(stream), nr_bytes, 0);
     }
 
-    if (nr_bytes < 0) {
-        if (sock->errCode != ETIMEDOUT) {
-            if (sock->get_socket()->catch_read_error(sock->errCode) == SW_WAIT) {
-                nr_bytes = 0;
-            } else {
-                stream->eof = 1;
-            }
+    if (nr_bytes < 0 && sock->errCode != ETIMEDOUT) {
+        if (sock->get_socket()->catch_read_error(sock->errCode) == SW_WAIT) {
+            nr_bytes = 0;
+        } else {
+            stream->eof = 1;
         }
     } else if (nr_bytes == 0) {
         stream->eof = 1;

From e1b9c37ad8ab62c09afbcbeb9e7b7c89361317b8 Mon Sep 17 00:00:00 2001
From: NathanFreeman <1056159381@qq.com>
Date: Sat, 17 Oct 2026 01:24:05 +0800
Subject: [PATCH 5/5] optimize code

---
 ext-src/swoole_runtime.cc                     |  18 +--
 .../stream_timed_out_reset.phpt               | 104 ++++++++++++++++++
 2 files changed, 114 insertions(+), 8 deletions(-)
 create mode 100644 tests/swoole_runtime/stream_timed_out_reset.phpt

diff --git a/ext-src/swoole_runtime.cc b/ext-src/swoole_runtime.cc
index 3421b95b25..aa63dd5277 100644
--- a/ext-src/swoole_runtime.cc
+++ b/ext-src/swoole_runtime.cc
@@ -505,19 +505,21 @@ 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) {
+                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;
-                    if (abstract->zero_timer) {
-                        abstract->stream.timeout_event = 1;
-                        didwrite = -1;
-                    }
+                    sock->set_err(ETIMEDOUT);
+                    didwrite = -1;
                 }
             } else {
                 report_errors = true;
@@ -1145,7 +1147,7 @@ static int socket_set_option(php_stream *stream, int option, int value, void *pt
             add_assoc_string(&tmp, "cipher_version", SSL_CIPHER_get_version(cipher));
             add_assoc_zval((zval *) ptrparam, "crypto", &tmp);
         }
-        add_assoc_bool((zval *) ptrparam, "timed_out", sock->errCode == ETIMEDOUT || abstract->stream.timeout_event == 1);
+        add_assoc_bool((zval *) ptrparam, "timed_out", sock->errCode == ETIMEDOUT);
         add_assoc_bool((zval *) ptrparam, "eof", stream->eof);
         add_assoc_bool((zval *) ptrparam, "blocked", true);
         break;
diff --git a/tests/swoole_runtime/stream_timed_out_reset.phpt b/tests/swoole_runtime/stream_timed_out_reset.phpt
new file mode 100644
index 0000000000..3906d1ab04
--- /dev/null
+++ b/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