LU-16796 ptlrpc: Change struct ptlrpc_request_set to use kref

This patch changes struct ptlrpc_request_set to use
kref instead of atomic_t

Signed-off-by: Arshad Hussain <arshad.hussain@aeoncomputing.com>
Change-Id: Icd8dc9d532121b9087455b951a1b7ee922ab532c
Reviewed-on: https://review.whamcloud.com/c/fs/lustre-release/+/53459
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>
Reviewed-by: James Simmons <jsimmons@infradead.org>
Reviewed-by: Neil Brown <neilb@suse.de>
Reviewed-by: Timothy Day <timday@amazon.com>

diff --git a/lustre/include/lustre_net.h b/lustre/include/lustre_net.h
index b34c20f..b93f7dd 100644 (file)
--- a/lustre/include/lustre_net.h
+++ b/lustre/include/lustre_net.h
@@ -609,7 +609,7 @@ union ptlrpc_async_args {
 struct ptlrpc_request_set;
 typedef int (*set_producer_func)(struct ptlrpc_request_set *, void *);
 
-/**
+/*
  * Definition of request set structure.
  * Request set is a list of requests (not necessary to the same target) that
  * once populated with RPCs could be sent in parallel.
@@ -621,32 +621,32 @@ typedef int (*set_producer_func)(struct ptlrpc_request_set *, void *);
  * returned.
  */
 struct ptlrpc_request_set {
-       atomic_t                set_refcount;
-       /** number of in queue requests */
+       struct kref             set_refcount;
+       /* number of in queue requests */
        atomic_t                set_new_count;
-       /** number of uncompleted requests */
+       /* number of uncompleted requests. */
        atomic_t                set_remaining;
-       /** wait queue to wait on for request events */
+       /* wait queue to wait on for request events */
        wait_queue_head_t       set_waitq;
-       /** List of requests in the set */
+       /* List of requests in the set */
        struct list_head        set_requests;
-       /**
+       /*
         * Lock for \a set_new_requests manipulations
         * locked so that any old caller can communicate requests to
         * the set holder who can then fold them into the lock-free set
         */
        spinlock_t              set_new_req_lock;
-       /** List of new yet unsent requests. Only used with ptlrpcd now. */
+       /* List of new yet unsent requests. Only used with ptlrpcd now. */
        struct list_head        set_new_requests;
 
-       /** rq_status of requests that have been freed already */
+       /* rq_status of requests that have been freed already */
        int                     set_rc;
-       /** Additional fields used by the flow control extension */
-       /** Maximum number of RPCs in flight */
+       /* Additional fields used by the flow control extension */
+       /* Maximum number of RPCs in flight */
        int                     set_max_inflight;
-       /** Callback function used to generate RPCs */
+       /* Callback function used to generate RPCs */
        set_producer_func       set_producer;
-       /** opaq argument passed to the producer callback */
+       /* opaq argument passed to the producer callback */
        void                    *set_producer_arg;
        unsigned int             set_allow_intr:1;
 };

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index 5b92485..9b01015 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -1077,7 +1077,7 @@ struct ptlrpc_request_set *ptlrpc_prep_set(void)
        OBD_CPT_ALLOC(set, cfs_cpt_tab, cpt, sizeof(*set));
        if (!set)
                RETURN(NULL);
-       atomic_set(&set->set_refcount, 1);
+       kref_init(&set->set_refcount);
        INIT_LIST_HEAD(&set->set_requests);
        init_waitqueue_head(&set->set_waitq);
        atomic_set(&set->set_new_count, 0);
@@ -1168,7 +1168,7 @@ void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
 
        LASSERT(atomic_read(&set->set_remaining) == 0);
 
-       ptlrpc_reqset_put(set);
+       kref_put(&set->set_refcount, ptlrpc_reqset_free);
        EXIT;
 }
 EXPORT_SYMBOL(ptlrpc_set_destroy);
@@ -3721,3 +3721,18 @@ int ptlrpcd_queue_work(void *handler)
        return 0;
 }
 EXPORT_SYMBOL(ptlrpcd_queue_work);
+
+/**
+ * ptlrpc_reqset_free() - Release memory allocated for ptlrpc_request_set
+ * @kref: kref when dropped below 1
+ *
+ * Used as a kref release callback, when the last user of ptlrpc_request_set
+ * is released.
+ */
+void ptlrpc_reqset_free(struct kref *kref)
+{
+       struct ptlrpc_request_set *set = container_of(kref,
+                                                     struct ptlrpc_request_set,
+                                                     set_refcount);
+       OBD_FREE_PTR(set);
+}

diff --git a/lustre/ptlrpc/ptlrpc_internal.h b/lustre/ptlrpc/ptlrpc_internal.h
index 1574c08..779b422 100644 (file)
--- a/lustre/ptlrpc/ptlrpc_internal.h
+++ b/lustre/ptlrpc/ptlrpc_internal.h
@@ -73,6 +73,7 @@ void ptlrpc_set_mbits(struct ptlrpc_request *req);
 void ptlrpc_assign_next_xid_nolock(struct ptlrpc_request *req);
 __u64 ptlrpc_known_replied_xid(struct obd_import *imp);
 void ptlrpc_add_unreplied(struct ptlrpc_request *req);
+void ptlrpc_reqset_free(struct kref *kerf);
 
 /* events.c */
 int ptlrpc_init_portals(void);
@@ -313,12 +314,6 @@ int nodemap_mod_init(void);
 void nodemap_mod_exit(void);
 #endif /* HAVE_SERVER_SUPPORT */
 
-static inline void ptlrpc_reqset_put(struct ptlrpc_request_set *set)
-{
-       if (atomic_dec_and_test(&set->set_refcount))
-               OBD_FREE_PTR(set);
-}
-
 /** initialise ptlrpc common fields */
 static inline void ptlrpc_req_comm_init(struct ptlrpc_request *req)
 {

diff --git a/lustre/ptlrpc/ptlrpcd.c b/lustre/ptlrpc/ptlrpcd.c
index 55ab470..bf6a530 100644 (file)
--- a/lustre/ptlrpc/ptlrpcd.c
+++ b/lustre/ptlrpc/ptlrpcd.c
@@ -236,12 +236,7 @@ void ptlrpcd_add_req(struct ptlrpc_request *req)
 }
 EXPORT_SYMBOL(ptlrpcd_add_req);
 
-static inline void ptlrpc_reqset_get(struct ptlrpc_request_set *set)
-{
-       atomic_inc(&set->set_refcount);
-}
-
-/**
+/*
  * Check if there is more work to do on ptlrpcd set.
  * Returns 1 if yes.
  */
@@ -335,7 +330,7 @@ static int ptlrpcd_check(struct lu_env *env, struct ptlrpcd_ctl *pc)
                                        continue;
                                }
 
-                               ptlrpc_reqset_get(ps);
+                               kref_get(&ps->set_refcount);
                                spin_unlock(&partner->pc_lock);
 
                                if (atomic_read(&ps->set_new_count)) {
@@ -346,7 +341,7 @@ static int ptlrpcd_check(struct lu_env *env, struct ptlrpcd_ctl *pc)
                                                       rc, partner->pc_index,
                                                       pc->pc_index);
                                }
-                               ptlrpc_reqset_put(ps);
+                               kref_put(&ps->set_refcount, ptlrpc_reqset_free);
                        } while (rc == 0 && pc->pc_cursor != first);
                }
        }