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