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