[PATCH v4] fs/pipe: unify the page pools into a single per-pipe pool

[PATCH v4] fs/pipe: unify the page pools into a single per-pipe pool

[Breno Leitao]

Pipes keep two separate page caches:
 a) The per-pipe, lock-protected tmp_page[2]
 b) An on-stack anon_pipe_prealloc burst pool of up to eight pages
    filled before the lock

Converge them into a single per-pipe pool (struct anon_pipe_prealloc
embedded in pipe_inode_info) with the same budget as before: up to
PIPE_PREALLOC_MAX (8) pages, trimmed back to PIPE_PREALLOC_KEEP (2)
after each operation. tmp_page[2] is removed.

Pages are still allocated and freed outside pipe->mutex; only the
assignment into the pool is done under it.

anon_pipe_prefill_and_lock() tops the pool up to the write's page count
-- and returns with pipe->mutex held, so a write acquires the lock only
once.

anon_pipe_trim_and_unlock() trims the pool under that same lock before
dropping it, then frees the excess.

Signed-off-by: Breno Leitao <leitao@debian.org>
Reviewed-by: Mateusz Guzik <mjguzik@gmail.com>
---
Changes in v4:
- Rename anon_pipe_trim_pool_and_unlock() to anon_pipe_trim_and_unlock()
  for naming symmetry (Guzik)
- Link to v3: https://patch.msgid.link/20260709-b4-pipe-unification-v3-1-80cafe097681@debian.org

Changes in v3:
- Squashes them into a single patch (Guzik)
- Fold prefill and trim into one mutex acquire per write. (Guzik)
- Link to v2: https://patch.msgid.link/20260707-b4-pipe-unification-v2-0-eb52bddeeefd@debian.org

Changes in v2:
- User READ_ONCE to read prealloc.count
- Trim the pool at the reader side
- Link to v1: https://lore.kernel.org/r/20260626-b4-pipe-unification-v1-0-d23fa6b1ee27@debian.org
---
 fs/pipe.c                 | 171 ++++++++++++++++++++--------------------------
 include/linux/pipe_fs_i.h |  21 +++++-
 2 files changed, 93 insertions(+), 99 deletions(-)

diff --git a/fs/pipe.c b/fs/pipe.c
index 429b0714ec575..3c6061cefe791 100644
--- a/fs/pipe.c
+++ b/fs/pipe.c
@@ -111,75 +111,95 @@ void pipe_double_lock(struct pipe_inode_info *pipe1,
 	pipe_lock(pipe2);
 }
 
-#define PIPE_PREALLOC_MAX 8
+static struct page *anon_pipe_prealloc_pop(struct anon_pipe_prealloc *prealloc)
+{
+	if (!prealloc->count)
+		return NULL;
 
-struct anon_pipe_prealloc {
-	struct page *pages[PIPE_PREALLOC_MAX];
-	unsigned int count;
-};
+	prealloc->count--;
+
+	return prealloc->pages[prealloc->count];
+}
+
+/* Push a page to the prealloc pool. Returns true if added, false if full. */
+static bool anon_pipe_prealloc_push(struct anon_pipe_prealloc *prealloc,
+				    struct page *page)
+{
+	if (prealloc->count >= PIPE_PREALLOC_MAX)
+		return false;
+	prealloc->pages[prealloc->count++] = page;
+	return true;
+}
 
 /*
- * Pre-allocate pages outside pipe->mutex for multi-page writes.
- * alloc_page() with GFP_HIGHUSER can sleep in reclaim and runs memcg
- * charging; doing it under the mutex stalls a concurrent reader.
- *
- * Loop alloc_page() instead of alloc_pages_bulk_*(): the bulk path refuses
- * __GFP_ACCOUNT under memcg (see commit 8dcb3060d81d "memcg: page_alloc:
- * skip bulk allocator for __GFP_ACCOUNT") and silently degrades to a single
- * page. A per-page loop keeps memcg accounting and the task NUMA mempolicy
- * honoured for every page; the per-call overhead is small compared to the
- * pipe->mutex hold-time being shrunk. Any shortfall is covered by the
- * in-lock alloc_page() fallback in anon_pipe_get_page().
+ * Top up the pipe's own pool, then take pipe->mutex and return with it held.
+ * The shortfall is allocated outside the lock; the push and the caller's write
+ * then run under a single lock acquisition, avoiding a separate prefill
+ * lock/unlock cycle. anon_pipe_get_page() drains the pool instead of allocating
+ * under the lock.
  */
-static void anon_pipe_get_page_prealloc(struct anon_pipe_prealloc *prealloc,
-					size_t total_len)
+static void anon_pipe_prefill_and_lock(struct pipe_inode_info *pipe, size_t total_len)
 {
-	unsigned int want, i;
-	struct page *page;
-
-	prealloc->count = 0;
-	if (total_len <= PAGE_SIZE)
-		return;
+	struct page *pages[PIPE_PREALLOC_MAX];
+	unsigned int want, have, need, n = 0;
 
 	want = min_t(unsigned int, DIV_ROUND_UP(total_len, PAGE_SIZE),
 		     PIPE_PREALLOC_MAX);
+	/* Unlocked read; the pool is refilled under the lock below. */
+	have = min_t(unsigned int, READ_ONCE(pipe->prealloc.count), want);
+	need = want - have;
+
+	if (!need) {
+		mutex_lock(&pipe->mutex);
+		return;
+	}
+
+	while (n < need) {
+		struct page *page = alloc_page(GFP_HIGHUSER | __GFP_ACCOUNT);
 
-	for (i = 0; i < want; i++) {
-		page = alloc_page(GFP_HIGHUSER | __GFP_ACCOUNT);
 		if (!page)
 			break;
-		prealloc->pages[prealloc->count++] = page;
+		pages[n++] = page;
 	}
+
+	mutex_lock(&pipe->mutex);
+	while (n && anon_pipe_prealloc_push(&pipe->prealloc, pages[n - 1]))
+		n--;
+
+	/*
+	 * Just flush any extra page that got affected by the TOCTOU
+	 * effect
+	 */
+	while (n)
+		put_page(pages[--n]);
 }
 
-static struct page *anon_pipe_prealloc_pop(struct anon_pipe_prealloc *prealloc)
+/*
+ * Called with pipe->mutex held. Trim the pool down to PIPE_PREALLOC_KEEP under
+ * the lock, drop it, then free the excess outside the critical section.
+ */
+static void anon_pipe_trim_and_unlock(struct pipe_inode_info *pipe)
 {
-	if (!prealloc->count)
-		return NULL;
+	struct page *excess[PIPE_PREALLOC_MAX];
+	unsigned int nexcess = 0;
 
-	prealloc->count--;
+	while (pipe->prealloc.count > PIPE_PREALLOC_KEEP)
+		excess[nexcess++] = anon_pipe_prealloc_pop(&pipe->prealloc);
+	mutex_unlock(&pipe->mutex);
 
-	return prealloc->pages[prealloc->count];
+	while (nexcess)
+		put_page(excess[--nexcess]);
 }
 
-static struct page *anon_pipe_get_page(struct pipe_inode_info *pipe,
-				       struct anon_pipe_prealloc *prealloc)
+static struct page *anon_pipe_get_page(struct pipe_inode_info *pipe)
 {
 	struct page *page;
 
-	/* Drain prealloc first to keep tmp_page[] hot for later small writes. */
-	page = anon_pipe_prealloc_pop(prealloc);
+	/* Drain the prealloc pool before allocating. Called with mutex held. */
+	page = anon_pipe_prealloc_pop(&pipe->prealloc);
 	if (page)
 		return page;
 
-	for (int i = 0; i < ARRAY_SIZE(pipe->tmp_page); i++) {
-		if (pipe->tmp_page[i]) {
-			page = pipe->tmp_page[i];
-			pipe->tmp_page[i] = NULL;
-			return page;
-		}
-	}
-
 	/* FWIW: This is called with pipe->mutex held */
 	return alloc_page(GFP_HIGHUSER | __GFP_ACCOUNT);
 }
@@ -187,48 +207,11 @@ static struct page *anon_pipe_get_page(struct pipe_inode_info *pipe,
 static void anon_pipe_put_page(struct pipe_inode_info *pipe,
 			       struct page *page)
 {
-	if (page_count(page) == 1) {
-		for (int i = 0; i < ARRAY_SIZE(pipe->tmp_page); i++) {
-			if (!pipe->tmp_page[i]) {
-				pipe->tmp_page[i] = page;
-				return;
-			}
-		}
-	}
-
-	put_page(page);
-}
-
-/*
- * Stash leftover prealloc pages in tmp_page[] so the next write to this
- * pipe gets a hot page without entering the allocator.
- */
-static void anon_pipe_refill_tmp_pages(struct pipe_inode_info *pipe,
-				       struct anon_pipe_prealloc *prealloc)
-{
-	int i, idx;
-
-	if (!prealloc->count)
+	if (page_count(page) == 1 &&
+	    anon_pipe_prealloc_push(&pipe->prealloc, page))
 		return;
 
-	for (i = 0; i < ARRAY_SIZE(pipe->tmp_page); i++) {
-		if (pipe->tmp_page[i])
-			continue;
-		if (!prealloc->count)
-			return;
-		idx = --prealloc->count;
-		pipe->tmp_page[i] = prealloc->pages[idx];
-		prealloc->pages[idx] = NULL;
-	}
-}
-
-/* Runs after mutex_unlock() to keep put_page() out of the critical section. */
-static void anon_pipe_free_pages(struct anon_pipe_prealloc *prealloc)
-{
-	while (prealloc->count) {
-		prealloc->count--;
-		put_page(prealloc->pages[prealloc->count]);
-	}
+	put_page(page);
 }
 
 static void anon_pipe_buf_release(struct pipe_inode_info *pipe,
@@ -485,7 +468,8 @@ anon_pipe_read(struct kiocb *iocb, struct iov_iter *to)
 	}
 	if (pipe_is_empty(pipe))
 		wake_next_reader = false;
-	mutex_unlock(&pipe->mutex);
+	/* Consumed buffers may have refilled the pool; trim it and unlock. */
+	anon_pipe_trim_and_unlock(pipe);
 
 	if (wake_writer)
 		wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
@@ -524,7 +508,6 @@ anon_pipe_write(struct kiocb *iocb, struct iov_iter *from)
 {
 	struct file *filp = iocb->ki_filp;
 	struct pipe_inode_info *pipe = filp->private_data;
-	struct anon_pipe_prealloc prealloc;
 	unsigned int head;
 	ssize_t ret = 0;
 	size_t total_len = iov_iter_count(from);
@@ -548,9 +531,7 @@ anon_pipe_write(struct kiocb *iocb, struct iov_iter *from)
 	if (unlikely(total_len == 0))
 		return 0;
 
-	anon_pipe_get_page_prealloc(&prealloc, total_len);
-
-	mutex_lock(&pipe->mutex);
+	anon_pipe_prefill_and_lock(pipe, total_len);
 
 	if (!pipe->readers) {
 		if ((iocb->ki_flags & IOCB_NOSIGNAL) == 0)
@@ -607,7 +588,7 @@ anon_pipe_write(struct kiocb *iocb, struct iov_iter *from)
 			struct page *page;
 			int copied;
 
-			page = anon_pipe_get_page(pipe, &prealloc);
+			page = anon_pipe_get_page(pipe);
 			if (unlikely(!page)) {
 				if (!ret)
 					ret = -ENOMEM;
@@ -671,11 +652,9 @@ anon_pipe_write(struct kiocb *iocb, struct iov_iter *from)
 		wake_next_writer = true;
 	}
 out:
-	anon_pipe_refill_tmp_pages(pipe, &prealloc);
 	if (pipe_is_full(pipe))
 		wake_next_writer = false;
-	mutex_unlock(&pipe->mutex);
-	anon_pipe_free_pages(&prealloc);
+	anon_pipe_trim_and_unlock(pipe);
 
 	/*
 	 * If we do do a wakeup event, we do a 'sync' wakeup, because we
@@ -956,10 +935,8 @@ void free_pipe_info(struct pipe_inode_info *pipe)
 	if (pipe->watch_queue)
 		put_watch_queue(pipe->watch_queue);
 #endif
-	for (i = 0; i < ARRAY_SIZE(pipe->tmp_page); i++) {
-		if (pipe->tmp_page[i])
-			__free_page(pipe->tmp_page[i]);
-	}
+	for (i = 0; i < pipe->prealloc.count; i++)
+		__free_page(pipe->prealloc.pages[i]);
 	kfree(pipe->bufs);
 	kfree(pipe);
 }
diff --git a/include/linux/pipe_fs_i.h b/include/linux/pipe_fs_i.h
index 7f6a92ac97047..4e2e4f44fbe5f 100644
--- a/include/linux/pipe_fs_i.h
+++ b/include/linux/pipe_fs_i.h
@@ -14,6 +14,9 @@
 #define PIPE_BUF_FLAG_LOSS	0x40	/* Message loss happened after this buffer */
 #endif
 
+#define PIPE_PREALLOC_MAX	8	/* max pages in prealloc pool */
+#define PIPE_PREALLOC_KEEP	2	/* keep at least this many after trim */
+
 /**
  *	struct pipe_buffer - a linux kernel pipe buffer
  *	@page: the page containing the data for the pipe buffer
@@ -58,6 +61,20 @@ union pipe_index {
 	};
 };
 
+/**
+ *	struct anon_pipe_prealloc - per-pipe page preallocation pool
+ *	@pages: array of cached pages (pool)
+ *	@count: number of pages currently in the pool
+ *
+ * Each pipe keeps a small bounded pool of preallocated pages to reduce
+ * allocation overhead during writes. The pool is bounded at PIPE_PREALLOC_MAX
+ * and trimmed down to PIPE_PREALLOC_KEEP after a write completes.
+ */
+struct anon_pipe_prealloc {
+	struct page *pages[PIPE_PREALLOC_MAX];
+	unsigned int count;
+};
+
 /**
  *	struct pipe_inode_info - a linux kernel pipe
  *	@mutex: mutex protecting the whole thing
@@ -68,7 +85,7 @@ union pipe_index {
  *	@max_usage: The maximum number of slots that may be used in the ring
  *	@ring_size: total number of buffers (should be a power of 2)
  *	@nr_accounted: The amount this pipe accounts for in user->pipe_bufs
- *	@tmp_page: cached released page
+ *	@prealloc: per-pipe page preallocation pool
  *	@readers: number of current readers of this pipe
  *	@writers: number of current writers of this pipe
  *	@files: number of struct file referring this pipe (protected by ->i_lock)
@@ -99,7 +116,7 @@ struct pipe_inode_info {
 #ifdef CONFIG_WATCH_QUEUE
 	bool note_loss;
 #endif
-	struct page *tmp_page[2];
+	struct anon_pipe_prealloc prealloc;
 	struct fasync_struct *fasync_readers;
 	struct fasync_struct *fasync_writers;
 	struct pipe_buffer *bufs;

---
base-commit: b9810cd75b9fb56a3425d391cba3f608502bd474
change-id: 20260625-b4-pipe-unification-aba7b8525de7

Best regards,
--  
Breno Leitao <leitao@debian.org>

Re: [PATCH v4] fs/pipe: unify the page pools into a single per-pipe pool

[Oleg Nesterov]

On 07/10, Breno Leitao wrote:
>
> Pipes keep two separate page caches:
>  a) The per-pipe, lock-protected tmp_page[2]
>  b) An on-stack anon_pipe_prealloc burst pool of up to eight pages
>     filled before the lock
>
> Converge them into a single per-pipe pool (struct anon_pipe_prealloc
> embedded in pipe_inode_info) with the same budget as before: up to
> PIPE_PREALLOC_MAX (8) pages, trimmed back to PIPE_PREALLOC_KEEP (2)
> after each operation. tmp_page[2] is removed.
>
> Pages are still allocated and freed outside pipe->mutex; only the
> assignment into the pool is done under it.
>
> anon_pipe_prefill_and_lock() tops the pool up to the write's page count
> -- and returns with pipe->mutex held, so a write acquires the lock only
> once.
>
> anon_pipe_trim_and_unlock() trims the pool under that same lock before
> dropping it, then frees the excess.
>
> Signed-off-by: Breno Leitao <leitao@debian.org>
> Reviewed-by: Mateusz Guzik <mjguzik@gmail.com>

I personally like this change. To me it really makes this logic more
simple/clear.

Reviewed-by: Oleg Nesterov <oleg@redhat.com>


One comment below...

> +static void anon_pipe_prefill_and_lock(struct pipe_inode_info *pipe, size_t total_len)
>  {
> -	unsigned int want, i;
> -	struct page *page;
> -
> -	prealloc->count = 0;
> -	if (total_len <= PAGE_SIZE)
> -		return;
> +	struct page *pages[PIPE_PREALLOC_MAX];
> +	unsigned int want, have, need, n = 0;
>
>  	want = min_t(unsigned int, DIV_ROUND_UP(total_len, PAGE_SIZE),
>  		     PIPE_PREALLOC_MAX);
> +	/* Unlocked read; the pool is refilled under the lock below. */
> +	have = min_t(unsigned int, READ_ONCE(pipe->prealloc.count), want);

And READ_ONCE() is enough correctness wise.

But AFAIK it is not enough to make KCSAN happy. anon_pipe_prealloc_pop/push
which modify ->count under pipe->mutex need WRITE_ONCE() to please KCSAN.

Or anon_pipe_prefill_and_lock() can use data_race(READ_ONCE()).

Oleg.