Whamcloud - gitweb
Branch Head
[fs/lustre-release.git] / lustre / quota / quota_context.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  lustre/quota/quota_context.c
5  *  Lustre Quota Context
6  *
7  *  Copyright (c) 2001-2005 Cluster File Systems, Inc.
8  *   Author: Niu YaWei <niu@clusterfs.com>
9  *
10  *   This file is part of Lustre, http://www.lustre.org.
11  *
12  *   No redistribution or use is permitted outside of Cluster File Systems, Inc.
13  *
14  */
15 #ifndef EXPORT_SYMTAB
16 # define EXPORT_SYMTAB
17 #endif
18
19 #define DEBUG_SUBSYSTEM S_MDS
20
21 #include <linux/version.h>
22 #include <linux/fs.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>
28
29 #include <obd_class.h>
30 #include <lustre_quota.h>
31 #include <lustre_fsfilt.h>
32 #include "quota_internal.h"
33
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 */
38
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;
42
43 struct lustre_qunit {
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 */
50 };
51
52 int should_translate_quota (struct obd_import *imp)
53 {
54         ENTRY;
55
56         LASSERT(imp);
57         if (imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_QUOTA64)
58                 RETURN(0);
59         else
60                 RETURN(1);
61 }
62
63 void qunit_cache_cleanup(void)
64 {
65         int i;
66         ENTRY;
67
68         spin_lock(&qunit_hash_lock);
69         for (i = 0; i < NR_DQHASH; i++)
70                 LASSERT(list_empty(qunit_hash + i));
71         spin_unlock(&qunit_hash_lock);
72
73         if (qunit_cachep) {
74                 int rc;
75                 rc = cfs_mem_cache_destroy(qunit_cachep);
76                 LASSERTF(rc == 0, "couldn't destory qunit_cache slab\n");
77                 qunit_cachep = NULL;
78         }
79         EXIT;
80 }
81
82 int qunit_cache_init(void)
83 {
84         int i;
85         ENTRY;
86
87         LASSERT(qunit_cachep == NULL);
88         qunit_cachep = cfs_mem_cache_create("ll_qunit_cache",
89                                          sizeof(struct lustre_qunit),
90                                          0, 0);
91         if (!qunit_cachep)
92                 RETURN(-ENOMEM);
93
94         spin_lock(&qunit_hash_lock);
95         for (i = 0; i < NR_DQHASH; i++)
96                 INIT_LIST_HEAD(qunit_hash + i);
97         spin_unlock(&qunit_hash_lock);
98         RETURN(0);
99 }
100
101 static inline int
102 qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
103              __attribute__((__const__));
104
105 static inline int
106 qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
107 {
108         unsigned int id = qdata->qd_id;
109         unsigned int type = qdata->qd_flags & QUOTA_IS_GRP;
110
111         unsigned long tmp = ((unsigned long)qctxt >> L1_CACHE_SHIFT) ^ id;
112         tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
113         return tmp;
114 }
115
116 /* compute the remaining quota for certain gid or uid b=11693 */
117 int compute_remquota(struct obd_device *obd,
118                      struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
119 {
120         struct super_block *sb = qctxt->lqc_sb;
121         __u64 usage, limit;
122         struct obd_quotactl *qctl;
123         int ret = QUOTA_RET_OK;
124         __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
125         ENTRY;
126
127         if (!sb_any_quota_enabled(sb))
128                 RETURN(QUOTA_RET_NOQUOTA);
129
130         /* ignore root user */
131         if (qdata->qd_id == 0 && qdata_type == USRQUOTA)
132                 RETURN(QUOTA_RET_NOLIMIT);
133
134         OBD_ALLOC_PTR(qctl);
135         if (qctl == NULL) 
136                 RETURN(-ENOMEM);
137
138         /* get fs quota usage & limit */
139         qctl->qc_cmd = Q_GETQUOTA;
140         qctl->qc_id = qdata->qd_id;
141         qctl->qc_type = qdata_type;
142         ret = fsfilt_quotactl(obd, sb, qctl);
143         if (ret) {
144                 if (ret == -ESRCH)      /* no limit */
145                         ret = QUOTA_RET_NOLIMIT;
146                 else
147                         CDEBUG(D_QUOTA, "can't get fs quota usage! (rc:%d)", 
148                                ret);
149                 GOTO(out, ret);
150         }
151
152         usage = qctl->qc_dqblk.dqb_curspace;
153         limit = qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS;
154         if (!limit){            /* no limit */
155                 ret = QUOTA_RET_NOLIMIT;
156                 GOTO(out, ret);
157         }
158
159         if (limit >= usage)
160                 qdata->qd_count = limit - usage;
161         else
162                 qdata->qd_count = 0;
163         EXIT;
164 out:
165         OBD_FREE_PTR(qctl);
166         return ret;
167 }
168
169 /* caller must hold qunit_hash_lock */
170 static inline struct lustre_qunit *find_qunit(unsigned int hashent,
171                                               struct lustre_quota_ctxt *qctxt,
172                                               struct qunit_data *qdata)
173 {
174         struct lustre_qunit *qunit = NULL;
175         struct qunit_data *tmp;
176
177         LASSERT_SPIN_LOCKED(&qunit_hash_lock);
178         list_for_each_entry(qunit, qunit_hash + hashent, lq_hash) {
179                 tmp = &qunit->lq_data;
180                 if (qunit->lq_ctxt == qctxt &&
181                     qdata->qd_id == tmp->qd_id && qdata->qd_flags == tmp->qd_flags)
182                         return qunit;
183         }
184         return NULL;
185 }
186
187 /* check_cur_qunit - check the current usage of qunit.
188  * @qctxt: quota context
189  * @qdata: the type of quota unit to be checked
190  *
191  * return: 1 - need acquire qunit;
192  *         2 - need release qunit;
193  *         0 - need do nothing.
194  *       < 0 - error.
195  */
196 static int
197 check_cur_qunit(struct obd_device *obd,
198                 struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
199 {
200         struct super_block *sb = qctxt->lqc_sb;
201         unsigned long qunit_sz, tune_sz;
202         __u64 usage, limit;
203         struct obd_quotactl *qctl;
204         int ret = 0;
205         __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
206         __u32 is_blk = (qdata->qd_flags & QUOTA_IS_BLOCK) >> 1;
207         ENTRY;
208
209         if (!sb_any_quota_enabled(sb))
210                 RETURN(0);
211
212         /* ignore root user */
213         if (qdata->qd_id == 0 && qdata_type == USRQUOTA)
214                 RETURN(0);
215
216         OBD_ALLOC_PTR(qctl);
217         if (qctl == NULL)
218                 RETURN(-ENOMEM);
219
220         /* get fs quota usage & limit */
221         qctl->qc_cmd = Q_GETQUOTA;
222         qctl->qc_id = qdata->qd_id;
223         qctl->qc_type = qdata_type;
224         ret = fsfilt_quotactl(obd, sb, qctl);
225         if (ret) {
226                 if (ret == -ESRCH)      /* no limit */
227                         ret = 0;
228                 else
229                         CERROR("can't get fs quota usage! (rc:%d)\n", ret);
230                 GOTO(out, ret);
231         }
232
233         if (is_blk) {
234                 usage = qctl->qc_dqblk.dqb_curspace;
235                 limit = qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS;
236                 qunit_sz = qctxt->lqc_bunit_sz;
237                 tune_sz = qctxt->lqc_btune_sz;
238
239                 LASSERT(!(qunit_sz % QUOTABLOCK_SIZE));
240         } else {
241                 usage = qctl->qc_dqblk.dqb_curinodes;
242                 limit = qctl->qc_dqblk.dqb_ihardlimit;
243                 qunit_sz = qctxt->lqc_iunit_sz;
244                 tune_sz = qctxt->lqc_itune_sz;
245         }
246
247         /* ignore the no quota limit case */
248         if (!limit)
249                 GOTO(out, ret = 0);
250
251         /* we don't count the MIN_QLIMIT */
252         if ((limit == MIN_QLIMIT && !is_blk) ||
253             (toqb(limit) == MIN_QLIMIT && is_blk))
254                 limit = 0;
255
256         LASSERT(qdata->qd_count == 0);
257         if (limit <= usage + tune_sz) {
258                 while (qdata->qd_count + limit <= usage + tune_sz)
259                         qdata->qd_count += qunit_sz;
260                 ret = 1;
261         } else if (limit > usage + qunit_sz + tune_sz) {
262                 while (limit - qdata->qd_count > usage + qunit_sz + tune_sz)
263                         qdata->qd_count += qunit_sz;
264                 ret = 2;
265         }
266         LASSERT(ret == 0 || qdata->qd_count);
267         EXIT;
268 out:
269         OBD_FREE_PTR(qctl);
270         return ret;
271 }
272
273 /* caller must hold qunit_hash_lock */
274 static struct lustre_qunit *dqacq_in_flight(struct lustre_quota_ctxt *qctxt,
275                                             struct qunit_data *qdata)
276 {
277         unsigned int hashent = qunit_hashfn(qctxt, qdata);
278         struct lustre_qunit *qunit;
279         ENTRY;
280
281         LASSERT_SPIN_LOCKED(&qunit_hash_lock);
282         qunit = find_qunit(hashent, qctxt, qdata);
283         RETURN(qunit);
284 }
285
286 static struct lustre_qunit *alloc_qunit(struct lustre_quota_ctxt *qctxt,
287                                         struct qunit_data *qdata, int opc)
288 {
289         struct lustre_qunit *qunit = NULL;
290         ENTRY;
291
292         OBD_SLAB_ALLOC(qunit, qunit_cachep, CFS_ALLOC_IO, sizeof(*qunit));
293         if (qunit == NULL)
294                 RETURN(NULL);
295
296         INIT_LIST_HEAD(&qunit->lq_hash);
297         INIT_LIST_HEAD(&qunit->lq_waiters);
298         atomic_set(&qunit->lq_refcnt, 1);
299         qunit->lq_ctxt = qctxt;
300         memcpy(&qunit->lq_data, qdata, sizeof(*qdata));
301         qunit->lq_opc = opc;
302
303         RETURN(qunit);
304 }
305
306 static inline void free_qunit(struct lustre_qunit *qunit)
307 {
308         OBD_SLAB_FREE(qunit, qunit_cachep, sizeof(*qunit));
309 }
310
311 static inline void qunit_get(struct lustre_qunit *qunit)
312 {
313         atomic_inc(&qunit->lq_refcnt);
314 }
315
316 static void qunit_put(struct lustre_qunit *qunit)
317 {
318         LASSERT(atomic_read(&qunit->lq_refcnt));
319         if (atomic_dec_and_test(&qunit->lq_refcnt))
320                 free_qunit(qunit);
321 }
322
323 static void
324 insert_qunit_nolock(struct lustre_quota_ctxt *qctxt, struct lustre_qunit *qunit)
325 {
326         struct list_head *head;
327
328         LASSERT(list_empty(&qunit->lq_hash));
329         head = qunit_hash + qunit_hashfn(qctxt, &qunit->lq_data);
330         list_add(&qunit->lq_hash, head);
331 }
332
333 static void remove_qunit_nolock(struct lustre_qunit *qunit)
334 {
335         LASSERT(!list_empty(&qunit->lq_hash));
336         list_del_init(&qunit->lq_hash);
337 }
338
339 struct qunit_waiter {
340         struct list_head qw_entry;
341         cfs_waitq_t      qw_waitq;
342         int qw_rc;
343 };
344
345 #define INC_QLIMIT(limit, count) (limit == MIN_QLIMIT) ? \
346                                  (limit = count) : (limit += count)
347
348
349 /* FIXME check if this mds is the master of specified id */
350 static int
351 is_master(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
352           unsigned int id, int type)
353 {
354         return qctxt->lqc_handler ? 1 : 0;
355 }
356
357 static int
358 schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
359                struct qunit_data *qdata, int opc, int wait);
360
361 static int split_before_schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
362                                        struct qunit_data *qdata, int opc, int wait)
363 {
364         int rc = 0, ret;
365         struct qunit_data tmp_qdata;
366         ENTRY;
367
368         LASSERT(qdata);
369         if (qctxt->lqc_import)
370                 while (should_translate_quota(qctxt->lqc_import) &&
371                        qdata->qd_count > MAX_QUOTA_COUNT32) {
372
373                         tmp_qdata = *qdata;
374                         tmp_qdata.qd_count = MAX_QUOTA_COUNT32;
375                         qdata->qd_count -= tmp_qdata.qd_count;
376                         ret = schedule_dqacq(obd, qctxt, &tmp_qdata, opc, wait);
377                         if (!rc)
378                                 rc = ret;
379                 }
380
381         if (qdata->qd_count){
382                 ret = schedule_dqacq(obd, qctxt, qdata, opc, wait);
383                 if (!rc)
384                         rc = ret;
385         }
386
387         RETURN(rc);
388 }
389
390 static int
391 dqacq_completion(struct obd_device *obd,
392                  struct lustre_quota_ctxt *qctxt,
393                  struct qunit_data *qdata, int rc, int opc)
394 {
395         struct lustre_qunit *qunit = NULL;
396         struct super_block *sb = qctxt->lqc_sb;
397         unsigned long qunit_sz;
398         struct qunit_waiter *qw, *tmp;
399         int err = 0;
400         __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
401         __u32 is_blk = (qdata->qd_flags & QUOTA_IS_BLOCK) >> 1;
402         __u64 qd_tmp = qdata->qd_count;
403         unsigned long div_r;
404         ENTRY;
405
406         LASSERT(qdata);
407         qunit_sz = is_blk ? qctxt->lqc_bunit_sz : qctxt->lqc_iunit_sz;
408         div_r = do_div(qd_tmp, qunit_sz);
409         LASSERT(!div_r);
410
411         /* update local operational quota file */
412         if (rc == 0) {
413                 __u32 count = QUSG(qdata->qd_count, is_blk);
414                 struct obd_quotactl *qctl;
415                 __u64 *hardlimit;
416
417                 OBD_ALLOC_PTR(qctl);
418                 if (qctl == NULL)
419                         GOTO(out, err = -ENOMEM);
420
421                 /* acq/rel qunit for specified uid/gid is serialized,
422                  * so there is no race between get fs quota limit and
423                  * set fs quota limit */
424                 qctl->qc_cmd = Q_GETQUOTA;
425                 qctl->qc_id = qdata->qd_id;
426                 qctl->qc_type = qdata_type;
427                 err = fsfilt_quotactl(obd, sb, qctl);
428                 if (err) {
429                         CERROR("error get quota fs limit! (rc:%d)\n", err);
430                         GOTO(out_mem, err);
431                 }
432
433                 if (is_blk) {
434                         qctl->qc_dqblk.dqb_valid = QIF_BLIMITS;
435                         hardlimit = &qctl->qc_dqblk.dqb_bhardlimit;
436                 } else {
437                         qctl->qc_dqblk.dqb_valid = QIF_ILIMITS;
438                         hardlimit = &qctl->qc_dqblk.dqb_ihardlimit;
439                 }
440
441                 switch (opc) {
442                 case QUOTA_DQACQ:
443                         INC_QLIMIT(*hardlimit, count);
444                         break;
445                 case QUOTA_DQREL:
446                         LASSERT(count < *hardlimit);
447                         *hardlimit -= count;
448                         break;
449                 default:
450                         LBUG();
451                 }
452
453                 /* clear quota limit */
454                 if (count == 0)
455                         *hardlimit = 0;
456
457                 qctl->qc_cmd = Q_SETQUOTA;
458                 err = fsfilt_quotactl(obd, sb, qctl);
459                 if (err)
460                         CERROR("error set quota fs limit! (rc:%d)\n", err);
461
462                 QDATA_DEBUG(qdata, "%s completion\n",
463                             opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
464 out_mem:
465                 OBD_FREE_PTR(qctl);
466         } else if (rc == -EDQUOT) {
467                 QDATA_DEBUG(qdata, "acquire qunit got EDQUOT.\n");
468         } else if (rc == -EBUSY) {
469                 QDATA_DEBUG(qdata, "it's is recovering, got EBUSY.\n");
470         } else {
471                 CERROR("acquire qunit got error! (rc:%d)\n", rc);
472         }
473 out:
474         /* remove the qunit from hash */
475         spin_lock(&qunit_hash_lock);
476
477         qunit = dqacq_in_flight(qctxt, qdata);
478         /* this qunit has been removed by qctxt_cleanup() */
479         if (!qunit) {
480                 spin_unlock(&qunit_hash_lock);
481                 RETURN(err);
482         }
483
484         LASSERT(opc == qunit->lq_opc);
485         remove_qunit_nolock(qunit);
486
487         /* wake up all waiters */
488         list_for_each_entry_safe(qw, tmp, &qunit->lq_waiters, qw_entry) {
489                 list_del_init(&qw->qw_entry);
490                 qw->qw_rc = rc;
491                 wake_up(&qw->qw_waitq);
492         }
493
494         spin_unlock(&qunit_hash_lock);
495
496         qunit_put(qunit);
497
498         /* don't reschedule in such cases:
499          *   - acq/rel failure, but not for quota recovery.
500          *   - local dqacq/dqrel.
501          *   - local disk io failure.
502          */
503         if (err || (rc && rc != -EBUSY) || 
504             is_master(obd, qctxt, qdata->qd_id, qdata_type))
505                 RETURN(err);
506
507         /* reschedule another dqacq/dqrel if needed */
508         qdata->qd_count = 0;
509         rc = check_cur_qunit(obd, qctxt, qdata);
510         if (rc > 0) {
511                 int opc;
512                 opc = rc == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
513                 rc = split_before_schedule_dqacq(obd, qctxt, qdata, opc, 0);
514                 QDATA_DEBUG(qdata, "reschedudle opc(%d) rc(%d)\n", opc, rc);
515         }
516         RETURN(err);
517 }
518
519 struct dqacq_async_args {
520         struct lustre_quota_ctxt *aa_ctxt;
521         struct lustre_qunit *aa_qunit;
522 };
523
524 static int dqacq_interpret(struct ptlrpc_request *req, void *data, int rc)
525 {
526         struct dqacq_async_args *aa = (struct dqacq_async_args *)data;
527         struct lustre_quota_ctxt *qctxt = aa->aa_ctxt;
528         struct lustre_qunit *qunit = aa->aa_qunit;
529         struct obd_device *obd = req->rq_import->imp_obd;
530         struct qunit_data *qdata = NULL;
531         struct qunit_data_old *qdata_old = NULL;
532         ENTRY;
533
534         LASSERT(req);
535         LASSERT(req->rq_import);
536         if ((req->rq_import->imp_connect_data.ocd_connect_flags & OBD_CONNECT_QUOTA64)  &&
537             !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT)) {
538                 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
539                 qdata = lustre_swab_reqbuf(req, REPLY_REC_OFF, sizeof(*qdata), lustre_swab_qdata);
540         } else {
541                 CDEBUG(D_QUOTA, "qd_count is 32bit!\n");
542                 qdata_old = lustre_swab_reqbuf(req, REPLY_REC_OFF, sizeof(struct qunit_data_old),
543                                                lustre_swab_qdata_old);
544                 qdata = lustre_quota_old_to_new(qdata_old);
545         }
546         if (qdata == NULL) {
547                 DEBUG_REQ(D_ERROR, req, "error unpacking qunit_data");
548                 RETURN(-EPROTO);
549         }
550
551         LASSERT(qdata->qd_id == qunit->lq_data.qd_id &&
552                 (qdata->qd_flags & QUOTA_IS_GRP) == (qunit->lq_data.qd_flags & QUOTA_IS_GRP) &&
553                 (qdata->qd_count == qunit->lq_data.qd_count ||
554                  qdata->qd_count == 0));
555
556         QDATA_DEBUG(qdata, "%s interpret rc(%d).\n",
557                     lustre_msg_get_opc(req->rq_reqmsg) == QUOTA_DQACQ ?
558                     "DQACQ" : "DQREL", rc);
559
560         rc = dqacq_completion(obd, qctxt, qdata, rc,
561                               lustre_msg_get_opc(req->rq_reqmsg));
562
563         RETURN(rc);
564 }
565
566 static int got_qunit(struct qunit_waiter *waiter)
567 {
568         int rc = 0;
569         ENTRY;
570         spin_lock(&qunit_hash_lock);
571         rc = list_empty(&waiter->qw_entry);
572         spin_unlock(&qunit_hash_lock);
573         RETURN(rc);
574 }
575
576 static int
577 schedule_dqacq(struct obd_device *obd,
578                struct lustre_quota_ctxt *qctxt,
579                struct qunit_data *qdata, int opc, int wait)
580 {
581         struct lustre_qunit *qunit, *empty;
582         struct qunit_waiter qw;
583         struct l_wait_info lwi = { 0 };
584         struct ptlrpc_request *req;
585         struct qunit_data *reqdata;
586         struct dqacq_async_args *aa;
587         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*reqdata) };
588         int rc = 0;
589         ENTRY;
590
591         INIT_LIST_HEAD(&qw.qw_entry);
592         init_waitqueue_head(&qw.qw_waitq);
593         qw.qw_rc = 0;
594
595         if ((empty = alloc_qunit(qctxt, qdata, opc)) == NULL)
596                 RETURN(-ENOMEM);
597
598         spin_lock(&qunit_hash_lock);
599
600         qunit = dqacq_in_flight(qctxt, qdata);
601         if (qunit) {
602                 if (wait)
603                         list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
604                 spin_unlock(&qunit_hash_lock);
605
606                 free_qunit(empty);
607                 goto wait_completion;
608         }
609         qunit = empty;
610         insert_qunit_nolock(qctxt, qunit);
611         if (wait)
612                 list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
613         spin_unlock(&qunit_hash_lock);
614
615         LASSERT(qunit);
616
617         /* master is going to dqacq/dqrel from itself */
618         if (is_master(obd, qctxt, qdata->qd_id, qdata->qd_flags & QUOTA_IS_GRP)) {
619                 int rc2;
620                 QDATA_DEBUG(qdata, "local %s.\n",
621                             opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
622                 rc = qctxt->lqc_handler(obd, qdata, opc);
623                 rc2 = dqacq_completion(obd, qctxt, qdata, rc, opc);
624                 RETURN((rc && rc != -EDQUOT) ? rc : rc2);
625         }
626
627         /* build dqacq/dqrel request */
628         LASSERT(qctxt->lqc_import);
629         req = ptlrpc_prep_req(qctxt->lqc_import, LUSTRE_MDS_VERSION, opc, 2,
630                               size, NULL);
631         if (!req) {
632                 dqacq_completion(obd, qctxt, qdata, -ENOMEM, opc);
633                 RETURN(-ENOMEM);
634         }
635
636         LASSERT(!should_translate_quota(qctxt->lqc_import) || 
637                 qdata->qd_count <= MAX_QUOTA_COUNT32);
638         if (should_translate_quota(qctxt->lqc_import) ||
639             OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT))
640         {
641                 struct qunit_data_old *reqdata_old, *tmp;
642                         
643                 reqdata_old = lustre_msg_buf(req->rq_reqmsg, REPLY_REC_OFF, 
644                                              sizeof(*reqdata_old));
645                 tmp = lustre_quota_new_to_old(qdata);
646                 *reqdata_old = *tmp;
647                 size[1] = sizeof(*reqdata_old);
648                 CDEBUG(D_QUOTA, "qd_count is 32bit!\n");
649         } else {
650                 reqdata = lustre_msg_buf(req->rq_reqmsg, REPLY_REC_OFF,
651                                          sizeof(*reqdata));
652                 *reqdata = *qdata;
653                 size[1] = sizeof(*reqdata);
654                 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
655         }
656         ptlrpc_req_set_repsize(req, 2, size);
657
658         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
659         aa = (struct dqacq_async_args *)&req->rq_async_args;
660         aa->aa_ctxt = qctxt;
661         aa->aa_qunit = qunit;
662
663         req->rq_interpret_reply = dqacq_interpret;
664         ptlrpcd_add_req(req);
665
666         QDATA_DEBUG(qdata, "%s scheduled.\n",
667                     opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
668 wait_completion:
669         if (wait && qunit) {
670                 struct qunit_data *p = &qunit->lq_data;
671                 QDATA_DEBUG(p, "wait for dqacq.\n");
672
673                 l_wait_event(qw.qw_waitq, got_qunit(&qw), &lwi);
674                 if (qw.qw_rc == 0)
675                         rc = -EAGAIN;
676
677                 CDEBUG(D_QUOTA, "wait dqacq done. (rc:%d)\n", qw.qw_rc);
678         }
679         RETURN(rc);
680 }
681
682 int
683 qctxt_adjust_qunit(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
684                    uid_t uid, gid_t gid, __u32 isblk, int wait)
685 {
686         int ret, rc = 0, i = USRQUOTA;
687         __u32 id[MAXQUOTAS] = { uid, gid };
688         struct qunit_data qdata[MAXQUOTAS];
689         ENTRY;
690
691         CLASSERT(MAXQUOTAS < 4);
692         if (!sb_any_quota_enabled(qctxt->lqc_sb))
693                 RETURN(0);
694
695         for (i = 0; i < MAXQUOTAS; i++) {
696                 qdata[i].qd_id = id[i];
697                 qdata[i].qd_flags = 0;
698                 qdata[i].qd_flags |= i;
699                 qdata[i].qd_flags |= isblk ? QUOTA_IS_BLOCK : 0;        
700                 qdata[i].qd_count = 0;
701
702                 ret = check_cur_qunit(obd, qctxt, &qdata[i]);
703                 if (ret > 0) {
704                         int opc;
705                         /* need acquire or release */
706                         opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
707                         ret = split_before_schedule_dqacq(obd, qctxt, &qdata[i], 
708                                                           opc, wait);
709                         if (!rc)
710                                 rc = ret;
711                 }
712         }
713
714         RETURN(rc);
715 }
716
717 int
718 qctxt_wait_pending_dqacq(struct lustre_quota_ctxt *qctxt, unsigned int id,
719                          unsigned short type, int isblk)
720 {
721         struct lustre_qunit *qunit = NULL;
722         struct qunit_waiter qw;
723         struct qunit_data qdata;
724         struct l_wait_info lwi = { 0 };
725         ENTRY;
726
727         INIT_LIST_HEAD(&qw.qw_entry);
728         init_waitqueue_head(&qw.qw_waitq);
729         qw.qw_rc = 0;
730
731         qdata.qd_id = id;
732         qdata.qd_flags = 0;
733         qdata.qd_flags |= type;
734         qdata.qd_flags |= isblk ? QUOTA_IS_BLOCK : 0;
735         qdata.qd_count = 0;
736
737         spin_lock(&qunit_hash_lock);
738
739         qunit = dqacq_in_flight(qctxt, &qdata);
740         if (qunit)
741                 list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
742
743         spin_unlock(&qunit_hash_lock);
744
745         if (qunit) {
746                 struct qunit_data *p = &qdata;
747                 QDATA_DEBUG(p, "wait for dqacq completion.\n");
748                 l_wait_event(qw.qw_waitq, got_qunit(&qw), &lwi);
749                 QDATA_DEBUG(p, "wait dqacq done. (rc:%d)\n", qw.qw_rc);
750         }
751         RETURN(0);
752 }
753
754 int
755 qctxt_init(struct lustre_quota_ctxt *qctxt, struct super_block *sb,
756            dqacq_handler_t handler)
757 {
758         int rc = 0;
759         ENTRY;
760
761         rc = ptlrpcd_addref();
762         if (rc)
763                 RETURN(rc);
764
765         qctxt->lqc_handler = handler;
766         qctxt->lqc_sb = sb;
767         qctxt->lqc_import = NULL;
768         qctxt->lqc_recovery = 0;
769         qctxt->lqc_atype = 0;
770         qctxt->lqc_status= 0;
771         qctxt->lqc_bunit_sz = default_bunit_sz;
772         qctxt->lqc_btune_sz = default_bunit_sz / 100 * default_btune_ratio;
773         qctxt->lqc_iunit_sz = default_iunit_sz;
774         qctxt->lqc_itune_sz = default_iunit_sz * default_itune_ratio / 100;
775
776         RETURN(0);
777 }
778
779 void qctxt_cleanup(struct lustre_quota_ctxt *qctxt, int force)
780 {
781         struct lustre_qunit *qunit, *tmp;
782         struct qunit_waiter *qw, *tmp2;
783         int i;
784         ENTRY;
785
786         spin_lock(&qunit_hash_lock);
787
788         for (i = 0; i < NR_DQHASH; i++) {
789                 list_for_each_entry_safe(qunit, tmp, &qunit_hash[i], lq_hash) {
790                         if (qunit->lq_ctxt != qctxt)
791                                 continue;
792
793                         remove_qunit_nolock(qunit);
794                         /* wake up all waiters */
795                         list_for_each_entry_safe(qw, tmp2, &qunit->lq_waiters,
796                                                  qw_entry) {
797                                 list_del_init(&qw->qw_entry);
798                                 qw->qw_rc = 0;
799                                 wake_up(&qw->qw_waitq);
800                         }
801                         qunit_put(qunit);
802                 }
803         }
804
805         spin_unlock(&qunit_hash_lock);
806
807         ptlrpcd_decref();
808
809         EXIT;
810 }
811
812 struct qslave_recov_thread_data {
813         struct obd_device *obd;
814         struct lustre_quota_ctxt *qctxt;
815         struct completion comp;
816 };
817
818 /* FIXME only recovery block quota by now */
819 static int qslave_recovery_main(void *arg)
820 {
821         struct qslave_recov_thread_data *data = arg;
822         struct obd_device *obd = data->obd;
823         struct lustre_quota_ctxt *qctxt = data->qctxt;
824         unsigned int type;
825         int rc = 0;
826         ENTRY;
827
828         ptlrpc_daemonize("qslave_recovd");
829
830         complete(&data->comp);
831
832         if (qctxt->lqc_recovery)
833                 RETURN(0);
834         qctxt->lqc_recovery = 1;
835
836         for (type = USRQUOTA; type < MAXQUOTAS; type++) {
837                 struct qunit_data qdata;
838                 struct quota_info *dqopt = sb_dqopt(qctxt->lqc_sb);
839                 struct list_head id_list;
840                 struct dquot_id *dqid, *tmp;
841                 int ret;
842
843                 LOCK_DQONOFF_MUTEX(dqopt);
844                 if (!sb_has_quota_enabled(qctxt->lqc_sb, type)) {
845                         UNLOCK_DQONOFF_MUTEX(dqopt);
846                         break;
847                 }
848
849                 LASSERT(dqopt->files[type] != NULL);
850                 INIT_LIST_HEAD(&id_list);
851 #ifndef KERNEL_SUPPORTS_QUOTA_READ 
852                 rc = fsfilt_qids(obd, dqopt->files[type], NULL, type, &id_list);
853 #else
854                 rc = fsfilt_qids(obd, NULL, dqopt->files[type], type, &id_list);
855 #endif
856                 UNLOCK_DQONOFF_MUTEX(dqopt);
857                 if (rc)
858                         CERROR("Get ids from quota file failed. (rc:%d)\n", rc);
859
860                 list_for_each_entry_safe(dqid, tmp, &id_list, di_link) {
861                         list_del_init(&dqid->di_link);
862                         /* skip slave recovery on itself */
863                         if (is_master(obd, qctxt, dqid->di_id, type))
864                                 goto free;
865                         if (rc && rc != -EBUSY)
866                                 goto free;
867
868                         qdata.qd_id = dqid->di_id;
869                         qdata.qd_flags = 0;
870                         qdata.qd_flags |= type;
871                         qdata.qd_flags |= QUOTA_IS_BLOCK;
872                         qdata.qd_count = 0;
873
874                         ret = check_cur_qunit(obd, qctxt, &qdata);
875                         if (ret > 0) {
876                                 int opc;
877                                 opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
878                                 rc = split_before_schedule_dqacq(obd, qctxt, &qdata, opc, 0);
879                         } else
880                                 rc = 0;
881
882                         if (rc)
883                                 CDEBUG(rc == -EBUSY ? D_QUOTA : D_ERROR,
884                                        "qslave recovery failed! (id:%d type:%d "
885                                        " rc:%d)\n", dqid->di_id, type, rc);
886 free:
887                         kfree(dqid);
888                 }
889         }
890
891         qctxt->lqc_recovery = 0;
892         RETURN(rc);
893 }
894
895 void
896 qslave_start_recovery(struct obd_device *obd, struct lustre_quota_ctxt *qctxt)
897 {
898         struct qslave_recov_thread_data data;
899         int rc;
900         ENTRY;
901
902         if (!sb_any_quota_enabled(qctxt->lqc_sb))
903                 goto exit;
904
905         data.obd = obd;
906         data.qctxt = qctxt;
907         init_completion(&data.comp);
908
909         rc = kernel_thread(qslave_recovery_main, &data, CLONE_VM|CLONE_FILES);
910         if (rc < 0) {
911                 CERROR("Cannot start quota recovery thread: rc %d\n", rc);
912                 goto exit;
913         }
914         wait_for_completion(&data.comp);
915 exit:
916         EXIT;
917 }
918