From b80d3b71d517acad161ed37a7b00e862f262eec4 Mon Sep 17 00:00:00 2001
From: panda <panda>
Date: Wed, 11 Nov 2009 00:11:41 +0000
Subject: [PATCH] b=21150 i=ZhiYong Tian i=Johann Lombardi

protect dqacq processing against mds-lov cleanup
---
 lustre/include/obd.h         |  1 +
 lustre/include/obd_class.h   | 27 ++++++++++++++++++++++++++-
 lustre/ldlm/ldlm_lib.c       | 10 +++++++---
 lustre/obdclass/obd_config.c |  1 +
 4 files changed, 35 insertions(+), 4 deletions(-)

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index a51cf93..c23c687 100644
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -1013,6 +1013,7 @@ struct obd_device {
         cfs_waitq_t             obd_llog_waitq;
 
         struct obd_device       *obd_observer;
+        struct rw_semaphore     obd_observer_link_sem;
         struct obd_notify_upcall obd_upcall;
         struct obd_export       *obd_self_export;
         /* list of exports in LRU order, for ping evictor, with obd_dev_lock */
diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index 92c4cde..c70135f 100644
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -1625,9 +1625,34 @@ static inline int obd_register_observer(struct obd_device *obd,
 {
         ENTRY;
         OBD_CHECK_DEV(obd);
-        if (obd->obd_observer && observer)
+        down_write(&obd->obd_observer_link_sem);
+        if (obd->obd_observer && observer) {
+                up_write(&obd->obd_observer_link_sem);
                 RETURN(-EALREADY);
+        }
         obd->obd_observer = observer;
+        up_write(&obd->obd_observer_link_sem);
+        RETURN(0);
+}
+
+static inline int obd_pin_observer(struct obd_device *obd,
+                                   struct obd_device **observer)
+{
+        ENTRY;
+        down_read(&obd->obd_observer_link_sem);
+        if (!obd->obd_observer) {
+                *observer = NULL;
+                up_read(&obd->obd_observer_link_sem);
+                RETURN(-ENOENT);
+        }
+        *observer = obd->obd_observer;
+        RETURN(0);
+}
+
+static inline int obd_unpin_observer(struct obd_device *obd)
+{
+        ENTRY;
+        up_read(&obd->obd_observer_link_sem);
         RETURN(0);
 }
 
diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index 9d96387..2a71dba 100644
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -2024,7 +2024,7 @@ int target_handle_dqacq_callback(struct ptlrpc_request *req)
 {
 #ifdef __KERNEL__
         struct obd_device *obd = req->rq_export->exp_obd;
-        struct obd_device *master_obd;
+        struct obd_device *master_obd = NULL, *lov_obd = NULL;
         struct lustre_quota_ctxt *qctxt;
         struct qunit_data *qdata = NULL;
         int rc = 0;
@@ -2055,13 +2055,13 @@ int target_handle_dqacq_callback(struct ptlrpc_request *req)
         }
 
         /* we use the observer */
-        if (!obd->obd_observer || !obd->obd_observer->obd_observer) {
+        if (obd_pin_observer(obd, &lov_obd) ||
+            obd_pin_observer(lov_obd, &master_obd)) {
                 CERROR("Can't find the observer, it is recovering\n");
                 req->rq_status = -EAGAIN;
                 GOTO(send_reply, rc = -EAGAIN);
         }
 
-        master_obd = obd->obd_observer->obd_observer;
         qctxt = &master_obd->u.obt.obt_qctxt;
 
         if (!qctxt->lqc_setup) {
@@ -2096,6 +2096,10 @@ int target_handle_dqacq_callback(struct ptlrpc_request *req)
  send_reply:
         rc = ptlrpc_reply(req);
 out:
+        if (master_obd)
+                obd_unpin_observer(lov_obd);
+        if (lov_obd)
+                obd_unpin_observer(obd);
         OBD_FREE(qdata, sizeof(struct qunit_data));
         RETURN(rc);
 #else
diff --git a/lustre/obdclass/obd_config.c b/lustre/obdclass/obd_config.c
index 727da48..27048f5 100644
--- a/lustre/obdclass/obd_config.c
+++ b/lustre/obdclass/obd_config.c
@@ -227,6 +227,7 @@ int class_attach(struct lustre_cfg *lcfg)
         cfs_waitq_init(&obd->obd_evict_inprogress_waitq);
         cfs_waitq_init(&obd->obd_llog_waitq);
         init_mutex(&obd->obd_llog_cat_process);
+        init_rwsem(&obd->obd_observer_link_sem);
         CFS_INIT_LIST_HEAD(&obd->obd_recovery_queue);
         CFS_INIT_LIST_HEAD(&obd->obd_delayed_reply_queue);
 
-- 
1.8.3.1