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, 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 hash table 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 #ifndef EXPORT_SYMTAB
  52 # define EXPORT_SYMTAB
  53 #endif
  54
  55 #define DEBUG_SUBSYSTEM S_LQUOTA
  56
  57 #include <obd_class.h>
  58 #include <lprocfs_status.h>
  59 #include "qmt_internal.h"
  60
  61 static void qmt_pool_free(const struct lu_env *, struct qmt_pool_info *);
  62
  63 /*
  64  * Static helper functions not used outside the scope of this file
  65  */
  66
  67 /*
  68  * Reference counter management for qmt_pool_info structures
  69  */
  70 static inline void qpi_getref(struct qmt_pool_info *pool)
  71 {
  72         cfs_atomic_inc(&pool->qpi_ref);
  73 }
  74
  75 static inline void qpi_putref(const struct lu_env *env,
  76                                struct qmt_pool_info *pool)
  77 {
  78         LASSERT(atomic_read(&pool->qpi_ref) > 0);
  79         if (cfs_atomic_dec_and_test(&pool->qpi_ref))
  80                 qmt_pool_free(env, pool);
  81 }
  82
  83 static inline void qpi_putref_locked(struct qmt_pool_info *pool)
  84 {
  85         LASSERT(cfs_atomic_read(&pool->qpi_ref) > 1);
  86         cfs_atomic_dec(&pool->qpi_ref);
  87 }
  88
  89 /*
  90  * Hash functions for qmt_pool_info management
  91  */
  92
  93 static unsigned qpi_hash_hash(cfs_hash_t *hs, const void *key, unsigned mask)
  94 {
  95         return cfs_hash_u32_hash(*((__u32 *)key), mask);
  96 }
  97
  98 static void *qpi_hash_key(cfs_hlist_node_t *hnode)
  99 {
 100         struct qmt_pool_info *pool;
 101         pool = cfs_hlist_entry(hnode, struct qmt_pool_info, qpi_hash);
 102         return &pool->qpi_key;
 103 }
 104
 105 static int qpi_hash_keycmp(const void *key, cfs_hlist_node_t *hnode)
 106 {
 107         struct qmt_pool_info *pool;
 108         pool = cfs_hlist_entry(hnode, struct qmt_pool_info, qpi_hash);
 109         return pool->qpi_key == *((__u32 *)key);
 110 }
 111
 112 static void *qpi_hash_object(cfs_hlist_node_t *hnode)
 113 {
 114         return cfs_hlist_entry(hnode, struct qmt_pool_info, qpi_hash);
 115 }
 116
 117 static void qpi_hash_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 118 {
 119         struct qmt_pool_info *pool;
 120         pool = cfs_hlist_entry(hnode, struct qmt_pool_info, qpi_hash);
 121         qpi_getref(pool);
 122 }
 123
 124 static void qpi_hash_put_locked(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 125 {
 126         struct qmt_pool_info *pool;
 127         pool = cfs_hlist_entry(hnode, struct qmt_pool_info, qpi_hash);
 128         qpi_putref_locked(pool);
 129 }
 130
 131 static void qpi_hash_exit(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 132 {
 133         CERROR("Should not have any item left!\n");
 134 }
 135
 136 /* vector of hash operations */
 137 static cfs_hash_ops_t qpi_hash_ops = {
 138         .hs_hash        = qpi_hash_hash,
 139         .hs_key         = qpi_hash_key,
 140         .hs_keycmp      = qpi_hash_keycmp,
 141         .hs_object      = qpi_hash_object,
 142         .hs_get         = qpi_hash_get,
 143         .hs_put_locked  = qpi_hash_put_locked,
 144         .hs_exit        = qpi_hash_exit
 145 };
 146
 147 /* some procfs helpers */
 148 static int lprocfs_qpi_rd_state(char *page, char **start, off_t off,
 149                                 int count, int *eof, void *data)
 150 {
 151         struct qmt_pool_info    *pool = (struct qmt_pool_info *)data;
 152         int                      type, i = 0;
 153
 154         LASSERT(pool != NULL);
 155
 156         i = snprintf(page, count,
 157                      "pool:\n"
 158                      "    id: %u\n"
 159                      "    type: %s\n"
 160                      "    ref: %d\n"
 161                      "    least qunit: %lu\n",
 162                      pool->qpi_key & 0x0000ffff,
 163                      RES_NAME(pool->qpi_key >> 16),
 164                      cfs_atomic_read(&pool->qpi_ref),
 165                      pool->qpi_least_qunit);
 166
 167
 168         for (type = 0; type < MAXQUOTAS; type++)
 169                 i += snprintf(page + i, count - i,
 170                               "    %s:\n"
 171                               "        #slv: %d\n"
 172                               "        #lqe: %d\n",
 173                               QTYPE_NAME(type),
 174                               pool->qpi_slv_nr[type],
 175                     cfs_atomic_read(&pool->qpi_site[type]->lqs_hash->hs_count));
 176
 177         return i;
 178 }
 179
 180 static struct lprocfs_vars lprocfs_quota_qpi_vars[] = {
 181         { "info", lprocfs_qpi_rd_state, 0, 0},
 182         { NULL }
 183 };
 184
 185 /*
 186  * Allocate a new qmt_pool_info structure and add it to the pool hash table
 187  * of the qmt.
 188  *
 189  * \param env       - is the environment passed by the caller
 190  * \param qmt       - is the quota master target
 191  * \param pool_id   - is the 16-bit pool identifier of the new pool to add
 192  * \param pool_type - is the resource type of this pool instance, either
 193  *                    LQUOTA_RES_MD or LQUOTA_RES_DT.
 194  *
 195  * \retval - 0 on success, appropriate error on failure
 196  */
 197 static int qmt_pool_alloc(const struct lu_env *env, struct qmt_device *qmt,
 198                           int pool_id, int pool_type)
 199 {
 200         struct qmt_thread_info  *qti = qmt_info(env);
 201         struct qmt_pool_info    *pool;
 202         int                      rc = 0;
 203         ENTRY;
 204
 205         OBD_ALLOC_PTR(pool);
 206         if (pool == NULL)
 207                 RETURN(-ENOMEM);
 208         CFS_INIT_LIST_HEAD(&pool->qpi_linkage);
 209
 210         /* assign key used by hash functions */
 211         pool->qpi_key = pool_id + (pool_type << 16);
 212
 213         /* initialize refcount to 1, hash table will then grab an additional
 214          * reference */
 215         atomic_set(&pool->qpi_ref, 1);
 216
 217         /* set up least qunit size to use for this pool */
 218         pool->qpi_least_qunit = LQUOTA_LEAST_QUNIT(pool_type);
 219
 220         /* create pool proc directory */
 221         sprintf(qti->qti_buf, "%s-0x%x", RES_NAME(pool_type), pool_id);
 222         pool->qpi_proc = lprocfs_register(qti->qti_buf, qmt->qmt_proc,
 223                                           lprocfs_quota_qpi_vars, pool);
 224         if (IS_ERR(pool->qpi_proc)) {
 225                 rc = PTR_ERR(pool->qpi_proc);
 226                 CERROR("%s: failed to create proc entry for pool %s (%d)\n",
 227                        qmt->qmt_svname, qti->qti_buf, rc);
 228                 pool->qpi_proc = NULL;
 229                 GOTO(out, rc);
 230         }
 231
 232         /* grab reference on master target that this pool belongs to */
 233         lu_device_get(qmt2lu_dev(qmt));
 234         lu_ref_add(&qmt2lu_dev(qmt)->ld_reference, "pool", pool);
 235         pool->qpi_qmt = qmt;
 236
 237         /* add to qmt hash table */
 238         rc = cfs_hash_add_unique(qmt->qmt_pool_hash, &pool->qpi_key,
 239                                  &pool->qpi_hash);
 240         if (rc) {
 241                 CERROR("%s: failed to add pool %s to qmt hash (%d)\n",
 242                        qmt->qmt_svname, qti->qti_buf, rc);
 243                 GOTO(out, rc);
 244         }
 245
 246         /* add to qmt pool list */
 247         cfs_list_add_tail(&pool->qpi_linkage, &qmt->qmt_pool_list);
 248         EXIT;
 249 out:
 250         if (rc)
 251                 /* this frees the pool structure since refcount is equal to 1 */
 252                 qpi_putref(env, pool);
 253         return rc;
 254 }
 255
 256 /*
 257  * Delete a qmt_pool_info instance and all structures associated.
 258  *
 259  * \param env  - is the environment passed by the caller
 260  * \param pool - is the qmt_pool_info structure to free
 261  */
 262 static void qmt_pool_free(const struct lu_env *env, struct qmt_pool_info *pool)
 263 {
 264         int     qtype;
 265         ENTRY;
 266
 267         /* release proc entry */
 268         if (pool->qpi_proc) {
 269                 lprocfs_remove(&pool->qpi_proc);
 270                 pool->qpi_proc = NULL;
 271         }
 272
 273         /* release per-quota type site used to manage quota entries as well as
 274          * references to global index files */
 275         for (qtype = 0; qtype < MAXQUOTAS; qtype++) {
 276                 /* release lqe storing grace time */
 277                 if (pool->qpi_grace_lqe[qtype] != NULL)
 278                         lqe_putref(pool->qpi_grace_lqe[qtype]);
 279
 280                 /* release site */
 281                 if (pool->qpi_site[qtype] != NULL &&
 282                     !IS_ERR(pool->qpi_site[qtype]))
 283                         lquota_site_free(env, pool->qpi_site[qtype]);
 284                 /* release reference to global index */
 285                 if (pool->qpi_glb_obj[qtype] != NULL &&
 286                     !IS_ERR(pool->qpi_glb_obj[qtype]))
 287                         lu_object_put(env, &pool->qpi_glb_obj[qtype]->do_lu);
 288         }
 289
 290         /* release reference on pool directory */
 291         if (pool->qpi_root != NULL && !IS_ERR(pool->qpi_root))
 292                 lu_object_put(env, &pool->qpi_root->do_lu);
 293
 294         /* release reference on the master target */
 295         if (pool->qpi_qmt != NULL) {
 296                 struct lu_device *ld = qmt2lu_dev(pool->qpi_qmt);
 297
 298                 lu_device_put(ld);
 299                 lu_ref_del(&ld->ld_reference, "pool", pool);
 300                 pool->qpi_qmt = NULL;
 301         }
 302
 303         LASSERT(cfs_list_empty(&pool->qpi_linkage));
 304         OBD_FREE_PTR(pool);
 305 }
 306
 307 /*
 308  * Look-up a pool in the hash table based on the pool ID and type.
 309  *
 310  * \param env     - is the environment passed by the caller
 311  * \param qmt     - is the quota master target
 312  * \param pool_id   - is the 16-bit identifier of the pool to look up
 313  * \param pool_type - is the type of this pool, either LQUOTA_RES_MD or
 314  *                    LQUOTA_RES_DT.
 315  */
 316 static struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
 317                                              struct qmt_device *qmt,
 318                                              int pool_id, int pool_type)
 319 {
 320         struct qmt_pool_info    *pool;
 321         __u32                    key;
 322         ENTRY;
 323
 324         LASSERT(qmt->qmt_pool_hash != NULL);
 325
 326         /* look-up pool in hash table */
 327         key = pool_id + (pool_type << 16);
 328         pool = cfs_hash_lookup(qmt->qmt_pool_hash, (void *)&key);
 329         if (pool == NULL) {
 330                 /* this qmt isn't managing this pool! */
 331                 CERROR("%s: looking up quota entry for a pool (0x%x/%d) which "
 332                        "isn't managed by this quota master target\n",
 333                        qmt->qmt_svname, pool_id, pool_type);
 334                 RETURN(ERR_PTR(-ENOENT));
 335         }
 336         RETURN(pool);
 337 }
 338
 339 /*
 340  * Functions implementing the pool API, used by the qmt handlers
 341  */
 342
 343 /*
 344  * Destroy all pools which are still in the hash table and free the pool
 345  * hash table.
 346  *
 347  * \param env - is the environment passed by the caller
 348  * \param qmt - is the quota master target
 349  *
 350  */
 351 void qmt_pool_fini(const struct lu_env *env, struct qmt_device *qmt)
 352 {
 353         struct qmt_pool_info    *pool;
 354         cfs_list_t              *pos, *n;
 355         ENTRY;
 356
 357         if (qmt->qmt_pool_hash == NULL)
 358                 RETURN_EXIT;
 359
 360         /* parse list of pool and destroy each element */
 361         cfs_list_for_each_safe(pos, n, &qmt->qmt_pool_list) {
 362                 pool = cfs_list_entry(pos, struct qmt_pool_info,
 363                                       qpi_linkage);
 364                 /* remove from hash */
 365                 cfs_hash_del(qmt->qmt_pool_hash, &pool->qpi_key,
 366                              &pool->qpi_hash);
 367
 368                 /* remove from list */
 369                 cfs_list_del_init(&pool->qpi_linkage);
 370
 371                 /* release extra reference taken in qmt_pool_alloc */
 372                 qpi_putref(env, pool);
 373         }
 374         LASSERT(cfs_list_empty(&qmt->qmt_pool_list));
 375
 376         cfs_hash_putref(qmt->qmt_pool_hash);
 377         qmt->qmt_pool_hash = NULL;
 378         EXIT;
 379 }
 380
 381 /*
 382  * Initialize pool configure for the quota master target. For now, we only
 383  * support the default data (i.e. all OSTs) and metadata (i.e. all the MDTs)
 384  * pool which are instantiated in this function.
 385  *
 386  * \param env - is the environment passed by the caller
 387  * \param qmt - is the quota master target for which we have to initializa the
 388  *              pool configuration
 389  *
 390  * \retval - 0 on success, appropriate error on failure
 391  */
 392 int qmt_pool_init(const struct lu_env *env, struct qmt_device *qmt)
 393 {
 394         int     res, rc = 0;
 395         ENTRY;
 396
 397         /* initialize pool hash table */
 398         qmt->qmt_pool_hash = cfs_hash_create("POOL_HASH",
 399                                              HASH_POOLS_CUR_BITS,
 400                                              HASH_POOLS_MAX_BITS,
 401                                              HASH_POOLS_BKT_BITS, 0,
 402                                              CFS_HASH_MIN_THETA,
 403                                              CFS_HASH_MAX_THETA,
 404                                              &qpi_hash_ops,
 405                                              CFS_HASH_DEFAULT);
 406         if (qmt->qmt_pool_hash == NULL) {
 407                 CERROR("%s: failed to create pool hash table\n",
 408                        qmt->qmt_svname);
 409                 RETURN(-ENOMEM);
 410         }
 411
 412         /* initialize pool list */
 413         CFS_INIT_LIST_HEAD(&qmt->qmt_pool_list);
 414
 415         /* Instantiate pool master for the default data and metadata pool (both
 416          * have pool ID equals to 0).
 417          * This code will have to be revisited once we support quota on
 418          * non-default pools */
 419         for (res = LQUOTA_FIRST_RES; res < LQUOTA_LAST_RES; res++) {
 420                 rc = qmt_pool_alloc(env, qmt, 0, res);
 421                 if (rc)
 422                         break;
 423         }
 424
 425         if (rc)
 426                 qmt_pool_fini(env, qmt);
 427
 428         RETURN(rc);
 429 }
 430
 431 static int qmt_slv_cnt(const struct lu_env *env, struct lu_fid *glb_fid,
 432                        char *slv_name, struct lu_fid *slv_fid, void *arg)
 433 {
 434         int *nr = arg;
 435
 436         /* one more slave */
 437         (*nr)++;
 438
 439         return 0;
 440 }
 441
 442 /*
 443  * Set up on-disk index files associated with each pool.
 444  *
 445  * \param env - is the environment passed by the caller
 446  * \param qmt - is the quota master target for which we have to initializa the
 447  *              pool configuration
 448  * \param qmt_root - is the on-disk directory created for the QMT.
 449  *
 450  * \retval - 0 on success, appropriate error on failure
 451  */
 452 int qmt_pool_prepare(const struct lu_env *env, struct qmt_device *qmt,
 453                      struct dt_object *qmt_root)
 454 {
 455         struct qmt_thread_info  *qti = qmt_info(env);
 456         struct lquota_glb_rec   *rec = &qti->qti_glb_rec;
 457         struct qmt_pool_info    *pool;
 458         struct dt_device        *dev = NULL;
 459         dt_obj_version_t         version;
 460         cfs_list_t              *pos;
 461         int                      rc = 0, qtype;
 462         ENTRY;
 463
 464         LASSERT(qmt->qmt_pool_hash != NULL);
 465
 466         /* iterate over each pool in the hash and allocate a quota site for each
 467          * one. This involves creating a global index file on disk */
 468         cfs_list_for_each(pos, &qmt->qmt_pool_list) {
 469                 struct dt_object        *obj;
 470                 int                      pool_type, pool_id;
 471                 struct lquota_entry     *lqe;
 472
 473                 pool = cfs_list_entry(pos, struct qmt_pool_info,
 474                                       qpi_linkage);
 475
 476                 pool_id   = pool->qpi_key & 0x0000ffff;
 477                 pool_type = pool->qpi_key >> 16;
 478                 if (dev == NULL)
 479                         dev = pool->qpi_qmt->qmt_child;
 480
 481                 /* allocate directory for this pool */
 482                 sprintf(qti->qti_buf, "%s-0x%x", RES_NAME(pool_type), pool_id);
 483                 obj = lquota_disk_dir_find_create(env, qmt->qmt_child, qmt_root,
 484                                                   qti->qti_buf);
 485                 if (IS_ERR(obj))
 486                         RETURN(PTR_ERR(obj));
 487                 pool->qpi_root = obj;
 488
 489                 for (qtype = 0; qtype < MAXQUOTAS; qtype++) {
 490                         /* Generating FID of global index in charge of storing
 491                          * settings for this quota type */
 492                         lquota_generate_fid(&qti->qti_fid, pool_id, pool_type,
 493                                             qtype);
 494
 495                         /* open/create the global index file for this quota
 496                          * type */
 497                         obj = lquota_disk_glb_find_create(env, dev,
 498                                                           pool->qpi_root,
 499                                                           &qti->qti_fid, false);
 500                         if (IS_ERR(obj)) {
 501                                 rc = PTR_ERR(obj);
 502                                 CERROR("%s: failed to create glb index copy for"
 503                                        " %s type (%d)\n", qmt->qmt_svname,
 504                                        QTYPE_NAME(qtype), rc);
 505                                 RETURN(rc);
 506                         }
 507
 508                         pool->qpi_glb_obj[qtype] = obj;
 509
 510                         version = dt_version_get(env, obj);
 511                         /* set default grace time for newly created index */
 512                         if (version == 0) {
 513                                 rec->qbr_hardlimit = 0;
 514                                 rec->qbr_softlimit = 0;
 515                                 rec->qbr_granted = 0;
 516                                 rec->qbr_time = pool_type == LQUOTA_RES_MD ?
 517                                         MAX_IQ_TIME : MAX_DQ_TIME;
 518
 519                                 rc = lquota_disk_write_glb(env, obj, 0, rec);
 520                                 if (rc) {
 521                                         CERROR("%s: failed to set default "
 522                                                "grace time for %s type (%d)\n",
 523                                                qmt->qmt_svname,
 524                                                QTYPE_NAME(qtype), rc);
 525                                         RETURN(rc);
 526                                 }
 527
 528                                 rc = lquota_disk_update_ver(env, dev, obj, 1);
 529                                 if (rc) {
 530                                         CERROR("%s: failed to set initial "
 531                                                "version for %s type (%d)\n",
 532                                                qmt->qmt_svname,
 533                                                QTYPE_NAME(qtype), rc);
 534                                         RETURN(rc);
 535                                 }
 536                         }
 537
 538                         /* create quota entry site for this quota type */
 539                         pool->qpi_site[qtype] = lquota_site_alloc(env, pool,
 540                                                                   true, qtype,
 541                                                                   &qmt_lqe_ops);
 542                         if (IS_ERR(pool->qpi_site[qtype])) {
 543                                 rc = PTR_ERR(pool->qpi_site[qtype]);
 544                                 CERROR("%s: failed to create site for %s type "
 545                                        "(%d)\n", qmt->qmt_svname,
 546                                        QTYPE_NAME(qtype), rc);
 547                                 RETURN(rc);
 548                         }
 549
 550                         /* count number of slaves which already connected to
 551                          * the master in the past */
 552                         pool->qpi_slv_nr[qtype] = 0;
 553                         rc = lquota_disk_for_each_slv(env, pool->qpi_root,
 554                                                       &qti->qti_fid,
 555                                                       qmt_slv_cnt,
 556                                                       &pool->qpi_slv_nr[qtype]);
 557                         if (rc) {
 558                                 CERROR("%s: failed to scan & count slave "
 559                                        "indexes for %s type (%d)\n",
 560                                        qmt->qmt_svname, QTYPE_NAME(qtype), rc);
 561                                 RETURN(rc);
 562                         }
 563
 564                         /* Global grace time is stored in quota settings of
 565                          * ID 0. */
 566                         qti->qti_id.qid_uid = 0;
 567
 568                         /* look-up quota entry storing grace time */
 569                         lqe = lqe_locate(env, pool->qpi_site[qtype],
 570                                          &qti->qti_id);
 571                         if (IS_ERR(lqe))
 572                                 RETURN(PTR_ERR(lqe));
 573                         pool->qpi_grace_lqe[qtype] = lqe;
 574 #ifdef LPROCFS
 575                         /* add procfs file to dump the global index, mostly for
 576                          * debugging purpose */
 577                         sprintf(qti->qti_buf, "glb-%s", QTYPE_NAME(qtype));
 578                         rc = lprocfs_seq_create(pool->qpi_proc, qti->qti_buf,
 579                                                 0444, &lprocfs_quota_seq_fops,
 580                                                 obj);
 581                         if (rc)
 582                                 CWARN("%s: Error adding procfs file for global"
 583                                       "quota index "DFID", rc:%d\n",
 584                                       qmt->qmt_svname, PFID(&qti->qti_fid), rc);
 585 #endif
 586                 }
 587         }
 588
 589         RETURN(0);
 590 }
 591
 592 /*
 593  * Handle new slave connection. Called when a slave enqueues the global quota
 594  * lock at the beginning of the reintegration procedure.
 595  *
 596  * \param env - is the environment passed by the caller
 597  * \parap qmt - is the quota master target handling this request
 598  * \param glb_fid - is the fid of the global index file
 599  * \param slv_fid - is the fid of the newly created slave index file
 600  * \param slv_ver - is the current version of the slave index file
 601  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 602  *                  target
 603  *
 604  * \retval - 0 on success, appropriate error on failure
 605  */
 606 int qmt_pool_new_conn(const struct lu_env *env, struct qmt_device *qmt,
 607                       struct lu_fid *glb_fid, struct lu_fid *slv_fid,
 608                       __u64 *slv_ver, struct obd_uuid *uuid)
 609 {
 610         struct qmt_pool_info    *pool;
 611         struct dt_object        *slv_obj;
 612         int                      pool_id, pool_type, qtype;
 613         bool                     created = false;
 614         int                      rc = 0;
 615
 616         /* extract pool info from global index FID */
 617         rc = lquota_extract_fid(glb_fid, &pool_id, &pool_type, &qtype);
 618         if (rc)
 619                 RETURN(rc);
 620
 621         /* look-up pool in charge of this global index FID */
 622         pool = qmt_pool_lookup(env, qmt, pool_id, pool_type);
 623         if (IS_ERR(pool))
 624                 RETURN(PTR_ERR(pool));
 625
 626         /* look-up slave index file */
 627         slv_obj = lquota_disk_slv_find(env, qmt->qmt_child, pool->qpi_root,
 628                                        glb_fid, uuid);
 629         if (IS_ERR(slv_obj) && PTR_ERR(slv_obj) == -ENOENT) {
 630                 /* create slave index file */
 631                 slv_obj = lquota_disk_slv_find_create(env, qmt->qmt_child,
 632                                                       pool->qpi_root, glb_fid,
 633                                                       uuid, false);
 634                 created = true;
 635         }
 636         if (IS_ERR(slv_obj)) {
 637                 rc = PTR_ERR(slv_obj);
 638                 CERROR("%s: failed to create quota slave index file for %s (%d)"
 639                        "\n", qmt->qmt_svname, obd_uuid2str(uuid), rc);
 640                 GOTO(out, rc);
 641         }
 642
 643         /* retrieve slave fid & current object version */
 644         memcpy(slv_fid, lu_object_fid(&slv_obj->do_lu), sizeof(*slv_fid));
 645         *slv_ver = dt_version_get(env, slv_obj);
 646         lu_object_put(env, &slv_obj->do_lu);
 647         if (created)
 648                 pool->qpi_slv_nr[qtype]++;
 649 out:
 650         qpi_putref(env, pool);
 651         RETURN(rc);
 652 }
 653
 654 /*
 655  * Look-up a lquota_entry in the pool hash and allocate it if not found.
 656  *
 657  * \param env - is the environment passed by the caller
 658  * \param qmt - is the quota master target for which we have to initializa the
 659  *              pool configuration
 660  * \param pool_id   - is the 16-bit identifier of the pool
 661  * \param pool_type - is the pool type, either LQUOTA_RES_MD or LQUOTA_RES_DT.
 662  * \param qtype     - is the quota type, either user or group.
 663  * \param qid       - is the quota ID to look-up
 664  *
 665  * \retval - valid pointer to lquota entry on success, appropriate error on
 666  *           failure
 667  */
 668 struct lquota_entry *qmt_pool_lqe_lookup(const struct lu_env *env,
 669                                          struct qmt_device *qmt,
 670                                          int pool_id, int pool_type,
 671                                          int qtype, union lquota_id *qid)
 672 {
 673         struct qmt_pool_info    *pool;
 674         struct lquota_entry     *lqe;
 675         ENTRY;
 676
 677         /* look-up pool responsible for this global index FID */
 678         pool = qmt_pool_lookup(env, qmt, pool_id, pool_type);
 679         if (IS_ERR(pool))
 680                 RETURN((void *)pool);
 681
 682         if (qid->qid_uid == 0) {
 683                 /* caller wants to access grace time, no need to look up the
 684                  * entry since we keep a reference on ID 0 all the time */
 685                 lqe = pool->qpi_grace_lqe[qtype];
 686                 lqe_getref(lqe);
 687                 GOTO(out, 0);
 688         }
 689
 690         /* now that we have the pool, let's look-up the quota entry in the
 691          * right quota site */
 692         lqe = lqe_locate(env, pool->qpi_site[qtype], qid);
 693 out:
 694         qpi_putref(env, pool);
 695         RETURN(lqe);
 696 }