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  *  lustre/quota/quota_master.c
   5  *  Lustre Quota Master request handler
   6  *
   7  *  Copyright (c) 2001-2005 Cluster File Systems, Inc.
   8  *   Author: Niu YaWei <niu@clusterfs.com>
   9  *
  10  *   This file is part of Lustre, http://www.lustre.org.
  11  *
  12  *   No redistribution or use is permitted outside of Cluster File Systems, Inc.
  13  *
  14  */
  15 #ifndef EXPORT_SYMTAB
  16 # define EXPORT_SYMTAB
  17 #endif
  18
  19 #define DEBUG_SUBSYSTEM S_MDS
  20
  21 #include <linux/version.h>
  22 #include <linux/fs.h>
  23 #include <asm/unistd.h>
  24 #include <linux/slab.h>
  25 #include <linux/quotaops.h>
  26 #include <linux/module.h>
  27 #include <linux/init.h>
  28 #include <linux/quota.h>
  29
  30 #include <obd_class.h>
  31 #include <lustre_quota.h>
  32 #include <lustre_fsfilt.h>
  33 #include <lustre_mds.h>
  34
  35 #include "quota_internal.h"
  36
  37 /* lock ordering:
  38  * mds->mds_qonoff_sem > dquot->dq_sem */
  39 static struct list_head lustre_dquot_hash[NR_DQHASH];
  40 static spinlock_t dquot_hash_lock = SPIN_LOCK_UNLOCKED;
  41
  42 kmem_cache_t *lustre_dquot_cachep;
  43
  44 int lustre_dquot_init(void)
  45 {
  46         int i;
  47         ENTRY;
  48
  49         LASSERT(lustre_dquot_cachep == NULL);
  50         lustre_dquot_cachep = kmem_cache_create("lustre_dquot_cache",
  51                                                 sizeof(struct lustre_dquot),
  52                                                 0, 0, NULL, NULL);
  53         if (!lustre_dquot_cachep)
  54                 return (-ENOMEM);
  55
  56         for (i = 0; i < NR_DQHASH; i++) {
  57                 INIT_LIST_HEAD(lustre_dquot_hash + i);
  58         }
  59         RETURN(0);
  60 }
  61
  62 void lustre_dquot_exit(void)
  63 {
  64         int i;
  65         ENTRY;
  66         /* FIXME cleanup work ?? */
  67
  68         for (i = 0; i < NR_DQHASH; i++) {
  69                 LASSERT(list_empty(lustre_dquot_hash + i));
  70         }
  71         if (lustre_dquot_cachep) {
  72 #ifdef HAVE_KMEM_CACHE_DESTROY_INT
  73                 int rc;
  74                 rc = kmem_cache_destroy(lustre_dquot_cachep);
  75                 LASSERTF(rc == 0,"couldn't destroy lustre_dquot_cachep slab\n");
  76 #else
  77                 kmem_cache_destroy(lustre_dquot_cachep);
  78 #endif
  79                 lustre_dquot_cachep = NULL;
  80         }
  81         EXIT;
  82 }
  83
  84 static inline int
  85 dquot_hashfn(struct lustre_quota_info *info, unsigned int id, int type)
  86              __attribute__((__const__));
  87
  88 static inline int
  89 dquot_hashfn(struct lustre_quota_info *info, unsigned int id, int type)
  90 {
  91         unsigned long tmp = ((unsigned long)info >> L1_CACHE_SHIFT) ^ id;
  92         tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
  93         return tmp;
  94 }
  95
  96 /* caller must hold dquot_hash_lock */
  97 static struct lustre_dquot *find_dquot(int hashent,
  98                                        struct lustre_quota_info *lqi, qid_t id,
  99                                        int type)
 100 {
 101         struct lustre_dquot *dquot;
 102         ENTRY;
 103
 104         LASSERT_SPIN_LOCKED(&dquot_hash_lock);
 105         list_for_each_entry(dquot, &lustre_dquot_hash[hashent], dq_hash) {
 106                 if (dquot->dq_info == lqi &&
 107                     dquot->dq_id == id && dquot->dq_type == type)
 108                         RETURN(dquot);
 109         }
 110         RETURN(NULL);
 111 }
 112
 113 static struct lustre_dquot *alloc_dquot(struct lustre_quota_info *lqi,
 114                                         qid_t id, int type)
 115 {
 116         struct lustre_dquot *dquot = NULL;
 117         ENTRY;
 118
 119         OBD_SLAB_ALLOC(dquot, lustre_dquot_cachep, SLAB_NOFS, sizeof(*dquot));
 120         if (dquot == NULL)
 121                 RETURN(NULL);
 122
 123         INIT_LIST_HEAD(&dquot->dq_hash);
 124         init_mutex_locked(&dquot->dq_sem);
 125         dquot->dq_refcnt = 1;
 126         dquot->dq_info = lqi;
 127         dquot->dq_id = id;
 128         dquot->dq_type = type;
 129         dquot->dq_status = DQ_STATUS_AVAIL;
 130
 131         RETURN(dquot);
 132 }
 133
 134 static void free_dquot(struct lustre_dquot *dquot)
 135 {
 136         OBD_SLAB_FREE(dquot, lustre_dquot_cachep, sizeof(*dquot));
 137 }
 138
 139 static void insert_dquot_nolock(struct lustre_dquot *dquot)
 140 {
 141         struct list_head *head = lustre_dquot_hash +
 142             dquot_hashfn(dquot->dq_info, dquot->dq_id, dquot->dq_type);
 143         LASSERT(list_empty(&dquot->dq_hash));
 144         list_add(&dquot->dq_hash, head);
 145 }
 146
 147 static void remove_dquot_nolock(struct lustre_dquot *dquot)
 148 {
 149         LASSERT(!list_empty(&dquot->dq_hash));
 150         list_del_init(&dquot->dq_hash);
 151 }
 152
 153 static void lustre_dqput(struct lustre_dquot *dquot)
 154 {
 155         ENTRY;
 156         spin_lock(&dquot_hash_lock);
 157         LASSERT(dquot->dq_refcnt);
 158         dquot->dq_refcnt--;
 159         if (!dquot->dq_refcnt) {
 160                 remove_dquot_nolock(dquot);
 161                 free_dquot(dquot);
 162         }
 163         spin_unlock(&dquot_hash_lock);
 164         EXIT;
 165 }
 166
 167 static struct lustre_dquot *lustre_dqget(struct obd_device *obd,
 168                                          struct lustre_quota_info *lqi,
 169                                          qid_t id, int type)
 170 {
 171         unsigned int hashent = dquot_hashfn(lqi, id, type);
 172         struct lustre_dquot *dquot, *empty;
 173         ENTRY;
 174
 175         if ((empty = alloc_dquot(lqi, id, type)) == NULL)
 176                 RETURN(ERR_PTR(-ENOMEM));
 177
 178         spin_lock(&dquot_hash_lock);
 179         if ((dquot = find_dquot(hashent, lqi, id, type)) != NULL) {
 180                 dquot->dq_refcnt++;
 181                 spin_unlock(&dquot_hash_lock);
 182                 free_dquot(empty);
 183         } else {
 184                 int rc;
 185
 186                 dquot = empty;
 187                 insert_dquot_nolock(dquot);
 188                 spin_unlock(&dquot_hash_lock);
 189
 190                 rc = fsfilt_dquot(obd, dquot, QFILE_RD_DQUOT);
 191                 up(&dquot->dq_sem);
 192                 if (rc) {
 193                         CERROR("can't read dquot from admin quotafile! "
 194                                "(rc:%d)\n", rc);
 195                         lustre_dqput(dquot);
 196                         RETURN(ERR_PTR(rc));
 197                 }
 198
 199         }
 200
 201         LASSERT(dquot);
 202         RETURN(dquot);
 203 }
 204
 205 int dqacq_handler(struct obd_device *obd, struct qunit_data *qdata, int opc)
 206 {
 207         struct mds_obd *mds = &obd->u.mds;
 208         struct lustre_quota_info *info = &mds->mds_quota_info;
 209         struct lustre_dquot *dquot = NULL;
 210         __u64 *usage = NULL;
 211         __u32 hlimit = 0, slimit = 0;
 212         __u32 qdata_type = qdata->qd_flags & QUOTA_IS_GRP;
 213         __u32 is_blk = (qdata->qd_flags & QUOTA_IS_BLOCK) >> 1;
 214         time_t *time = NULL;
 215         unsigned int grace = 0;
 216         int rc = 0;
 217         ENTRY;
 218
 219         OBD_FAIL_RETURN(OBD_FAIL_OBD_DQACQ, -EIO);
 220
 221         /* slaves never acquires qunit for user root */
 222         LASSERT(qdata->qd_id || qdata_type);
 223
 224         dquot = lustre_dqget(obd, info, qdata->qd_id, qdata_type);
 225         if (IS_ERR(dquot))
 226                 RETURN(PTR_ERR(dquot));
 227
 228         DQUOT_DEBUG(dquot, "get dquot in dqacq_handler\n");
 229         QINFO_DEBUG(dquot->dq_info, "get dquot in dqadq_handler\n");
 230
 231         down(&mds->mds_qonoff_sem);
 232         down(&dquot->dq_sem);
 233
 234         if (dquot->dq_status & DQ_STATUS_RECOVERY) {
 235                 DQUOT_DEBUG(dquot, "this dquot is under recovering.\n");
 236                 GOTO(out, rc = -EBUSY);
 237         }
 238
 239         if (is_blk) {
 240                 grace = info->qi_info[qdata_type].dqi_bgrace;
 241                 usage = &dquot->dq_dqb.dqb_curspace;
 242                 hlimit = dquot->dq_dqb.dqb_bhardlimit;
 243                 slimit = dquot->dq_dqb.dqb_bsoftlimit;
 244                 time = &dquot->dq_dqb.dqb_btime;
 245         } else {
 246                 grace = info->qi_info[qdata_type].dqi_igrace;
 247                 usage = (__u64 *) & dquot->dq_dqb.dqb_curinodes;
 248                 hlimit = dquot->dq_dqb.dqb_ihardlimit;
 249                 slimit = dquot->dq_dqb.dqb_isoftlimit;
 250                 time = &dquot->dq_dqb.dqb_itime;
 251         }
 252
 253         /* if the quota limit in admin quotafile is zero, we just inform
 254          * slave to clear quota limit with zero qd_count */
 255         if (hlimit == 0 && slimit == 0) {
 256                 qdata->qd_count = 0;
 257                 GOTO(out, rc);
 258         }
 259
 260         switch (opc) {
 261         case QUOTA_DQACQ:
 262                 if (hlimit &&
 263                     QUSG(*usage + qdata->qd_count, is_blk) > hlimit)
 264                         GOTO(out, rc = -EDQUOT);
 265
 266                 if (slimit &&
 267                     QUSG(*usage + qdata->qd_count, is_blk) > slimit) {
 268                         if (*time && cfs_time_current_sec() >= *time)
 269                                 GOTO(out, rc = -EDQUOT);
 270                         else if (!*time)
 271                                 *time = cfs_time_current_sec() + grace;
 272                 }
 273
 274                 *usage += qdata->qd_count;
 275                 break;
 276         case QUOTA_DQREL:
 277                 /* The usage in administrative file might be incorrect before
 278                  * recovery done */
 279                 if (*usage - qdata->qd_count < 0)
 280                         *usage = 0;
 281                 else
 282                         *usage -= qdata->qd_count;
 283
 284                 /* (usage <= soft limit) but not (usage < soft limit) */
 285                 if (!slimit || QUSG(*usage, is_blk) <= slimit)
 286                         *time = 0;
 287                 break;
 288         default:
 289                 LBUG();
 290         }
 291
 292         rc = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
 293         EXIT;
 294 out:
 295         up(&dquot->dq_sem);
 296         up(&mds->mds_qonoff_sem);
 297         lustre_dqput(dquot);
 298         return rc;
 299 }
 300
 301 int mds_quota_adjust(struct obd_device *obd, unsigned int qcids[],
 302                      unsigned int qpids[], int rc, int opc)
 303 {
 304         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
 305         int rc2 = 0;
 306         ENTRY;
 307
 308         if (rc && rc != -EDQUOT && rc != ENOLCK)
 309                 RETURN(0);
 310
 311         switch (opc) {
 312         case FSFILT_OP_RENAME:
 313                 /* acquire/release block quota on owner of original parent */
 314                 rc2 = qctxt_adjust_qunit(obd, qctxt, qpids[2], qpids[3], 1, 0);
 315                 /* fall-through */
 316         case FSFILT_OP_SETATTR:
 317                 /* acquire/release file quota on original owner */
 318                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids[0], qpids[1], 0, 0);
 319                 /* fall-through */
 320         case FSFILT_OP_CREATE:
 321         case FSFILT_OP_UNLINK:
 322                 /* acquire/release file/block quota on owner of child (or current owner) */
 323                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids[0], qcids[1], 0, 0);
 324                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qcids[0], qcids[1], 1, 0);
 325                 /* acquire/release block quota on owner of parent (or original owner) */
 326                 rc2 |= qctxt_adjust_qunit(obd, qctxt, qpids[0], qpids[1], 1, 0);
 327                 break;
 328         default:
 329                 LBUG();
 330                 break;
 331         }
 332
 333         if (rc2)
 334                 CERROR("mds adjust qunit failed! (opc:%d rc:%d)\n", opc, rc2);
 335         RETURN(0);
 336 }
 337
 338 int filter_quota_adjust(struct obd_device *obd, unsigned int qcids[],
 339                         unsigned int qpids[], int rc, int opc)
 340 {
 341         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
 342         int rc2 = 0;
 343         ENTRY;
 344
 345         if (rc && rc != -EDQUOT)
 346                 RETURN(0);
 347
 348         switch (opc) {
 349         case FSFILT_OP_SETATTR:
 350                 /* acquire/release block quota on original & current owner */
 351                 rc = qctxt_adjust_qunit(obd, qctxt, qcids[0], qcids[1], 1, 0);
 352                 rc2 = qctxt_adjust_qunit(obd, qctxt, qpids[0], qpids[1], 1, 0);
 353                 break;
 354         case FSFILT_OP_UNLINK:
 355                 /* release block quota on this owner */
 356         case FSFILT_OP_CREATE: /* XXX for write operation on obdfilter */
 357                 /* acquire block quota on this owner */
 358                 rc = qctxt_adjust_qunit(obd, qctxt, qcids[0], qcids[1], 1, 0);
 359                 break;
 360         default:
 361                 LBUG();
 362                 break;
 363         }
 364
 365         if (rc || rc2)
 366                 CERROR("filter adjust qunit failed! (opc:%d rc%d)\n",
 367                        opc, rc ?: rc2);
 368         RETURN(0);
 369 }
 370
 371 #define LUSTRE_ADMIN_QUOTAFILES {\
 372         "admin_quotafile.usr",  /* user admin quotafile */\
 373         "admin_quotafile.grp"   /* group admin quotafile */\
 374 }
 375 static const char prefix[] = "OBJECTS/";
 376
 377 int init_admin_quotafiles(struct obd_device *obd, struct obd_quotactl *oqctl)
 378 {
 379         struct mds_obd *mds = &obd->u.mds;
 380         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 381         const char *quotafiles[] = LUSTRE_ADMIN_QUOTAFILES;
 382         struct lvfs_run_ctxt saved;
 383         char name[64];
 384         int i, rc = 0;
 385         struct dentry *dparent = mds->mds_objects_dir;
 386         struct inode *iparent = dparent->d_inode;
 387         ENTRY;
 388
 389         LASSERT(iparent);
 390         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 391
 392         down(&mds->mds_qonoff_sem);
 393         for (i = 0; i < MAXQUOTAS; i++) {
 394                 struct dentry *de;
 395                 struct file *fp;
 396
 397                 if (!Q_TYPESET(oqctl, i))
 398                         continue;
 399
 400                 /* quota file has been opened ? */
 401                 if (qinfo->qi_files[i]) {
 402                         CWARN("init %s admin quotafile while quota on.\n",
 403                               i == USRQUOTA ? "user" : "group");
 404                         continue;
 405                 }
 406
 407                 /* lookup quota file */
 408                 rc = 0;
 409                 LOCK_INODE_MUTEX(iparent);
 410                 de = lookup_one_len(quotafiles[i], dparent,
 411                                     strlen(quotafiles[i]));
 412                 UNLOCK_INODE_MUTEX(iparent);
 413                 if (IS_ERR(de) || de->d_inode == NULL ||
 414                     !S_ISREG(de->d_inode->i_mode))
 415                         rc = IS_ERR(de) ? PTR_ERR(de) : -ENOENT;
 416                 if (!IS_ERR(de))
 417                         dput(de);
 418
 419                 if (rc && rc != -ENOENT) {
 420                         CERROR("error lookup quotafile %s! (rc:%d)\n",
 421                                name, rc);
 422                         break;
 423                 } else if (!rc) {
 424                         continue;
 425                 }
 426
 427                 LASSERT(strlen(quotafiles[i]) + sizeof(prefix) <= sizeof(name));
 428                 sprintf(name, "%s%s", prefix, quotafiles[i]);
 429
 430                 LASSERT(rc == -ENOENT);
 431                 /* create quota file */
 432                 fp = filp_open(name, O_CREAT | O_EXCL, 0644);
 433                 if (IS_ERR(fp) || !S_ISREG(fp->f_dentry->d_inode->i_mode)) {
 434                         rc = PTR_ERR(fp);
 435                         CERROR("error creating admin quotafile %s (rc:%d)\n",
 436                                name, rc);
 437                         break;
 438                 }
 439
 440                 qinfo->qi_files[i] = fp;
 441                 rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_INIT_INFO);
 442                 filp_close(fp, 0);
 443                 qinfo->qi_files[i] = NULL;
 444
 445                 if (rc) {
 446                         CERROR("error init %s admin quotafile! (rc:%d)\n",
 447                                i == USRQUOTA ? "user" : "group", rc);
 448                         break;
 449                 }
 450         }
 451         up(&mds->mds_qonoff_sem);
 452
 453         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 454         RETURN(rc);
 455 }
 456
 457 static int close_quota_files(struct obd_quotactl *oqctl,
 458                              struct lustre_quota_info *qinfo)
 459 {
 460         int i, rc = 0;
 461         ENTRY;
 462
 463         for (i = 0; i < MAXQUOTAS; i++) {
 464                 if (!Q_TYPESET(oqctl, i))
 465                         continue;
 466                 if (qinfo->qi_files[i] == NULL) {
 467                         rc = -ESRCH;
 468                         continue;
 469                 }
 470                 filp_close(qinfo->qi_files[i], 0);
 471                 qinfo->qi_files[i] = NULL;
 472         }
 473         RETURN(rc);
 474 }
 475
 476 int mds_admin_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
 477 {
 478         struct mds_obd *mds = &obd->u.mds;
 479         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 480         const char *quotafiles[] = LUSTRE_ADMIN_QUOTAFILES;
 481         char name[64];
 482         int i, rc = 0;
 483         struct inode *iparent = mds->mds_objects_dir->d_inode;
 484         ENTRY;
 485
 486         LASSERT(iparent);
 487
 488         /* open admin quota files and read quotafile info */
 489         for (i = 0; i < MAXQUOTAS; i++) {
 490                 struct file *fp;
 491
 492                 if (!Q_TYPESET(oqctl, i))
 493                         continue;
 494
 495                 LASSERT(strlen(quotafiles[i]) + sizeof(prefix) <= sizeof(name));
 496                 sprintf(name, "%s%s", prefix, quotafiles[i]);
 497
 498                 if (qinfo->qi_files[i] != NULL) {
 499                         rc = -EBUSY;
 500                         break;
 501                 }
 502
 503                 fp = filp_open(name, O_RDWR | O_EXCL, 0644);
 504                 if (IS_ERR(fp) || !S_ISREG(fp->f_dentry->d_inode->i_mode)) {
 505                         rc = PTR_ERR(fp);
 506                         CDEBUG(rc == -ENOENT ? D_QUOTA : D_ERROR,
 507                                "open %s failed! (rc:%d)\n", name, rc);
 508                         break;
 509                 }
 510                 qinfo->qi_files[i] = fp;
 511
 512                 rc = fsfilt_quotainfo(obd, qinfo, i, QFILE_RD_INFO);
 513                 if (rc) {
 514                         CERROR("error read quotainfo of %s! (rc:%d)\n",
 515                                name, rc);
 516                         break;
 517                 }
 518         }
 519
 520         if (rc && rc != -EBUSY)
 521                 close_quota_files(oqctl, qinfo);
 522
 523         RETURN(rc);
 524 }
 525
 526 static int mds_admin_quota_off(struct obd_device *obd,
 527                                struct obd_quotactl *oqctl)
 528 {
 529         struct mds_obd *mds = &obd->u.mds;
 530         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 531         int rc;
 532         ENTRY;
 533
 534         /* close admin quota files */
 535         rc = close_quota_files(oqctl, qinfo);
 536         RETURN(rc);
 537 }
 538
 539 int mds_quota_on(struct obd_device *obd, struct obd_quotactl *oqctl)
 540 {
 541         struct mds_obd *mds = &obd->u.mds;
 542         struct obd_device_target *obt = &obd->u.obt;
 543         struct lvfs_run_ctxt saved;
 544         int rc;
 545         ENTRY;
 546
 547         if (!atomic_dec_and_test(&obt->obt_quotachecking)) {
 548                 CDEBUG(D_INFO, "other people are doing quotacheck\n");
 549                 atomic_inc(&obt->obt_quotachecking);
 550                 RETURN(-EBUSY);
 551         }
 552
 553         down(&mds->mds_qonoff_sem);
 554         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 555         rc = mds_admin_quota_on(obd, oqctl);
 556         if (rc)
 557                 goto out;
 558
 559         rc = obd_quotactl(mds->mds_osc_exp, oqctl);
 560         if (rc)
 561                 goto out;
 562
 563         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 564         if (!rc)
 565                 obt->obt_qctxt.lqc_status = 1;
 566 out:
 567         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 568         up(&mds->mds_qonoff_sem);
 569         atomic_inc(&obt->obt_quotachecking);
 570         RETURN(rc);
 571 }
 572
 573 int mds_quota_off(struct obd_device *obd, struct obd_quotactl *oqctl)
 574 {
 575         struct mds_obd *mds = &obd->u.mds;
 576         struct obd_device_target *obt = &obd->u.obt;
 577         struct lvfs_run_ctxt saved;
 578         int rc, rc2;
 579         ENTRY;
 580
 581         if (!atomic_dec_and_test(&obt->obt_quotachecking)) {
 582                 CDEBUG(D_INFO, "other people are doing quotacheck\n");
 583                 atomic_inc(&obt->obt_quotachecking);
 584                 RETURN(-EBUSY);
 585         }
 586
 587         down(&mds->mds_qonoff_sem);
 588         /* close admin quota files */
 589         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 590         mds_admin_quota_off(obd, oqctl);
 591
 592         rc = obd_quotactl(mds->mds_osc_exp, oqctl);
 593         rc2 = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 594         if (!rc2)
 595                 obt->obt_qctxt.lqc_status = 0;
 596
 597         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 598         up(&mds->mds_qonoff_sem);
 599         atomic_inc(&obt->obt_quotachecking);
 600
 601         RETURN(rc ?: rc2);
 602 }
 603
 604 int mds_set_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
 605 {
 606         struct mds_obd *mds = &obd->u.mds;
 607         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 608         struct obd_dqinfo *dqinfo = &oqctl->qc_dqinfo;
 609         int rc;
 610         ENTRY;
 611
 612         down(&mds->mds_qonoff_sem);
 613         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
 614                 rc = -ESRCH;
 615                 goto out;
 616         }
 617
 618         qinfo->qi_info[oqctl->qc_type].dqi_bgrace = dqinfo->dqi_bgrace;
 619         qinfo->qi_info[oqctl->qc_type].dqi_igrace = dqinfo->dqi_igrace;
 620         qinfo->qi_info[oqctl->qc_type].dqi_flags = dqinfo->dqi_flags;
 621
 622         rc = fsfilt_quotainfo(obd, qinfo, oqctl->qc_type, QFILE_WR_INFO);
 623
 624 out:
 625         up(&mds->mds_qonoff_sem);
 626         RETURN(rc);
 627 }
 628
 629 int mds_get_dqinfo(struct obd_device *obd, struct obd_quotactl *oqctl)
 630 {
 631         struct mds_obd *mds = &obd->u.mds;
 632         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 633         struct obd_dqinfo *dqinfo = &oqctl->qc_dqinfo;
 634         int rc = 0;
 635         ENTRY;
 636
 637         down(&mds->mds_qonoff_sem);
 638         if (qinfo->qi_files[oqctl->qc_type] == NULL) {
 639                 rc = -ESRCH;
 640                 goto out;
 641         }
 642
 643         dqinfo->dqi_bgrace = qinfo->qi_info[oqctl->qc_type].dqi_bgrace;
 644         dqinfo->dqi_igrace = qinfo->qi_info[oqctl->qc_type].dqi_igrace;
 645         dqinfo->dqi_flags = qinfo->qi_info[oqctl->qc_type].dqi_flags;
 646
 647 out:
 648         up(&mds->mds_qonoff_sem);
 649         RETURN(rc);
 650 }
 651
 652 static int mds_init_slave_ilimits(struct obd_device *obd,
 653                                   struct obd_quotactl *oqctl, int set)
 654 {
 655         /* XXX: for file limits only adjust local now */
 656         unsigned int uid = 0, gid = 0;
 657         struct obd_quotactl *ioqc = NULL;
 658         int rc;
 659         ENTRY;
 660
 661         /* if we are going to set zero limit, needn't init slaves */
 662         if (!oqctl->qc_dqblk.dqb_ihardlimit && !oqctl->qc_dqblk.dqb_isoftlimit)
 663                 RETURN(0);
 664
 665         if (!set)
 666                 goto acquire;
 667
 668         OBD_ALLOC_PTR(ioqc);
 669         if (!ioqc)
 670                 RETURN(-ENOMEM);
 671
 672         ioqc->qc_cmd = Q_INITQUOTA;
 673         ioqc->qc_id = oqctl->qc_id;
 674         ioqc->qc_type = oqctl->qc_type;
 675         ioqc->qc_dqblk.dqb_valid = QIF_ILIMITS;
 676         ioqc->qc_dqblk.dqb_ihardlimit = MIN_QLIMIT;
 677
 678         /* set local limit to MIN_QLIMIT */
 679         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, ioqc);
 680         if (rc)
 681                 GOTO(out, rc);
 682 acquire:
 683         /* trigger local qunit pre-acquire */
 684         if (oqctl->qc_type == USRQUOTA)
 685                 uid = oqctl->qc_id;
 686         else
 687                 gid = oqctl->qc_id;
 688
 689         rc = qctxt_adjust_qunit(obd, &obd->u.obt.obt_qctxt, uid, gid, 0, 0);
 690         if (rc) {
 691                 CERROR("error mds adjust local file quota! (rc:%d)\n", rc);
 692                 GOTO(out, rc);
 693         }
 694         /* FIXME initialize all slaves in CMD */
 695         EXIT;
 696 out:
 697         if (ioqc)
 698                 OBD_FREE_PTR(ioqc);
 699         return rc;
 700 }
 701
 702 static int mds_init_slave_blimits(struct obd_device *obd,
 703                                   struct obd_quotactl *oqctl, int set)
 704 {
 705         struct mds_obd *mds = &obd->u.mds;
 706         struct obd_quotactl *ioqc;
 707         unsigned int uid = 0, gid = 0;
 708         int rc;
 709         ENTRY;
 710
 711         /* if we are going to set zero limit, needn't init slaves */
 712         if (!oqctl->qc_dqblk.dqb_bhardlimit && !oqctl->qc_dqblk.dqb_bsoftlimit)
 713                 RETURN(0);
 714
 715         OBD_ALLOC_PTR(ioqc);
 716         if (!ioqc)
 717                 RETURN(-ENOMEM);
 718
 719         ioqc->qc_cmd = Q_INITQUOTA;
 720         ioqc->qc_id = oqctl->qc_id;
 721         ioqc->qc_type = oqctl->qc_type;
 722         ioqc->qc_dqblk.dqb_valid = QIF_BLIMITS;
 723         ioqc->qc_dqblk.dqb_bhardlimit = set ? MIN_QLIMIT : 0;
 724
 725         /* set local limit to MIN_QLIMIT */
 726         if (set) {
 727                 rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, ioqc);
 728                 if (rc)
 729                         GOTO(out, rc);
 730         }
 731
 732         /* trigger local qunit pre-acquire */
 733         if (oqctl->qc_type == USRQUOTA)
 734                 uid = oqctl->qc_id;
 735         else
 736                 gid = oqctl->qc_id;
 737
 738         rc = qctxt_adjust_qunit(obd, &obd->u.obt.obt_qctxt, uid, gid, 1, 0);
 739         if (rc) {
 740                 CERROR("error mds adjust local block quota! (rc:%d)\n", rc);
 741                 GOTO(out, rc);
 742         }
 743
 744         /* initialize all slave's limit */
 745         rc = obd_quotactl(mds->mds_osc_exp, ioqc);
 746         EXIT;
 747 out:
 748         OBD_FREE_PTR(ioqc);
 749         return rc;
 750 }
 751
 752 int mds_set_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
 753 {
 754         struct mds_obd *mds = &obd->u.mds;
 755         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 756         __u32 ihardlimit, isoftlimit, bhardlimit, bsoftlimit;
 757         time_t btime, itime;
 758         struct lustre_dquot *dquot;
 759         struct obd_dqblk *dqblk = &oqctl->qc_dqblk;
 760         int set, rc;
 761         ENTRY;
 762
 763         down(&mds->mds_qonoff_sem);
 764         if (qinfo->qi_files[oqctl->qc_type] == NULL)
 765                 GOTO(out_sem, rc = -ESRCH);
 766
 767         dquot = lustre_dqget(obd, qinfo, oqctl->qc_id, oqctl->qc_type);
 768         if (IS_ERR(dquot))
 769                 GOTO(out_sem, rc = PTR_ERR(dquot));
 770         DQUOT_DEBUG(dquot, "get dquot in mds_set_blk\n");
 771         QINFO_DEBUG(dquot->dq_info, "get dquot in mds_set_blk\n");
 772
 773         down(&dquot->dq_sem);
 774
 775         if (dquot->dq_status) {
 776                 up(&dquot->dq_sem);
 777                 lustre_dqput(dquot);
 778                 GOTO(out_sem, rc = -EBUSY);
 779         }
 780         dquot->dq_status |= DQ_STATUS_SET;
 781
 782         ihardlimit = dquot->dq_dqb.dqb_ihardlimit;
 783         isoftlimit = dquot->dq_dqb.dqb_isoftlimit;
 784         bhardlimit = dquot->dq_dqb.dqb_bhardlimit;
 785         bsoftlimit = dquot->dq_dqb.dqb_bsoftlimit;
 786         btime = dquot->dq_dqb.dqb_btime;
 787         itime = dquot->dq_dqb.dqb_itime;
 788
 789         if (dqblk->dqb_valid & QIF_BTIME)
 790                 dquot->dq_dqb.dqb_btime = dqblk->dqb_btime;
 791         if (dqblk->dqb_valid & QIF_ITIME)
 792                 dquot->dq_dqb.dqb_itime = dqblk->dqb_itime;
 793
 794         if (dqblk->dqb_valid & QIF_BLIMITS) {
 795                 dquot->dq_dqb.dqb_bhardlimit = dqblk->dqb_bhardlimit;
 796                 dquot->dq_dqb.dqb_bsoftlimit = dqblk->dqb_bsoftlimit;
 797                 /* clear usage (limit pool) */
 798                 if (!dquot->dq_dqb.dqb_bhardlimit &&
 799                     !dquot->dq_dqb.dqb_bsoftlimit)
 800                         dquot->dq_dqb.dqb_curspace = 0;
 801
 802                 /* clear grace time */
 803                 if (!dqblk->dqb_bsoftlimit ||
 804                     toqb(dquot->dq_dqb.dqb_curspace) <= dqblk->dqb_bsoftlimit)
 805                         dquot->dq_dqb.dqb_btime = 0;
 806                 /* set grace only if user hasn't provided his own */
 807                 else if (!(dqblk->dqb_valid & QIF_BTIME))
 808                         dquot->dq_dqb.dqb_btime = cfs_time_current_sec() +
 809                                 qinfo->qi_info[dquot->dq_type].dqi_bgrace;
 810         }
 811
 812         if (dqblk->dqb_valid & QIF_ILIMITS) {
 813                 dquot->dq_dqb.dqb_ihardlimit = dqblk->dqb_ihardlimit;
 814                 dquot->dq_dqb.dqb_isoftlimit = dqblk->dqb_isoftlimit;
 815                 /* clear usage (limit pool) */
 816                 if (!dquot->dq_dqb.dqb_ihardlimit &&
 817                     !dquot->dq_dqb.dqb_isoftlimit)
 818                         dquot->dq_dqb.dqb_curinodes = 0;
 819
 820                 if (!dqblk->dqb_isoftlimit ||
 821                     dquot->dq_dqb.dqb_curinodes <= dqblk->dqb_isoftlimit)
 822                         dquot->dq_dqb.dqb_itime = 0;
 823                 else if (!(dqblk->dqb_valid & QIF_ITIME))
 824                         dquot->dq_dqb.dqb_itime = cfs_time_current_sec() +
 825                                 qinfo->qi_info[dquot->dq_type].dqi_igrace;
 826         }
 827
 828         rc = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
 829
 830         up(&dquot->dq_sem);
 831
 832         if (rc) {
 833                 CERROR("set limit failed! (rc:%d)\n", rc);
 834                 goto out;
 835         }
 836
 837         up(&mds->mds_qonoff_sem);
 838         if (dqblk->dqb_valid & QIF_ILIMITS) {
 839                 set = !(ihardlimit || isoftlimit);
 840                 rc = mds_init_slave_ilimits(obd, oqctl, set);
 841                 if (rc) {
 842                         CERROR("init slave ilimits failed! (rc:%d)\n", rc);
 843                         goto revoke_out;
 844                 }
 845         }
 846
 847         if (dqblk->dqb_valid & QIF_BLIMITS) {
 848                 set = !(bhardlimit || bsoftlimit);
 849                 rc = mds_init_slave_blimits(obd, oqctl, set);
 850                 if (rc) {
 851                         CERROR("init slave blimits failed! (rc:%d)\n", rc);
 852                         goto revoke_out;
 853                 }
 854         }
 855         down(&mds->mds_qonoff_sem);
 856
 857 revoke_out:
 858         if (rc) {
 859                 /* cancel previous setting */
 860                 down(&dquot->dq_sem);
 861                 dquot->dq_dqb.dqb_ihardlimit = ihardlimit;
 862                 dquot->dq_dqb.dqb_isoftlimit = isoftlimit;
 863                 dquot->dq_dqb.dqb_bhardlimit = bhardlimit;
 864                 dquot->dq_dqb.dqb_bsoftlimit = bsoftlimit;
 865                 dquot->dq_dqb.dqb_btime = btime;
 866                 dquot->dq_dqb.dqb_itime = itime;
 867                 fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
 868                 up(&dquot->dq_sem);
 869         }
 870 out:
 871         down(&dquot->dq_sem);
 872         dquot->dq_status &= ~DQ_STATUS_SET;
 873         up(&dquot->dq_sem);
 874         lustre_dqput(dquot);
 875         EXIT;
 876 out_sem:
 877         up(&mds->mds_qonoff_sem);
 878         return rc;
 879 }
 880
 881 static int mds_get_space(struct obd_device *obd, struct obd_quotactl *oqctl)
 882 {
 883         struct obd_quotactl *soqc;
 884         struct lvfs_run_ctxt saved;
 885         int rc;
 886         ENTRY;
 887
 888         OBD_ALLOC_PTR(soqc);
 889         if (!soqc)
 890                 RETURN(-ENOMEM);
 891
 892         soqc->qc_cmd = Q_GETOQUOTA;
 893         soqc->qc_id = oqctl->qc_id;
 894         soqc->qc_type = oqctl->qc_type;
 895
 896         rc = obd_quotactl(obd->u.mds.mds_osc_exp, soqc);
 897         if (rc)
 898                GOTO(out, rc);
 899
 900         oqctl->qc_dqblk.dqb_curspace = soqc->qc_dqblk.dqb_curspace;
 901
 902         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 903         soqc->qc_dqblk.dqb_curspace = 0;
 904         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, soqc);
 905         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 906
 907         if (rc)
 908                 GOTO(out, rc);
 909
 910         oqctl->qc_dqblk.dqb_curinodes += soqc->qc_dqblk.dqb_curinodes;
 911         oqctl->qc_dqblk.dqb_curspace += soqc->qc_dqblk.dqb_curspace;
 912         EXIT;
 913 out:
 914         OBD_FREE_PTR(soqc);
 915         return rc;
 916 }
 917
 918 int mds_get_dqblk(struct obd_device *obd, struct obd_quotactl *oqctl)
 919 {
 920         struct mds_obd *mds = &obd->u.mds;
 921         struct lustre_quota_info *qinfo = &mds->mds_quota_info;
 922         struct lustre_dquot *dquot;
 923         struct obd_dqblk *dqblk = &oqctl->qc_dqblk;
 924         int rc;
 925         ENTRY;
 926
 927         down(&mds->mds_qonoff_sem);
 928         if (qinfo->qi_files[oqctl->qc_type] == NULL)
 929                 GOTO(out, rc = -ESRCH);
 930
 931         dquot = lustre_dqget(obd, qinfo, oqctl->qc_id, oqctl->qc_type);
 932         if (IS_ERR(dquot))
 933                 GOTO(out, rc = PTR_ERR(dquot));
 934
 935         down(&dquot->dq_sem);
 936         dqblk->dqb_ihardlimit = dquot->dq_dqb.dqb_ihardlimit;
 937         dqblk->dqb_isoftlimit = dquot->dq_dqb.dqb_isoftlimit;
 938         dqblk->dqb_bhardlimit = dquot->dq_dqb.dqb_bhardlimit;
 939         dqblk->dqb_bsoftlimit = dquot->dq_dqb.dqb_bsoftlimit;
 940         dqblk->dqb_btime = dquot->dq_dqb.dqb_btime;
 941         dqblk->dqb_itime = dquot->dq_dqb.dqb_itime;
 942         up(&dquot->dq_sem);
 943
 944         lustre_dqput(dquot);
 945
 946         /* the usages in admin quota file is inaccurate */
 947         dqblk->dqb_curinodes = 0;
 948         dqblk->dqb_curspace = 0;
 949         rc = mds_get_space(obd, oqctl);
 950         EXIT;
 951 out:
 952         up(&mds->mds_qonoff_sem);
 953         return rc;
 954 }
 955
 956 int mds_get_obd_quota(struct obd_device *obd, struct obd_quotactl *oqctl)
 957 {
 958         struct lvfs_run_ctxt saved;
 959         int rc;
 960         ENTRY;
 961
 962         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 963         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 964         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 965
 966         RETURN(rc);
 967 }
 968
 969
 970 /* FIXME we only recovery block limit by now, need recovery inode
 971  * limits also after CMD involved in */
 972 static int
 973 dquot_recovery(struct obd_device *obd, unsigned int id, unsigned short type)
 974 {
 975         struct mds_obd *mds = &obd->u.mds;
 976         struct lustre_quota_info *qinfo= &obd->u.mds.mds_quota_info;
 977         struct lustre_dquot *dquot;
 978         struct obd_quotactl *qctl;
 979         __u64 total_limits = 0;
 980         int rc;
 981         ENTRY;
 982
 983         OBD_ALLOC_PTR(qctl);
 984         if (qctl == NULL)
 985                 RETURN(-ENOMEM);
 986
 987         dquot = lustre_dqget(obd, qinfo, id, type);
 988         if (IS_ERR(dquot)) {
 989                 CERROR("Get dquot failed. (rc:%ld)\n", PTR_ERR(dquot));
 990                 OBD_FREE_PTR(qctl);
 991                 RETURN(PTR_ERR(dquot));
 992         }
 993
 994         down(&dquot->dq_sem);
 995
 996         /* don't recovery the dquot without limits or under setting */
 997         if (!(dquot->dq_dqb.dqb_bhardlimit || dquot->dq_dqb.dqb_bsoftlimit) ||
 998             dquot->dq_status)
 999                 GOTO(skip, rc = 0);
1000         dquot->dq_status |= DQ_STATUS_RECOVERY;
1001
1002         up(&dquot->dq_sem);
1003
1004         /* get real bhardlimit from all slaves. */
1005         qctl->qc_cmd = Q_GETOQUOTA;
1006         qctl->qc_type = type;
1007         qctl->qc_id = id;
1008         qctl->qc_stat = QUOTA_RECOVERING;
1009         rc = obd_quotactl(obd->u.mds.mds_osc_exp, qctl);
1010         if (rc)
1011                 GOTO(out, rc);
1012         total_limits = qctl->qc_dqblk.dqb_bhardlimit;
1013
1014         /* get real bhardlimit from master */
1015         rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, qctl);
1016         if (rc)
1017                 GOTO(out, rc);
1018         total_limits += qctl->qc_dqblk.dqb_bhardlimit;
1019
1020         /* amend the usage of the administrative quotafile */
1021         down(&mds->mds_qonoff_sem);
1022         down(&dquot->dq_sem);
1023
1024         dquot->dq_dqb.dqb_curspace = total_limits << QUOTABLOCK_BITS;
1025
1026         rc = fsfilt_dquot(obd, dquot, QFILE_WR_DQUOT);
1027         if (rc)
1028                 CERROR("write dquot failed! (rc:%d)\n", rc);
1029
1030         up(&dquot->dq_sem);
1031         up(&mds->mds_qonoff_sem);
1032         EXIT;
1033 out:
1034         down(&dquot->dq_sem);
1035         dquot->dq_status &= ~DQ_STATUS_RECOVERY;
1036 skip:
1037         up(&dquot->dq_sem);
1038
1039         lustre_dqput(dquot);
1040         OBD_FREE_PTR(qctl);
1041         return rc;
1042 }
1043
1044 struct qmaster_recov_thread_data {
1045         struct obd_device *obd;
1046         struct completion comp;
1047 };
1048
1049 static int qmaster_recovery_main(void *arg)
1050 {
1051         struct qmaster_recov_thread_data *data = arg;
1052         struct obd_device *obd = data->obd;
1053         int rc = 0;
1054         unsigned short type;
1055         ENTRY;
1056
1057         ptlrpc_daemonize("qmaster_recovd");
1058
1059         complete(&data->comp);
1060
1061         for (type = USRQUOTA; type < MAXQUOTAS; type++) {
1062                 struct mds_obd *mds = &obd->u.mds;
1063                 struct lustre_quota_info *qinfo = &mds->mds_quota_info;
1064                 struct list_head id_list;
1065                 struct dquot_id *dqid, *tmp;
1066
1067                 down(&mds->mds_qonoff_sem);
1068                 if (qinfo->qi_files[type] == NULL) {
1069                         up(&mds->mds_qonoff_sem);
1070                         continue;
1071                 }
1072                 INIT_LIST_HEAD(&id_list);
1073                 rc = fsfilt_qids(obd, qinfo->qi_files[type], NULL, type,
1074                                  &id_list);
1075                 up(&mds->mds_qonoff_sem);
1076
1077                 if (rc)
1078                         CERROR("error get ids from admin quotafile.(%d)\n", rc);
1079
1080                 list_for_each_entry_safe(dqid, tmp, &id_list, di_link) {
1081                         list_del_init(&dqid->di_link);
1082                         if (rc)
1083                                 goto free;
1084
1085                         rc = dquot_recovery(obd, dqid->di_id, type);
1086                         if (rc)
1087                                 CERROR("qmaster recovery failed! (id:%d type:%d"
1088                                        " rc:%d)\n", dqid->di_id, type, rc);
1089 free:
1090                         kfree(dqid);
1091                 }
1092         }
1093         RETURN(rc);
1094 }
1095
1096 int mds_quota_recovery(struct obd_device *obd)
1097 {
1098         struct lov_obd *lov = &obd->u.mds.mds_osc_obd->u.lov;
1099         struct qmaster_recov_thread_data data;
1100         int rc = 0;
1101         ENTRY;
1102
1103         mutex_down(&lov->lov_lock);
1104         if (lov->desc.ld_tgt_count != lov->desc.ld_active_tgt_count) {
1105                 CWARN("Not all osts are active, abort quota recovery\n");
1106                 mutex_up(&lov->lov_lock);
1107                 RETURN(rc);
1108         }
1109         mutex_up(&lov->lov_lock);
1110
1111         data.obd = obd;
1112         init_completion(&data.comp);
1113
1114         rc = kernel_thread(qmaster_recovery_main, &data, CLONE_VM|CLONE_FILES);
1115         if (rc < 0)
1116                 CERROR("Cannot start quota recovery thread: rc %d\n", rc);
1117
1118         wait_for_completion(&data.comp);
1119         RETURN(rc);
1120 }