quota_context.c

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

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *  lustre/quota/quota_context.c
 *  Lustre Quota Context
 *
 *  Copyright (c) 2001-2005 Cluster File Systems, Inc.
 *   Author: Niu YaWei <[e-mail]>
 *
 *   This file is part of Lustre, http://www.lustre.org.
 *
 *   No redistribution or use is permitted outside of Cluster File Systems, Inc.
 *
 */
#ifndef EXPORT_SYMTAB
# define EXPORT_SYMTAB
#endif

#define DEBUG_SUBSYSTEM S_MDS

#include <linux/version.h>
#include <linux/fs.h>
#include <asm/unistd.h>
#include <linux/slab.h>
#include <linux/quotaops.h>
#include <linux/module.h>
#include <linux/init.h>

#include <obd_class.h>
#include <lustre_quota.h>
#include <lustre_fsfilt.h>
#include <class_hash.h>
#include "quota_internal.h"

extern struct lustre_hash_operations lqs_hash_operations;

unsigned long default_bunit_sz = 128 * 1024 * 1024; /* 128M bytes */
unsigned long default_btune_ratio = 50;             /* 50 percentage */
unsigned long default_iunit_sz = 5120;              /* 5120 inodes */
unsigned long default_itune_ratio = 50;             /* 50 percentage */

cfs_mem_cache_t *qunit_cachep = NULL;
struct list_head qunit_hash[NR_DQHASH];
spinlock_t qunit_hash_lock = SPIN_LOCK_UNLOCKED;

/* please sync qunit_state with qunit_state_names */
enum qunit_state {
        QUNIT_CREATED      = 0,   /* a qunit is created */
        QUNIT_IN_HASH      = 1,   /* a qunit is added into qunit hash, that means
                                   * a quota req will be sent or is flying */
        QUNIT_RM_FROM_HASH = 2,   /* a qunit is removed from qunit hash, that
                                   * means a quota req is handled and comes
                                   * back */
        QUNIT_FINISHED     = 3,   /* qunit can wake up all threads waiting
                                   * for it */
};

static const char *qunit_state_names[] = {
        [QUNIT_CREATED]      = "CREATED",
        [QUNIT_IN_HASH]      = "IN_HASH",
        [QUNIT_RM_FROM_HASH] = "RM_FROM_HASH",
        [QUNIT_FINISHED]     = "FINISHED",
};

struct lustre_qunit {
        struct list_head lq_hash;          /* Hash list in memory */
        atomic_t lq_refcnt;                /* Use count */
        struct lustre_quota_ctxt *lq_ctxt; /* Quota context this applies to */
        struct qunit_data lq_data;         /* See qunit_data */
        unsigned int lq_opc;               /* QUOTA_DQACQ, QUOTA_DQREL */
        cfs_waitq_t lq_waitq;              /* Threads waiting for this qunit */
        spinlock_t lq_lock;                /* Protect the whole structure */
        enum qunit_state lq_state;         /* Present the status of qunit */
        int lq_rc;                         /* The rc of lq_data */
};

#define QUNIT_SET_STATE(qunit, state)                                   \
do {                                                                    \
        spin_lock(&qunit->lq_lock);                                     \
        QDATA_DEBUG((&qunit->lq_data), "qunit(%p) lq_state(%s->%s), "   \
                    "lq_rc(%d)\n",                                      \
                    qunit, qunit_state_names[qunit->lq_state],          \
                    qunit_state_names[state], qunit->lq_rc);            \
        qunit->lq_state = state;                                        \
        spin_unlock(&qunit->lq_lock);                                   \
} while(0)

#define QUNIT_SET_STATE_AND_RC(qunit, state, rc)                        \
do {                                                                    \
        spin_lock(&qunit->lq_lock);                                     \
        qunit->lq_rc = rc;                                              \
        QDATA_DEBUG((&qunit->lq_data), "qunit(%p) lq_state(%s->%s), "   \
                    "lq_rc(%d)\n",                                      \
                    qunit, qunit_state_names[qunit->lq_state],          \
                    qunit_state_names[state], qunit->lq_rc);            \
        qunit->lq_state = state;                                        \
        spin_unlock(&qunit->lq_lock);                                   \
} while(0)


int should_translate_quota (struct obd_import *imp)
{
        ENTRY;

        LASSERT(imp);
#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(1, 7, 0, 0)
        if (imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_QUOTA64 &&
            !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT))
#else
        if (imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_QUOTA64)
#endif
                RETURN(0);
        else
                RETURN(1);
}

void qunit_cache_cleanup(void)
{
        int i;
        ENTRY;

        spin_lock(&qunit_hash_lock);
        for (i = 0; i < NR_DQHASH; i++)
                LASSERT(list_empty(qunit_hash + i));
        spin_unlock(&qunit_hash_lock);

        if (qunit_cachep) {
                int rc;
                rc = cfs_mem_cache_destroy(qunit_cachep);
                LASSERTF(rc == 0, "couldn't destory qunit_cache slab\n");
                qunit_cachep = NULL;
        }
        EXIT;
}

int qunit_cache_init(void)
{
        int i;
        ENTRY;

        LASSERT(qunit_cachep == NULL);
        qunit_cachep = cfs_mem_cache_create("ll_qunit_cache",
                                            sizeof(struct lustre_qunit),
                                            0, 0);
        if (!qunit_cachep)
                RETURN(-ENOMEM);

        spin_lock(&qunit_hash_lock);
        for (i = 0; i < NR_DQHASH; i++)
                INIT_LIST_HEAD(qunit_hash + i);
        spin_unlock(&qunit_hash_lock);
        RETURN(0);
}

static inline int
qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
             __attribute__((__const__));

static inline int
qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
{
        unsigned int id = qdata->qd_id;
        unsigned int type = QDATA_IS_GRP(qdata);

        unsigned long tmp = ((unsigned long)qctxt >> L1_CACHE_SHIFT) ^ id;
        tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
        return tmp;
}

/* caller must hold qunit_hash_lock */
static inline struct lustre_qunit *find_qunit(unsigned int hashent,
                                              struct lustre_quota_ctxt *qctxt,
                                              struct qunit_data *qdata)
{
        struct lustre_qunit *qunit = NULL;
        struct qunit_data *tmp;

        LASSERT_SPIN_LOCKED(&qunit_hash_lock);
        list_for_each_entry(qunit, qunit_hash + hashent, lq_hash) {
                tmp = &qunit->lq_data;
                if (qunit->lq_ctxt == qctxt &&
                    qdata->qd_id == tmp->qd_id &&
                    (qdata->qd_flags & LQUOTA_QUNIT_FLAGS) ==
                    (tmp->qd_flags & LQUOTA_QUNIT_FLAGS))
                        return qunit;
        }
        return NULL;
}

/* check_cur_qunit - check the current usage of qunit.
 * @qctxt: quota context
 * @qdata: the type of quota unit to be checked
 *
 * return: 1 - need acquire qunit;
 * 	   2 - need release qunit;
 * 	   0 - need do nothing.
 * 	 < 0 - error.
 */
static int
check_cur_qunit(struct obd_device *obd,
                struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
{
        struct super_block *sb = qctxt->lqc_sb;
        unsigned long qunit_sz, tune_sz;
        __u64 usage, limit, limit_org, pending_write = 0;
        long long record = 0;
        struct obd_quotactl *qctl;
        struct lustre_qunit_size *lqs = NULL;
        int ret = 0;
        ENTRY;

        if (!sb_any_quota_enabled(sb))
                RETURN(0);

        OBD_ALLOC_PTR(qctl);
        if (qctl == NULL)
                RETURN(-ENOMEM);

        /* get fs quota usage & limit */
        qctl->qc_cmd = Q_GETQUOTA;
        qctl->qc_id = qdata->qd_id;
        qctl->qc_type = QDATA_IS_GRP(qdata);
        ret = fsfilt_quotactl(obd, sb, qctl);
        if (ret) {
                if (ret == -ESRCH)      /* no limit */
                        ret = 0;
                else
                        CERROR("can't get fs quota usage! (rc:%d)\n", ret);
                GOTO(out, ret);
        }

        if (QDATA_IS_BLK(qdata)) {
                usage = qctl->qc_dqblk.dqb_curspace;
                limit = qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS;
        } else {
                usage = qctl->qc_dqblk.dqb_curinodes;
                limit = qctl->qc_dqblk.dqb_ihardlimit;
        }

        /* ignore the no quota limit case; and it can avoid creating
         * unnecessary lqs for uid/gid */
        if (!limit)
                GOTO(out, ret = 0);

 search_lqs:
        quota_search_lqs(qdata, NULL, qctxt, &lqs);
        if (!lqs) {
                CDEBUG(D_QUOTA, "Can't find the lustre qunit size!\n");
                ret = quota_create_lqs(qdata, NULL, qctxt, &lqs);
                if (ret == -EALREADY) {
                        ret = 0;
                        goto search_lqs;
                }
                if (ret < 0)
                        GOTO (out, ret);
        }
        spin_lock(&lqs->lqs_lock);

        if (QDATA_IS_BLK(qdata)) {
                qunit_sz = lqs->lqs_bunit_sz;
                tune_sz  = lqs->lqs_btune_sz;
                pending_write = lqs->lqs_bwrite_pending * CFS_PAGE_SIZE;
                record   = lqs->lqs_blk_rec;
                LASSERT(!(qunit_sz % QUOTABLOCK_SIZE));
        } else {
                /* we didn't need change inode qunit size now */
                qunit_sz = lqs->lqs_iunit_sz;
                tune_sz  = lqs->lqs_itune_sz;
                pending_write = lqs->lqs_iwrite_pending;
                record   = lqs->lqs_ino_rec;
        }

        /* we don't count the MIN_QLIMIT */
        if ((limit == MIN_QLIMIT && !QDATA_IS_BLK(qdata)) ||
            (toqb(limit) == MIN_QLIMIT && QDATA_IS_BLK(qdata)))
                limit = 0;

        usage += pending_write;
        limit_org = limit;
        /* when a releasing quota req is sent, before it returned
           limit is assigned a small value. limit will overflow */
        if (limit + record < 0)
                usage -= record;
        else
                limit += record;

        LASSERT(qdata->qd_count == 0);
        if (limit <= usage + tune_sz) {
                while (qdata->qd_count + limit <=
                       usage + tune_sz)
                        qdata->qd_count += qunit_sz;
                ret = 1;
        } else if (limit > usage + qunit_sz + tune_sz &&
                   limit_org > qdata->qd_count + qunit_sz) {
                while (limit - qdata->qd_count > usage + qunit_sz + tune_sz &&
                       limit_org > qdata->qd_count + qunit_sz)
                        qdata->qd_count += qunit_sz;
                ret = 2;
        }
        CDEBUG(D_QUOTA, "type: %c, limit: "LPU64", usage: "LPU64
               ", pending_write: "LPU64", record: "LPD64
               ", qunit_sz: %lu, tune_sz: %lu, ret: %d.\n",
               QDATA_IS_BLK(qdata) ? 'b' : 'i', limit, usage, pending_write,
               record, qunit_sz, tune_sz, ret);
        LASSERT(ret == 0 || qdata->qd_count);

        spin_unlock(&lqs->lqs_lock);
        lqs_putref(lqs);
        EXIT;
 out:
        OBD_FREE_PTR(qctl);
        return ret;
}

/* compute the remaining quota for certain gid or uid b=11693 */
int compute_remquota(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
                     struct qunit_data *qdata, int isblk)
{
        struct super_block *sb = qctxt->lqc_sb;
        __u64 usage, limit;
        struct obd_quotactl *qctl;
        int ret = QUOTA_RET_OK;
        ENTRY;

        if (!sb_any_quota_enabled(sb))
                RETURN(QUOTA_RET_NOQUOTA);

        /* ignore root user */
        if (qdata->qd_id == 0 && QDATA_IS_GRP(qdata) == USRQUOTA)
                RETURN(QUOTA_RET_NOLIMIT);

        OBD_ALLOC_PTR(qctl);
        if (qctl == NULL)
                RETURN(-ENOMEM);

        /* get fs quota usage & limit */
        qctl->qc_cmd = Q_GETQUOTA;
        qctl->qc_id = qdata->qd_id;
        qctl->qc_type = QDATA_IS_GRP(qdata);
        ret = fsfilt_quotactl(obd, sb, qctl);
        if (ret) {
                if (ret == -ESRCH)      /* no limit */
                        ret = QUOTA_RET_NOLIMIT;
                else
                        CDEBUG(D_QUOTA, "can't get fs quota usage! (rc:%d)",
                               ret);
                GOTO(out, ret);
        }

        usage = isblk ? qctl->qc_dqblk.dqb_curspace :
                qctl->qc_dqblk.dqb_curinodes;
        limit = isblk ? qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS :
                qctl->qc_dqblk.dqb_ihardlimit;
        if (!limit){            /* no limit */
                ret = QUOTA_RET_NOLIMIT;
                GOTO(out, ret);
        }

        if (limit >= usage)
                qdata->qd_count = limit - usage;
        else
                qdata->qd_count = 0;
        EXIT;
out:
        OBD_FREE_PTR(qctl);
        return ret;
}

/* caller must hold qunit_hash_lock */
static struct lustre_qunit *dqacq_in_flight(struct lustre_quota_ctxt *qctxt,
                                            struct qunit_data *qdata)
{
        unsigned int hashent = qunit_hashfn(qctxt, qdata);
        struct lustre_qunit *qunit;
        ENTRY;

        LASSERT_SPIN_LOCKED(&qunit_hash_lock);
        qunit = find_qunit(hashent, qctxt, qdata);
        RETURN(qunit);
}

static struct lustre_qunit *alloc_qunit(struct lustre_quota_ctxt *qctxt,
                                        struct qunit_data *qdata, int opc)
{
        struct lustre_qunit *qunit = NULL;
        ENTRY;