From 6f759812fc54b95aed39158d4ff3c375e367065f Mon Sep 17 00:00:00 2001
From: alex <alex>
Date: Fri, 14 Oct 2005 10:20:29 +0000
Subject: [PATCH] b=9516 r=alex

 - limit number of in-flight async destroy rpcs MDS issues
   to destroy OST objects
---
 lustre/include/linux/obd.h |  4 ++++
 lustre/ldlm/ldlm_lib.c     |  4 ++++
 lustre/mds/mds_reint.c     | 15 +++++++++------
 lustre/osc/osc_request.c   | 39 +++++++++++++++++++++++++++++++++++++++
 4 files changed, 56 insertions(+), 6 deletions(-)

diff --git a/lustre/include/linux/obd.h b/lustre/include/linux/obd.h
index a401829..ef6e876 100644
--- a/lustre/include/linux/obd.h
+++ b/lustre/include/linux/obd.h
@@ -356,6 +356,10 @@ struct client_obd {
         struct osc_creator       cl_oscc;
         int                      cl_async:1;
 
+        int                      cl_dstr_in_flight;
+        int                      cl_max_dstr_in_flight;
+        wait_queue_head_t        cl_wait_for_destroy_slot;
+
         /* debug stuff */
         struct timeval           cl_last_write_time;
         unsigned long            cl_write_gap_sum;
diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index acd8f02..5a6e255 100644
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -256,6 +256,10 @@ int client_obd_setup(struct obd_device *obddev, obd_count len, void *buf)
         spin_lock_init(&cli->cl_read_page_hist.oh_lock);
         spin_lock_init(&cli->cl_write_page_hist.oh_lock);
 
+        cli->cl_dstr_in_flight = 0;
+        cli->cl_max_dstr_in_flight = OST_MAX_THREADS;
+        init_waitqueue_head(&cli->cl_wait_for_destroy_slot);
+
         memset(&cli->cl_last_write_time, 0,
                sizeof(cli->cl_last_write_time));
         
diff --git a/lustre/mds/mds_reint.c b/lustre/mds/mds_reint.c
index f219db1..4d0d480 100644
--- a/lustre/mds/mds_reint.c
+++ b/lustre/mds/mds_reint.c
@@ -2235,6 +2235,7 @@ static int mds_reint_unlink(struct mds_update_record *rec, int offset,
         int rc = 0, cleanup_phase = 0;
         int unlink_by_id = 0;
         int update_mode;
+        int destroy_objects = 0;
         ENTRY;
 	
         LASSERT(offset == 1 || offset == 3);
@@ -2449,12 +2450,9 @@ static int mds_reint_unlink(struct mds_update_record *rec, int offset,
                                              &lcl) > 0){
                         body->valid |= OBD_MD_FLCOOKIE;
                 }
-                
-                rc = mds_destroy_object(obd, child_inode, 1);
-                if (rc) {
-                        CERROR("can't remove OST object, err %d\n",
-                               rc);
-                }
+       
+                /* destroy OST objects when diskfs transaction is closed */
+                destroy_objects = 1;
 
                 if (child_inode->i_nlink == 0)
                         mds_fidmap_del(obd, &body->id1);
@@ -2486,6 +2484,11 @@ cleanup:
         case 5: /* pending_dir semaphore */
                 up(&mds->mds_pending_dir->d_inode->i_sem);
         case 4: /* child inode semaphore */
+                if (destroy_objects) {
+                        rc = mds_destroy_object(obd, child_inode, 1);
+                        if (rc)
+                                CERROR("can't remove OST object, err %d\n", rc);
+                }
                 UP_READ_I_ALLOC_SEM(child_inode);
                  /* handle splitted dir */
                 if (rc == 0) {
diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index 6a18804..8ed5024 100644
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -495,9 +495,46 @@ static int osc_sync(struct obd_export *exp, struct obdo *oa,
         return rc;
 }
 
+static int osc_destroy_interpret(struct ptlrpc_request *req,
+                                 void *arg, int rc)
+{
+        struct obd_import *imp = req->rq_import; 
+        struct client_obd *cli = &imp->imp_obd->u.cli;
+        ENTRY;
+        spin_lock(&cli->cl_loi_list_lock);
+        cli->cl_dstr_in_flight--;
+        LASSERT(cli->cl_dstr_in_flight >= 0);
+        LASSERT(cli->cl_dstr_in_flight <= cli->cl_dstr_in_flight);
+        spin_unlock(&cli->cl_loi_list_lock);
+        wake_up(&cli->cl_wait_for_destroy_slot);
+        RETURN(rc);
+}
+
+static void osc_grab_destroy_slot(struct client_obd *cli)
+{
+        spin_lock(&cli->cl_loi_list_lock);
+        while (1) {
+                struct l_wait_info lwi = { 0 };
+
+                if (cli->cl_dstr_in_flight < cli->cl_max_dstr_in_flight) {
+                        cli->cl_dstr_in_flight++;
+                        break;
+                }
+
+                spin_unlock(&cli->cl_loi_list_lock);
+                l_wait_event(cli->cl_wait_for_destroy_slot,
+                                cli->cl_dstr_in_flight < cli->cl_max_dstr_in_flight,
+                                &lwi);
+                spin_lock(&cli->cl_loi_list_lock);
+        }
+        spin_unlock(&cli->cl_loi_list_lock);
+}
+
 static int osc_destroy(struct obd_export *exp, struct obdo *oa,
                        struct lov_stripe_md *ea, struct obd_trans_info *oti)
 {
+        struct obd_import *imp = class_exp2cliimp(exp);
+        struct client_obd *cli = &imp->imp_obd->u.cli;
         struct ptlrpc_request *request;
         struct ost_body *body;
         int rc, size = sizeof(*body);
@@ -526,6 +563,8 @@ static int osc_destroy(struct obd_export *exp, struct obdo *oa,
         request->rq_replen = lustre_msg_size(1, &size);
 
         if (oti != NULL && (oti->oti_flags & OBD_MODE_ASYNC)) {
+                osc_grab_destroy_slot(cli);
+                request->rq_interpret_reply = osc_destroy_interpret;
                 ptlrpcd_add_req(request);
                 rc = 0;
         } else {
-- 
1.8.3.1