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