lustre/quota/quota_ctl.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36 #ifndef EXPORT_SYMTAB
  37 # define EXPORT_SYMTAB
  38 #endif
  39 #define DEBUG_SUBSYSTEM S_LQUOTA
  40
  41 #ifdef __KERNEL__
  42 # include <linux/version.h>
  43 # include <linux/module.h>
  44 # include <linux/init.h>
  45 # include <linux/fs.h>
  46 # include <linux/quota.h>
  47 #  include <linux/smp_lock.h>
  48 #  include <linux/buffer_head.h>
  49 #  include <linux/workqueue.h>
  50 #  include <linux/mount.h>
  51 #else /* __KERNEL__ */
  52 # include <liblustre.h>
  53 #endif
  54
  55 #include <obd_class.h>
  56 #include <lustre_mds.h>
  57 #include <lustre_dlm.h>
  58 #include <lustre_cfg.h>
  59 #include <obd_ost.h>
  60 #include <lustre_fsfilt.h>
  61 #include <lustre_quota.h>
  62 #include "quota_internal.h"
  63
  64 #ifdef HAVE_QUOTA_SUPPORT
  65 #ifdef __KERNEL__
  66
  67 /* When quotaon, build a lqs for every uid/gid who has been set limitation
  68  * for quota. After quota_search_lqs, it will hold one ref for the lqs.
  69  * It will be released when qctxt_cleanup() is executed b=18574 */
  70 void build_lqs(struct obd_device *obd)
  71 {
  72         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
  73         struct list_head id_list;
  74         int i, rc;
  75
  76         INIT_LIST_HEAD(&id_list);
  77         for (i = 0; i < MAXQUOTAS; i++) {
  78                 struct dquot_id *dqid, *tmp;
  79
  80                 if (sb_dqopt(qctxt->lqc_sb)->files[i] == NULL)
  81                         continue;
  82
  83 #ifndef KERNEL_SUPPORTS_QUOTA_READ
  84                 rc = fsfilt_qids(obd, sb_dqopt(qctxt->lqc_sb)->files[i], NULL,
  85                                  i, &id_list);
  86 #else
  87                 rc = fsfilt_qids(obd, NULL, sb_dqopt(qctxt->lqc_sb)->files[i],
  88                                  i, &id_list);
  89 #endif
  90                 if (rc) {
  91                         CERROR("%s: failed to get %s qids\n", obd->obd_name,
  92                                i ? "group" : "user");
  93                         continue;
  94                 }
  95
  96                 list_for_each_entry_safe(dqid, tmp, &id_list,
  97                                          di_link) {
  98                         struct lustre_qunit_size *lqs;
  99
 100                         list_del_init(&dqid->di_link);
 101                         lqs = quota_search_lqs(LQS_KEY(i, dqid->di_id),
 102                                                qctxt, 1);
 103                         if (lqs && !IS_ERR(lqs)) {
 104                                 lqs->lqs_flags |= dqid->di_flag;
 105                                 lqs_putref(lqs);
 106                         } else {
 107                                 CERROR("%s: failed to create a lqs for %sid %u"
 108                                        "\n", obd->obd_name, i ? "g" : "u",
 109                                        dqid->di_id);
 110                         }
 111
 112                         OBD_FREE_PTR(dqid);
 113                 }
 114         }
 115 }
 116
 117 int mds_quota_ctl(struct obd_export *exp, struct obd_quotactl *oqctl)
 118 {
 119         struct obd_device *obd = exp->exp_obd;
 120         struct obd_device_target *obt = &obd->u.obt;
 121         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
 122         struct timeval work_start;
 123         struct timeval work_end;
 124         long timediff;
 125         int rc = 0;
 126         ENTRY;
 127
 128         do_gettimeofday(&work_start);
 129         switch (oqctl->qc_cmd) {
 130         case Q_QUOTAON:
 131                 rc = mds_quota_on(obd, oqctl);
 132                 /* when quotaon, create lqs for every quota uid/gid b=18574 */
 133                 build_lqs(obd);
 134                 break;
 135         case Q_QUOTAOFF:
 136                 oqctl->qc_id = obt->obt_qfmt; /* override qfmt version */
 137                 mds_quota_off(obd, oqctl);
 138                 break;
 139         case Q_SETINFO:
 140                 rc = mds_set_dqinfo(obd, oqctl);
 141                 break;
 142         case Q_GETINFO:
 143                 rc = mds_get_dqinfo(obd, oqctl);
 144                 break;
 145         case Q_SETQUOTA:
 146                 rc = mds_set_dqblk(obd, oqctl);
 147                 break;
 148         case Q_GETQUOTA:
 149                 rc = mds_get_dqblk(obd, oqctl);
 150                 break;
 151         case Q_GETOINFO:
 152         case Q_GETOQUOTA:
 153                 rc = mds_get_obd_quota(obd, oqctl);
 154                 break;
 155         case LUSTRE_Q_INVALIDATE:
 156                 rc = mds_quota_invalidate(obd, oqctl);
 157                 break;
 158         case LUSTRE_Q_FINVALIDATE:
 159                 oqctl->qc_id = obt->obt_qfmt; /* override qfmt version */
 160                 rc = mds_quota_finvalidate(obd, oqctl);
 161                 break;
 162         default:
 163                 CERROR("%s: unsupported mds_quotactl command: %d\n",
 164                        obd->obd_name, oqctl->qc_cmd);
 165                 RETURN(-EFAULT);
 166         }
 167
 168         if (rc)
 169                 CDEBUG(D_INFO, "mds_quotactl admin quota command %d, id %u, "
 170                                "type %d, failed: rc = %d\n",
 171                        oqctl->qc_cmd, oqctl->qc_id, oqctl->qc_type, rc);
 172         do_gettimeofday(&work_end);
 173         timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
 174         lprocfs_counter_add(qctxt->lqc_stats, LQUOTA_QUOTA_CTL, timediff);
 175
 176         RETURN(rc);
 177 }
 178
 179 int filter_quota_ctl(struct obd_export *exp, struct obd_quotactl *oqctl)
 180 {
 181         struct obd_device *obd = exp->exp_obd;
 182         struct obd_device_target *obt = &obd->u.obt;
 183         struct lvfs_run_ctxt saved;
 184         struct lustre_quota_ctxt *qctxt = &obd->u.obt.obt_qctxt;
 185         struct lustre_qunit_size *lqs;
 186         void *handle = NULL;
 187         struct timeval work_start;
 188         struct timeval work_end;
 189         long timediff;
 190         int rc = 0;
 191         ENTRY;
 192
 193         do_gettimeofday(&work_start);
 194         switch (oqctl->qc_cmd) {
 195         case Q_QUOTAON:
 196                 oqctl->qc_id = obt->obt_qfmt;
 197                 rc = generic_quota_on(obd, oqctl, 0);
 198                 /* when quotaon, create lqs for every quota uid/gid b=18574 */
 199                 build_lqs(obd);
 200                 break;
 201         case Q_FINVALIDATE:
 202         case Q_QUOTAOFF:
 203                 if (!atomic_dec_and_test(&obt->obt_quotachecking)) {
 204                         CDEBUG(D_INFO, "other people are doing quotacheck\n");
 205                         atomic_inc(&obt->obt_quotachecking);
 206                         rc = -EBUSY;
 207                         break;
 208                 }
 209                 if (oqctl->qc_cmd == Q_FINVALIDATE &&
 210                     (obt->obt_qctxt.lqc_flags & UGQUOTA2LQC(oqctl->qc_type))) {
 211                         atomic_inc(&obt->obt_quotachecking);
 212                         rc = -EBUSY;
 213                         break;
 214                 }
 215                 oqctl->qc_id = obt->obt_qfmt; /* override qfmt version */
 216         case Q_GETOINFO:
 217         case Q_GETOQUOTA:
 218         case Q_GETQUOTA:
 219                 /* In recovery scenario, this pending dqacq/dqrel might have
 220                  * been processed by master successfully before it's dquot
 221                  * on master enter recovery mode. We must wait for this
 222                  * dqacq/dqrel done then return the correct limits to master */
 223                 if (oqctl->qc_stat == QUOTA_RECOVERING)
 224                         handle = quota_barrier(&obd->u.obt.obt_qctxt, oqctl, 1);
 225
 226                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 227                 rc = fsfilt_quotactl(obd, obt->obt_sb, oqctl);
 228                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 229
 230                 if (oqctl->qc_stat == QUOTA_RECOVERING)
 231                         quota_unbarrier(handle);
 232
 233                 if (oqctl->qc_cmd == Q_QUOTAOFF ||
 234                     oqctl->qc_cmd == Q_FINVALIDATE) {
 235                         if (!rc && oqctl->qc_cmd == Q_QUOTAOFF)
 236                                 obt->obt_qctxt.lqc_flags &= ~UGQUOTA2LQC(oqctl->qc_type);
 237                         atomic_inc(&obt->obt_quotachecking);
 238                 }
 239                 break;
 240         case Q_SETQUOTA:
 241                 /* currently, it is only used for nullifying the quota */
 242                 handle = quota_barrier(&obd->u.obt.obt_qctxt, oqctl, 1);
 243
 244                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 245                 rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 246
 247                 if (!rc) {
 248                         oqctl->qc_cmd = Q_SYNC;
 249                         fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 250                         oqctl->qc_cmd = Q_SETQUOTA;
 251                 }
 252                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 253                 quota_unbarrier(handle);
 254
 255                 lqs = quota_search_lqs(LQS_KEY(oqctl->qc_type, oqctl->qc_id),
 256                                        qctxt, 0);
 257                 if (lqs == NULL || IS_ERR(lqs)){
 258                         CERROR("fail to create lqs during setquota operation "
 259                                "for %sid %u\n", oqctl->qc_type ? "g" : "u",
 260                                oqctl->qc_id);
 261                 } else {
 262                         lqs->lqs_flags &= ~QB_SET;
 263                         lqs_putref(lqs);
 264                 }
 265
 266                 break;
 267         case Q_INITQUOTA:
 268                 {
 269                 unsigned int uid = 0, gid = 0;
 270
 271                 /* Initialize quota limit to MIN_QLIMIT */
 272                 LASSERT(oqctl->qc_dqblk.dqb_valid == QIF_BLIMITS);
 273                 LASSERT(oqctl->qc_dqblk.dqb_bsoftlimit == 0);
 274
 275                 if (!oqctl->qc_dqblk.dqb_bhardlimit)
 276                         goto adjust;
 277
 278                /* There might be a pending dqacq/dqrel (which is going to
 279                  * clear stale limits on slave). we should wait for it's
 280                  * completion then initialize limits */
 281                 handle = quota_barrier(&obd->u.obt.obt_qctxt, oqctl, 1);
 282                 LASSERT(oqctl->qc_dqblk.dqb_bhardlimit == MIN_QLIMIT);
 283                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 284                 rc = fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 285
 286                 /* Update on-disk quota, in case of lose the changed limits
 287                  * (MIN_QLIMIT) on crash, which cannot be recovered.*/
 288                 if (!rc) {
 289                         oqctl->qc_cmd = Q_SYNC;
 290                         fsfilt_quotactl(obd, obd->u.obt.obt_sb, oqctl);
 291                         oqctl->qc_cmd = Q_INITQUOTA;
 292                 }
 293                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 294                 quota_unbarrier(handle);
 295
 296                 if (rc)
 297                         RETURN(rc);
 298 adjust:
 299                 lqs = quota_search_lqs(LQS_KEY(oqctl->qc_type, oqctl->qc_id),
 300                                        qctxt, 1);
 301                 if (lqs == NULL || IS_ERR(lqs)){
 302                         CERROR("fail to create lqs during setquota operation "
 303                                "for %sid %u\n", oqctl->qc_type ? "g" : "u",
 304                                oqctl->qc_id);
 305                         break;
 306                 } else {
 307                         lqs->lqs_flags |= QB_SET;
 308                         if (OBD_FAIL_CHECK(OBD_FAIL_QUOTA_WITHOUT_CHANGE_QS)) {
 309                                 lqs->lqs_bunit_sz = qctxt->lqc_bunit_sz;
 310                                 lqs->lqs_btune_sz = qctxt->lqc_btune_sz;
 311                                 lqs->lqs_iunit_sz = qctxt->lqc_iunit_sz;
 312                                 lqs->lqs_itune_sz = qctxt->lqc_itune_sz;
 313                         }
 314                         lqs_putref(lqs);
 315                 }
 316
 317                 /* Trigger qunit pre-acquire */
 318                 if (oqctl->qc_type == USRQUOTA)
 319                         uid = oqctl->qc_id;
 320                 else
 321                         gid = oqctl->qc_id;
 322
 323                 rc = qctxt_adjust_qunit(obd, &obd->u.obt.obt_qctxt,
 324                                         uid, gid, 1, 0, NULL);
 325                 if (rc == -EDQUOT || rc == -EBUSY) {
 326                         CDEBUG(D_QUOTA, "rc: %d.\n", rc);
 327                         rc = 0;
 328                 }
 329
 330                 break;
 331                 }
 332         default:
 333                 CERROR("%s: unsupported filter_quotactl command: %d\n",
 334                        obd->obd_name, oqctl->qc_cmd);
 335                 RETURN(-EFAULT);
 336         }
 337         do_gettimeofday(&work_end);
 338         timediff = cfs_timeval_sub(&work_end, &work_start, NULL);
 339         lprocfs_counter_add(qctxt->lqc_stats, LQUOTA_QUOTA_CTL, timediff);
 340
 341         RETURN(rc);
 342 }
 343 #endif /* __KERNEL__ */
 344 #endif
 345
 346 int client_quota_ctl(struct obd_export *exp, struct obd_quotactl *oqctl)
 347 {
 348         struct ptlrpc_request *req;
 349         struct obd_quotactl *oqc;
 350         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*oqctl) };
 351         int ver, opc, rc, resends = 0;
 352         ENTRY;
 353
 354         if (!strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_MDC_NAME)) {
 355                 ver = LUSTRE_MDS_VERSION,
 356                 opc = MDS_QUOTACTL;
 357         } else if (!strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_OSC_NAME)) {
 358                 ver = LUSTRE_OST_VERSION,
 359                 opc = OST_QUOTACTL;
 360         } else {
 361                 RETURN(-EINVAL);
 362         }
 363
 364 restart_request:
 365
 366         req = ptlrpc_prep_req(class_exp2cliimp(exp), ver, opc, 2, size, NULL);
 367         if (!req)
 368                 GOTO(out, rc = -ENOMEM);
 369
 370         oqc = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, sizeof(*oqctl));
 371         *oqc = *oqctl;
 372
 373         ptlrpc_req_set_repsize(req, 2, size);
 374         ptlrpc_at_set_req_timeout(req);
 375         req->rq_no_resend = 1;
 376
 377         rc = ptlrpc_queue_wait(req);
 378         if (rc) {
 379                 CERROR("ptlrpc_queue_wait failed, rc: %d\n", rc);
 380                 GOTO(out, rc);
 381         }
 382
 383         oqc = NULL;
 384         if (req->rq_repmsg)
 385                 oqc = lustre_swab_repbuf(req, REPLY_REC_OFF, sizeof(*oqc),
 386                                          lustre_swab_obd_quotactl);
 387         if (oqc == NULL) {
 388                 CERROR ("Can't unpack obd_quotactl\n");
 389                 GOTO(out, rc = -EPROTO);
 390         }
 391
 392         *oqctl = *oqc;
 393         EXIT;
 394 out:
 395         ptlrpc_req_finished(req);
 396
 397         if (client_quota_recoverable_error(rc)) {
 398                 resends++;
 399                 if (!client_quota_should_resend(resends, &exp->exp_obd->u.cli)) {
 400                         CERROR("too many resend retries, returning error "
 401                                "(cmd = %d, id = %u, type = %d)\n",
 402                                oqctl->qc_cmd, oqctl->qc_id, oqctl->qc_type);
 403                         RETURN(-EIO);
 404                 }
 405
 406                 goto restart_request;
 407         }
 408
 409         return rc;
 410 }
 411
 412 int lov_quota_ctl(struct obd_export *exp, struct obd_quotactl *oqctl)
 413 {
 414         struct obd_device *obd = class_exp2obd(exp);
 415         struct lov_obd *lov = &obd->u.lov;
 416         __u64 curspace = 0;
 417         __u64 bhardlimit = 0;
 418         int i, rc = 0;
 419         ENTRY;
 420
 421         if (oqctl->qc_cmd != LUSTRE_Q_QUOTAON &&
 422             oqctl->qc_cmd != LUSTRE_Q_QUOTAOFF &&
 423             oqctl->qc_cmd != Q_GETOQUOTA &&
 424             oqctl->qc_cmd != Q_INITQUOTA &&
 425             oqctl->qc_cmd != LUSTRE_Q_SETQUOTA &&
 426             oqctl->qc_cmd != Q_FINVALIDATE) {
 427                 CERROR("bad quota opc %x for lov obd", oqctl->qc_cmd);
 428                 RETURN(-EFAULT);
 429         }
 430
 431         obd_getref(obd);
 432         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 433                 int err;
 434
 435                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active) {
 436                         if (oqctl->qc_cmd == Q_GETOQUOTA) {
 437                                 CERROR("ost %d is inactive\n", i);
 438                                 rc = -EIO;
 439                         } else {
 440                                 CDEBUG(D_HA, "ost %d is inactive\n", i);
 441                         }
 442                         continue;
 443                 }
 444
 445                 err = obd_quotactl(lov->lov_tgts[i]->ltd_exp, oqctl);
 446                 if (err) {
 447                         if (lov->lov_tgts[i]->ltd_active && !rc)
 448                                 rc = err;
 449                         continue;
 450                 }
 451
 452                 if (oqctl->qc_cmd == Q_GETOQUOTA) {
 453                         curspace += oqctl->qc_dqblk.dqb_curspace;
 454                         bhardlimit += oqctl->qc_dqblk.dqb_bhardlimit;
 455                 }
 456         }
 457         obd_putref(obd);
 458
 459         if (oqctl->qc_cmd == Q_GETOQUOTA) {
 460                 oqctl->qc_dqblk.dqb_curspace = curspace;
 461                 oqctl->qc_dqblk.dqb_bhardlimit = bhardlimit;
 462         }
 463         RETURN(rc);
 464 }