From 65796fee91090d26393b5d1b23bf6e72d28640e9 Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Mon, 7 Sep 2026 15:58:31 +0300
Subject: [PATCH 1/5] Cover concurrent reads from one shared file handle

Two coroutines reading the same handle lose and duplicate data: both derive
the same buffer address from a state neither has finished updating, so 1024
of 2000 distinct records survive. The test states the contract rather than
the counts: every byte reaches exactly one reader.

Needs the buffer lock from true-async/php-src#32.
---
 .../085-shared_handle_concurrent_reads.phpt   | 66 +++++++++++++++++++
 1 file changed, 66 insertions(+)
 create mode 100644 tests/io/085-shared_handle_concurrent_reads.phpt

diff --git a/tests/io/085-shared_handle_concurrent_reads.phpt b/tests/io/085-shared_handle_concurrent_reads.phpt
new file mode 100644
index 0000000..fa5b1dd
--- /dev/null
+++ b/tests/io/085-shared_handle_concurrent_reads.phpt
@@ -0,0 +1,66 @@
+--TEST--
+Concurrent reads from one shared file handle deliver every byte exactly once
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+
+echo "Start\n";
+
+/* Records of a fixed width, each holding its own index, so that a record read
+ * twice or lost is visible in the counts below. */
+const RECORDS      = 2000;
+const RECORD_SIZE  = 16;
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+$fp = fopen($tmpfile, 'w');
+for ($i = 0; $i < RECORDS; $i++) {
+    fwrite($fp, sprintf("%015d\n", $i));
+}
+fclose($fp);
+
+$handle = fopen($tmpfile, 'r');
+
+$reader = function () use ($handle) {
+    $data = '';
+    while (!feof($handle)) {
+        $chunk = fread($handle, 4096);
+        if ($chunk === false) {
+            return false;
+        }
+        $data .= $chunk;
+    }
+    return $data;
+};
+
+[$results, $exceptions] = await_all([spawn($reader), spawn($reader)]);
+
+$records = [];
+$bytes   = 0;
+foreach ($results as $data) {
+    if ($data === false) {
+        echo "fread() failed\n";
+        continue;
+    }
+    $bytes += strlen($data);
+    foreach (str_split($data, RECORD_SIZE) as $record) {
+        $records[] = $record;
+    }
+}
+
+printf("bytes: %d of %d\n", $bytes, RECORDS * RECORD_SIZE);
+printf("records: %d, unique: %d\n", count($records), count(array_unique($records)));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($handle);
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+bytes: 32000 of 32000
+records: 2000, unique: 2000
+Exceptions: 0
+End

From 0df1b1212753506c3a4cafe1d6a4ea969cec46b4 Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Mon, 7 Sep 2026 23:42:20 +0300
Subject: [PATCH 2/5] Cover a close that lands while another coroutine is in
 the IO

Closing a handle from a second coroutine frees the stream inside the first
one's ops->read or ops->write, and the buffered layer then reached the
freed stream through its lock and through the write path's eof check. The
test reads and writes under such a close; on the previous php-src code it
segfaults under a sanitizer build.

Needs true-async/php-src#32.
---
 tests/io/086-close_during_io.phpt | 71 +++++++++++++++++++++++++++++++
 1 file changed, 71 insertions(+)
 create mode 100644 tests/io/086-close_during_io.phpt

diff --git a/tests/io/086-close_during_io.phpt b/tests/io/086-close_during_io.phpt
new file mode 100644
index 0000000..4b42a12
--- /dev/null
+++ b/tests/io/086-close_during_io.phpt
@@ -0,0 +1,71 @@
+--TEST--
+Closing a handle from another coroutine while it is being read and written
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+const ROUNDS = 20;
+
+$source = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($source, str_repeat("0123456789abcdef", 200000));
+
+$target = tempnam(sys_get_temp_dir(), 'async_io_test_');
+
+/* The closing coroutine runs while the other one is parked inside the read or
+ * the write, which is where the stream is freed. */
+$closeAfterTwoSwitches = function ($handle) {
+    suspend();
+    suspend();
+    @fclose($handle);
+    return true;
+};
+
+for ($round = 0; $round < ROUNDS; $round++) {
+    $handle = fopen($source, 'r');
+    $reader = spawn(function () use ($handle) {
+        $total = 0;
+        while (true) {
+            $chunk = @fread($handle, 4096);
+            if ($chunk === false || $chunk === '') {
+                break;
+            }
+            $total += strlen($chunk);
+        }
+        return $total;
+    });
+    await_all([$reader, spawn(fn() => $closeAfterTwoSwitches($handle))]);
+}
+
+echo "Read survived\n";
+
+for ($round = 0; $round < ROUNDS; $round++) {
+    $handle = fopen($target, 'w');
+    $writer = spawn(function () use ($handle) {
+        $chunk = str_repeat('x', 200000);
+        for ($i = 0; $i < 20; $i++) {
+            if (@fwrite($handle, $chunk) === false) {
+                break;
+            }
+        }
+        return true;
+    });
+    await_all([$writer, spawn(fn() => $closeAfterTwoSwitches($handle))]);
+}
+
+echo "Write survived\n";
+
+unlink($source);
+unlink($target);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+Read survived
+Write survived
+End

From ea8a7b9fa2e590736054507518545aad3d2eecde Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Tue, 8 Sep 2026 11:49:29 +0300
Subject: [PATCH 3/5] Cover the stream paths that reached around the buffer
 lock

Six coroutine tests over one shared handle: a read filter appended and removed
while another coroutine is inside fread(), a cast to a file descriptor during a
read, and a close landing while a copy, a stream_get_contents() or a write
filter flush is parked. Each one loses data, hangs or crashes without the lock
those paths now take.
---
 tests/io/087-filter_append_during_read.phpt  | 48 +++++++++++++
 tests/io/088-cast_during_read.phpt           | 71 ++++++++++++++++++
 tests/io/089-close_during_copy.phpt          | 48 +++++++++++++
 tests/io/090-close_during_get_contents.phpt  | 52 ++++++++++++++
 tests/io/091-filter_remove_during_read.phpt  | 76 ++++++++++++++++++++
 tests/io/092-filter_remove_during_write.phpt | 68 ++++++++++++++++++
 tests/io/093-filter_append_during_close.phpt | 46 ++++++++++++
 7 files changed, 409 insertions(+)
 create mode 100644 tests/io/087-filter_append_during_read.phpt
 create mode 100644 tests/io/088-cast_during_read.phpt
 create mode 100644 tests/io/089-close_during_copy.phpt
 create mode 100644 tests/io/090-close_during_get_contents.phpt
 create mode 100644 tests/io/091-filter_remove_during_read.phpt
 create mode 100644 tests/io/092-filter_remove_during_write.phpt
 create mode 100644 tests/io/093-filter_append_during_close.phpt

diff --git a/tests/io/087-filter_append_during_read.phpt b/tests/io/087-filter_append_during_read.phpt
new file mode 100644
index 0000000..9cb264a
--- /dev/null
+++ b/tests/io/087-filter_append_during_read.phpt
@@ -0,0 +1,48 @@
+--TEST--
+Appending a read filter while another coroutine is inside fread()
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+/* The first read leaves 8182 bytes buffered, so the filter append has
+ * pre-buffered data to wind through the chain, and the second read is parked
+ * inside ops->read while it happens. */
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+$body = '';
+for ($i = 0; $i < 4000; $i++) {
+    $body .= sprintf("%015d\n", $i);
+}
+file_put_contents($tmpfile, $body);
+
+$handle = fopen($tmpfile, 'r');
+fread($handle, 10);
+
+$reader = spawn(fn() => @fread($handle, 20000));
+$appender = spawn(function () use ($handle) {
+    suspend();
+    return @stream_filter_append($handle, 'convert.base64-encode', STREAM_FILTER_READ) !== false;
+});
+
+[$results, $exceptions] = await_all([$reader, $appender]);
+
+$data = (string) $results[0];
+printf("length: %d\n", strlen($data));
+printf("raw: %s\n", var_export($data === substr($body, 10, strlen($data)), true));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($handle);
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+length: 20000
+raw: true
+Exceptions: 0
+End
diff --git a/tests/io/088-cast_during_read.phpt b/tests/io/088-cast_during_read.phpt
new file mode 100644
index 0000000..bea4bf4
--- /dev/null
+++ b/tests/io/088-cast_during_read.phpt
@@ -0,0 +1,71 @@
+--TEST--
+Casting a handle to a file descriptor while another coroutine is inside fread()
+--SKIPIF--
+<?php
+if (!function_exists('proc_open')) die('skip proc_open() is disabled');
+if (PHP_OS_FAMILY === 'Windows') die('skip POSIX only');
+?>
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+/* proc_open() casts the handle with PHP_STREAM_AS_FD, which flushes it, seeks
+ * the descriptor and drops the read buffer under the parked reader. */
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+$body = '';
+for ($i = 0; $i < 20000; $i++) {
+    $body .= sprintf("%015d\n", $i);
+}
+file_put_contents($tmpfile, $body);
+
+$handle = fopen($tmpfile, 'r');
+
+$reader = spawn(function () use ($handle) {
+    $data = '';
+    while (true) {
+        $chunk = @fread($handle, 4096);
+        if ($chunk === false || $chunk === '') {
+            break;
+        }
+        $data .= $chunk;
+    }
+    return $data;
+});
+
+$caster = spawn(function () use ($handle) {
+    suspend();
+    $descriptors = [0 => $handle, 1 => ['pipe', 'w'], 2 => ['pipe', 'w']];
+    /* The child must not read the descriptor: it shares the file offset, and
+     * bytes it consumed would be missing from the reader for good. */
+    $process = @proc_open('exit 0', $descriptors, $pipes);
+    if (is_resource($process)) {
+        fclose($pipes[1]);
+        fclose($pipes[2]);
+        proc_close($process);
+    }
+    return true;
+});
+
+[$results, $exceptions] = await_all([$reader, $caster]);
+
+$data = (string) $results[0];
+printf("bytes: %d of %d\n", strlen($data), strlen($body));
+printf("in order: %s\n", var_export($data === substr($body, 0, strlen($data)), true));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($handle);
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+bytes: 320000 of 320000
+in order: true
+Exceptions: 0
+End
diff --git a/tests/io/089-close_during_copy.phpt b/tests/io/089-close_during_copy.phpt
new file mode 100644
index 0000000..ec74faf
--- /dev/null
+++ b/tests/io/089-close_during_copy.phpt
@@ -0,0 +1,48 @@
+--TEST--
+Closing the source handle while another coroutine is inside stream_copy_to_stream()
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+const ROUNDS = 10;
+
+$source = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($source, str_repeat('0123456789abcdef', 200000));
+$target = tempnam(sys_get_temp_dir(), 'async_io_test_');
+
+for ($round = 0; $round < ROUNDS; $round++) {
+    $in = fopen($source, 'r');
+    $out = fopen($target, 'w');
+
+    /* A non-empty read buffer keeps the copy out of the descriptor-level fast
+     * path, so it alternates parking reads and writes over the two handles. */
+    fread($in, 10);
+
+    $copier = spawn(fn() => @stream_copy_to_stream($in, $out));
+    $closer = spawn(function () use ($in) {
+        suspend();
+        suspend();
+        @fclose($in);
+        return true;
+    });
+
+    await_all([$copier, $closer]);
+    @fclose($out);
+}
+
+echo "Copy survived\n";
+
+unlink($source);
+unlink($target);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+Copy survived
+End
diff --git a/tests/io/090-close_during_get_contents.phpt b/tests/io/090-close_during_get_contents.phpt
new file mode 100644
index 0000000..1fa44fd
--- /dev/null
+++ b/tests/io/090-close_during_get_contents.phpt
@@ -0,0 +1,52 @@
+--TEST--
+Closing a handle while another coroutine is inside stream_get_contents()
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+const ROUNDS = 10;
+
+$source = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($source, str_repeat('0123456789abcdef', 200000));
+
+/* php_stream_read() reports the bytes it delivered before parking, so the
+ * caller loops once more over a handle another coroutine has already freed. */
+$closeWhileParked = function ($handle) {
+    suspend();
+    suspend();
+    @fclose($handle);
+    return true;
+};
+
+for ($round = 0; $round < ROUNDS; $round++) {
+    $handle = fopen($source, 'r');
+    fread($handle, 10);
+    $reader = spawn(fn() => strlen((string) @stream_get_contents($handle)));
+    await_all([$reader, spawn(fn() => $closeWhileParked($handle))]);
+}
+
+echo "Unbounded read survived\n";
+
+for ($round = 0; $round < ROUNDS; $round++) {
+    $handle = fopen($source, 'r');
+    fread($handle, 10);
+    $reader = spawn(fn() => strlen((string) @stream_get_contents($handle, 20000)));
+    await_all([$reader, spawn(fn() => $closeWhileParked($handle))]);
+}
+
+echo "Bounded read survived\n";
+
+unlink($source);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+Unbounded read survived
+Bounded read survived
+End
diff --git a/tests/io/091-filter_remove_during_read.phpt b/tests/io/091-filter_remove_during_read.phpt
new file mode 100644
index 0000000..36b6211
--- /dev/null
+++ b/tests/io/091-filter_remove_during_read.phpt
@@ -0,0 +1,76 @@
+--TEST--
+Removing a read filter while another coroutine is inside fread()
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+/* The filter holds everything it is fed and releases it in one bucket when the
+ * chain is flushed, so the removal lands a large block in the read buffer while
+ * the other coroutine is parked in ops->read. */
+class Hoarder extends php_user_filter
+{
+    private string $held = '';
+
+    public function filter($in, $out, &$consumed, $closing): int
+    {
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $consumed += $bucket->datalen;
+            $this->held .= $bucket->data;
+        }
+
+        if ($closing) {
+            if ($this->held === '') {
+                return PSFS_FEED_ME;
+            }
+            stream_bucket_append($out, stream_bucket_new($this->stream, $this->held));
+            $this->held = '';
+            return PSFS_PASS_ON;
+        }
+
+        if (strlen($this->held) > 1) {
+            stream_bucket_append($out, stream_bucket_new($this->stream, $this->held[0]));
+            $this->held = substr($this->held, 1);
+            return PSFS_PASS_ON;
+        }
+
+        return PSFS_FEED_ME;
+    }
+}
+
+stream_filter_register('hoarder', 'Hoarder');
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($tmpfile, str_repeat('0123456789abcdef', 200000));
+
+$handle = fopen($tmpfile, 'r');
+$filter = stream_filter_append($handle, 'hoarder', STREAM_FILTER_READ);
+for ($i = 0; $i < 12; $i++) {
+    fread($handle, 1);
+}
+
+$reader = spawn(fn() => strlen((string) @fread($handle, 4096)));
+$remover = spawn(function () use ($filter) {
+    suspend();
+    return @stream_filter_remove($filter);
+});
+
+[$results, $exceptions] = await_all([$reader, $remover]);
+
+printf("read: %s\n", var_export($results[0] > 0, true));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($handle);
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+read: true
+Exceptions: 0
+End
diff --git a/tests/io/092-filter_remove_during_write.phpt b/tests/io/092-filter_remove_during_write.phpt
new file mode 100644
index 0000000..2cdd84a
--- /dev/null
+++ b/tests/io/092-filter_remove_during_write.phpt
@@ -0,0 +1,68 @@
+--TEST--
+Closing a handle while a write filter is being flushed out of another coroutine
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+const ROUNDS = 10;
+
+/* Removing the filter flushes the whole hoarded block through ops->write, which
+ * parks; the close then frees the stream the flush writes its position back to. */
+class Hoarder extends php_user_filter
+{
+    private string $held = '';
+
+    public function filter($in, $out, &$consumed, $closing): int
+    {
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $consumed += $bucket->datalen;
+            $this->held .= $bucket->data;
+        }
+
+        if ($closing) {
+            if ($this->held === '') {
+                return PSFS_FEED_ME;
+            }
+            stream_bucket_append($out, stream_bucket_new($this->stream, $this->held));
+            $this->held = '';
+            return PSFS_PASS_ON;
+        }
+
+        return PSFS_FEED_ME;
+    }
+}
+
+stream_filter_register('hoarder', 'Hoarder');
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+
+for ($round = 0; $round < ROUNDS; $round++) {
+    $handle = fopen($tmpfile, 'w');
+    $filter = stream_filter_append($handle, 'hoarder', STREAM_FILTER_WRITE);
+    fwrite($handle, str_repeat('abcdefgh', 400000));
+
+    $remover = spawn(fn() => @stream_filter_remove($filter));
+    $closer = spawn(function () use ($handle) {
+        suspend();
+        @fclose($handle);
+        return true;
+    });
+
+    await_all([$remover, $closer]);
+}
+
+echo "Filter removal survived\n";
+
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+Filter removal survived
+End
diff --git a/tests/io/093-filter_append_during_close.phpt b/tests/io/093-filter_append_during_close.phpt
new file mode 100644
index 0000000..d48b516
--- /dev/null
+++ b/tests/io/093-filter_append_during_close.phpt
@@ -0,0 +1,46 @@
+--TEST--
+Appending a filter that waits for the buffer lock while the handle is closed
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+const ROUNDS = 10;
+
+$source = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($source, str_repeat('0123456789abcdef', 200000));
+
+/* The append queues behind the reader's lock and wakes into a closed stream,
+ * so it never reaches the filter chain it would otherwise be unlinked from. */
+for ($round = 0; $round < ROUNDS; $round++) {
+    $handle = fopen($source, 'r');
+
+    $reader = spawn(fn() => strlen((string) @fread($handle, 3200000)));
+    $appender = spawn(function () use ($handle) {
+        suspend();
+        return @stream_filter_append($handle, 'string.toupper', STREAM_FILTER_READ);
+    });
+    $closer = spawn(function () use ($handle) {
+        suspend();
+        suspend();
+        @fclose($handle);
+        return true;
+    });
+
+    await_all([$reader, $appender, $closer]);
+}
+
+echo "Append survived\n";
+
+unlink($source);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+Append survived
+End

From 73d98c533e077d6ae327badd342f564fc14b9a1a Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Tue, 8 Sep 2026 17:24:17 +0300
Subject: [PATCH 4/5] Cover a chunk size change and a filter removed twice
 during a read

---
 tests/io/094-chunk_size_during_read.phpt | 80 ++++++++++++++++++++++++
 tests/io/095-filter_removed_twice.phpt   | 56 +++++++++++++++++
 2 files changed, 136 insertions(+)
 create mode 100644 tests/io/094-chunk_size_during_read.phpt
 create mode 100644 tests/io/095-filter_removed_twice.phpt

diff --git a/tests/io/094-chunk_size_during_read.phpt b/tests/io/094-chunk_size_during_read.phpt
new file mode 100644
index 0000000..7268c76
--- /dev/null
+++ b/tests/io/094-chunk_size_during_read.phpt
@@ -0,0 +1,80 @@
+--TEST--
+Changing the chunk size while another coroutine is inside a filtered read
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+/* The filter holds what it is fed and releases one byte per call, so the fill
+ * loop keeps reading; suspending hands the resizer its turn in the middle of
+ * that loop, and the counter proves the loop went on afterwards. */
+class Drip extends php_user_filter
+{
+    public static int $callsAfterResize = 0;
+
+    private string $held = '';
+
+    public function filter($in, $out, &$consumed, $closing): int
+    {
+        /* Every call, so the reader yields between two reads whatever the file
+         * backend does with a page-cache hit. */
+        suspend();
+
+        if ($GLOBALS['resized']) {
+            self::$callsAfterResize++;
+        }
+
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $consumed += $bucket->datalen;
+            $this->held .= $bucket->data;
+        }
+
+        if (strlen($this->held) > 1) {
+            stream_bucket_append($out, stream_bucket_new($this->stream, $this->held[0]));
+            $this->held = substr($this->held, 1);
+            return PSFS_PASS_ON;
+        }
+
+        return PSFS_FEED_ME;
+    }
+}
+
+stream_filter_register('drip', 'Drip');
+
+$GLOBALS['resized'] = false;
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($tmpfile, str_repeat('x', 65536));
+
+$handle = fopen($tmpfile, 'r');
+stream_filter_append($handle, 'drip', STREAM_FILTER_READ);
+
+$reader = spawn(fn() => strlen((string) fread($handle, 4096)));
+$resizer = spawn(function () use ($handle) {
+    suspend();
+    $ok = stream_set_chunk_size($handle, 4000000);
+    $GLOBALS['resized'] = true;
+    return $ok;
+});
+
+[$results, $exceptions] = await_all([$reader, $resizer]);
+
+printf("read: %s\n", var_export($results[0] > 0, true));
+printf("read after the resize: %s\n", var_export(Drip::$callsAfterResize > 0, true));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($handle);
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+read: true
+read after the resize: true
+Exceptions: 0
+End
diff --git a/tests/io/095-filter_removed_twice.phpt b/tests/io/095-filter_removed_twice.phpt
new file mode 100644
index 0000000..8dd8397
--- /dev/null
+++ b/tests/io/095-filter_removed_twice.phpt
@@ -0,0 +1,56 @@
+--TEST--
+Two coroutines removing the same filter from one handle
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+/* The filter suspends inside the flush that the removal starts with, so the
+ * second remover has taken hold of the same filter by the time the first one
+ * frees it, and finds it gone when it wakes with the lock. */
+class Parker extends php_user_filter
+{
+    public function filter($in, $out, &$consumed, $closing): int
+    {
+        suspend();
+
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            $consumed += $bucket->datalen;
+            stream_bucket_append($out, $bucket);
+        }
+
+        return PSFS_PASS_ON;
+    }
+}
+
+stream_filter_register('parker', 'Parker');
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($tmpfile, str_repeat('0123456789abcdef', 4096));
+
+$handle = fopen($tmpfile, 'r');
+$filter = stream_filter_append($handle, 'parker', STREAM_FILTER_READ);
+fread($handle, 10);
+
+$first = spawn(fn() => @stream_filter_remove($filter));
+$second = spawn(fn() => @stream_filter_remove($filter));
+
+[$results, $exceptions] = await_all([$first, $second]);
+
+printf("removed once: %s\n", var_export(count(array_filter($results)) === 1, true));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($handle);
+unlink($tmpfile);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+removed once: true
+Exceptions: 0
+End

From af1d74b40ddf1bf1f50a0a1c7f39dfe1b604d209 Mon Sep 17 00:00:00 2001
From: Edmond <1571649+edmonddantes@users.noreply.github.com>
Date: Tue, 8 Sep 2026 18:06:31 +0300
Subject: [PATCH 5/5] Cover a socket, crossed copies and a FIFO under two
 coroutines

---
 tests/io/096-socket_read_and_write.phpt | 45 ++++++++++++++++++++++++
 tests/io/097-crossed_copies.phpt        | 45 ++++++++++++++++++++++++
 tests/io/098-fifo_read_and_write.phpt   | 46 +++++++++++++++++++++++++
 3 files changed, 136 insertions(+)
 create mode 100644 tests/io/096-socket_read_and_write.phpt
 create mode 100644 tests/io/097-crossed_copies.phpt
 create mode 100644 tests/io/098-fifo_read_and_write.phpt

diff --git a/tests/io/096-socket_read_and_write.phpt b/tests/io/096-socket_read_and_write.phpt
new file mode 100644
index 0000000..202fccc
--- /dev/null
+++ b/tests/io/096-socket_read_and_write.phpt
@@ -0,0 +1,45 @@
+--TEST--
+Reading and writing one socket from two coroutines at the same time
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await;
+use function Async\delay;
+
+echo "Start\n";
+
+/* The peer answers only what it has received, so the reader is released by the
+ * writer: a lock over the whole handle would hold the writer behind the reader
+ * and neither would ever run. */
+[$near, $far] = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+
+$peer = spawn(function () use ($far) {
+    $request = fread($far, 100);
+    fwrite($far, "reply to $request");
+});
+
+$reader = spawn(fn() => fgets($near));
+
+$writer = spawn(function () use ($near) {
+    delay(20);
+    return fwrite($near, "request\n");
+});
+
+$answer = await($reader);
+$written = await($writer);
+await($peer);
+
+printf("written: %d\n", $written);
+printf("answer: %s", $answer);
+
+fclose($near);
+fclose($far);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+written: 8
+answer: reply to request
+End
diff --git a/tests/io/097-crossed_copies.phpt b/tests/io/097-crossed_copies.phpt
new file mode 100644
index 0000000..0a0365f
--- /dev/null
+++ b/tests/io/097-crossed_copies.phpt
@@ -0,0 +1,45 @@
+--TEST--
+Two copies running in opposite directions over the same pair of handles
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+
+echo "Start\n";
+
+/* Each copy holds one side of both handles; taken in different orders they
+ * would wait for each other for the rest of the request. */
+$one = tempnam(sys_get_temp_dir(), 'async_io_test_');
+$two = tempnam(sys_get_temp_dir(), 'async_io_test_');
+file_put_contents($one, str_repeat('a', 200000));
+file_put_contents($two, str_repeat('b', 200000));
+
+$first = fopen($one, 'r+');
+$second = fopen($two, 'r+');
+
+/* A byte out of each read buffer keeps both copies off the descriptor-level
+ * fast path and makes the write side reach for the read side. */
+fread($first, 1);
+fread($second, 1);
+
+[$results, $exceptions] = await_all([
+    spawn(fn() => @stream_copy_to_stream($first, $second)),
+    spawn(fn() => @stream_copy_to_stream($second, $first)),
+]);
+
+printf("copies finished: %d\n", count($results));
+echo "Exceptions: " . count($exceptions) . "\n";
+
+fclose($first);
+fclose($second);
+unlink($one);
+unlink($two);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+copies finished: 2
+Exceptions: 0
+End
diff --git a/tests/io/098-fifo_read_and_write.phpt b/tests/io/098-fifo_read_and_write.phpt
new file mode 100644
index 0000000..92cc65b
--- /dev/null
+++ b/tests/io/098-fifo_read_and_write.phpt
@@ -0,0 +1,46 @@
+--TEST--
+Reading and writing one FIFO from two coroutines at the same time
+--SKIPIF--
+<?php
+if (PHP_OS_FAMILY === 'Windows') die('skip POSIX only');
+if (!function_exists('posix_mkfifo')) die('skip ext/posix required');
+?>
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await;
+use function Async\delay;
+
+echo "Start\n";
+
+/* A FIFO comes from the plain files wrapper but has no descriptor offset to
+ * share, so the writer must not queue behind the reader it has to release. */
+$path = tempnam(sys_get_temp_dir(), 'async_io_test_');
+unlink($path);
+posix_mkfifo($path, 0600);
+
+$handle = fopen($path, 'r+');
+
+$reader = spawn(fn() => fread($handle, 5));
+$writer = spawn(function () use ($handle) {
+    delay(20);
+    return fwrite($handle, "hello");
+});
+
+$read = await($reader);
+$written = await($writer);
+
+printf("written: %d\n", $written);
+printf("read: %s\n", $read);
+
+fclose($handle);
+unlink($path);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+written: 5
+read: hello
+End