quota_interface.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 948 行 · 第 1/3 页

                memset(oqctl, 0, sizeof(*oqctl));

                oqctl->qc_cmd = Q_GETQUOTA;
                oqctl->qc_type = cnt;
                oqctl->qc_id = (cnt == USRQUOTA) ? oa->o_uid : oa->o_gid;
                err = fsfilt_quotactl(obd, obt->obt_sb, oqctl);
                if (err) {
                        if (!rc)
                                rc = err;
                        continue;
                }

                /* set over quota flags for a uid/gid */
                oa->o_valid |= (cnt == USRQUOTA) ?
                               OBD_MD_FLUSRQUOTA : OBD_MD_FLGRPQUOTA;
                if (oqctl->qc_dqblk.dqb_bhardlimit &&
                   (toqb(oqctl->qc_dqblk.dqb_curspace) >=
                    oqctl->qc_dqblk.dqb_bhardlimit))
                        oa->o_flags |= (cnt == USRQUOTA) ?
                                OBD_FL_NO_USRQUOTA : OBD_FL_NO_GRPQUOTA;
        }
        OBD_FREE_PTR(oqctl);
        RETURN(rc);
}

static int filter_quota_acquire(struct obd_device *obd, unsigned int uid,
                                unsigned int gid)
{
        struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
        int rc;
        ENTRY;

        rc = qctxt_adjust_qunit(obd, qctxt, uid, gid, LQUOTA_FLAGS_BLK, 1);
        RETURN(rc);
}

/* check whether the left quota of certain uid and gid can satisfy a block_write
 * or inode_create rpc. When need to acquire quota, return QUOTA_RET_ACQUOTA */
static int quota_check_common(struct obd_device *obd, unsigned int uid,
                              unsigned int gid, int count, int cycle, int isblk)
{
        struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
        int i;
        __u32 id[MAXQUOTAS] = { uid, gid };
        struct qunit_data qdata[MAXQUOTAS];
        int rc = 0, rc2[2] = { 0, 0 };
        ENTRY;

        CLASSERT(MAXQUOTAS < 4);
        if (!sb_any_quota_enabled(qctxt->lqc_sb))
                RETURN(rc);

        for (i = 0; i < MAXQUOTAS; i++) {
                struct lustre_qunit_size *lqs = NULL;

                qdata[i].qd_id = id[i];
                qdata[i].qd_flags = i;
                if (isblk)
                        QDATA_SET_BLK(&qdata[i]);
                qdata[i].qd_count = 0;

                /* ignore root user */
                if (qdata[i].qd_id == 0 && !QDATA_IS_GRP(&qdata[i]))
                        continue;

                quota_search_lqs(&qdata[i], NULL, qctxt, &lqs);
                if (!lqs)
                        continue;

                rc2[i] = compute_remquota(obd, qctxt, &qdata[i], isblk);
                spin_lock(&lqs->lqs_lock);
                if (!cycle) {
                        rc = QUOTA_RET_INC_PENDING;
                        if (isblk)
                                lqs->lqs_bwrite_pending += count;
                        else
                                lqs->lqs_iwrite_pending += count;
                }

                CDEBUG(D_QUOTA, "write pending: %lu, qd_count: "LPU64".\n",
                       isblk ? lqs->lqs_bwrite_pending : lqs->lqs_iwrite_pending,
                       qdata[i].qd_count);
                if (rc2[i] == QUOTA_RET_OK) {
                        if (isblk && qdata[i].qd_count <
                            lqs->lqs_bwrite_pending * CFS_PAGE_SIZE)
                                rc2[i] = QUOTA_RET_ACQUOTA;
                        if (!isblk && qdata[i].qd_count <
                            lqs->lqs_iwrite_pending)
                                rc2[i] = QUOTA_RET_ACQUOTA;
                }

                spin_unlock(&lqs->lqs_lock);

                /* When cycle is zero, lqs_*_pending will be changed. We will
                 * get reference of the lqs here and put reference of lqs in
                 * quota_pending_commit b=14784 */
                if (!cycle)
                        lqs_getref(lqs);

                /* this is for quota_search_lqs */
                lqs_putref(lqs);
        }

        if (rc2[0] == QUOTA_RET_ACQUOTA || rc2[1] == QUOTA_RET_ACQUOTA)
                RETURN(rc | QUOTA_RET_ACQUOTA);
        else
                RETURN(rc);
}

static int quota_chk_acq_common(struct obd_device *obd, unsigned int uid,
                                unsigned int gid, int count, int *pending,
                                int isblk, quota_acquire acquire)
{
        int rc = 0, cycle = 0, count_err = 0;
        ENTRY;

        /* Unfortunately, if quota master is too busy to handle the
         * pre-dqacq in time and quota hash on ost is used up, we
         * have to wait for the completion of in flight dqacq/dqrel,
         * in order to get enough quota for write b=12588 */
        while ((rc = quota_check_common(obd, uid, gid, count, cycle, isblk)) &
               QUOTA_RET_ACQUOTA) {

                if (rc & QUOTA_RET_INC_PENDING)
                        *pending = 1;

                cycle++;
                if (isblk)
                        OBD_FAIL_TIMEOUT(OBD_FAIL_OST_HOLD_WRITE_RPC, 90);
                /* after acquire(), we should run quota_check_common again
                 * so that we confirm there are enough quota to finish write */
                rc = acquire(obd, uid, gid);

                /* please reference to dqacq_completion for the below */
                /* a new request is finished, try again */
                if (rc == -EAGAIN) {
                        CDEBUG(D_QUOTA, "finish a quota req, try again\n");
                        continue;
                }

                /* it is out of quota already */
                if (rc == -EDQUOT) {
                        CDEBUG(D_QUOTA, "out of quota,  return -EDQUOT\n");
                        break;
                }

                /* -EBUSY and others, try 10 times */
                if (rc < 0 && count_err < 10) {
                        CDEBUG(D_QUOTA, "rc: %d, count_err: %d\n", rc, count_err++);
                        cfs_schedule_timeout(CFS_TASK_INTERRUPTIBLE, HZ);
                        continue;
                }

                if (count_err >= 10 || cycle >= 1000) {
                        CDEBUG(D_ERROR, "we meet 10 errors or run too many"
                               " cycles when acquiring quota, quit checking with"
                               " rc: %d, cycle: %d.\n", rc, cycle);
                        break;
                }

                CDEBUG(D_QUOTA, "recheck quota with rc: %d, cycle: %d\n", rc,
                       cycle);
        }

        if (!cycle && rc & QUOTA_RET_INC_PENDING)
                *pending = 1;

        RETURN(rc);
}


static int filter_quota_check(struct obd_device *obd, unsigned int uid,
                              unsigned int gid, int npage, int *flag,
                              quota_acquire acquire)
{
        return quota_chk_acq_common(obd, uid, gid, npage, flag, LQUOTA_FLAGS_BLK,
                                    acquire);
}

/* when a block_write or inode_create rpc is finished, adjust the record for
 * pending blocks and inodes*/
static int quota_pending_commit(struct obd_device *obd, unsigned int uid,
                                unsigned int gid, int count, int isblk)
{
        struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
        int i;
        __u32 id[MAXQUOTAS] = { uid, gid };
        struct qunit_data qdata[MAXQUOTAS];
        ENTRY;

        CLASSERT(MAXQUOTAS < 4);
        if (!sb_any_quota_enabled(qctxt->lqc_sb))
                RETURN(0);

        for (i = 0; i < MAXQUOTAS; i++) {
                struct lustre_qunit_size *lqs = NULL;

                qdata[i].qd_id = id[i];
                qdata[i].qd_flags = i;
                if (isblk)
                        QDATA_SET_BLK(&qdata[i]);
                qdata[i].qd_count = 0;

                if (qdata[i].qd_id == 0 && !QDATA_IS_GRP(&qdata[i]))
                        continue;

                quota_search_lqs(&qdata[i], NULL, qctxt, &lqs);
                if (lqs) {
                        int flag = 0;
                        spin_lock(&lqs->lqs_lock);
                        CDEBUG(D_QUOTA, "pending: %lu, count: %d.\n",
                               isblk ? lqs->lqs_bwrite_pending :
                               lqs->lqs_iwrite_pending, count);

                        if (isblk) {
                                if (lqs->lqs_bwrite_pending >= count) {
                                        lqs->lqs_bwrite_pending -= count;
                                        flag = 1;
                                } else {
                                        CDEBUG(D_ERROR,
                                               "there are too many blocks!\n");
                                }
                        } else {
                                if (lqs->lqs_iwrite_pending >= count) {
                                        lqs->lqs_iwrite_pending -= count;
                                        flag = 1;
                                } else {
                                        CDEBUG(D_ERROR,
                                               "there are too many files!\n");
                                }
                        }

                        spin_unlock(&lqs->lqs_lock);
                        lqs_putref(lqs);
                        /* When lqs_*_pening is changed back, we'll putref lqs
                         * here b=14784 */
                        if (flag)
                                lqs_putref(lqs);
                }
        }

        RETURN(0);
}

static int filter_quota_pending_commit(struct obd_device *obd, unsigned int uid,
                                       unsigned int gid, int npage)
{
        return quota_pending_commit(obd, uid, gid, npage, LQUOTA_FLAGS_BLK);
}

static int mds_quota_init(void)
{
        return lustre_dquot_init();
}

static int mds_quota_exit(void)
{
        lustre_dquot_exit();
        return 0;
}

static int mds_quota_setup(struct obd_device *obd)
{
        struct obd_device_target *obt = &obd->u.obt;
        struct mds_obd *mds = &obd->u.mds;
        int rc;
        ENTRY;

        mds->mds_quota_info.qi_version = LUSTRE_QUOTA_V2;
        atomic_set(&obt->obt_quotachecking, 1);
        /* initialize quota master and quota context */
        sema_init(&mds->mds_qonoff_sem, 1);
        rc = qctxt_init(&obt->obt_qctxt, obt->obt_sb, dqacq_handler);
        if (rc) {
                CERROR("initialize quota context failed! (rc:%d)\n", rc);
                RETURN(rc);
        }
        RETURN(rc);
}

static int mds_quota_cleanup(struct obd_device *obd)
{
        qctxt_cleanup(&obd->u.obt.obt_qctxt, 0);
        RETURN(0);
}

static int mds_quota_fs_cleanup(struct obd_device *obd)
{
        struct mds_obd *mds = &obd->u.mds;
        int i;
        ENTRY;

        /* close admin quota files */
        down(&mds->mds_qonoff_sem);
        for (i = 0; i < MAXQUOTAS; i++) {
                if (mds->mds_quota_info.qi_files[i]) {
                        filp_close(mds->mds_quota_info.qi_files[i], 0);
                        mds->mds_quota_info.qi_files[i] = NULL;
                }
        }
        up(&mds->mds_qonoff_sem);
        RETURN(0);
}

static int mds_quota_check(struct obd_device *obd, unsigned int uid,
                           unsigned int gid, int inodes, int *flag,
                           quota_acquire acquire)
{
        return quota_chk_acq_common(obd, uid, gid, inodes, flag, 0, acquire);
}

static int mds_quota_acquire(struct obd_device *obd, unsigned int uid,
                             unsigned int gid)
{
        struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
        int rc;