quota_context.c

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

                goto wait_completion;
        }
        qunit = empty;
        insert_qunit_nolock(qctxt, qunit);
        spin_unlock(&qunit_hash_lock);

        LASSERT(qunit);

        quota_search_lqs(qdata, NULL, qctxt, &lqs);
        if (lqs) {
                spin_lock(&lqs->lqs_lock);
                quota_compute_lqs(qdata, lqs, 1, (opc == QUOTA_DQACQ) ? 1 : 0);
                /* when this qdata returned from mds, it will call lqs_putref */
                lqs_getref(lqs);
                spin_unlock(&lqs->lqs_lock);
                /* this is for quota_search_lqs */
                lqs_putref(lqs);
        } else {
                CDEBUG(D_ERROR, "Can't find the lustre qunit size!\n");
        }

        QDATA_DEBUG(qdata, "obd(%s): send %s quota req\n",
                    obd->obd_name, (opc == QUOTA_DQACQ) ? "acq" : "rel");
        /* master is going to dqacq/dqrel from itself */
        if (is_master(obd, qctxt, qdata->qd_id, QDATA_IS_GRP(qdata))) {
                int rc2;
                QDATA_DEBUG(qdata, "local %s.\n",
                            opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
                rc = qctxt->lqc_handler(obd, qdata, opc);
                rc2 = dqacq_completion(obd, qctxt, qdata, rc, opc);
                RETURN(rc ? 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);
                remove_qunit_nolock(qunit);
                spin_unlock(&qunit_hash_lock);

                compute_lqs_after_removing_qunit(qunit);

                QUNIT_SET_STATE_AND_RC(qunit, QUNIT_FINISHED, -EAGAIN);
                wake_up(&qunit->lq_waitq);

                qunit_put(qunit);
                RETURN(-EAGAIN);
        }
        imp = class_import_get(qctxt->lqc_import);
        spin_unlock(&qctxt->lqc_lock);

        /* build dqacq/dqrel request */
        LASSERT(imp);
        size[1] = quota_get_qunit_data_size(imp->
                                            imp_connect_data.ocd_connect_flags);

        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);
        }

        if (QDATA_IS_BLK(qdata))
                factor = MAX_QUOTA_COUNT32 / qctxt->lqc_bunit_sz *
                        qctxt->lqc_bunit_sz;
        else
                factor = MAX_QUOTA_COUNT32 / qctxt->lqc_iunit_sz *
                        qctxt->lqc_iunit_sz;

        LASSERTF(!should_translate_quota(imp) || qdata->qd_count <= factor,
                 "qd_count: "LPU64"; should_translate_quota: %d.\n",
                 qdata->qd_count, should_translate_quota(imp));
        rc = quota_copy_qdata(req, qdata, QUOTA_REQUEST, QUOTA_IMPORT);
        if (rc < 0) {
                CDEBUG(D_ERROR, "Can't pack qunit_data\n");
                RETURN(-EPROTO);
        }
        ptlrpc_req_set_repsize(req, 2, size);
        class_import_put(imp);

        if (wait && qunit) 
                qunit_get(qunit);

        CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
        aa = (struct dqacq_async_args *)&req->rq_async_args;
        aa->aa_ctxt = qctxt;
        aa->aa_qunit = qunit;

        req->rq_interpret_reply = dqacq_interpret;
        ptlrpcd_add_req(req);

        QDATA_DEBUG(qdata, "%s scheduled.\n",
                    opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
wait_completion:
        if (wait && qunit) {
                struct qunit_data *p = &qunit->lq_data;

                QDATA_DEBUG(p, "qunit(%p) is waiting for dqacq.\n", qunit);
                l_wait_event(qunit->lq_waitq, got_qunit(qunit), &lwi);
                /* rc = -EAGAIN, it means a quota req is finished;
                 * rc = -EDQUOT, it means out of quota
                 * rc = -EBUSY, it means recovery is happening
                 * other rc < 0, it means real errors, functions who call
                 * schedule_dqacq should take care of this */
                spin_lock(&qunit->lq_lock);
                if (qunit->lq_rc == 0)
                        rc = -EAGAIN;
                else
                        rc = qunit->lq_rc;
                spin_unlock(&qunit->lq_lock);
                CDEBUG(D_QUOTA, "qunit(%p) finishes waiting. (rc:%d)\n",
                       qunit, rc);
                qunit_put(qunit);
        }
        RETURN(rc);
}

int
qctxt_adjust_qunit(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
                   uid_t uid, gid_t gid, __u32 isblk, int wait)
{
        int ret, rc = 0, i = USRQUOTA;
        __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++) {
                qdata[i].qd_id = id[i];
                qdata[i].qd_flags = i;
                if (isblk)
                        QDATA_SET_BLK(&qdata[i]);
                qdata[i].qd_count = 0;

                ret = check_cur_qunit(obd, qctxt, &qdata[i]);
                if (ret > 0) {
                        int opc;
                        /* need acquire or release */
                        opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
                        ret = split_before_schedule_dqacq(obd, qctxt, &qdata[i], 
                                                          opc, wait);
                        if (!rc)
                                rc = ret;
                } else if (wait == 1) {
                        /* when wait equates 1, that means mds_quota_acquire
                         * or filter_quota_acquire is calling it. */
                        qctxt_wait_pending_dqacq(qctxt, id[i], i, isblk);
                }
        }

        RETURN(rc);
}

int
qctxt_wait_pending_dqacq(struct lustre_quota_ctxt *qctxt, unsigned int id,
                         unsigned short type, int isblk)
{
        struct lustre_qunit *qunit = NULL;
        struct qunit_data qdata;
        struct l_wait_info lwi = { 0 };
        ENTRY;

        qdata.qd_id = id;
        qdata.qd_flags = type;
        if (isblk)
                QDATA_SET_BLK(&qdata);
        qdata.qd_count = 0;

        spin_lock(&qunit_hash_lock);
        qunit = dqacq_in_flight(qctxt, &qdata);
        if (qunit)
                /* grab reference on this qunit to handle races with
                 * dqacq_completion(). Otherwise, this qunit could be freed just
                 * after we release the qunit_hash_lock */
                qunit_get(qunit);
        spin_unlock(&qunit_hash_lock);

        if (qunit) {
                struct qunit_data *p = &qunit->lq_data;

                QDATA_DEBUG(p, "qunit(%p) is waiting for dqacq.\n", qunit);
                l_wait_event(qunit->lq_waitq, got_qunit(qunit), &lwi);
                CDEBUG(D_QUOTA, "qunit(%p) finishes waiting. (rc:%d)\n",
                       qunit, qunit->lq_rc);
                qunit_put(qunit);
        }
        RETURN(0);
}

int
qctxt_init(struct lustre_quota_ctxt *qctxt, struct super_block *sb,
           dqacq_handler_t handler)
{
        int rc = 0;
        ENTRY;

        LASSERT(qctxt);

        rc = ptlrpcd_addref();
        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;
        qctxt->lqc_recovery = 0;
        qctxt->lqc_switch_qs = 1; /* Change qunit size in default setting */
        qctxt->lqc_cqs_boundary_factor = 4;
        qctxt->lqc_cqs_least_bunit = PTLRPC_MAX_BRW_SIZE;
        qctxt->lqc_cqs_least_iunit = 1;
        qctxt->lqc_cqs_qs_factor = 2;
        qctxt->lqc_atype = 0;
        qctxt->lqc_status= 0;
        qctxt->lqc_bunit_sz = default_bunit_sz;
        qctxt->lqc_btune_sz = default_bunit_sz / 100 * default_btune_ratio;
        qctxt->lqc_iunit_sz = default_iunit_sz;
        qctxt->lqc_itune_sz = default_iunit_sz * default_itune_ratio / 100;
        qctxt->lqc_switch_seconds = 300; /* enlarging will wait 5 minutes
                                          * after the last shrinking */
        rc = lustre_hash_init(&LQC_HASH_BODY(qctxt), "LQS_HASH",128,
                              &lqs_hash_operations);
        if (rc) {
                CDEBUG(D_ERROR, "initialize hash lqs on ost error!\n");
                lustre_hash_exit(&LQC_HASH_BODY(qctxt));
        }
        spin_unlock(&qctxt->lqc_lock);

        RETURN(rc);
}

void qctxt_cleanup(struct lustre_quota_ctxt *qctxt, int force)
{
        struct lustre_qunit *qunit, *tmp;
        struct list_head tmp_list;
        int i;
        ENTRY;

        INIT_LIST_HEAD(&tmp_list);

        spin_lock(&qunit_hash_lock);
        for (i = 0; i < NR_DQHASH; i++) {
                list_for_each_entry_safe(qunit, tmp, &qunit_hash[i], lq_hash) {
                        if (qunit->lq_ctxt != qctxt)
                                continue;
                        remove_qunit_nolock(qunit);
                        list_add(&qunit->lq_hash, &tmp_list);
                }
        }
        spin_unlock(&qunit_hash_lock);

        list_for_each_entry_safe(qunit, tmp, &tmp_list, lq_hash) {
                list_del_init(&qunit->lq_hash);
                compute_lqs_after_removing_qunit(qunit);

                /* wake up all waiters */
                QUNIT_SET_STATE_AND_RC(qunit, QUNIT_FINISHED, 0);
                wake_up(&qunit->lq_waitq);
                qunit_put(qunit);
        }

        lustre_hash_exit(&LQC_HASH_BODY(qctxt));
        ptlrpcd_decref();

        EXIT;
}

struct qslave_recov_thread_data {
        struct obd_device *obd;
        struct lustre_quota_ctxt *qctxt;
        struct completion comp;
};

/* FIXME only recovery block quota by now */
static int qslave_recovery_main(void *arg)
{
        struct qslave_recov_thread_data *data = arg;
        struct obd_device *obd = data->obd;
        struct lustre_quota_ctxt *qctxt = data->qctxt;
        unsigned int type;
        int rc = 0;
        ENTRY;

        ptlrpc_daemonize("qslave_recovd");

        complete(&data->comp);

        if (qctxt->lqc_recovery)
                RETURN(0);
        qctxt->lqc_recovery = 1;

        for (type = USRQUOTA; type < MAXQUOTAS; type++) {
                struct qunit_data qdata;
                struct quota_info *dqopt = sb_dqopt(qctxt->lqc_sb);
                struct list_head id_list;
                struct dquot_id *dqid, *tmp;
                int ret;

                LOCK_DQONOFF_MUTEX(dqopt);
                if (!sb_has_quota_enabled(qctxt->lqc_sb, type)) {
                        UNLOCK_DQONOFF_MUTEX(dqopt);
                        break;
                }

                LASSERT(dqopt->files[type] != NULL);
                INIT_LIST_HEAD(&id_list);
#ifndef KERNEL_SUPPORTS_QUOTA_READ
                rc = fsfilt_qids(obd, dqopt->files[type], NULL, type, &id_list);
#else
                rc = fsfilt_qids(obd, NULL, dqopt->files[type], type, &id_list);
#endif
                UNLOCK_DQONOFF_MUTEX(dqopt);
                if (rc)
                        CERROR("Get ids from quota file failed. (rc:%d)\n", rc);

                list_for_each_entry_safe(dqid, tmp, &id_list, di_link) {
                        list_del_init(&dqid->di_link);
                        /* skip slave recovery on itself */
                        if (is_master(obd, qctxt, dqid->di_id, type))
                                goto free;
                        if (rc && rc != -EBUSY)
                                goto free;

                        qdata.qd_id = dqid->di_id;
                        qdata.qd_flags = type;
                        QDATA_SET_BLK(&qdata);
                        qdata.qd_count = 0;

                        ret = check_cur_qunit(obd, qctxt, &qdata);
                        if (ret > 0) {
                                int opc;
                                opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
                                rc = split_before_schedule_dqacq(obd, qctxt,
                                                                 &qdata, opc,
                                                                 0);
                                if (rc == -EDQUOT)
                                        rc = 0;
                        } else {
                                rc = 0;
                        }

                        if (rc)
                                CDEBUG(rc == -EBUSY ? D_QUOTA : D_ERROR,
                                       "qslave recovery failed! (id:%d type:%d "
                                       " rc:%d)\n", dqid->di_id, type, rc);
free:
                        kfree(dqid);
                }
        }

        qctxt->lqc_recovery = 0;
        RETURN(rc);
}

void
qslave_start_recovery(struct obd_device *obd, struct lustre_quota_ctxt *qctxt)
{
        struct qslave_recov_thread_data data;
        int rc;
        ENTRY;

        if (!sb_any_quota_enabled(qctxt->lqc_sb))
                goto exit;

        data.obd = obd;
        data.qctxt = qctxt;
        init_completion(&data.comp);

        rc = kernel_thread(qslave_recovery_main, &data, CLONE_VM|CLONE_FILES);
        if (rc < 0) {
                CERROR("Cannot start quota recovery thread: rc %d\n", rc);
                goto exit;
        }
        wait_for_completion(&data.comp);
exit:
        EXIT;
}