1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * lustre/quota/quota_context.c
7 * Copyright (c) 2001-2005 Cluster File Systems, Inc.
8 * Author: Niu YaWei <niu@clusterfs.com>
10 * This file is part of Lustre, http://www.lustre.org.
12 * No redistribution or use is permitted outside of Cluster File Systems, Inc.
16 # define EXPORT_SYMTAB
19 #define DEBUG_SUBSYSTEM S_MDS
21 #include <linux/version.h>
23 #include <asm/unistd.h>
24 #include <linux/slab.h>
25 #include <linux/quotaops.h>
26 #include <linux/module.h>
27 #include <linux/init.h>
29 #include <obd_class.h>
30 #include <lustre_quota.h>
31 #include <lustre_fsfilt.h>
32 #include "quota_internal.h"
34 unsigned long default_bunit_sz = 100 * 1024 * 1024; /* 100M bytes */
35 unsigned long default_btune_ratio = 50; /* 50 percentage */
36 unsigned long default_iunit_sz = 5000; /* 5000 inodes */
37 unsigned long default_itune_ratio = 50; /* 50 percentage */
39 cfs_mem_cache_t *qunit_cachep = NULL;
40 struct list_head qunit_hash[NR_DQHASH];
41 spinlock_t qunit_hash_lock = SPIN_LOCK_UNLOCKED;
44 struct list_head lq_hash; /* Hash list in memory */
45 atomic_t lq_refcnt; /* Use count */
46 struct lustre_quota_ctxt *lq_ctxt; /* Quota context this applies to */
47 struct qunit_data lq_data; /* See qunit_data */
48 unsigned int lq_opc; /* QUOTA_DQACQ, QUOTA_DQREL */
49 struct list_head lq_waiters; /* All write threads waiting for this qunit */
52 int should_translate_quota (struct obd_import *imp)
57 if ((imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_QUOTA64) &&
58 !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT))
64 void qunit_cache_cleanup(void)
69 spin_lock(&qunit_hash_lock);
70 for (i = 0; i < NR_DQHASH; i++)
71 LASSERT(list_empty(qunit_hash + i));
72 spin_unlock(&qunit_hash_lock);
76 rc = cfs_mem_cache_destroy(qunit_cachep);
77 LASSERTF(rc == 0, "couldn't destory qunit_cache slab\n");
83 int qunit_cache_init(void)
88 LASSERT(qunit_cachep == NULL);
89 qunit_cachep = cfs_mem_cache_create("ll_qunit_cache",
90 sizeof(struct lustre_qunit),
95 spin_lock(&qunit_hash_lock);
96 for (i = 0; i < NR_DQHASH; i++)
97 CFS_INIT_LIST_HEAD(qunit_hash + i);
98 spin_unlock(&qunit_hash_lock);
103 qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
104 __attribute__((__const__));
107 qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
109 unsigned int id = qdata->qd_id;
110 unsigned int type = qdata->qd_flags & QUOTA_IS_GRP;
112 unsigned long tmp = ((unsigned long)qctxt >> L1_CACHE_SHIFT) ^ id;
113 tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
117 /* compute the remaining quota for certain gid or uid b=11693 */
118 int compute_remquota(struct obd_device *obd,
119 struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
121 struct super_block *sb = qctxt->lqc_sb;
123 struct obd_quotactl *qctl;
124 int ret = QUOTA_RET_OK;
125 __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
128 if (!sb_any_quota_enabled(sb))
129 RETURN(QUOTA_RET_NOQUOTA);
131 /* ignore root user */
132 if (qdata->qd_id == 0 && qdata_type == USRQUOTA)
133 RETURN(QUOTA_RET_NOLIMIT);
139 /* get fs quota usage & limit */
140 qctl->qc_cmd = Q_GETQUOTA;
141 qctl->qc_id = qdata->qd_id;
142 qctl->qc_type = qdata_type;
143 ret = fsfilt_quotactl(obd, sb, qctl);
145 if (ret == -ESRCH) /* no limit */
146 ret = QUOTA_RET_NOLIMIT;
148 CDEBUG(D_QUOTA, "can't get fs quota usage! (rc:%d)",
153 usage = qctl->qc_dqblk.dqb_curspace;
154 limit = qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS;
155 if (!limit){ /* no limit */
156 ret = QUOTA_RET_NOLIMIT;
161 qdata->qd_count = limit - usage;
170 /* caller must hold qunit_hash_lock */
171 static inline struct lustre_qunit *find_qunit(unsigned int hashent,
172 struct lustre_quota_ctxt *qctxt,
173 struct qunit_data *qdata)
175 struct lustre_qunit *qunit = NULL;
176 struct qunit_data *tmp;
178 LASSERT_SPIN_LOCKED(&qunit_hash_lock);
179 list_for_each_entry(qunit, qunit_hash + hashent, lq_hash) {
180 tmp = &qunit->lq_data;
181 if (qunit->lq_ctxt == qctxt &&
182 qdata->qd_id == tmp->qd_id && qdata->qd_flags == tmp->qd_flags)
188 /* check_cur_qunit - check the current usage of qunit.
189 * @qctxt: quota context
190 * @qdata: the type of quota unit to be checked
192 * return: 1 - need acquire qunit;
193 * 2 - need release qunit;
194 * 0 - need do nothing.
198 check_cur_qunit(struct obd_device *obd,
199 struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
201 struct super_block *sb = qctxt->lqc_sb;
202 unsigned long qunit_sz, tune_sz;
204 struct obd_quotactl *qctl;
206 __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
207 __u32 is_blk = (qdata->qd_flags & QUOTA_IS_BLOCK) >> 1;
210 if (!sb_any_quota_enabled(sb))
217 /* get fs quota usage & limit */
218 qctl->qc_cmd = Q_GETQUOTA;
219 qctl->qc_id = qdata->qd_id;
220 qctl->qc_type = qdata_type;
221 ret = fsfilt_quotactl(obd, sb, qctl);
223 if (ret == -ESRCH) /* no limit */
226 CERROR("can't get fs quota usage! (rc:%d)\n", ret);
231 usage = qctl->qc_dqblk.dqb_curspace;
232 limit = qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS;
233 qunit_sz = qctxt->lqc_bunit_sz;
234 tune_sz = qctxt->lqc_btune_sz;
236 LASSERT(!(qunit_sz % QUOTABLOCK_SIZE));
238 usage = qctl->qc_dqblk.dqb_curinodes;
239 limit = qctl->qc_dqblk.dqb_ihardlimit;
240 qunit_sz = qctxt->lqc_iunit_sz;
241 tune_sz = qctxt->lqc_itune_sz;
244 /* ignore the no quota limit case */
248 /* we don't count the MIN_QLIMIT */
249 if ((limit == MIN_QLIMIT && !is_blk) ||
250 (toqb(limit) == MIN_QLIMIT && is_blk))
253 LASSERT(qdata->qd_count == 0);
254 if (limit <= usage + tune_sz) {
255 while (qdata->qd_count + limit <= usage + tune_sz)
256 qdata->qd_count += qunit_sz;
258 } else if (limit > usage + qunit_sz + tune_sz) {
259 while (limit - qdata->qd_count > usage + qunit_sz + tune_sz)
260 qdata->qd_count += qunit_sz;
263 LASSERT(ret == 0 || qdata->qd_count);
270 /* caller must hold qunit_hash_lock */
271 static struct lustre_qunit *dqacq_in_flight(struct lustre_quota_ctxt *qctxt,
272 struct qunit_data *qdata)
274 unsigned int hashent = qunit_hashfn(qctxt, qdata);
275 struct lustre_qunit *qunit;
278 LASSERT_SPIN_LOCKED(&qunit_hash_lock);
279 qunit = find_qunit(hashent, qctxt, qdata);
283 static struct lustre_qunit *alloc_qunit(struct lustre_quota_ctxt *qctxt,
284 struct qunit_data *qdata, int opc)
286 struct lustre_qunit *qunit = NULL;
289 OBD_SLAB_ALLOC(qunit, qunit_cachep, CFS_ALLOC_IO, sizeof(*qunit));
293 CFS_INIT_LIST_HEAD(&qunit->lq_hash);
294 CFS_INIT_LIST_HEAD(&qunit->lq_waiters);
295 atomic_set(&qunit->lq_refcnt, 1);
296 qunit->lq_ctxt = qctxt;
297 memcpy(&qunit->lq_data, qdata, sizeof(*qdata));
303 static inline void free_qunit(struct lustre_qunit *qunit)
305 OBD_SLAB_FREE(qunit, qunit_cachep, sizeof(*qunit));
308 static inline void qunit_get(struct lustre_qunit *qunit)
310 atomic_inc(&qunit->lq_refcnt);
313 static void qunit_put(struct lustre_qunit *qunit)
315 LASSERT(atomic_read(&qunit->lq_refcnt));
316 if (atomic_dec_and_test(&qunit->lq_refcnt))
321 insert_qunit_nolock(struct lustre_quota_ctxt *qctxt, struct lustre_qunit *qunit)
323 struct list_head *head;
325 LASSERT(list_empty(&qunit->lq_hash));
326 head = qunit_hash + qunit_hashfn(qctxt, &qunit->lq_data);
327 list_add(&qunit->lq_hash, head);
330 static void remove_qunit_nolock(struct lustre_qunit *qunit)
332 LASSERT(!list_empty(&qunit->lq_hash));
333 list_del_init(&qunit->lq_hash);
336 struct qunit_waiter {
337 struct list_head qw_entry;
338 cfs_waitq_t qw_waitq;
342 #define INC_QLIMIT(limit, count) (limit == MIN_QLIMIT) ? \
343 (limit = count) : (limit += count)
346 /* FIXME check if this mds is the master of specified id */
348 is_master(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
349 unsigned int id, int type)
351 return qctxt->lqc_handler ? 1 : 0;
355 schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
356 struct qunit_data *qdata, int opc, int wait);
358 static int split_before_schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
359 struct qunit_data *qdata, int opc, int wait)
362 unsigned long factor;
363 struct qunit_data tmp_qdata;
366 LASSERT(qdata && qdata->qd_count);
367 QDATA_DEBUG(qdata, "%s quota split.\n",
368 (qdata->qd_flags & QUOTA_IS_BLOCK) ? "block" : "inode");
369 if (qdata->qd_flags & QUOTA_IS_BLOCK)
370 factor = MAX_QUOTA_COUNT32 / qctxt->lqc_bunit_sz *
373 factor = MAX_QUOTA_COUNT32 / qctxt->lqc_iunit_sz *
376 if (qctxt->lqc_import && should_translate_quota(qctxt->lqc_import) &&
377 qdata->qd_count > factor) {
379 tmp_qdata.qd_count = factor;
380 qdata->qd_count -= tmp_qdata.qd_count;
381 QDATA_DEBUG((&tmp_qdata), "be split.\n");
382 rc = schedule_dqacq(obd, qctxt, &tmp_qdata, opc, wait);
384 QDATA_DEBUG(qdata, "don't be split.\n");
385 rc = schedule_dqacq(obd, qctxt, qdata, opc, wait);
392 dqacq_completion(struct obd_device *obd,
393 struct lustre_quota_ctxt *qctxt,
394 struct qunit_data *qdata, int rc, int opc)
396 struct lustre_qunit *qunit = NULL;
397 struct super_block *sb = qctxt->lqc_sb;
398 unsigned long qunit_sz;
399 struct qunit_waiter *qw, *tmp;
401 __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
402 __u32 is_blk = (qdata->qd_flags & QUOTA_IS_BLOCK) >> 1;
403 __u64 qd_tmp = qdata->qd_count;
408 qunit_sz = is_blk ? qctxt->lqc_bunit_sz : qctxt->lqc_iunit_sz;
409 div_r = do_div(qd_tmp, qunit_sz);
410 LASSERTF(!div_r, "qunit_sz: %lu, return qunit_sz: "LPU64"\n",
413 /* update local operational quota file */
415 __u32 count = QUSG(qdata->qd_count, is_blk);
416 struct obd_quotactl *qctl;
421 GOTO(out, err = -ENOMEM);
423 /* acq/rel qunit for specified uid/gid is serialized,
424 * so there is no race between get fs quota limit and
425 * set fs quota limit */
426 qctl->qc_cmd = Q_GETQUOTA;
427 qctl->qc_id = qdata->qd_id;
428 qctl->qc_type = qdata_type;
429 err = fsfilt_quotactl(obd, sb, qctl);
431 CERROR("error get quota fs limit! (rc:%d)\n", err);
436 qctl->qc_dqblk.dqb_valid = QIF_BLIMITS;
437 hardlimit = &qctl->qc_dqblk.dqb_bhardlimit;
439 qctl->qc_dqblk.dqb_valid = QIF_ILIMITS;
440 hardlimit = &qctl->qc_dqblk.dqb_ihardlimit;
445 CDEBUG(D_QUOTA, "%s(acq):count: %d, hardlimt: "LPU64
446 ",type: %s.\n", obd->obd_name, count, *hardlimit,
447 qdata_type ? "grp": "usr");
448 INC_QLIMIT(*hardlimit, count);
451 CDEBUG(D_QUOTA, "%s(rel):count: %d, hardlimt: "LPU64
452 ",type: %s.\n", obd->obd_name, count, *hardlimit,
453 qdata_type ? "grp": "usr");
454 LASSERTF(count < *hardlimit,
455 "count: %d, hardlimit: "LPU64".\n",
463 /* clear quota limit */
467 qctl->qc_cmd = Q_SETQUOTA;
468 err = fsfilt_quotactl(obd, sb, qctl);
470 CERROR("error set quota fs limit! (rc:%d)\n", err);
472 QDATA_DEBUG(qdata, "%s completion\n",
473 opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
476 } else if (rc == -EDQUOT) {
477 QDATA_DEBUG(qdata, "acquire qunit got EDQUOT.\n");
478 } else if (rc == -EBUSY) {
479 QDATA_DEBUG(qdata, "it's is recovering, got EBUSY.\n");
481 CERROR("acquire qunit got error! (rc:%d)\n", rc);
484 /* remove the qunit from hash */
485 spin_lock(&qunit_hash_lock);
487 qunit = dqacq_in_flight(qctxt, qdata);
488 /* this qunit has been removed by qctxt_cleanup() */
490 spin_unlock(&qunit_hash_lock);
494 LASSERT(opc == qunit->lq_opc);
495 remove_qunit_nolock(qunit);
497 /* wake up all waiters */
498 list_for_each_entry_safe(qw, tmp, &qunit->lq_waiters, qw_entry) {
499 list_del_init(&qw->qw_entry);
501 wake_up(&qw->qw_waitq);
504 spin_unlock(&qunit_hash_lock);
508 /* don't reschedule in such cases:
509 * - acq/rel failure, but not for quota recovery.
510 * - local dqacq/dqrel.
511 * - local disk io failure.
513 if (err || (rc && rc != -EBUSY) ||
514 is_master(obd, qctxt, qdata->qd_id, qdata_type))
517 /* reschedule another dqacq/dqrel if needed */
519 rc = check_cur_qunit(obd, qctxt, qdata);
522 opc = rc == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
523 rc = split_before_schedule_dqacq(obd, qctxt, qdata, opc, 0);
524 QDATA_DEBUG(qdata, "reschedudle opc(%d) rc(%d)\n", opc, rc);
529 struct dqacq_async_args {
530 struct lustre_quota_ctxt *aa_ctxt;
531 struct lustre_qunit *aa_qunit;
534 static int dqacq_interpret(struct ptlrpc_request *req, void *data, int rc)
536 struct dqacq_async_args *aa = (struct dqacq_async_args *)data;
537 struct lustre_quota_ctxt *qctxt = aa->aa_ctxt;
538 struct lustre_qunit *qunit = aa->aa_qunit;
539 struct obd_device *obd = req->rq_import->imp_obd;
540 struct qunit_data *qdata = NULL;
541 struct qunit_data_old *qdata_old = NULL;
545 LASSERT(req->rq_import);
547 if ((req->rq_import->imp_connect_data.ocd_connect_flags &
548 OBD_CONNECT_QUOTA64) &&
549 !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT)) {
550 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
552 qdata = req_capsule_server_swab_get(&req->rq_pill,
554 (void*)lustre_swab_qdata);
556 CDEBUG(D_QUOTA, "qd_count is 32bit!\n");
558 qdata = req_capsule_server_swab_get(&req->rq_pill,
560 (void*)lustre_swab_qdata_old);
561 qdata = lustre_quota_old_to_new(qdata_old);
564 DEBUG_REQ(D_ERROR, req, "error unpacking qunit_data");
568 LASSERT(qdata->qd_id == qunit->lq_data.qd_id &&
569 (qdata->qd_flags & QUOTA_IS_GRP) ==
570 (qunit->lq_data.qd_flags & QUOTA_IS_GRP) &&
571 (qdata->qd_count == qunit->lq_data.qd_count ||
572 qdata->qd_count == 0));
574 QDATA_DEBUG(qdata, "%s interpret rc(%d).\n",
575 lustre_msg_get_opc(req->rq_reqmsg) == QUOTA_DQACQ ?
576 "DQACQ" : "DQREL", rc);
578 rc = dqacq_completion(obd, qctxt, qdata, rc,
579 lustre_msg_get_opc(req->rq_reqmsg));
584 static int got_qunit(struct qunit_waiter *waiter)
588 spin_lock(&qunit_hash_lock);
589 rc = list_empty(&waiter->qw_entry);
590 spin_unlock(&qunit_hash_lock);
595 schedule_dqacq(struct obd_device *obd,
596 struct lustre_quota_ctxt *qctxt,
597 struct qunit_data *qdata, int opc, int wait)
599 struct lustre_qunit *qunit, *empty;
600 struct qunit_waiter qw;
601 struct l_wait_info lwi = { 0 };
602 struct ptlrpc_request *req;
603 struct qunit_data *reqdata;
604 struct dqacq_async_args *aa;
605 unsigned long factor;
609 CFS_INIT_LIST_HEAD(&qw.qw_entry);
610 init_waitqueue_head(&qw.qw_waitq);
613 if ((empty = alloc_qunit(qctxt, qdata, opc)) == NULL)
616 spin_lock(&qunit_hash_lock);
618 qunit = dqacq_in_flight(qctxt, qdata);
621 list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
622 spin_unlock(&qunit_hash_lock);
625 goto wait_completion;
628 insert_qunit_nolock(qctxt, qunit);
630 list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
631 spin_unlock(&qunit_hash_lock);
635 /* master is going to dqacq/dqrel from itself */
636 if (is_master(obd, qctxt, qdata->qd_id, qdata->qd_flags & QUOTA_IS_GRP))
639 QDATA_DEBUG(qdata, "local %s.\n",
640 opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
641 rc = qctxt->lqc_handler(obd, qdata, opc);
642 rc2 = dqacq_completion(obd, qctxt, qdata, rc, opc);
643 RETURN((rc && rc != -EDQUOT) ? rc : rc2);
646 /* build dqacq/dqrel request */
647 LASSERT(qctxt->lqc_import);
649 req = ptlrpc_request_alloc_pack(qctxt->lqc_import, &RQF_MDS_QUOTA_DQACQ,
650 LUSTRE_MDS_VERSION, opc);
652 dqacq_completion(obd, qctxt, qdata, -ENOMEM, opc);
656 if (qdata->qd_flags & QUOTA_IS_BLOCK)
657 factor = MAX_QUOTA_COUNT32 / qctxt->lqc_bunit_sz *
660 factor = MAX_QUOTA_COUNT32 / qctxt->lqc_iunit_sz *
663 LASSERT(!should_translate_quota(qctxt->lqc_import) ||
664 qdata->qd_count <= factor);
665 if (should_translate_quota(qctxt->lqc_import))
667 struct qunit_data_old *reqdata_old, *tmp;
669 reqdata_old = req_capsule_client_get(&req->rq_pill,
672 tmp = lustre_quota_new_to_old(qdata);
674 req_capsule_set_size(&req->rq_pill, &RMF_QUNIT_DATA, RCL_SERVER,
675 sizeof(*reqdata_old));
676 CDEBUG(D_QUOTA, "qd_count is 32bit!\n");
678 reqdata = req_capsule_client_get(&req->rq_pill,
682 req_capsule_set_size(&req->rq_pill, &RMF_QUNIT_DATA, RCL_SERVER,
684 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
686 ptlrpc_request_set_replen(req);
688 CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
689 aa = (struct dqacq_async_args *)&req->rq_async_args;
691 aa->aa_qunit = qunit;
693 req->rq_interpret_reply = dqacq_interpret;
694 ptlrpcd_add_req(req);
696 QDATA_DEBUG(qdata, "%s scheduled.\n",
697 opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
700 struct qunit_data *p = &qunit->lq_data;
701 QDATA_DEBUG(p, "wait for dqacq.\n");
703 l_wait_event(qw.qw_waitq, got_qunit(&qw), &lwi);
707 CDEBUG(D_QUOTA, "wait dqacq done. (rc:%d)\n", qw.qw_rc);
713 qctxt_adjust_qunit(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
714 uid_t uid, gid_t gid, __u32 isblk, int wait)
716 int ret, rc = 0, i = USRQUOTA;
717 __u32 id[MAXQUOTAS] = { uid, gid };
718 struct qunit_data qdata[MAXQUOTAS];
721 CLASSERT(MAXQUOTAS < 4);
722 if (!sb_any_quota_enabled(qctxt->lqc_sb))
725 for (i = 0; i < MAXQUOTAS; i++) {
726 qdata[i].qd_id = id[i];
727 qdata[i].qd_flags = 0;
728 qdata[i].qd_flags |= i;
729 qdata[i].qd_flags |= isblk ? QUOTA_IS_BLOCK : 0;
730 qdata[i].qd_count = 0;
732 ret = check_cur_qunit(obd, qctxt, &qdata[i]);
735 /* need acquire or release */
736 opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
737 ret = split_before_schedule_dqacq(obd, qctxt, &qdata[i],
748 qctxt_wait_pending_dqacq(struct lustre_quota_ctxt *qctxt, unsigned int id,
749 unsigned short type, int isblk)
751 struct lustre_qunit *qunit = NULL;
752 struct qunit_waiter qw;
753 struct qunit_data qdata;
754 struct l_wait_info lwi = { 0 };
757 CFS_INIT_LIST_HEAD(&qw.qw_entry);
758 init_waitqueue_head(&qw.qw_waitq);
763 qdata.qd_flags |= type;
764 qdata.qd_flags |= isblk ? QUOTA_IS_BLOCK : 0;
767 spin_lock(&qunit_hash_lock);
769 qunit = dqacq_in_flight(qctxt, &qdata);
771 list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
773 spin_unlock(&qunit_hash_lock);
776 struct qunit_data *p = &qdata;
777 QDATA_DEBUG(p, "wait for dqacq completion.\n");
778 l_wait_event(qw.qw_waitq, got_qunit(&qw), &lwi);
779 QDATA_DEBUG(p, "wait dqacq done. (rc:%d)\n", qw.qw_rc);
785 qctxt_init(struct lustre_quota_ctxt *qctxt, struct super_block *sb,
786 dqacq_handler_t handler)
791 rc = ptlrpcd_addref();
795 qctxt->lqc_handler = handler;
797 qctxt->lqc_import = NULL;
798 qctxt->lqc_recovery = 0;
799 qctxt->lqc_atype = 0;
800 qctxt->lqc_status= 0;
801 qctxt->lqc_bunit_sz = default_bunit_sz;
802 qctxt->lqc_btune_sz = default_bunit_sz / 100 * default_btune_ratio;
803 qctxt->lqc_iunit_sz = default_iunit_sz;
804 qctxt->lqc_itune_sz = default_iunit_sz * default_itune_ratio / 100;
809 void qctxt_cleanup(struct lustre_quota_ctxt *qctxt, int force)
811 struct lustre_qunit *qunit, *tmp;
812 struct qunit_waiter *qw, *tmp2;
816 spin_lock(&qunit_hash_lock);
818 for (i = 0; i < NR_DQHASH; i++) {
819 list_for_each_entry_safe(qunit, tmp, &qunit_hash[i], lq_hash) {
820 if (qunit->lq_ctxt != qctxt)
823 remove_qunit_nolock(qunit);
824 /* wake up all waiters */
825 list_for_each_entry_safe(qw, tmp2, &qunit->lq_waiters,
827 list_del_init(&qw->qw_entry);
829 wake_up(&qw->qw_waitq);
835 spin_unlock(&qunit_hash_lock);
842 struct qslave_recov_thread_data {
843 struct obd_device *obd;
844 struct lustre_quota_ctxt *qctxt;
845 struct completion comp;
848 /* FIXME only recovery block quota by now */
849 static int qslave_recovery_main(void *arg)
851 struct qslave_recov_thread_data *data = arg;
852 struct obd_device *obd = data->obd;
853 struct lustre_quota_ctxt *qctxt = data->qctxt;
858 ptlrpc_daemonize("qslave_recovd");
860 complete(&data->comp);
862 if (qctxt->lqc_recovery)
864 qctxt->lqc_recovery = 1;
866 for (type = USRQUOTA; type < MAXQUOTAS; type++) {
867 struct qunit_data qdata;
868 struct quota_info *dqopt = sb_dqopt(qctxt->lqc_sb);
869 struct list_head id_list;
870 struct dquot_id *dqid, *tmp;
873 LOCK_DQONOFF_MUTEX(dqopt);
874 if (!sb_has_quota_enabled(qctxt->lqc_sb, type)) {
875 UNLOCK_DQONOFF_MUTEX(dqopt);
879 LASSERT(dqopt->files[type] != NULL);
880 CFS_INIT_LIST_HEAD(&id_list);
881 #ifndef KERNEL_SUPPORTS_QUOTA_READ
882 rc = fsfilt_qids(obd, dqopt->files[type], NULL, type, &id_list);
884 rc = fsfilt_qids(obd, NULL, dqopt->files[type], type, &id_list);
886 UNLOCK_DQONOFF_MUTEX(dqopt);
888 CERROR("Get ids from quota file failed. (rc:%d)\n", rc);
890 list_for_each_entry_safe(dqid, tmp, &id_list, di_link) {
891 list_del_init(&dqid->di_link);
892 /* skip slave recovery on itself */
893 if (is_master(obd, qctxt, dqid->di_id, type))
895 if (rc && rc != -EBUSY)
898 qdata.qd_id = dqid->di_id;
900 qdata.qd_flags |= type;
901 qdata.qd_flags |= QUOTA_IS_BLOCK;
904 ret = check_cur_qunit(obd, qctxt, &qdata);
907 opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
908 rc = split_before_schedule_dqacq(obd, qctxt, &qdata, opc, 0);
913 CDEBUG(rc == -EBUSY ? D_QUOTA : D_ERROR,
914 "qslave recovery failed! (id:%d type:%d "
915 " rc:%d)\n", dqid->di_id, type, rc);
921 qctxt->lqc_recovery = 0;
926 qslave_start_recovery(struct obd_device *obd, struct lustre_quota_ctxt *qctxt)
928 struct qslave_recov_thread_data data;
932 if (!sb_any_quota_enabled(qctxt->lqc_sb))
937 init_completion(&data.comp);
939 rc = kernel_thread(qslave_recovery_main, &data, CLONE_VM|CLONE_FILES);
941 CERROR("Cannot start quota recovery thread: rc %d\n", rc);
944 wait_for_completion(&data.comp);