Cover concurrent reads from one shared file handle (#280)

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

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

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

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

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

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