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