lustre/quota/quota_master.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 (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/quota/quota_master.c
  39  *
  40  * Lustre Quota Master request handler
  41  *
  42  * Author: Niu YaWei <niu@clusterfs.com>
  43  */
  44
  45 #ifndef EXPORT_SYMTAB
  46 # define EXPORT_SYMTAB
  47 #endif
  48
  49 #define DEBUG_SUBSYSTEM S_LQUOTA
  50
  51 #include <linux/version.h>
  52 #include <linux/fs.h>
  53 #include <asm/unistd.h>
  54 #include <linux/slab.h>
  55 #include <linux/quotaops.h>
  56 #include <linux/module.h>
  57 #include <linux/init.h>
  58 #include <linux/quota.h>
  59
  60 #include <obd_class.h>
  61 #include <lustre_quota.h>
  62 #include <lustre_fsfilt.h>
  63 #include <lustre_mds.h>
  64
  65 #include "quota_internal.h"
  66
  67 #ifdef HAVE_QUOTA_SUPPORT
  68
  69 /* lock ordering: mds->mds_qonoff_sem > dquot->dq_sem > lqs->lqs_lock */
  70 static cfs_list_t lustre_dquot_hash[NR_DQHASH];
  71 static cfs_rwlock_t dquot_hash_lock = CFS_RW_LOCK_UNLOCKED;
  72
  73 cfs_mem_cache_t *lustre_dquot_cachep;
  74
  75 int lustre_dquot_init(void)
  76 {
  77         int i;
  78         ENTRY;
  79
  80         LASSERT(lustre_dquot_cachep == NULL);
  81         lustre_dquot_cachep = cfs_mem_cache_create("lustre_dquot_cache",
  82                                                    sizeof(struct lustre_dquot),
  83                                                    0, 0);
  84         if (!lustre_dquot_cachep)
  85                 return (-ENOMEM);
  86
  87         for (i = 0; i < NR_DQHASH; i++) {
  88                 CFS_INIT_LIST_HEAD(lustre_dquot_hash + i);
  89         }
  90         RETURN(0);
  91 }
  92
  93 void lustre_dquot_exit(void)
  94 {
  95         int i;
  96         ENTRY;
  97         /* FIXME cleanup work ?? */
  98
  99         for (i = 0; i < NR_DQHASH; i++) {
 100                 LASSERT(cfs_list_empty(lustre_dquot_hash + i));
 101         }
 102         if (lustre_dquot_cachep) {
 103                 int rc;
 104                 rc = cfs_mem_cache_destroy(lustre_dquot_cachep);
 105                 LASSERTF(rc == 0,"couldn't destroy lustre_dquot_cachep slab\n");
 106                 lustre_dquot_cachep = NULL;
 107         }
 108         EXIT;
 109 }
 110
 111 static inline int
 112 dquot_hashfn(struct lustre_quota_info *info, unsigned int id, int type)
 113              __attribute__((__const__));
 114
 115 static inline int
 116 dquot_hashfn(struct lustre_quota_info *info, unsigned int id, int type)
 117 {
 118         unsigned long tmp = ((unsigned long)info >> L1_CACHE_SHIFT) ^ id;
 119         tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
 120         return tmp;
 121 }
 122
 123 /* caller must hold dquot_hash_lock */
 124 static struct lustre_dquot *find_dquot(int hashent,
 125                                        struct lustre_quota_info *lqi, qid_t id,
 126                                        int type)
 127 {
 128         struct lustre_dquot *dquot;
 129         ENTRY;
 130
 131         cfs_list_for_each_entry(dquot, &lustre_dquot_hash[hashent], dq_hash) {
 132                 if (dquot->dq_info == lqi &&
 133                     dquot->dq_id == id && dquot->dq_type == type)
 134                         RETURN(dquot);
 135         }
 136         RETURN(NULL);
 137 }
 138
 139 static struct lustre_dquot *alloc_dquot(struct lustre_quota_info *lqi,
 140                                         qid_t id, int type)
 141 {
 142         struct lustre_dquot *dquot = NULL;
 143         ENTRY;
 144
 145         OBD_SLAB_ALLOC_PTR_GFP(dquot, lustre_dquot_cachep, CFS_ALLOC_IO);
 146         if (dquot == NULL)
 147                 RETURN(NULL);
 148
 149         CFS_INIT_LIST_HEAD(&dquot->dq_hash);
 150         cfs_init_mutex_locked(&dquot->dq_sem);
 151         cfs_atomic_set(&dquot->dq_refcnt, 1);
 152         dquot->dq_info = lqi;
 153         dquot->dq_id = id;
 154         dquot->dq_type = type;
 155
 156         RETURN(dquot);
 157 }
 158
 159 static void free_dquot(struct lustre_dquot *dquot)
 160 {
 161         OBD_SLAB_FREE(dquot, lustre_dquot_cachep, sizeof(*dquot));
 162 }
 163
 164 static void insert_dquot_nolock(struct lustre_dquot *dquot)
 165 {
 166         cfs_list_t *head = lustre_dquot_hash +
 167             dquot_hashfn(dquot->dq_info, dquot->dq_id, dquot->dq_type);
 168         LASSERT(cfs_list_empty(&dquot->dq_hash));
 169         cfs_list_add(&dquot->dq_hash, head);
 170 }
 171
 172 static void remove_dquot_nolock(struct lustre_dquot *dquot)
 173 {
 174         LASSERT(!cfs_list_empty(&dquot->dq_hash));
 175         cfs_list_del_init(&dquot->dq_hash);
 176 }
 177
 178 static void lustre_dqput(struct lustre_dquot *dquot)
 179 {
 180         ENTRY;
 181         cfs_write_lock(&dquot_hash_lock);
 182         LASSERT(cfs_atomic_read(&dquot->dq_refcnt));
 183         cfs_atomic_dec(&dquot->dq_refcnt);
 184         if (cfs_atomic_read(&dquot->dq_refcnt) == 0) {
 185                 remove_dquot_nolock(dquot);
 186                 free_dquot(dquot);
 187         }
 188         cfs_write_unlock(&dquot_hash_lock);
 189         EXIT;
 190 }
 191
 192 static struct lustre_dquot *lustre_dqget(struct obd_device *obd,
 193                                          struct lustre_quota_info *lqi,
 194                                          qid_t id, int type, int can_fake)
 195 {
 196         unsigned int hashent = dquot_hashfn(lqi, id, type);
 197         struct lustre_dquot *dquot, *empty;
 198         int free_dq = 0;
 199         ENTRY;
 200
 201         if ((empty = alloc_dquot(lqi, id, type)) == NULL)
 202                 RETURN(ERR_PTR(-ENOMEM));
 203
 204         cfs_read_lock(&dquot_hash_lock);
 205         if ((dquot = find_dquot(hashent, lqi, id, type)) != NULL) {
 206                 cfs_atomic_inc(&dquot->dq_refcnt);
 207                 cfs_read_unlock(&dquot_hash_lock);
 208                 free_dq = 1;
 209         } else {
 210                 int rc;
 211
 212                 cfs_read_unlock(&dquot_hash_lock);
 213
 214                 dquot = empty;
 215                 rc = fsfilt_dquot(obd, dquot, QFILE_RD_DQUOT);
 216                 cfs_up(&dquot->dq_sem);
 217                 if (rc) {
 218                         CERROR("can't read dquot from admin quotafile! "
 219                                "(rc:%d)\n", rc);
 220                         free_dquot(dquot);
 221                         RETURN(ERR_PTR(rc));
 222                 } else {
 223                         cfs_write_lock(&dquot_hash_lock);
 224                         if ((dquot = find_dquot(hashent, lqi, id, type)) != NULL) {
 225                                 cfs_atomic_inc(&dquot->dq_refcnt);
 226                                 free_dq = 1;
 227                         } else {
 228                                 dquot = empty;
 229                                 insert_dquot_nolock(dquot);
 230                         }
 231                         cfs_write_unlock(&dquot_hash_lock);
 232                 }
 233
 234         }
 235
 236         LASSERT(dquot);
 237         if (!can_fake && cfs_test_bit(DQ_FAKE_B, &dquot->dq_flags)) {
 238                 DQUOT_DEBUG(dquot, "It is a fake dquot: unexpected!\n");
 239                 lustre_dqput(dquot);
 240                 dquot = ERR_PTR(-ENOENT);
 241         }
 242
 243         if (free_dq)
 244                 free_dquot(empty);
 245
 246
 247         RETURN(dquot);
 248 }
 249
 250 static void init_oqaq(struct quota_adjust_qunit *oqaq,
 251                       struct lustre_quota_ctxt *qctxt,
 252                       qid_t id, int type)
 253 {
 254         struct lustre_qunit_size *lqs = NULL;
 255
 256         oqaq->qaq_id = id;
 257         oqaq->qaq_flags = type;
 258         lqs = quota_search_lqs(LQS_KEY(type, id), qctxt, 0);
 259         if (lqs && !IS_ERR(lqs)) {
 260                 cfs_spin_lock(&lqs->lqs_lock);
 261                 oqaq->qaq_bunit_sz = lqs->lqs_bunit_sz;
 262                 oqaq->qaq_iunit_sz = lqs->lqs_iunit_sz;
 263                 oqaq->qaq_flags    = lqs->lqs_flags;
 264                 cfs_spin_unlock(&lqs->lqs_lock);
 265                 lqs_putref(lqs);
 266         } else {
 267                 CDEBUG(D_QUOTA, "Can't find the lustre qunit size!\n");
 268                 oqaq->qaq_bunit_sz = qctxt->lqc_bunit_sz;
 269                 oqaq->qaq_iunit_sz = qctxt->lqc_iunit_sz;
 270         }
 271 }
 272
 273 int dqacq_adjust_qunit_sz(struct obd_device *obd, qid_t id, int type,
 274                           __u32 is_blk)
 275 {
 276         struct mds_obd *mds = &obd->u.mds;
 277         struct lustre_quota_ctxt *qctxt = &mds->mds_obt.obt_qctxt;
 278         __u32 ost_num = mds->mds_lov_objid_count, mdt_num = 1;
 279         struct quota_adjust_qunit *oqaq = NULL;
 280         unsigned int qid[MAXQUOTAS] = { 0, 0 };
 281         struct lustre_quota_info *info = &mds->mds_quota_info;
 282         struct lustre_dquot *dquot = NULL;
 283         int adjust_res = 0;
 284         int rc = 0;
 285         ENTRY;
 286
 287         LASSERT(mds);
 288         cfs_down_read(&mds->mds_qonoff_sem);
 289         dquot = lustre_dqget(obd, info, id, type, 0);
 290         if (IS_ERR(dquot))
 291                 RETURN(PTR_ERR(dquot));
 292
 293         cfs_up_read(&mds->mds_qonoff_sem);
 294         OBD_ALLOC_PTR(oqaq);
 295         if (!oqaq)
 296                 GOTO(out, rc = -ENOMEM);
 297
 298         cfs_down(&dquot->dq_sem);
 299         init_oqaq(oqaq, qctxt, id, type);
 300
 301         rc = dquot_create_oqaq(qctxt, dquot, ost_num, mdt_num,
 302                                is_blk ? LQUOTA_FLAGS_ADJBLK :
 303                                LQUOTA_FLAGS_ADJINO, oqaq);
 304
 305         if (rc < 0) {
 306                 CERROR("create oqaq failed! (rc:%d)\n", rc);
 307                 GOTO(out_sem, rc);
 308         }
 309         QAQ_DEBUG(oqaq, "show oqaq.\n")
 310
 311         if (!QAQ_IS_ADJBLK(oqaq) && !QAQ_IS_ADJINO(oqaq))
 312                 GOTO(out_sem, rc);
 313
 314         /* adjust the mds slave qunit size */
 315         adjust_res = quota_adjust_slave_lqs(oqaq, qctxt);
 316         if (adjust_res <= 0) {
 317                 if (adjust_res < 0) {
 318                         rc = adjust_res;
 319                         CERROR("adjust mds slave's qunit size failed! "
 320                                "(rc:%d)\n", rc);
 321                 } else {
 322                         CDEBUG(D_QUOTA, "qunit doesn't need to be adjusted.\n");
 323                 }
 324                 GOTO(out_sem, rc);
 325         }
 326
 327         if (type)
 328                 qid[GRPQUOTA] = dquot->dq_id;
 329         else
 330                 qid[USRQUOTA] = dquot->dq_id;
 331
 332         cfs_up(&dquot->dq_sem);
 333
 334         rc = qctxt_adjust_qunit(obd, qctxt, qid, is_blk, 0, NULL);
 335         if (rc == -EDQUOT || rc == -EBUSY) {
 336                 CDEBUG(D_QUOTA, "rc: %d.\n", rc);
 337                 rc = 0;
 338         }
 339         if (rc) {
 340                 CERROR("%s: mds fail to adjust file quota! (rc:%d)\n",
 341                        obd->obd_name, rc);
 342                 GOTO(out, rc);
 343         }
 344
 345         /* only when block qunit is reduced, boardcast to osts */
 346         if ((adjust_res & LQS_BLK_DECREASE) && QAQ_IS_ADJBLK(oqaq))
 347                 rc = obd_quota_adjust_qunit(mds->mds_lov_exp, oqaq, qctxt, NULL);
 348
 349 out:
 350         lustre_dqput(dquot);
 351         if (oqaq)
 352                 OBD_FREE_PTR(oqaq);
 353
 354         RETURN(rc);
 355 out_sem:
 356         cfs_up(&dquot->dq_sem);
 357         goto out;
 358 }
 359
 360 int dqacq_handler(struct obd_device *obd, struct qunit_data *qdata, int opc)
 361 {
 362         struct mds_obd *mds = &obd->u.mds;
 363         struct lustre_quota_ctxt *qctxt = &mds->mds_obt.obt_qctxt;
 364         struct lustre_quota_info *info = &mds->mds_quota_info;
 365         struct lustre_dquot *dquot = NULL;
 366         __u64 *usage = NULL;
 367         __u64 hlimit = 0, slimit = 0;
 368         time_t *time = NULL;
 369         unsigned int grace = 0;
 370         struct lustre_qunit_size *lqs = NULL;
 371         int rc = 0;
 372         ENTRY;
 373
 374         if (OBD_FAIL_CHECK(OBD_FAIL_OBD_DQACQ))
 375                 RETURN(-EIO);
 376
 377         if (!ll_sb_has_quota_active(qctxt->lqc_sb,
 378                                     QDATA_IS_GRP(qdata) ? GRPQUOTA : USRQUOTA))
 379                 RETURN(-EIO);
 380
 381         lqs = quota_search_lqs(LQS_KEY(QDATA_IS_GRP(qdata), qdata->qd_id),
 382                                qctxt, 0);
 383         if (lqs == NULL)
 384                 rc = -ENOENT;
 385         if (IS_ERR(lqs))
 386                 rc = PTR_ERR(lqs);
 387         if (rc)
 388                 RETURN(rc);
 389
 390         cfs_spin_lock(&lqs->lqs_lock);
 391         if (LQS_IS_RECOVERY(lqs)) {
 392                 cfs_spin_unlock(&lqs->lqs_lock);
 393                 LQS_DEBUG(lqs, "this lqs is under recovery\n");
 394                 GOTO(skip, rc = -EBUSY);
 395         }
 396         cfs_spin_unlock(&lqs->lqs_lock);
 397
 398         cfs_down_write(&mds->mds_qonoff_sem);
 399         dquot = lustre_dqget(obd, info, qdata->qd_id, QDATA_IS_GRP(qdata), 0);
 400         if (IS_ERR(dquot)) {
 401                 cfs_up_write(&mds->mds_qonoff_sem);
 402                 GOTO(skip, rc = PTR_ERR(dquot));
 403         }
 404
 405         DQUOT_DEBUG(dquot, "get dquot in dqacq_handler\n");
 406         QINFO_DEBUG(dquot->dq_info, "get dquot in dqadq_handler\n");
 407
 408         cfs_down(&dquot->dq_sem);
 409
 410         if (QDATA_IS_BLK(qdata)) {
 411                 grace = info->qi_info[QDATA_IS_GRP(qdata)].dqi_bgrace;
 412                 usage = &dquot->dq_dqb.dqb_curspace;
 413                 hlimit = dquot->dq_dqb.dqb_bhardlimit;
 414                 slimit = dquot->dq_dqb.dqb_bsoftlimit;
 415                 time = &dquot->dq_dqb.dqb_btime;
 416         } else {
 417                 grace = info->qi_info[QDATA_IS_GRP(qdata)].dqi_igrace;
 418                 usage = (__u64 *) & dquot->dq_dqb.dqb_curinodes;
 419                 hlimit = dquot->dq_dqb.dqb_ihardlimit;
 420                 slimit = dquot->dq_dqb.dqb_isoftlimit;
 421                 time = &dquot->dq_dqb.dqb_itime;
 422         }
 423
 424         /* if the quota limit in admin quotafile is zero, we just inform
 425          * slave to clear quota limit with zero qd_count */
 426         if (hlimit == 0 && slimit == 0) {
 427                 qdata->qd_count = 0;
 428                 GOTO(out, rc);
 429         }
 430
 431         switch (opc) {
 432         case QUOTA_DQACQ:
 433                 if (hlimit &&
 434                     QUSG(*usage + qdata->qd_count, QDATA_IS_BLK(qdata)) > hlimit)
 435                 {
 436                         if (QDATA_IS_CHANGE_QS(qdata) &&
 437                             QUSG(*usage, QDATA_IS_BLK(qdata)) < hlimit)
 438                                 qdata->qd_count = (hlimit -
 439                                         QUSG(*usage, QDATA_IS_BLK(qdata)))
 440                                         * (QDATA_IS_BLK(qdata) ?
 441                                            QUOTABLOCK_SIZE : 1);
 442                         else
 443                                 GOTO(out, rc = -EDQUOT);
 444                 }
 445
 446                 if (slimit &&
 447                     QUSG(*usage + qdata->qd_count, QDATA_IS_BLK(qdata)) > slimit) {
 448                         if (*time && cfs_time_current_sec() >= *time)
 449                                 GOTO(out, rc = -EDQUOT);
 450                         else if (!*time)
 451                                 *time = cfs_time_current_sec() + grace;
 452                 }
 453
 454                 *usage += qdata->qd_count;
 455                 break;
 456         case QUOTA_DQREL:
 457                 /* The usage in administrative file might be incorrect before
 458                  * recovery done */
 459                 if (*usage < qdata->qd_count)
 460                         *usage = 0;
 461                 else
 462                         *usage -= qdata->qd_count;
 463
 464                 /* (usage <= soft limit) but not (usage < soft limit) */
 465                 if (!slimit || QUSG(*usage, QDATA_IS_BLK(qdata)) <= slimit)
 466                         *time = 0;
 467                 break;
 468         default:
 469                 LBUG();
 470         }
 471
 472         rc = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
 473         EXIT;
 474 out:
 475         cfs_up(&dquot->dq_sem);
 476         cfs_up_write(&mds->mds_qonoff_sem);
 477         lustre_dqput(dquot);
 478         if (rc != -EDQUOT)
 479                 dqacq_adjust_qunit_sz(obd, qdata->qd_id, QDATA_IS_GRP(qdata),
 480                                       QDATA_IS_BLK(qdata));
 481
 482         cfs_spin_lock(&lqs->lqs_lock);
 483         qdata->qd_qunit  = QDATA_IS_BLK(qdata) ? lqs->lqs_bunit_sz :
 484                 lqs->lqs_iunit_sz;
 485         cfs_spin_unlock(&lqs->lqs_lock);
 486
 487         if (QDATA_IS_BLK(qdata))
 488                 QDATA_SET_ADJBLK(qdata);
 489         else
 490                 QDATA_SET_ADJINO(qdata);
 491
 492         QDATA_DEBUG(qdata, "alloc/release qunit in dqacq_handler\n");
 493 skip:
 494         lqs_putref(lqs);
 495
 496         return rc;
 497 }
 498
 499 int mds_quota_adjust(struct obd_device *obd, const unsigned int qcids[],
 500                      const unsigned int qpids[], int rc, int opc)
 501 {
 502         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
 503         int rc2 = 0;
 504         ENTRY;
 505
 506         if (rc && rc != -EDQUOT && rc != ENOLCK)
 507                 RETURN(0);
 508
 509         switch (opc) {
 510         case FSFILT_OP_SETATTR:
 511                 /* release file quota on original owner */
 512                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids, 0, 0, NULL);
 513                 /* release block quota on original owner */
 514                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids, 1, 0, NULL);
 515                 /* acquire file quota on current owner */
 516                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 0, 0, NULL);
 517                 /* acquire block quota on current owner */
 518                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0, NULL);
 519                 break;
 520         case FSFILT_OP_UNLINK_PARTIAL_CHILD:
 521                 /* release file quota on child */
 522                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 0, 0, NULL);
 523                 /* rlease block quota on child */
 524                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0, NULL);
 525                 break;
 526         case FSFILT_OP_CREATE_PARTIAL_CHILD:
 527                 /* acquire file quota on child */
 528                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 0, 0, NULL);
 529                 /* acquire block quota on child */
 530                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0, NULL);
 531                 break;
 532         case FSFILT_OP_LINK:
 533                 /* acquire block quota on parent */
 534                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids, 1, 0, NULL);
 535                 break;
 536         case FSFILT_OP_UNLINK:
 537                 /* release block quota on parent */
 538                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids, 1, 0, NULL);
 539                 /* release file quota on child */
 540                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 0, 0, NULL);
 541                 if (qpids[0] != qcids[0] || qpids[1] != qcids[1])
 542                         /* release block quota on child */
 543                         rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0,
 544                                                   NULL);
 545                 break;
 546         case FSFILT_OP_UNLINK_PARTIAL_PARENT:
 547                 /* release block quota on parent */
 548                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids, 1, 0, NULL);
 549                 break;
 550         case FSFILT_OP_CREATE:
 551                 /* acquire block quota on parent */
 552                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids, 1, 0, NULL);
 553                 /* acquire file quota on child */
 554                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 0, 0, NULL);
 555                 if (qpids[0] != qcids[0] || qpids[1] != qcids[1])
 556                         /* acquire block quota on child */
 557                         rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0,
 558                                                   NULL);
 559                 break;
 560         default:
 561                 LBUG();
 562                 break;
 563         }
 564
 565         if (rc2)
 566                 CDEBUG(D_QUOTA,
 567                        "mds adjust qunit %ssuccessfully! (opc:%d rc:%d)\n",
 568                        rc2 == QUOTA_REQ_RETURNED ? "" : "un", opc, rc2);
 569         RETURN(0);
 570 }
 571
 572 int filter_quota_adjust(struct obd_device *obd, const unsigned int qcids[],
 573                         const unsigned int qpids[], int rc, int opc)
 574 {
 575         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
 576         int rc2 = 0;
 577         ENTRY;
 578
 579         if (rc && rc != -EDQUOT)
 580                 RETURN(0);
 581
 582         switch (opc) {
 583         case FSFILT_OP_SETATTR:
 584                 /* acquire/release block quota on original & current owner */
 585                 rc = qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0, NULL);
 586                 rc2 = qctxt_adjust_qunit(obd, qctxt, qpids, 1, 0, NULL);
 587                 break;
 588         case FSFILT_OP_UNLINK:
 589                 /* release block quota on this owner */
 590         case FSFILT_OP_CREATE: /* XXX for write operation on obdfilter */
 591                 /* acquire block quota on this owner */
 592                 rc = qctxt_adjust_qunit(obd, qctxt, qcids, 1, 0, NULL);
 593                 break;
 594         default:
 595                 LBUG();
 596                 break;
 597         }
 598
 599         if (rc || rc2) {
 600                 if (!rc)
 601                         rc = rc2;
 602                 CDEBUG(D_QUOTA,
 603                        "filter adjust qunit %ssuccessfully! (opc:%d rc%d)\n",
 604                        rc == QUOTA_REQ_RETURNED ? "" : "un", opc, rc);
 605         }
 606
 607         RETURN(0);
 608 }
 609
 610 static const char prefix[] = "OBJECTS/";
 611
 612 int mds_quota_invalidate(struct obd_device *obd, struct obd_quotactl *oqctl)
 613 {
 614         struct mds_obd *mds = &obd->u.mds;
 615         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 616         struct obd_device_target *obt = &obd->u.obt;
 617         int rc = 0, i, rc1 = 0;
 618         char *quotafile[] = LUSTRE_ADMIN_QUOTAFILES_V2;
 619         char name[64];
 620         struct lvfs_run_ctxt saved;
 621         ENTRY;
 622
 623         LASSERT(qinfo->qi_version == LUSTRE_QUOTA_V2);
 624
 625         if (oqctl->qc_type != USRQUOTA &&
 626             oqctl->qc_type != GRPQUOTA &&
 627             oqctl->qc_type != UGQUOTA)
 628                 RETURN(-EINVAL);
 629
 630         cfs_down(&obt->obt_quotachecking);
 631         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 632         cfs_down_write(&mds->mds_qonoff_sem);
 633
 634         for (i = 0; i < MAXQUOTAS; i++) {
 635                 struct file *fp;
 636
 637                 if (!Q_TYPESET(oqctl, i))
 638                         continue;
 639
 640                 /* quota file has been opened ? */
 641                 if (qinfo->qi_files[i]) {
 642                         CWARN("quota[%d] is on yet\n", i);
 643                         rc1 = -EBUSY;
 644                         continue;
 645                 }
 646
 647                 LASSERT(strlen(quotafile[i]) + sizeof(prefix) <= sizeof(name));
 648                 sprintf(name, "%s%s", prefix, quotafile[i]);
 649
 650                 fp = filp_open(name, O_CREAT | O_TRUNC | O_RDWR, 0644);
 651                 if (IS_ERR(fp)) {
 652                         rc = PTR_ERR(fp);
 653                         CERROR("%s: error invalidating admin quotafile %s (rc:%d)\n",
 654                                obd->obd_name, name, rc);
 655                 }
 656                 else
 657                         filp_close(fp, 0);
 658         }
 659
 660         cfs_up_write(&mds->mds_qonoff_sem);
 661         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 662         cfs_up(&obt->obt_quotachecking);
 663         RETURN(rc ? : rc1);
 664 }
 665
 666 int mds_quota_finvalidate(struct obd_device *obd, struct obd_quotactl *oqctl)
 667 {
 668         struct mds_obd *mds = &obd->u.mds;
 669         struct obd_device_target *obt = &obd->u.obt;
 670         int rc;
 671         struct lvfs_run_ctxt saved;
 672         ENTRY;
 673
 674         if (oqctl->qc_type != USRQUOTA &&
 675             oqctl->qc_type != GRPQUOTA &&
 676             oqctl->qc_type != UGQUOTA)
 677                 RETURN(-EINVAL);
 678
 679         cfs_down(&obt->obt_quotachecking);
 680         if (obt->obt_qctxt.lqc_flags & UGQUOTA2LQC(oqctl->qc_type))
 681                 GOTO(out, rc = -EBUSY);
 682         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 683         cfs_down_write(&mds->mds_qonoff_sem);
 684
 685         oqctl->qc_cmd = Q_FINVALIDATE;
 686         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 687         if (!rc)
 688                 rc = obd_quotactl(mds->mds_lov_exp, oqctl);
 689
 690         cfs_up_write(&mds->mds_qonoff_sem);
 691         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 692 out:
 693         cfs_up(&obt->obt_quotachecking);
 694         RETURN(rc);
 695 }
 696
 697 int init_admin_quotafiles(struct obd_device *obd, struct obd_quotactl *oqctl)
 698 {
 699         struct mds_obd *mds = &obd->u.mds;
 700         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 701         const char *quotafile[] = LUSTRE_ADMIN_QUOTAFILES_V2;
 702         struct lvfs_run_ctxt saved;
 703         char name[64];
 704         int i, rc = 0;
 705         ENTRY;
 706
 707         LASSERT(qinfo->qi_version == LUSTRE_QUOTA_V2);
 708
 709         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 710         cfs_down_write(&mds->mds_qonoff_sem);
 711
 712         for (i = 0; i < MAXQUOTAS && !rc; i++) {
 713                 struct file *fp;
 714
 715                 if (!Q_TYPESET(oqctl, i))
 716                         continue;
 717
 718                 /* quota file has been opened ? */
 719                 if (qinfo->qi_files[i]) {
 720                         CWARN("init %s admin quotafile while quota on.\n",
 721                               i == USRQUOTA ? "user" : "group");
 722                         continue;
 723                 }
 724
 725                 LASSERT(strlen(quotafile[i]) + sizeof(prefix) <= sizeof(name));
 726                 sprintf(name, "%s%s", prefix, quotafile[i]);
 727
 728                 /* check if quota file exists and is correct */
 729                 fp = filp_open(name, O_RDONLY, 0);
 730                 if (!IS_ERR(fp)) {
 731                         /* irregular file is not the right place for quota */
 732                         if (!S_ISREG(fp->f_dentry->d_inode->i_mode)) {
 733                                 CERROR("admin quota file %s is not "
 734                                        "regular!", name);
 735                                 filp_close(fp, 0);
 736                                 rc = -EINVAL;
 737                                 break;
 738                         }
 739                         qinfo->qi_files[i] = fp;
 740                         rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_CHK);
 741                         qinfo->qi_files[i] = NULL;
 742                         filp_close(fp, 0);
 743                 }
 744                 else
 745                         rc = PTR_ERR(fp);
 746
 747                 if (!rc)
 748                         continue;
 749
 750                 /* -EINVAL may be returned by quotainfo for bad quota file */
 751                 if (rc != -ENOENT && rc != -EINVAL) {
 752                         CERROR("%s: error opening old quota file %s (%d)\n",
 753                                obd->obd_name, name, rc);
 754                         break;
 755                 }
 756
 757                 CDEBUG(D_INFO, "%s new quota file %s\n", name,
 758                        rc == -ENOENT ? "creating" : "overwriting");
 759
 760                 /* create quota file overwriting old if needed */
 761                 fp = filp_open(name, O_CREAT | O_TRUNC | O_RDWR, 0644);
 762                 if (IS_ERR(fp)) {
 763                         rc = PTR_ERR(fp);
 764                         CERROR("%s: error creating admin quotafile %s (rc:%d)\n",
 765                                obd->obd_name, name, rc);
 766                         break;
 767                 }
 768
 769                 qinfo->qi_files[i] = fp;
 770
 771                 rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_INIT_INFO);
 772                 if (rc)
 773                         CERROR("error init %s admin quotafile! (rc:%d)\n",
 774                                i == USRQUOTA ? "user" : "group", rc);
 775
 776                 filp_close(fp, 0);
 777                 qinfo->qi_files[i] = NULL;
 778         }
 779
 780         cfs_up_write(&mds->mds_qonoff_sem);
 781         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 782         RETURN(rc);
 783 }
 784
 785 static int close_quota_files(struct obd_quotactl *oqctl,
 786                              struct lustre_quota_info *qinfo)
 787 {
 788         int i, rc = 0;
 789         ENTRY;
 790
 791         for (i = 0; i < MAXQUOTAS; i++) {
 792                 if (!Q_TYPESET(oqctl, i))
 793                         continue;
 794                 if (qinfo->qi_files[i] == NULL) {
 795                         CDEBUG(D_QUOTA, "quota[%d] is off already\n", i);
 796                         rc = -EALREADY;
 797                         continue;
 798                 }
 799                 filp_close(qinfo->qi_files[i], 0);
 800                 qinfo->qi_files[i] = NULL;
 801         }
 802         RETURN(rc);
 803 }
 804
 805 int mds_admin_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
 806 {
 807         struct mds_obd *mds = &obd->u.mds;
 808         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 809         const char *quotafile[] = LUSTRE_ADMIN_QUOTAFILES_V2;
 810         char name[64];
 811         int i, rc = 0, rc1 = 0;
 812         ENTRY;
 813
 814         LASSERT(qinfo->qi_version == LUSTRE_QUOTA_V2);
 815
 816         /* open admin quota files and read quotafile info */
 817         for (i = 0; i < MAXQUOTAS; i++) {
 818                 struct file *fp;
 819
 820                 if (!Q_TYPESET(oqctl, i) || qinfo->qi_files[i] != NULL)
 821                         continue;
 822
 823                 LASSERT(strlen(quotafile[i])
 824                         + sizeof(prefix) <= sizeof(name));
 825                 sprintf(name, "%s%s", prefix, quotafile[i]);
 826                 fp = filp_open(name, O_RDWR, 0);
 827                 if (IS_ERR(fp) || !S_ISREG(fp->f_dentry->d_inode->i_mode)) {
 828                         rc = IS_ERR(fp) ? PTR_ERR(fp) : -EINVAL;
 829                         CERROR("error open/create %s! (rc:%d)\n", name, rc);
 830                         break;
 831                 }
 832                 qinfo->qi_files[i] = fp;
 833
 834                 rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_CHK);
 835                 if (rc) {
 836                         CERROR("invalid quota file %s! (rc:%d)\n", name, rc);
 837                         break;
 838                 }
 839
 840                 rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_RD_INFO);
 841                 if (rc) {
 842                         CERROR("error read quotainfo of %s! (rc:%d)\n", name,
 843                                rc);
 844                         break;
 845                 }
 846         }
 847
 848         if (rc && rc1 != -EALREADY)
 849                 close_quota_files(oqctl, qinfo);
 850
 851         RETURN(rc ? : rc1);
 852 }
 853
 854 int mds_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
 855 {
 856         int rc;
 857         ENTRY;
 858
 859         if (oqctl->qc_type != USRQUOTA &&
 860             oqctl->qc_type != GRPQUOTA &&
 861             oqctl->qc_type != UGQUOTA)
 862                 RETURN(-EINVAL);
 863
 864         rc = generic_quota_on(obd, oqctl, 1);
 865
 866         RETURN(rc);
 867 }
 868
 869
 870 int mds_admin_quota_off(struct obd_device *obd,
 871                         struct obd_quotactl *oqctl)
 872 {
 873         struct mds_obd *mds = &obd->u.mds;
 874         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 875         int rc;
 876         ENTRY;
 877
 878         /* close admin quota files */
 879         rc = close_quota_files(oqctl, qinfo);
 880         RETURN(rc);
 881 }
 882
 883
 884 /* with obt->obt_quotachecking held */
 885 int do_mds_quota_off(struct obd_device *obd, struct obd_quotactl *oqctl)
 886 {
 887         struct mds_obd *mds = &obd->u.mds;
 888         struct obd_device_target *obt = &obd->u.obt;
 889         struct lustre_quota_ctxt *qctxt = &obt->obt_qctxt;
 890         struct lvfs_run_ctxt saved;
 891         int rc = 0, rc1 = 0, rc2 = 0;
 892         ENTRY;
 893
 894         LASSERT_SEM_LOCKED(&obt->obt_quotachecking);
 895
 896         if (oqctl->qc_type != USRQUOTA &&
 897             oqctl->qc_type != GRPQUOTA &&
 898             oqctl->qc_type != UGQUOTA)
 899                 RETURN(-EINVAL);
 900
 901         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 902         cfs_down_write(&mds->mds_qonoff_sem);
 903         /* close admin quota files */
 904         rc2 = mds_admin_quota_off(obd, oqctl);
 905         if (rc2 && rc2 != -EALREADY) {
 906                 CWARN("mds quota[%d] is failed to be off for %d\n", oqctl->qc_type, rc2);
 907                 GOTO(out, rc2);
 908         }
 909
 910         rc1 = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 911         if (!rc1) {
 912                 obt->obt_qctxt.lqc_flags &= ~UGQUOTA2LQC(oqctl->qc_type);
 913         } else if (quota_is_off(qctxt, oqctl)) {
 914                 CWARN("mds local quota[%d] is off already\n", oqctl->qc_type);
 915                 rc1 = -EALREADY;
 916         } else {
 917                 if (rc2 != -EALREADY) {
 918                         CWARN("mds local quota[%d] is failed to be off for %d\n",
 919                               oqctl->qc_type, rc1);
 920                         oqctl->qc_cmd = Q_QUOTAON;
 921                         mds_admin_quota_on(obd, oqctl);
 922                         oqctl->qc_cmd = Q_QUOTAOFF;
 923                 }
 924                 GOTO(out, rc1);
 925         }
 926
 927         rc = obd_quotactl(mds->mds_lov_exp, oqctl);
 928         if (rc && rc != -EALREADY) {
 929                 CWARN("mds remote quota[%d] is failed to be off for %d\n",
 930                       oqctl->qc_type, rc);
 931                 oqctl->qc_cmd = Q_QUOTAON;
 932                 if (rc2 != -EALREADY)
 933                         mds_admin_quota_on(obd, oqctl);
 934                 if (rc1 != -EALREADY) {
 935                         fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 936                         qctxt->lqc_flags |= UGQUOTA2LQC(oqctl->qc_type);
 937                 }
 938                 oqctl->qc_cmd = Q_QUOTAOFF;
 939         }
 940         EXIT;
 941
 942 out:
 943         CDEBUG(D_QUOTA, "%s: quotaoff type:flags:rc %u:%lu:%d\n",
 944                obd->obd_name, oqctl->qc_type, qctxt->lqc_flags, rc);
 945         cfs_up_write(&mds->mds_qonoff_sem);
 946         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 947         return rc ? : (rc1 ? : rc2);
 948 }
 949
 950 int mds_quota_off(struct obd_device *obd, struct obd_quotactl *oqctl)
 951 {
 952         struct obd_device_target *obt = &obd->u.obt;
 953         int rc;
 954         ENTRY;
 955
 956         cfs_down(&obt->obt_quotachecking);
 957         rc = do_mds_quota_off(obd, oqctl);
 958         cfs_up(&obt->obt_quotachecking);
 959         RETURN(rc);
 960 }
 961
 962 int mds_set_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
 963 {
 964         struct mds_obd *mds = &obd->u.mds;
 965         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 966         struct obd_dqinfo *dqinfo = &oqctl->qc_dqinfo;
 967         int rc;
 968         ENTRY;
 969
 970         if (oqctl->qc_type != USRQUOTA &&
 971             oqctl->qc_type != GRPQUOTA)
 972                 RETURN(-EINVAL);
 973
 974         cfs_down_write(&mds->mds_qonoff_sem);
 975         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
 976                 CWARN("quota[%u] is off\n", oqctl->qc_type);
 977                 GOTO(out, rc = -ESRCH);
 978         }
 979
 980         qinfo->qi_info[oqctl->qc_type].dqi_bgrace = dqinfo->dqi_bgrace;
 981         qinfo->qi_info[oqctl->qc_type].dqi_igrace = dqinfo->dqi_igrace;
 982         qinfo->qi_info[oqctl->qc_type].dqi_flags = dqinfo->dqi_flags;
 983
 984         rc = fsfilt_quotainfo(obd, qinfo, oqctl->qc_type, QFILE_WR_INFO);
 985         EXIT;
 986
 987 out:
 988         cfs_up_write(&mds->mds_qonoff_sem);
 989         return rc;
 990 }
 991
 992 int mds_get_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
 993 {
 994         struct mds_obd *mds = &obd->u.mds;
 995         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 996         struct obd_dqinfo *dqinfo = &oqctl->qc_dqinfo;
 997         int rc = 0;
 998         ENTRY;
 999
1000         if (oqctl->qc_type != USRQUOTA &&
1001             oqctl->qc_type != GRPQUOTA)
1002                 RETURN(-EINVAL);
1003
1004         cfs_down_read(&mds->mds_qonoff_sem);
1005         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
1006                 CWARN("quota[%u] is off\n", oqctl->qc_type);
1007                 GOTO(out, rc = -ESRCH);
1008         }
1009
1010         dqinfo->dqi_bgrace = qinfo->qi_info[oqctl->qc_type].dqi_bgrace;
1011         dqinfo->dqi_igrace = qinfo->qi_info[oqctl->qc_type].dqi_igrace;
1012         dqinfo->dqi_flags = qinfo->qi_info[oqctl->qc_type].dqi_flags;
1013         EXIT;
1014
1015 out:
1016         cfs_up_read(&mds->mds_qonoff_sem);
1017         return rc;
1018 }
1019
1020 int dquot_create_oqaq(struct lustre_quota_ctxt *qctxt,
1021                       struct lustre_dquot *dquot, __u32 ost_num, __u32 mdt_num,
1022                       int type, struct quota_adjust_qunit *oqaq)
1023 {
1024         __u64 bunit_curr_o, iunit_curr_o;
1025         unsigned long shrink_qunit_limit = qctxt->lqc_cqs_boundary_factor;
1026         unsigned long cqs_factor = qctxt->lqc_cqs_qs_factor;
1027         __u64 blimit = dquot->dq_dqb.dqb_bhardlimit ?
1028                 dquot->dq_dqb.dqb_bhardlimit : dquot->dq_dqb.dqb_bsoftlimit;
1029         __u64 ilimit = dquot->dq_dqb.dqb_ihardlimit ?
1030                 dquot->dq_dqb.dqb_ihardlimit : dquot->dq_dqb.dqb_isoftlimit;
1031         int rc = 0;
1032         ENTRY;
1033
1034         if (!dquot || !oqaq)
1035                 RETURN(-EINVAL);
1036         LASSERT_SEM_LOCKED(&dquot->dq_sem);
1037         LASSERT(oqaq->qaq_iunit_sz);
1038         LASSERT(oqaq->qaq_bunit_sz);
1039
1040         /* don't change qunit size */
1041         if (!qctxt->lqc_switch_qs)
1042                 RETURN(rc);
1043
1044         bunit_curr_o = oqaq->qaq_bunit_sz;
1045         iunit_curr_o = oqaq->qaq_iunit_sz;
1046
1047         if (dquot->dq_type == GRPQUOTA)
1048                 QAQ_SET_GRP(oqaq);
1049
1050         if ((type & LQUOTA_FLAGS_ADJBLK) && blimit) {
1051                 __u64 b_limitation =
1052                         oqaq->qaq_bunit_sz * (ost_num + 1) * shrink_qunit_limit;
1053                 /* enlarge block qunit size */
1054                 while (blimit >
1055                        QUSG(dquot->dq_dqb.dqb_curspace + 2 * b_limitation, 1)) {
1056                         oqaq->qaq_bunit_sz =
1057                                 QUSG(oqaq->qaq_bunit_sz * cqs_factor, 1)
1058                                 << QUOTABLOCK_BITS;
1059                         if (oqaq->qaq_bunit_sz >= qctxt->lqc_bunit_sz)
1060                                 break;
1061                         b_limitation = oqaq->qaq_bunit_sz * (ost_num + 1) *
1062                                 shrink_qunit_limit;
1063                 }
1064
1065                 if (oqaq->qaq_bunit_sz > qctxt->lqc_bunit_sz)
1066                         oqaq->qaq_bunit_sz = qctxt->lqc_bunit_sz;
1067
1068                 /* shrink block qunit size */
1069                 while (blimit <
1070                        QUSG(dquot->dq_dqb.dqb_curspace + b_limitation, 1)) {
1071                         do_div(oqaq->qaq_bunit_sz , cqs_factor);
1072                         oqaq->qaq_bunit_sz = QUSG(oqaq->qaq_bunit_sz, 1) <<
1073                                 QUOTABLOCK_BITS;
1074                         b_limitation = oqaq->qaq_bunit_sz * (ost_num + 1) *
1075                                 shrink_qunit_limit;
1076                         if (oqaq->qaq_bunit_sz <  qctxt->lqc_cqs_least_bunit)
1077                                 break;
1078                 }
1079
1080                 if (oqaq->qaq_bunit_sz < qctxt->lqc_cqs_least_bunit)
1081                         oqaq->qaq_bunit_sz = qctxt->lqc_cqs_least_bunit;
1082
1083                 if (bunit_curr_o != oqaq->qaq_bunit_sz)
1084                         QAQ_SET_ADJBLK(oqaq);
1085
1086         }
1087
1088         if ((type & LQUOTA_FLAGS_ADJINO) && ilimit) {
1089                 __u64 i_limitation =
1090                         oqaq->qaq_iunit_sz * mdt_num * shrink_qunit_limit;
1091                 /* enlarge file qunit size */
1092                 while (ilimit > dquot->dq_dqb.dqb_curinodes
1093                        + 2 * i_limitation) {
1094                         oqaq->qaq_iunit_sz = oqaq->qaq_iunit_sz * cqs_factor;
1095                         if (oqaq->qaq_iunit_sz >= qctxt->lqc_iunit_sz)
1096                                 break;
1097                         i_limitation = oqaq->qaq_iunit_sz * mdt_num *
1098                                 shrink_qunit_limit;
1099                 }
1100
1101                 if (oqaq->qaq_iunit_sz > qctxt->lqc_iunit_sz)
1102                         oqaq->qaq_iunit_sz = qctxt->lqc_iunit_sz;
1103
1104                 /* shrink file qunit size */
1105                 while (ilimit < dquot->dq_dqb.dqb_curinodes
1106                        + i_limitation) {
1107                         do_div(oqaq->qaq_iunit_sz, cqs_factor);
1108                         i_limitation = oqaq->qaq_iunit_sz * mdt_num *
1109                                        shrink_qunit_limit;
1110                         if (oqaq->qaq_iunit_sz < qctxt->lqc_cqs_least_iunit)
1111                                 break;
1112                 }
1113
1114                 if (oqaq->qaq_iunit_sz < qctxt->lqc_cqs_least_iunit)
1115                         oqaq->qaq_iunit_sz = qctxt->lqc_cqs_least_iunit;
1116
1117                 if (iunit_curr_o != oqaq->qaq_iunit_sz)
1118                         QAQ_SET_ADJINO(oqaq);
1119
1120         }
1121
1122         QAQ_DEBUG(oqaq, "the oqaq computed\n");
1123
1124         RETURN(rc);
1125 }
1126
1127 static int mds_init_slave_ilimits(struct obd_device *obd,
1128                                   struct obd_quotactl *oqctl, int set)
1129 {
1130         /* XXX: for file limits only adjust local now */
1131         struct obd_device_target *obt = &obd->u.obt;
1132         struct lustre_quota_ctxt *qctxt = &obt->obt_qctxt;
1133         unsigned int id[MAXQUOTAS] = { 0, 0 };
1134         struct obd_quotactl *ioqc = NULL;
1135         struct lustre_qunit_size *lqs;
1136         int flag;
1137         int rc;
1138         ENTRY;
1139
1140         /* if we are going to set zero limit, needn't init slaves */
1141         if (!oqctl->qc_dqblk.dqb_ihardlimit && !oqctl->qc_dqblk.dqb_isoftlimit &&
1142             !set)
1143                 RETURN(0);
1144
1145         OBD_ALLOC_PTR(ioqc);
1146         if (!ioqc)
1147                 RETURN(-ENOMEM);
1148
1149         flag = oqctl->qc_dqblk.dqb_ihardlimit ||
1150                oqctl->qc_dqblk.dqb_isoftlimit || !set;
1151         ioqc->qc_cmd = flag ? Q_INITQUOTA : Q_SETQUOTA;
1152         ioqc->qc_id = oqctl->qc_id;
1153         ioqc->qc_type = oqctl->qc_type;
1154         ioqc->qc_dqblk.dqb_valid = QIF_ILIMITS;
1155         ioqc->qc_dqblk.dqb_ihardlimit = flag ? MIN_QLIMIT : 0;
1156
1157         /* build lqs for mds */
1158         lqs = quota_search_lqs(LQS_KEY(oqctl->qc_type, oqctl->qc_id),
1159                                qctxt, flag ? 1 : 0);
1160         if (lqs && !IS_ERR(lqs)) {
1161                 if (flag)
1162                         lqs->lqs_flags |= QI_SET;
1163                 else
1164                         lqs->lqs_flags &= ~QI_SET;
1165                 lqs_putref(lqs);
1166         } else {
1167                 CERROR("fail to %s lqs for inode(%s id: %u)!\n",
1168                        flag ? "create" : "search",
1169                        oqctl->qc_type ? "group" : "user",
1170                        oqctl->qc_id);
1171                 GOTO(out, rc = PTR_ERR(lqs));
1172         }
1173
1174         /* set local limit to MIN_QLIMIT */
1175         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, ioqc);
1176         if (rc)
1177                 GOTO(out, rc);
1178
1179         /* trigger local qunit pre-acquire */
1180         if (oqctl->qc_type == USRQUOTA)
1181                 id[USRQUOTA] = oqctl->qc_id;
1182         else
1183                 id[GRPQUOTA] = oqctl->qc_id;
1184
1185         rc = qctxt_adjust_qunit(obd, &obd->u.obt.obt_qctxt, id, 0, 0, NULL);
1186         if (rc == -EDQUOT || rc == -EBUSY) {
1187                 CDEBUG(D_QUOTA, "rc: %d.\n", rc);
1188                 rc = 0;
1189         }
1190         if (rc) {
1191                 CDEBUG(D_QUOTA,"error mds adjust local file quota! (rc:%d)\n",
1192                        rc);
1193                 GOTO(out, rc);
1194         }
1195         /* FIXME initialize all slaves in CMD */
1196         EXIT;
1197 out:
1198         if (ioqc)
1199                 OBD_FREE_PTR(ioqc);
1200         return rc;
1201 }
1202
1203 static int mds_init_slave_blimits(struct obd_device *obd,
1204                                   struct obd_quotactl *oqctl, int set)
1205 {
1206         struct obd_device_target *obt = &obd->u.obt;
1207         struct lustre_quota_ctxt *qctxt = &obt->obt_qctxt;
1208         struct mds_obd *mds = &obd->u.mds;
1209         struct obd_quotactl *ioqc;
1210         struct lustre_qunit_size *lqs;
1211         unsigned int id[MAXQUOTAS] = { 0, 0 };
1212         int rc;
1213         int flag;
1214         ENTRY;
1215
1216         /* if we are going to set zero limit, needn't init slaves */
1217         if (!oqctl->qc_dqblk.dqb_bhardlimit && !oqctl->qc_dqblk.dqb_bsoftlimit &&
1218             !set)
1219                 RETURN(0);
1220
1221         OBD_ALLOC_PTR(ioqc);
1222         if (!ioqc)
1223                 RETURN(-ENOMEM);
1224
1225         flag = oqctl->qc_dqblk.dqb_bhardlimit ||
1226                oqctl->qc_dqblk.dqb_bsoftlimit || !set;
1227         ioqc->qc_cmd = flag ? Q_INITQUOTA : Q_SETQUOTA;
1228         ioqc->qc_id = oqctl->qc_id;
1229         ioqc->qc_type = oqctl->qc_type;
1230         ioqc->qc_dqblk.dqb_valid = QIF_BLIMITS;
1231         ioqc->qc_dqblk.dqb_bhardlimit = flag ? MIN_QLIMIT : 0;
1232
1233         /* build lqs for mds */
1234         lqs = quota_search_lqs(LQS_KEY(oqctl->qc_type, oqctl->qc_id),
1235                                qctxt, flag ? 1 : 0);
1236         if (lqs && !IS_ERR(lqs)) {
1237                 if (flag)
1238                         lqs->lqs_flags |= QB_SET;
1239                 else
1240                         lqs->lqs_flags &= ~QB_SET;
1241                 lqs_putref(lqs);
1242         } else {
1243                 CERROR("fail to %s lqs for block(%s id: %u)!\n",
1244                        flag ? "create" : "search",
1245                        oqctl->qc_type ? "group" : "user",
1246                        oqctl->qc_id);
1247                 GOTO(out, rc = PTR_ERR(lqs));
1248         }
1249
1250         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, ioqc);
1251         if (rc)
1252                 GOTO(out, rc);
1253
1254         /* trigger local qunit pre-acquire */
1255         if (oqctl->qc_type == USRQUOTA)
1256                 id[USRQUOTA] = oqctl->qc_id;
1257         else
1258                 id[GRPQUOTA] = oqctl->qc_id;
1259
1260         /* initialize all slave's limit */
1261         rc = obd_quotactl(mds->mds_lov_exp, ioqc);
1262
1263         rc = qctxt_adjust_qunit(obd, &obd->u.obt.obt_qctxt, id, 1, 0, NULL);
1264         if (rc == -EDQUOT || rc == -EBUSY) {
1265                 CDEBUG(D_QUOTA, "rc: %d.\n", rc);
1266                 rc = 0;
1267         }
1268         if (rc) {
1269                 CERROR("error mds adjust local block quota! (rc:%d)\n", rc);
1270                 GOTO(out, rc);
1271         }
1272
1273         EXIT;
1274 out:
1275         OBD_FREE_PTR(ioqc);
1276         return rc;
1277 }
1278
1279 static void adjust_lqs(struct obd_device *obd, struct quota_adjust_qunit *qaq)
1280 {
1281         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
1282         int rc = 0;
1283
1284         QAQ_SET_CREATE_LQS(qaq);
1285         /* adjust local lqs */
1286         rc = quota_adjust_slave_lqs(qaq, qctxt);
1287         if (rc < 0)
1288                 CERROR("adjust master's qunit size failed!(rc=%d)\n", rc);
1289
1290         /* adjust remote lqs */
1291         if (QAQ_IS_ADJBLK(qaq)) {
1292                 rc = obd_quota_adjust_qunit(obd->u.mds.mds_lov_exp, qaq, qctxt, NULL);
1293                 if (rc < 0)
1294                         CERROR("adjust slaves' qunit size failed!(rc=%d)\n", rc);
1295
1296         }
1297 }
1298
1299 int mds_set_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
1300 {
1301         struct mds_obd *mds = &obd->u.mds;
1302         struct lustre_quota_ctxt *qctxt = &mds->mds_obt.obt_qctxt;
1303         struct obd_device *lov_obd = class_exp2obd(mds->mds_lov_exp);
1304         struct lov_obd *lov = &lov_obd->u.lov;
1305         struct quota_adjust_qunit *oqaq = NULL;
1306         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
1307         __u64 ihardlimit, isoftlimit, bhardlimit, bsoftlimit;
1308         time_t btime, itime;
1309         struct lustre_dquot *dquot;
1310         struct obd_dqblk *dqblk = &oqctl->qc_dqblk;
1311         /* orig_set means if quota was set before; now_set means we are
1312          * setting/cancelling quota */
1313         int orig_set, now_set;
1314         struct lustre_qunit_size *lqs;
1315         int rc = 0, rc2 = 0, flag = 0;
1316         ENTRY;
1317
1318         if (oqctl->qc_type != USRQUOTA &&
1319             oqctl->qc_type != GRPQUOTA)
1320                 RETURN(-EINVAL);
1321
1322         OBD_ALLOC_PTR(oqaq);
1323         if (!oqaq)
1324                 RETURN(-ENOMEM);
1325
1326         cfs_down_write(&mds->mds_qonoff_sem);
1327         init_oqaq(oqaq, qctxt, oqctl->qc_id, oqctl->qc_type);
1328
1329         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
1330                 CWARN("quota[%u] is off\n", oqctl->qc_type);
1331                 GOTO(out_sem, rc = -ESRCH);
1332         }
1333
1334         dquot = lustre_dqget(obd, qinfo, oqctl->qc_id, oqctl->qc_type, 1);
1335         if (IS_ERR(dquot))
1336                 GOTO(out_sem, rc = PTR_ERR(dquot));
1337         DQUOT_DEBUG(dquot, "get dquot in mds_set_blk\n");
1338         QINFO_DEBUG(dquot->dq_info, "get dquot in mds_set_blk\n");
1339
1340         lqs = quota_search_lqs(LQS_KEY(oqctl->qc_type, oqctl->qc_id), qctxt, 1);
1341         if (lqs == NULL)
1342                 rc = -ENOENT;
1343         if (IS_ERR(lqs))
1344                 rc = PTR_ERR(lqs);
1345         if (rc)
1346                 GOTO(out, rc);
1347
1348         cfs_down(&dquot->dq_sem);
1349         cfs_spin_lock(&lqs->lqs_lock);
1350         if (LQS_IS_SETQUOTA(lqs) || LQS_IS_RECOVERY(lqs)) {
1351                 cfs_spin_unlock(&lqs->lqs_lock);
1352                 cfs_up(&dquot->dq_sem);
1353                 GOTO(skip, rc = -EBUSY);
1354         }
1355         LQS_SET_SETQUOTA(lqs);
1356         cfs_spin_unlock(&lqs->lqs_lock);
1357
1358         ihardlimit = dquot->dq_dqb.dqb_ihardlimit;
1359         isoftlimit = dquot->dq_dqb.dqb_isoftlimit;
1360         bhardlimit = dquot->dq_dqb.dqb_bhardlimit;
1361         bsoftlimit = dquot->dq_dqb.dqb_bsoftlimit;
1362         btime = dquot->dq_dqb.dqb_btime;
1363         itime = dquot->dq_dqb.dqb_itime;
1364
1365         if (dqblk->dqb_valid & QIF_BTIME)
1366                 dquot->dq_dqb.dqb_btime = dqblk->dqb_btime;
1367         if (dqblk->dqb_valid & QIF_ITIME)
1368                 dquot->dq_dqb.dqb_itime = dqblk->dqb_itime;
1369
1370         if (dqblk->dqb_valid & QIF_BLIMITS) {
1371                 dquot->dq_dqb.dqb_bhardlimit = dqblk->dqb_bhardlimit;
1372                 dquot->dq_dqb.dqb_bsoftlimit = dqblk->dqb_bsoftlimit;
1373                 /* clear usage (limit pool) */
1374                 if (!dquot->dq_dqb.dqb_bhardlimit &&
1375                     !dquot->dq_dqb.dqb_bsoftlimit)
1376                         dquot->dq_dqb.dqb_curspace = 0;
1377
1378                 /* clear grace time */
1379                 if (!dqblk->dqb_bsoftlimit ||
1380                     toqb(dquot->dq_dqb.dqb_curspace) <= dqblk->dqb_bsoftlimit)
1381                         dquot->dq_dqb.dqb_btime = 0;
1382                 /* set grace only if user hasn't provided his own */
1383                 else if (!(dqblk->dqb_valid & QIF_BTIME))
1384                         dquot->dq_dqb.dqb_btime = cfs_time_current_sec() +
1385                                 qinfo->qi_info[dquot->dq_type].dqi_bgrace;
1386
1387                 flag |= LQUOTA_FLAGS_ADJBLK;
1388         }
1389
1390         if (dqblk->dqb_valid & QIF_ILIMITS) {
1391                 dquot->dq_dqb.dqb_ihardlimit = dqblk->dqb_ihardlimit;
1392                 dquot->dq_dqb.dqb_isoftlimit = dqblk->dqb_isoftlimit;
1393                 /* clear usage (limit pool) */
1394                 if (!dquot->dq_dqb.dqb_ihardlimit &&
1395                     !dquot->dq_dqb.dqb_isoftlimit)
1396                         dquot->dq_dqb.dqb_curinodes = 0;
1397
1398                 if (!dqblk->dqb_isoftlimit ||
1399                     dquot->dq_dqb.dqb_curinodes <= dqblk->dqb_isoftlimit)
1400                         dquot->dq_dqb.dqb_itime = 0;
1401                 else if (!(dqblk->dqb_valid & QIF_ITIME))
1402                         dquot->dq_dqb.dqb_itime = cfs_time_current_sec() +
1403                                 qinfo->qi_info[dquot->dq_type].dqi_igrace;
1404
1405                 flag |= LQUOTA_FLAGS_ADJINO;
1406         }
1407         QAQ_DEBUG(oqaq, "before dquot_create_oqaq\n");
1408         rc = dquot_create_oqaq(qctxt, dquot, lov->desc.ld_tgt_count, 1,
1409                                flag, oqaq);
1410         QAQ_DEBUG(oqaq, "after dquot_create_oqaq\n");
1411         if (rc < 0)
1412                 CDEBUG(D_QUOTA, "adjust qunit size failed! (rc:%d)\n", rc);
1413
1414
1415         rc = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
1416
1417         cfs_up(&dquot->dq_sem);
1418
1419         if (rc) {
1420                 CERROR("set limit failed! (rc:%d)\n", rc);
1421                 GOTO(update_fail, rc);
1422         }
1423
1424         cfs_up_write(&mds->mds_qonoff_sem);
1425         adjust_lqs(obd, oqaq);
1426
1427         orig_set = ihardlimit || isoftlimit;
1428         now_set  = dqblk->dqb_ihardlimit || dqblk->dqb_isoftlimit;
1429         if (dqblk->dqb_valid & QIF_ILIMITS && orig_set != now_set) {
1430                 cfs_down(&dquot->dq_sem);
1431                 dquot->dq_dqb.dqb_curinodes = 0;
1432                 cfs_up(&dquot->dq_sem);
1433                 rc = mds_init_slave_ilimits(obd, oqctl, orig_set);
1434                 if (rc) {
1435                         CERROR("init slave ilimits failed! (rc:%d)\n", rc);
1436                         goto revoke_out;
1437                 }
1438         }
1439
1440         orig_set = bhardlimit || bsoftlimit;
1441         now_set  = dqblk->dqb_bhardlimit || dqblk->dqb_bsoftlimit;
1442         if (dqblk->dqb_valid & QIF_BLIMITS && orig_set != now_set) {
1443                 cfs_down(&dquot->dq_sem);
1444                 dquot->dq_dqb.dqb_curspace = 0;
1445                 cfs_up(&dquot->dq_sem);
1446                 rc = mds_init_slave_blimits(obd, oqctl, orig_set);
1447                 if (rc) {
1448                         CERROR("init slave blimits failed! (rc:%d)\n", rc);
1449                         goto revoke_out;
1450                 }
1451         }
1452
1453 revoke_out:
1454         cfs_down_write(&mds->mds_qonoff_sem);
1455         cfs_down(&dquot->dq_sem);
1456         if (rc) {
1457                 /* cancel previous setting */
1458                 dquot->dq_dqb.dqb_ihardlimit = ihardlimit;
1459                 dquot->dq_dqb.dqb_isoftlimit = isoftlimit;
1460                 dquot->dq_dqb.dqb_bhardlimit = bhardlimit;
1461                 dquot->dq_dqb.dqb_bsoftlimit = bsoftlimit;
1462                 dquot->dq_dqb.dqb_btime = btime;
1463                 dquot->dq_dqb.dqb_itime = itime;
1464         }
1465         rc2 = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
1466         cfs_up(&dquot->dq_sem);
1467 update_fail:
1468         cfs_spin_lock(&lqs->lqs_lock);
1469         LQS_CLEAR_SETQUOTA(lqs);
1470         cfs_spin_unlock(&lqs->lqs_lock);
1471 skip:
1472         lqs_putref(lqs);
1473 out:
1474         lustre_dqput(dquot);
1475         EXIT;
1476 out_sem:
1477         cfs_up_write(&mds->mds_qonoff_sem);
1478
1479         if (oqaq)
1480                 OBD_FREE_PTR(oqaq);
1481
1482         return rc ? rc : rc2;
1483 }
1484
1485 static int mds_get_space(struct obd_device *obd, struct obd_quotactl *oqctl)
1486 {
1487         struct obd_quotactl *soqc;
1488         struct lvfs_run_ctxt saved;
1489         int rc, rc1;
1490         ENTRY;
1491
1492         OBD_ALLOC_PTR(soqc);
1493         if (!soqc)
1494                 RETURN(-ENOMEM);
1495
1496         soqc->qc_cmd = Q_GETOQUOTA;
1497         soqc->qc_id = oqctl->qc_id;
1498         soqc->qc_type = oqctl->qc_type;
1499
1500         /* get block usage from OSS */
1501         soqc->qc_dqblk.dqb_curspace = 0;
1502         rc = obd_quotactl(obd->u.mds.mds_lov_exp, soqc);
1503         if (!rc || rc == -EREMOTEIO) {
1504                 oqctl->qc_dqblk.dqb_curspace = soqc->qc_dqblk.dqb_curspace;
1505                 oqctl->qc_dqblk.dqb_valid |= QIF_SPACE;
1506         }
1507
1508         /* get block/inode usage from MDS */
1509         soqc->qc_dqblk.dqb_curspace = 0;
1510         soqc->qc_dqblk.dqb_curinodes = 0;
1511         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1512         rc1 = fsfilt_quotactl(obd, obd->u.obt.obt_sb, soqc);
1513         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1514         if (!rc1) {
1515                 oqctl->qc_dqblk.dqb_curspace += soqc->qc_dqblk.dqb_curspace;
1516                 oqctl->qc_dqblk.dqb_curinodes = soqc->qc_dqblk.dqb_curinodes;
1517                 oqctl->qc_dqblk.dqb_valid |= QIF_INODES;
1518         }
1519
1520         OBD_FREE_PTR(soqc);
1521
1522         RETURN(rc ? : rc1);
1523 }
1524
1525 int mds_get_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
1526 {
1527         struct mds_obd *mds = &obd->u.mds;
1528         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
1529         struct lustre_dquot *dquot;
1530         struct obd_dqblk *dqblk = &oqctl->qc_dqblk;
1531         int rc;
1532         ENTRY;
1533
1534         if (oqctl->qc_type != USRQUOTA &&
1535             oqctl->qc_type != GRPQUOTA)
1536                 RETURN(-EINVAL);
1537
1538         cfs_down_read(&mds->mds_qonoff_sem);
1539         dqblk->dqb_valid = 0;
1540         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
1541                 CWARN("quota[%u] is off\n", oqctl->qc_type);
1542                 GOTO(out, rc = -ESRCH);
1543         }
1544
1545         dquot = lustre_dqget(obd, qinfo, oqctl->qc_id, oqctl->qc_type, 1);
1546         if (IS_ERR(dquot))
1547                 GOTO(out, rc = PTR_ERR(dquot));
1548
1549         cfs_down(&dquot->dq_sem);
1550         dqblk->dqb_ihardlimit = dquot->dq_dqb.dqb_ihardlimit;
1551         dqblk->dqb_isoftlimit = dquot->dq_dqb.dqb_isoftlimit;
1552         dqblk->dqb_bhardlimit = dquot->dq_dqb.dqb_bhardlimit;
1553         dqblk->dqb_bsoftlimit = dquot->dq_dqb.dqb_bsoftlimit;
1554         dqblk->dqb_btime = dquot->dq_dqb.dqb_btime;
1555         dqblk->dqb_itime = dquot->dq_dqb.dqb_itime;
1556         dqblk->dqb_valid |= QIF_LIMITS | QIF_TIMES;
1557         cfs_up(&dquot->dq_sem);
1558
1559         lustre_dqput(dquot);
1560         cfs_up_read(&mds->mds_qonoff_sem);
1561
1562         /* the usages in admin quota file is inaccurate */
1563         dqblk->dqb_curinodes = 0;
1564         dqblk->dqb_curspace = 0;
1565         rc = mds_get_space(obd, oqctl);
1566
1567         RETURN(rc);
1568
1569 out:
1570         cfs_up_read(&mds->mds_qonoff_sem);
1571         return rc;
1572 }
1573
1574 int mds_get_obd_quota(struct obd_device *obd, struct obd_quotactl *oqctl)
1575 {
1576         struct lvfs_run_ctxt saved;
1577         int rc;
1578         ENTRY;
1579
1580         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1581         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
1582         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
1583
1584         RETURN(rc);
1585 }
1586
1587
1588 /* FIXME we only recovery block limit by now, need recovery inode
1589  * limits also after CMD involved in */
1590 static int
1591 dquot_recovery(struct obd_device *obd, unsigned int id, unsigned short type)
1592 {
1593         struct mds_obd *mds = &obd->u.mds;
1594         struct lustre_quota_ctxt *qctxt = &mds->mds_obt.obt_qctxt;
1595         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
1596         struct lustre_qunit_size *lqs;
1597         struct lustre_dquot *dquot;
1598         struct obd_quotactl *qctl;
1599         __u64 total_limits = 0;
1600         int rc = 0;
1601         ENTRY;
1602
1603         OBD_ALLOC_PTR(qctl);
1604         if (qctl == NULL)
1605                 RETURN(-ENOMEM);
1606
1607         dquot = lustre_dqget(obd, qinfo, id, type, 0);
1608         if (IS_ERR(dquot)) {
1609                 CERROR("Get dquot failed. (rc:%ld)\n", PTR_ERR(dquot));
1610                 OBD_FREE_PTR(qctl);
1611                 RETURN(PTR_ERR(dquot));
1612         }
1613
1614         lqs = quota_search_lqs(LQS_KEY(type, id), qctxt, 1);
1615         if (lqs == NULL)
1616                 rc = -ENOENT;
1617         if (IS_ERR(lqs))
1618                 rc = PTR_ERR(lqs);
1619         if (rc)
1620                 GOTO(skip, rc);
1621
1622         cfs_down(&dquot->dq_sem);
1623
1624         /* don't recover the dquot without limits or quota is setting or
1625          * another recovery is already going on */
1626         if (!(dquot->dq_dqb.dqb_bhardlimit || dquot->dq_dqb.dqb_bsoftlimit) ||
1627             LQS_IS_SETQUOTA(lqs) || LQS_IS_RECOVERY(lqs)) {
1628                 cfs_up(&dquot->dq_sem);
1629                 GOTO(skip1, rc = 0);
1630         }
1631
1632         cfs_spin_lock(&lqs->lqs_lock);
1633         LQS_SET_RECOVERY(lqs);
1634         cfs_spin_unlock(&lqs->lqs_lock);
1635         cfs_up(&dquot->dq_sem);
1636
1637         /* release mds_qonoff_sem during obd_quotactl ops here */
1638         cfs_up_write(&mds->mds_qonoff_sem);
1639
1640         /* get real bhardlimit from all slaves. */
1641         qctl->qc_cmd = Q_GETOQUOTA;
1642         qctl->qc_type = type;
1643         qctl->qc_id = id;
1644         qctl->qc_stat = QUOTA_RECOVERING;
1645         rc = obd_quotactl(mds->mds_lov_exp, qctl);
1646         cfs_down_write(&mds->mds_qonoff_sem);
1647         if (rc)
1648                 GOTO(out, rc);
1649         total_limits = qctl->qc_dqblk.dqb_bhardlimit;
1650
1651         /* get real bhardlimit from master */
1652         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, qctl);
1653         if (rc)
1654                 GOTO(out, rc);
1655         total_limits += qctl->qc_dqblk.dqb_bhardlimit;
1656
1657         /* amend the usage of the administrative quotafile */
1658         cfs_down(&dquot->dq_sem);
1659
1660         dquot->dq_dqb.dqb_curspace = total_limits << QUOTABLOCK_BITS;
1661
1662         rc = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
1663         if (rc)
1664                 CERROR("write dquot failed! (rc:%d)\n", rc);
1665
1666         cfs_up(&dquot->dq_sem);
1667         EXIT;
1668 out:
1669         cfs_spin_lock(&lqs->lqs_lock);
1670         LQS_CLEAR_RECOVERY(lqs);
1671         cfs_spin_unlock(&lqs->lqs_lock);
1672 skip1:
1673         lqs_putref(lqs);
1674 skip:
1675         lustre_dqput(dquot);
1676         OBD_FREE_PTR(qctl);
1677         return rc;
1678 }
1679
1680 struct qmaster_recov_thread_data {
1681         struct obd_device *obd;
1682         cfs_completion_t comp;
1683 };
1684
1685 static int qmaster_recovery_main(void *arg)
1686 {
1687         struct qmaster_recov_thread_data *data = arg;
1688         struct obd_device *obd = data->obd;
1689         struct mds_obd *mds = &obd->u.mds;
1690         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
1691         int rc = 0;
1692         unsigned short type;
1693         ENTRY;
1694
1695         cfs_daemonize_ctxt("qmaster_recovd");
1696
1697         /* for mds */
1698         class_incref(obd, "qmaster_recovd_mds", obd);
1699         /* for lov */
1700         class_incref(mds->mds_lov_obd, "qmaster_recovd_lov", mds->mds_lov_obd);
1701
1702         cfs_complete(&data->comp);
1703
1704         cfs_down_write(&mds->mds_qonoff_sem);
1705         for (type = USRQUOTA; type < MAXQUOTAS; type++) {
1706                 cfs_list_t id_list;
1707                 struct dquot_id *dqid, *tmp;
1708
1709                 if (qinfo->qi_files[type] == NULL)
1710                         continue;
1711
1712                 CFS_INIT_LIST_HEAD(&id_list);
1713                 rc = fsfilt_qids(obd, qinfo->qi_files[type], NULL, type,
1714                                  &id_list);
1715                 if (rc)
1716                         CERROR("error get ids from admin quotafile.(%d)\n", rc);
1717
1718                 cfs_list_for_each_entry_safe(dqid, tmp, &id_list, di_link) {
1719                         cfs_list_del_init(&dqid->di_link);
1720                         if (rc)
1721                                 goto free;
1722
1723                         rc = dquot_recovery(obd, dqid->di_id, type);
1724                         if (rc)
1725                                 CERROR("%s: qmaster recovery failed for %sid %d"
1726                                        " rc:%d)\n", obd->obd_name,
1727                                        type ? "g" : "u", dqid->di_id, rc);
1728 free:
1729                         OBD_FREE_PTR(dqid);
1730                 }
1731         }
1732         cfs_up_write(&mds->mds_qonoff_sem);
1733         class_decref(mds->mds_lov_obd, "qmaster_recovd_lov", mds->mds_lov_obd);
1734         class_decref(obd, "qmaster_recovd_mds", obd);
1735         RETURN(rc);
1736 }
1737
1738 int mds_quota_recovery(struct obd_device *obd)
1739 {
1740         struct mds_obd *mds = &obd->u.mds;
1741         struct qmaster_recov_thread_data data;
1742         int rc = 0;
1743         ENTRY;
1744
1745         if (!ll_sb_any_quota_active(obd->u.obt.obt_qctxt.lqc_sb))
1746                 RETURN(0);
1747
1748         if (unlikely(!mds->mds_quota || obd->obd_stopping))
1749                 RETURN(rc);
1750
1751         cfs_mutex_down(&obd->obd_dev_sem);
1752         if (mds->mds_lov_desc.ld_active_tgt_count != mds->mds_lov_objid_count) {
1753                 CWARN("Only %u/%u OSTs are active, abort quota recovery\n",
1754                       mds->mds_lov_desc.ld_active_tgt_count,
1755                       mds->mds_lov_objid_count);
1756                 cfs_mutex_up(&obd->obd_dev_sem);
1757                 RETURN(rc);
1758         }
1759         cfs_mutex_up(&obd->obd_dev_sem);
1760
1761         data.obd = obd;
1762         cfs_init_completion(&data.comp);
1763
1764         rc = cfs_create_thread(qmaster_recovery_main, &data,
1765                                CFS_DAEMON_FLAGS);
1766         if (rc < 0)
1767                 CERROR("%s: cannot start quota recovery thread: rc %d\n",
1768                        obd->obd_name, rc);
1769
1770         cfs_wait_for_completion(&data.comp);
1771         RETURN(rc);
1772 }
1773
1774 #endif /* HAVE_QUOTA_SUPPORT */