Keep the bytes a cancelled read took from the descriptor

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **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.
 
   A Windows stream read still hands the caller's buffer to libuv, where a console line read runs on a thread of its own and an overlapped read keeps the address past `uv_read_stop`. The same treatment does not work there: a stream read carries no in-flight marker, so the dispose would free the buffer at cancel time — earlier than today — and the window would widen. Left open in #286.

libuv_reactor.c

@@ -4591,6 +4591,7 @@ zend_async_trigger_event_t *libuv_new_trigger_event(size_t extra_size)
 static bool libuv_io_close(zend_async_io_t *io_base);
 static void io_close_cb(uv_handle_t *pipe_handle);
 static bool io_file_write_dispatch(async_io_t *io, async_io_req_t *req);
+static bool io_file_read_dispatch(async_io_t *io, async_io_req_t *req);
 
 /* {{{ IO event methods */
 static bool libuv_io_event_start(zend_async_event_t *event)
@@ -4686,14 +4687,29 @@ static bool libuv_io_event_dispose(zend_async_event_t *event)
 		}
 	}
 
+	/* Dispose any file-read requests still queued behind the reader. */
+	if (io->read_q_head != NULL) {
+		async_io_req_t *qreq = io->read_q_head;
+		io->read_q_head = NULL;
+		io->read_q_tail = NULL;
+		while (qreq != NULL) {
+			async_io_req_t *qnext = qreq->q_next;
+			qreq->q_next = NULL;
+			if (qreq->base.dispose != NULL) {
+				qreq->base.dispose(&qreq->base);
+			}
+			qreq = qnext;
+		}
+	}
+
 	/* Dispose any file-write requests still queued behind the writer. */
 	if (io->write_q_head != NULL) {
 		async_io_req_t *qreq = io->write_q_head;
 		io->write_q_head = NULL;
 		io->write_q_tail = NULL;
 		while (qreq != NULL) {
-			async_io_req_t *qnext = qreq->write_q_next;
-			qreq->write_q_next = NULL;
+			async_io_req_t *qnext = qreq->q_next;
+			qreq->q_next = NULL;
 			if (qreq->base.dispose != NULL) {
 				qreq->base.dispose(&qreq->base);
 			}
@@ -4715,6 +4731,37 @@ static bool libuv_io_event_dispose(zend_async_event_t *event)
 
 /* }}} */
 
+/* Take a request out of one of the handle's FIFOs. Answers whether it stood
+ * there, so the caller stops looking: q_next is cleared here, and clearing it
+ * for a request that stands in the other queue would cut the list in two. */
+static bool io_file_queue_unlink(async_io_req_t **head, async_io_req_t **tail, async_io_req_t *req)
+{
+	if (*head == req) {
+		*head = req->q_next;
+		if (*head == NULL) {
+			*tail = NULL;
+		}
+		req->q_next = NULL;
+		return true;
+	}
+
+	async_io_req_t *prev = *head;
+	while (prev != NULL && prev->q_next != req) {
+		prev = prev->q_next;
+	}
+
+	if (prev == NULL) {
+		return false;
+	}
+
+	prev->q_next = req->q_next;
+	if (*tail == req) {
+		*tail = prev;
+	}
+	req->q_next = NULL;
+	return true;
+}
+
 /* {{{ IO request dispose */
 static void libuv_io_req_dispose(zend_async_io_req_t *base_req)
 {
@@ -4744,29 +4791,15 @@ static void libuv_io_req_dispose(zend_async_io_req_t *base_req)
 		io->active_req = NULL;
 	}
 
-	/* A file-write request still waiting its turn in the handle's pending
-	 * queue (its coroutine was cancelled before the write was dispatched)
-	 * — unlink it so the queue never dereferences this freed request. */
-	if (req->io != NULL && req->io->write_q_head != NULL) {
-		async_io_t *io = req->io;
-		if (io->write_q_head == req) {
-			io->write_q_head = req->write_q_next;
-			if (io->write_q_head == NULL) {
-				io->write_q_tail = NULL;
-			}
+	/* A file request still waiting its turn in one of the handle's queues (its
+	 * coroutine was cancelled before the operation was dispatched) — unlink it
+	 * so the queue never dereferences this freed request. */
+	if (req->io != NULL) {
+		if (io_file_queue_unlink(&req->io->read_q_head, &req->io->read_q_tail, req)) {
+			/* nothing else: a request stands in one queue at a time */
 		} else {
-			async_io_req_t *p = io->write_q_head;
-			while (p != NULL && p->write_q_next != req) {
-				p = p->write_q_next;
-			}
-			if (p != NULL) {
-				p->write_q_next = req->write_q_next;
-				if (io->write_q_tail == req) {
-					io->write_q_tail = p;
-				}
-			}
+			io_file_queue_unlink(&req->io->write_q_head, &req->io->write_q_tail, req);
 		}
-		req->write_q_next = NULL;
 	}
 
 	/* Vectored fire-and-forget early-teardown path: writev request that never
@@ -5184,18 +5217,28 @@ static void io_file_read_cb(uv_fs_t *fs_request)
 	if (fs_request->result >= 0) {
 		req->base.transferred = (ssize_t) fs_request->result;
 		if (fs_request->result > 0) {
-			/* An abandoned request has no reader left, and base.buf may already
-			 * be freed — see the same guard in io_file_stat_cb. */
-			if (req->fs_buf != NULL && req->base.buf != NULL
-					&& !(req->uv_flags & ASYNC_IO_REQ_F_DISPOSE_PENDING)) {
-				memcpy(req->base.buf, req->fs_buf, (size_t) fs_request->result);
+			const bool abandoned = (req->uv_flags & ASYNC_IO_REQ_F_DISPOSE_PENDING) != 0;
+
+			if (!abandoned) {
+				/* base.buf may already be freed for an abandoned request — see
+				 * the same guard in io_file_stat_cb. */
+				if (req->fs_buf != NULL && req->base.buf != NULL) {
+					memcpy(req->base.buf, req->fs_buf, (size_t) fs_request->result);
+				}
+			} else if (io->fs_in_flight == 0 && io->crt_fd >= 0) {
+				/* Nobody took these bytes, and the read moved the descriptor
+				 * offset the next reader counts from: put it back (#288). The
+				 * queue keeps the next read out of the way, but a write, an
+				 * fsync or a sendfile on the same handle moves that offset too,
+				 * so the rewind waits for the pool to be empty. */
+				zend_lseek(io->crt_fd, -(zend_off_t) fs_request->result, SEEK_CUR);
 			}
 
 			/* Update tracked offset from kernel position. */
-			const zend_off_t pos = zend_lseek(io->crt_fd, 0, SEEK_CUR);
+			const zend_off_t pos = io->crt_fd >= 0 ? zend_lseek(io->crt_fd, 0, SEEK_CUR) : -1;
 			if (pos >= 0) {
 				io->handle.file.offset = pos;
-			} else {
+			} else if (!abandoned) {
 				io->handle.file.offset += fs_request->result;
 			}
 		}
@@ -5209,6 +5252,27 @@ static void io_file_read_cb(uv_fs_t *fs_request)
 	uv_fs_req_cleanup(fs_request);
 	ZEND_ASYNC_DECREASE_EVENT_COUNT(&io->base.event);
 
+	/* Hand off to the next queued read, now that the offset is settled. A read
+	 * that fails to submit is completed with its own exception and notified
+	 * here; draining continues past it. */
+	bool dispatched = false;
+	while (io->read_q_head != NULL) {
+		async_io_req_t *next = io->read_q_head;
+		io->read_q_head = next->q_next;
+		if (io->read_q_head == NULL) {
+			io->read_q_tail = NULL;
+		}
+		next->q_next = NULL;
+		if (io_file_read_dispatch(io, next)) {
+			dispatched = true;
+			break;
+		}
+		ZEND_ASYNC_CALLBACKS_NOTIFY(&io->base.event, &next->base, next->base.exception);
+	}
+	if (!dispatched) {
+		io->file_read_in_flight = false;
+	}
+
 	io_file_release_fd(io);
 
 	/* Awaiter gone mid-read → finish the deferred dispose, else NOTIFY. */
@@ -5262,11 +5326,11 @@ static void io_file_write_cb(uv_fs_t *fs_request)
 	bool dispatched = false;
 	while (io->write_q_head != NULL) {
 		async_io_req_t *next = io->write_q_head;
-		io->write_q_head = next->write_q_next;
+		io->write_q_head = next->q_next;
 		if (io->write_q_head == NULL) {
 			io->write_q_tail = NULL;
 		}
-		next->write_q_next = NULL;
+		next->q_next = NULL;
 		if (io_file_write_dispatch(io, next)) {
 			dispatched = true;
 			break;
@@ -6099,8 +6163,42 @@ static zend_async_io_req_t *libuv_io_read(zend_async_io_t *io_base, char *buf, s
 		req->fs_buf = pemalloc(max_size, 1);
 	}
 
-	const uv_buf_t read_buffer =
-			uv_buf_init(req->fs_buf != NULL ? req->fs_buf : req->base.buf, (unsigned int) max_size);
+	/* One read at a time per handle: they share the descriptor offset, so a
+	 * second read submitted while the first is in flight starts past bytes the
+	 * first may still have to give back (#288). */
+	if (io->file_read_in_flight) {
+		req->q_next = NULL;
+		if (io->read_q_tail != NULL) {
+			io->read_q_tail->q_next = req;
+		} else {
+			io->read_q_head = req;
+		}
+		io->read_q_tail = req;
+		return &req->base;
+	}
+
+	io->file_read_in_flight = true;
+	if (UNEXPECTED(!io_file_read_dispatch(io, req))) {
+		/* Submit failed: req is completed with its exception, and nothing else
+		 * is queued — clear the in-flight flag. The caller observes the error
+		 * through req->base.exception. */
+		io->file_read_in_flight = false;
+	}
+
+	return &req->base;
+}
+
+/* }}} */
+
+/* {{{ io_file_read_dispatch
+ * Submit one pending file read to libuv. Exactly one is in flight per handle
+ * (io->file_read_in_flight), so the reads never race the shared kernel offset
+ * and an abandoned one can put it back. Returns true when submitted; false on
+ * a submit error, the request carrying the exception for its waiter. */
+static bool io_file_read_dispatch(async_io_t *io, async_io_req_t *req)
+{
+	const uv_buf_t read_buffer = uv_buf_init(req->fs_buf != NULL ? req->fs_buf : req->base.buf,
+			(unsigned int) req->max_size);
 	req->fs_req.data = req;
 
 	/* Use offset=-1 so libuv calls read() instead of pread().
@@ -6109,9 +6207,11 @@ static zend_async_io_req_t *libuv_io_read(zend_async_io_t *io_base, char *buf, s
 			uv_fs_read(UVLOOP, &req->fs_req, io->crt_fd, &read_buffer, 1, -1, io_file_read_cb);
 
 	if (UNEXPECTED(error < 0)) {
-		async_throw_error("Failed to start file read: %s", uv_strerror(error));
-		libuv_io_req_dispose(&req->base);
-		return NULL;
+		req->base.transferred = -1;
+		req->base.completed   = true;
+		req->base.exception   = async_new_exception(
+				async_ce_input_output_exception, "Failed to start file read: %s", uv_strerror(error));
+		return false;
 	}
 
 	/* Pin the io for the op: FILE ios have no uv_close rendezvous, so an owner
@@ -6121,9 +6221,8 @@ static zend_async_io_req_t *libuv_io_read(zend_async_io_t *io_base, char *buf, s
 	io->fs_in_flight++;
 	ZEND_ASYNC_EVENT_ADD_REF(&io->base.event);
 	ZEND_ASYNC_INCREASE_EVENT_COUNT(&io->base.event);
-	return &req->base;
+	return true;
 }
-
 /* }}} */
 
 /* {{{ io_file_write_dispatch
@@ -6295,9 +6394,9 @@ static zend_async_io_req_t *libuv_io_write(zend_async_io_t *io_base, const char
 	 * arrives while another is in flight waits in the FIFO and is
 	 * dispatched by io_file_write_cb when the current one completes. */
 	if (io->file_write_in_flight) {
-		req->write_q_next = NULL;
+		req->q_next = NULL;
 		if (io->write_q_tail != NULL) {
-			io->write_q_tail->write_q_next = req;
+			io->write_q_tail->q_next = req;
 		} else {
 			io->write_q_head = req;
 		}

libuv_reactor.h

@@ -187,6 +187,13 @@ struct _async_io_t
 	async_io_req_t *write_q_tail;
 	bool file_write_in_flight;
 
+	/* File-read serialization, for the same shared offset: a read abandoned
+	 * mid-flight has to put the offset back before the next read starts, or the
+	 * bytes it took go to nobody (#288). Extra reads wait in this FIFO. */
+	async_io_req_t *read_q_head;
+	async_io_req_t *read_q_tail;
+	bool file_read_in_flight;
+
 	/* Thread-pool requests in flight: read, write, flush, stat, and both sides
 	 * of a sendfile. Non-zero means a worker still names crt_fd and the
 	 * caller's buffer, and the close waits. */
@@ -236,8 +243,9 @@ struct _async_io_req_t
 	 * Completion / dispose paths release each ref. Zero means non-writev. */
 	uint16_t writev_nbufs;
 
-	/* Link in async_io_t::write_q_* while this file write waits its turn. */
-	async_io_req_t *write_q_next;
+	/* Link in async_io_t::write_q_* or ::read_q_* while this file operation
+	 * waits its turn. A request is one or the other, never both. */
+	async_io_req_t *q_next;
 
 	union
 	{

tests/io/100-cancel_keeps_the_position.phpt

@@ -0,0 +1,50 @@
+--TEST--
+A cancelled read leaves the bytes it took to the next reader
+--FILE--
+<?php
+
+use function Async\spawn;
+use function Async\await;
+use function Async\await_all;
+use function Async\suspend;
+
+echo "Start\n";
+
+/* Every record carries its own number, so a hole in the file shows up as a
+ * jump rather than as bytes that merely look wrong. */
+$source = tempnam(sys_get_temp_dir(), 'async_io_test_');
+$content = '';
+for ($i = 0; $i < 65536; $i++) {
+    $content .= sprintf('%015d ', $i);
+}
+file_put_contents($source, $content);
+
+/* A read submitted with offset -1 moves the descriptor offset, so a reader
+ * that leaves without its bytes takes them from everyone else. */
+$handle = fopen($source, 'r');
+stream_set_chunk_size($handle, 65536);
+
+$reader = spawn(function () use ($handle) {
+    try {
+        @fread($handle, 65536);
+    } catch (Throwable) {
+    }
+});
+
+suspend();
+$reader->cancel();
+await_all([$reader]);
+
+$next = await(spawn(fn() => fread($handle, 16)));
+
+printf("next=%s ftell=%d\n", rtrim($next), ftell($handle));
+
+fclose($handle);
+unlink($source);
+echo "End\n";
+
+?>
+--EXPECT--
+Start
+next=000000000000000 ftell=16
+End