[PATCH 0/3] sunrpc: cache infrastructure scalability improvements

[PATCH 0/3] sunrpc: cache infrastructure scalability improvements

[Jeff Layton]

I've been working on trying to retrofit a netlink upcall into the sunrpc
cache infrastructure. While crawling over that code, I noticed that it
relies on both a global spinlock and waitqueue. The first two patches
convert those to be per-cache_detail instead.

The last patch splits up the cache_detail->queue into two lists: one to
hold cache_readers and one for cache_requests. This simplifies the code,
and the new sequence number that helps the readers track position may
help with implementing netlink upcalls.

Please consider these for v7.1.

Signed-off-by: Jeff Layton <jlayton@kernel.org>
---
Jeff Layton (3):
      sunrpc: convert queue_lock from global spinlock to per-cache_detail lock
      sunrpc: convert queue_wait from global to per-cache_detail waitqueue
      sunrpc: split cache_detail queue into request and reader lists

 include/linux/sunrpc/cache.h |   7 +-
 net/sunrpc/cache.c           | 179 +++++++++++++++++++------------------------
 2 files changed, 85 insertions(+), 101 deletions(-)
---
base-commit: cca65706e7b428b96c951016fc372cc766b94b71
change-id: 20260220-sunrpc-cache-fe4cd44413d3

Best regards,
-- 
Jeff Layton <jlayton@kernel.org>

[PATCH 1/3] sunrpc: convert queue_lock from global spinlock to per-cache_detail lock

[Jeff Layton]

The global queue_lock serializes all upcall queue operations across
every cache_detail instance. Convert it to a per-cache_detail spinlock
so that different caches (e.g. auth.unix.ip vs nfsd.fh) no longer
contend with each other on queue operations.

Signed-off-by: Jeff Layton <jlayton@kernel.org>
---
 include/linux/sunrpc/cache.h |  1 +
 net/sunrpc/cache.c           | 47 ++++++++++++++++++++++----------------------
 2 files changed, 24 insertions(+), 24 deletions(-)

diff --git a/include/linux/sunrpc/cache.h b/include/linux/sunrpc/cache.h
index e783132e481ff2593fdc5d323f7b3a08f85d4cd8..3d32dd1f7b05d35562d2064fed69877b3950fb51 100644
--- a/include/linux/sunrpc/cache.h
+++ b/include/linux/sunrpc/cache.h
@@ -113,6 +113,7 @@ struct cache_detail {
 
 	/* fields for communication over channel */
 	struct list_head	queue;
+	spinlock_t		queue_lock;
 
 	atomic_t		writers;		/* how many time is /channel open */
 	time64_t		last_close;		/* if no writers, when did last close */
diff --git a/net/sunrpc/cache.c b/net/sunrpc/cache.c
index 7c73d1c39687343db02d1f1423b58213b7a35f42..6add2fe311425dc3aec63efce2c4bed06a3d3ba5 100644
--- a/net/sunrpc/cache.c
+++ b/net/sunrpc/cache.c
@@ -400,6 +400,7 @@ void sunrpc_init_cache_detail(struct cache_detail *cd)
 {
 	spin_lock_init(&cd->hash_lock);
 	INIT_LIST_HEAD(&cd->queue);
+	spin_lock_init(&cd->queue_lock);
 	spin_lock(&cache_list_lock);
 	cd->nextcheck = 0;
 	cd->entries = 0;
@@ -803,8 +804,6 @@ void cache_clean_deferred(void *owner)
  *
  */
 
-static DEFINE_SPINLOCK(queue_lock);
-
 struct cache_queue {
 	struct list_head	list;
 	int			reader;	/* if 0, then request */
@@ -847,7 +846,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 	inode_lock(inode); /* protect against multiple concurrent
 			      * readers on this file */
  again:
-	spin_lock(&queue_lock);
+	spin_lock(&cd->queue_lock);
 	/* need to find next request */
 	while (rp->q.list.next != &cd->queue &&
 	       list_entry(rp->q.list.next, struct cache_queue, list)
@@ -856,7 +855,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 		list_move(&rp->q.list, next);
 	}
 	if (rp->q.list.next == &cd->queue) {
-		spin_unlock(&queue_lock);
+		spin_unlock(&cd->queue_lock);
 		inode_unlock(inode);
 		WARN_ON_ONCE(rp->offset);
 		return 0;
@@ -865,7 +864,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 	WARN_ON_ONCE(rq->q.reader);
 	if (rp->offset == 0)
 		rq->readers++;
-	spin_unlock(&queue_lock);
+	spin_unlock(&cd->queue_lock);
 
 	if (rq->len == 0) {
 		err = cache_request(cd, rq);
@@ -876,9 +875,9 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 
 	if (rp->offset == 0 && !test_bit(CACHE_PENDING, &rq->item->flags)) {
 		err = -EAGAIN;
-		spin_lock(&queue_lock);
+		spin_lock(&cd->queue_lock);
 		list_move(&rp->q.list, &rq->q.list);
-		spin_unlock(&queue_lock);
+		spin_unlock(&cd->queue_lock);
 	} else {
 		if (rp->offset + count > rq->len)
 			count = rq->len - rp->offset;
@@ -888,26 +887,26 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 		rp->offset += count;
 		if (rp->offset >= rq->len) {
 			rp->offset = 0;
-			spin_lock(&queue_lock);
+			spin_lock(&cd->queue_lock);
 			list_move(&rp->q.list, &rq->q.list);
-			spin_unlock(&queue_lock);
+			spin_unlock(&cd->queue_lock);
 		}
 		err = 0;
 	}
  out:
 	if (rp->offset == 0) {
 		/* need to release rq */
-		spin_lock(&queue_lock);
+		spin_lock(&cd->queue_lock);
 		rq->readers--;
 		if (rq->readers == 0 &&
 		    !test_bit(CACHE_PENDING, &rq->item->flags)) {
 			list_del(&rq->q.list);
-			spin_unlock(&queue_lock);
+			spin_unlock(&cd->queue_lock);
 			cache_put(rq->item, cd);
 			kfree(rq->buf);
 			kfree(rq);
 		} else
-			spin_unlock(&queue_lock);
+			spin_unlock(&cd->queue_lock);
 	}
 	if (err == -EAGAIN)
 		goto again;
@@ -988,7 +987,7 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
 	if (!rp)
 		return mask;
 
-	spin_lock(&queue_lock);
+	spin_lock(&cd->queue_lock);
 
 	for (cq= &rp->q; &cq->list != &cd->queue;
 	     cq = list_entry(cq->list.next, struct cache_queue, list))
@@ -996,7 +995,7 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
 			mask |= EPOLLIN | EPOLLRDNORM;
 			break;
 		}
-	spin_unlock(&queue_lock);
+	spin_unlock(&cd->queue_lock);
 	return mask;
 }
 
@@ -1011,7 +1010,7 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
 	if (cmd != FIONREAD || !rp)
 		return -EINVAL;
 
-	spin_lock(&queue_lock);
+	spin_lock(&cd->queue_lock);
 
 	/* only find the length remaining in current request,
 	 * or the length of the next request
@@ -1024,7 +1023,7 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
 			len = cr->len - rp->offset;
 			break;
 		}
-	spin_unlock(&queue_lock);
+	spin_unlock(&cd->queue_lock);
 
 	return put_user(len, (int __user *)arg);
 }
@@ -1046,9 +1045,9 @@ static int cache_open(struct inode *inode, struct file *filp,
 		rp->offset = 0;
 		rp->q.reader = 1;
 
-		spin_lock(&queue_lock);
+		spin_lock(&cd->queue_lock);
 		list_add(&rp->q.list, &cd->queue);
-		spin_unlock(&queue_lock);
+		spin_unlock(&cd->queue_lock);
 	}
 	if (filp->f_mode & FMODE_WRITE)
 		atomic_inc(&cd->writers);
@@ -1062,7 +1061,7 @@ static int cache_release(struct inode *inode, struct file *filp,
 	struct cache_reader *rp = filp->private_data;
 
 	if (rp) {
-		spin_lock(&queue_lock);
+		spin_lock(&cd->queue_lock);
 		if (rp->offset) {
 			struct cache_queue *cq;
 			for (cq= &rp->q; &cq->list != &cd->queue;
@@ -1075,7 +1074,7 @@ static int cache_release(struct inode *inode, struct file *filp,
 			rp->offset = 0;
 		}
 		list_del(&rp->q.list);
-		spin_unlock(&queue_lock);
+		spin_unlock(&cd->queue_lock);
 
 		filp->private_data = NULL;
 		kfree(rp);
@@ -1097,7 +1096,7 @@ static void cache_dequeue(struct cache_detail *detail, struct cache_head *ch)
 	struct cache_request *cr;
 	LIST_HEAD(dequeued);
 
-	spin_lock(&queue_lock);
+	spin_lock(&detail->queue_lock);
 	list_for_each_entry_safe(cq, tmp, &detail->queue, list)
 		if (!cq->reader) {
 			cr = container_of(cq, struct cache_request, q);
@@ -1110,7 +1109,7 @@ static void cache_dequeue(struct cache_detail *detail, struct cache_head *ch)
 				continue;
 			list_move(&cr->q.list, &dequeued);
 		}
-	spin_unlock(&queue_lock);
+	spin_unlock(&detail->queue_lock);
 	while (!list_empty(&dequeued)) {
 		cr = list_entry(dequeued.next, struct cache_request, q.list);
 		list_del(&cr->q.list);
@@ -1235,7 +1234,7 @@ static int cache_pipe_upcall(struct cache_detail *detail, struct cache_head *h)
 	crq->buf = buf;
 	crq->len = 0;
 	crq->readers = 0;
-	spin_lock(&queue_lock);
+	spin_lock(&detail->queue_lock);
 	if (test_bit(CACHE_PENDING, &h->flags)) {
 		crq->item = cache_get(h);
 		list_add_tail(&crq->q.list, &detail->queue);
@@ -1243,7 +1242,7 @@ static int cache_pipe_upcall(struct cache_detail *detail, struct cache_head *h)
 	} else
 		/* Lost a race, no longer PENDING, so don't enqueue */
 		ret = -EAGAIN;
-	spin_unlock(&queue_lock);
+	spin_unlock(&detail->queue_lock);
 	wake_up(&queue_wait);
 	if (ret == -EAGAIN) {
 		kfree(buf);

-- 
2.53.0

[PATCH 2/3] sunrpc: convert queue_wait from global to per-cache_detail waitqueue

[Jeff Layton]

The queue_wait waitqueue is currently a file-scoped global, so a
wake_up for one cache_detail wakes pollers on all caches. Convert it
to a per-cache_detail field so that only pollers on the relevant cache
are woken.

Signed-off-by: Jeff Layton <jlayton@kernel.org>
---
 include/linux/sunrpc/cache.h | 2 ++
 net/sunrpc/cache.c           | 7 +++----
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/include/linux/sunrpc/cache.h b/include/linux/sunrpc/cache.h
index 3d32dd1f7b05d35562d2064fed69877b3950fb51..031379efba24d40f64ce346cf1032261d4b98d05 100644
--- a/include/linux/sunrpc/cache.h
+++ b/include/linux/sunrpc/cache.h
@@ -16,6 +16,7 @@
 #include <linux/atomic.h>
 #include <linux/kstrtox.h>
 #include <linux/proc_fs.h>
+#include <linux/wait.h>
 
 /*
  * Each cache requires:
@@ -114,6 +115,7 @@ struct cache_detail {
 	/* fields for communication over channel */
 	struct list_head	queue;
 	spinlock_t		queue_lock;
+	wait_queue_head_t	queue_wait;
 
 	atomic_t		writers;		/* how many time is /channel open */
 	time64_t		last_close;		/* if no writers, when did last close */
diff --git a/net/sunrpc/cache.c b/net/sunrpc/cache.c
index 6add2fe311425dc3aec63efce2c4bed06a3d3ba5..aef2607b3d7ffb61a42b9ea2ec17947465c026dc 100644
--- a/net/sunrpc/cache.c
+++ b/net/sunrpc/cache.c
@@ -401,6 +401,7 @@ void sunrpc_init_cache_detail(struct cache_detail *cd)
 	spin_lock_init(&cd->hash_lock);
 	INIT_LIST_HEAD(&cd->queue);
 	spin_lock_init(&cd->queue_lock);
+	init_waitqueue_head(&cd->queue_wait);
 	spin_lock(&cache_list_lock);
 	cd->nextcheck = 0;
 	cd->entries = 0;
@@ -970,8 +971,6 @@ static ssize_t cache_write(struct file *filp, const char __user *buf,
 	return ret;
 }
 
-static DECLARE_WAIT_QUEUE_HEAD(queue_wait);
-
 static __poll_t cache_poll(struct file *filp, poll_table *wait,
 			       struct cache_detail *cd)
 {
@@ -979,7 +978,7 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
 	struct cache_reader *rp = filp->private_data;
 	struct cache_queue *cq;
 
-	poll_wait(filp, &queue_wait, wait);
+	poll_wait(filp, &cd->queue_wait, wait);
 
 	/* alway allow write */
 	mask = EPOLLOUT | EPOLLWRNORM;
@@ -1243,7 +1242,7 @@ static int cache_pipe_upcall(struct cache_detail *detail, struct cache_head *h)
 		/* Lost a race, no longer PENDING, so don't enqueue */
 		ret = -EAGAIN;
 	spin_unlock(&detail->queue_lock);
-	wake_up(&queue_wait);
+	wake_up(&detail->queue_wait);
 	if (ret == -EAGAIN) {
 		kfree(buf);
 		kfree(crq);

-- 
2.53.0

[PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists

[Jeff Layton]

Replace the single interleaved queue (which mixed cache_request and
cache_reader entries distinguished by a ->reader flag) with two
dedicated lists: cd->requests for upcall requests and cd->readers
for open file handles.

Readers now track their position via a monotonically increasing
sequence number (next_seqno) rather than by their position in the
shared list. Each cache_request is assigned a seqno when enqueued,
and a new cache_next_request() helper finds the next request at or
after a given seqno.

This eliminates the cache_queue wrapper struct entirely, simplifies
the reader-skipping loops in cache_read/cache_poll/cache_ioctl/
cache_release, and makes the data flow easier to reason about.

Also, remove an obsolete comment. CACHE_UPCALLING hasn't existed
since before the git era started.

Signed-off-by: Jeff Layton <jlayton@kernel.org>
---
 include/linux/sunrpc/cache.h |   4 +-
 net/sunrpc/cache.c           | 125 ++++++++++++++++++-------------------------
 2 files changed, 56 insertions(+), 73 deletions(-)

diff --git a/include/linux/sunrpc/cache.h b/include/linux/sunrpc/cache.h
index 031379efba24d40f64ce346cf1032261d4b98d05..b1e595c2615bd4be4d9ad19f71a8f4d08bd74a9b 100644
--- a/include/linux/sunrpc/cache.h
+++ b/include/linux/sunrpc/cache.h
@@ -113,9 +113,11 @@ struct cache_detail {
 	int			entries;
 
 	/* fields for communication over channel */
-	struct list_head	queue;
+	struct list_head	requests;
+	struct list_head	readers;
 	spinlock_t		queue_lock;
 	wait_queue_head_t	queue_wait;
+	u64			next_seqno;
 
 	atomic_t		writers;		/* how many time is /channel open */
 	time64_t		last_close;		/* if no writers, when did last close */
diff --git a/net/sunrpc/cache.c b/net/sunrpc/cache.c
index aef2607b3d7ffb61a42b9ea2ec17947465c026dc..09389ce8b961fe0cb5a472bcf2d3dd0b3faa13a6 100644
--- a/net/sunrpc/cache.c
+++ b/net/sunrpc/cache.c
@@ -399,9 +399,11 @@ static struct delayed_work cache_cleaner;
 void sunrpc_init_cache_detail(struct cache_detail *cd)
 {
 	spin_lock_init(&cd->hash_lock);
-	INIT_LIST_HEAD(&cd->queue);
+	INIT_LIST_HEAD(&cd->requests);
+	INIT_LIST_HEAD(&cd->readers);
 	spin_lock_init(&cd->queue_lock);
 	init_waitqueue_head(&cd->queue_wait);
+	cd->next_seqno = 0;
 	spin_lock(&cache_list_lock);
 	cd->nextcheck = 0;
 	cd->entries = 0;
@@ -796,29 +798,20 @@ void cache_clean_deferred(void *owner)
  * On read, you get a full request, or block.
  * On write, an update request is processed.
  * Poll works if anything to read, and always allows write.
- *
- * Implemented by linked list of requests.  Each open file has
- * a ->private that also exists in this list.  New requests are added
- * to the end and may wakeup and preceding readers.
- * New readers are added to the head.  If, on read, an item is found with
- * CACHE_UPCALLING clear, we free it from the list.
- *
  */
 
-struct cache_queue {
-	struct list_head	list;
-	int			reader;	/* if 0, then request */
-};
 struct cache_request {
-	struct cache_queue	q;
+	struct list_head	list;
 	struct cache_head	*item;
-	char			* buf;
+	char			*buf;
 	int			len;
 	int			readers;
+	u64			seqno;
 };
 struct cache_reader {
-	struct cache_queue	q;
+	struct list_head	list;
 	int			offset;	/* if non-0, we have a refcnt on next request */
+	u64			next_seqno;
 };
 
 static int cache_request(struct cache_detail *detail,
@@ -833,6 +826,17 @@ static int cache_request(struct cache_detail *detail,
 	return PAGE_SIZE - len;
 }
 
+static struct cache_request *
+cache_next_request(struct cache_detail *cd, u64 seqno)
+{
+	struct cache_request *rq;
+
+	list_for_each_entry(rq, &cd->requests, list)
+		if (rq->seqno >= seqno)
+			return rq;
+	return NULL;
+}
+
 static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 			  loff_t *ppos, struct cache_detail *cd)
 {
@@ -849,20 +853,13 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
  again:
 	spin_lock(&cd->queue_lock);
 	/* need to find next request */
-	while (rp->q.list.next != &cd->queue &&
-	       list_entry(rp->q.list.next, struct cache_queue, list)
-	       ->reader) {
-		struct list_head *next = rp->q.list.next;
-		list_move(&rp->q.list, next);
-	}
-	if (rp->q.list.next == &cd->queue) {
+	rq = cache_next_request(cd, rp->next_seqno);
+	if (!rq) {
 		spin_unlock(&cd->queue_lock);
 		inode_unlock(inode);
 		WARN_ON_ONCE(rp->offset);
 		return 0;
 	}
-	rq = container_of(rp->q.list.next, struct cache_request, q.list);
-	WARN_ON_ONCE(rq->q.reader);
 	if (rp->offset == 0)
 		rq->readers++;
 	spin_unlock(&cd->queue_lock);
@@ -877,7 +874,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 	if (rp->offset == 0 && !test_bit(CACHE_PENDING, &rq->item->flags)) {
 		err = -EAGAIN;
 		spin_lock(&cd->queue_lock);
-		list_move(&rp->q.list, &rq->q.list);
+		rp->next_seqno = rq->seqno + 1;
 		spin_unlock(&cd->queue_lock);
 	} else {
 		if (rp->offset + count > rq->len)
@@ -889,7 +886,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 		if (rp->offset >= rq->len) {
 			rp->offset = 0;
 			spin_lock(&cd->queue_lock);
-			list_move(&rp->q.list, &rq->q.list);
+			rp->next_seqno = rq->seqno + 1;
 			spin_unlock(&cd->queue_lock);
 		}
 		err = 0;
@@ -901,7 +898,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
 		rq->readers--;
 		if (rq->readers == 0 &&
 		    !test_bit(CACHE_PENDING, &rq->item->flags)) {
-			list_del(&rq->q.list);
+			list_del(&rq->list);
 			spin_unlock(&cd->queue_lock);
 			cache_put(rq->item, cd);
 			kfree(rq->buf);
@@ -976,7 +973,6 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
 {
 	__poll_t mask;
 	struct cache_reader *rp = filp->private_data;
-	struct cache_queue *cq;
 
 	poll_wait(filp, &cd->queue_wait, wait);
 
@@ -988,12 +984,8 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
 
 	spin_lock(&cd->queue_lock);
 
-	for (cq= &rp->q; &cq->list != &cd->queue;
-	     cq = list_entry(cq->list.next, struct cache_queue, list))
-		if (!cq->reader) {
-			mask |= EPOLLIN | EPOLLRDNORM;
-			break;
-		}
+	if (cache_next_request(cd, rp->next_seqno))
+		mask |= EPOLLIN | EPOLLRDNORM;
 	spin_unlock(&cd->queue_lock);
 	return mask;
 }
@@ -1004,7 +996,7 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
 {
 	int len = 0;
 	struct cache_reader *rp = filp->private_data;
-	struct cache_queue *cq;
+	struct cache_request *rq;
 
 	if (cmd != FIONREAD || !rp)
 		return -EINVAL;
@@ -1014,14 +1006,9 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
 	/* only find the length remaining in current request,
 	 * or the length of the next request
 	 */
-	for (cq= &rp->q; &cq->list != &cd->queue;
-	     cq = list_entry(cq->list.next, struct cache_queue, list))
-		if (!cq->reader) {
-			struct cache_request *cr =
-				container_of(cq, struct cache_request, q);
-			len = cr->len - rp->offset;
-			break;
-		}
+	rq = cache_next_request(cd, rp->next_seqno);
+	if (rq)
+		len = rq->len - rp->offset;
 	spin_unlock(&cd->queue_lock);
 
 	return put_user(len, (int __user *)arg);
@@ -1042,10 +1029,10 @@ static int cache_open(struct inode *inode, struct file *filp,
 			return -ENOMEM;
 		}
 		rp->offset = 0;
-		rp->q.reader = 1;
+		rp->next_seqno = 0;
 
 		spin_lock(&cd->queue_lock);
-		list_add(&rp->q.list, &cd->queue);
+		list_add(&rp->list, &cd->readers);
 		spin_unlock(&cd->queue_lock);
 	}
 	if (filp->f_mode & FMODE_WRITE)
@@ -1062,17 +1049,14 @@ static int cache_release(struct inode *inode, struct file *filp,
 	if (rp) {
 		spin_lock(&cd->queue_lock);
 		if (rp->offset) {
-			struct cache_queue *cq;
-			for (cq= &rp->q; &cq->list != &cd->queue;
-			     cq = list_entry(cq->list.next, struct cache_queue, list))
-				if (!cq->reader) {
-					container_of(cq, struct cache_request, q)
-						->readers--;
-					break;
-				}
+			struct cache_request *rq;
+
+			rq = cache_next_request(cd, rp->next_seqno);
+			if (rq)
+				rq->readers--;
 			rp->offset = 0;
 		}
-		list_del(&rp->q.list);
+		list_del(&rp->list);
 		spin_unlock(&cd->queue_lock);
 
 		filp->private_data = NULL;
@@ -1091,27 +1075,24 @@ static int cache_release(struct inode *inode, struct file *filp,
 
 static void cache_dequeue(struct cache_detail *detail, struct cache_head *ch)
 {
-	struct cache_queue *cq, *tmp;
-	struct cache_request *cr;
+	struct cache_request *cr, *tmp;
 	LIST_HEAD(dequeued);
 
 	spin_lock(&detail->queue_lock);
-	list_for_each_entry_safe(cq, tmp, &detail->queue, list)
-		if (!cq->reader) {
-			cr = container_of(cq, struct cache_request, q);
-			if (cr->item != ch)
-				continue;
-			if (test_bit(CACHE_PENDING, &ch->flags))
-				/* Lost a race and it is pending again */
-				break;
-			if (cr->readers != 0)
-				continue;
-			list_move(&cr->q.list, &dequeued);
-		}
+	list_for_each_entry_safe(cr, tmp, &detail->requests, list) {
+		if (cr->item != ch)
+			continue;
+		if (test_bit(CACHE_PENDING, &ch->flags))
+			/* Lost a race and it is pending again */
+			break;
+		if (cr->readers != 0)
+			continue;
+		list_move(&cr->list, &dequeued);
+	}
 	spin_unlock(&detail->queue_lock);
 	while (!list_empty(&dequeued)) {
-		cr = list_entry(dequeued.next, struct cache_request, q.list);
-		list_del(&cr->q.list);
+		cr = list_entry(dequeued.next, struct cache_request, list);
+		list_del(&cr->list);
 		cache_put(cr->item, detail);
 		kfree(cr->buf);
 		kfree(cr);
@@ -1229,14 +1210,14 @@ static int cache_pipe_upcall(struct cache_detail *detail, struct cache_head *h)
 		return -EAGAIN;
 	}
 
-	crq->q.reader = 0;
 	crq->buf = buf;
 	crq->len = 0;
 	crq->readers = 0;
 	spin_lock(&detail->queue_lock);
 	if (test_bit(CACHE_PENDING, &h->flags)) {
 		crq->item = cache_get(h);
-		list_add_tail(&crq->q.list, &detail->queue);
+		crq->seqno = detail->next_seqno++;
+		list_add_tail(&crq->list, &detail->requests);
 		trace_cache_entry_upcall(detail, h);
 	} else
 		/* Lost a race, no longer PENDING, so don't enqueue */

-- 
2.53.0

Re: [PATCH 0/3] sunrpc: cache infrastructure scalability improvements

[Chuck Lever]

From: Chuck Lever <chuck.lever@oracle.com>

On Fri, 20 Feb 2026 07:26:02 -0500, Jeff Layton wrote:
> I've been working on trying to retrofit a netlink upcall into the sunrpc
> cache infrastructure. While crawling over that code, I noticed that it
> relies on both a global spinlock and waitqueue. The first two patches
> convert those to be per-cache_detail instead.
> 
> The last patch splits up the cache_detail->queue into two lists: one to
> hold cache_readers and one for cache_requests. This simplifies the code,
> and the new sequence number that helps the readers track position may
> help with implementing netlink upcalls.
> 
> [...]

Applied to nfsd-testing, thanks!

[1/3] sunrpc: convert queue_lock from global spinlock to per-cache_detail lock
      commit: 8da8f32e9a2702259cdf97e2f8f492ef9c79db65
[2/3] sunrpc: convert queue_wait from global to per-cache_detail waitqueue
      commit: 802261d8b58dd2f41a52a0c92776e0fb45619efe
[3/3] sunrpc: split cache_detail queue into request and reader lists
      commit: 0eb3d9dc71ada02909e4dfe9cb54e703ec717ed4

--
Chuck Lever

Re: [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists

[NeilBrown]

On Fri, 20 Feb 2026, Jeff Layton wrote:
> Replace the single interleaved queue (which mixed cache_request and
> cache_reader entries distinguished by a ->reader flag) with two
> dedicated lists: cd->requests for upcall requests and cd->readers
> for open file handles.
> 
> Readers now track their position via a monotonically increasing
> sequence number (next_seqno) rather than by their position in the
> shared list. Each cache_request is assigned a seqno when enqueued,
> and a new cache_next_request() helper finds the next request at or
> after a given seqno.
> 
> This eliminates the cache_queue wrapper struct entirely, simplifies
> the reader-skipping loops in cache_read/cache_poll/cache_ioctl/
> cache_release, and makes the data flow easier to reason about.
> 
> Also, remove an obsolete comment. CACHE_UPCALLING hasn't existed
> since before the git era started.
> 
> Signed-off-by: Jeff Layton <jlayton@kernel.org>
> ---
>  include/linux/sunrpc/cache.h |   4 +-
>  net/sunrpc/cache.c           | 125 ++++++++++++++++++-------------------------
>  2 files changed, 56 insertions(+), 73 deletions(-)
> 
> diff --git a/include/linux/sunrpc/cache.h b/include/linux/sunrpc/cache.h
> index 031379efba24d40f64ce346cf1032261d4b98d05..b1e595c2615bd4be4d9ad19f71a8f4d08bd74a9b 100644
> --- a/include/linux/sunrpc/cache.h
> +++ b/include/linux/sunrpc/cache.h
> @@ -113,9 +113,11 @@ struct cache_detail {
>  	int			entries;
>  
>  	/* fields for communication over channel */
> -	struct list_head	queue;
> +	struct list_head	requests;
> +	struct list_head	readers;
>  	spinlock_t		queue_lock;
>  	wait_queue_head_t	queue_wait;
> +	u64			next_seqno;
>  
>  	atomic_t		writers;		/* how many time is /channel open */
>  	time64_t		last_close;		/* if no writers, when did last close */
> diff --git a/net/sunrpc/cache.c b/net/sunrpc/cache.c
> index aef2607b3d7ffb61a42b9ea2ec17947465c026dc..09389ce8b961fe0cb5a472bcf2d3dd0b3faa13a6 100644
> --- a/net/sunrpc/cache.c
> +++ b/net/sunrpc/cache.c
> @@ -399,9 +399,11 @@ static struct delayed_work cache_cleaner;
>  void sunrpc_init_cache_detail(struct cache_detail *cd)
>  {
>  	spin_lock_init(&cd->hash_lock);
> -	INIT_LIST_HEAD(&cd->queue);
> +	INIT_LIST_HEAD(&cd->requests);
> +	INIT_LIST_HEAD(&cd->readers);
>  	spin_lock_init(&cd->queue_lock);
>  	init_waitqueue_head(&cd->queue_wait);
> +	cd->next_seqno = 0;
>  	spin_lock(&cache_list_lock);
>  	cd->nextcheck = 0;
>  	cd->entries = 0;
> @@ -796,29 +798,20 @@ void cache_clean_deferred(void *owner)
>   * On read, you get a full request, or block.
>   * On write, an update request is processed.
>   * Poll works if anything to read, and always allows write.
> - *
> - * Implemented by linked list of requests.  Each open file has
> - * a ->private that also exists in this list.  New requests are added
> - * to the end and may wakeup and preceding readers.
> - * New readers are added to the head.  If, on read, an item is found with
> - * CACHE_UPCALLING clear, we free it from the list.
> - *
>   */
>  
> -struct cache_queue {
> -	struct list_head	list;
> -	int			reader;	/* if 0, then request */
> -};
>  struct cache_request {
> -	struct cache_queue	q;
> +	struct list_head	list;
>  	struct cache_head	*item;
> -	char			* buf;
> +	char			*buf;
>  	int			len;
>  	int			readers;
> +	u64			seqno;
>  };
>  struct cache_reader {
> -	struct cache_queue	q;
> +	struct list_head	list;
>  	int			offset;	/* if non-0, we have a refcnt on next request */
> +	u64			next_seqno;
>  };
>  
>  static int cache_request(struct cache_detail *detail,
> @@ -833,6 +826,17 @@ static int cache_request(struct cache_detail *detail,
>  	return PAGE_SIZE - len;
>  }
>  
> +static struct cache_request *
> +cache_next_request(struct cache_detail *cd, u64 seqno)
> +{
> +	struct cache_request *rq;
> +
> +	list_for_each_entry(rq, &cd->requests, list)
> +		if (rq->seqno >= seqno)
> +			return rq;
> +	return NULL;
> +}
> +
>  static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
>  			  loff_t *ppos, struct cache_detail *cd)
>  {
> @@ -849,20 +853,13 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
>   again:
>  	spin_lock(&cd->queue_lock);
>  	/* need to find next request */
> -	while (rp->q.list.next != &cd->queue &&
> -	       list_entry(rp->q.list.next, struct cache_queue, list)
> -	       ->reader) {
> -		struct list_head *next = rp->q.list.next;
> -		list_move(&rp->q.list, next);
> -	}
> -	if (rp->q.list.next == &cd->queue) {
> +	rq = cache_next_request(cd, rp->next_seqno);
> +	if (!rq) {
>  		spin_unlock(&cd->queue_lock);
>  		inode_unlock(inode);
>  		WARN_ON_ONCE(rp->offset);
>  		return 0;
>  	}
> -	rq = container_of(rp->q.list.next, struct cache_request, q.list);
> -	WARN_ON_ONCE(rq->q.reader);
>  	if (rp->offset == 0)
>  		rq->readers++;
>  	spin_unlock(&cd->queue_lock);
> @@ -877,7 +874,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
>  	if (rp->offset == 0 && !test_bit(CACHE_PENDING, &rq->item->flags)) {
>  		err = -EAGAIN;
>  		spin_lock(&cd->queue_lock);
> -		list_move(&rp->q.list, &rq->q.list);
> +		rp->next_seqno = rq->seqno + 1;
>  		spin_unlock(&cd->queue_lock);

I don't think we need the spin_lock here any more.


>  	} else {
>  		if (rp->offset + count > rq->len)
> @@ -889,7 +886,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
>  		if (rp->offset >= rq->len) {
>  			rp->offset = 0;
>  			spin_lock(&cd->queue_lock);
> -			list_move(&rp->q.list, &rq->q.list);
> +			rp->next_seqno = rq->seqno + 1;
>  			spin_unlock(&cd->queue_lock);

Nor here.


>  		}
>  		err = 0;
> @@ -901,7 +898,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
>  		rq->readers--;
>  		if (rq->readers == 0 &&
>  		    !test_bit(CACHE_PENDING, &rq->item->flags)) {
> -			list_del(&rq->q.list);
> +			list_del(&rq->list);
>  			spin_unlock(&cd->queue_lock);
>  			cache_put(rq->item, cd);
>  			kfree(rq->buf);
> @@ -976,7 +973,6 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
>  {
>  	__poll_t mask;
>  	struct cache_reader *rp = filp->private_data;
> -	struct cache_queue *cq;
>  
>  	poll_wait(filp, &cd->queue_wait, wait);
>  
> @@ -988,12 +984,8 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
>  
>  	spin_lock(&cd->queue_lock);
>  
> -	for (cq= &rp->q; &cq->list != &cd->queue;
> -	     cq = list_entry(cq->list.next, struct cache_queue, list))
> -		if (!cq->reader) {
> -			mask |= EPOLLIN | EPOLLRDNORM;
> -			break;
> -		}
> +	if (cache_next_request(cd, rp->next_seqno))
> +		mask |= EPOLLIN | EPOLLRDNORM;
>  	spin_unlock(&cd->queue_lock);
>  	return mask;
>  }
> @@ -1004,7 +996,7 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
>  {
>  	int len = 0;
>  	struct cache_reader *rp = filp->private_data;
> -	struct cache_queue *cq;
> +	struct cache_request *rq;
>  
>  	if (cmd != FIONREAD || !rp)
>  		return -EINVAL;
> @@ -1014,14 +1006,9 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
>  	/* only find the length remaining in current request,
>  	 * or the length of the next request
>  	 */
> -	for (cq= &rp->q; &cq->list != &cd->queue;
> -	     cq = list_entry(cq->list.next, struct cache_queue, list))
> -		if (!cq->reader) {
> -			struct cache_request *cr =
> -				container_of(cq, struct cache_request, q);
> -			len = cr->len - rp->offset;
> -			break;
> -		}
> +	rq = cache_next_request(cd, rp->next_seqno);
> +	if (rq)
> +		len = rq->len - rp->offset;
>  	spin_unlock(&cd->queue_lock);
>  
>  	return put_user(len, (int __user *)arg);
> @@ -1042,10 +1029,10 @@ static int cache_open(struct inode *inode, struct file *filp,
>  			return -ENOMEM;
>  		}
>  		rp->offset = 0;
> -		rp->q.reader = 1;
> +		rp->next_seqno = 0;
>  
>  		spin_lock(&cd->queue_lock);
> -		list_add(&rp->q.list, &cd->queue);
> +		list_add(&rp->list, &cd->readers);
>  		spin_unlock(&cd->queue_lock);
>  	}
>  	if (filp->f_mode & FMODE_WRITE)
> @@ -1062,17 +1049,14 @@ static int cache_release(struct inode *inode, struct file *filp,
>  	if (rp) {
>  		spin_lock(&cd->queue_lock);
>  		if (rp->offset) {
> -			struct cache_queue *cq;
> -			for (cq= &rp->q; &cq->list != &cd->queue;
> -			     cq = list_entry(cq->list.next, struct cache_queue, list))
> -				if (!cq->reader) {
> -					container_of(cq, struct cache_request, q)
> -						->readers--;
> -					break;
> -				}
> +			struct cache_request *rq;
> +
> +			rq = cache_next_request(cd, rp->next_seqno);
> +			if (rq)
> +				rq->readers--;

Hmm..  The other place where we decrement ->readers we then check if it
is zero and if CACHE_PENDING is clear - and do something.
I suspect we should do that here.
This bug (if I'm right and it is a bug) if there before you patch, but
now might be a good time to fix it?

Thanks.  Nice cleanups.

NeilBrown


>  			rp->offset = 0;
>  		}
> -		list_del(&rp->q.list);
> +		list_del(&rp->list);
>  		spin_unlock(&cd->queue_lock);
>  
>  		filp->private_data = NULL;
> @@ -1091,27 +1075,24 @@ static int cache_release(struct inode *inode, struct file *filp,
>  
>  static void cache_dequeue(struct cache_detail *detail, struct cache_head *ch)
>  {
> -	struct cache_queue *cq, *tmp;
> -	struct cache_request *cr;
> +	struct cache_request *cr, *tmp;
>  	LIST_HEAD(dequeued);
>  
>  	spin_lock(&detail->queue_lock);
> -	list_for_each_entry_safe(cq, tmp, &detail->queue, list)
> -		if (!cq->reader) {
> -			cr = container_of(cq, struct cache_request, q);
> -			if (cr->item != ch)
> -				continue;
> -			if (test_bit(CACHE_PENDING, &ch->flags))
> -				/* Lost a race and it is pending again */
> -				break;
> -			if (cr->readers != 0)
> -				continue;
> -			list_move(&cr->q.list, &dequeued);
> -		}
> +	list_for_each_entry_safe(cr, tmp, &detail->requests, list) {
> +		if (cr->item != ch)
> +			continue;
> +		if (test_bit(CACHE_PENDING, &ch->flags))
> +			/* Lost a race and it is pending again */
> +			break;
> +		if (cr->readers != 0)
> +			continue;
> +		list_move(&cr->list, &dequeued);
> +	}
>  	spin_unlock(&detail->queue_lock);
>  	while (!list_empty(&dequeued)) {
> -		cr = list_entry(dequeued.next, struct cache_request, q.list);
> -		list_del(&cr->q.list);
> +		cr = list_entry(dequeued.next, struct cache_request, list);
> +		list_del(&cr->list);
>  		cache_put(cr->item, detail);
>  		kfree(cr->buf);
>  		kfree(cr);
> @@ -1229,14 +1210,14 @@ static int cache_pipe_upcall(struct cache_detail *detail, struct cache_head *h)
>  		return -EAGAIN;
>  	}
>  
> -	crq->q.reader = 0;
>  	crq->buf = buf;
>  	crq->len = 0;
>  	crq->readers = 0;
>  	spin_lock(&detail->queue_lock);
>  	if (test_bit(CACHE_PENDING, &h->flags)) {
>  		crq->item = cache_get(h);
> -		list_add_tail(&crq->q.list, &detail->queue);
> +		crq->seqno = detail->next_seqno++;
> +		list_add_tail(&crq->list, &detail->requests);
>  		trace_cache_entry_upcall(detail, h);
>  	} else
>  		/* Lost a race, no longer PENDING, so don't enqueue */
> 
> -- 
> 2.53.0
> 
> 

Re: [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists

[Jeff Layton]

On Sun, 2026-02-22 at 09:41 +1100, NeilBrown wrote:
> On Fri, 20 Feb 2026, Jeff Layton wrote:
> > Replace the single interleaved queue (which mixed cache_request and
> > cache_reader entries distinguished by a ->reader flag) with two
> > dedicated lists: cd->requests for upcall requests and cd->readers
> > for open file handles.
> > 
> > Readers now track their position via a monotonically increasing
> > sequence number (next_seqno) rather than by their position in the
> > shared list. Each cache_request is assigned a seqno when enqueued,
> > and a new cache_next_request() helper finds the next request at or
> > after a given seqno.
> > 
> > This eliminates the cache_queue wrapper struct entirely, simplifies
> > the reader-skipping loops in cache_read/cache_poll/cache_ioctl/
> > cache_release, and makes the data flow easier to reason about.
> > 
> > Also, remove an obsolete comment. CACHE_UPCALLING hasn't existed
> > since before the git era started.
> > 
> > Signed-off-by: Jeff Layton <jlayton@kernel.org>
> > ---
> >  include/linux/sunrpc/cache.h |   4 +-
> >  net/sunrpc/cache.c           | 125 ++++++++++++++++++-------------------------
> >  2 files changed, 56 insertions(+), 73 deletions(-)
> > 
> > diff --git a/include/linux/sunrpc/cache.h b/include/linux/sunrpc/cache.h
> > index 031379efba24d40f64ce346cf1032261d4b98d05..b1e595c2615bd4be4d9ad19f71a8f4d08bd74a9b 100644
> > --- a/include/linux/sunrpc/cache.h
> > +++ b/include/linux/sunrpc/cache.h
> > @@ -113,9 +113,11 @@ struct cache_detail {
> >  	int			entries;
> >  
> >  	/* fields for communication over channel */
> > -	struct list_head	queue;
> > +	struct list_head	requests;
> > +	struct list_head	readers;
> >  	spinlock_t		queue_lock;
> >  	wait_queue_head_t	queue_wait;
> > +	u64			next_seqno;
> >  
> >  	atomic_t		writers;		/* how many time is /channel open */
> >  	time64_t		last_close;		/* if no writers, when did last close */
> > diff --git a/net/sunrpc/cache.c b/net/sunrpc/cache.c
> > index aef2607b3d7ffb61a42b9ea2ec17947465c026dc..09389ce8b961fe0cb5a472bcf2d3dd0b3faa13a6 100644
> > --- a/net/sunrpc/cache.c
> > +++ b/net/sunrpc/cache.c
> > @@ -399,9 +399,11 @@ static struct delayed_work cache_cleaner;
> >  void sunrpc_init_cache_detail(struct cache_detail *cd)
> >  {
> >  	spin_lock_init(&cd->hash_lock);
> > -	INIT_LIST_HEAD(&cd->queue);
> > +	INIT_LIST_HEAD(&cd->requests);
> > +	INIT_LIST_HEAD(&cd->readers);
> >  	spin_lock_init(&cd->queue_lock);
> >  	init_waitqueue_head(&cd->queue_wait);
> > +	cd->next_seqno = 0;
> >  	spin_lock(&cache_list_lock);
> >  	cd->nextcheck = 0;
> >  	cd->entries = 0;
> > @@ -796,29 +798,20 @@ void cache_clean_deferred(void *owner)
> >   * On read, you get a full request, or block.
> >   * On write, an update request is processed.
> >   * Poll works if anything to read, and always allows write.
> > - *
> > - * Implemented by linked list of requests.  Each open file has
> > - * a ->private that also exists in this list.  New requests are added
> > - * to the end and may wakeup and preceding readers.
> > - * New readers are added to the head.  If, on read, an item is found with
> > - * CACHE_UPCALLING clear, we free it from the list.
> > - *
> >   */
> >  
> > -struct cache_queue {
> > -	struct list_head	list;
> > -	int			reader;	/* if 0, then request */
> > -};
> >  struct cache_request {
> > -	struct cache_queue	q;
> > +	struct list_head	list;
> >  	struct cache_head	*item;
> > -	char			* buf;
> > +	char			*buf;
> >  	int			len;
> >  	int			readers;
> > +	u64			seqno;
> >  };
> >  struct cache_reader {
> > -	struct cache_queue	q;
> > +	struct list_head	list;
> >  	int			offset;	/* if non-0, we have a refcnt on next request */
> > +	u64			next_seqno;
> >  };
> >  
> >  static int cache_request(struct cache_detail *detail,
> > @@ -833,6 +826,17 @@ static int cache_request(struct cache_detail *detail,
> >  	return PAGE_SIZE - len;
> >  }
> >  
> > +static struct cache_request *
> > +cache_next_request(struct cache_detail *cd, u64 seqno)
> > +{
> > +	struct cache_request *rq;
> > +
> > +	list_for_each_entry(rq, &cd->requests, list)
> > +		if (rq->seqno >= seqno)
> > +			return rq;
> > +	return NULL;
> > +}
> > +
> >  static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
> >  			  loff_t *ppos, struct cache_detail *cd)
> >  {
> > @@ -849,20 +853,13 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
> >   again:
> >  	spin_lock(&cd->queue_lock);
> >  	/* need to find next request */
> > -	while (rp->q.list.next != &cd->queue &&
> > -	       list_entry(rp->q.list.next, struct cache_queue, list)
> > -	       ->reader) {
> > -		struct list_head *next = rp->q.list.next;
> > -		list_move(&rp->q.list, next);
> > -	}
> > -	if (rp->q.list.next == &cd->queue) {
> > +	rq = cache_next_request(cd, rp->next_seqno);
> > +	if (!rq) {
> >  		spin_unlock(&cd->queue_lock);
> >  		inode_unlock(inode);
> >  		WARN_ON_ONCE(rp->offset);
> >  		return 0;
> >  	}
> > -	rq = container_of(rp->q.list.next, struct cache_request, q.list);
> > -	WARN_ON_ONCE(rq->q.reader);
> >  	if (rp->offset == 0)
> >  		rq->readers++;
> >  	spin_unlock(&cd->queue_lock);
> > @@ -877,7 +874,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
> >  	if (rp->offset == 0 && !test_bit(CACHE_PENDING, &rq->item->flags)) {
> >  		err = -EAGAIN;
> >  		spin_lock(&cd->queue_lock);
> > -		list_move(&rp->q.list, &rq->q.list);
> > +		rp->next_seqno = rq->seqno + 1;
> >  		spin_unlock(&cd->queue_lock);
> 
> I don't think we need the spin_lock here any more.
> 

Good point in both places.

> 
> >  	} else {
> >  		if (rp->offset + count > rq->len)
> > @@ -889,7 +886,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
> >  		if (rp->offset >= rq->len) {
> >  			rp->offset = 0;
> >  			spin_lock(&cd->queue_lock);
> > -			list_move(&rp->q.list, &rq->q.list);
> > +			rp->next_seqno = rq->seqno + 1;
> >  			spin_unlock(&cd->queue_lock);
> 
> Nor here.
> 
> 
> >  		}
> >  		err = 0;
> > @@ -901,7 +898,7 @@ static ssize_t cache_read(struct file *filp, char __user *buf, size_t count,
> >  		rq->readers--;
> >  		if (rq->readers == 0 &&
> >  		    !test_bit(CACHE_PENDING, &rq->item->flags)) {
> > -			list_del(&rq->q.list);
> > +			list_del(&rq->list);
> >  			spin_unlock(&cd->queue_lock);
> >  			cache_put(rq->item, cd);
> >  			kfree(rq->buf);
> > @@ -976,7 +973,6 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
> >  {
> >  	__poll_t mask;
> >  	struct cache_reader *rp = filp->private_data;
> > -	struct cache_queue *cq;
> >  
> >  	poll_wait(filp, &cd->queue_wait, wait);
> >  
> > @@ -988,12 +984,8 @@ static __poll_t cache_poll(struct file *filp, poll_table *wait,
> >  
> >  	spin_lock(&cd->queue_lock);
> >  
> > -	for (cq= &rp->q; &cq->list != &cd->queue;
> > -	     cq = list_entry(cq->list.next, struct cache_queue, list))
> > -		if (!cq->reader) {
> > -			mask |= EPOLLIN | EPOLLRDNORM;
> > -			break;
> > -		}
> > +	if (cache_next_request(cd, rp->next_seqno))
> > +		mask |= EPOLLIN | EPOLLRDNORM;
> >  	spin_unlock(&cd->queue_lock);
> >  	return mask;
> >  }
> > @@ -1004,7 +996,7 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
> >  {
> >  	int len = 0;
> >  	struct cache_reader *rp = filp->private_data;
> > -	struct cache_queue *cq;
> > +	struct cache_request *rq;
> >  
> >  	if (cmd != FIONREAD || !rp)
> >  		return -EINVAL;
> > @@ -1014,14 +1006,9 @@ static int cache_ioctl(struct inode *ino, struct file *filp,
> >  	/* only find the length remaining in current request,
> >  	 * or the length of the next request
> >  	 */
> > -	for (cq= &rp->q; &cq->list != &cd->queue;
> > -	     cq = list_entry(cq->list.next, struct cache_queue, list))
> > -		if (!cq->reader) {
> > -			struct cache_request *cr =
> > -				container_of(cq, struct cache_request, q);
> > -			len = cr->len - rp->offset;
> > -			break;
> > -		}
> > +	rq = cache_next_request(cd, rp->next_seqno);
> > +	if (rq)
> > +		len = rq->len - rp->offset;
> >  	spin_unlock(&cd->queue_lock);
> >  
> >  	return put_user(len, (int __user *)arg);
> > @@ -1042,10 +1029,10 @@ static int cache_open(struct inode *inode, struct file *filp,
> >  			return -ENOMEM;
> >  		}
> >  		rp->offset = 0;
> > -		rp->q.reader = 1;
> > +		rp->next_seqno = 0;
> >  
> >  		spin_lock(&cd->queue_lock);
> > -		list_add(&rp->q.list, &cd->queue);
> > +		list_add(&rp->list, &cd->readers);
> >  		spin_unlock(&cd->queue_lock);
> >  	}
> >  	if (filp->f_mode & FMODE_WRITE)
> > @@ -1062,17 +1049,14 @@ static int cache_release(struct inode *inode, struct file *filp,
> >  	if (rp) {
> >  		spin_lock(&cd->queue_lock);
> >  		if (rp->offset) {
> > -			struct cache_queue *cq;
> > -			for (cq= &rp->q; &cq->list != &cd->queue;
> > -			     cq = list_entry(cq->list.next, struct cache_queue, list))
> > -				if (!cq->reader) {
> > -					container_of(cq, struct cache_request, q)
> > -						->readers--;
> > -					break;
> > -				}
> > +			struct cache_request *rq;
> > +
> > +			rq = cache_next_request(cd, rp->next_seqno);
> > +			if (rq)
> > +				rq->readers--;
> 
> Hmm..  The other place where we decrement ->readers we then check if it
> is zero and if CACHE_PENDING is clear - and do something.
> I suspect we should do that here.
> This bug (if I'm right and it is a bug) if there before you patch, but
> now might be a good time to fix it?
> 

Good catch. I'll add a patch to fix that preceding these patches, so it
can go to stable (if we think that's worthwhile).
 
> Thanks.  Nice cleanups.
> 


Thanks for the review! I'll send a v2 with the spinlock fixes and the
above bugfix.



> NeilBrown
> 
> 
> >  			rp->offset = 0;
> >  		}
> > -		list_del(&rp->q.list);
> > +		list_del(&rp->list);
> >  		spin_unlock(&cd->queue_lock);
> >  
> >  		filp->private_data = NULL;
> > @@ -1091,27 +1075,24 @@ static int cache_release(struct inode *inode, struct file *filp,
> >  
> >  static void cache_dequeue(struct cache_detail *detail, struct cache_head *ch)
> >  {
> > -	struct cache_queue *cq, *tmp;
> > -	struct cache_request *cr;
> > +	struct cache_request *cr, *tmp;
> >  	LIST_HEAD(dequeued);
> >  
> >  	spin_lock(&detail->queue_lock);
> > -	list_for_each_entry_safe(cq, tmp, &detail->queue, list)
> > -		if (!cq->reader) {
> > -			cr = container_of(cq, struct cache_request, q);
> > -			if (cr->item != ch)
> > -				continue;
> > -			if (test_bit(CACHE_PENDING, &ch->flags))
> > -				/* Lost a race and it is pending again */
> > -				break;
> > -			if (cr->readers != 0)
> > -				continue;
> > -			list_move(&cr->q.list, &dequeued);
> > -		}
> > +	list_for_each_entry_safe(cr, tmp, &detail->requests, list) {
> > +		if (cr->item != ch)
> > +			continue;
> > +		if (test_bit(CACHE_PENDING, &ch->flags))
> > +			/* Lost a race and it is pending again */
> > +			break;
> > +		if (cr->readers != 0)
> > +			continue;
> > +		list_move(&cr->list, &dequeued);
> > +	}
> >  	spin_unlock(&detail->queue_lock);
> >  	while (!list_empty(&dequeued)) {
> > -		cr = list_entry(dequeued.next, struct cache_request, q.list);
> > -		list_del(&cr->q.list);
> > +		cr = list_entry(dequeued.next, struct cache_request, list);
> > +		list_del(&cr->list);
> >  		cache_put(cr->item, detail);
> >  		kfree(cr->buf);
> >  		kfree(cr);
> > @@ -1229,14 +1210,14 @@ static int cache_pipe_upcall(struct cache_detail *detail, struct cache_head *h)
> >  		return -EAGAIN;
> >  	}
> >  
> > -	crq->q.reader = 0;
> >  	crq->buf = buf;
> >  	crq->len = 0;
> >  	crq->readers = 0;
> >  	spin_lock(&detail->queue_lock);
> >  	if (test_bit(CACHE_PENDING, &h->flags)) {
> >  		crq->item = cache_get(h);
> > -		list_add_tail(&crq->q.list, &detail->queue);
> > +		crq->seqno = detail->next_seqno++;
> > +		list_add_tail(&crq->list, &detail->requests);
> >  		trace_cache_entry_upcall(detail, h);
> >  	} else
> >  		/* Lost a race, no longer PENDING, so don't enqueue */
> > 
> > -- 
> > 2.53.0
> > 
> > 

-- 
Jeff Layton <jlayton@kernel.org>