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_LQUOTA
  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 <class_hash.h>
  61 #include <lprocfs_status.h>
  62 #include "quota_internal.h"
  63
  64 #ifdef HAVE_QUOTA_SUPPORT
  65
  66 static lustre_hash_ops_t lqs_hash_ops;
  67
  68 unsigned long default_bunit_sz = 128 * 1024 * 1024; /* 128M bytes */
  69 unsigned long default_btune_ratio = 50;             /* 50 percentage */
  70 unsigned long default_iunit_sz = 5120;              /* 5120 inodes */
  71 unsigned long default_itune_ratio = 50;             /* 50 percentage */
  72
  73 cfs_mem_cache_t *qunit_cachep = NULL;
  74 struct list_head qunit_hash[NR_DQHASH];
  75 spinlock_t qunit_hash_lock = SPIN_LOCK_UNLOCKED;
  76
  77 /* please sync qunit_state with qunit_state_names */
  78 enum qunit_state {
  79         QUNIT_CREATED      = 0,   /* a qunit is created */
  80         QUNIT_IN_HASH      = 1,   /* a qunit is added into qunit hash, that means
  81                                    * a quota req will be sent or is flying */
  82         QUNIT_RM_FROM_HASH = 2,   /* a qunit is removed from qunit hash, that
  83                                    * means a quota req is handled and comes
  84                                    * back */
  85         QUNIT_FINISHED     = 3,   /* qunit can wake up all threads waiting
  86                                    * for it */
  87 };
  88
  89 static const char *qunit_state_names[] = {
  90         [QUNIT_CREATED]      = "CREATED",
  91         [QUNIT_IN_HASH]      = "IN_HASH",
  92         [QUNIT_RM_FROM_HASH] = "RM_FROM_HASH",
  93         [QUNIT_FINISHED]     = "FINISHED",
  94 };
  95
  96 struct lustre_qunit {
  97         struct list_head lq_hash;          /* Hash list in memory */
  98         atomic_t lq_refcnt;                /* Use count */
  99         struct lustre_quota_ctxt *lq_ctxt; /* Quota context this applies to */
 100         struct qunit_data lq_data;         /* See qunit_data */
 101         unsigned int lq_opc;               /* QUOTA_DQACQ, QUOTA_DQREL */
 102         cfs_waitq_t lq_waitq;              /* Threads waiting for this qunit */
 103         spinlock_t lq_lock;                /* Protect the whole structure */
 104         enum qunit_state lq_state;         /* Present the status of qunit */
 105         int lq_rc;                         /* The rc of lq_data */
 106 };
 107
 108 #define QUNIT_SET_STATE(qunit, state)                                   \
 109 do {                                                                    \
 110         spin_lock(&qunit->lq_lock);                                     \
 111         QDATA_DEBUG((&qunit->lq_data), "qunit(%p) lq_state(%s->%s), "   \
 112                     "lq_rc(%d)\n",                                      \
 113                     qunit, qunit_state_names[qunit->lq_state],          \
 114                     qunit_state_names[state], qunit->lq_rc);            \
 115         qunit->lq_state = state;                                        \
 116         spin_unlock(&qunit->lq_lock);                                   \
 117 } while(0)
 118
 119 #define QUNIT_SET_STATE_AND_RC(qunit, state, rc)                        \
 120 do {                                                                    \
 121         spin_lock(&qunit->lq_lock);                                     \
 122         qunit->lq_rc = rc;                                              \
 123         QDATA_DEBUG((&qunit->lq_data), "qunit(%p) lq_state(%s->%s), "   \
 124                     "lq_rc(%d)\n",                                      \
 125                     qunit, qunit_state_names[qunit->lq_state],          \
 126                     qunit_state_names[state], qunit->lq_rc);            \
 127         qunit->lq_state = state;                                        \
 128         spin_unlock(&qunit->lq_lock);                                   \
 129 } while(0)
 130
 131
 132 int should_translate_quota (struct obd_import *imp)
 133 {
 134         ENTRY;
 135
 136         LASSERT(imp);
 137         if (imp->imp_connect_data.ocd_connect_flags & OBD_CONNECT_QUOTA64)
 138                 RETURN(0);
 139         else
 140                 RETURN(1);
 141 }
 142
 143 void qunit_cache_cleanup(void)
 144 {
 145         int i;
 146         ENTRY;
 147
 148         spin_lock(&qunit_hash_lock);
 149         for (i = 0; i < NR_DQHASH; i++)
 150                 LASSERT(list_empty(qunit_hash + i));
 151         spin_unlock(&qunit_hash_lock);
 152
 153         if (qunit_cachep) {
 154                 int rc;
 155                 rc = cfs_mem_cache_destroy(qunit_cachep);
 156                 LASSERTF(rc == 0, "couldn't destory qunit_cache slab\n");
 157                 qunit_cachep = NULL;
 158         }
 159         EXIT;
 160 }
 161
 162 int qunit_cache_init(void)
 163 {
 164         int i;
 165         ENTRY;
 166
 167         LASSERT(qunit_cachep == NULL);
 168         qunit_cachep = cfs_mem_cache_create("ll_qunit_cache",
 169                                             sizeof(struct lustre_qunit),
 170                                             0, 0);
 171         if (!qunit_cachep)
 172                 RETURN(-ENOMEM);
 173
 174         spin_lock(&qunit_hash_lock);
 175         for (i = 0; i < NR_DQHASH; i++)
 176                 INIT_LIST_HEAD(qunit_hash + i);
 177         spin_unlock(&qunit_hash_lock);
 178         RETURN(0);
 179 }
 180
 181 static inline int
 182 qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
 183              __attribute__((__const__));
 184
 185 static inline int
 186 qunit_hashfn(struct lustre_quota_ctxt *qctxt, struct qunit_data *qdata)
 187 {
 188         unsigned int id = qdata->qd_id;
 189         unsigned int type = QDATA_IS_GRP(qdata);
 190
 191         unsigned long tmp = ((unsigned long)qctxt >> L1_CACHE_SHIFT) ^ id;
 192         tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
 193         return tmp;
 194 }
 195
 196 /* caller must hold qunit_hash_lock */
 197 static inline struct lustre_qunit *find_qunit(unsigned int hashent,
 198                                               struct lustre_quota_ctxt *qctxt,
 199                                               struct qunit_data *qdata)
 200 {
 201         struct lustre_qunit *qunit = NULL;
 202         struct qunit_data *tmp;
 203
 204         LASSERT_SPIN_LOCKED(&qunit_hash_lock);
 205         list_for_each_entry(qunit, qunit_hash + hashent, lq_hash) {
 206                 tmp = &qunit->lq_data;
 207                 if (qunit->lq_ctxt == qctxt &&
 208                     qdata->qd_id == tmp->qd_id &&
 209                     (qdata->qd_flags & LQUOTA_QUNIT_FLAGS) ==
 210                     (tmp->qd_flags & LQUOTA_QUNIT_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, limit_org, pending_write = 0;
 232         long long record = 0;
 233         struct obd_quotactl *qctl;
 234         struct lustre_qunit_size *lqs = NULL;
 235         int ret = 0;
 236         ENTRY;
 237
 238         if (!sb_any_quota_enabled(sb))
 239                 RETURN(0);
 240
 241         spin_lock(&qctxt->lqc_lock);
 242         if (!qctxt->lqc_valid){
 243                 spin_unlock(&qctxt->lqc_lock);
 244                 RETURN(0);
 245         }
 246         spin_unlock(&qctxt->lqc_lock);
 247
 248         OBD_ALLOC_PTR(qctl);
 249         if (qctl == NULL)
 250                 RETURN(-ENOMEM);
 251
 252         /* get fs quota usage & limit */
 253         qctl->qc_cmd = Q_GETQUOTA;
 254         qctl->qc_id = qdata->qd_id;
 255         qctl->qc_type = QDATA_IS_GRP(qdata);
 256         ret = fsfilt_quotactl(obd, sb, qctl);
 257         if (ret) {
 258                 if (ret == -ESRCH)      /* no limit */
 259                         ret = 0;
 260                 else
 261                         CERROR("can't get fs quota usage! (rc:%d)\n", ret);
 262                 GOTO(out, ret);
 263         }
 264
 265         if (QDATA_IS_BLK(qdata)) {
 266                 usage = qctl->qc_dqblk.dqb_curspace;
 267                 limit = qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS;
 268         } else {
 269                 usage = qctl->qc_dqblk.dqb_curinodes;
 270                 limit = qctl->qc_dqblk.dqb_ihardlimit;
 271         }
 272
 273         /* ignore the no quota limit case; and it can avoid creating
 274          * unnecessary lqs for uid/gid */
 275         if (!limit)
 276                 GOTO(out, ret = 0);
 277
 278  search_lqs:
 279         quota_search_lqs(qdata, NULL, qctxt, &lqs);
 280         if (!lqs) {
 281                 CDEBUG(D_QUOTA, "Can't find the lustre qunit size!\n");
 282                 ret = quota_create_lqs(qdata, NULL, qctxt, &lqs);
 283                 if (ret == -EALREADY) {
 284                         ret = 0;
 285                         goto search_lqs;
 286                 }
 287                 if (ret < 0)
 288                         GOTO (out, ret);
 289         }
 290         spin_lock(&lqs->lqs_lock);
 291
 292         if (QDATA_IS_BLK(qdata)) {
 293                 qunit_sz = lqs->lqs_bunit_sz;
 294                 tune_sz  = lqs->lqs_btune_sz;
 295                 pending_write = lqs->lqs_bwrite_pending;
 296                 record   = lqs->lqs_blk_rec;
 297                 LASSERT(!(qunit_sz % QUOTABLOCK_SIZE));
 298         } else {
 299                 /* we didn't need change inode qunit size now */
 300                 qunit_sz = lqs->lqs_iunit_sz;
 301                 tune_sz  = lqs->lqs_itune_sz;
 302                 pending_write = lqs->lqs_iwrite_pending;
 303                 record   = lqs->lqs_ino_rec;
 304         }
 305
 306         /* we don't count the MIN_QLIMIT */
 307         if ((limit == MIN_QLIMIT && !QDATA_IS_BLK(qdata)) ||
 308             (toqb(limit) == MIN_QLIMIT && QDATA_IS_BLK(qdata)))
 309                 limit = 0;
 310
 311         usage += pending_write;
 312         limit_org = limit;
 313         /* when a releasing quota req is sent, before it returned
 314            limit is assigned a small value. limit will overflow */
 315         if (limit + record < 0)
 316                 usage -= record;
 317         else
 318                 limit += record;
 319
 320         LASSERT(qdata->qd_count == 0);
 321         if (limit <= usage + tune_sz) {
 322                 while (qdata->qd_count + limit <=
 323                        usage + tune_sz)
 324                         qdata->qd_count += qunit_sz;
 325                 ret = 1;
 326         } else if (limit > usage + qunit_sz + tune_sz &&
 327                    limit_org > qdata->qd_count + qunit_sz) {
 328                 while (limit - qdata->qd_count > usage + qunit_sz + tune_sz &&
 329                        limit_org > qdata->qd_count + qunit_sz)
 330                         qdata->qd_count += qunit_sz;
 331                 ret = 2;
 332                 /* if there are other pending writes for this uid/gid, releasing
 333                  * quota is put off until the last pending write b=16645 */
 334                 if (ret == 2 && pending_write) {
 335                         CDEBUG(D_QUOTA, "delay quota release\n");
 336                         ret = 0;
 337                 }
 338         }
 339         CDEBUG(D_QUOTA, "type: %c, limit: "LPU64", usage: "LPU64
 340                ", pending_write: "LPU64", record: "LPD64
 341                ", qunit_sz: %lu, tune_sz: %lu, ret: %d.\n",
 342                QDATA_IS_BLK(qdata) ? 'b' : 'i', limit, usage, pending_write,
 343                record, qunit_sz, tune_sz, ret);
 344         LASSERT(ret == 0 || qdata->qd_count);
 345
 346         spin_unlock(&lqs->lqs_lock);
 347         lqs_putref(lqs);
 348         EXIT;
 349  out:
 350         OBD_FREE_PTR(qctl);
 351         return ret;
 352 }
 353
 354 /* compute the remaining quota for certain gid or uid b=11693 */
 355 int compute_remquota(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
 356                      struct qunit_data *qdata, int isblk)
 357 {
 358         struct super_block *sb = qctxt->lqc_sb;
 359         __u64 usage, limit;
 360         struct obd_quotactl *qctl;
 361         int ret = QUOTA_RET_OK;
 362         ENTRY;
 363
 364         if (!sb_any_quota_enabled(sb))
 365                 RETURN(QUOTA_RET_NOQUOTA);
 366
 367         /* ignore root user */
 368         if (qdata->qd_id == 0 && QDATA_IS_GRP(qdata) == USRQUOTA)
 369                 RETURN(QUOTA_RET_NOLIMIT);
 370
 371         OBD_ALLOC_PTR(qctl);
 372         if (qctl == NULL)
 373                 RETURN(-ENOMEM);
 374
 375         /* get fs quota usage & limit */
 376         qctl->qc_cmd = Q_GETQUOTA;
 377         qctl->qc_id = qdata->qd_id;
 378         qctl->qc_type = QDATA_IS_GRP(qdata);
 379         ret = fsfilt_quotactl(obd, sb, qctl);
 380         if (ret) {
 381                 if (ret == -ESRCH)      /* no limit */
 382                         ret = QUOTA_RET_NOLIMIT;
 383                 else
 384                         CDEBUG(D_QUOTA, "can't get fs quota usage! (rc:%d)",
 385                                ret);
 386                 GOTO(out, ret);
 387         }
 388
 389         usage = isblk ? qctl->qc_dqblk.dqb_curspace :
 390                 qctl->qc_dqblk.dqb_curinodes;
 391         limit = isblk ? qctl->qc_dqblk.dqb_bhardlimit << QUOTABLOCK_BITS :
 392                 qctl->qc_dqblk.dqb_ihardlimit;
 393         if (!limit){            /* no limit */
 394                 ret = QUOTA_RET_NOLIMIT;
 395                 GOTO(out, ret);
 396         }
 397
 398         if (limit >= usage)
 399                 qdata->qd_count = limit - usage;
 400         else
 401                 qdata->qd_count = 0;
 402         EXIT;
 403 out:
 404         OBD_FREE_PTR(qctl);
 405         return ret;
 406 }
 407
 408 static struct lustre_qunit *alloc_qunit(struct lustre_quota_ctxt *qctxt,
 409                                         struct qunit_data *qdata, int opc)
 410 {
 411         struct lustre_qunit *qunit = NULL;
 412         ENTRY;
 413
 414         OBD_SLAB_ALLOC(qunit, qunit_cachep, CFS_ALLOC_IO, sizeof(*qunit));
 415         if (qunit == NULL)
 416                 RETURN(NULL);
 417
 418         INIT_LIST_HEAD(&qunit->lq_hash);
 419         init_waitqueue_head(&qunit->lq_waitq);
 420         atomic_set(&qunit->lq_refcnt, 1);
 421         qunit->lq_ctxt = qctxt;
 422         memcpy(&qunit->lq_data, qdata, sizeof(*qdata));
 423         qunit->lq_opc = opc;
 424         qunit->lq_lock = SPIN_LOCK_UNLOCKED;
 425         QUNIT_SET_STATE_AND_RC(qunit, QUNIT_CREATED, 0);
 426         RETURN(qunit);
 427 }
 428
 429 static inline void free_qunit(struct lustre_qunit *qunit)
 430 {
 431         OBD_SLAB_FREE(qunit, qunit_cachep, sizeof(*qunit));
 432 }
 433
 434 static inline void qunit_get(struct lustre_qunit *qunit)
 435 {
 436         atomic_inc(&qunit->lq_refcnt);
 437 }
 438
 439 static void qunit_put(struct lustre_qunit *qunit)
 440 {
 441         LASSERT(atomic_read(&qunit->lq_refcnt));
 442         if (atomic_dec_and_test(&qunit->lq_refcnt))
 443                 free_qunit(qunit);
 444 }
 445
 446 /* caller must hold qunit_hash_lock and release ref of qunit after using it */
 447 static struct lustre_qunit *dqacq_in_flight(struct lustre_quota_ctxt *qctxt,
 448                                             struct qunit_data *qdata)
 449 {
 450         unsigned int hashent = qunit_hashfn(qctxt, qdata);
 451         struct lustre_qunit *qunit;
 452         ENTRY;
 453
 454         LASSERT_SPIN_LOCKED(&qunit_hash_lock);
 455         qunit = find_qunit(hashent, qctxt, qdata);
 456         if (qunit)
 457                 qunit_get(qunit);
 458         RETURN(qunit);
 459 }
 460
 461 static void
 462 insert_qunit_nolock(struct lustre_quota_ctxt *qctxt, struct lustre_qunit *qunit)
 463 {
 464         struct list_head *head;
 465
 466         LASSERT(list_empty(&qunit->lq_hash));
 467         qunit_get(qunit);
 468         head = qunit_hash + qunit_hashfn(qctxt, &qunit->lq_data);
 469         list_add(&qunit->lq_hash, head);
 470         QUNIT_SET_STATE(qunit, QUNIT_IN_HASH);
 471 }
 472
 473 static void compute_lqs_after_removing_qunit(struct lustre_qunit *qunit)
 474 {
 475         struct lustre_qunit_size *lqs = NULL;
 476
 477         quota_search_lqs(&qunit->lq_data, NULL, qunit->lq_ctxt, &lqs);
 478         if (lqs) {
 479                 spin_lock(&lqs->lqs_lock);
 480                 if (qunit->lq_opc == QUOTA_DQACQ)
 481                         quota_compute_lqs(&qunit->lq_data, lqs, 0, 1);
 482                 if (qunit->lq_opc == QUOTA_DQREL)
 483                         quota_compute_lqs(&qunit->lq_data, lqs, 0, 0);
 484                 spin_unlock(&lqs->lqs_lock);
 485                 /* this is for quota_search_lqs */
 486                 lqs_putref(lqs);
 487                 /* this is for schedule_dqacq */
 488                 lqs_putref(lqs);
 489         }
 490
 491 }
 492
 493 static void remove_qunit_nolock(struct lustre_qunit *qunit)
 494 {
 495         LASSERT(!list_empty(&qunit->lq_hash));
 496         LASSERT_SPIN_LOCKED(&qunit_hash_lock);
 497
 498         list_del_init(&qunit->lq_hash);
 499         QUNIT_SET_STATE(qunit, QUNIT_RM_FROM_HASH);
 500         qunit_put(qunit);
 501 }
 502
 503 #define INC_QLIMIT(limit, count) (limit == MIN_QLIMIT) ? \
 504                                  (limit = count) : (limit += count)
 505
 506
 507 /* FIXME check if this mds is the master of specified id */
 508 static int
 509 is_master(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
 510           unsigned int id, int type)
 511 {
 512         return qctxt->lqc_handler ? 1 : 0;
 513 }
 514
 515 static int
 516 schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
 517                struct qunit_data *qdata, int opc, int wait,
 518                struct obd_trans_info *oti);
 519
 520 static int
 521 dqacq_completion(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
 522                  struct qunit_data *qdata, int rc, int opc)
 523 {
 524         struct lustre_qunit *qunit = NULL;
 525         struct super_block *sb = qctxt->lqc_sb;
 526         int err = 0;
 527         struct quota_adjust_qunit *oqaq = NULL;
 528         int rc1 = 0;
 529         ENTRY;
 530
 531         LASSERT(qdata);
 532         QDATA_DEBUG(qdata, "obd(%s): complete %s quota req\n",
 533                     obd->obd_name, (opc == QUOTA_DQACQ) ? "acq" : "rel");
 534
 535         /* update local operational quota file */
 536         if (rc == 0) {
 537                 __u64 count = QUSG(qdata->qd_count, QDATA_IS_BLK(qdata));
 538                 struct obd_quotactl *qctl;
 539                 __u64 *hardlimit;
 540
 541                 OBD_ALLOC_PTR(qctl);
 542                 if (qctl == NULL)
 543                         GOTO(out, err = -ENOMEM);
 544
 545                 /* acq/rel qunit for specified uid/gid is serialized,
 546                  * so there is no race between get fs quota limit and
 547                  * set fs quota limit */
 548                 qctl->qc_cmd = Q_GETQUOTA;
 549                 qctl->qc_id = qdata->qd_id;
 550                 qctl->qc_type = QDATA_IS_GRP(qdata);
 551                 err = fsfilt_quotactl(obd, sb, qctl);
 552                 if (err) {
 553                         CERROR("error get quota fs limit! (rc:%d)\n", err);
 554                         GOTO(out_mem, err);
 555                 }
 556
 557                 if (QDATA_IS_BLK(qdata)) {
 558                         qctl->qc_dqblk.dqb_valid = QIF_BLIMITS;
 559                         hardlimit = &qctl->qc_dqblk.dqb_bhardlimit;
 560                 } else {
 561                         qctl->qc_dqblk.dqb_valid = QIF_ILIMITS;
 562                         hardlimit = &qctl->qc_dqblk.dqb_ihardlimit;
 563                 }
 564
 565                 CDEBUG(D_QUOTA, "hardlimt: "LPU64"\n", *hardlimit);
 566
 567                 if (*hardlimit == 0)
 568                         goto out_mem;
 569
 570                 switch (opc) {
 571                 case QUOTA_DQACQ:
 572                         INC_QLIMIT(*hardlimit, count);
 573                         break;
 574                 case QUOTA_DQREL:
 575                         LASSERTF(count < *hardlimit,
 576                                  "id(%u) flag(%u) type(%c) isblk(%c) "
 577                                  "count("LPU64") qd_qunit("LPU64") "
 578                                  "hardlimit("LPU64").\n",
 579                                  qdata->qd_id, qdata->qd_flags,
 580                                  QDATA_IS_GRP(qdata) ? 'g' : 'u',
 581                                  QDATA_IS_BLK(qdata) ? 'b': 'i',
 582                                  qdata->qd_count, qdata->qd_qunit, *hardlimit);
 583                         *hardlimit -= count;
 584                         break;
 585                 default:
 586                         LBUG();
 587                 }
 588
 589                 /* clear quota limit */
 590                 if (count == 0)
 591                         *hardlimit = 0;
 592
 593                 qctl->qc_cmd = Q_SETQUOTA;
 594                 err = fsfilt_quotactl(obd, sb, qctl);
 595                 if (err)
 596                         CERROR("error set quota fs limit! (rc:%d)\n", err);
 597
 598                 QDATA_DEBUG(qdata, "%s completion\n",
 599                             opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
 600 out_mem:
 601                 OBD_FREE_PTR(qctl);
 602         } else if (rc == -EDQUOT) {
 603                 QDATA_DEBUG(qdata, "acquire qunit got EDQUOT.\n");
 604         } else if (rc == -EBUSY) {
 605                 QDATA_DEBUG(qdata, "it's is recovering, got EBUSY.\n");
 606         } else {
 607                 CERROR("acquire qunit got error! (rc:%d)\n", rc);
 608         }
 609 out:
 610         /* remove the qunit from hash */
 611         spin_lock(&qunit_hash_lock);
 612
 613         qunit = dqacq_in_flight(qctxt, qdata);
 614         /* this qunit has been removed by qctxt_cleanup() */
 615         if (!qunit) {
 616                 spin_unlock(&qunit_hash_lock);
 617                 QDATA_DEBUG(qdata, "%s is discarded because qunit isn't found\n",
 618                             opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
 619                 RETURN(err);
 620         }
 621
 622         LASSERT(opc == qunit->lq_opc);
 623         /* remove this qunit from lq_hash so that new processes cannot be added
 624          * to qunit->lq_waiters */
 625         remove_qunit_nolock(qunit);
 626         spin_unlock(&qunit_hash_lock);
 627
 628         compute_lqs_after_removing_qunit(qunit);
 629
 630         /* wake up all waiters */
 631         QUNIT_SET_STATE_AND_RC(qunit, QUNIT_FINISHED, rc);
 632         wake_up(&qunit->lq_waitq);
 633
 634         /* this is for dqacq_in_flight() */
 635         qunit_put(qunit);
 636         /* this is for alloc_qunit() */
 637         qunit_put(qunit);
 638         if (rc < 0 && rc != -EDQUOT)
 639                  RETURN(err);
 640
 641         /* don't reschedule in such cases:
 642          *   - acq/rel failure and qunit isn't changed,
 643          *     but not for quota recovery.
 644          *   - local dqacq/dqrel.
 645          *   - local disk io failure.
 646          */
 647          OBD_ALLOC_PTR(oqaq);
 648          if (!oqaq)
 649                  RETURN(-ENOMEM);
 650          qdata_to_oqaq(qdata, oqaq);
 651          /* adjust the qunit size in slaves */
 652          rc1 = quota_adjust_slave_lqs(oqaq, qctxt);
 653          OBD_FREE_PTR(oqaq);
 654          if (rc1 < 0) {
 655                  CERROR("adjust slave's qunit size failed!(rc:%d)\n", rc1);
 656                  RETURN(rc1);
 657          }
 658          if (err || (rc && rc != -EBUSY && rc1 == 0) ||
 659              is_master(obd, qctxt, qdata->qd_id, QDATA_IS_GRP(qdata)))
 660                 RETURN(err);
 661
 662         /* reschedule another dqacq/dqrel if needed */
 663         qdata->qd_count = 0;
 664         qdata->qd_flags &= LQUOTA_QUNIT_FLAGS;
 665         rc1 = check_cur_qunit(obd, qctxt, qdata);
 666         if (rc1 > 0) {
 667                 int opc;
 668                 opc = rc1 == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
 669                 rc1 = schedule_dqacq(obd, qctxt, qdata, opc, 0, NULL);
 670                 QDATA_DEBUG(qdata, "reschedudle opc(%d) rc(%d)\n", opc, rc1);
 671         }
 672         RETURN(err);
 673 }
 674
 675 struct dqacq_async_args {
 676         struct lustre_quota_ctxt *aa_ctxt;
 677         struct lustre_qunit *aa_qunit;
 678 };
 679
 680 static int dqacq_interpret(struct ptlrpc_request *req, void *data, int rc)
 681 {
 682         struct dqacq_async_args *aa = (struct dqacq_async_args *)data;
 683         struct lustre_quota_ctxt *qctxt = aa->aa_ctxt;
 684         struct lustre_qunit *qunit = aa->aa_qunit;
 685         struct obd_device *obd = req->rq_import->imp_obd;
 686         struct qunit_data *qdata = NULL;
 687         int rc1 = 0;
 688         ENTRY;
 689
 690         LASSERT(req);
 691         LASSERT(req->rq_import);
 692
 693         /* there are several forms of qunit(historic causes), so we need to
 694          * adjust qunit from slaves to the same form here */
 695         OBD_ALLOC(qdata, sizeof(struct qunit_data));
 696         if (!qdata)
 697                 RETURN(-ENOMEM);
 698
 699         /* if a quota req timeouts or is dropped, we should update quota
 700          * statistics which will be handled in dqacq_completion. And in
 701          * this situation we should get qdata from request instead of
 702          * reply */
 703         rc1 = quota_get_qdata(req, qdata,
 704                               (rc != 0) ? QUOTA_REQUEST : QUOTA_REPLY,
 705                               QUOTA_IMPORT);
 706         if (rc1 < 0) {
 707                 DEBUG_REQ(D_ERROR, req,
 708                           "error unpacking qunit_data(rc: %d)\n", rc1);
 709                 GOTO(exit, rc = rc1);
 710         }
 711
 712         QDATA_DEBUG(qdata, "qdata: interpret rc(%d).\n", rc);
 713         QDATA_DEBUG((&qunit->lq_data), "lq_data: \n");
 714
 715         if (qdata->qd_id != qunit->lq_data.qd_id ||
 716             OBD_FAIL_CHECK_ONCE(OBD_FAIL_QUOTA_RET_QDATA)) {
 717                 CDEBUG(D_ERROR, "the returned qd_id isn't expected!"
 718                        "(qdata: %u, lq_data: %u)\n", qdata->qd_id,
 719                        qunit->lq_data.qd_id);
 720                 qdata->qd_id = qunit->lq_data.qd_id;
 721                 rc = -EPROTO;
 722         }
 723         if (QDATA_IS_GRP(qdata) != QDATA_IS_GRP(&qunit->lq_data)) {
 724                 CDEBUG(D_ERROR, "the returned grp/usr isn't expected!"
 725                        "(qdata: %u, lq_data: %u)\n", qdata->qd_flags,
 726                        qunit->lq_data.qd_flags);
 727                 if (QDATA_IS_GRP(&qunit->lq_data))
 728                         QDATA_SET_GRP(qdata);
 729                 else
 730                         QDATA_CLR_GRP(qdata);
 731                 rc = -EPROTO;
 732         }
 733         if (qdata->qd_count > qunit->lq_data.qd_count) {
 734                 CDEBUG(D_ERROR, "the returned qd_count isn't expected!"
 735                        "(qdata: "LPU64", lq_data: "LPU64")\n", qdata->qd_count,
 736                        qunit->lq_data.qd_count);
 737                 rc = -EPROTO;
 738         }
 739
 740         rc = dqacq_completion(obd, qctxt, qdata, rc,
 741                               lustre_msg_get_opc(req->rq_reqmsg));
 742
 743 exit:
 744         OBD_FREE(qdata, sizeof(struct qunit_data));
 745
 746         RETURN(rc);
 747 }
 748
 749 /* check if quota master is online */
 750 int check_qm(struct lustre_quota_ctxt *qctxt)
 751 {
 752         int rc;
 753         ENTRY;
 754
 755         spin_lock(&qctxt->lqc_lock);
 756         /* quit waiting when mds is back or qctxt is cleaned up */
 757         rc = qctxt->lqc_import || !qctxt->lqc_valid;
 758         spin_unlock(&qctxt->lqc_lock);
 759
 760         RETURN(rc);
 761 }
 762
 763 static int got_qunit(struct lustre_qunit *qunit)
 764 {
 765         int rc;
 766         ENTRY;
 767
 768         spin_lock(&qunit->lq_lock);
 769         switch (qunit->lq_state) {
 770         case QUNIT_IN_HASH:
 771         case QUNIT_RM_FROM_HASH:
 772                 rc = 0;
 773                 break;
 774         case QUNIT_FINISHED:
 775                 rc = 1;
 776                 break;
 777         default:
 778                 rc = 0;
 779                 CERROR("invalid qunit state %d\n", qunit->lq_state);
 780         }
 781         spin_unlock(&qunit->lq_lock);
 782         RETURN(rc);
 783 }
 784
 785 static int
 786 schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
 787                struct qunit_data *qdata, int opc, int wait,
 788                struct obd_trans_info *oti)
 789 {
 790         struct lustre_qunit *qunit, *empty;
 791         struct l_wait_info lwi = { 0 };
 792         struct ptlrpc_request *req;
 793         struct dqacq_async_args *aa;
 794         int size[2] = { sizeof(struct ptlrpc_body), 0 };
 795         struct obd_import *imp = NULL;
 796         struct lustre_qunit_size *lqs = NULL;
 797         struct timeval work_start;
 798         struct timeval work_end;
 799         long timediff;
 800         int rc = 0;
 801         ENTRY;
 802
 803         LASSERT(opc == QUOTA_DQACQ || opc == QUOTA_DQREL);
 804         do_gettimeofday(&work_start);
 805         if ((empty = alloc_qunit(qctxt, qdata, opc)) == NULL)
 806                 RETURN(-ENOMEM);
 807
 808         spin_lock(&qunit_hash_lock);
 809         qunit = dqacq_in_flight(qctxt, qdata);
 810         if (qunit) {
 811                 spin_unlock(&qunit_hash_lock);
 812                 qunit_put(empty);
 813
 814                 goto wait_completion;
 815         }
 816         qunit = empty;
 817         qunit_get(qunit);
 818         insert_qunit_nolock(qctxt, qunit);
 819         spin_unlock(&qunit_hash_lock);
 820
 821         quota_search_lqs(qdata, NULL, qctxt, &lqs);
 822         if (lqs) {
 823                 spin_lock(&lqs->lqs_lock);
 824                 quota_compute_lqs(qdata, lqs, 1, (opc == QUOTA_DQACQ) ? 1 : 0);
 825                 /* when this qdata returned from mds, it will call lqs_putref */
 826                 lqs_getref(lqs);
 827                 spin_unlock(&lqs->lqs_lock);
 828                 /* this is for quota_search_lqs */
 829                 lqs_putref(lqs);
 830         } else {
 831                 CDEBUG(D_ERROR, "Can't find the lustre qunit size!\n");
 832         }
 833
 834         QDATA_DEBUG(qdata, "obd(%s): send %s quota req\n",
 835                     obd->obd_name, (opc == QUOTA_DQACQ) ? "acq" : "rel");
 836         /* master is going to dqacq/dqrel from itself */
 837         if (is_master(obd, qctxt, qdata->qd_id, QDATA_IS_GRP(qdata))) {
 838                 int rc2;
 839                 QDATA_DEBUG(qdata, "local %s.\n",
 840                             opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
 841                 QDATA_SET_CHANGE_QS(qdata);
 842                 rc = qctxt->lqc_handler(obd, qdata, opc);
 843                 rc2 = dqacq_completion(obd, qctxt, qdata, rc, opc);
 844                 /* this is for qunit_get() */
 845                 qunit_put(qunit);
 846
 847                 do_gettimeofday(&work_end);
 848                 timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
 849                 if (opc == QUOTA_DQACQ)
 850                         lprocfs_counter_add(qctxt->lqc_stats,
 851                                             wait ? LQUOTA_SYNC_ACQ : LQUOTA_ASYNC_ACQ,
 852                                             timediff);
 853                 else
 854                         lprocfs_counter_add(qctxt->lqc_stats,
 855                                             wait ? LQUOTA_SYNC_REL : LQUOTA_ASYNC_REL,
 856                                             timediff);
 857                 RETURN(rc ? rc : rc2);
 858         }
 859
 860         spin_lock(&qctxt->lqc_lock);
 861         if (!qctxt->lqc_import) {
 862                 spin_unlock(&qctxt->lqc_lock);
 863                 QDATA_DEBUG(qdata, "lqc_import is invalid.\n");
 864
 865                 spin_lock(&qunit_hash_lock);
 866                 remove_qunit_nolock(qunit);
 867                 spin_unlock(&qunit_hash_lock);
 868
 869                 compute_lqs_after_removing_qunit(qunit);
 870
 871                 QUNIT_SET_STATE_AND_RC(qunit, QUNIT_FINISHED, -EAGAIN);
 872                 wake_up(&qunit->lq_waitq);
 873
 874                 /* this is for qunit_get() */
 875                 qunit_put(qunit);
 876                 /* this for alloc_qunit() */
 877                 qunit_put(qunit);
 878                 spin_lock(&qctxt->lqc_lock);
 879                 if (wait && !qctxt->lqc_import) {
 880                         spin_unlock(&qctxt->lqc_lock);
 881
 882                         LASSERT(oti && oti->oti_thread &&
 883                                 oti->oti_thread->t_watchdog);
 884
 885                         lc_watchdog_disable(oti->oti_thread->t_watchdog);
 886                         CDEBUG(D_QUOTA, "sleep for quota master\n");
 887                         l_wait_event(qctxt->lqc_wait_for_qmaster,
 888                                      check_qm(qctxt), &lwi);
 889                         CDEBUG(D_QUOTA, "wake up when quota master is back\n");
 890                         lc_watchdog_touch(oti->oti_thread->t_watchdog);
 891                 } else {
 892                         spin_unlock(&qctxt->lqc_lock);
 893                 }
 894
 895                 RETURN(-EAGAIN);
 896         }
 897         imp = class_import_get(qctxt->lqc_import);
 898         spin_unlock(&qctxt->lqc_lock);
 899
 900         /* build dqacq/dqrel request */
 901         LASSERT(imp);
 902         size[1] = quota_get_qunit_data_size(imp->
 903                                             imp_connect_data.ocd_connect_flags);
 904
 905         req = ptlrpc_prep_req(imp, LUSTRE_MDS_VERSION, opc, 2,
 906                               size, NULL);
 907         if (!req) {
 908                 dqacq_completion(obd, qctxt, qdata, -ENOMEM, opc);
 909                 class_import_put(imp);
 910                 /* this is for qunit_get() */
 911                 qunit_put(qunit);
 912                 RETURN(-ENOMEM);
 913         }
 914
 915         rc = quota_copy_qdata(req, qdata, QUOTA_REQUEST, QUOTA_IMPORT);
 916         if (rc < 0) {
 917                 CDEBUG(D_ERROR, "Can't pack qunit_data(rc: %d)\n", rc);
 918                 dqacq_completion(obd, qctxt, qdata, rc, opc);
 919                 class_import_put(imp);
 920                 /* this is for qunit_get() */
 921                 qunit_put(qunit);
 922                 RETURN(rc);
 923         }
 924         ptlrpc_req_set_repsize(req, 2, size);
 925         req->rq_no_resend = req->rq_no_delay = 1;
 926         class_import_put(imp);
 927
 928         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
 929         aa = ptlrpc_req_async_args(req);
 930         aa->aa_ctxt = qctxt;
 931         aa->aa_qunit = qunit;
 932
 933         req->rq_interpret_reply = dqacq_interpret;
 934         ptlrpcd_add_req(req);
 935
 936         QDATA_DEBUG(qdata, "%s scheduled.\n",
 937                     opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
 938 wait_completion:
 939         if (wait && qunit) {
 940                 struct qunit_data *p = &qunit->lq_data;
 941
 942                 QDATA_DEBUG(p, "qunit(%p) is waiting for dqacq.\n", qunit);
 943                 l_wait_event(qunit->lq_waitq, got_qunit(qunit), &lwi);
 944                 /* rc = -EAGAIN, it means a quota req is finished;
 945                  * rc = -EDQUOT, it means out of quota
 946                  * rc = -EBUSY, it means recovery is happening
 947                  * other rc < 0, it means real errors, functions who call
 948                  * schedule_dqacq should take care of this */
 949                 spin_lock(&qunit->lq_lock);
 950                 if (qunit->lq_rc == 0)
 951                         rc = -EAGAIN;
 952                 else
 953                         rc = qunit->lq_rc;
 954                 spin_unlock(&qunit->lq_lock);
 955                 CDEBUG(D_QUOTA, "qunit(%p) finishes waiting. (rc:%d)\n",
 956                        qunit, rc);
 957         }
 958
 959         qunit_put(qunit);
 960         do_gettimeofday(&work_end);
 961         timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
 962         if (opc == QUOTA_DQACQ)
 963                 lprocfs_counter_add(qctxt->lqc_stats,
 964                                     wait ? LQUOTA_SYNC_ACQ : LQUOTA_ASYNC_ACQ,
 965                                     timediff);
 966         else
 967                 lprocfs_counter_add(qctxt->lqc_stats,
 968                                     wait ? LQUOTA_SYNC_REL : LQUOTA_ASYNC_REL,
 969                                     timediff);
 970
 971         RETURN(rc);
 972 }
 973
 974 int
 975 qctxt_adjust_qunit(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
 976                    uid_t uid, gid_t gid, __u32 isblk, int wait,
 977                    struct obd_trans_info *oti)
 978 {
 979         int rc = 0, i = USRQUOTA;
 980         __u32 id[MAXQUOTAS] = { uid, gid };
 981         struct qunit_data qdata[MAXQUOTAS];
 982         ENTRY;
 983
 984         CLASSERT(MAXQUOTAS < 4);
 985         if (!sb_any_quota_enabled(qctxt->lqc_sb))
 986                 RETURN(0);
 987
 988         for (i = 0; i < MAXQUOTAS; i++) {
 989                 qdata[i].qd_id = id[i];
 990                 qdata[i].qd_flags = i;
 991                 if (isblk)
 992                         QDATA_SET_BLK(&qdata[i]);
 993                 qdata[i].qd_count = 0;
 994
 995                 rc = check_cur_qunit(obd, qctxt, &qdata[i]);
 996                 if (rc > 0) {
 997                         int opc;
 998                         /* need acquire or release */
 999                         opc = rc == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
1000                         rc = schedule_dqacq(obd, qctxt, &qdata[i], opc,
1001                                             wait,oti);
1002                         if (rc < 0)
1003                                 RETURN(rc);
1004                 } else if (wait == 1) {
1005                         /* when wait equates 1, that means mds_quota_acquire
1006                          * or filter_quota_acquire is calling it. */
1007                         rc = qctxt_wait_pending_dqacq(qctxt, id[i], i, isblk);
1008                         if (rc < 0)
1009                                 RETURN(rc);
1010                 }
1011         }
1012
1013         RETURN(rc);
1014 }
1015
1016 int
1017 qctxt_wait_pending_dqacq(struct lustre_quota_ctxt *qctxt, unsigned int id,
1018                          unsigned short type, int isblk)
1019 {
1020         struct lustre_qunit *qunit = NULL;
1021         struct qunit_data qdata;
1022         struct timeval work_start;
1023         struct timeval work_end;
1024         long timediff;
1025         struct l_wait_info lwi = { 0 };
1026         int rc = 0;
1027         ENTRY;
1028
1029         do_gettimeofday(&work_start);
1030         qdata.qd_id = id;
1031         qdata.qd_flags = type;
1032         if (isblk)
1033                 QDATA_SET_BLK(&qdata);
1034         qdata.qd_count = 0;
1035
1036         spin_lock(&qunit_hash_lock);
1037         qunit = dqacq_in_flight(qctxt, &qdata);
1038         spin_unlock(&qunit_hash_lock);
1039
1040         if (qunit) {
1041                 struct qunit_data *p = &qunit->lq_data;
1042
1043                 QDATA_DEBUG(p, "qunit(%p) is waiting for dqacq.\n", qunit);
1044                 l_wait_event(qunit->lq_waitq, got_qunit(qunit), &lwi);
1045                 CDEBUG(D_QUOTA, "qunit(%p) finishes waiting. (rc:%d)\n",
1046                        qunit, qunit->lq_rc);
1047                 /* keep same as schedule_dqacq() b=17030 */
1048                 spin_lock(&qunit->lq_lock);
1049                 if (qunit->lq_rc == 0)
1050                         rc = -EAGAIN;
1051                 else
1052                         rc = qunit->lq_rc;
1053                 spin_unlock(&qunit->lq_lock);
1054                 /* this is for dqacq_in_flight() */
1055                 qunit_put(qunit);
1056                 do_gettimeofday(&work_end);
1057                 timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
1058                 lprocfs_counter_add(qctxt->lqc_stats,
1059                                     isblk ? LQUOTA_WAIT_PENDING_BLK_QUOTA :
1060                                             LQUOTA_WAIT_PENDING_INO_QUOTA,
1061                                     timediff);
1062         } else {
1063                 do_gettimeofday(&work_end);
1064                 timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
1065                 lprocfs_counter_add(qctxt->lqc_stats,
1066                                     isblk ? LQUOTA_NOWAIT_PENDING_BLK_QUOTA :
1067                                             LQUOTA_NOWAIT_PENDING_INO_QUOTA,
1068                                     timediff);
1069         }
1070
1071         RETURN(rc);
1072 }
1073
1074 int
1075 qctxt_init(struct obd_device *obd, dqacq_handler_t handler)
1076 {
1077         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
1078         struct super_block *sb = obd->u.obt.obt_sb;
1079         int rc = 0;
1080         ENTRY;
1081
1082         LASSERT(qctxt);
1083
1084         rc = ptlrpcd_addref();
1085         if (rc)
1086                 RETURN(rc);
1087
1088         cfs_waitq_init(&qctxt->lqc_wait_for_qmaster);
1089         spin_lock_init(&qctxt->lqc_lock);
1090         spin_lock(&qctxt->lqc_lock);
1091         qctxt->lqc_handler = handler;
1092         qctxt->lqc_sb = sb;
1093         qctxt->lqc_import = NULL;
1094         qctxt->lqc_recovery = 0;
1095         qctxt->lqc_switch_qs = 1; /* Change qunit size in default setting */
1096         qctxt->lqc_valid = 1;
1097         qctxt->lqc_cqs_boundary_factor = 4;
1098         qctxt->lqc_cqs_least_bunit = PTLRPC_MAX_BRW_SIZE;
1099         qctxt->lqc_cqs_least_iunit = 2;
1100         qctxt->lqc_cqs_qs_factor = 2;
1101         qctxt->lqc_flags = 0;
1102         QUOTA_MASTER_UNREADY(qctxt);
1103         qctxt->lqc_bunit_sz = default_bunit_sz;
1104         qctxt->lqc_btune_sz = default_bunit_sz / 100 * default_btune_ratio;
1105         qctxt->lqc_iunit_sz = default_iunit_sz;
1106         qctxt->lqc_itune_sz = default_iunit_sz * default_itune_ratio / 100;
1107         qctxt->lqc_switch_seconds = 300; /* enlarging will wait 5 minutes
1108                                           * after the last shrinking */
1109         qctxt->lqc_sync_blk = 0;
1110         spin_unlock(&qctxt->lqc_lock);
1111
1112         qctxt->lqc_lqs_hash = lustre_hash_init("LQS_HASH", 7, 7,
1113                                                &lqs_hash_ops, 0);
1114         if (!qctxt->lqc_lqs_hash)
1115                 CERROR("initialize hash lqs for %s error!\n", obd->obd_name);
1116
1117 #ifdef LPROCFS
1118         if (lquota_proc_setup(obd, is_master(obd, qctxt, 0, 0)))
1119                 CERROR("initialize proc for %s error!\n", obd->obd_name);
1120 #endif
1121
1122         RETURN(rc);
1123 }
1124
1125 void qctxt_cleanup(struct lustre_quota_ctxt *qctxt, int force)
1126 {
1127         struct lustre_qunit *qunit, *tmp;
1128         struct list_head tmp_list;
1129         int i;
1130         ENTRY;
1131
1132         INIT_LIST_HEAD(&tmp_list);
1133
1134         spin_lock(&qctxt->lqc_lock);
1135         qctxt->lqc_valid = 0;
1136         spin_unlock(&qctxt->lqc_lock);
1137
1138         spin_lock(&qunit_hash_lock);
1139         for (i = 0; i < NR_DQHASH; i++) {
1140                 list_for_each_entry_safe(qunit, tmp, &qunit_hash[i], lq_hash) {
1141                         if (qunit->lq_ctxt != qctxt)
1142                                 continue;
1143                         remove_qunit_nolock(qunit);
1144                         list_add(&qunit->lq_hash, &tmp_list);
1145                 }
1146         }
1147         spin_unlock(&qunit_hash_lock);
1148
1149         list_for_each_entry_safe(qunit, tmp, &tmp_list, lq_hash) {
1150                 list_del_init(&qunit->lq_hash);
1151                 compute_lqs_after_removing_qunit(qunit);
1152
1153                 /* wake up all waiters */
1154                 QUNIT_SET_STATE_AND_RC(qunit, QUNIT_FINISHED, 0);
1155                 wake_up(&qunit->lq_waitq);
1156                 qunit_put(qunit);
1157         }
1158
1159         lustre_hash_exit(qctxt->lqc_lqs_hash);
1160
1161         /* after qctxt_cleanup, qctxt might be freed, then check_qm() is
1162          * unpredicted. So we must wait until lqc_wait_for_qmaster is empty */
1163         while (cfs_waitq_active(&qctxt->lqc_wait_for_qmaster)) {
1164                 cfs_waitq_signal(&qctxt->lqc_wait_for_qmaster);
1165                 cfs_schedule_timeout(CFS_TASK_INTERRUPTIBLE,
1166                                      cfs_time_seconds(1));
1167         }
1168
1169         ptlrpcd_decref();
1170
1171 #ifdef LPROCFS
1172         if (lquota_proc_cleanup(qctxt))
1173                 CERROR("cleanup proc error!\n");
1174 #endif
1175
1176         EXIT;
1177 }
1178
1179 struct qslave_recov_thread_data {
1180         struct obd_device *obd;
1181         struct lustre_quota_ctxt *qctxt;
1182         struct completion comp;
1183 };
1184
1185 /* FIXME only recovery block quota by now */
1186 static int qslave_recovery_main(void *arg)
1187 {
1188         struct qslave_recov_thread_data *data = arg;
1189         struct obd_device *obd = data->obd;
1190         struct lustre_quota_ctxt *qctxt = data->qctxt;
1191         unsigned int type;
1192         int rc = 0;
1193         ENTRY;
1194
1195         ptlrpc_daemonize("qslave_recovd");
1196
1197         complete(&data->comp);
1198
1199         if (qctxt->lqc_recovery)
1200                 RETURN(0);
1201         qctxt->lqc_recovery = 1;
1202
1203         for (type = USRQUOTA; type < MAXQUOTAS; type++) {
1204                 struct qunit_data qdata;
1205                 struct quota_info *dqopt = sb_dqopt(qctxt->lqc_sb);
1206                 struct list_head id_list;
1207                 struct dquot_id *dqid, *tmp;
1208                 int ret;
1209
1210                 LOCK_DQONOFF_MUTEX(dqopt);
1211                 if (!sb_has_quota_enabled(qctxt->lqc_sb, type)) {
1212                         UNLOCK_DQONOFF_MUTEX(dqopt);
1213                         break;
1214                 }
1215
1216                 LASSERT(dqopt->files[type] != NULL);
1217                 INIT_LIST_HEAD(&id_list);
1218 #ifndef KERNEL_SUPPORTS_QUOTA_READ
1219                 rc = fsfilt_qids(obd, dqopt->files[type], NULL, type, &id_list);
1220 #else
1221                 rc = fsfilt_qids(obd, NULL, dqopt->files[type], type, &id_list);
1222 #endif
1223                 UNLOCK_DQONOFF_MUTEX(dqopt);
1224                 if (rc)
1225                         CERROR("Get ids from quota file failed. (rc:%d)\n", rc);
1226
1227                 list_for_each_entry_safe(dqid, tmp, &id_list, di_link) {
1228                         list_del_init(&dqid->di_link);
1229                         /* skip slave recovery on itself */
1230                         if (is_master(obd, qctxt, dqid->di_id, type))
1231                                 goto free;
1232                         if (rc && rc != -EBUSY)
1233                                 goto free;
1234
1235                         qdata.qd_id = dqid->di_id;
1236                         qdata.qd_flags = type;
1237                         QDATA_SET_BLK(&qdata);
1238                         qdata.qd_count = 0;
1239
1240                         ret = check_cur_qunit(obd, qctxt, &qdata);
1241                         if (ret > 0) {
1242                                 int opc;
1243                                 opc = ret == 1 ? QUOTA_DQACQ : QUOTA_DQREL;
1244                                 rc = schedule_dqacq(obd, qctxt, &qdata, opc,
1245                                                     0, NULL);
1246                                 if (rc == -EDQUOT)
1247                                         rc = 0;
1248                         } else {
1249                                 rc = 0;
1250                         }
1251
1252                         if (rc)
1253                                 CDEBUG(rc == -EBUSY ? D_QUOTA : D_ERROR,
1254                                        "qslave recovery failed! (id:%d type:%d "
1255                                        " rc:%d)\n", dqid->di_id, type, rc);
1256 free:
1257                         kfree(dqid);
1258                 }
1259         }
1260
1261         qctxt->lqc_recovery = 0;
1262         RETURN(rc);
1263 }
1264
1265 void
1266 qslave_start_recovery(struct obd_device *obd, struct lustre_quota_ctxt *qctxt)
1267 {
1268         struct qslave_recov_thread_data data;
1269         int rc;
1270         ENTRY;
1271
1272         if (!sb_any_quota_enabled(qctxt->lqc_sb))
1273                 goto exit;
1274
1275         data.obd = obd;
1276         data.qctxt = qctxt;
1277         init_completion(&data.comp);
1278
1279         rc = kernel_thread(qslave_recovery_main, &data, CLONE_VM|CLONE_FILES);
1280         if (rc < 0) {
1281                 CERROR("Cannot start quota recovery thread: rc %d\n", rc);
1282                 goto exit;
1283         }
1284         wait_for_completion(&data.comp);
1285 exit:
1286         EXIT;
1287 }
1288
1289
1290 /*
1291  * lqs<->qctxt hash operations
1292  */
1293
1294 /* string hashing using djb2 hash algorithm */
1295 static unsigned
1296 lqs_hash(lustre_hash_t *lh, void *key, unsigned mask)
1297 {
1298         struct quota_adjust_qunit *lqs_key;
1299         unsigned hash;
1300         ENTRY;
1301
1302         LASSERT(key);
1303         lqs_key = (struct quota_adjust_qunit *)key;
1304         hash = (QAQ_IS_GRP(lqs_key) ? 5381 : 5387) * lqs_key->qaq_id;
1305
1306         RETURN(hash & mask);
1307 }
1308
1309 static int
1310 lqs_compare(void *key, struct hlist_node *hnode)
1311 {
1312         struct quota_adjust_qunit *lqs_key;
1313         struct lustre_qunit_size *q;
1314         int rc;
1315         ENTRY;
1316
1317         LASSERT(key);
1318         lqs_key = (struct quota_adjust_qunit *)key;
1319         q = hlist_entry(hnode, struct lustre_qunit_size, lqs_hash);
1320
1321         spin_lock(&q->lqs_lock);
1322         rc = ((lqs_key->qaq_id == q->lqs_id) &&
1323               (QAQ_IS_GRP(lqs_key) == LQS_IS_GRP(q)));
1324         spin_unlock(&q->lqs_lock);
1325
1326         RETURN(rc);
1327 }
1328
1329 static void *
1330 lqs_get(struct hlist_node *hnode)
1331 {
1332         struct lustre_qunit_size *q =
1333             hlist_entry(hnode, struct lustre_qunit_size, lqs_hash);
1334         ENTRY;
1335
1336         atomic_inc(&q->lqs_refcount);
1337         CDEBUG(D_QUOTA, "lqs=%p refcount %d\n",
1338                q, atomic_read(&q->lqs_refcount));
1339
1340         RETURN(q);
1341 }
1342
1343 static void *
1344 lqs_put(struct hlist_node *hnode)
1345 {
1346         struct lustre_qunit_size *q =
1347             hlist_entry(hnode, struct lustre_qunit_size, lqs_hash);
1348         ENTRY;
1349
1350         LASSERT(atomic_read(&q->lqs_refcount) > 0);
1351         atomic_dec(&q->lqs_refcount);
1352         CDEBUG(D_QUOTA, "lqs=%p refcount %d\n",
1353                q, atomic_read(&q->lqs_refcount));
1354
1355         RETURN(q);
1356 }
1357
1358 static void
1359 lqs_exit(struct hlist_node *hnode)
1360 {
1361         struct lustre_qunit_size *q;
1362         ENTRY;
1363
1364         q = hlist_entry(hnode, struct lustre_qunit_size, lqs_hash);
1365         /*
1366          * Nothing should be left. User of lqs put it and
1367          * lqs also was deleted from table by this time
1368          * so we should have 0 refs.
1369          */
1370         LASSERTF(atomic_read(&q->lqs_refcount) == 0,
1371                  "Busy lqs %p with %d refs\n", q,
1372                  atomic_read(&q->lqs_refcount));
1373         OBD_FREE_PTR(q);
1374         EXIT;
1375 }
1376
1377 static lustre_hash_ops_t lqs_hash_ops = {
1378         .lh_hash    = lqs_hash,
1379         .lh_compare = lqs_compare,
1380         .lh_get     = lqs_get,
1381         .lh_put     = lqs_put,
1382         .lh_exit    = lqs_exit
1383 };
1384 #endif /* HAVE_QUOTA_SUPPORT */