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 inline int qmt_sarr_pool_init(struct qmt_pool_info *qpi);
  58 static inline int qmt_sarr_pool_add(struct qmt_pool_info *qpi,
  59                                     int idx, int min);
  60 static inline int qmt_sarr_pool_rem(struct qmt_pool_info *qpi, int idx);
  61 static inline void qmt_sarr_pool_free(struct qmt_pool_info *qpi);
  62 static inline int qmt_sarr_check_idx(struct qmt_pool_info *qpi, int idx);
  63 static inline void qmt_stop_pool_recalc(struct qmt_pool_info *qpi);
  64
  65 /*
  66  * Static helper functions not used outside the scope of this file
  67  */
  68
  69 static inline void qpi_putref_locked(struct qmt_pool_info *pool)
  70 {
  71         LASSERT(atomic_read(&pool->qpi_ref) > 1);
  72         atomic_dec(&pool->qpi_ref);
  73 }
  74
  75 /* some procfs helpers */
  76 static int qpi_state_seq_show(struct seq_file *m, void *data)
  77 {
  78         struct qmt_pool_info    *pool = m->private;
  79         int                      type;
  80
  81         LASSERT(pool != NULL);
  82         if (unlikely(!test_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags)))
  83                 return -ENOENT;
  84
  85         seq_printf(m, "pool:\n"
  86                    "    id: %u\n"
  87                    "    type: %s\n"
  88                    "    ref: %d\n"
  89                    "    least qunit: %lu\n",
  90                    0,
  91                    RES_NAME(pool->qpi_rtype),
  92                    atomic_read(&pool->qpi_ref),
  93                    pool->qpi_least_qunit);
  94
  95         for (type = 0; type < LL_MAXQUOTAS; type++)
  96                 seq_printf(m, "    %s:\n"
  97                            "        #slv: %d\n"
  98                            "        #lqe: %d\n",
  99                            qtype_name(type),
 100                            qpi_slv_nr(pool, type),
 101                     atomic_read(&pool->qpi_site[type]->lqs_hash->hs_count));
 102
 103         return 0;
 104 }
 105 LPROC_SEQ_FOPS_RO(qpi_state);
 106
 107 static int qpi_soft_least_qunit_seq_show(struct seq_file *m, void *data)
 108 {
 109         struct qmt_pool_info    *pool = m->private;
 110         LASSERT(pool != NULL);
 111         if (unlikely(!test_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags)))
 112                 return -ENOENT;
 113
 114         seq_printf(m, "%lu\n", pool->qpi_soft_least_qunit);
 115         return 0;
 116 }
 117
 118 static ssize_t
 119 qpi_soft_least_qunit_seq_write(struct file *file, const char __user *buffer,
 120                                size_t count, loff_t *off)
 121 {
 122         struct seq_file *m = file->private_data;
 123         struct qmt_pool_info *pool = m->private;
 124         long long least_qunit, qunit;
 125         int rc;
 126
 127         LASSERT(pool != NULL);
 128         if (unlikely(!test_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags)))
 129                 return -ENOENT;
 130
 131         /* Not tuneable for inode limit */
 132         if (pool->qpi_rtype != LQUOTA_RES_DT)
 133                 return -EINVAL;
 134
 135         rc = kstrtoll_from_user(buffer, count, 0, &least_qunit);
 136         if (rc)
 137                 return rc;
 138
 139         /* Miminal qpi_soft_least_qunit */
 140         qunit = pool->qpi_least_qunit << 2;
 141         /* The value must be power of miminal qpi_soft_least_qunit, see
 142          * how the qunit is adjusted in qmt_adjust_qunit(). */
 143         while (qunit > 0 && qunit < least_qunit)
 144                 qunit <<= 2;
 145         if (qunit <= 0)
 146                 qunit = INT_MAX & ~3;
 147
 148         pool->qpi_soft_least_qunit = qunit;
 149         return count;
 150 }
 151 LPROC_SEQ_FOPS(qpi_soft_least_qunit);
 152
 153 static struct lprocfs_vars lprocfs_quota_qpi_vars[] = {
 154         { .name =       "info",
 155           .fops =       &qpi_state_fops },
 156         { .name =       "soft_least_qunit",
 157           .fops =       &qpi_soft_least_qunit_fops },
 158         { NULL }
 159 };
 160
 161 /*
 162  * Allocate a new qmt_pool_info structure and add it to qmt_pool_list.
 163  *
 164  * \param env       - is the environment passed by the caller
 165  * \param qmt       - is the quota master target
 166  * \param pool_type - is the resource type of this pool instance, either
 167  *                    LQUOTA_RES_MD or LQUOTA_RES_DT.
 168  *
 169  * \retval - 0 on success, appropriate error on failure
 170  */
 171 static int qmt_pool_alloc(const struct lu_env *env, struct qmt_device *qmt,
 172                           char *pool_name, int pool_type)
 173 {
 174         struct qmt_thread_info  *qti = qmt_info(env);
 175         struct qmt_pool_info    *pool;
 176         int                      rc = 0;
 177         ENTRY;
 178
 179         OBD_ALLOC_PTR(pool);
 180         if (pool == NULL)
 181                 RETURN(-ENOMEM);
 182         INIT_LIST_HEAD(&pool->qpi_linkage);
 183         init_rwsem(&pool->qpi_recalc_sem);
 184
 185         pool->qpi_rtype = pool_type;
 186         pool->qpi_flags = 0;
 187
 188         /* initialize refcount to 1, hash table will then grab an additional
 189          * reference */
 190         atomic_set(&pool->qpi_ref, 1);
 191
 192         /* set up least qunit size to use for this pool */
 193         pool->qpi_least_qunit = LQUOTA_LEAST_QUNIT(pool_type);
 194         if (pool_type == LQUOTA_RES_DT)
 195                 pool->qpi_soft_least_qunit = pool->qpi_least_qunit << 2;
 196         else
 197                 pool->qpi_soft_least_qunit = pool->qpi_least_qunit;
 198
 199         /* grab reference on master target that this pool belongs to */
 200         lu_device_get(qmt2lu_dev(qmt));
 201         lu_ref_add(&qmt2lu_dev(qmt)->ld_reference, "pool", pool);
 202         pool->qpi_qmt = qmt;
 203
 204         /* create pool proc directory */
 205         snprintf(qti->qti_buf, LQUOTA_NAME_MAX, "%s-%s",
 206                  RES_NAME(pool_type), pool_name);
 207         strncpy(pool->qpi_name, pool_name, QPI_MAXNAME);
 208         pool->qpi_proc = lprocfs_register(qti->qti_buf, qmt->qmt_proc,
 209                                           lprocfs_quota_qpi_vars, pool);
 210         if (IS_ERR(pool->qpi_proc)) {
 211                 rc = PTR_ERR(pool->qpi_proc);
 212                 CERROR("%s: failed to create proc entry for pool %s (%d)\n",
 213                        qmt->qmt_svname, qti->qti_buf, rc);
 214                 pool->qpi_proc = NULL;
 215                 GOTO(out, rc);
 216         }
 217
 218         rc = qmt_sarr_pool_init(pool);
 219         if (rc)
 220                 GOTO(out, rc);
 221
 222         /* add to qmt pool list */
 223         down_write(&qmt->qmt_pool_lock);
 224         list_add_tail(&pool->qpi_linkage, &qmt->qmt_pool_list);
 225         up_write(&qmt->qmt_pool_lock);
 226         EXIT;
 227 out:
 228         if (rc)
 229                 /* this frees the pool structure since refcount is equal to 1 */
 230                 qpi_putref(env, pool);
 231         return rc;
 232 }
 233
 234 /*
 235  * Delete a qmt_pool_info instance and all structures associated.
 236  *
 237  * \param env  - is the environment passed by the caller
 238  * \param pool - is the qmt_pool_info structure to free
 239  */
 240 void qmt_pool_free(const struct lu_env *env, struct qmt_pool_info *pool)
 241 {
 242         struct  qmt_device *qmt = pool->qpi_qmt;
 243         int     qtype;
 244         ENTRY;
 245
 246         /* remove from list */
 247         down_write(&qmt->qmt_pool_lock);
 248         list_del_init(&pool->qpi_linkage);
 249         up_write(&qmt->qmt_pool_lock);
 250
 251         if (atomic_read(&pool->qpi_ref) > 0)
 252                 RETURN_EXIT;
 253
 254         qmt_stop_pool_recalc(pool);
 255         qmt_sarr_pool_free(pool);
 256
 257         /* release proc entry */
 258         if (pool->qpi_proc) {
 259                 lprocfs_remove(&pool->qpi_proc);
 260                 pool->qpi_proc = NULL;
 261         }
 262
 263         /* release per-quota type site used to manage quota entries as well as
 264          * references to global index files */
 265         for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
 266                 /* release lqe storing grace time */
 267                 if (pool->qpi_grace_lqe[qtype] != NULL)
 268                         lqe_putref(pool->qpi_grace_lqe[qtype]);
 269
 270                 /* release site */
 271                 if (pool->qpi_site[qtype] != NULL &&
 272                     !IS_ERR(pool->qpi_site[qtype]))
 273                         lquota_site_free(env, pool->qpi_site[qtype]);
 274                 /* release reference to global index */
 275                 if (pool->qpi_glb_obj[qtype] != NULL &&
 276                     !IS_ERR(pool->qpi_glb_obj[qtype]))
 277                         dt_object_put(env, pool->qpi_glb_obj[qtype]);
 278         }
 279
 280         /* release reference on pool directory */
 281         if (pool->qpi_root != NULL && !IS_ERR(pool->qpi_root))
 282                 dt_object_put(env, pool->qpi_root);
 283
 284         /* release reference on the master target */
 285         if (pool->qpi_qmt != NULL) {
 286                 struct lu_device *ld = qmt2lu_dev(pool->qpi_qmt);
 287
 288                 lu_ref_del(&ld->ld_reference, "pool", pool);
 289                 lu_device_put(ld);
 290                 pool->qpi_qmt = NULL;
 291         }
 292
 293         LASSERT(list_empty(&pool->qpi_linkage));
 294         OBD_FREE_PTR(pool);
 295 }
 296
 297 static inline void qti_pools_init(const struct lu_env *env)
 298 {
 299         struct qmt_thread_info  *qti = qmt_info(env);
 300
 301         qti->qti_pools_cnt = 0;
 302         qti->qti_pools_num = QMT_MAX_POOL_NUM;
 303 }
 304
 305 #define qti_pools(qti)  (qti->qti_pools_num > QMT_MAX_POOL_NUM ? \
 306                                 qti->qti_pools : qti->qti_pools_small)
 307 #define qti_pools_env(env) \
 308         (qmt_info(env)->qti_pools_num > QMT_MAX_POOL_NUM ? \
 309                 qmt_info(env)->qti_pools : qmt_info(env)->qti_pools_small)
 310 #define qti_pools_cnt(env)      (qmt_info(env)->qti_pools_cnt)
 311
 312 static inline int qti_pools_add(const struct lu_env *env,
 313                                 struct qmt_pool_info *qpi)
 314 {
 315         struct qmt_thread_info  *qti = qmt_info(env);
 316         struct qmt_pool_info    **pools = qti->qti_pools;
 317
 318         pools = qti_pools(qti);
 319         LASSERTF(qti->qti_pools_num >= QMT_MAX_POOL_NUM,
 320                  "Forgot init? %p\n", qti);
 321
 322         if (qti->qti_pools_cnt >= qti->qti_pools_num) {
 323                 OBD_ALLOC(pools, sizeof(qpi) * qti->qti_pools_num * 2);
 324                 if (!pools)
 325                         return -ENOMEM;
 326                 memcpy(pools, qti_pools(qti), qti->qti_pools_cnt * sizeof(qpi));
 327                 /* Don't need to free, if it is the very 1st allocation */
 328                 if (qti->qti_pools_num > QMT_MAX_POOL_NUM)
 329                         OBD_FREE(qti->qti_pools,
 330                                  qti->qti_pools_num * sizeof(qpi));
 331                 qti->qti_pools = pools;
 332                 qti->qti_pools_num *= 2;
 333         }
 334
 335         qpi_getref(qpi);
 336         /* Take this to protect pool's lqes against changing by
 337          * recalculation thread. This would be unlocked at
 338          * qti_pools_fini. */
 339         down_read(&qpi->qpi_recalc_sem);
 340         if (qmt_pool_global(qpi) && qti_pools_cnt(env) > 0) {
 341                 pools[qti->qti_pools_cnt++] = pools[0];
 342                 /* Store global pool always at index 0 */
 343                 pools[0] = qpi;
 344         } else {
 345                 pools[qti->qti_pools_cnt++] = qpi;
 346         }
 347
 348         CDEBUG(D_QUOTA, "Pool %s is added, pools %p qti_pools %p pool_num %d\n",
 349                qpi->qpi_name, pools, qti->qti_pools, qti->qti_pools_cnt);
 350
 351         return 0;
 352 }
 353
 354 static inline void qti_pools_fini(const struct lu_env *env)
 355 {
 356         struct qmt_thread_info  *qti = qmt_info(env);
 357         struct qmt_pool_info    **pools = qti->qti_pools;
 358         int i;
 359
 360         LASSERT(qti->qti_pools_cnt > 0);
 361
 362         pools = qti_pools(qti);
 363         for (i = 0; i < qti->qti_pools_cnt; i++) {
 364                 up_read(&pools[i]->qpi_recalc_sem);
 365                 qpi_putref(env, pools[i]);
 366         }
 367
 368         if (qti->qti_pools_num > QMT_MAX_POOL_NUM)
 369                 OBD_FREE(qti->qti_pools,
 370                          qti->qti_pools_num * sizeof(struct qmt_pool_info *));
 371 }
 372
 373 /*
 374  * Look-up a pool in a list based on the type.
 375  *
 376  * \param env   - is the environment passed by the caller
 377  * \param qmt   - is the quota master target
 378  * \param rtype - is the type of this pool, either LQUOTA_RES_MD or
 379  *                    LQUOTA_RES_DT.
 380  * \param pool_name - is the pool name to search for
 381  * \param idx   - OST or MDT index to search for. When it is >= 0, function
 382  *              returns array with pointers to all pools that include
 383  *              targets with requested index.
 384  * \param add   - add to qti_pool_arr if true
 385  */
 386 struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
 387                                              struct qmt_device *qmt,
 388                                              int rtype,
 389                                              char *pool_name,
 390                                              int idx, bool add)
 391 {
 392         struct qmt_pool_info    *pos, *pool;
 393         int rc = 0;
 394         ENTRY;
 395
 396         down_read(&qmt->qmt_pool_lock);
 397         if (list_empty(&qmt->qmt_pool_list)) {
 398                 up_read(&qmt->qmt_pool_lock);
 399                 RETURN(ERR_PTR(-ENOENT));
 400         }
 401
 402         CDEBUG(D_QUOTA, "type %d name %s index %d\n",
 403                rtype, pool_name ?: "<none>", idx);
 404         /* Now just find a pool with correct type in a list. Further we need
 405          * to go through the list and find a pool that includes requested OST
 406          * or MDT. Possibly this would return a list of pools that includes
 407          * needed target(OST/MDT). */
 408         pool = NULL;
 409         if (idx == -1 && !pool_name)
 410                 pool_name = GLB_POOL_NAME;
 411
 412         list_for_each_entry(pos, &qmt->qmt_pool_list, qpi_linkage) {
 413                 if (pos->qpi_rtype != rtype)
 414                         continue;
 415
 416                 if (idx >= 0 && !qmt_sarr_check_idx(pos, idx)) {
 417                         rc = qti_pools_add(env, pos);
 418                         if (rc)
 419                                 break;
 420                         continue;
 421                 }
 422
 423                 if (pool_name && !strncmp(pool_name, pos->qpi_name,
 424                                           LOV_MAXPOOLNAME)) {
 425                         pool = pos;
 426                         if (add) {
 427                                 rc = qti_pools_add(env, pos);
 428                                 if (rc)
 429                                         break;
 430                         } else {
 431                                 qpi_getref(pool);
 432                         }
 433                         break;
 434                 }
 435         }
 436         up_read(&qmt->qmt_pool_lock);
 437
 438         if (rc)
 439                 GOTO(out_err, rc);
 440
 441         if (idx >= 0 && qti_pools_cnt(env))
 442                 pool = qti_pools_env(env)[0];
 443
 444         RETURN(pool ? : ERR_PTR(-ENOENT));
 445 out_err:
 446         CERROR("%s: cannot add pool %s: err = %d\n",
 447                 qmt->qmt_svname, pos->qpi_name, rc);
 448         return ERR_PTR(rc);
 449 }
 450
 451 /*
 452  * Functions implementing the pool API, used by the qmt handlers
 453  */
 454
 455 /*
 456  * Destroy all pools which are still in the pool list.
 457  *
 458  * \param env - is the environment passed by the caller
 459  * \param qmt - is the quota master target
 460  *
 461  */
 462 void qmt_pool_fini(const struct lu_env *env, struct qmt_device *qmt)
 463 {
 464         struct qmt_pool_info *pool, *tmp;
 465         ENTRY;
 466
 467         /* parse list of pool and destroy each element */
 468         list_for_each_entry_safe(pool, tmp, &qmt->qmt_pool_list, qpi_linkage) {
 469                 /* release extra reference taken in qmt_pool_alloc */
 470                 qpi_putref(env, pool);
 471         }
 472         LASSERT(list_empty(&qmt->qmt_pool_list));
 473
 474         EXIT;
 475 }
 476
 477 /*
 478  * Initialize pool configure for the quota master target. For now, we only
 479  * support the default data (i.e. all OSTs) and metadata (i.e. all the MDTs)
 480  * pool which are instantiated in this function.
 481  *
 482  * \param env - is the environment passed by the caller
 483  * \param qmt - is the quota master target for which we have to initialize the
 484  *              pool configuration
 485  *
 486  * \retval - 0 on success, appropriate error on failure
 487  */
 488 int qmt_pool_init(const struct lu_env *env, struct qmt_device *qmt)
 489 {
 490         int     res, rc = 0;
 491         ENTRY;
 492
 493         INIT_LIST_HEAD(&qmt->qmt_pool_list);
 494         init_rwsem(&qmt->qmt_pool_lock);
 495
 496         /* Instantiate pool master for the default data and metadata pool.
 497          * This code will have to be revisited once we support quota on
 498          * non-default pools */
 499         for (res = LQUOTA_FIRST_RES; res < LQUOTA_LAST_RES; res++) {
 500                 rc = qmt_pool_alloc(env, qmt, GLB_POOL_NAME, res);
 501                 if (rc)
 502                         break;
 503         }
 504
 505         if (rc)
 506                 qmt_pool_fini(env, qmt);
 507
 508         RETURN(rc);
 509 }
 510
 511 static int qmt_slv_cnt(const struct lu_env *env, struct lu_fid *glb_fid,
 512                        char *slv_name, struct lu_fid *slv_fid, void *arg)
 513 {
 514         struct obd_uuid uuid;
 515         int (*nr)[QMT_STYPE_CNT][LL_MAXQUOTAS] = arg;
 516         int stype, qtype;
 517         int rc;
 518
 519         rc = lquota_extract_fid(glb_fid, NULL, &qtype);
 520         LASSERT(!rc);
 521
 522         obd_str2uuid(&uuid, slv_name);
 523         stype = qmt_uuid2idx(&uuid, NULL);
 524         if (stype < 0)
 525                 return stype;
 526         /* one more slave */
 527         (*nr)[stype][qtype]++;
 528         CDEBUG(D_QUOTA, "slv_name %s stype %d qtype %d nr %d\n",
 529                         slv_name, stype, qtype, (*nr)[stype][qtype]);
 530
 531         return 0;
 532 }
 533
 534 /*
 535  * Set up on-disk index files associated with each pool.
 536  *
 537  * \param env - is the environment passed by the caller
 538  * \param qmt - is the quota master target for which we have to initialize the
 539  *              pool configuration
 540  * \param qmt_root - is the on-disk directory created for the QMT.
 541  * \param name - is the pool name that we need to setup. Setup all pools
 542  *               in qmt_pool_list when name is NULL.
 543  *
 544  * \retval - 0 on success, appropriate error on failure
 545  */
 546 int qmt_pool_prepare(const struct lu_env *env, struct qmt_device *qmt,
 547                      struct dt_object *qmt_root, char *name)
 548 {
 549         struct qmt_thread_info  *qti = qmt_info(env);
 550         struct lquota_glb_rec   *rec = &qti->qti_glb_rec;
 551         struct qmt_pool_info    *pool;
 552         struct dt_device        *dev = NULL;
 553         dt_obj_version_t         version;
 554         struct list_head        *pos;
 555         int                      rc = 0, i, qtype;
 556         ENTRY;
 557
 558         /* iterate over each pool in the list and allocate a quota site for each
 559          * one. This involves creating a global index file on disk */
 560         list_for_each(pos, &qmt->qmt_pool_list) {
 561                 struct dt_object        *obj;
 562                 struct lquota_entry     *lqe;
 563                 char                    *pool_name;
 564                 int                      rtype;
 565
 566                 pool = list_entry(pos, struct qmt_pool_info,
 567                                   qpi_linkage);
 568
 569                 pool_name = pool->qpi_name;
 570                 if (name && strncmp(pool_name, name, LOV_MAXPOOLNAME))
 571                         continue;
 572                 rtype = pool->qpi_rtype;
 573                 if (dev == NULL)
 574                         dev = pool->qpi_qmt->qmt_child;
 575
 576                 /* allocate directory for this pool */
 577                 snprintf(qti->qti_buf, LQUOTA_NAME_MAX, "%s-%s",
 578                          RES_NAME(rtype), pool_name);
 579                 obj = lquota_disk_dir_find_create(env, qmt->qmt_child, qmt_root,
 580                                                   qti->qti_buf);
 581                 if (IS_ERR(obj))
 582                         RETURN(PTR_ERR(obj));
 583                 pool->qpi_root = obj;
 584
 585                 for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
 586                         /* Generating FID of global index in charge of storing
 587                          * settings for this quota type */
 588                         lquota_generate_fid(&qti->qti_fid, rtype, qtype);
 589
 590                         /* open/create the global index file for this quota
 591                          * type. If name is set, it means we came here from
 592                          * qmt_pool_new and can create glb index with a
 593                          * local generated FID. */
 594                         obj = lquota_disk_glb_find_create(env, dev,
 595                                                           pool->qpi_root,
 596                                                           &qti->qti_fid,
 597                                                           name ? true : false);
 598                         if (IS_ERR(obj)) {
 599                                 rc = PTR_ERR(obj);
 600                                 CERROR("%s: failed to create glb index copy for %s type: rc = %d\n",
 601                                        qmt->qmt_svname, qtype_name(qtype), rc);
 602                                 RETURN(rc);
 603                         }
 604
 605                         pool->qpi_glb_obj[qtype] = obj;
 606
 607                         version = dt_version_get(env, obj);
 608                         /* set default grace time for newly created index */
 609                         if (version == 0) {
 610                                 rec->qbr_hardlimit = 0;
 611                                 rec->qbr_softlimit = 0;
 612                                 rec->qbr_granted = 0;
 613                                 rec->qbr_time = rtype == LQUOTA_RES_MD ?
 614                                         MAX_IQ_TIME : MAX_DQ_TIME;
 615
 616                                 rc = lquota_disk_write_glb(env, obj, 0, rec);
 617                                 if (rc) {
 618                                         CERROR("%s: failed to set default grace time for %s type: rc = %d\n",
 619                                                qmt->qmt_svname, qtype_name(qtype), rc);
 620                                         RETURN(rc);
 621                                 }
 622
 623                                 rc = lquota_disk_update_ver(env, dev, obj, 1);
 624                                 if (rc) {
 625                                         CERROR("%s: failed to set initial version for %s type: rc = %d\n",
 626                                                qmt->qmt_svname, qtype_name(qtype), rc);
 627                                         RETURN(rc);
 628                                 }
 629                         }
 630
 631                         /* create quota entry site for this quota type */
 632                         pool->qpi_site[qtype] = lquota_site_alloc(env, pool,
 633                                                                   true, qtype,
 634                                                                   &qmt_lqe_ops);
 635                         if (IS_ERR(pool->qpi_site[qtype])) {
 636                                 rc = PTR_ERR(pool->qpi_site[qtype]);
 637                                 CERROR("%s: failed to create site for %s type: rc = %d\n",
 638                                        qmt->qmt_svname, qtype_name(qtype), rc);
 639                                 RETURN(rc);
 640                         }
 641
 642                         /* count number of slaves which already connected to
 643                          * the master in the past */
 644                         for (i = 0; i < QMT_STYPE_CNT; i++)
 645                                 pool->qpi_slv_nr[i][qtype] = 0;
 646
 647                         rc = lquota_disk_for_each_slv(env, pool->qpi_root,
 648                                                       &qti->qti_fid,
 649                                                       qmt_slv_cnt,
 650                                                       &pool->qpi_slv_nr);
 651                         if (rc) {
 652                                 CERROR("%s: failed to scan & count slave indexes for %s type: rc = %d\n",
 653                                        qmt->qmt_svname, qtype_name(qtype), rc);
 654                                 RETURN(rc);
 655                         }
 656
 657                         /* Global grace time is stored in quota settings of
 658                          * ID 0. */
 659                         qti->qti_id.qid_uid = 0;
 660
 661                         /* look-up quota entry storing grace time */
 662                         lqe = lqe_locate(env, pool->qpi_site[qtype],
 663                                          &qti->qti_id);
 664                         if (IS_ERR(lqe))
 665                                 RETURN(PTR_ERR(lqe));
 666                         pool->qpi_grace_lqe[qtype] = lqe;
 667 #ifdef CONFIG_PROC_FS
 668                         /* add procfs file to dump the global index, mostly for
 669                          * debugging purpose */
 670                         snprintf(qti->qti_buf, MTI_NAME_MAXLEN,
 671                                  "glb-%s", qtype_name(qtype));
 672                         rc = lprocfs_seq_create(pool->qpi_proc, qti->qti_buf,
 673                                                 0444, &lprocfs_quota_seq_fops,
 674                                                 obj);
 675                         if (rc)
 676                                 CWARN("%s: Error adding procfs file for global quota index "DFID": rc = %d\n",
 677                                       qmt->qmt_svname, PFID(&qti->qti_fid), rc);
 678 #endif
 679                 }
 680                 set_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags);
 681                 if (name)
 682                         break;
 683         }
 684
 685         RETURN(0);
 686 }
 687
 688 /*
 689  * Handle new slave connection. Called when a slave enqueues the global quota
 690  * lock at the beginning of the reintegration procedure.
 691  *
 692  * \param env - is the environment passed by the caller
 693  * \parap qmt - is the quota master target handling this request
 694  * \param glb_fid - is the fid of the global index file
 695  * \param slv_fid - is the fid of the newly created slave index file
 696  * \param slv_ver - is the current version of the slave index file
 697  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 698  *                  target
 699  *
 700  * \retval - 0 on success, appropriate error on failure
 701  */
 702 int qmt_pool_new_conn(const struct lu_env *env, struct qmt_device *qmt,
 703                       struct lu_fid *glb_fid, struct lu_fid *slv_fid,
 704                       __u64 *slv_ver, struct obd_uuid *uuid)
 705 {
 706         struct qmt_pool_info    *pool;
 707         struct dt_object        *slv_obj;
 708         int                      pool_type, qtype, stype;
 709         bool                     created = false;
 710         int                      idx, i, rc = 0;
 711
 712         stype = qmt_uuid2idx(uuid, &idx);
 713         if (stype < 0)
 714                 RETURN(stype);
 715
 716         /* extract pool info from global index FID */
 717         rc = lquota_extract_fid(glb_fid, &pool_type, &qtype);
 718         if (rc)
 719                 RETURN(rc);
 720
 721         /* look-up pool in charge of this global index FID */
 722         qti_pools_init(env);
 723         pool = qmt_pool_lookup_arr(env, qmt, pool_type, idx);
 724         if (IS_ERR(pool))
 725                 RETURN(PTR_ERR(pool));
 726
 727         /* look-up slave index file */
 728         slv_obj = lquota_disk_slv_find(env, qmt->qmt_child, pool->qpi_root,
 729                                        glb_fid, uuid);
 730         if (IS_ERR(slv_obj) && PTR_ERR(slv_obj) == -ENOENT) {
 731                 /* create slave index file */
 732                 slv_obj = lquota_disk_slv_find_create(env, qmt->qmt_child,
 733                                                       pool->qpi_root, glb_fid,
 734                                                       uuid, false);
 735                 created = true;
 736         }
 737         if (IS_ERR(slv_obj)) {
 738                 rc = PTR_ERR(slv_obj);
 739                 CERROR("%s: failed to create quota slave index file for %s (%d)"
 740                        "\n", qmt->qmt_svname, obd_uuid2str(uuid), rc);
 741                 GOTO(out, rc);
 742         }
 743
 744         /* retrieve slave fid & current object version */
 745         memcpy(slv_fid, lu_object_fid(&slv_obj->do_lu), sizeof(*slv_fid));
 746         *slv_ver = dt_version_get(env, slv_obj);
 747         dt_object_put(env, slv_obj);
 748         if (created)
 749                 for (i = 0; i < qti_pools_cnt(env); i++)
 750                         qti_pools_env(env)[i]->qpi_slv_nr[stype][qtype]++;
 751 out:
 752         qti_pools_fini(env);
 753         RETURN(rc);
 754 }
 755
 756 /*
 757  * Look-up a lquota_entry in the pool hash and allocate it if not found.
 758  *
 759  * \param env - is the environment passed by the caller
 760  * \param qmt - is the quota master target for which we have to initialize the
 761  *              pool configuration
 762  * \param pool_type - is the pool type, either LQUOTA_RES_MD or LQUOTA_RES_DT.
 763  * \param qtype     - is the quota type, either user or group.
 764  * \param qid       - is the quota ID to look-up
 765  *
 766  * \retval - valid pointer to lquota entry on success, appropriate error on
 767  *           failure
 768  */
 769 struct lquota_entry *qmt_pool_lqe_lookup(const struct lu_env *env,
 770                                          struct qmt_device *qmt,
 771                                          int pool_type, int qtype,
 772                                          union lquota_id *qid,
 773                                          char *pool_name)
 774 {
 775         struct qmt_pool_info    *pool;
 776         struct lquota_entry     *lqe;
 777         ENTRY;
 778
 779         /* look-up pool responsible for this global index FID */
 780         pool = qmt_pool_lookup_name(env, qmt, pool_type, pool_name);
 781         if (IS_ERR(pool))
 782                 RETURN(ERR_CAST(pool));
 783
 784         if (qid->qid_uid == 0) {
 785                 /* caller wants to access grace time, no need to look up the
 786                  * entry since we keep a reference on ID 0 all the time */
 787                 lqe = pool->qpi_grace_lqe[qtype];
 788                 lqe_getref(lqe);
 789                 GOTO(out, lqe);
 790         }
 791
 792         /* now that we have the pool, let's look-up the quota entry in the
 793          * right quota site */
 794         lqe = lqe_locate(env, pool->qpi_site[qtype], qid);
 795 out:
 796         qpi_putref(env, pool);
 797         RETURN(lqe);
 798 }
 799
 800 int qmt_pool_lqes_lookup(const struct lu_env *env,
 801                          struct qmt_device *qmt,
 802                          int rtype, int stype,
 803                          int qtype, union lquota_id *qid,
 804                          char *pool_name, int idx)
 805 {
 806         struct qmt_pool_info    *pool;
 807         struct lquota_entry     *lqe;
 808         int rc, i;
 809         ENTRY;
 810
 811         /* Until MDT pools are not emplemented, all MDTs belong to
 812          * global pool, thus lookup lqes only from global pool. */
 813         if (rtype == LQUOTA_RES_DT && stype == QMT_STYPE_MDT)
 814                 idx = -1;
 815
 816         qti_pools_init(env);
 817         rc = 0;
 818         /* look-up pool responsible for this global index FID */
 819         pool = qmt_pool_lookup_arr(env, qmt, rtype, idx);
 820         if (IS_ERR(pool)) {
 821                 qti_pools_fini(env);
 822                 RETURN(PTR_ERR(pool));
 823         }
 824
 825         /* now that we have the pool, let's look-up the quota entry in the
 826          * right quota site */
 827         qti_lqes_init(env);
 828         for (i = 0; i < qti_pools_cnt(env); i++) {
 829                 pool = qti_pools_env(env)[i];
 830                 lqe = lqe_locate(env, pool->qpi_site[qtype], qid);
 831                 if (IS_ERR(lqe)) {
 832                         qti_lqes_fini(env);
 833                         GOTO(out, rc = PTR_ERR(lqe));
 834                 }
 835                 qti_lqes_add(env, lqe);
 836         }
 837         LASSERT(qti_lqes_glbl(env)->lqe_is_global);
 838
 839 out:
 840         qti_pools_fini(env);
 841         RETURN(rc);
 842 }
 843
 844 static int lqes_cmp(const void *arg1, const void *arg2)
 845 {
 846         const struct lquota_entry *lqe1, *lqe2;
 847
 848         lqe1 = *(const struct lquota_entry **)arg1;
 849         lqe2 = *(const struct lquota_entry **)arg2;
 850         if (lqe1->lqe_qunit > lqe2->lqe_qunit)
 851                 return 1;
 852         if (lqe1->lqe_qunit < lqe2->lqe_qunit)
 853                 return -1;
 854         return 0;
 855 }
 856
 857 void qmt_lqes_sort(const struct lu_env *env)
 858 {
 859         sort(qti_lqes(env), qti_lqes_cnt(env), sizeof(void *), lqes_cmp, NULL);
 860         /* global lqe was moved during sorting */
 861         if (!qti_lqes_glbl(env)->lqe_is_global) {
 862                 int i;
 863                 for (i = 0; i < qti_lqes_cnt(env); i++) {
 864                         if (qti_lqes(env)[i]->lqe_is_global) {
 865                                 qti_glbl_lqe_idx(env) = i;
 866                                 break;
 867                         }
 868                 }
 869         }
 870 }
 871
 872 int qmt_pool_lqes_lookup_spec(const struct lu_env *env, struct qmt_device *qmt,
 873                               int rtype, int qtype, union lquota_id *qid)
 874 {
 875         struct qmt_pool_info    *pos;
 876         struct lquota_entry     *lqe;
 877         int rc = 0;
 878
 879         qti_lqes_init(env);
 880         down_read(&qmt->qmt_pool_lock);
 881         if (list_empty(&qmt->qmt_pool_list)) {
 882                 up_read(&qmt->qmt_pool_lock);
 883                 RETURN(-ENOENT);
 884         }
 885
 886         list_for_each_entry(pos, &qmt->qmt_pool_list, qpi_linkage) {
 887                 if (pos->qpi_rtype != rtype)
 888                         continue;
 889                 /* Don't take into account pools without slaves */
 890                 if (!qpi_slv_nr(pos, qtype))
 891                         continue;
 892                 lqe = lqe_find(env, pos->qpi_site[qtype], qid);
 893                 /* ENOENT is valid case for lqe from non global pool
 894                  * that hasn't limits, i.e. not enforced. Continue even
 895                  * in case of error - we can handle already found lqes */
 896                 if (IS_ERR_OR_NULL(lqe)) {
 897                         /* let know that something went wrong */
 898                         rc = lqe ? PTR_ERR(lqe) : -ENOENT;
 899                         continue;
 900                 }
 901                 if (!lqe->lqe_enforced) {
 902                         /* no settings for this qid_uid */
 903                         lqe_putref(lqe);
 904                         continue;
 905                 }
 906                 qti_lqes_add(env, lqe);
 907                 CDEBUG(D_QUOTA, "adding lqe %p from pool %s\n",
 908                                  lqe, pos->qpi_name);
 909         }
 910         up_read(&qmt->qmt_pool_lock);
 911         RETURN(rc);
 912 }
 913
 914 /**
 915  * Allocate a new pool for the specified device.
 916  *
 917  * Allocate a new pool_desc structure for the specified \a new_pool
 918  * device to create a pool with the given \a poolname.  The new pool
 919  * structure is created with a single reference, and is freed when the
 920  * reference count drops to zero.
 921  *
 922  * \param[in] obd       Lustre OBD device on which to add a pool iterator
 923  * \param[in] poolname  the name of the pool to be created
 924  *
 925  * \retval              0 in case of success
 926  * \retval              negative error code in case of error
 927  */
 928 int qmt_pool_new(struct obd_device *obd, char *poolname)
 929 {
 930         struct qmt_device       *qmt = lu2qmt_dev(obd->obd_lu_dev);
 931         struct qmt_pool_info *qpi;
 932         struct lu_env env;
 933         int rc;
 934         ENTRY;
 935
 936         if (strnlen(poolname, LOV_MAXPOOLNAME + 1) > LOV_MAXPOOLNAME)
 937                 RETURN(-ENAMETOOLONG);
 938
 939         rc = lu_env_init(&env, LCT_MD_THREAD);
 940         if (rc) {
 941                 CERROR("%s: can't init env: rc = %d\n", obd->obd_name, rc);
 942                 RETURN(rc);
 943         }
 944
 945         qpi = qmt_pool_lookup_name(&env, qmt, LQUOTA_RES_DT, poolname);
 946         if (!IS_ERR(qpi)) {
 947                 /* Valid case when several MDTs are mounted
 948                  * at the same node. */
 949                 CDEBUG(D_QUOTA, "pool %s already exists\n", poolname);
 950                 qpi_putref(&env, qpi);
 951                 GOTO(out_env, rc = -EEXIST);
 952         }
 953         if (PTR_ERR(qpi) != -ENOENT) {
 954                 CWARN("%s: pool %s lookup failed: rc = %ld\n",
 955                       obd->obd_name, poolname, PTR_ERR(qpi));
 956                 GOTO(out_env, rc = PTR_ERR(qpi));
 957         }
 958
 959         /* Now allocate and prepare only DATA pool.
 960          * Further when MDT pools will be ready we need to add
 961          * a cycle here and setup pools of both types. Another
 962          * approach is to find out pool of which type should be
 963          * created. */
 964         rc = qmt_pool_alloc(&env, qmt, poolname, LQUOTA_RES_DT);
 965         if (rc) {
 966                 CERROR("%s: can't alloc pool %s: rc = %d\n",
 967                        obd->obd_name, poolname, rc);
 968                 GOTO(out_env, rc);
 969         }
 970
 971         rc = qmt_pool_prepare(&env, qmt, qmt->qmt_root, poolname);
 972         if (rc) {
 973                 CERROR("%s: can't prepare pool for %s: rc = %d\n",
 974                        obd->obd_name, poolname, rc);
 975                 GOTO(out_err, rc);
 976         }
 977
 978         CDEBUG(D_QUOTA, "Quota pool "LOV_POOLNAMEF" added\n",
 979                poolname);
 980
 981         GOTO(out_env, rc);
 982 out_err:
 983         qpi = qmt_pool_lookup_name(&env, qmt, LQUOTA_RES_DT, poolname);
 984         if (!IS_ERR(qpi)) {
 985                 qpi_putref(&env, qpi);
 986                 qpi_putref(&env, qpi);
 987         }
 988 out_env:
 989         lu_env_fini(&env);
 990         return rc;
 991 }
 992
 993 static int
 994 qmt_obj_recalc(const struct lu_env *env, struct dt_object *obj,
 995                struct lquota_site *site)
 996 {
 997         struct qmt_thread_info *qti = qmt_info(env);
 998         union lquota_id *qid = &qti->qti_id;
 999         const struct dt_it_ops *iops;
1000         struct dt_key *key;
1001         struct dt_it *it;
1002         __u64 granted;
1003         int rc;
1004         ENTRY;
1005
1006         iops = &obj->do_index_ops->dio_it;
1007
1008         it = iops->init(env, obj, 0);
1009         if (IS_ERR(it)) {
1010                 CWARN("quota: initialize it for "DFID" failed: rc = %ld\n",
1011                       PFID(&qti->qti_fid), PTR_ERR(it));
1012                 RETURN(PTR_ERR(it));
1013         }
1014
1015         rc = iops->load(env, it, 0);
1016         if (rc < 0) {
1017                 CWARN("quota: load first entry for "DFID" failed: rc = %d\n",
1018                       PFID(&qti->qti_fid), rc);
1019                 GOTO(out, rc);
1020         } else if (rc == 0) {
1021                 rc = iops->next(env, it);
1022                 if (rc != 0)
1023                         GOTO(out, rc = (rc < 0) ? rc : 0);
1024         }
1025
1026         do {
1027                 struct lquota_entry *lqe;
1028
1029                 key = iops->key(env, it);
1030                 if (IS_ERR(key)) {
1031                         CWARN("quota: error key for "DFID": rc = %ld\n",
1032                               PFID(&qti->qti_fid), PTR_ERR(key));
1033                         GOTO(out, rc = PTR_ERR(key));
1034                 }
1035
1036                 /* skip the root user/group */
1037                 if (*((__u64 *)key) == 0)
1038                         goto next;
1039
1040                 qid->qid_uid = *((__u64 *)key);
1041
1042                 rc = qmt_slv_read(env, qid, obj, &granted);
1043                 if (!granted)
1044                         goto next;
1045
1046                 lqe = lqe_locate(env, site, qid);
1047                 if (IS_ERR(lqe))
1048                         GOTO(out, rc = PTR_ERR(lqe));
1049                 lqe_write_lock(lqe);
1050                 lqe->lqe_recalc_granted += granted;
1051                 lqe_write_unlock(lqe);
1052                 lqe_putref(lqe);
1053 next:
1054                 rc = iops->next(env, it);
1055                 if (rc < 0)
1056                         CWARN("quota: failed to parse index "DFID
1057                               ", ->next error: rc = %d\n",
1058                               PFID(&qti->qti_fid), rc);
1059         } while (rc == 0 && !kthread_should_stop());
1060
1061 out:
1062         iops->put(env, it);
1063         iops->fini(env, it);
1064         RETURN(rc);
1065 }
1066
1067 static int qmt_site_recalc_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
1068                               struct hlist_node *hnode, void *data)
1069 {
1070         struct lquota_entry     *lqe;
1071         struct lu_env *env = data;
1072
1073         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
1074         LASSERT(atomic_read(&lqe->lqe_ref) > 0);
1075
1076         lqe_write_lock(lqe);
1077         if (lqe->lqe_granted != lqe->lqe_recalc_granted) {
1078                 struct qmt_device *qmt = lqe2qpi(lqe)->qpi_qmt;
1079                 struct thandle *th;
1080                 bool need_notify = false;
1081                 int rc;
1082
1083                 LQUOTA_DEBUG(lqe, "lqe_recalc_granted %llu\n",
1084                              lqe->lqe_recalc_granted);
1085                 lqe->lqe_granted = lqe->lqe_recalc_granted;
1086                 /* Always returns true, if there is no slaves in a pool */
1087                 need_notify |= qmt_adjust_qunit(env, lqe);
1088                 need_notify |= qmt_adjust_edquot(lqe, ktime_get_real_seconds());
1089                 if (need_notify) {
1090                         /* Find all lqes with lqe_id to reseed lgd array */
1091                         rc = qmt_pool_lqes_lookup_spec(env, qmt, lqe_rtype(lqe),
1092                                                 lqe_qtype(lqe), &lqe->lqe_id);
1093                         if (!rc && qti_lqes_glbl(env)->lqe_glbl_data) {
1094                                 qmt_seed_glbe(env,
1095                                         qti_lqes_glbl(env)->lqe_glbl_data);
1096                                 qmt_id_lock_notify(qmt, qti_lqes_glbl(env));
1097                         }
1098                         qti_lqes_fini(env);
1099                 }
1100                 th = dt_trans_create(env, qmt->qmt_child);
1101                 if (IS_ERR(th))
1102                         goto out;
1103
1104                 rc = lquota_disk_declare_write(env, th,
1105                                                LQE_GLB_OBJ(lqe),
1106                                                &lqe->lqe_id);
1107                 if (rc)
1108                         GOTO(out_stop, rc);
1109
1110                 rc = dt_trans_start_local(env, qmt->qmt_child, th);
1111                 if (rc)
1112                         GOTO(out_stop, rc);
1113
1114                 qmt_glb_write(env, th, lqe, 0, NULL);
1115 out_stop:
1116                 dt_trans_stop(env, qmt->qmt_child, th);
1117         }
1118 out:
1119         lqe->lqe_recalc_granted = 0;
1120         lqe_write_unlock(lqe);
1121
1122         return 0;
1123 }
1124
1125 #define MDT_DEV_NAME_LEN (LUSTRE_MAXFSNAME + sizeof("-MDT0000"))
1126 static struct obd_device *qmt_get_mgc(struct qmt_device *qmt)
1127 {
1128         char mdt_name[MDT_DEV_NAME_LEN];
1129         struct lustre_mount_info *lmi;
1130         struct obd_device *obd;
1131         int rc;
1132         ENTRY;
1133
1134         rc = server_name2fsname(qmt->qmt_svname, mdt_name, NULL);
1135         if (rc) {
1136                 CERROR("quota: cannot get server name from %s: rc = %d\n",
1137                        qmt->qmt_svname, rc);
1138                 RETURN(ERR_PTR(rc));
1139         }
1140
1141         strlcat(mdt_name, "-MDT0000", MDT_DEV_NAME_LEN);
1142         lmi = server_get_mount(mdt_name);
1143         if (lmi == NULL) {
1144                 rc = -ENOENT;
1145                 CERROR("%s: cannot get mount info from %s: rc = %d\n",
1146                        qmt->qmt_svname, mdt_name, rc);
1147                 RETURN(ERR_PTR(rc));
1148         }
1149         obd = s2lsi(lmi->lmi_sb)->lsi_mgc;
1150         lustre_put_lsi(lmi->lmi_sb);
1151
1152         RETURN(obd);
1153 }
1154
1155 static int qmt_pool_recalc(void *args)
1156 {
1157         struct qmt_pool_info *pool, *glbl_pool;
1158         struct rw_semaphore *sem = NULL;
1159         struct obd_device *obd;
1160         struct lu_env env;
1161         int i, rc, qtype, slaves_cnt;
1162         ENTRY;
1163
1164         pool = args;
1165
1166         rc = lu_env_init(&env, LCT_MD_THREAD);
1167         if (rc) {
1168                 CERROR("%s: cannot init env: rc = %d\n",
1169                        pool->qpi_qmt->qmt_svname, rc);
1170                 GOTO(out, rc);
1171         }
1172
1173         obd = qmt_get_mgc(pool->qpi_qmt);
1174         if (IS_ERR(obd))
1175                 GOTO(out, rc = PTR_ERR(obd));
1176         else
1177                 /* Waiting for the end of processing mgs config.
1178                  * It is needed to be sure all pools are configured. */
1179                 while (obd->obd_process_conf)
1180                         schedule_timeout_uninterruptible(cfs_time_seconds(1));
1181
1182         sem = qmt_sarr_rwsem(pool);
1183         LASSERT(sem);
1184         down_read(sem);
1185         /* Hold this to be sure that OSTs from this pool
1186          * can't do acquire/release.
1187          *
1188          * I guess below write semaphore could be a bottleneck
1189          * as qmt_dqacq would be blocked trying to hold
1190          * read_lock at qmt_pool_lookup->qti_pools_add.
1191          * But on the other hand adding/removing OSTs to the pool is
1192          * a rare operation. If finally this would be a problem,
1193          * we can consider another approach. For example we can
1194          * iterate through the POOL's lqes. Take lqe, hold lqe_write_lock
1195          * and go through appropriate OSTs. I don't use this approach now
1196          * as newly created pool hasn't lqes entries. So firstly we need
1197          * to get this lqes from the global pool index file. This
1198          * solution looks more complex, so leave it as it is. */
1199         down_write(&pool->qpi_recalc_sem);
1200
1201         glbl_pool = qmt_pool_lookup_glb(&env, pool->qpi_qmt, pool->qpi_rtype);
1202         if (IS_ERR(glbl_pool))
1203                 GOTO(out, rc = PTR_ERR(glbl_pool));
1204
1205         slaves_cnt = qmt_sarr_count(pool);
1206         CDEBUG(D_QUOTA, "Starting pool recalculation for %d slaves in %s\n",
1207                slaves_cnt, pool->qpi_name);
1208
1209         for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
1210                 for (i = 0; i < slaves_cnt; i++) {
1211                         struct qmt_thread_info  *qti = qmt_info(&env);
1212                         struct dt_object *slv_obj;
1213                         struct obd_uuid uuid;
1214                         int idx;
1215
1216                         if (kthread_should_stop())
1217                                 GOTO(out_stop, rc = 0);
1218                         idx = qmt_sarr_get_idx(pool, i);
1219                         LASSERT(idx >= 0);
1220
1221                         /* We don't need fsname here - anyway
1222                          * lquota_disk_slv_filename ignores it. */
1223                         snprintf(uuid.uuid, UUID_MAX, "-OST%04x_UUID", idx);
1224                         lquota_generate_fid(&qti->qti_fid, pool->qpi_rtype,
1225                                             qtype);
1226                         /* look-up index file associated with acquiring slave */
1227                         slv_obj = lquota_disk_slv_find(&env,
1228                                                 glbl_pool->qpi_qmt->qmt_child,
1229                                                 glbl_pool->qpi_root,
1230                                                 &qti->qti_fid,
1231                                                 &uuid);
1232                         if (IS_ERR(slv_obj))
1233                                 GOTO(out_stop, rc = PTR_ERR(slv_obj));
1234
1235                         CDEBUG(D_QUOTA, "slv_obj is found %p for uuid %s\n",
1236                                slv_obj, uuid.uuid);
1237                         qmt_obj_recalc(&env, slv_obj, pool->qpi_site[qtype]);
1238                         dt_object_put(&env, slv_obj);
1239                 }
1240                 /* Now go trough the site hash and compare lqe_granted
1241                  * with lqe_calc_granted. Write new value if disagree */
1242
1243                 cfs_hash_for_each(pool->qpi_site[qtype]->lqs_hash,
1244                                   qmt_site_recalc_cb, &env);
1245         }
1246         GOTO(out_stop, rc);
1247 out_stop:
1248         qpi_putref(&env, glbl_pool);
1249 out:
1250         if (xchg(&pool->qpi_recalc_task, NULL) == NULL)
1251                 /*
1252                  * Someone is waiting for us to stop - be sure not to exit
1253                  * before kthread_stop() gets a ref on the task.  No event
1254                  * will happen on 'pool, this is just a convenient way to
1255                  * wait.
1256                  */
1257                 wait_var_event(pool, kthread_should_stop());
1258
1259         clear_bit(QPI_FLAG_RECALC_OFFSET, &pool->qpi_flags);
1260         /* Pool can't be changed, since sem has been down.
1261          * Thus until up_read, no one can restart recalc thread.
1262          */
1263         if (sem) {
1264                 up_read(sem);
1265                 up_write(&pool->qpi_recalc_sem);
1266         }
1267
1268         /* qpi_getref has been called in qmt_start_pool_recalc,
1269          * however we can't call qpi_putref if lu_env_init failed.
1270          */
1271         if (env.le_ctx.lc_state == LCS_ENTERED) {
1272                 qpi_putref(&env, pool);
1273                 lu_env_fini(&env);
1274         }
1275
1276         return rc;
1277 }
1278
1279 static int qmt_start_pool_recalc(struct lu_env *env, struct qmt_pool_info *qpi)
1280 {
1281         struct task_struct *task;
1282         int rc = 0;
1283
1284         if (!test_and_set_bit(QPI_FLAG_RECALC_OFFSET, &qpi->qpi_flags)) {
1285                 LASSERT(!qpi->qpi_recalc_task);
1286
1287                 qpi_getref(qpi);
1288                 task = kthread_create(qmt_pool_recalc, qpi,
1289                                       "qsd_reint_%s", qpi->qpi_name);
1290                 if (IS_ERR(task)) {
1291                         clear_bit(QPI_FLAG_RECALC_OFFSET, &qpi->qpi_flags);
1292                         rc = PTR_ERR(task);
1293                         qpi_putref(env, qpi);
1294                 } else {
1295                         qpi->qpi_recalc_task = task;
1296                         /* Using park/unpark to start the thread ensures that
1297                          * the thread function does get calls, so the
1298                          * ref on qpi will be dropped
1299                          */
1300                         kthread_park(task);
1301                         kthread_unpark(task);
1302                 }
1303         }
1304
1305         RETURN(rc);
1306 }
1307
1308 static inline void qmt_stop_pool_recalc(struct qmt_pool_info *qpi)
1309 {
1310         struct task_struct *task;
1311
1312         task = xchg(&qpi->qpi_recalc_task, NULL);
1313         if (task)
1314                 kthread_stop(task);
1315 }
1316
1317 static int qmt_pool_slv_nr_change(const struct lu_env *env,
1318                                   struct qmt_pool_info *pool,
1319                                   int idx, bool add)
1320 {
1321         struct qmt_pool_info *glbl_pool;
1322         int qtype;
1323
1324         glbl_pool = qmt_pool_lookup_glb(env, pool->qpi_qmt, LQUOTA_RES_DT);
1325         if (IS_ERR(glbl_pool))
1326                 RETURN(PTR_ERR(glbl_pool));
1327
1328         for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
1329                 struct qmt_thread_info  *qti = qmt_info(env);
1330                 struct dt_object *slv_obj;
1331                 struct obd_uuid uuid;
1332
1333                 /* We don't need fsname here - anyway
1334                  * lquota_disk_slv_filename ignores it. */
1335                 snprintf(uuid.uuid, UUID_MAX, "-OST%04x_UUID", idx);
1336                 lquota_generate_fid(&qti->qti_fid, pool->qpi_rtype,
1337                                     qtype);
1338                 /* look-up index file associated with acquiring slave */
1339                 slv_obj = lquota_disk_slv_find(env,
1340                                         glbl_pool->qpi_qmt->qmt_child,
1341                                         glbl_pool->qpi_root,
1342                                         &qti->qti_fid,
1343                                         &uuid);
1344                 if (IS_ERR(slv_obj))
1345                         continue;
1346
1347                 if (add)
1348                         pool->qpi_slv_nr[QMT_STYPE_OST][qtype]++;
1349                 else
1350                         pool->qpi_slv_nr[QMT_STYPE_OST][qtype]--;
1351                 dt_object_put(env, slv_obj);
1352         }
1353         qpi_putref(env, glbl_pool);
1354
1355         return 0;
1356 }
1357
1358 static int qmt_pool_add_rem(struct obd_device *obd, char *poolname,
1359                             char *slavename, bool add)
1360 {
1361         struct qmt_device       *qmt = lu2qmt_dev(obd->obd_lu_dev);
1362         struct qmt_pool_info    *qpi;
1363         struct lu_env            env;
1364         int                      rc, idx;
1365         ENTRY;
1366
1367         if (strnlen(poolname, LOV_MAXPOOLNAME + 1) > LOV_MAXPOOLNAME)
1368                 RETURN(-ENAMETOOLONG);
1369
1370         CDEBUG(D_QUOTA, add ? "%s: pool %s, adding %s\n" :
1371                               "%s: pool %s, removing %s\n",
1372               obd->obd_name, poolname, slavename);
1373
1374         rc = server_name2index(slavename, &idx, NULL);
1375         if (rc != LDD_F_SV_TYPE_OST)
1376                 RETURN(-EINVAL);
1377
1378         rc = lu_env_init(&env, LCT_MD_THREAD);
1379         if (rc) {
1380                 CERROR("%s: cannot init env: rc = %d\n", obd->obd_name, rc);
1381                 RETURN(rc);
1382         }
1383
1384         qpi = qmt_pool_lookup_name(&env, qmt, LQUOTA_RES_DT, poolname);
1385         if (IS_ERR(qpi)) {
1386                 CWARN("%s: can't find pool %s: rc = %long\n",
1387                       obd->obd_name, poolname, PTR_ERR(qpi));
1388                 GOTO(out, rc = PTR_ERR(qpi));
1389         }
1390
1391         rc = add ? qmt_sarr_pool_add(qpi, idx, 32) :
1392                    qmt_sarr_pool_rem(qpi, idx);
1393         if (rc) {
1394                 CERROR("%s: can't %s %s pool %s: rc = %d\n",
1395                        add ? "add to" : "remove", obd->obd_name,
1396                        slavename, poolname, rc);
1397                 GOTO(out_putref, rc);
1398         }
1399         qmt_pool_slv_nr_change(&env, qpi, idx, add);
1400         qmt_start_pool_recalc(&env, qpi);
1401
1402 out_putref:
1403         qpi_putref(&env, qpi);
1404 out:
1405         lu_env_fini(&env);
1406         RETURN(rc);
1407 }
1408
1409
1410
1411 /**
1412  * Add a single target device to the named pool.
1413  *
1414  * \param[in] obd       OBD device on which to add the pool
1415  * \param[in] poolname  name of the pool to which to add the target \a slavename
1416  * \param[in] slavename name of the target device to be added
1417  *
1418  * \retval              0 if \a slavename was (previously) added to the pool
1419  * \retval              negative error number on failure
1420  */
1421 int qmt_pool_add(struct obd_device *obd, char *poolname, char *slavename)
1422 {
1423         return qmt_pool_add_rem(obd, poolname, slavename, true);
1424 }
1425
1426 /**
1427  * Remove the named target from the specified pool.
1428  *
1429  * \param[in] obd       OBD device from which to remove \a poolname
1430  * \param[in] poolname  name of the pool to be changed
1431  * \param[in] slavename name of the target to remove from \a poolname
1432  *
1433  * \retval              0 on successfully removing \a slavename from the pool
1434  * \retval              negative number on error (e.g. \a slavename not in pool)
1435  */
1436 int qmt_pool_rem(struct obd_device *obd, char *poolname, char *slavename)
1437 {
1438         return qmt_pool_add_rem(obd, poolname, slavename, false);
1439 }
1440
1441 /**
1442  * Remove the named pool from the QMT device.
1443  *
1444  * \param[in] obd       OBD device on which pool was previously created
1445  * \param[in] poolname  name of pool to remove from \a obd
1446  *
1447  * \retval              0 on successfully removing the pool
1448  * \retval              negative error numbers for failures
1449  */
1450 int qmt_pool_del(struct obd_device *obd, char *poolname)
1451 {
1452         struct qmt_device       *qmt = lu2qmt_dev(obd->obd_lu_dev);
1453         struct qmt_pool_info    *qpi;
1454         struct lu_fid            fid;
1455         char                     buf[LQUOTA_NAME_MAX];
1456         struct lu_env            env;
1457         int                      rc;
1458         int                      qtype;
1459         ENTRY;
1460
1461         if (strnlen(poolname, LOV_MAXPOOLNAME + 1) > LOV_MAXPOOLNAME)
1462                 RETURN(-ENAMETOOLONG);
1463
1464         CDEBUG(D_QUOTA, "Removing quota pool "LOV_POOLNAMEF"\n",
1465                poolname);
1466
1467         rc = lu_env_init(&env, LCT_MD_THREAD);
1468         if (rc) {
1469                 CERROR("%s: cannot init env: rc = %d\n", obd->obd_name, rc);
1470                 RETURN(rc);
1471         }
1472
1473         /* look-up pool in charge of this global index FID */
1474         qpi = qmt_pool_lookup_name(&env, qmt, LQUOTA_RES_DT, poolname);
1475         if (IS_ERR(qpi)) {
1476                 /* Valid case for several MDTs at the same node -
1477                  * pool removed by the 1st MDT in config */
1478                 CDEBUG(D_QUOTA, "Cannot find pool %s\n", poolname);
1479                 lu_env_fini(&env);
1480                 RETURN(PTR_ERR(qpi));
1481         }
1482
1483         for (qtype = 0; qtype < LL_MAXQUOTAS; qtype++) {
1484                 lquota_generate_fid(&fid, LQUOTA_RES_DT, qtype);
1485                 snprintf(buf, LQUOTA_NAME_MAX, "0x%x", fid.f_oid);
1486                 rc = local_object_unlink(&env, qmt->qmt_child,
1487                                          qpi->qpi_root, buf);
1488                 if (rc)
1489                         CWARN("%s: cannot unlink %s from pool %s: rc = %d\n",
1490                               obd->obd_name, buf, poolname, rc);
1491         }
1492
1493         /* put ref from look-up */
1494         qpi_putref(&env, qpi);
1495         /* put last ref to free qpi */
1496         qpi_putref(&env, qpi);
1497
1498         snprintf(buf, LQUOTA_NAME_MAX, "%s-%s",
1499                  RES_NAME(LQUOTA_RES_DT), poolname);
1500         rc = local_object_unlink(&env, qmt->qmt_child, qmt->qmt_root, buf);
1501         if (rc)
1502                 CWARN("%s: cannot unlink dir %s: rc = %d\n",
1503                       obd->obd_name, poolname, rc);
1504
1505         lu_env_fini(&env);
1506         RETURN(0);
1507 }
1508
1509 static inline int qmt_sarr_pool_init(struct qmt_pool_info *qpi)
1510 {
1511
1512         /* No need to initialize sarray for global pool
1513          * as it always includes all slaves */
1514         if (qmt_pool_global(qpi))
1515                 return 0;
1516
1517         switch (qpi->qpi_rtype) {
1518         case LQUOTA_RES_DT:
1519                 return lu_tgt_pool_init(&qpi->qpi_sarr.osts, 0);
1520         case LQUOTA_RES_MD:
1521         default:
1522                 return 0;
1523         }
1524 }
1525
1526 static inline int qmt_sarr_pool_add(struct qmt_pool_info *qpi, int idx, int min)
1527 {
1528         switch (qpi->qpi_rtype) {
1529         case LQUOTA_RES_DT:
1530                 return lu_tgt_pool_add(&qpi->qpi_sarr.osts, idx, min);
1531         case LQUOTA_RES_MD:
1532         default:
1533                 return 0;
1534         }
1535 }
1536
1537 static inline int qmt_sarr_pool_rem(struct qmt_pool_info *qpi, int idx)
1538 {
1539         switch (qpi->qpi_rtype) {
1540         case LQUOTA_RES_DT:
1541                 return lu_tgt_pool_remove(&qpi->qpi_sarr.osts, idx);
1542         case LQUOTA_RES_MD:
1543         default:
1544                 return 0;
1545         }
1546 }
1547
1548 static inline void qmt_sarr_pool_free(struct qmt_pool_info *qpi)
1549 {
1550         if (qmt_pool_global(qpi))
1551                 return;
1552
1553         switch (qpi->qpi_rtype) {
1554         case LQUOTA_RES_DT:
1555                 if (qpi->qpi_sarr.osts.op_array)
1556                         lu_tgt_pool_free(&qpi->qpi_sarr.osts);
1557                 return;
1558         case LQUOTA_RES_MD:
1559         default:
1560                 return;
1561         }
1562 }
1563
1564 static inline int qmt_sarr_check_idx(struct qmt_pool_info *qpi, int idx)
1565 {
1566         if (qmt_pool_global(qpi))
1567                 return 0;
1568
1569         switch (qpi->qpi_rtype) {
1570         case LQUOTA_RES_DT:
1571                 return lu_tgt_check_index(idx, &qpi->qpi_sarr.osts);
1572         case LQUOTA_RES_MD:
1573         default:
1574                 return 0;
1575         }
1576 }
1577
1578 struct rw_semaphore *qmt_sarr_rwsem(struct qmt_pool_info *qpi)
1579 {
1580         switch (qpi->qpi_rtype) {
1581         case LQUOTA_RES_DT:
1582                 /* to protect ost_pool use */
1583                 return &qpi->qpi_sarr.osts.op_rw_sem;
1584         case LQUOTA_RES_MD:
1585         default:
1586                 return NULL;
1587         }
1588 }
1589
1590 int qmt_sarr_get_idx(struct qmt_pool_info *qpi, int arr_idx)
1591 {
1592
1593         if (qmt_pool_global(qpi))
1594                 return arr_idx;
1595
1596         switch (qpi->qpi_rtype) {
1597         case LQUOTA_RES_DT:
1598                 LASSERTF(arr_idx < qpi->qpi_sarr.osts.op_count && arr_idx >= 0,
1599                          "idx invalid %d op_count %d\n", arr_idx,
1600                          qpi->qpi_sarr.osts.op_count);
1601                 return qpi->qpi_sarr.osts.op_array[arr_idx];
1602         case LQUOTA_RES_MD:
1603         default:
1604                 return -EINVAL;
1605         }
1606 }
1607
1608 /* Number of slaves in a pool */
1609 unsigned int qmt_sarr_count(struct qmt_pool_info *qpi)
1610 {
1611         switch (qpi->qpi_rtype) {
1612         case LQUOTA_RES_DT:
1613                 return qpi->qpi_sarr.osts.op_count;
1614         case LQUOTA_RES_MD:
1615         default:
1616                 return -EINVAL;
1617         }
1618 }