xprtrdma: Convert send buffer free list to llist

rpcrdma_buffer_get() and rpcrdma_buffer_put() both take rb_lock to
pop/push from the rb_send_bufs free list. Under high I/O concurrency
(e.g., nconnect=N with small random writes), this spinlock is contended
between the request submission path and the transport completion path.

Replace the list_head with an llist_head. The put side uses
lockless llist_add(), which is safe for concurrent producers. The
get side retains the spinlock to satisfy the llist single-consumer
contract portably; submitters continue to serialize there. Completion
handlers returning buffers no longer contend on rb_lock, eliminating
contention on the return path.

rb_lock remains for the MR free list and the tracking lists used
during setup and teardown. rb_free_reps already uses llist_head, so
the llist idiom is established in this structure. The precedent is the
data structure, not the locking: rb_free_reps serializes its single
consumer through the re_receiving gate in rpcrdma_post_recvs, whereas
rb_send_bufs serializes its consumer with rb_lock. Both satisfy the
llist single-consumer contract.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
Signed-off-by: Anna Schumaker <anna.schumaker@hammerspace.com>

diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 92c691d2521f0c16347355e2655a8cae56e7ff81..993bc5c444a4bb1b8faa2953f900abc72d4f1fa4 100644 (file)
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -1120,7 +1120,7 @@ int rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
        INIT_LIST_HEAD(&buf->rb_all_mrs);
        INIT_WORK(&buf->rb_refresh_worker, rpcrdma_mr_refresh_worker);
 
-       INIT_LIST_HEAD(&buf->rb_send_bufs);
+       init_llist_head(&buf->rb_send_bufs);
        INIT_LIST_HEAD(&buf->rb_allreqs);
        INIT_LIST_HEAD(&buf->rb_all_reps);
 
@@ -1134,7 +1134,7 @@ int rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
                                         RPCRDMA_V1_DEF_INLINE_SIZE * 2);
                if (!req)
                        goto out;
-               list_add(&req->rl_list, &buf->rb_send_bufs);
+               llist_add(&req->rl_node, &buf->rb_send_bufs);
        }
 
        init_llist_head(&buf->rb_free_reps);
@@ -1214,16 +1214,14 @@ static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt)
 void
 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
 {
-       rpcrdma_reps_destroy(buf);
+       struct rpcrdma_req *req, *next;
+       struct llist_node *node;
 
-       while (!list_empty(&buf->rb_send_bufs)) {
-               struct rpcrdma_req *req;
+       rpcrdma_reps_destroy(buf);
 
-               req = list_first_entry(&buf->rb_send_bufs,
-                                      struct rpcrdma_req, rl_list);
-               list_del(&req->rl_list);
+       node = llist_del_all(&buf->rb_send_bufs);
+       llist_for_each_entry_safe(req, next, node, rl_node)
                rpcrdma_req_destroy(req);
-       }
 }
 
 /**
@@ -1270,15 +1268,15 @@ void rpcrdma_reply_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req)
 struct rpcrdma_req *
 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
 {
-       struct rpcrdma_req *req;
+       struct llist_node *node;
 
+       /* Calls to llist_del_first are required to be serialized */
        spin_lock(&buffers->rb_lock);
-       req = list_first_entry_or_null(&buffers->rb_send_bufs,
-                                      struct rpcrdma_req, rl_list);
-       if (req)
-               list_del_init(&req->rl_list);
+       node = llist_del_first(&buffers->rb_send_bufs);
        spin_unlock(&buffers->rb_lock);
-       return req;
+       if (!node)
+               return NULL;
+       return llist_entry(node, struct rpcrdma_req, rl_node);
 }
 
 /**
@@ -1291,9 +1289,7 @@ void rpcrdma_buffer_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req)
 {
        rpcrdma_reply_put(buffers, req);
 
-       spin_lock(&buffers->rb_lock);
-       list_add(&req->rl_list, &buffers->rb_send_bufs);
-       spin_unlock(&buffers->rb_lock);
+       llist_add(&req->rl_node, &buffers->rb_send_bufs);
 }
 
 /* Returns a pointer to a rpcrdma_regbuf object, or NULL.

diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h
index f879d9b9f57ef4001051df5a30f052426ab0c5c7..ae036719f840c70109b220a87693b610b619affb 100644 (file)
--- a/net/sunrpc/xprtrdma/xprt_rdma.h
+++ b/net/sunrpc/xprtrdma/xprt_rdma.h
@@ -332,7 +332,7 @@ enum {
 
 struct rpcrdma_buffer;
 struct rpcrdma_req {
-       struct list_head        rl_list;
+       struct llist_node       rl_node;
        struct rpc_rqst         rl_slot;
        struct rpcrdma_rep      *rl_reply;
        struct xdr_stream       rl_stream;
@@ -374,14 +374,14 @@ rpcrdma_mr_pop(struct list_head *list)
 }
 
 /*
- * struct rpcrdma_buffer -- holds list/queue of pre-registered memory for
- * inline requests/replies, and client/server credits.
+ * struct rpcrdma_buffer -- holds pre-registered memory for inline
+ * requests/replies, and client/server credits.
  *
  * One of these is associated with a transport instance
  */
 struct rpcrdma_buffer {
        spinlock_t              rb_lock;
-       struct list_head        rb_send_bufs;
+       struct llist_head       rb_send_bufs;
        struct list_head        rb_mrs;
 
        unsigned long           rb_sc_head;