lustre/quota/qmt_pool.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 /*
  32  * A Quota Master Target has a list(qmt_pool_list) where it stores qmt_pool_info
  33  * structures. There is one such structure for each pool managed by the QMT.
  34  *
  35  * Each pool can have different quota types enforced (typically user & group
  36  * quota). A pool is in charge of managing lquota_entry structures for each
  37  * quota type. This is done by creating one lquota_entry site per quota
  38  * type. A site stores entries in a hash table and read quota settings from disk
  39  * when a given ID isn't present in the hash.
  40  *
  41  * The pool API exported here is the following:
  42  * - qmt_pool_init(): initializes the general QMT structures used to manage
  43  *                    pools.
  44  * - qmt_pool_fini(): frees the structures allocated by qmt_pool_fini().
  45  * - qmt_pool_prepare(): sets up the on-disk indexes associated with each pool.
  46  * - qmt_pool_new_conn(): is used to create a new slave index file.
  47  * - qmt_pool_lqe_lookup(): returns an up-to-date lquota entry associated with
  48  *                          a given ID.
  49  */
  50
  51 #define DEBUG_SUBSYSTEM S_LQUOTA
  52
  53 #include <obd_class.h>
  54 #include <lprocfs_status.h>
  55 #include "qmt_internal.h"
  56
  57 static void qmt_pool_free(const struct lu_env *, struct qmt_pool_info *);
  58
  59 /*
  60  * Static helper functions not used outside the scope of this file
  61  */
  62
  63 /*
  64  * Reference counter management for qmt_pool_info structures
  65  */
  66 static inline void qpi_getref(struct qmt_pool_info *pool)
  67 {
  68         atomic_inc(&pool->qpi_ref);
  69 }
  70
  71 static inline void qpi_putref(const struct lu_env *env,
  72                               struct qmt_pool_info *pool)
  73 {
  74         LASSERT(atomic_read(&pool->qpi_ref) > 0);
  75         if (atomic_dec_and_test(&pool->qpi_ref))
  76                 qmt_pool_free(env, pool);
  77 }
  78
  79 static inline void qpi_putref_locked(struct qmt_pool_info *pool)
  80 {
  81         LASSERT(atomic_read(&pool->qpi_ref) > 1);
  82         atomic_dec(&pool->qpi_ref);
  83 }
  84
  85 /* some procfs helpers */
  86 static int qpi_state_seq_show(struct seq_file *m, void *data)
  87 {
  88         struct qmt_pool_info    *pool = m->private;
  89         int                      type;
  90
  91         LASSERT(pool != NULL);
  92
  93         seq_printf(m, "pool:\n"
  94                    "    id: %u\n"
  95                    "    type: %s\n"
  96                    "    ref: %d\n"
  97                    "    least qunit: %lu\n",
  98                    0,
  99                    RES_NAME(pool->qpi_rtype),
 100                    atomic_read(&pool->qpi_ref),
 101                    pool->qpi_least_qunit);
 102
 103         for (type = 0; type < LL_MAXQUOTAS; type++)
 104                 seq_printf(m, "    %s:\n"
 105                            "        #slv: %d\n"
 106                            "        #lqe: %d\n",
 107                            qtype_name(type),
 108                            pool->qpi_slv_nr[type],
 109                     atomic_read(&pool->qpi_site[type]->lqs_hash->hs_count));
 110
 111         return 0;
 112 }
 113 LPROC_SEQ_FOPS_RO(qpi_state);
 114
 115 static int qpi_soft_least_qunit_seq_show(struct seq_file *m, void *data)
 116 {
 117         struct qmt_pool_info    *pool = m->private;
 118         LASSERT(pool != NULL);
 119
 120         seq_printf(m, "%lu\n", pool->qpi_soft_least_qunit);
 121         return 0;
 122 }
 123
 124 static ssize_t
 125 qpi_soft_least_qunit_seq_write(struct file *file, const char __user *buffer,
 126                                size_t count, loff_t *off)
 127 {
 128         struct seq_file *m = file->private_data;
 129         struct qmt_pool_info *pool = m->private;
 130         long long least_qunit;
 131         int qunit, rc;
 132
 133         LASSERT(pool != NULL);
 134
 135         /* Not tuneable for inode limit */
 136         if (pool->qpi_rtype != LQUOTA_RES_DT)
 137                 return -EINVAL;
 138
 139         rc = kstrtoll_from_user(buffer, count, 0, &least_qunit);
 140         if (rc)
 141                 return rc;
 142
 143         /* Miminal qpi_soft_least_qunit */
 144         qunit = pool->qpi_least_qunit << 2;
 145         /* The value must be power of miminal qpi_soft_least_qunit, see
 146          * how the qunit is adjusted in qmt_adjust_qunit(). */
 147         while (qunit > 0 && qunit < least_qunit)
 148                 qunit <<= 2;
 149         if (qunit <= 0)
 150                 qunit = INT_MAX & ~3;
 151
 152         pool->qpi_soft_least_qunit = qunit;
 153         return count;
 154 }
 155 LPROC_SEQ_FOPS(qpi_soft_least_qunit);
 156
 157 static struct lprocfs_vars lprocfs_quota_qpi_vars[] = {
 158         { .name =       "info",
 159           .fops =       &qpi_state_fops },
 160         { .name =       "soft_least_qunit",
 161           .fops =       &qpi_soft_least_qunit_fops },
 162         { NULL }
 163 };
 164
 165 /*
 166  * Allocate a new qmt_pool_info structure and add it to qmt_pool_list.
 167  *
 168  * \param env       - is the environment passed by the caller
 169  * \param qmt       - is the quota master target
 170  * \param pool_type - is the resource type of this pool instance, either
 171  *                    LQUOTA_RES_MD or LQUOTA_RES_DT.
 172  *
 173  * \retval - 0 on success, appropriate error on failure
 174  */
 175 static int qmt_pool_alloc(const struct lu_env *env, struct qmt_device *qmt,
 176                           char *pool_name, int pool_type)
 177 {
 178         struct qmt_thread_info  *qti = qmt_info(env);
 179         struct qmt_pool_info    *pool;
 180         int                      rc = 0;
 181         ENTRY;
 182
 183         OBD_ALLOC_PTR(pool);
 184         if (pool == NULL)
 185                 RETURN(-ENOMEM);
 186         INIT_LIST_HEAD(&pool->qpi_linkage);
 187
 188         pool->qpi_rtype = pool_type;
 189
 190         /* initialize refcount to 1, hash table will then grab an additional
 191          * reference */
 192         atomic_set(&pool->qpi_ref, 1);
 193
 194         /* set up least qunit size to use for this pool */
 195         pool->qpi_least_qunit = LQUOTA_LEAST_QUNIT(pool_type);
 196         if (pool_type == LQUOTA_RES_DT)
 197                 pool->qpi_soft_least_qunit = pool->qpi_least_qunit << 2;
 198         else
 199                 pool->qpi_soft_least_qunit = pool->qpi_least_qunit;
 200
 201         /* grab reference on master target that this pool belongs to */
 202         lu_device_get(qmt2lu_dev(qmt));
 203         lu_ref_add(&qmt2lu_dev(qmt)->ld_reference, "pool", pool);
 204         pool->qpi_qmt = qmt;
 205
 206         /* create pool proc directory */
 207         snprintf(qti->qti_buf, LQUOTA_NAME_MAX, "%s-%s",
 208                  RES_NAME(pool_type), pool_name);
 209         strncpy(pool->qpi_name, pool_name, QPI_MAXNAME);
 210         pool->qpi_proc = lprocfs_register(qti->qti_buf, qmt->qmt_proc,
 211                                           lprocfs_quota_qpi_vars, pool);
 212         if (IS_ERR(pool->qpi_proc)) {
 213                 rc = PTR_ERR(pool->qpi_proc);
 214                 CERROR("%s: failed to create proc entry for pool %s (%d)\n",
 215                        qmt->qmt_svname, qti->qti_buf, rc);
 216                 pool->qpi_proc = NULL;
 217                 GOTO(out, rc);
 218         }
 219
 220         /* add to qmt pool list */
 221         write_lock(&qmt->qmt_pool_lock);
 222         list_add_tail(&pool->qpi_linkage, &qmt->qmt_pool_list);
 223         write_unlock(&qmt->qmt_pool_lock);
 224         EXIT;
 225 out:
 226         if (rc)
 227                 /* this frees the pool structure since refcount is equal to 1 */
 228                 qpi_putref(env, pool);
 229         return rc;
 230 }
 231
 232 /*
 233  * Delete a qmt_pool_info instance and all structures associated.
 234  *
 235  * \param env  - is the environment passed by the caller
 236  * \param pool - is the qmt_pool_info structure to free
 237  */
 238 static void qmt_pool_free(const struct lu_env *env, struct qmt_pool_info *pool)
 239 {
 240         struct  qmt_device *qmt = pool->qpi_qmt;
 241         int     qtype;
 242         ENTRY;
 243
 244         /* remove from list */
 245         write_lock(&qmt->qmt_pool_lock);
 246         list_del_init(&pool->qpi_linkage);
 247         write_unlock(&qmt->qmt_pool_lock);
 248
 249         if (atomic_read(&pool->qpi_ref) > 0)
 250                 RETURN_EXIT;
 251
 252         /* release proc entry */
 253         if (pool->qpi_proc) {
 254                 lprocfs_remove(&pool->qpi_proc);
 255                 pool->qpi_proc = NULL;
 256         }
 257
 258         /* release per-quota type site used to manage quota entries as well as
 259          * references to global index files */
 260         for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
 261                 /* release lqe storing grace time */
 262                 if (pool->qpi_grace_lqe[qtype] != NULL)
 263                         lqe_putref(pool->qpi_grace_lqe[qtype]);
 264
 265                 /* release site */
 266                 if (pool->qpi_site[qtype] != NULL &&
 267                     !IS_ERR(pool->qpi_site[qtype]))
 268                         lquota_site_free(env, pool->qpi_site[qtype]);
 269                 /* release reference to global index */
 270                 if (pool->qpi_glb_obj[qtype] != NULL &&
 271                     !IS_ERR(pool->qpi_glb_obj[qtype]))
 272                         dt_object_put(env, pool->qpi_glb_obj[qtype]);
 273         }
 274
 275         /* release reference on pool directory */
 276         if (pool->qpi_root != NULL && !IS_ERR(pool->qpi_root))
 277                 dt_object_put(env, pool->qpi_root);
 278
 279         /* release reference on the master target */
 280         if (pool->qpi_qmt != NULL) {
 281                 struct lu_device *ld = qmt2lu_dev(pool->qpi_qmt);
 282
 283                 lu_device_put(ld);
 284                 lu_ref_del(&ld->ld_reference, "pool", pool);
 285                 pool->qpi_qmt = NULL;
 286         }
 287
 288         LASSERT(list_empty(&pool->qpi_linkage));
 289         OBD_FREE_PTR(pool);
 290 }
 291
 292 /*
 293  * Look-up a pool in a list based on the type.
 294  *
 295  * \param env     - is the environment passed by the caller
 296  * \param qmt     - is the quota master target
 297  * \param pool_type - is the type of this pool, either LQUOTA_RES_MD or
 298  *                    LQUOTA_RES_DT.
 299  */
 300 static struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
 301                                              struct qmt_device *qmt,
 302                                              int pool_type)
 303 {
 304         struct qmt_pool_info    *pos, *pool;
 305         ENTRY;
 306
 307         read_lock(&qmt->qmt_pool_lock);
 308         if (list_empty(&qmt->qmt_pool_list)) {
 309                 read_unlock(&qmt->qmt_pool_lock);
 310                 RETURN(ERR_PTR(-ENOENT));
 311         }
 312
 313         /* Now just find a pool with correct type in a list. Further we need
 314          * to go through the list and find a pool that includes requested OST
 315          * or MDT. Possibly this would return a list of pools that includes
 316          * needed target(OST/MDT). */
 317         pool = NULL;
 318         list_for_each_entry(pos, &qmt->qmt_pool_list, qpi_linkage) {
 319                 if (pos->qpi_rtype == pool_type) {
 320                         pool = pos;
 321                         qpi_getref(pool);
 322                         break;
 323                 }
 324         }
 325         read_unlock(&qmt->qmt_pool_lock);
 326
 327         RETURN(pool);
 328 }
 329
 330 /*
 331  * Functions implementing the pool API, used by the qmt handlers
 332  */
 333
 334 /*
 335  * Destroy all pools which are still in the pool list.
 336  *
 337  * \param env - is the environment passed by the caller
 338  * \param qmt - is the quota master target
 339  *
 340  */
 341 void qmt_pool_fini(const struct lu_env *env, struct qmt_device *qmt)
 342 {
 343         struct qmt_pool_info *pool, *tmp;
 344         ENTRY;
 345
 346         /* parse list of pool and destroy each element */
 347         list_for_each_entry_safe(pool, tmp, &qmt->qmt_pool_list, qpi_linkage) {
 348                 /* release extra reference taken in qmt_pool_alloc */
 349                 qpi_putref(env, pool);
 350         }
 351         LASSERT(list_empty(&qmt->qmt_pool_list));
 352
 353         EXIT;
 354 }
 355
 356 /*
 357  * Initialize pool configure for the quota master target. For now, we only
 358  * support the default data (i.e. all OSTs) and metadata (i.e. all the MDTs)
 359  * pool which are instantiated in this function.
 360  *
 361  * \param env - is the environment passed by the caller
 362  * \param qmt - is the quota master target for which we have to initialize the
 363  *              pool configuration
 364  *
 365  * \retval - 0 on success, appropriate error on failure
 366  */
 367 int qmt_pool_init(const struct lu_env *env, struct qmt_device *qmt)
 368 {
 369         int     res, rc = 0;
 370         ENTRY;
 371
 372         INIT_LIST_HEAD(&qmt->qmt_pool_list);
 373         rwlock_init(&qmt->qmt_pool_lock);
 374
 375         /* Instantiate pool master for the default data and metadata pool.
 376          * This code will have to be revisited once we support quota on
 377          * non-default pools */
 378         for (res = LQUOTA_FIRST_RES; res < LQUOTA_LAST_RES; res++) {
 379                 rc = qmt_pool_alloc(env, qmt, "0x0", res);
 380                 if (rc)
 381                         break;
 382         }
 383
 384         if (rc)
 385                 qmt_pool_fini(env, qmt);
 386
 387         RETURN(rc);
 388 }
 389
 390 static int qmt_slv_cnt(const struct lu_env *env, struct lu_fid *glb_fid,
 391                        char *slv_name, struct lu_fid *slv_fid, void *arg)
 392 {
 393         int *nr = arg;
 394
 395         /* one more slave */
 396         (*nr)++;
 397
 398         return 0;
 399 }
 400
 401 /*
 402  * Set up on-disk index files associated with each pool.
 403  *
 404  * \param env - is the environment passed by the caller
 405  * \param qmt - is the quota master target for which we have to initialize the
 406  *              pool configuration
 407  * \param qmt_root - is the on-disk directory created for the QMT.
 408  *
 409  * \retval - 0 on success, appropriate error on failure
 410  */
 411 int qmt_pool_prepare(const struct lu_env *env, struct qmt_device *qmt,
 412                      struct dt_object *qmt_root)
 413 {
 414         struct qmt_thread_info  *qti = qmt_info(env);
 415         struct lquota_glb_rec   *rec = &qti->qti_glb_rec;
 416         struct qmt_pool_info    *pool;
 417         struct dt_device        *dev = NULL;
 418         dt_obj_version_t         version;
 419         struct list_head        *pos;
 420         int                      rc = 0, qtype;
 421         ENTRY;
 422
 423         /* iterate over each pool in the list and allocate a quota site for each
 424          * one. This involves creating a global index file on disk */
 425         list_for_each(pos, &qmt->qmt_pool_list) {
 426                 struct dt_object        *obj;
 427                 struct lquota_entry     *lqe;
 428                 char                    *pool_name;
 429                 int                      pool_type;
 430
 431                 pool = list_entry(pos, struct qmt_pool_info,
 432                                   qpi_linkage);
 433
 434                 pool_name = pool->qpi_name;
 435                 pool_type = pool->qpi_rtype;
 436                 if (dev == NULL)
 437                         dev = pool->qpi_qmt->qmt_child;
 438
 439                 /* allocate directory for this pool */
 440                 snprintf(qti->qti_buf, LQUOTA_NAME_MAX, "%s-%s",
 441                          RES_NAME(pool_type), pool_name);
 442                 obj = lquota_disk_dir_find_create(env, qmt->qmt_child, qmt_root,
 443                                                   qti->qti_buf);
 444                 if (IS_ERR(obj))
 445                         RETURN(PTR_ERR(obj));
 446                 pool->qpi_root = obj;
 447
 448                 for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
 449                         /* Generating FID of global index in charge of storing
 450                          * settings for this quota type */
 451                         lquota_generate_fid(&qti->qti_fid, pool_type, qtype);
 452
 453                         /* open/create the global index file for this quota
 454                          * type */
 455                         obj = lquota_disk_glb_find_create(env, dev,
 456                                                           pool->qpi_root,
 457                                                           &qti->qti_fid, false);
 458                         if (IS_ERR(obj)) {
 459                                 rc = PTR_ERR(obj);
 460                                 CERROR("%s: failed to create glb index copy for %s type: rc = %d\n",
 461                                        qmt->qmt_svname, qtype_name(qtype), rc);
 462                                 RETURN(rc);
 463                         }
 464
 465                         pool->qpi_glb_obj[qtype] = obj;
 466
 467                         version = dt_version_get(env, obj);
 468                         /* set default grace time for newly created index */
 469                         if (version == 0) {
 470                                 rec->qbr_hardlimit = 0;
 471                                 rec->qbr_softlimit = 0;
 472                                 rec->qbr_granted = 0;
 473                                 rec->qbr_time = pool_type == LQUOTA_RES_MD ?
 474                                         MAX_IQ_TIME : MAX_DQ_TIME;
 475
 476                                 rc = lquota_disk_write_glb(env, obj, 0, rec);
 477                                 if (rc) {
 478                                         CERROR("%s: failed to set default grace time for %s type: rc = %d\n",
 479                                                qmt->qmt_svname, qtype_name(qtype), rc);
 480                                         RETURN(rc);
 481                                 }
 482
 483                                 rc = lquota_disk_update_ver(env, dev, obj, 1);
 484                                 if (rc) {
 485                                         CERROR("%s: failed to set initial version for %s type: rc = %d\n",
 486                                                qmt->qmt_svname, qtype_name(qtype), rc);
 487                                         RETURN(rc);
 488                                 }
 489                         }
 490
 491                         /* create quota entry site for this quota type */
 492                         pool->qpi_site[qtype] = lquota_site_alloc(env, pool,
 493                                                                   true, qtype,
 494                                                                   &qmt_lqe_ops);
 495                         if (IS_ERR(pool->qpi_site[qtype])) {
 496                                 rc = PTR_ERR(pool->qpi_site[qtype]);
 497                                 CERROR("%s: failed to create site for %s type: rc = %d\n",
 498                                        qmt->qmt_svname, qtype_name(qtype), rc);
 499                                 RETURN(rc);
 500                         }
 501
 502                         /* count number of slaves which already connected to
 503                          * the master in the past */
 504                         pool->qpi_slv_nr[qtype] = 0;
 505                         rc = lquota_disk_for_each_slv(env, pool->qpi_root,
 506                                                       &qti->qti_fid,
 507                                                       qmt_slv_cnt,
 508                                                       &pool->qpi_slv_nr[qtype]);
 509                         if (rc) {
 510                                 CERROR("%s: failed to scan & count slave indexes for %s type: rc = %d\n",
 511                                        qmt->qmt_svname, qtype_name(qtype), rc);
 512                                 RETURN(rc);
 513                         }
 514
 515                         /* Global grace time is stored in quota settings of
 516                          * ID 0. */
 517                         qti->qti_id.qid_uid = 0;
 518
 519                         /* look-up quota entry storing grace time */
 520                         lqe = lqe_locate(env, pool->qpi_site[qtype],
 521                                          &qti->qti_id);
 522                         if (IS_ERR(lqe))
 523                                 RETURN(PTR_ERR(lqe));
 524                         pool->qpi_grace_lqe[qtype] = lqe;
 525 #ifdef CONFIG_PROC_FS
 526                         /* add procfs file to dump the global index, mostly for
 527                          * debugging purpose */
 528                         snprintf(qti->qti_buf, MTI_NAME_MAXLEN,
 529                                  "glb-%s", qtype_name(qtype));
 530                         rc = lprocfs_seq_create(pool->qpi_proc, qti->qti_buf,
 531                                                 0444, &lprocfs_quota_seq_fops,
 532                                                 obj);
 533                         if (rc)
 534                                 CWARN("%s: Error adding procfs file for global quota index "DFID": rc = %d\n",
 535                                       qmt->qmt_svname, PFID(&qti->qti_fid), rc);
 536 #endif
 537                 }
 538         }
 539
 540         RETURN(0);
 541 }
 542
 543 /*
 544  * Handle new slave connection. Called when a slave enqueues the global quota
 545  * lock at the beginning of the reintegration procedure.
 546  *
 547  * \param env - is the environment passed by the caller
 548  * \parap qmt - is the quota master target handling this request
 549  * \param glb_fid - is the fid of the global index file
 550  * \param slv_fid - is the fid of the newly created slave index file
 551  * \param slv_ver - is the current version of the slave index file
 552  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 553  *                  target
 554  *
 555  * \retval - 0 on success, appropriate error on failure
 556  */
 557 int qmt_pool_new_conn(const struct lu_env *env, struct qmt_device *qmt,
 558                       struct lu_fid *glb_fid, struct lu_fid *slv_fid,
 559                       __u64 *slv_ver, struct obd_uuid *uuid)
 560 {
 561         struct qmt_pool_info    *pool;
 562         struct dt_object        *slv_obj;
 563         int                      pool_type, qtype;
 564         bool                     created = false;
 565         int                      rc = 0;
 566
 567         /* extract pool info from global index FID */
 568         rc = lquota_extract_fid(glb_fid, &pool_type, &qtype);
 569         if (rc)
 570                 RETURN(rc);
 571
 572         /* look-up pool in charge of this global index FID */
 573         pool = qmt_pool_lookup(env, qmt, pool_type);
 574         if (IS_ERR(pool))
 575                 RETURN(PTR_ERR(pool));
 576
 577         /* look-up slave index file */
 578         slv_obj = lquota_disk_slv_find(env, qmt->qmt_child, pool->qpi_root,
 579                                        glb_fid, uuid);
 580         if (IS_ERR(slv_obj) && PTR_ERR(slv_obj) == -ENOENT) {
 581                 /* create slave index file */
 582                 slv_obj = lquota_disk_slv_find_create(env, qmt->qmt_child,
 583                                                       pool->qpi_root, glb_fid,
 584                                                       uuid, false);
 585                 created = true;
 586         }
 587         if (IS_ERR(slv_obj)) {
 588                 rc = PTR_ERR(slv_obj);
 589                 CERROR("%s: failed to create quota slave index file for %s (%d)"
 590                        "\n", qmt->qmt_svname, obd_uuid2str(uuid), rc);
 591                 GOTO(out, rc);
 592         }
 593
 594         /* retrieve slave fid & current object version */
 595         memcpy(slv_fid, lu_object_fid(&slv_obj->do_lu), sizeof(*slv_fid));
 596         *slv_ver = dt_version_get(env, slv_obj);
 597         dt_object_put(env, slv_obj);
 598         if (created)
 599                 pool->qpi_slv_nr[qtype]++;
 600 out:
 601         qpi_putref(env, pool);
 602         RETURN(rc);
 603 }
 604
 605 /*
 606  * Look-up a lquota_entry in the pool hash and allocate it if not found.
 607  *
 608  * \param env - is the environment passed by the caller
 609  * \param qmt - is the quota master target for which we have to initialize the
 610  *              pool configuration
 611  * \param pool_type - is the pool type, either LQUOTA_RES_MD or LQUOTA_RES_DT.
 612  * \param qtype     - is the quota type, either user or group.
 613  * \param qid       - is the quota ID to look-up
 614  *
 615  * \retval - valid pointer to lquota entry on success, appropriate error on
 616  *           failure
 617  */
 618 struct lquota_entry *qmt_pool_lqe_lookup(const struct lu_env *env,
 619                                          struct qmt_device *qmt,
 620                                          int pool_type, int qtype,
 621                                          union lquota_id *qid)
 622 {
 623         struct qmt_pool_info    *pool;
 624         struct lquota_entry     *lqe;
 625         ENTRY;
 626
 627         /* look-up pool responsible for this global index FID */
 628         pool = qmt_pool_lookup(env, qmt, pool_type);
 629         if (IS_ERR(pool))
 630                 RETURN((void *)pool);
 631
 632         if (qid->qid_uid == 0) {
 633                 /* caller wants to access grace time, no need to look up the
 634                  * entry since we keep a reference on ID 0 all the time */
 635                 lqe = pool->qpi_grace_lqe[qtype];
 636                 lqe_getref(lqe);
 637                 GOTO(out, lqe);
 638         }
 639
 640         /* now that we have the pool, let's look-up the quota entry in the
 641          * right quota site */
 642         lqe = lqe_locate(env, pool->qpi_site[qtype], qid);
 643 out:
 644         qpi_putref(env, pool);
 645         RETURN(lqe);
 646 }