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, 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 #ifndef EXPORT_SYMTAB
  32 # define EXPORT_SYMTAB
  33 #endif
  34
  35 #define DEBUG_SUBSYSTEM S_LQUOTA
  36
  37 #include "qsd_internal.h"
  38
  39 /*
  40  * Initialize qsd-specific fields of quota entry.
  41  *
  42  * \param lqe - is the quota entry to initialize
  43  * \param arg - is the pointer to the qsd_qtype_info structure
  44  */
  45 static void qsd_lqe_init(struct lquota_entry *lqe, void *arg)
  46 {
  47         LASSERT(!lqe_is_master(lqe));
  48
  49         /* initialize slave parameters */
  50         rwlock_init(&lqe->lqe_lock);
  51         memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh));
  52         lqe->lqe_pending_write = 0;
  53         lqe->lqe_pending_req   = 0;
  54         cfs_waitq_init(&lqe->lqe_waiters);
  55         lqe->lqe_usage    = 0;
  56         lqe->lqe_nopreacq = false;
  57 }
  58
  59 /*
  60  * Update a slave quota entry. This is done by reading enforcement status from
  61  * the copy of the global index and then how much is the slave currenly owns
  62  * for this user from the slave index copy.
  63  *
  64  * \param env - the environment passed by the caller
  65  * \param lqe - is the quota entry to refresh
  66  * \param arg - is the pointer to the qsd_qtype_info structure
  67  */
  68 static int qsd_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
  69                         void *arg)
  70 {
  71         struct qsd_thread_info *qti = qsd_info(env);
  72         struct qsd_qtype_info  *qqi = (struct qsd_qtype_info *)arg;
  73         int                     rc;
  74
  75         LASSERT(!lqe_is_master(lqe));
  76
  77         /* read record from global index copy to know whether quota is
  78          * enforced for this user */
  79         rc = lquota_disk_read(env, qqi->qqi_glb_obj, &lqe->lqe_id,
  80                               (struct dt_rec *)&qti->qti_glb_rec);
  81
  82         switch(rc) {
  83         case -ENOENT:
  84                 /* no such entry, assume quota isn't enforced for this user */
  85                 lqe->lqe_enforced = false;
  86                 break;
  87         case 0:
  88                 if (qti->qti_glb_rec.qbr_hardlimit == 0 &&
  89                     qti->qti_glb_rec.qbr_softlimit == 0)
  90                         /* quota isn't enforced for this use */
  91                         lqe->lqe_enforced = false;
  92                 else
  93                         lqe->lqe_enforced = true;
  94                 break;
  95         default:
  96                 LQUOTA_ERROR(lqe, "failed to read quota entry from global "
  97                              "index copy, rc:%d", rc);
  98                 return rc;
  99         }
 100
 101         /* read record from slave index copy to find out how much space is
 102          * currently owned by this slave */
 103         rc = lquota_disk_read(env, qqi->qqi_slv_obj, &lqe->lqe_id,
 104                               (struct dt_rec *)&qti->qti_slv_rec);
 105         switch(rc) {
 106         case -ENOENT:
 107                 lqe->lqe_granted = 0;
 108                 break;
 109         case 0:
 110                 lqe->lqe_granted = qti->qti_slv_rec.qsr_granted;
 111                 break;
 112         default:
 113                 LQUOTA_ERROR(lqe, "failed to read quota entry from slave "
 114                              "index copy, rc:%d", rc);
 115                 return rc;
 116         }
 117
 118         /* don't know what the qunit value is yet */
 119         qsd_set_qunit(lqe, 0);
 120
 121         /* read current disk-usage from disk */
 122         rc = qsd_refresh_usage(env, lqe);
 123         if (rc)
 124                 return rc;
 125
 126         LQUOTA_DEBUG(lqe, "successfully read from disk");
 127         return 0;
 128 }
 129
 130 /*
 131  * Print lqe information for debugging.
 132  *
 133  * \param lqe - is the quota entry to debug
 134  * \param arg - is the pointer to the qsd_qtype_info structure
 135  * \param msgdata - debug message
 136  * \param fmt     - format of debug message
 137  */
 138 static void qsd_lqe_debug(struct lquota_entry *lqe, void *arg,
 139                           struct libcfs_debug_msg_data *msgdata,
 140                           const char *fmt, va_list args)
 141 {
 142         struct qsd_qtype_info   *qqi = (struct qsd_qtype_info *)arg;
 143
 144         libcfs_debug_vmsg2(msgdata, fmt, args,
 145                            "qsd:%s qtype:%s id:"LPU64" enforced:%d granted:"
 146                            LPU64" pending:"LPU64" waiting:"LPU64" req:%d usage:"
 147                            LPU64" qunit:"LPU64" qtune:"LPU64" edquot:%d\n",
 148                            qqi->qqi_qsd->qsd_svname, QTYPE_NAME(qqi->qqi_qtype),
 149                            lqe->lqe_id.qid_uid, lqe->lqe_enforced,
 150                            lqe->lqe_granted, lqe->lqe_pending_write,
 151                            lqe->lqe_waiting_write, lqe->lqe_pending_req,
 152                            lqe->lqe_usage, lqe->lqe_qunit, lqe->lqe_qtune,
 153                            lqe->lqe_edquot);
 154 }
 155
 156 /*
 157  * Vector of quota entry operations supported on the slave
 158  */
 159 struct lquota_entry_operations qsd_lqe_ops = {
 160         .lqe_init               = qsd_lqe_init,
 161         .lqe_read               = qsd_lqe_read,
 162         .lqe_debug              = qsd_lqe_debug,
 163 };
 164
 165 int qsd_write_version(const struct lu_env *env, struct qsd_qtype_info *qqi,
 166                       __u64 ver, bool global)
 167 {
 168         struct qsd_instance *qsd = qqi->qqi_qsd;
 169         struct dt_object    *obj = global ? qqi->qqi_glb_obj :
 170                                             qqi->qqi_slv_obj;
 171         int                  rc;
 172         ENTRY;
 173
 174         rc = lquota_disk_update_ver(env, qsd->qsd_dev, obj, ver);
 175         if (rc)
 176                 RETURN(rc);
 177
 178         qsd_bump_version(qqi, ver, global);
 179         RETURN(0);
 180 }
 181
 182 /*
 183  * Consult current disk space consumed by a given identifier.
 184  *
 185  * \param env   - the environment passed by the caller
 186  * \param qqi   - is the pointer to the qsd_qtype_info structure associated
 187  *                with the identifier.
 188  * \param lqe   - is the quota entry associated with the identifier
 189  */
 190 int qsd_refresh_usage(const struct lu_env *env, struct lquota_entry *lqe)
 191 {
 192         struct qsd_thread_info  *qti = qsd_info(env);
 193         struct lquota_acct_rec  *rec = &qti->qti_acct_rec;
 194         struct qsd_qtype_info   *qqi = lqe2qqi(lqe);
 195         int                      rc = 0;
 196         ENTRY;
 197
 198         LASSERT(qqi->qqi_acct_obj);
 199
 200         /* read disk usage */
 201         rc = lquota_disk_read(env, qqi->qqi_acct_obj, &lqe->lqe_id,
 202                               (struct dt_rec *)rec);
 203         switch(rc) {
 204         case -ENOENT:
 205                 lqe->lqe_usage = 0;
 206                 rc = 0;
 207                 break;
 208         case 0:
 209                 if (qqi->qqi_qsd->qsd_is_md)
 210                         lqe->lqe_usage = rec->ispace;
 211                 else
 212                         lqe->lqe_usage = toqb(rec->bspace);
 213                 break;
 214         default:
 215                 LQUOTA_ERROR(lqe, "failed to read disk usage, rc:%d", rc);
 216                 RETURN(rc);
 217         }
 218
 219         LQUOTA_DEBUG(lqe, "disk usage: "LPU64, lqe->lqe_usage);
 220         RETURN(0);
 221 }
 222
 223 /*
 224  * Update slave or global index copy.
 225  *
 226  * \param env    - the environment passed by the caller
 227  * \param qqi    - is the qsd_type_info structure managing the index to be
 228  *                 update
 229  * \param qid    - is the identifier for which we need to update the quota
 230  *                 settings
 231  * \param global - is set to true when updating the global index copy and to
 232  *                 false for the slave index copy.
 233  * \param ver    - is the new version of the index. If equal to 0, the version
 234  *                 of the index isn't changed
 235  * \param rec    - is the updated record to insert in the index file
 236  */
 237 int qsd_update_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
 238                      union lquota_id *qid, bool global, __u64 ver, void *rec)
 239 {
 240         struct thandle          *th = NULL;
 241         struct dt_object        *obj;
 242         __u64                   *new_verp = NULL;
 243         int                      flags = 0;
 244         int                      rc;
 245         ENTRY;
 246
 247         obj = global ? qqi->qqi_glb_obj : qqi->qqi_slv_obj;
 248
 249         /* allocate transaction */
 250         th = dt_trans_create(env, qqi->qqi_qsd->qsd_dev);
 251         if (IS_ERR(th))
 252                 RETURN(PTR_ERR(th));
 253
 254         /* reserve enough credits to update record in index file */
 255         rc = lquota_disk_declare_write(env, th, obj, qid);
 256         if (rc)
 257                 GOTO(out, rc);
 258
 259         /* start local transaction */
 260         rc = dt_trans_start_local(env, qqi->qqi_qsd->qsd_dev, th);
 261         if (rc)
 262                 GOTO(out, rc);
 263
 264         if (global) {
 265                 /* Update record in global index copy */
 266                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
 267
 268                 CDEBUG(D_QUOTA, "%s: updating global index hardlimit: "LPU64", "
 269                        "softlimit: "LPU64" for id "LPU64"\n",
 270                        qqi->qqi_qsd->qsd_svname, glb_rec->qbr_hardlimit,
 271                        glb_rec->qbr_softlimit, qid->qid_uid);
 272         } else {
 273                 /* Update record in slave index copy */
 274                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
 275
 276                 CDEBUG(D_QUOTA, "%s: update granted to "LPU64" for id "LPU64
 277                        "\n", qqi->qqi_qsd->qsd_svname, slv_rec->qsr_granted,
 278                        qid->qid_uid);
 279         }
 280
 281         if (ver != 0) {
 282                 new_verp = &ver;
 283                 flags = LQUOTA_SET_VER;
 284         }
 285
 286         /* write new record to index file */
 287         rc = lquota_disk_write(env, th, obj, qid, (struct dt_rec *)rec, flags,
 288                                new_verp);
 289         EXIT;
 290 out:
 291         dt_trans_stop(env, qqi->qqi_qsd->qsd_dev, th);
 292         if (rc)
 293                 CERROR("%s: failed to update %s index copy for id "LPU64", rc:"
 294                        "%d\n", qqi->qqi_qsd->qsd_svname,
 295                        global ? "global" : "slave", qid->qid_uid, rc);
 296         else if (flags == LQUOTA_SET_VER)
 297                 qsd_bump_version(qqi, ver, global);
 298         return rc;
 299 }
 300
 301 /*
 302  * Update in-memory lquota entry with new quota setting from record \rec.
 303  * The record can either be a global record (i.e. lquota_glb_rec) or a slave
 304  * index record (i.e. lquota_slv_rec). In the former case, \global should be
 305  * set to true.
 306  *
 307  * \param env    - the environment passed by the caller
 308  * \param lqe    - is the quota entry associated with the identifier
 309  * \param global - is set to true when updating the record is of type
 310  *                 lquota_glb_rec. Otherwise, it is a lquota_slv_rec record.
 311  * \param rec    - is the updated record received from the master.
 312  */
 313 int qsd_update_lqe(const struct lu_env *env, struct lquota_entry *lqe,
 314                    bool global, void *rec)
 315 {
 316         ENTRY;
 317
 318         LASSERT(lqe != NULL);
 319         LASSERT(!lqe_is_master(lqe));
 320
 321         /* updating lqe is always serialized, no locking needed. */
 322         if (global) {
 323                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
 324
 325                 /* change enforcement status based on new hard/soft limit */
 326                 lqe->lqe_enforced = (glb_rec->qbr_hardlimit ||
 327                                      glb_rec->qbr_softlimit) ? true : false;
 328
 329                 LQUOTA_DEBUG(lqe, "updating global index hardlimit: "LPU64", "
 330                              "softlimit: "LPU64, glb_rec->qbr_hardlimit,
 331                              glb_rec->qbr_softlimit);
 332         } else {
 333                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
 334
 335                 lqe->lqe_granted = slv_rec->qsr_granted;
 336
 337                 LQUOTA_DEBUG(lqe, "updating slave index, granted:"LPU64"",
 338                              slv_rec->qsr_granted);
 339         }
 340
 341         RETURN(0);
 342 }