b=21147 several fixes and enhancements for quota

1) allow consecutive quotaons
2) a locking fix
3) call build_lqs from a single point in the code

i=Johann Lombardi
i=ZhiYong Tian

diff --git a/lustre/quota/lproc_quota.c b/lustre/quota/lproc_quota.c
index d911365..6f0114a 100644 (file)
--- a/lustre/quota/lproc_quota.c
+++ b/lustre/quota/lproc_quota.c
@@ -202,86 +202,88 @@ int lprocfs_quota_rd_type(char *page, char **start, off_t off, int count,
 }
 EXPORT_SYMBOL(lprocfs_quota_rd_type);
 
-static int auto_quota_on(struct obd_device *obd, int type,
-                         struct super_block *sb, int is_master)
+/*
+ * generic_quota_on is very lazy and tolerant about current quota settings
+ * @global means to turn on quotas on each OST additionally to local quotas;
+ * should not be called from filter_quota_ctl on MDS nodes (as it starts
+ * admin quotas on MDS nodes).
+ */
+int generic_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl, int global)
 {
-        struct obd_quotactl *oqctl;
-        struct lvfs_run_ctxt saved;
-        int rc = 0, rc1 = 0, id;
         struct obd_device_target *obt = &obd->u.obt;
-        struct lustre_quota_ctxt *qctxt = &obt->obt_qctxt;
-        struct mds_obd *mds = NULL;
-        ENTRY;
-
-        LASSERT(type == USRQUOTA || type == GRPQUOTA || type == UGQUOTA);
-
-        OBD_ALLOC_PTR(oqctl);
-        if (!oqctl)
-                RETURN(-ENOMEM);
+        struct lvfs_run_ctxt saved;
+        int id, is_master, rc = 0, local; /* means we need a local quotaon */
 
         cfs_down(&obt->obt_quotachecking);
-        id = UGQUOTA2LQC(type);
-        /* quota already turned on */
-        if ((obt->obt_qctxt.lqc_flags & id) == id)
-                GOTO(out, rc);
+        id = UGQUOTA2LQC(oqctl->qc_type);
+        local = (obt->obt_qctxt.lqc_flags & id) != id;
 
-        oqctl->qc_type = type;
         oqctl->qc_cmd = Q_QUOTAON;
         oqctl->qc_id = obt->obt_qfmt;
 
-        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-
+        is_master = !strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME);
         if (is_master) {
-                mds = &obd->u.mds;
-                cfs_down(&mds->mds_qonoff_sem);
-                /* turn on cluster wide quota */
-                rc1 = mds_admin_quota_on(obd, oqctl);
-                if (rc1 && rc1 != -EALREADY) {
-                        CDEBUG_LIMIT(rc1 == -ENOENT ? D_QUOTA : D_ERROR,
-                                     "%s: auto-enable admin quota failed with "
-                                     "rc=%d\n", obd->obd_name, rc1);
-                        GOTO(out_ctxt, rc1);
+                cfs_down(&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",
+                                       rc);
                 }
         }
 
-        /* turn on local quota */
-        rc = fsfilt_quotactl(obd, sb, oqctl);
-        if (!rc) {
-                obt->obt_qctxt.lqc_flags |= UGQUOTA2LQC(type);
-                build_lqs(obd);
-        } else if (rc == -EBUSY && quota_is_on(qctxt, oqctl)) {
-                CWARN("%s: mds local quota[%d] is on already\n",
-                      obd->obd_name, oqctl->qc_type);
-                rc = -EALREADY;
-        } else {
-                CDEBUG_LIMIT(rc == -ENOENT ? D_QUOTA : D_ERROR,
-                             "%s: auto-enable local quota failed with rc=%d\n",
-                             obd->obd_name, rc);
-                if (rc1 == -EALREADY) {
-                        oqctl->qc_cmd = Q_QUOTAOFF;
-                        mds_admin_quota_off(obd, oqctl);
+        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)
+                                        CERROR("%s: %s quotaon failed with"
+                                               " rc=%d\n", obd->obd_name,
+                                               global ? "global" : "local", rc);
+                        } else {
+                                obt->obt_qctxt.lqc_flags |= UGQUOTA2LQC(oqctl->qc_type);
+                                build_lqs(obd);
+                        }
+                        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                 }
-                if (rc == -ENOENT)
-                        CWARN("%s: quotaon failed because quota files don't "
-                              "exist, please run quotacheck firstly\n",
-                              obd->obd_name);
 
-                GOTO(out_ctxt, rc);
+                if (rc == 0 && global && is_master)
+                        rc = obd_quotactl(obd->u.mds.mds_osc_exp, oqctl);
         }
 
-        EXIT;
-
-out_ctxt:
-        if (mds != NULL)
-                cfs_up(&mds->mds_qonoff_sem);
-        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
+        if (is_master)
+                cfs_up(&obd->u.mds.mds_qonoff_sem);
 
-out:
         cfs_up(&obt->obt_quotachecking);
-        OBD_FREE_PTR(oqctl);
+
         return rc;
 }
 
+static int auto_quota_on(struct obd_device *obd, int type)
+{
+        struct obd_quotactl *oqctl;
+        int rc;
+        ENTRY;
+
+        LASSERT(type == USRQUOTA || type == GRPQUOTA || type == UGQUOTA);
+
+        OBD_ALLOC_PTR(oqctl);
+        if (!oqctl)
+                RETURN(-ENOMEM);
+
+        oqctl->qc_type = type;
+
+        rc = generic_quota_on(obd, oqctl, 0);
+
+        OBD_FREE_PTR(oqctl);
+        RETURN(rc);
+}
+
 int lprocfs_quota_wr_type(struct file *file, const char *buffer,
                           unsigned long count, void *data)
 {
@@ -323,7 +325,7 @@ int lprocfs_quota_wr_type(struct file *file, const char *buffer,
         }
 
         if (type != 0) {
-                int rc = auto_quota_on(obd, type - 1, obt->obt_sb, is_mds);
+                int rc = auto_quota_on(obd, type - 1);
 
                 if (rc && rc != -EALREADY && rc != -ENOENT)
                         return rc;

diff --git a/lustre/quota/quota_ctl.c b/lustre/quota/quota_ctl.c
index 3650a67..91fee1d 100644 (file)
--- a/lustre/quota/quota_ctl.c
+++ b/lustre/quota/quota_ctl.c
@@ -143,8 +143,11 @@ int filter_quota_ctl(struct obd_device *unused, struct obd_export *exp,
 
         cfs_gettimeofday(&work_start);
         switch (oqctl->qc_cmd) {
-        case Q_FINVALIDATE:
         case Q_QUOTAON:
+                oqctl->qc_id = obt->obt_qfmt;
+                rc = generic_quota_on(obd, oqctl, 0);
+                break;
+        case Q_FINVALIDATE:
         case Q_QUOTAOFF:
                 cfs_down(&obt->obt_quotachecking);
                 if (oqctl->qc_cmd == Q_FINVALIDATE &&
@@ -172,20 +175,9 @@ int filter_quota_ctl(struct obd_device *unused, struct obd_export *exp,
                 if (oqctl->qc_stat == QUOTA_RECOVERING)
                         quota_unbarrier(handle);
 
-                if (oqctl->qc_cmd == Q_QUOTAON || oqctl->qc_cmd == Q_QUOTAOFF ||
+                if (oqctl->qc_cmd == Q_QUOTAOFF ||
                     oqctl->qc_cmd == Q_FINVALIDATE) {
-                        if (oqctl->qc_cmd == Q_QUOTAON) {
-                                if (!rc) {
-                                        obt->obt_qctxt.lqc_flags |=
-                                                UGQUOTA2LQC(oqctl->qc_type);
-                                        /* when quotaon, create lqs for every
-                                         * quota uid/gid b=18574 */
-                                        build_lqs(obd);
-                                } else if (rc == -EBUSY &&
-                                         quota_is_on(qctxt, oqctl)) {
-                                                rc = -EALREADY;
-                                }
-                        } else if (oqctl->qc_cmd == Q_QUOTAOFF) {
+                        if (oqctl->qc_cmd == Q_QUOTAOFF) {
                                 if (!rc)
                                         obt->obt_qctxt.lqc_flags &=
                                                 ~UGQUOTA2LQC(oqctl->qc_type);

diff --git a/lustre/quota/quota_internal.h b/lustre/quota/quota_internal.h
index 4a046e5..babb923 100644 (file)
--- a/lustre/quota/quota_internal.h
+++ b/lustre/quota/quota_internal.h
@@ -146,6 +146,7 @@ int mds_get_obd_quota(struct obd_device *obd, struct obd_quotactl *oqctl);
 int dquot_create_oqaq(struct lustre_quota_ctxt *qctxt, struct lustre_dquot
                       *dquot, __u32 ost_num, __u32 mdt_num, int type,
                       struct quota_adjust_qunit *oqaq);
+int generic_quota_on(struct obd_device *, struct obd_quotactl *, int);
 #endif
 
 /* quota_ctl.c */

diff --git a/lustre/quota/quota_master.c b/lustre/quota/quota_master.c
index cc90f76..8d58de0 100644 (file)
--- a/lustre/quota/quota_master.c
+++ b/lustre/quota/quota_master.c
@@ -783,19 +783,12 @@ int mds_admin_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
         for (i = 0; i < MAXQUOTAS; i++) {
                 struct file *fp;
 
-                if (!Q_TYPESET(oqctl, i))
+                if (!Q_TYPESET(oqctl, i) || qinfo->qi_files[i] != NULL)
                         continue;
 
                 LASSERT(strlen(quotafile[i])
                         + sizeof(prefix) <= sizeof(name));
                 sprintf(name, "%s%s", prefix, quotafile[i]);
-
-                if (qinfo->qi_files[i] != NULL) {
-                        CWARN("quota[%d] is on already\n", i);
-                        rc1 = -EALREADY;
-                        continue;
-                }
-
                 fp = filp_open(name, O_RDWR, 0);
                 if (IS_ERR(fp) || !S_ISREG(fp->f_dentry->d_inode->i_mode)) {
                         rc = IS_ERR(fp) ? PTR_ERR(fp) : -EINVAL;
@@ -824,26 +817,9 @@ int mds_admin_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
         RETURN(rc ? : rc1);
 }
 
-int mds_admin_quota_off(struct obd_device *obd,
-                        struct obd_quotactl *oqctl)
-{
-        struct mds_obd *mds = &obd->u.mds;
-        struct lustre_quota_info *qinfo = &mds->mds_quota_info;
-        int rc;
-        ENTRY;
-
-        /* close admin quota files */
-        rc = close_quota_files(oqctl, qinfo);
-        RETURN(rc);
-}
-
 int mds_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
 {
-        struct mds_obd *mds = &obd->u.mds;
-        struct obd_device_target *obt = &obd->u.obt;
-        struct lustre_quota_ctxt *qctxt = &obt->obt_qctxt;
-        struct lvfs_run_ctxt saved;
-        int rc = 0, rc1 = 0, rc2 = 0;
+        int rc;
         ENTRY;
 
         if (oqctl->qc_type != USRQUOTA &&
@@ -851,57 +827,26 @@ int mds_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
             oqctl->qc_type != UGQUOTA)
                 RETURN(-EINVAL);
 
-        cfs_down(&obt->obt_quotachecking);
-        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-        cfs_down(&mds->mds_qonoff_sem);
-        rc2 = mds_admin_quota_on(obd, oqctl);
-        if (rc2 && rc2 != -EALREADY) {
-                CWARN("mds quota[%d] is failed to be on for %d\n", oqctl->qc_type, rc2);
-                GOTO(out, rc2);
-        }
+        rc = generic_quota_on(obd, oqctl, 1);
 
-        rc1 = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
-        if (!rc1) {
-                qctxt->lqc_flags |= UGQUOTA2LQC(oqctl->qc_type);
-                /* when quotaon, create lqs for every quota uid/gid b=18574 */
-                build_lqs(obd);
-        } else if (rc1 == -EBUSY && quota_is_on(qctxt, oqctl)) {
-                CWARN("mds local quota[%d] is on already\n", oqctl->qc_type);
-                rc1 = -EALREADY;
-        } else {
-                if (rc2 != -EALREADY) {
-                        CWARN("mds local quota[%d] is failed to be on for %d\n",
-                              oqctl->qc_type, rc1);
-                        oqctl->qc_cmd = Q_QUOTAOFF;
-                        mds_admin_quota_off(obd, oqctl);
-                        oqctl->qc_cmd = Q_QUOTAON;
-                }
-                GOTO(out, rc1);
-        }
+        RETURN(rc);
+}
 
-        rc = obd_quotactl(mds->mds_osc_exp, oqctl);
-        if (rc && rc != -EALREADY) {
-                CWARN("mds remote quota[%d] is failed to be on for %d\n",
-                      oqctl->qc_type, rc);
-                oqctl->qc_cmd = Q_QUOTAOFF;
-                if (rc2 != -EALREADY)
-                        mds_admin_quota_off(obd, oqctl);
-                if (rc1 != -EALREADY) {
-                        fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
-                        qctxt->lqc_flags &= ~UGQUOTA2LQC(oqctl->qc_type);
-                }
-                oqctl->qc_cmd = Q_QUOTAON;
-        }
 
-        EXIT;
+int mds_admin_quota_off(struct obd_device *obd,
+                        struct obd_quotactl *oqctl)
+{
+        struct mds_obd *mds = &obd->u.mds;
+        struct lustre_quota_info *qinfo = &mds->mds_quota_info;
+        int rc;
+        ENTRY;
 
-out:
-        cfs_up(&mds->mds_qonoff_sem);
-        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
-        cfs_up(&obt->obt_quotachecking);
-        return rc ? : (rc1 ? : rc2);
+        /* close admin quota files */
+        rc = close_quota_files(oqctl, qinfo);
+        RETURN(rc);
 }
 
+
 /* with obt->obt_quotachecking held */
 int do_mds_quota_off(struct obd_device *obd, struct obd_quotactl *oqctl)
 {