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