Read entries of one phar from several coroutines

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Two coroutines decompressing entries of one phar wrote over each other** (#283). Every compressed entry is decompressed into one stream shared by the archive, at the offset the entry takes from that stream before it starts writing, and every call of that span parks: seek to the end, remember the offset, append the decompression filter, copy from the archive stream, flush, remove the filter. A second entry entering the span appended its own filter to the same write chain and put its bytes at the offset the first had remembered, so the first read back a length that matched nothing: four gzipped entries of one archive, read by four coroutines, answered `internal corruption of phar "..." (actual filesize mismatch on file "...")` for three of the four on every run. php-src holds the span as one unit now - both sides of the archive stream around it, both sides of the uncompressed-file stream inside them, taken in the order the stream layer itself takes them - and holds an archive reference for the call, because `phar_get_entry_data()` takes its own only after the entry has been opened: without it the coroutine that finished first dropped the last reference and `phar_archive_delref()` closed the archive stream the others were queued on. Reading an archive is serialized per archive as a result, stored entries included. Evidence: `tests/io/101-phar_entries_decompressed_at_once.phpt`, three broken entries of four before the change and 10 green runs of 10 after. Needs php-src `353ccf7b070`.
+
 - **A cancelled read swallowed the bytes the worker had already taken** (#288). A file read is submitted with `offset = -1`, so `read(2)` moves the descriptor offset; a cancelled or timed-out coroutine returned `-1` and disposed the request, and those bytes went nowhere. The next reader on that handle carried on past them and `ftell()` answered a number unrelated to the descriptor: after a cancelled 64 KB read on a file whose every 16 bytes carry their own record number, the next read returned the record at byte 65536 while `ftell()` said 32. Three parts: a read that finished before the exception arrived hands its count back instead of `-1`, since the bytes are already in the caller's buffer — unless a read filter is attached, where the filter call under a pending exception would mark the stream fatally broken; a read abandoned mid-flight seeks the descriptor back by what it took; and file reads are serialized per handle the way file writes already are, so nothing reads from the advanced offset before the rewind. Evidence: `tests/io/100-cancel_keeps_the_position.phpt`, red 4 runs in 5 before, green 10 in 10 after. The queue costs nothing measurable — one read at a time per handle is what the stream's read-side lock already enforces, and 256 MB in 8 KB chunks on a release build is 2226.3 ms before and 2171.6 ms after, medians of five runs.
 
 - **A cancelled coroutine left its buffer to a thread-pool worker** (#286). `uv_cancel` does not stop a worker that has started, so a read or a write cancelled mid-flight returned while the worker still named the caller's memory: the next `fclose` freed `stream->readbuf` under a `uv_fs_read` — ASAN reports `WRITE of size <chunk>` from `uv__fs_read` — and a cancelled `fwrite` left `uv_fs_write` reading a filter bucket that had already gone, which `strace` shows as `write(...) = -1 EFAULT` (18 of 20 rounds) and which on a recycled block would put another allocation's bytes in the file. A thread-pool read now lands in a buffer of the request's own and the completion copies it across unless the request was abandoned; a thread-pool write copies its payload at submit, before the queue, so that no allocation happens inside a completion callback. Both buffers go with the request, which `libuv_io_req_dispose` frees only after the operation completes, and both come from `malloc` rather than the request allocator: a chunk-sized block per operation would otherwise count twice against `memory_limit`, and freeing it through ZendMM cost an `mmap`/`munmap` pair per read above the 2 MB huge-block threshold. A fire-and-forget write keeps its buffer as before — the reactor already owns it. Nothing in php-src changes. Cost on a release ZTS build, 256 MB read or written, medians of five interleaved runs: 8 KB chunks 1629.7 ms to 1658.8 ms reading and 1471.9 ms to 1505.8 ms writing to `/dev/null`; 1 MB chunks 54.2 ms to 69.8 ms; 4 MB chunks 128.8 ms to 152.6 ms — the residue is the copy itself, which the approach cannot avoid. The footprint is the other half of the price: an in-flight operation now holds a twin of a buffer `memory_limit` already counts, and the twin is outside that limit — a script's real peak is up to twice its accounted one, bounded by the chunk size times the operations in flight. Evidence: `tests/io/099-cancel_during_io.phpt`, which needs the ASAN job to fail: the defect is silent on an ordinary build.

tests/io/101-phar_entries_decompressed_at_once.phpt

@@ -0,0 +1,111 @@
+--TEST--
+Coroutines reading compressed and stored entries of one phar at once
+--SKIPIF--
+<?php
+if (!extension_loaded('phar')) die('skip phar extension required');
+if (!extension_loaded('zlib')) die('skip zlib extension required');
+?>
+--INI--
+phar.readonly=0
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await_all;
+
+echo "Start\n";
+
+/* Every entry of an archive is decompressed into one stream shared by the
+ * archive, at the offset the entry remembers before it starts writing. The
+ * bodies differ per entry, so bytes taken from another entry's span show up as
+ * a mismatch rather than as a shorter read. */
+$archive = sys_get_temp_dir() . '/async_phar_' . getmypid() . '.phar';
+@unlink($archive);
+
+$bodies = [];
+for ($i = 0; $i < 4; $i++) {
+    $bodies["file$i.txt"] = str_repeat("entry-$i-", 40000);
+}
+
+$phar = new Phar($archive);
+foreach ($bodies as $name => $body) {
+    $phar[$name] = $body;
+}
+$phar->compressFiles(Phar::GZ);
+unset($phar);
+
+$readers = [];
+foreach ($bodies as $name => $body) {
+    $readers[] = spawn(function () use ($archive, $name, $body) {
+        $read = @file_get_contents('phar://' . $archive . '/' . $name);
+        if ($read === $body) {
+            return "$name intact";
+        }
+
+        return "$name broken: " . strlen((string) $read) . " bytes of " . strlen($body);
+    });
+}
+
+[$results, $errors] = await_all($readers);
+
+foreach ($results as $result) {
+    echo $result, "\n";
+}
+
+foreach ($errors as $error) {
+    echo 'error: ', $error->getMessage(), "\n";
+}
+
+@unlink($archive);
+
+/* A stored entry is read through the archive stream directly, while a
+ * compressed one is read through the decompression above: the two paths take
+ * the archive stream in the same order, or one of them waits for a side the
+ * other will not release. */
+$mixed = sys_get_temp_dir() . '/async_phar_mixed_' . getmypid() . '.phar';
+@unlink($mixed);
+
+$stored = str_repeat('stored-', 60000);
+$packed = str_repeat('packed-', 60000);
+
+$phar = new Phar($mixed);
+$phar['stored.txt'] = $stored;
+$phar['packed.txt'] = $packed;
+$phar['packed.txt']->compress(Phar::GZ);
+unset($phar);
+
+$handle = fopen('phar://' . $mixed . '/stored.txt', 'rb');
+
+[$results, $errors] = await_all([
+    spawn(fn() => @file_get_contents('phar://' . $mixed . '/packed.txt') === $packed
+        ? 'packed.txt intact' : 'packed.txt broken'),
+    spawn(function () use ($handle, $stored) {
+        fseek($handle, 120000);
+
+        return fread($handle, 1024) === substr($stored, 120000, 1024)
+            ? 'stored.txt intact' : 'stored.txt broken';
+    }),
+]);
+
+foreach ($results as $result) {
+    echo $result, "\n";
+}
+
+foreach ($errors as $error) {
+    echo 'error: ', $error->getMessage(), "\n";
+}
+
+fclose($handle);
+@unlink($mixed);
+
+echo "End\n";
+?>
+--EXPECT--
+Start
+file0.txt intact
+file1.txt intact
+file2.txt intact
+file3.txt intact
+packed.txt intact
+stored.txt intact
+End