* [PATCH 1/3] sunrpc: convert queue_lock from global spinlock to per-cache_detail lock
2026-02-20 12:26 [PATCH 0/3] sunrpc: cache infrastructure scalability improvements Jeff Layton
@ 2026-02-20 12:26 ` Jeff Layton
2026-02-20 12:26 ` [PATCH 2/3] sunrpc: convert queue_wait from global to per-cache_detail waitqueue Jeff Layton
` (2 subsequent siblings)
3 siblings, 0 replies; 7+ messages in thread
From: Jeff Layton @ 2026-02-20 12:26 UTC (permalink / raw)
To: Chuck Lever, NeilBrown, Olga Kornievskaia, Dai Ngo, Tom Talpey,
Trond Myklebust, Anna Schumaker
Cc: linux-nfs, linux-kernel, 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
^ permalink raw reply [flat|nested] 7+ messages in thread* [PATCH 2/3] sunrpc: convert queue_wait from global to per-cache_detail waitqueue
2026-02-20 12:26 [PATCH 0/3] sunrpc: cache infrastructure scalability improvements Jeff Layton
2026-02-20 12:26 ` [PATCH 1/3] sunrpc: convert queue_lock from global spinlock to per-cache_detail lock Jeff Layton
@ 2026-02-20 12:26 ` Jeff Layton
2026-02-20 12:26 ` [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists Jeff Layton
2026-02-20 19:37 ` [PATCH 0/3] sunrpc: cache infrastructure scalability improvements Chuck Lever
3 siblings, 0 replies; 7+ messages in thread
From: Jeff Layton @ 2026-02-20 12:26 UTC (permalink / raw)
To: Chuck Lever, NeilBrown, Olga Kornievskaia, Dai Ngo, Tom Talpey,
Trond Myklebust, Anna Schumaker
Cc: linux-nfs, linux-kernel, 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
^ permalink raw reply [flat|nested] 7+ messages in thread* [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists
2026-02-20 12:26 [PATCH 0/3] sunrpc: cache infrastructure scalability improvements Jeff Layton
2026-02-20 12:26 ` [PATCH 1/3] sunrpc: convert queue_lock from global spinlock to per-cache_detail lock Jeff Layton
2026-02-20 12:26 ` [PATCH 2/3] sunrpc: convert queue_wait from global to per-cache_detail waitqueue Jeff Layton
@ 2026-02-20 12:26 ` Jeff Layton
2026-02-21 22:41 ` NeilBrown
2026-02-20 19:37 ` [PATCH 0/3] sunrpc: cache infrastructure scalability improvements Chuck Lever
3 siblings, 1 reply; 7+ messages in thread
From: Jeff Layton @ 2026-02-20 12:26 UTC (permalink / raw)
To: Chuck Lever, NeilBrown, Olga Kornievskaia, Dai Ngo, Tom Talpey,
Trond Myklebust, Anna Schumaker
Cc: linux-nfs, linux-kernel, 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
^ permalink raw reply [flat|nested] 7+ messages in thread* Re: [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists
2026-02-20 12:26 ` [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists Jeff Layton
@ 2026-02-21 22:41 ` NeilBrown
2026-02-23 14:07 ` Jeff Layton
0 siblings, 1 reply; 7+ messages in thread
From: NeilBrown @ 2026-02-21 22:41 UTC (permalink / raw)
To: Jeff Layton
Cc: Chuck Lever, Olga Kornievskaia, Dai Ngo, Tom Talpey,
Trond Myklebust, Anna Schumaker, linux-nfs, linux-kernel,
Jeff Layton
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
>
>
^ permalink raw reply [flat|nested] 7+ messages in thread* Re: [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists
2026-02-21 22:41 ` NeilBrown
@ 2026-02-23 14:07 ` Jeff Layton
0 siblings, 0 replies; 7+ messages in thread
From: Jeff Layton @ 2026-02-23 14:07 UTC (permalink / raw)
To: NeilBrown
Cc: Chuck Lever, Olga Kornievskaia, Dai Ngo, Tom Talpey,
Trond Myklebust, Anna Schumaker, linux-nfs, linux-kernel
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>
^ permalink raw reply [flat|nested] 7+ messages in thread
* Re: [PATCH 0/3] sunrpc: cache infrastructure scalability improvements
2026-02-20 12:26 [PATCH 0/3] sunrpc: cache infrastructure scalability improvements Jeff Layton
` (2 preceding siblings ...)
2026-02-20 12:26 ` [PATCH 3/3] sunrpc: split cache_detail queue into request and reader lists Jeff Layton
@ 2026-02-20 19:37 ` Chuck Lever
3 siblings, 0 replies; 7+ messages in thread
From: Chuck Lever @ 2026-02-20 19:37 UTC (permalink / raw)
To: NeilBrown, Olga Kornievskaia, Dai Ngo, Tom Talpey,
Trond Myklebust, Anna Schumaker, Jeff Layton
Cc: Chuck Lever, linux-nfs, linux-kernel
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
^ permalink raw reply [flat|nested] 7+ messages in thread