lustre/quota/qsd_entry.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2012, 2017, Intel Corporation.
  25  * Use is subject to license terms.
  26  *
  27  * Author: Johann Lombardi <johann.lombardi@intel.com>
  28  * Author: Niu    Yawei    <yawei.niu@intel.com>
  29  */
  30
  31 #define DEBUG_SUBSYSTEM S_LQUOTA
  32
  33 #include "qsd_internal.h"
  34
  35 /*
  36  * Initialize qsd-specific fields of quota entry.
  37  *
  38  * \param lqe - is the quota entry to initialize
  39  * \param arg - is the pointer to the qsd_qtype_info structure
  40  */
  41 static void qsd_lqe_init(struct lquota_entry *lqe, void *arg)
  42 {
  43         LASSERT(!lqe_is_master(lqe));
  44
  45         /* initialize slave parameters */
  46         rwlock_init(&lqe->lqe_lock);
  47         memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh));
  48         lqe->lqe_pending_write = 0;
  49         lqe->lqe_pending_req   = 0;
  50         init_waitqueue_head(&lqe->lqe_waiters);
  51         lqe->lqe_usage    = 0;
  52         lqe->lqe_nopreacq = false;
  53         mutex_init(&lqe->lqe_glbl_data_lock);
  54 }
  55
  56 /*
  57  * Update a slave quota entry. This is done by reading enforcement status from
  58  * the copy of the global index and the amount the slave currenly owns
  59  * for this user from the slave index copy.
  60  *
  61  * \param env - the environment passed by the caller
  62  * \param lqe - is the quota entry to refresh
  63  * \param arg - is the pointer to the qsd_qtype_info structure
  64  * \param need_crt - needed to be compat with qmt_lqe_read
  65  */
  66 static int qsd_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
  67                         void *arg, bool need_crt)
  68 {
  69         struct qsd_thread_info *qti = qsd_info(env);
  70         struct qsd_qtype_info  *qqi = (struct qsd_qtype_info *)arg;
  71         int                     rc;
  72
  73         LASSERT(!lqe_is_master(lqe));
  74
  75         /* read record from global index copy to know whether quota is
  76          * enforced for this user */
  77         rc = lquota_disk_read(env, qqi->qqi_glb_obj, &lqe->lqe_id,
  78                               (struct dt_rec *)&qti->qti_glb_rec);
  79
  80         switch(rc) {
  81         case -ENOENT:
  82                 /* no such entry, assume quota isn't enforced for this user */
  83                 lqe->lqe_enforced = false;
  84                 break;
  85         case 0:
  86                 if (lqe->lqe_id.qid_uid == 0) {
  87                         qqi->qqi_default_hardlimit =
  88                                                 qti->qti_glb_rec.qbr_hardlimit;
  89                         qqi->qqi_default_softlimit =
  90                                                 qti->qti_glb_rec.qbr_softlimit;
  91                         qqi->qqi_default_gracetime =
  92                                                 qti->qti_glb_rec.qbr_granted;
  93                 }
  94
  95                 if (lqe->lqe_id.qid_uid != 0 &&
  96                     (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) &
  97                                                 LQUOTA_FLAG_RESET))
  98                         lqe->lqe_is_reset = true;
  99
 100                 if (lqe->lqe_id.qid_uid != 0 &&
 101                     (qti->qti_glb_rec.qbr_hardlimit != 0 ||
 102                      qti->qti_glb_rec.qbr_softlimit != 0))
 103                         lqe->lqe_enforced = true;
 104                 else
 105                         lqe->lqe_enforced = false;
 106                 break;
 107         default:
 108                 LQUOTA_ERROR(lqe, "failed to read quota entry from global "
 109                              "index copy, rc:%d", rc);
 110                 return rc;
 111         }
 112
 113         if (lqe->lqe_id.qid_uid != 0 &&
 114             (rc == -ENOENT ||
 115              (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) & LQUOTA_FLAG_DEFAULT &&
 116               qti->qti_glb_rec.qbr_hardlimit == 0 &&
 117               qti->qti_glb_rec.qbr_softlimit == 0))) {
 118                 struct lquota_entry *lqe_def;
 119                 union lquota_id qid = { {0} };
 120
 121                 /* ensure the lqe storing the default quota setting loaded */
 122                 lqe_def = lqe_locate(env, qqi->qqi_site, &qid);
 123
 124                 lqe->lqe_is_default = true;
 125
 126                 if (qqi->qqi_default_hardlimit != 0 ||
 127                     qqi->qqi_default_softlimit != 0) {
 128                         LQUOTA_DEBUG(lqe, "enforced by default quota");
 129                         lqe->lqe_enforced = true;
 130                 }
 131
 132                 if (!IS_ERR(lqe_def))
 133                         lqe_putref(lqe_def);
 134         }
 135
 136         /* read record from slave index copy to find out how much space is
 137          * currently owned by this slave */
 138         rc = lquota_disk_read(env, qqi->qqi_slv_obj, &lqe->lqe_id,
 139                               (struct dt_rec *)&qti->qti_slv_rec);
 140         switch(rc) {
 141         case -ENOENT:
 142                 lqe->lqe_granted = 0;
 143                 break;
 144         case 0:
 145                 lqe->lqe_granted = qti->qti_slv_rec.qsr_granted;
 146                 break;
 147         default:
 148                 LQUOTA_ERROR(lqe, "failed to read quota entry from slave "
 149                              "index copy, rc:%d", rc);
 150                 return rc;
 151         }
 152
 153         /* don't know what the qunit value is yet */
 154         qsd_set_qunit(lqe, 0);
 155
 156         /* read current disk-usage from disk */
 157         rc = qsd_refresh_usage(env, lqe);
 158         if (rc)
 159                 return rc;
 160
 161         LQUOTA_DEBUG(lqe, "successfully read from disk");
 162         return 0;
 163 }
 164
 165 /*
 166  * Print lqe information for debugging.
 167  *
 168  * \param lqe - is the quota entry to debug
 169  * \param arg - is the pointer to the qsd_qtype_info structure
 170  * \param msgdata - debug message
 171  * \param fmt     - format of debug message
 172  */
 173 static void qsd_lqe_debug(struct lquota_entry *lqe, void *arg,
 174                           struct libcfs_debug_msg_data *msgdata,
 175                           struct va_format *vaf)
 176 {
 177         struct qsd_qtype_info   *qqi = (struct qsd_qtype_info *)arg;
 178
 179         libcfs_debug_msg(msgdata,
 180                          "%pV qsd:%s qtype:%s id:%llu enforced:%d granted: %llu pending:%llu waiting:%llu req:%d usage: %llu qunit:%llu qtune:%llu edquot:%d default:%s revoke:%d\n",
 181                          vaf,
 182                          qqi->qqi_qsd->qsd_svname, qtype_name(qqi->qqi_qtype),
 183                          lqe->lqe_id.qid_uid, lqe->lqe_enforced,
 184                          lqe->lqe_granted, lqe->lqe_pending_write,
 185                          lqe->lqe_waiting_write, lqe->lqe_pending_req,
 186                          lqe->lqe_usage, lqe->lqe_qunit, lqe->lqe_qtune,
 187                          lqe->lqe_edquot, lqe->lqe_is_default ? "yes" : "no",
 188                          lqe->lqe_revoke);
 189 }
 190
 191 /*
 192  * Vector of quota entry operations supported on the slave
 193  */
 194 const struct lquota_entry_operations qsd_lqe_ops = {
 195         .lqe_init               = qsd_lqe_init,
 196         .lqe_read               = qsd_lqe_read,
 197         .lqe_debug              = qsd_lqe_debug,
 198 };
 199
 200 int qsd_write_version(const struct lu_env *env, struct qsd_qtype_info *qqi,
 201                       __u64 ver, bool global)
 202 {
 203         struct qsd_instance *qsd = qqi->qqi_qsd;
 204         struct dt_object    *obj = global ? qqi->qqi_glb_obj :
 205                                             qqi->qqi_slv_obj;
 206         int                  rc;
 207         ENTRY;
 208
 209         rc = lquota_disk_update_ver(env, qsd->qsd_dev, obj, ver);
 210         if (rc)
 211                 RETURN(rc);
 212
 213         qsd_bump_version(qqi, ver, global);
 214         RETURN(0);
 215 }
 216
 217 /*
 218  * Consult current disk space consumed by a given identifier.
 219  *
 220  * \param env   - the environment passed by the caller
 221  * \param qqi   - is the pointer to the qsd_qtype_info structure associated
 222  *                with the identifier.
 223  * \param lqe   - is the quota entry associated with the identifier
 224  */
 225 int qsd_refresh_usage(const struct lu_env *env, struct lquota_entry *lqe)
 226 {
 227         struct qsd_thread_info  *qti = qsd_info(env);
 228         struct lquota_acct_rec  *rec = &qti->qti_acct_rec;
 229         struct qsd_qtype_info   *qqi = lqe2qqi(lqe);
 230         int                      rc = 0;
 231         ENTRY;
 232
 233         LASSERT(qqi->qqi_acct_obj);
 234
 235         /* read disk usage */
 236         rc = lquota_disk_read(env, qqi->qqi_acct_obj, &lqe->lqe_id,
 237                               (struct dt_rec *)rec);
 238         switch(rc) {
 239         case -ENOENT:
 240                 lqe->lqe_usage = 0;
 241                 rc = 0;
 242                 break;
 243         case 0:
 244                 if (qqi->qqi_qsd->qsd_is_md)
 245                         lqe->lqe_usage = rec->ispace;
 246                 else
 247                         lqe->lqe_usage = toqb(rec->bspace);
 248                 break;
 249         default:
 250                 LQUOTA_ERROR(lqe, "failed to read disk usage, rc:%d", rc);
 251                 RETURN(rc);
 252         }
 253
 254         LQUOTA_DEBUG(lqe, "disk usage: %llu", lqe->lqe_usage);
 255         RETURN(0);
 256 }
 257
 258 /*
 259  * Update slave or global index copy.
 260  *
 261  * \param env    - the environment passed by the caller
 262  * \param qqi    - is the qsd_type_info structure managing the index to be
 263  *                 update
 264  * \param qid    - is the identifier for which we need to update the quota
 265  *                 settings
 266  * \param global - is set to true when updating the global index copy and to
 267  *                 false for the slave index copy.
 268  * \param ver    - is the new version of the index. If equal to 0, the version
 269  *                 of the index isn't changed
 270  * \param rec    - is the updated record to insert in the index file
 271  */
 272 int qsd_update_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
 273                      union lquota_id *qid, bool global, __u64 ver, void *rec)
 274 {
 275         struct thandle          *th = NULL;
 276         struct dt_object        *obj;
 277         __u64                   *new_verp = NULL;
 278         int                      flags = 0;
 279         int                      rc;
 280         ENTRY;
 281
 282         obj = global ? qqi->qqi_glb_obj : qqi->qqi_slv_obj;
 283
 284         /* allocate transaction */
 285         th = dt_trans_create(env, qqi->qqi_qsd->qsd_dev);
 286         if (IS_ERR(th))
 287                 RETURN(PTR_ERR(th));
 288
 289         /* reserve enough credits to update record in index file */
 290         rc = lquota_disk_declare_write(env, th, obj, qid);
 291         if (rc)
 292                 GOTO(out, rc);
 293
 294         /* start local transaction */
 295         rc = dt_trans_start_local(env, qqi->qqi_qsd->qsd_dev, th);
 296         if (rc)
 297                 GOTO(out, rc);
 298
 299         if (global) {
 300                 /* Update record in global index copy */
 301                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
 302
 303                 CDEBUG(D_QUOTA, "%s: updating global index hardlimit: %llu, "
 304                        "softlimit: %llu for id %llu\n",
 305                        qqi->qqi_qsd->qsd_svname, glb_rec->qbr_hardlimit,
 306                        glb_rec->qbr_softlimit, qid->qid_uid);
 307         } else {
 308                 /* Update record in slave index copy */
 309                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
 310
 311                 CDEBUG(D_QUOTA, "%s: update granted to %llu for id %llu"
 312                        "\n", qqi->qqi_qsd->qsd_svname, slv_rec->qsr_granted,
 313                        qid->qid_uid);
 314         }
 315
 316         if (ver != 0) {
 317                 new_verp = &ver;
 318                 flags = LQUOTA_SET_VER;
 319         }
 320
 321         /* write new record to index file */
 322         rc = lquota_disk_write(env, th, obj, qid, (struct dt_rec *)rec, flags,
 323                                new_verp);
 324         EXIT;
 325 out:
 326         dt_trans_stop(env, qqi->qqi_qsd->qsd_dev, th);
 327         if (rc)
 328                 CERROR("%s: failed to update %s index copy for id %llu, : rc = %d\n",
 329                        qqi->qqi_qsd->qsd_svname,
 330                        global ? "global" : "slave", qid->qid_uid, rc);
 331         else if (flags == LQUOTA_SET_VER)
 332                 qsd_bump_version(qqi, ver, global);
 333         return rc;
 334 }
 335
 336 /*
 337  * Update in-memory lquota entry with new quota setting from record \rec.
 338  * The record can either be a global record (i.e. lquota_glb_rec) or a slave
 339  * index record (i.e. lquota_slv_rec). In the former case, \global should be
 340  * set to true.
 341  *
 342  * \param env    - the environment passed by the caller
 343  * \param lqe    - is the quota entry associated with the identifier
 344  * \param global - is set to true when updating the record is of type
 345  *                 lquota_glb_rec. Otherwise, it is a lquota_slv_rec record.
 346  * \param rec    - is the updated record received from the master.
 347  */
 348 int qsd_update_lqe(const struct lu_env *env, struct lquota_entry *lqe,
 349                    bool global, void *rec)
 350 {
 351         ENTRY;
 352
 353         LASSERT(lqe != NULL);
 354         LASSERT(!lqe_is_master(lqe));
 355
 356         /* updating lqe is always serialized, no locking needed. */
 357         if (global) {
 358                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
 359
 360                 /* doesn't change quota enforcement if the quota entry is still
 361                  * using default quota. */
 362                 if (LQUOTA_FLAG(glb_rec->qbr_time) & LQUOTA_FLAG_DEFAULT &&
 363                     glb_rec->qbr_hardlimit == 0 && glb_rec->qbr_softlimit == 0)
 364                         RETURN(0);
 365
 366                 LQUOTA_DEBUG(lqe, "the ID has been set quota, so clear the"
 367                              " default quota flag");
 368                 lqe->lqe_is_default = false;
 369
 370                 /* change enforcement status based on new hard/soft limit */
 371                 if (lqe->lqe_id.qid_uid != 0 && (glb_rec->qbr_hardlimit != 0 ||
 372                     glb_rec->qbr_softlimit != 0))
 373                         lqe->lqe_enforced = true;
 374                 else
 375                         lqe->lqe_enforced = false;
 376
 377                 LQUOTA_DEBUG(lqe, "updating global index hardlimit: %llu, "
 378                              "softlimit: %llu", glb_rec->qbr_hardlimit,
 379                              glb_rec->qbr_softlimit);
 380         } else {
 381                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
 382
 383                 lqe->lqe_granted = slv_rec->qsr_granted;
 384
 385                 LQUOTA_DEBUG(lqe, "updating slave index, granted:%llu",
 386                              slv_rec->qsr_granted);
 387         }
 388
 389         RETURN(0);
 390 }