Whamcloud - gitweb
5cf9505bd8907f6104e81118f97c8a7c257fb784
[fs/lustre-release.git] / lustre / quota / quota_interface.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  */
36
37 #ifndef EXPORT_SYMTAB
38 # define EXPORT_SYMTAB
39 #endif
40 #define DEBUG_SUBSYSTEM S_LQUOTA
41
42 #ifdef __KERNEL__
43 # include <linux/version.h>
44 # include <linux/module.h>
45 # include <linux/init.h>
46 # include <linux/fs.h>
47 # include <linux/jbd.h>
48 # if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
49 #  include <linux/smp_lock.h>
50 #  include <linux/buffer_head.h>
51 #  include <linux/workqueue.h>
52 #  include <linux/mount.h>
53 # else
54 #  include <linux/locks.h>
55 # endif
56 #else /* __KERNEL__ */
57 # include <liblustre.h>
58 #endif
59
60 #include <obd_class.h>
61 #include <lustre_mds.h>
62 #include <lustre_dlm.h>
63 #include <lustre_cfg.h>
64 #include <obd_ost.h>
65 #include <lustre_fsfilt.h>
66 #include <lustre_quota.h>
67 #include <lprocfs_status.h>
68 #include "quota_internal.h"
69
70 #ifdef __KERNEL__
71
72 #ifdef HAVE_QUOTA_SUPPORT
73
74 static cfs_time_t last_print = 0;
75 static spinlock_t last_print_lock = SPIN_LOCK_UNLOCKED;
76
77 static int filter_quota_setup(struct obd_device *obd)
78 {
79         int rc = 0;
80         struct obd_device_target *obt = &obd->u.obt;
81         ENTRY;
82
83         init_rwsem(&obt->obt_rwsem);
84         obt->obt_qfmt = LUSTRE_QUOTA_V2;
85         atomic_set(&obt->obt_quotachecking, 1);
86         rc = qctxt_init(obd, NULL);
87         if (rc)
88                 CERROR("initialize quota context failed! (rc:%d)\n", rc);
89
90         RETURN(rc);
91 }
92
93 static int filter_quota_cleanup(struct obd_device *obd)
94 {
95         ENTRY;
96         qctxt_cleanup(&obd->u.obt.obt_qctxt, 0);
97         RETURN(0);
98 }
99
100 static int filter_quota_setinfo(struct obd_device *obd, void *data)
101 {
102         struct obd_export *exp = data;
103         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
104         struct obd_import *imp;
105         ENTRY;
106
107         /* setup the quota context import */
108         spin_lock(&qctxt->lqc_lock);
109         qctxt->lqc_import = exp->exp_imp_reverse;
110         spin_unlock(&qctxt->lqc_lock);
111         CDEBUG(D_QUOTA, "%s: lqc_import(%p) of obd(%p) is reactivated now, \n",
112                obd->obd_name,exp->exp_imp_reverse, obd);
113
114         /* make imp's connect flags equal relative exp's connect flags
115          * adding it to avoid the scan export list
116          */
117         imp = qctxt->lqc_import;
118         if (likely(imp))
119                 imp->imp_connect_data.ocd_connect_flags |=
120                         (exp->exp_connect_flags &
121                          (OBD_CONNECT_QUOTA64 | OBD_CONNECT_CHANGE_QS));
122
123         cfs_waitq_signal(&qctxt->lqc_wait_for_qmaster);
124         /* start quota slave recovery thread. (release high limits) */
125         qslave_start_recovery(obd, qctxt);
126         RETURN(0);
127 }
128
129 static int filter_quota_clearinfo(struct obd_export *exp, struct obd_device *obd)
130 {
131         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
132         ENTRY;
133
134         /* lquota may be not set up before destroying export, b=14896 */
135         if (!obd->obd_set_up)
136                 RETURN(0);
137
138         /* when exp->exp_imp_reverse is destroyed, the corresponding lqc_import
139          * should be invalid b=12374 */
140         if (qctxt->lqc_import && qctxt->lqc_import == exp->exp_imp_reverse) {
141                 spin_lock(&qctxt->lqc_lock);
142                 qctxt->lqc_import = NULL;
143                 spin_unlock(&qctxt->lqc_lock);
144                 CDEBUG(D_QUOTA, "%s: lqc_import of obd(%p) is invalid now.\n",
145                        obd->obd_name, obd);
146         }
147         RETURN(0);
148 }
149
150 static int filter_quota_enforce(struct obd_device *obd, unsigned int ignore)
151 {
152         ENTRY;
153
154         if (!sb_any_quota_enabled(obd->u.obt.obt_sb))
155                 RETURN(0);
156
157         if (ignore) {
158                 CDEBUG(D_QUOTA, "blocks will be written with ignoring quota.\n");
159                 cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
160         } else {
161                 cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
162         }
163
164         RETURN(0);
165 }
166
167 static int filter_quota_getflag(struct obd_device *obd, struct obdo *oa)
168 {
169         struct obd_device_target *obt = &obd->u.obt;
170         struct lustre_quota_ctxt *qctxt = &obt->obt_qctxt;
171         int err, cnt, rc = 0;
172         struct obd_quotactl *oqctl;
173         ENTRY;
174
175         if (!sb_any_quota_enabled(obt->obt_sb))
176                 RETURN(0);
177
178         OBD_ALLOC_PTR(oqctl);
179         if (!oqctl) {
180                 CERROR("Not enough memory!");
181                 RETURN(-ENOMEM);
182         }
183
184         /* set over quota flags for a uid/gid */
185         oa->o_valid |= OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA;
186         oa->o_flags &= ~(OBD_FL_NO_USRQUOTA | OBD_FL_NO_GRPQUOTA);
187
188         for (cnt = 0; cnt < MAXQUOTAS; cnt++) {
189                 struct quota_adjust_qunit oqaq_tmp;
190                 struct lustre_qunit_size *lqs = NULL;
191
192                 oqaq_tmp.qaq_flags = cnt;
193                 oqaq_tmp.qaq_id = (cnt == USRQUOTA) ? oa->o_uid : oa->o_gid;
194
195                 quota_search_lqs(NULL, &oqaq_tmp, qctxt, &lqs);
196                 if (lqs) {
197                         spin_lock(&lqs->lqs_lock);
198                         if (lqs->lqs_bunit_sz <= qctxt->lqc_sync_blk) {
199                                 oa->o_flags |= (cnt == USRQUOTA) ?
200                                         OBD_FL_NO_USRQUOTA : OBD_FL_NO_GRPQUOTA;
201                                 spin_unlock(&lqs->lqs_lock);
202                                 CDEBUG(D_QUOTA, "set sync flag: bunit(%lu), "
203                                        "sync_blk(%d)\n", lqs->lqs_bunit_sz,
204                                        qctxt->lqc_sync_blk);
205                                 /* this is for quota_search_lqs */
206                                 lqs_putref(lqs);
207                                 continue;
208                         }
209                         spin_unlock(&lqs->lqs_lock);
210                         /* this is for quota_search_lqs */
211                         lqs_putref(lqs);
212                 }
213
214                 memset(oqctl, 0, sizeof(*oqctl));
215
216                 oqctl->qc_cmd = Q_GETQUOTA;
217                 oqctl->qc_type = cnt;
218                 oqctl->qc_id = (cnt == USRQUOTA) ? oa->o_uid : oa->o_gid;
219                 err = fsfilt_quotactl(obd, obt->obt_sb, oqctl);
220                 if (err) {
221                         if (!rc)
222                                 rc = err;
223                         oa->o_valid &= ~((cnt == USRQUOTA) ? OBD_MD_FLUSRQUOTA :
224                                                              OBD_MD_FLGRPQUOTA);
225                         continue;
226                 }
227
228                 if (oqctl->qc_dqblk.dqb_bhardlimit &&
229                    (toqb(oqctl->qc_dqblk.dqb_curspace) >=
230                     oqctl->qc_dqblk.dqb_bhardlimit))
231                         oa->o_flags |= (cnt == USRQUOTA) ?
232                                 OBD_FL_NO_USRQUOTA : OBD_FL_NO_GRPQUOTA;
233         }
234         OBD_FREE_PTR(oqctl);
235         RETURN(rc);
236 }
237
238 /**
239  * check whether the left quota of certain uid and gid can satisfy a block_write
240  * or inode_create rpc. When need to acquire quota, return QUOTA_RET_ACQUOTA
241  */
242 static int quota_check_common(struct obd_device *obd, unsigned int uid,
243                               unsigned int gid, int count, int cycle, int isblk,
244                               struct inode *inode, int frags, int *pending)
245 {
246         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
247         int i;
248         __u32 id[MAXQUOTAS] = { uid, gid };
249         struct qunit_data qdata[MAXQUOTAS];
250         int mb = 0;
251         int rc = 0, rc2[2] = { 0, 0 };
252         ENTRY;
253
254         CLASSERT(MAXQUOTAS < 4);
255         if (!sb_any_quota_enabled(qctxt->lqc_sb))
256                 RETURN(rc);
257
258         spin_lock(&qctxt->lqc_lock);
259         if (!qctxt->lqc_valid){
260                 spin_unlock(&qctxt->lqc_lock);
261                 RETURN(rc);
262         }
263         spin_unlock(&qctxt->lqc_lock);
264
265         for (i = 0; i < MAXQUOTAS; i++) {
266                 struct lustre_qunit_size *lqs = NULL;
267
268                 qdata[i].qd_id = id[i];
269                 qdata[i].qd_flags = i;
270                 if (isblk)
271                         QDATA_SET_BLK(&qdata[i]);
272                 qdata[i].qd_count = 0;
273
274                 /* ignore root user */
275                 if (qdata[i].qd_id == 0 && !QDATA_IS_GRP(&qdata[i]))
276                         continue;
277
278                 quota_search_lqs(&qdata[i], NULL, qctxt, &lqs);
279                 if (!lqs)
280                         continue;
281
282                 rc2[i] = compute_remquota(obd, qctxt, &qdata[i], isblk);
283                 spin_lock(&lqs->lqs_lock);
284                 if (!cycle) {
285                         if (isblk) {
286                                 *pending = count * CFS_PAGE_SIZE;
287                                 /* in order to complete this write, we need extra
288                                  * meta blocks. This function can get it through
289                                  * data needed to be written b=16542 */
290                                 mb = *pending;
291                                 LASSERT(inode && frags > 0);
292                                 if (fsfilt_get_mblk(obd, qctxt->lqc_sb, &mb,
293                                                     inode, frags) < 0)
294                                         CDEBUG(D_ERROR,
295                                                "can't get extra meta blocks.\n");
296                                 else
297                                         *pending += mb;
298                                 lqs->lqs_bwrite_pending += *pending;
299                         } else {
300                                 *pending = count;
301                                 lqs->lqs_iwrite_pending += *pending;
302                         }
303                 }
304                 if (rc2[i] == QUOTA_RET_OK) {
305                         if (isblk && qdata[i].qd_count < lqs->lqs_bwrite_pending)
306                                 rc2[i] = QUOTA_RET_ACQUOTA;
307                         if (!isblk && qdata[i].qd_count <
308                             lqs->lqs_iwrite_pending)
309                                 rc2[i] = QUOTA_RET_ACQUOTA;
310                 }
311                 spin_unlock(&lqs->lqs_lock);
312                 CDEBUG(D_QUOTA, "count: %d, lqs pending: %lu, qd_count: "LPU64
313                        ", metablocks: %d, isblk: %d, pending: %d.\n", count,
314                        isblk ? lqs->lqs_bwrite_pending : lqs->lqs_iwrite_pending,
315                        qdata[i].qd_count, mb, isblk, *pending);
316
317                 /* When cycle is zero, lqs_*_pending will be changed. We will
318                  * get reference of the lqs here and put reference of lqs in
319                  * quota_pending_commit b=14784 */
320                 if (!cycle)
321                         lqs_getref(lqs);
322
323                 /* this is for quota_search_lqs */
324                 lqs_putref(lqs);
325         }
326
327         if (rc2[0] == QUOTA_RET_ACQUOTA || rc2[1] == QUOTA_RET_ACQUOTA)
328                 RETURN(QUOTA_RET_ACQUOTA);
329         else
330                 RETURN(rc);
331 }
332
333 static int quota_chk_acq_common(struct obd_device *obd, unsigned int uid,
334                                 unsigned int gid, int count, int *pending,
335                                 quota_acquire acquire,
336                                 struct obd_trans_info *oti, int isblk,
337                                 struct inode *inode, int frags)
338 {
339         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
340         struct timeval work_start;
341         struct timeval work_end;
342         long timediff;
343         struct l_wait_info lwi = { 0 };
344         int rc = 0, cycle = 0, count_err = 1;
345         ENTRY;
346
347         CDEBUG(D_QUOTA, "check quota for %s\n", obd->obd_name);
348         *pending = 0;
349         /* Unfortunately, if quota master is too busy to handle the
350          * pre-dqacq in time and quota hash on ost is used up, we
351          * have to wait for the completion of in flight dqacq/dqrel,
352          * in order to get enough quota for write b=12588 */
353         do_gettimeofday(&work_start);
354         while ((rc = quota_check_common(obd, uid, gid, count, cycle, isblk,
355                                         inode, frags, pending)) &
356                QUOTA_RET_ACQUOTA) {
357
358                 spin_lock(&qctxt->lqc_lock);
359                 if (!qctxt->lqc_import && oti) {
360                         spin_unlock(&qctxt->lqc_lock);
361
362                         LASSERT(oti && oti->oti_thread &&
363                                 oti->oti_thread->t_watchdog);
364
365                         lc_watchdog_disable(oti->oti_thread->t_watchdog);
366                         CDEBUG(D_QUOTA, "sleep for quota master\n");
367                         l_wait_event(qctxt->lqc_wait_for_qmaster, check_qm(qctxt),
368                                      &lwi);
369                         CDEBUG(D_QUOTA, "wake up when quota master is back\n");
370                         lc_watchdog_touch(oti->oti_thread->t_watchdog);
371                 } else {
372                         spin_unlock(&qctxt->lqc_lock);
373                 }
374
375                 cycle++;
376                 if (isblk)
377                         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_HOLD_WRITE_RPC, 90);
378                 /* after acquire(), we should run quota_check_common again
379                  * so that we confirm there are enough quota to finish write */
380                 rc = acquire(obd, uid, gid, oti, isblk);
381
382                 /* please reference to dqacq_completion for the below */
383                 /* a new request is finished, try again */
384                 if (rc == -EAGAIN) {
385                         CDEBUG(D_QUOTA, "finish a quota req, try again\n");
386                         continue;
387                 }
388
389                 /* it is out of quota already */
390                 if (rc == -EDQUOT) {
391                         CDEBUG(D_QUOTA, "out of quota,  return -EDQUOT\n");
392                         break;
393                 }
394
395                 /* -EBUSY and others, wait a second and try again */
396                 if (rc < 0) {
397                         cfs_waitq_t        waitq;
398                         struct l_wait_info lwi;
399
400                         if (oti && oti->oti_thread && oti->oti_thread->t_watchdog)
401                                 lc_watchdog_touch(oti->oti_thread->t_watchdog);
402                         CDEBUG(D_QUOTA, "rc: %d, count_err: %d\n", rc,
403                                count_err++);
404
405                         init_waitqueue_head(&waitq);
406                         lwi = LWI_TIMEOUT(cfs_time_seconds(min(cycle, 10)), NULL,
407                                           NULL);
408                         l_wait_event(waitq, 0, &lwi);
409                 }
410
411                 if (rc < 0 || cycle % 10 == 2) {
412                         spin_lock(&last_print_lock);
413                         if (last_print == 0 ||
414                             cfs_time_before((last_print + cfs_time_seconds(30)),
415                                             cfs_time_current())) {
416                                 last_print = cfs_time_current();
417                                 spin_unlock(&last_print_lock);
418                                 CWARN("still haven't managed to acquire quota "
419                                       "space from the quota master after %d "
420                                       "retries (err=%d, rc=%d)\n",
421                                       cycle, count_err - 1, rc);
422                         } else {
423                                 spin_unlock(&last_print_lock);
424                         }
425                 }
426
427                 CDEBUG(D_QUOTA, "recheck quota with rc: %d, cycle: %d\n", rc,
428                        cycle);
429         }
430         do_gettimeofday(&work_end);
431         timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
432         lprocfs_counter_add(qctxt->lqc_stats,
433                             isblk ? LQUOTA_WAIT_FOR_CHK_BLK :
434                                     LQUOTA_WAIT_FOR_CHK_INO,
435                             timediff);
436
437         RETURN(rc);
438 }
439
440 /**
441  * when a block_write or inode_create rpc is finished, adjust the record for
442  * pending blocks and inodes
443  */
444 static int quota_pending_commit(struct obd_device *obd, unsigned int uid,
445                                 unsigned int gid, int pending, int isblk)
446 {
447         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
448         struct timeval work_start;
449         struct timeval work_end;
450         long timediff;
451         int i;
452         __u32 id[MAXQUOTAS] = { uid, gid };
453         struct qunit_data qdata[MAXQUOTAS];
454         ENTRY;
455
456         CDEBUG(D_QUOTA, "commit pending quota for  %s\n", obd->obd_name);
457         CLASSERT(MAXQUOTAS < 4);
458         if (!sb_any_quota_enabled(qctxt->lqc_sb))
459                 RETURN(0);
460
461         do_gettimeofday(&work_start);
462         for (i = 0; i < MAXQUOTAS; i++) {
463                 struct lustre_qunit_size *lqs = NULL;
464
465                 qdata[i].qd_id = id[i];
466                 qdata[i].qd_flags = i;
467                 if (isblk)
468                         QDATA_SET_BLK(&qdata[i]);
469                 qdata[i].qd_count = 0;
470
471                 if (qdata[i].qd_id == 0 && !QDATA_IS_GRP(&qdata[i]))
472                         continue;
473
474                 quota_search_lqs(&qdata[i], NULL, qctxt, &lqs);
475                 if (lqs) {
476                         int flag = 0;
477                         spin_lock(&lqs->lqs_lock);
478                         if (isblk) {
479                                 if (lqs->lqs_bwrite_pending >= pending) {
480                                         lqs->lqs_bwrite_pending -= pending;
481                                         spin_unlock(&lqs->lqs_lock);
482                                         flag = 1;
483                                 } else {
484                                         spin_unlock(&lqs->lqs_lock);
485                                         CDEBUG(D_ERROR,
486                                                "there are too many blocks!\n");
487                                 }
488                         } else {
489                                 if (lqs->lqs_iwrite_pending >= pending) {
490                                         lqs->lqs_iwrite_pending -= pending;
491                                         spin_unlock(&lqs->lqs_lock);
492                                         flag = 1;
493                                 } else {
494                                         spin_unlock(&lqs->lqs_lock);
495                                         CDEBUG(D_ERROR,
496                                                "there are too many files!\n");
497                                 }
498                         }
499                         CDEBUG(D_QUOTA, "lqs pending: %lu, pending: %d, "
500                                "isblk: %d.\n",
501                                isblk ? lqs->lqs_bwrite_pending :
502                                lqs->lqs_iwrite_pending, pending, isblk);
503
504                         lqs_putref(lqs);
505                         /* When lqs_*_pening is changed back, we'll putref lqs
506                          * here b=14784 */
507                         if (flag)
508                                 lqs_putref(lqs);
509                 }
510         }
511         do_gettimeofday(&work_end);
512         timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
513         lprocfs_counter_add(qctxt->lqc_stats,
514                             isblk ? LQUOTA_WAIT_FOR_COMMIT_BLK :
515                                     LQUOTA_WAIT_FOR_COMMIT_INO,
516                             timediff);
517
518         RETURN(0);
519 }
520
521 static int mds_quota_init(void)
522 {
523         return lustre_dquot_init();
524 }
525
526 static int mds_quota_exit(void)
527 {
528         lustre_dquot_exit();
529         return 0;
530 }
531
532 static int mds_quota_setup(struct obd_device *obd)
533 {
534         struct obd_device_target *obt = &obd->u.obt;
535         struct mds_obd *mds = &obd->u.mds;
536         int rc;
537         ENTRY;
538
539         if (unlikely(mds->mds_quota)) {
540                 CWARN("try to reinitialize quota context!\n");
541                 RETURN(0);
542         }
543
544         init_rwsem(&obt->obt_rwsem);
545         obt->obt_qfmt = LUSTRE_QUOTA_V2;
546         mds->mds_quota_info.qi_version = LUSTRE_QUOTA_V2;
547         atomic_set(&obt->obt_quotachecking, 1);
548         /* initialize quota master and quota context */
549         sema_init(&mds->mds_qonoff_sem, 1);
550         rc = qctxt_init(obd, dqacq_handler);
551         if (rc) {
552                 CERROR("initialize quota context failed! (rc:%d)\n", rc);
553                 RETURN(rc);
554         }
555         mds->mds_quota = 1;
556         RETURN(rc);
557 }
558
559 static int mds_quota_cleanup(struct obd_device *obd)
560 {
561         ENTRY;
562         if (unlikely(!obd->u.mds.mds_quota))
563                 RETURN(0);
564
565         qctxt_cleanup(&obd->u.obt.obt_qctxt, 0);
566         RETURN(0);
567 }
568
569 static int mds_quota_setinfo(struct obd_device *obd, void *data)
570 {
571         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
572         ENTRY;
573
574         if (unlikely(!obd->u.mds.mds_quota))
575                 RETURN(0);
576
577         if (data != NULL)
578                 QUOTA_MASTER_READY(qctxt);
579         else
580                 QUOTA_MASTER_UNREADY(qctxt);
581         RETURN(0);
582 }
583
584 static int mds_quota_fs_cleanup(struct obd_device *obd)
585 {
586         struct mds_obd *mds = &obd->u.mds;
587         struct obd_quotactl oqctl;
588         ENTRY;
589
590         if (unlikely(!mds->mds_quota))
591                 RETURN(0);
592
593         mds->mds_quota = 0;
594         memset(&oqctl, 0, sizeof(oqctl));
595         oqctl.qc_type = UGQUOTA;
596
597         down(&mds->mds_qonoff_sem);
598         mds_admin_quota_off(obd, &oqctl);
599         up(&mds->mds_qonoff_sem);
600         RETURN(0);
601 }
602
603 static int quota_acquire_common(struct obd_device *obd, unsigned int uid,
604                                 unsigned int gid, struct obd_trans_info *oti,
605                                 int isblk)
606 {
607         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
608         int rc;
609         ENTRY;
610
611         rc = qctxt_adjust_qunit(obd, qctxt, uid, gid, isblk, 1, oti);
612         RETURN(rc);
613 }
614
615 #endif /* HAVE_QUOTA_SUPPORT */
616 #endif /* __KERNEL__ */
617
618 struct osc_quota_info {
619         struct list_head        oqi_hash;       /* hash list */
620         struct client_obd      *oqi_cli;        /* osc obd */
621         unsigned int            oqi_id;         /* uid/gid of a file */
622         short                   oqi_type;       /* quota type */
623 };
624
625 spinlock_t qinfo_list_lock = SPIN_LOCK_UNLOCKED;
626
627 static struct list_head qinfo_hash[NR_DQHASH];
628 /* SLAB cache for client quota context */
629 cfs_mem_cache_t *qinfo_cachep = NULL;
630
631 static inline int hashfn(struct client_obd *cli, unsigned long id, int type)
632                          __attribute__((__const__));
633
634 static inline int hashfn(struct client_obd *cli, unsigned long id, int type)
635 {
636         unsigned long tmp = ((unsigned long)cli>>6) ^ id;
637         tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
638         return tmp;
639 }
640
641 /* caller must hold qinfo_list_lock */
642 static inline void insert_qinfo_hash(struct osc_quota_info *oqi)
643 {
644         struct list_head *head = qinfo_hash +
645                 hashfn(oqi->oqi_cli, oqi->oqi_id, oqi->oqi_type);
646
647         LASSERT_SPIN_LOCKED(&qinfo_list_lock);
648         list_add(&oqi->oqi_hash, head);
649 }
650
651 /* caller must hold qinfo_list_lock */
652 static inline void remove_qinfo_hash(struct osc_quota_info *oqi)
653 {
654         LASSERT_SPIN_LOCKED(&qinfo_list_lock);
655         list_del_init(&oqi->oqi_hash);
656 }
657
658 /* caller must hold qinfo_list_lock */
659 static inline struct osc_quota_info *find_qinfo(struct client_obd *cli,
660                                                 unsigned int id, int type)
661 {
662         unsigned int hashent = hashfn(cli, id, type);
663         struct osc_quota_info *oqi;
664         ENTRY;
665
666         LASSERT_SPIN_LOCKED(&qinfo_list_lock);
667         list_for_each_entry(oqi, &qinfo_hash[hashent], oqi_hash) {
668                 if (oqi->oqi_cli == cli &&
669                     oqi->oqi_id == id && oqi->oqi_type == type)
670                         return oqi;
671         }
672         RETURN(NULL);
673 }
674
675 static struct osc_quota_info *alloc_qinfo(struct client_obd *cli,
676                                           unsigned int id, int type)
677 {
678         struct osc_quota_info *oqi;
679         ENTRY;
680
681         OBD_SLAB_ALLOC(oqi, qinfo_cachep, CFS_ALLOC_STD, sizeof(*oqi));
682         if(!oqi)
683                 RETURN(NULL);
684
685         CFS_INIT_LIST_HEAD(&oqi->oqi_hash);
686         oqi->oqi_cli = cli;
687         oqi->oqi_id = id;
688         oqi->oqi_type = type;
689
690         RETURN(oqi);
691 }
692
693 static void free_qinfo(struct osc_quota_info *oqi)
694 {
695         OBD_SLAB_FREE(oqi, qinfo_cachep, sizeof(*oqi));
696 }
697
698 int osc_quota_chkdq(struct client_obd *cli, unsigned int uid, unsigned int gid)
699 {
700         unsigned int id;
701         int cnt, rc = QUOTA_OK;
702         ENTRY;
703
704         spin_lock(&qinfo_list_lock);
705         for (cnt = 0; cnt < MAXQUOTAS; cnt++) {
706                 struct osc_quota_info *oqi = NULL;
707
708                 id = (cnt == USRQUOTA) ? uid : gid;
709                 oqi = find_qinfo(cli, id, cnt);
710                 if (oqi) {
711                         rc = NO_QUOTA;
712                         break;
713                 }
714         }
715         spin_unlock(&qinfo_list_lock);
716
717         RETURN(rc);
718 }
719
720 int osc_quota_setdq(struct client_obd *cli, unsigned int uid, unsigned int gid,
721                     obd_flag valid, obd_flag flags)
722 {
723         unsigned int id;
724         obd_flag noquota;
725         int cnt, rc = 0;
726         ENTRY;
727
728
729         for (cnt = 0; cnt < MAXQUOTAS; cnt++) {
730                 struct osc_quota_info *oqi, *old;
731
732                 if (!(valid & ((cnt == USRQUOTA) ?
733                     OBD_MD_FLUSRQUOTA : OBD_MD_FLGRPQUOTA)))
734                         continue;
735
736                 id = (cnt == USRQUOTA) ? uid : gid;
737                 noquota = (cnt == USRQUOTA) ?
738                     (flags & OBD_FL_NO_USRQUOTA) : (flags & OBD_FL_NO_GRPQUOTA);
739
740                 oqi = alloc_qinfo(cli, id, cnt);
741                 if (oqi) {
742                         spin_lock(&qinfo_list_lock);
743
744                         old = find_qinfo(cli, id, cnt);
745                         if (old && !noquota)
746                                 remove_qinfo_hash(old);
747                         else if (!old && noquota)
748                                 insert_qinfo_hash(oqi);
749
750                         spin_unlock(&qinfo_list_lock);
751
752                         if (old || !noquota)
753                                 free_qinfo(oqi);
754                         if (old && !noquota)
755                                 free_qinfo(old);
756                 } else {
757                         CERROR("not enough mem!\n");
758                         rc = -ENOMEM;
759                         break;
760                 }
761         }
762
763         RETURN(rc);
764 }
765
766 int osc_quota_cleanup(struct obd_device *obd)
767 {
768         struct client_obd *cli = &obd->u.cli;
769         struct osc_quota_info *oqi, *n;
770         int i;
771         ENTRY;
772
773         spin_lock(&qinfo_list_lock);
774         for (i = 0; i < NR_DQHASH; i++) {
775                 list_for_each_entry_safe(oqi, n, &qinfo_hash[i], oqi_hash) {
776                         if (oqi->oqi_cli != cli)
777                                 continue;
778                         remove_qinfo_hash(oqi);
779                         free_qinfo(oqi);
780                 }
781         }
782         spin_unlock(&qinfo_list_lock);
783
784         RETURN(0);
785 }
786
787 int osc_quota_init(void)
788 {
789         int i;
790         ENTRY;
791
792         LASSERT(qinfo_cachep == NULL);
793         qinfo_cachep = cfs_mem_cache_create("osc_quota_info",
794                                             sizeof(struct osc_quota_info),
795                                             0, 0);
796         if (!qinfo_cachep)
797                 RETURN(-ENOMEM);
798
799         for (i = 0; i < NR_DQHASH; i++)
800                 CFS_INIT_LIST_HEAD(qinfo_hash + i);
801
802         RETURN(0);
803 }
804
805 int osc_quota_exit(void)
806 {
807         struct osc_quota_info *oqi, *n;
808         int i, rc;
809         ENTRY;
810
811         spin_lock(&qinfo_list_lock);
812         for (i = 0; i < NR_DQHASH; i++) {
813                 list_for_each_entry_safe(oqi, n, &qinfo_hash[i], oqi_hash) {
814                         remove_qinfo_hash(oqi);
815                         free_qinfo(oqi);
816                 }
817         }
818         spin_unlock(&qinfo_list_lock);
819
820         rc = cfs_mem_cache_destroy(qinfo_cachep);
821         LASSERTF(rc == 0, "couldn't destory qinfo_cachep slab\n");
822         qinfo_cachep = NULL;
823
824         RETURN(0);
825 }
826
827 #ifdef __KERNEL__
828 #ifdef HAVE_QUOTA_SUPPORT
829 quota_interface_t mds_quota_interface = {
830         .quota_init     = mds_quota_init,
831         .quota_exit     = mds_quota_exit,
832         .quota_setup    = mds_quota_setup,
833         .quota_cleanup  = mds_quota_cleanup,
834         .quota_check    = target_quota_check,
835         .quota_ctl      = mds_quota_ctl,
836         .quota_setinfo  = mds_quota_setinfo,
837         .quota_fs_cleanup = mds_quota_fs_cleanup,
838         .quota_recovery = mds_quota_recovery,
839         .quota_adjust   = mds_quota_adjust,
840         .quota_chkquota = quota_chk_acq_common,
841         .quota_acquire  = quota_acquire_common,
842         .quota_pending_commit = quota_pending_commit,
843 };
844
845 quota_interface_t filter_quota_interface = {
846         .quota_setup    = filter_quota_setup,
847         .quota_cleanup  = filter_quota_cleanup,
848         .quota_check    = target_quota_check,
849         .quota_ctl      = filter_quota_ctl,
850         .quota_setinfo  = filter_quota_setinfo,
851         .quota_clearinfo = filter_quota_clearinfo,
852         .quota_enforce  = filter_quota_enforce,
853         .quota_getflag  = filter_quota_getflag,
854         .quota_acquire  = quota_acquire_common,
855         .quota_adjust   = filter_quota_adjust,
856         .quota_chkquota = quota_chk_acq_common,
857         .quota_adjust_qunit   = filter_quota_adjust_qunit,
858         .quota_pending_commit = quota_pending_commit,
859 };
860 #endif
861 #endif /* __KERNEL__ */
862
863 quota_interface_t mdc_quota_interface = {
864         .quota_ctl      = client_quota_ctl,
865         .quota_check    = client_quota_check,
866         .quota_poll_check = client_quota_poll_check,
867 };
868
869 quota_interface_t lmv_quota_interface = {
870         .quota_ctl      = lmv_quota_ctl,
871         .quota_check    = lmv_quota_check,
872 };
873
874 quota_interface_t osc_quota_interface = {
875         .quota_ctl      = client_quota_ctl,
876         .quota_check    = client_quota_check,
877         .quota_poll_check = client_quota_poll_check,
878         .quota_init     = osc_quota_init,
879         .quota_exit     = osc_quota_exit,
880         .quota_chkdq    = osc_quota_chkdq,
881         .quota_setdq    = osc_quota_setdq,
882         .quota_cleanup  = osc_quota_cleanup,
883         .quota_adjust_qunit = client_quota_adjust_qunit,
884 };
885
886 quota_interface_t lov_quota_interface = {
887         .quota_ctl      = lov_quota_ctl,
888         .quota_check    = lov_quota_check,
889         .quota_adjust_qunit = lov_quota_adjust_qunit,
890 };
891
892 #ifdef __KERNEL__
893
894 cfs_proc_dir_entry_t *lquota_type_proc_dir = NULL;
895
896 static int __init init_lustre_quota(void)
897 {
898 #ifdef HAVE_QUOTA_SUPPORT
899         int rc = 0;
900
901         lquota_type_proc_dir = lprocfs_register(OBD_LQUOTA_DEVICENAME,
902                                                 proc_lustre_root,
903                                                 NULL, NULL);
904         if (IS_ERR(lquota_type_proc_dir)) {
905                 CERROR("LProcFS failed in lquota-init\n");
906                 rc = PTR_ERR(lquota_type_proc_dir);
907                 return rc;
908         }
909
910         rc = qunit_cache_init();
911         if (rc)
912                 return rc;
913
914         PORTAL_SYMBOL_REGISTER(filter_quota_interface);
915         PORTAL_SYMBOL_REGISTER(mds_quota_interface);
916 #endif
917         PORTAL_SYMBOL_REGISTER(mdc_quota_interface);
918         PORTAL_SYMBOL_REGISTER(lmv_quota_interface);
919         PORTAL_SYMBOL_REGISTER(osc_quota_interface);
920         PORTAL_SYMBOL_REGISTER(lov_quota_interface);
921         return 0;
922 }
923
924 static void /*__exit*/ exit_lustre_quota(void)
925 {
926         PORTAL_SYMBOL_UNREGISTER(mdc_quota_interface);
927         PORTAL_SYMBOL_UNREGISTER(lmv_quota_interface);
928         PORTAL_SYMBOL_UNREGISTER(osc_quota_interface);
929         PORTAL_SYMBOL_UNREGISTER(lov_quota_interface);
930 #ifdef HAVE_QUOTA_SUPPORT
931         PORTAL_SYMBOL_UNREGISTER(filter_quota_interface);
932         PORTAL_SYMBOL_UNREGISTER(mds_quota_interface);
933
934         qunit_cache_cleanup();
935
936         if (lquota_type_proc_dir)
937                 lprocfs_remove(&lquota_type_proc_dir);
938 #endif
939 }
940
941 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
942 MODULE_DESCRIPTION("Lustre Quota");
943 MODULE_LICENSE("GPL");
944
945 cfs_module(lquota, "1.0.0", init_lustre_quota, exit_lustre_quota);
946
947 #ifdef HAVE_QUOTA_SUPPORT
948 EXPORT_SYMBOL(mds_quota_interface);
949 EXPORT_SYMBOL(filter_quota_interface);
950 #endif
951 EXPORT_SYMBOL(mdc_quota_interface);
952 EXPORT_SYMBOL(lmv_quota_interface);
953 EXPORT_SYMBOL(osc_quota_interface);
954 EXPORT_SYMBOL(lov_quota_interface);
955 #endif /* __KERNEL */