Branch b1_6

Quota slave depends qctxt->lqc_import to send its quota request.
This pointer will be invalid if mds did failover or broke its
connect to osts, which leads to LBUG.
b=12374
i=green
i=adilger

diff --git a/lustre/ChangeLog b/lustre/ChangeLog
index 7d2f827..a738ab5 100644 (file)
--- a/lustre/ChangeLog
+++ b/lustre/ChangeLog
@@ -35,6 +35,15 @@ Details    : Updated patch inode-nr_unused-2.6.9-rhel4.patch
             Updated kernel config files
 
 Severity   : normal
+Frequency  : rare
+Bugzilla   : 12374
+Description: lquota slave complains LBUG when reconnecting with mds
+             or failover in mds.
+Details    : quota slave depends qctxt->lqc_import to send its quota request.
+             This pointer will be invalid if mds did failover or broke its
+             connect to osts, which leads to LBUG.
+
+Severity   : normal    
 Frequency  : when qunit size is too small(less than 20M)
 Bugzilla   : 12588
 Description: write is stopped by improper -EDQUOT

diff --git a/lustre/include/lustre_quota.h b/lustre/include/lustre_quota.h
index 1ead9b0..c7e97c1 100644 (file)
--- a/lustre/include/lustre_quota.h
+++ b/lustre/include/lustre_quota.h
@@ -136,6 +136,7 @@ struct lustre_quota_ctxt {
                       lqc_atype:2,      /* Turn on user/group quota at setup automatically, 
                                          * 0: none, 1: user quota, 2: group quota, 3: both */
                       lqc_status:1;     /* Quota status. 0:Off, 1:On */
+        spinlock_t    lqc_lock;         /* guard lqc_imp_valid now */
         unsigned long lqc_iunit_sz;     /* Unit size of file quota */
         unsigned long lqc_itune_sz;     /* Trigger dqacq when available file quota less than
                                          * this value, trigger dqrel when available file quota
@@ -198,6 +199,9 @@ typedef struct {
         /* For quota slave, set import, trigger quota recovery */
         int (*quota_setinfo) (struct obd_export *, struct obd_device *);
         
+        /* For quota slave, clear import when relative import is invalid */
+        int (*quota_clearinfo) (struct obd_export *, struct obd_device *);
+        
         /* For quota slave, set proper thread resoure capability */
         int (*quota_enforce) (struct obd_device *, unsigned int);
         
@@ -365,6 +369,18 @@ static inline int lquota_setinfo(quota_interface_t *interface,
         RETURN(rc);
 }
 
+static inline int lquota_clearinfo(quota_interface_t *interface,
+                                   struct obd_export *exp, 
+                                   struct obd_device *obd) 
+{
+        int rc;
+        ENTRY;
+
+        QUOTA_CHECK_OP(interface, clearinfo);
+        rc = QUOTA_OP(interface, clearinfo)(exp, obd);
+        RETURN(rc);
+}
+
 static inline int lquota_enforce(quota_interface_t *interface, 
                                  struct obd_device *obd,
                                  unsigned int ignore)

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index 37befda..45b1ec1 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -2200,6 +2200,8 @@ static int filter_destroy_export(struct obd_export *exp)
                        exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
                        exp, exp->exp_filter_data.fed_pending);
 
+        lquota_clearinfo(filter_quota_interface_ref, exp, exp->exp_obd);
+
         target_destroy_export(exp);
 
         if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid))

diff --git a/lustre/quota/quota_context.c b/lustre/quota/quota_context.c
index 4a692eb..36098c7 100644 (file)
--- a/lustre/quota/quota_context.c
+++ b/lustre/quota/quota_context.c
@@ -586,6 +586,7 @@ schedule_dqacq(struct obd_device *obd,
         struct qunit_data *reqdata;
         struct dqacq_async_args *aa;
         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*reqdata) };
+        struct obd_import *imp = NULL;
         int rc = 0;
         ENTRY;
 
@@ -625,18 +626,38 @@ schedule_dqacq(struct obd_device *obd,
                 RETURN((rc && rc != -EDQUOT) ? rc : rc2);
         }
 
+        spin_lock(&qctxt->lqc_lock);
+        if (!qctxt->lqc_import) {
+                spin_unlock(&qctxt->lqc_lock);
+                QDATA_DEBUG(qdata, "lqc_import is invalid.\n");
+                spin_lock(&qunit_hash_lock);
+                if (wait)
+                        list_del_init(&qw.qw_entry);
+                remove_qunit_nolock(qunit);
+                free_qunit(empty);
+                qunit = NULL;
+                spin_unlock(&qunit_hash_lock);
+                RETURN(-EAGAIN);
+        } else {
+                imp = class_import_get(qctxt->lqc_import);
+        }
+        spin_unlock(&qctxt->lqc_lock);
+
         /* build dqacq/dqrel request */
-        LASSERT(qctxt->lqc_import);
-        req = ptlrpc_prep_req(qctxt->lqc_import, LUSTRE_MDS_VERSION, opc, 2,
+        LASSERT(imp);
+        req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, opc, 2,
                               size, NULL);
         if (!req) {
                 dqacq_completion(obd, qctxt, qdata, -ENOMEM, opc);
+                class_import_put(imp);
                 RETURN(-ENOMEM);
         }
 
-        LASSERT(!should_translate_quota(qctxt->lqc_import) || 
-                qdata->qd_count <= MAX_QUOTA_COUNT32);
-        if (should_translate_quota(qctxt->lqc_import) ||
+        LASSERTF(!should_translate_quota(imp) ||
+                 qdata->qd_count <= MAX_QUOTA_COUNT32,
+                 "qd_count: "LPU64"; should_translate_quota: %d.\n",
+                 qdata->qd_count, should_translate_quota(imp));
+        if (should_translate_quota(imp) ||
             OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT))
         {
                 struct qunit_data_old *reqdata_old, *tmp;
@@ -655,6 +676,7 @@ schedule_dqacq(struct obd_device *obd,
                 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
         }
         ptlrpc_req_set_repsize(req, 2, size);
+        class_import_put(imp);
 
         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
         aa = (struct dqacq_async_args *)&req->rq_async_args;
@@ -763,6 +785,8 @@ qctxt_init(struct lustre_quota_ctxt *qctxt, struct super_block *sb,
         if (rc)
                 RETURN(rc);
 
+        spin_lock_init(&qctxt->lqc_lock);
+        spin_lock(&qctxt->lqc_lock);
         qctxt->lqc_handler = handler;
         qctxt->lqc_sb = sb;
         qctxt->lqc_import = NULL;
@@ -774,6 +798,7 @@ qctxt_init(struct lustre_quota_ctxt *qctxt, struct super_block *sb,
         qctxt->lqc_iunit_sz = default_iunit_sz;
         qctxt->lqc_itune_sz = default_iunit_sz * default_itune_ratio / 100;
         qctxt->lqc_limit_sz = default_limit_sz;
+        spin_unlock(&qctxt->lqc_lock);
 
         RETURN(0);
 }

diff --git a/lustre/quota/quota_interface.c b/lustre/quota/quota_interface.c
index eaa4b46..8acd727 100644 (file)
--- a/lustre/quota/quota_interface.c
+++ b/lustre/quota/quota_interface.c
@@ -340,7 +340,9 @@ static int filter_quota_setinfo(struct obd_export *exp, struct obd_device *obd)
         struct obd_import *imp;
 
         /* setup the quota context import */
+        spin_lock(&obd->u.obt.obt_qctxt.lqc_lock);
         obd->u.obt.obt_qctxt.lqc_import = exp->exp_imp_reverse;
+        spin_unlock(&obd->u.obt.obt_qctxt.lqc_lock);
 
         /* make imp's connect flags equal relative exp's connect flags 
          * adding it to avoid the scan export list
@@ -354,6 +356,22 @@ static int filter_quota_setinfo(struct obd_export *exp, struct obd_device *obd)
         qslave_start_recovery(obd, &obd->u.obt.obt_qctxt);
         return 0;
 }
+
+static int filter_quota_clearinfo(struct obd_export *exp, struct obd_device *obd)
+{
+        struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
+
+        /* when exp->exp_imp_reverse is destroyed, the corresponding lqc_import
+         * should be invalid b=12374 */
+        if (qctxt->lqc_import == exp->exp_imp_reverse) {
+                spin_lock(&qctxt->lqc_lock);
+                qctxt->lqc_import = NULL;
+                spin_unlock(&qctxt->lqc_lock);
+        }
+
+        return 0;
+}
+
 static int filter_quota_enforce(struct obd_device *obd, unsigned int ignore)
 {
         ENTRY;
@@ -743,6 +761,7 @@ quota_interface_t filter_quota_interface = {
         .quota_check    = target_quota_check,
         .quota_ctl      = filter_quota_ctl,
         .quota_setinfo  = filter_quota_setinfo,
+        .quota_clearinfo = filter_quota_clearinfo,
         .quota_enforce  = filter_quota_enforce,
         .quota_getflag  = filter_quota_getflag,
         .quota_acquire  = filter_quota_acquire,