b=22634 hold "mds_qonoff_sem" when call "lustre_read_quota()", and check parameter...

1) replace "cfs_semaphore_t" with "cfs_rw_semaphore_t" for "mds_qonoff_sem" to enhance the parallel processing of quota related operations
2) hold "mds_qonoff_sem" when call "lustre_read_quota()", and check parameter properly in such function

i=robert.read
i=landen

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index e79f512..040fe05 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -529,7 +529,7 @@ struct mds_obd {
 
 
         struct lustre_quota_info         mds_quota_info;
-        cfs_semaphore_t                  mds_qonoff_sem;
+        cfs_rw_semaphore_t               mds_qonoff_sem;
         cfs_semaphore_t                  mds_health_sem;
         unsigned long                    mds_fl_user_xattr:1,
                                          mds_fl_acl:1,

diff --git a/lustre/lvfs/fsfilt_ext3.c b/lustre/lvfs/fsfilt_ext3.c
index bde0093..67bddb9 100644 (file)
--- a/lustre/lvfs/fsfilt_ext3.c
+++ b/lustre/lvfs/fsfilt_ext3.c
@@ -2219,6 +2219,12 @@ ssize_t lustre_read_quota(struct file *f, struct inode *inode, int type,
         loff_t p = pos;
         int rc;
 
+        if (!f && !inode) {
+                CERROR("lustre_read_quota failed for no quota file!\n");
+                libcfs_debug_dumpstack(NULL);
+                return -EINVAL;
+        }
+
         /* Support for both adm and op quota files must be provided */
         if (f) {
                 rc = fsfilt_ext3_read_record(f, buf, count, &p);

diff --git a/lustre/lvfs/lustre_quota_fmt.c b/lustre/lvfs/lustre_quota_fmt.c
index d169bb7..26944fc 100644 (file)
--- a/lustre/lvfs/lustre_quota_fmt.c
+++ b/lustre/lvfs/lustre_quota_fmt.c
@@ -103,15 +103,8 @@ int check_quota_file(struct file *f, struct inode *inode, int type,
         static const uint quota_magics[] = LUSTRE_INITQMAGICS;
         const uint *quota_versions = lustre_initqversions[version];
 
-        if (!f && !inode) {
-                CERROR("check_quota_file failed!\n");
-                libcfs_debug_dumpstack(NULL);
-                return -EINVAL;
-        }
-
         size = lustre_read_quota(f, inode, type, (char *)&dqhead,
                                  sizeof(struct lustre_disk_dqheader), 0);
-
         if (size != sizeof(struct lustre_disk_dqheader))
                 return -EINVAL;
         if (le32_to_cpu(dqhead.dqh_magic) != quota_magics[type] ||

diff --git a/lustre/quota/lproc_quota.c b/lustre/quota/lproc_quota.c
index 6f0114a..49d16ab 100644 (file)
--- a/lustre/quota/lproc_quota.c
+++ b/lustre/quota/lproc_quota.c
@@ -215,6 +215,7 @@ int generic_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl, int glo
         int id, is_master, rc = 0, local; /* means we need a local quotaon */
 
         cfs_down(&obt->obt_quotachecking);
+        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         id = UGQUOTA2LQC(oqctl->qc_type);
         local = (obt->obt_qctxt.lqc_flags & id) != id;
 
@@ -223,12 +224,10 @@ int generic_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl, int glo
 
         is_master = !strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME);
         if (is_master) {
-                cfs_down(&obd->u.mds.mds_qonoff_sem);
+                cfs_down_write(&obd->u.mds.mds_qonoff_sem);
                 if (local) {
-                        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                         /* turn on cluster wide quota */
                         rc = mds_admin_quota_on(obd, oqctl);
-                        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                         if (rc && rc != -ENOENT)
                                 CERROR("%s: %s admin quotaon failed. rc=%d\n",
                                        obd->obd_name, global ? "global":"local",
@@ -238,7 +237,6 @@ int generic_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl, int glo
 
         if (rc == 0) {
                 if (local) {
-                        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                         rc = fsfilt_quotactl(obd, obt->obt_sb, oqctl);
                         if (rc) {
                                 if (rc != -ENOENT)
@@ -249,7 +247,6 @@ int generic_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl, int glo
                                 obt->obt_qctxt.lqc_flags |= UGQUOTA2LQC(oqctl->qc_type);
                                 build_lqs(obd);
                         }
-                        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                 }
 
                 if (rc == 0 && global && is_master)
@@ -257,8 +254,9 @@ int generic_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl, int glo
         }
 
         if (is_master)
-                cfs_up(&obd->u.mds.mds_qonoff_sem);
+                cfs_up_write(&obd->u.mds.mds_qonoff_sem);
 
+        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         cfs_up(&obt->obt_quotachecking);
 
         return rc;

diff --git a/lustre/quota/quota_check.c b/lustre/quota/quota_check.c
index f00c7ce..17f20d8 100644 (file)
--- a/lustre/quota/quota_check.c
+++ b/lustre/quota/quota_check.c
@@ -177,6 +177,8 @@ int target_quota_check(struct obd_device *obd, struct obd_export *exp,
         rc = cfs_kernel_thread(target_quotacheck_thread, qta,
                                CLONE_VM|CLONE_FILES);
         if (rc >= 0) {
+                /* target_quotacheck_thread will drop the ref on exp and release
+                 * obt_quotachecking */
                 CDEBUG(D_INFO, "%s: target_quotacheck_thread: %d\n",
                        obd->obd_name, rc);
                 RETURN(0);

diff --git a/lustre/quota/quota_interface.c b/lustre/quota/quota_interface.c
index b29ab8e..11825d5 100644 (file)
--- a/lustre/quota/quota_interface.c
+++ b/lustre/quota/quota_interface.c
@@ -655,7 +655,7 @@ static int mds_quota_setup(struct obd_device *obd)
         mds->mds_quota_info.qi_version = LUSTRE_QUOTA_V2;
         cfs_sema_init(&obt->obt_quotachecking, 1);
         /* initialize quota master and quota context */
-        cfs_sema_init(&mds->mds_qonoff_sem, 1);
+        cfs_init_rwsem(&mds->mds_qonoff_sem);
         rc = qctxt_init(obd, dqacq_handler);
         if (rc) {
                 CERROR("%s: initialize quota context failed! (rc:%d)\n",
@@ -704,9 +704,9 @@ static int mds_quota_fs_cleanup(struct obd_device *obd)
         memset(&oqctl, 0, sizeof(oqctl));
         oqctl.qc_type = UGQUOTA;
 
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
         mds_admin_quota_off(obd, &oqctl);
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
         RETURN(0);
 }
 

diff --git a/lustre/quota/quota_master.c b/lustre/quota/quota_master.c
index c72e477..c9249ca 100644 (file)
--- a/lustre/quota/quota_master.c
+++ b/lustre/quota/quota_master.c
@@ -265,10 +265,12 @@ int dqacq_adjust_qunit_sz(struct obd_device *obd, qid_t id, int type,
         ENTRY;
 
         LASSERT(mds);
+        cfs_down_read(&mds->mds_qonoff_sem);
         dquot = lustre_dqget(obd, info, id, type);
         if (IS_ERR(dquot))
                 RETURN(PTR_ERR(dquot));
 
+        cfs_up_read(&mds->mds_qonoff_sem);
         OBD_ALLOC_PTR(oqaq);
         if (!oqaq)
                 GOTO(out, rc = -ENOMEM);
@@ -352,14 +354,16 @@ int dqacq_handler(struct obd_device *obd, struct qunit_data *qdata, int opc)
         if (OBD_FAIL_CHECK(OBD_FAIL_OBD_DQACQ))
                 RETURN(-EIO);
 
+        cfs_down_write(&mds->mds_qonoff_sem);
         dquot = lustre_dqget(obd, info, qdata->qd_id, QDATA_IS_GRP(qdata));
-        if (IS_ERR(dquot))
+        if (IS_ERR(dquot)) {
+                cfs_up_write(&mds->mds_qonoff_sem);
                 RETURN(PTR_ERR(dquot));
+        }
 
         DQUOT_DEBUG(dquot, "get dquot in dqacq_handler\n");
         QINFO_DEBUG(dquot->dq_info, "get dquot in dqadq_handler\n");
 
-        cfs_down(&mds->mds_qonoff_sem);
         cfs_down(&dquot->dq_sem);
 
         if (dquot->dq_status & DQ_STATUS_RECOVERY) {
@@ -433,12 +437,11 @@ int dqacq_handler(struct obd_device *obd, struct qunit_data *qdata, int opc)
         EXIT;
 out:
         cfs_up(&dquot->dq_sem);
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
         lustre_dqput(dquot);
         if (rc != -EDQUOT)
                 dqacq_adjust_qunit_sz(obd, qdata->qd_id, QDATA_IS_GRP(qdata),
                                       QDATA_IS_BLK(qdata));
-
         lqs = quota_search_lqs(LQS_KEY(QDATA_IS_GRP(qdata), qdata->qd_id),
                                qctxt, 0);
         if (lqs == NULL || IS_ERR(lqs)) {
@@ -597,7 +600,7 @@ int mds_quota_invalidate(struct obd_device *obd, struct obd_quotactl *oqctl)
 
         cfs_down(&obt->obt_quotachecking);
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
 
         for (i = 0; i < MAXQUOTAS; i++) {
                 struct file *fp;
@@ -625,7 +628,7 @@ int mds_quota_invalidate(struct obd_device *obd, struct obd_quotactl *oqctl)
                         filp_close(fp, 0);
         }
 
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         cfs_up(&obt->obt_quotachecking);
         RETURN(rc ? : rc1);
@@ -648,14 +651,14 @@ int mds_quota_finvalidate(struct obd_device *obd, struct obd_quotactl *oqctl)
         if (obt->obt_qctxt.lqc_flags & UGQUOTA2LQC(oqctl->qc_type))
                 GOTO(out, rc = -EBUSY);
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
 
         oqctl->qc_cmd = Q_FINVALIDATE;
         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
         if (!rc)
                 rc = obd_quotactl(mds->mds_osc_exp, oqctl);
 
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 out:
         cfs_up(&obt->obt_quotachecking);
@@ -675,8 +678,7 @@ int init_admin_quotafiles(struct obd_device *obd, struct obd_quotactl *oqctl)
         LASSERT(qinfo->qi_version == LUSTRE_QUOTA_V2);
 
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
 
         for (i = 0; i < MAXQUOTAS && !rc; i++) {
                 struct file *fp;
@@ -707,7 +709,7 @@ int init_admin_quotafiles(struct obd_device *obd, struct obd_quotactl *oqctl)
                         }
                         qinfo->qi_files[i] = fp;
                         rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_CHK);
-                        qinfo->qi_files[i] = 0;
+                        qinfo->qi_files[i] = NULL;
                         filp_close(fp, 0);
                 }
                 else
@@ -745,8 +747,8 @@ int init_admin_quotafiles(struct obd_device *obd, struct obd_quotactl *oqctl)
                 filp_close(fp, 0);
                 qinfo->qi_files[i] = NULL;
         }
-        cfs_up(&mds->mds_qonoff_sem);
 
+        cfs_up_write(&mds->mds_qonoff_sem);
         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         RETURN(rc);
 }
@@ -868,7 +870,7 @@ int do_mds_quota_off(struct obd_device *obd, struct obd_quotactl *oqctl)
                 RETURN(-EINVAL);
 
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
         /* close admin quota files */
         rc2 = mds_admin_quota_off(obd, oqctl);
         if (rc2 && rc2 != -EALREADY) {
@@ -909,7 +911,7 @@ int do_mds_quota_off(struct obd_device *obd, struct obd_quotactl *oqctl)
         EXIT;
 
 out:
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         return rc ? : (rc1 ? : rc2);
 }
@@ -938,7 +940,7 @@ int mds_set_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
             oqctl->qc_type != GRPQUOTA)
                 RETURN(-EINVAL);
 
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
                 CWARN("quota[%u] is off\n", oqctl->qc_type);
                 GOTO(out, rc = -ESRCH);
@@ -952,7 +954,7 @@ int mds_set_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
         EXIT;
 
 out:
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
         return rc;
 }
 
@@ -968,7 +970,7 @@ int mds_get_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
             oqctl->qc_type != GRPQUOTA)
                 RETURN(-EINVAL);
 
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_read(&mds->mds_qonoff_sem);
         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
                 CWARN("quota[%u] is off\n", oqctl->qc_type);
                 GOTO(out, rc = -ESRCH);
@@ -980,7 +982,7 @@ int mds_get_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
         EXIT;
 
 out:
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_read(&mds->mds_qonoff_sem);
         return rc;
 }
 
@@ -1284,7 +1286,8 @@ int mds_set_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
         OBD_ALLOC_PTR(oqaq);
         if (!oqaq)
                 RETURN(-ENOMEM);
-        cfs_down(&mds->mds_qonoff_sem);
+
+        cfs_down_write(&mds->mds_qonoff_sem);
         init_oqaq(oqaq, qctxt, oqctl->qc_id, oqctl->qc_type);
 
         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
@@ -1373,8 +1376,7 @@ int mds_set_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
                 goto out;
         }
 
-        cfs_up(&mds->mds_qonoff_sem);
-
+        cfs_up_write(&mds->mds_qonoff_sem);
         adjust_lqs(obd, oqaq);
 
         orig_set = ihardlimit || isoftlimit;
@@ -1404,7 +1406,7 @@ int mds_set_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
         }
 
 revoke_out:
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_write(&mds->mds_qonoff_sem);
         cfs_down(&dquot->dq_sem);
         if (rc) {
                 /* cancel previous setting */
@@ -1425,7 +1427,7 @@ out:
         lustre_dqput(dquot);
         EXIT;
 out_sem:
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_write(&mds->mds_qonoff_sem);
 
         if (oqaq)
                 OBD_FREE_PTR(oqaq);
@@ -1486,7 +1488,7 @@ int mds_get_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
             oqctl->qc_type != GRPQUOTA)
                 RETURN(-EINVAL);
 
-        cfs_down(&mds->mds_qonoff_sem);
+        cfs_down_read(&mds->mds_qonoff_sem);
         dqblk->dqb_valid = 0;
         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
                 CWARN("quota[%u] is off\n", oqctl->qc_type);
@@ -1508,16 +1510,16 @@ int mds_get_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
         cfs_up(&dquot->dq_sem);
 
         lustre_dqput(dquot);
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_read(&mds->mds_qonoff_sem);
 
         /* the usages in admin quota file is inaccurate */
         dqblk->dqb_curinodes = 0;
         dqblk->dqb_curspace = 0;
         rc = mds_get_space(obd, oqctl);
-        EXIT;
-        return rc;
+        RETURN(rc);
+
 out:
-        cfs_up(&mds->mds_qonoff_sem);
+        cfs_up_read(&mds->mds_qonoff_sem);
         return rc;
 }
 
@@ -1569,12 +1571,16 @@ dquot_recovery(struct obd_device *obd, unsigned int id, unsigned short type)
 
         cfs_up(&dquot->dq_sem);
 
+        /* release mds_qonoff_sem during obd_quotactl ops here */
+        cfs_up_write(&mds->mds_qonoff_sem);
+
         /* get real bhardlimit from all slaves. */
         qctl->qc_cmd = Q_GETOQUOTA;
         qctl->qc_type = type;
         qctl->qc_id = id;
         qctl->qc_stat = QUOTA_RECOVERING;
         rc = obd_quotactl(mds->mds_osc_exp, qctl);
+        cfs_down_write(&mds->mds_qonoff_sem);
         if (rc)
                 GOTO(out, rc);
         total_limits = qctl->qc_dqblk.dqb_bhardlimit;
@@ -1586,7 +1592,6 @@ dquot_recovery(struct obd_device *obd, unsigned int id, unsigned short type)
         total_limits += qctl->qc_dqblk.dqb_bhardlimit;
 
         /* amend the usage of the administrative quotafile */
-        cfs_down(&mds->mds_qonoff_sem);
         cfs_down(&dquot->dq_sem);
 
         dquot->dq_dqb.dqb_curspace = total_limits << QUOTABLOCK_BITS;
@@ -1596,7 +1601,6 @@ dquot_recovery(struct obd_device *obd, unsigned int id, unsigned short type)
                 CERROR("write dquot failed! (rc:%d)\n", rc);
 
         cfs_up(&dquot->dq_sem);
-        cfs_up(&mds->mds_qonoff_sem);
         EXIT;
 out:
         cfs_down(&dquot->dq_sem);
@@ -1633,20 +1637,17 @@ static int qmaster_recovery_main(void *arg)
 
         cfs_complete(&data->comp);
 
+        cfs_down_write(&mds->mds_qonoff_sem);
         for (type = USRQUOTA; type < MAXQUOTAS; type++) {
                 cfs_list_t id_list;
                 struct dquot_id *dqid, *tmp;
 
-                cfs_down(&mds->mds_qonoff_sem);
-                if (qinfo->qi_files[type] == NULL) {
-                        cfs_up(&mds->mds_qonoff_sem);
+                if (qinfo->qi_files[type] == NULL)
                         continue;
-                }
+
                 CFS_INIT_LIST_HEAD(&id_list);
                 rc = fsfilt_qids(obd, qinfo->qi_files[type], NULL, type,
                                  &id_list);
-                cfs_up(&mds->mds_qonoff_sem);
-
                 if (rc)
                         CERROR("error get ids from admin quotafile.(%d)\n", rc);
 
@@ -1664,6 +1665,7 @@ free:
                         OBD_FREE_PTR(dqid);
                 }
         }
+        cfs_up_write(&mds->mds_qonoff_sem);
         class_decref(mds->mds_osc_obd, "qmaster_recovd_lov", mds->mds_osc_obd);
         class_decref(obd, "qmaster_recovd_mds", obd);
         RETURN(rc);