Serialise buffered stream access across coroutines

ext/standard/streamsfuncs.c

@@ -1372,7 +1372,13 @@ static void apply_filter_to_stream(bool append, INTERNAL_FUNCTION_PARAMETERS)
 		if (append) {
 			zend_result ret = php_stream_filter_append_ex(&stream->readfilters, filter);
 			if (ret != SUCCESS) {
-				php_stream_filter_remove(filter, 1);
+				if (filter->chain == NULL) {
+					/* Never linked: the stream was closed before the append
+					 * started, and the chain to unlink from is gone with it. */
+					php_stream_filter_free(filter);
+				} else {
+					php_stream_filter_remove(filter, 1);
+				}
 				RETURN_FALSE;
 			}
 		} else {
@@ -1389,7 +1395,13 @@ static void apply_filter_to_stream(bool append, INTERNAL_FUNCTION_PARAMETERS)
 		if (append) {
 			zend_result ret = php_stream_filter_append_ex(&stream->writefilters, filter);
 			if (ret != SUCCESS) {
-				php_stream_filter_remove(filter, 1);
+				if (filter->chain == NULL) {
+					/* Never linked: the stream was closed before the append
+					 * started, and the chain to unlink from is gone with it. */
+					php_stream_filter_free(filter);
+				} else {
+					php_stream_filter_remove(filter, 1);
+				}
 				RETURN_FALSE;
 			}
 		} else {

ext/standard/tests/filters/stream_filter_remove_buffer_length.phpt

@@ -0,0 +1,79 @@
+--TEST--
+Flushing a read filter keeps the read buffer length in step with the buffer
+--FILE--
+<?php
+
+/* The filter holds everything it is fed and releases it in one bucket when the
+ * chain is flushed, so the flush has to grow the stream read buffer. */
+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');
+
+const SIZE = 3200000;
+
+$tmpfile = tempnam(sys_get_temp_dir(), 'filter_buflen_');
+file_put_contents($tmpfile, str_repeat('0123456789abcdef', SIZE / 16));
+
+$handle = fopen($tmpfile, 'r');
+$filter = stream_filter_append($handle, 'hoarder', STREAM_FILTER_READ);
+
+$taken = 0;
+for ($i = 0; $i < 12; $i++) {
+    $taken += strlen(fread($handle, 1));
+}
+
+var_dump(stream_filter_remove($filter));
+
+$rest = 0;
+while (true) {
+    $chunk = fread($handle, 4096);
+    if ($chunk === false || $chunk === '') {
+        break;
+    }
+    $rest += strlen($chunk);
+}
+
+printf("delivered: %d of %d\n", $taken + $rest, SIZE);
+printf("eof: %s\n", var_export(feof($handle), true));
+
+fclose($handle);
+
+?>
+--CLEAN--
+<?php
+foreach (glob(sys_get_temp_dir() . '/filter_buflen_*') as $leftover) {
+    unlink($leftover);
+}
+?>
+--EXPECT--
+bool(true)
+delivered: 3200000 of 3200000
+eof: true

main/php_streams.h

@@ -244,6 +244,10 @@ struct _php_stream  {
 	/* how much data to read when filling buffer */
 	size_t chunk_size;
 
+	/* Async: serialises access to the buffer fields above. Owner, depth and
+	 * waiters live in the event; see stream_buffer_lock() in streams.c. */
+	struct _zend_async_event_s *buffer_lock;
+
 #if ZEND_DEBUG
 	const char *open_filename;
 	uint32_t open_lineno;
@@ -262,8 +266,23 @@ struct _php_stream  {
 #define PHP_STREAM_FCLOSE_FDOPEN	1
 #define PHP_STREAM_FCLOSE_FOPENCOOKIE 2
 
-/* allocate a new stream for a particular ops */
 BEGIN_EXTERN_C()
+
+/* Async: serialises the buffered stream API - the read buffer, the position and
+ * the filter chain - across coroutines; unrelated to php_stream_lock(), which is
+ * the advisory file lock. acquire() returns false when the caller
+ * must abandon its operation without touching the stream; on true it stores a
+ * lock that release() must be given back, NULL included: a caller outside a
+ * coroutine has nothing to serialise. The lock is recursive for its owner and
+ * holds a reference of its own, so is_closed() still answers after a call that
+ * parked, including one that returned to a stream fclose() has freed. */
+typedef struct _php_stream_buffer_lock php_stream_buffer_lock_t;
+PHPAPI bool php_stream_buffer_lock_acquire(php_stream *stream, php_stream_buffer_lock_t **held);
+PHPAPI void php_stream_buffer_lock_release(php_stream_buffer_lock_t *lock);
+/* True once the stream behind the lock has been closed and freed. */
+PHPAPI bool php_stream_buffer_lock_is_closed(php_stream_buffer_lock_t *lock);
+
+/* allocate a new stream for a particular ops */
 PHPAPI php_stream *_php_stream_alloc(const php_stream_ops *ops, void *abstract,
 		const char *persistent_id, const char *mode STREAMS_DC);
 END_EXTERN_C()

main/streams/cast.c

@@ -195,12 +195,31 @@ PHPAPI zend_result php_stream_cast(php_stream *stream, int castas, void **ret, i
 
 	/* synchronize our buffer (if possible) */
 	if (ret && castas != PHP_STREAM_AS_FD_FOR_SELECT && castas != PHP_STREAM_AS_FD_FOR_COPY) {
+		php_stream_buffer_lock_t *lock;
+
+		/* The seek moves the descriptor a parked reader is about to credit its
+		 * bytes against, and the reset drops the bytes it has not taken yet. */
+		if (UNEXPECTED(!php_stream_buffer_lock_acquire(stream, &lock))) {
+			return FAILURE;
+		}
+
 		php_stream_flush(stream);
-		if (stream->ops->seek && (stream->flags & PHP_STREAM_FLAG_NO_SEEK) == 0) {
+
+		if (!php_stream_buffer_lock_is_closed(lock) && stream->ops->seek
+				&& (stream->flags & PHP_STREAM_FLAG_NO_SEEK) == 0) {
 			zend_off_t dummy;
 
 			stream->ops->seek(stream, stream->position, SEEK_SET, &dummy);
-			stream->readpos = stream->writepos = 0;
+			if (!php_stream_buffer_lock_is_closed(lock)) {
+				stream->readpos = stream->writepos = 0;
+			}
+		}
+
+		const bool closed = php_stream_buffer_lock_is_closed(lock);
+		php_stream_buffer_lock_release(lock);
+
+		if (UNEXPECTED(closed)) {
+			return FAILURE;
 		}
 	}
 

main/streams/filter.c

@@ -348,9 +348,33 @@ PHPAPI void php_stream_filter_prepend(php_stream_filter_chain *chain, php_stream
 	php_stream_filter_prepend_ex(chain, filter);
 }
 
+static void php_stream_filter_flush_drain(php_stream_bucket_brigade *inp, php_stream_bucket_brigade *outp)
+{
+	php_stream_bucket *bucket;
+
+	while ((bucket = inp->head)) {
+		php_stream_bucket_unlink(bucket);
+		php_stream_bucket_delref(bucket);
+	}
+	while ((bucket = outp->head)) {
+		php_stream_bucket_unlink(bucket);
+		php_stream_bucket_delref(bucket);
+	}
+}
+
 PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, php_stream_filter *filter)
 {
 	php_stream *stream = chain->stream;
+	php_stream_buffer_lock_t *lock;
+
+	/* Both the acquire and the wind below suspend, and the filter is still
+	 * outside the chain here: a close landing meanwhile frees the chain, not the
+	 * filter, and the caller keeps a filter it can still free. The wind itself
+	 * reallocates readbuf and resets the positions a parked reader credits its
+	 * bytes to, which is what the lock is for. */
+	if (UNEXPECTED(!php_stream_buffer_lock_acquire(stream, &lock))) {
+		return FAILURE;
+	}
 
 	filter->prev = chain->tail;
 	filter->next = NULL;
@@ -375,25 +399,24 @@ PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, p
 		php_stream_bucket_append(brig_inp, bucket);
 		status = filter->fops->filter(stream, filter, brig_inp, brig_outp, &consumed, PSFS_FLAG_NORMAL);
 
+		if (UNEXPECTED(php_stream_buffer_lock_is_closed(lock))) {
+			/* A userspace filter suspended, and the stream was closed meanwhile. */
+			php_stream_filter_flush_drain(brig_inp, brig_outp);
+			php_stream_buffer_lock_release(lock);
+			return FAILURE;
+		}
+
 		if (stream->readpos + consumed > (uint32_t)stream->writepos) {
 			/* No behaving filter should cause this. */
 			status = PSFS_ERR_FATAL;
 		}
 
 		switch (status) {
 			case PSFS_ERR_FATAL:
-				while (brig_in.head) {
-					bucket = brig_in.head;
-					php_stream_bucket_unlink(bucket);
-					php_stream_bucket_delref(bucket);
-				}
-				while (brig_out.head) {
-					bucket = brig_out.head;
-					php_stream_bucket_unlink(bucket);
-					php_stream_bucket_delref(bucket);
-				}
+				php_stream_filter_flush_drain(&brig_in, &brig_out);
 				php_stream_warn(stream, FilterFailed,
 						"Filter failed to process pre-buffered data");
+				php_stream_buffer_lock_release(lock);
 				return FAILURE;
 			case PSFS_FEED_ME:
 				/* We don't actually need data yet,
@@ -433,12 +456,18 @@ PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, p
 		}
 	}
 
+	php_stream_buffer_lock_release(lock);
+
 	return SUCCESS;
 }
 
 PHPAPI void php_stream_filter_append(php_stream_filter_chain *chain, php_stream_filter *filter)
 {
 	if (php_stream_filter_append_ex(chain, filter) != SUCCESS) {
+		if (filter->chain != chain) {
+			/* Never linked: the stream was closed before the append started. */
+			return;
+		}
 		if (chain->head == filter) {
 			chain->head = NULL;
 			chain->tail = NULL;
@@ -468,15 +497,33 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 	chain = filter->chain;
 	stream = chain->stream;
 
+	php_stream_buffer_lock_t *lock;
+
+	/* The flush writes into the read buffer and out through ops->write, both of
+	 * which belong to whoever holds the lock. */
+	if (UNEXPECTED(!php_stream_buffer_lock_acquire(stream, &lock))) {
+		return FAILURE;
+	}
+
 	for(current = filter; current; current = current->next) {
 		php_stream_filter_status_t status;
 
 		status = current->fops->filter(stream, current, inp, outp, NULL, flags);
+		if (UNEXPECTED(php_stream_buffer_lock_is_closed(lock))) {
+			/* A userspace filter suspended, and the stream was closed meanwhile. */
+			php_stream_filter_flush_drain(inp, outp);
+			php_stream_buffer_lock_release(lock);
+			return FAILURE;
+		}
 		if (status == PSFS_FEED_ME) {
 			/* We've flushed the data far enough */
+			php_stream_filter_flush_drain(inp, outp);
+			php_stream_buffer_lock_release(lock);
 			return SUCCESS;
 		}
 		if (status == PSFS_ERR_FATAL) {
+			php_stream_filter_flush_drain(inp, outp);
+			php_stream_buffer_lock_release(lock);
 			return FAILURE;
 		}
 		/* Otherwise we have data available to PASS_ON
@@ -499,6 +546,7 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 
 	if (flushed_size == 0) {
 		/* Unlikely, but possible */
+		php_stream_buffer_lock_release(lock);
 		return SUCCESS;
 	}
 
@@ -511,8 +559,10 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 			stream->readpos = 0;
 		}
 		if (flushed_size > (stream->readbuflen - stream->writepos)) {
-			/* Grow the buffer */
-			stream->readbuf = perealloc(stream->readbuf, stream->writepos + flushed_size + stream->chunk_size, stream->is_persistent);
+			/* Grow the buffer. readbuflen follows the allocation: the next fill
+			 * derives its free space from it and would wrap a stale one. */
+			stream->readbuflen = stream->writepos + flushed_size + stream->chunk_size;
+			stream->readbuf = perealloc(stream->readbuf, stream->readbuflen, stream->is_persistent);
 		}
 		while ((bucket = inp->head)) {
 			memcpy(stream->readbuf + stream->writepos, bucket->buf, bucket->buflen);
@@ -524,6 +574,13 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 		/* Send flushed data to the stream */
 		while ((bucket = inp->head)) {
 			ssize_t count = stream->ops->write(stream, bucket->buf, bucket->buflen);
+			if (UNEXPECTED(php_stream_buffer_lock_is_closed(lock))) {
+				/* fclose() from another coroutine freed the stream while the
+				 * write was parked; the position below is gone with it. */
+				php_stream_filter_flush_drain(inp, outp);
+				php_stream_buffer_lock_release(lock);
+				return FAILURE;
+			}
 			if (count > 0) {
 				stream->position += count;
 			}
@@ -532,6 +589,8 @@ PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finis
 		}
 	}
 
+	php_stream_buffer_lock_release(lock);
+
 	return SUCCESS;
 }
 

main/streams/php_stream_filter_api.h

@@ -138,6 +138,10 @@ BEGIN_EXTERN_C()
 PHPAPI void php_stream_filter_prepend(php_stream_filter_chain *chain, php_stream_filter *filter);
 PHPAPI void php_stream_filter_prepend_ex(php_stream_filter_chain *chain, php_stream_filter *filter);
 PHPAPI void php_stream_filter_append(php_stream_filter_chain *chain, php_stream_filter *filter);
+/* Appends the filter to the chain and winds any buffered data through it.
+ * On FAILURE the filter may or may not have reached the chain: filter->chain is
+ * the chain when it did, and NULL when the stream was closed before the append
+ * started. A caller that unlinks the filter must test it. */
 PHPAPI zend_result php_stream_filter_append_ex(php_stream_filter_chain *chain, php_stream_filter *filter);
 PHPAPI zend_result php_stream_filter_flush(php_stream_filter *filter, bool finish);
 PHPAPI php_stream_filter *php_stream_filter_remove(php_stream_filter *filter, bool call_dtor);