lustre/quota/qmt_entry.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2012, 2016, 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 #define DEBUG_SUBSYSTEM S_LQUOTA
  32
  33 #include "qmt_internal.h"
  34
  35 static void qmt_work_lvbo_free(struct work_struct *work)
  36 {
  37         struct lqe_glbl_data *lgd;
  38         struct lquota_entry *lqe;
  39
  40         lqe = container_of(work, struct lquota_entry, lqe_work);
  41         mutex_lock(&lqe->lqe_glbl_data_lock);
  42         lgd = lqe->lqe_glbl_data;
  43         lqe->lqe_glbl_data = NULL;
  44         mutex_unlock(&lqe->lqe_glbl_data_lock);
  45         qmt_free_lqe_gd(lgd);
  46
  47         if (unlikely(lgd == NULL)) {
  48                 struct qmt_pool_info *pool;
  49
  50                 pool = (struct qmt_pool_info *)lqe->lqe_site->lqs_parent;
  51                 CWARN("%s: lvbo for (id=%llx) not fully inited\n",
  52                       pool->qpi_qmt->qmt_svname,
  53                       lqe->lqe_id.qid_uid);
  54         }
  55
  56         /* release lqe reference */
  57         lqe_putref(lqe);
  58 }
  59
  60 /*
  61  * Initialize qmt-specific fields of quota entry.
  62  *
  63  * \param lqe - is the quota entry to initialize
  64  * \param arg - is the pointer to the qmt_pool_info structure
  65  */
  66 static void qmt_lqe_init(struct lquota_entry *lqe, void *arg)
  67 {
  68         LASSERT(lqe_is_master(lqe));
  69
  70         lqe->lqe_revoke_time = 0;
  71         init_rwsem(&lqe->lqe_sem);
  72         mutex_init(&lqe->lqe_glbl_data_lock);
  73         INIT_WORK(&lqe->lqe_work, qmt_work_lvbo_free);
  74 }
  75
  76 /* Apply the default quota setting to the specified quota entry
  77  *
  78  * \param env           - is the environment passed by the caller
  79  * \param pool          - is the quota pool of the quota entry
  80  * \param lqe           - is the lquota_entry object to apply default quota on
  81  * \param create_record - if true, an global quota record will be created and
  82  *                        write to the disk.
  83  *
  84  * \retval 0            : success
  85  * \retval -ve          : other appropriate errors
  86  */
  87 int qmt_lqe_set_default(const struct lu_env *env, struct qmt_pool_info *pool,
  88                         struct lquota_entry *lqe, bool create_record)
  89 {
  90         struct lquota_entry     *lqe_def;
  91         int                     rc = 0;
  92
  93         ENTRY;
  94
  95         if (lqe->lqe_id.qid_uid == 0)
  96                 RETURN(0);
  97
  98         lqe_def = pool->qpi_grace_lqe[lqe_qtype(lqe)];
  99
 100         LQUOTA_DEBUG(lqe, "inherit default quota");
 101
 102         lqe->lqe_is_default = true;
 103         lqe->lqe_hardlimit = lqe_def->lqe_hardlimit;
 104         lqe->lqe_softlimit = lqe_def->lqe_softlimit;
 105
 106         if (create_record) {
 107                 lqe->lqe_uptodate = true;
 108                 rc = qmt_set_with_lqe(env, pool->qpi_qmt, lqe, 0, 0,
 109                                       LQUOTA_GRACE_FLAG(0, LQUOTA_FLAG_DEFAULT),
 110                                       QIF_TIMES, true, false);
 111
 112                 if (rc != 0)
 113                         LQUOTA_ERROR(lqe, "failed to create the global quota"
 114                                      " record: %d", rc);
 115         }
 116
 117         if (lqe->lqe_hardlimit == 0 && lqe->lqe_softlimit == 0)
 118                 lqe->lqe_enforced = false;
 119         else
 120                 lqe->lqe_enforced = true;
 121
 122         RETURN(rc);
 123 }
 124
 125 /*
 126  * Update a lquota entry. This is done by reading quota settings from the global
 127  * index. The lquota entry must be write locked.
 128  *
 129  * \param env - the environment passed by the caller
 130  * \param lqe - is the quota entry to refresh
 131  * \param arg - is the pointer to the qmt_pool_info structure
 132  * \param find - don't create lqe on disk in case of ENOENT if true
 133  */
 134 static int qmt_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
 135                         void *arg, bool find)
 136 {
 137         struct qmt_thread_info  *qti = qmt_info(env);
 138         struct qmt_pool_info    *pool = (struct qmt_pool_info *)arg;
 139         int                      rc;
 140         ENTRY;
 141
 142         LASSERT(lqe_is_master(lqe));
 143
 144         if (qmt_pool_global(pool))
 145                 lqe->lqe_is_global = 1;
 146
 147         /* read record from disk */
 148         rc = lquota_disk_read(env, pool->qpi_glb_obj[lqe->lqe_site->lqs_qtype],
 149                               &lqe->lqe_id, (struct dt_rec *)&qti->qti_glb_rec);
 150
 151         switch (rc) {
 152         case -ENOENT:
 153                 if (find)
 154                         RETURN(-ENOENT);
 155                 qmt_lqe_set_default(env, pool, lqe, true);
 156                 break;
 157         case 0:
 158                 /* copy quota settings from on-disk record */
 159                 lqe->lqe_granted   = qti->qti_glb_rec.qbr_granted;
 160                 lqe->lqe_hardlimit = qti->qti_glb_rec.qbr_hardlimit;
 161                 lqe->lqe_softlimit = qti->qti_glb_rec.qbr_softlimit;
 162                 lqe->lqe_gracetime = LQUOTA_GRACE(qti->qti_glb_rec.qbr_time);
 163
 164                 if (lqe->lqe_hardlimit == 0 && lqe->lqe_softlimit == 0 &&
 165                     (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) &
 166                      LQUOTA_FLAG_DEFAULT))
 167                         qmt_lqe_set_default(env, pool, lqe, false);
 168                 else if (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) &
 169                                                         LQUOTA_FLAG_RESET)
 170                         lqe->lqe_is_reset = true;
 171                 break;
 172         default:
 173                 LQUOTA_ERROR(lqe, "failed to read quota entry from disk, rc:%d",
 174                              rc);
 175                 RETURN(rc);
 176         }
 177
 178         if (lqe->lqe_id.qid_uid == 0 ||
 179             (lqe->lqe_hardlimit == 0 && lqe->lqe_softlimit == 0))
 180                 /* {hard,soft}limit=0 means no quota enforced */
 181                 lqe->lqe_enforced = false;
 182         else
 183                 lqe->lqe_enforced  = true;
 184
 185         LQUOTA_DEBUG(lqe, "read");
 186         RETURN(0);
 187 }
 188
 189 /*
 190  * Print lqe information for debugging.
 191  *
 192  * \param lqe - is the quota entry to debug
 193  * \param arg - is the pointer to the qmt_pool_info structure
 194  * \param msgdata - debug message
 195  * \param fmt     - format of debug message
 196  */
 197 static void qmt_lqe_debug(struct lquota_entry *lqe, void *arg,
 198                           struct libcfs_debug_msg_data *msgdata,
 199                           struct va_format *vaf)
 200 {
 201         struct qmt_pool_info    *pool = (struct qmt_pool_info *)arg;
 202
 203         libcfs_debug_msg(msgdata,
 204                          "%pV qmt:%s pool:%s-%s id:%llu enforced:%d hard:%llu soft:%llu granted:%llu time:%llu qunit: %llu edquot:%d may_rel:%llu revoke:%lld default:%s\n",
 205                          vaf, pool->qpi_qmt->qmt_svname,
 206                          RES_NAME(pool->qpi_rtype),
 207                          pool->qpi_name,
 208                          lqe->lqe_id.qid_uid, lqe->lqe_enforced,
 209                          lqe->lqe_hardlimit, lqe->lqe_softlimit,
 210                          lqe->lqe_granted, lqe->lqe_gracetime,
 211                          lqe->lqe_qunit, lqe->lqe_edquot, lqe->lqe_may_rel,
 212                          lqe->lqe_revoke_time,
 213                          lqe->lqe_is_default ? "yes" : "no");
 214 }
 215
 216 /*
 217  * Vector of quota entry operations supported on the master
 218  */
 219 const struct lquota_entry_operations qmt_lqe_ops = {
 220         .lqe_init       = qmt_lqe_init,
 221         .lqe_read       = qmt_lqe_read,
 222         .lqe_debug      = qmt_lqe_debug,
 223 };
 224
 225 /*
 226  * Reserve enough credits to update records in both the global index and
 227  * the slave index identified by \slv_obj
 228  *
 229  * \param env     - is the environment passed by the caller
 230  * \param lqe     - is the quota entry associated with the identifier
 231  *                  subject to the change. If it is NULL lqes array is
 232  *                  taken from env with qti_lqes_env(env).
 233  * \param slv_obj - is the dt_object associated with the index file
 234  * \param sync    - make transaction sync if true
 235  */
 236 struct thandle *qmt_trans_start_with_slv(const struct lu_env *env,
 237                                          struct lquota_entry *lqe,
 238                                          struct dt_object *slv_obj,
 239                                          bool sync)
 240 {
 241         struct qmt_device       *qmt;
 242         struct thandle          *th;
 243         struct lquota_entry     **lqes;
 244         struct qmt_lqe_restore  *restore;
 245         int                      rc, i, lqes_cnt;
 246         ENTRY;
 247
 248         restore = qti_lqes_rstr(env);
 249         if (!lqe) {
 250                 lqes_cnt = qti_lqes_cnt(env);
 251                 lqes = qti_lqes(env);
 252         } else {
 253                 lqes_cnt = 1;
 254                 lqes = &lqe;
 255         }
 256
 257         /* qmt is the same for all lqes, so take it from the 1st */
 258         qmt = lqe2qpi(lqes[0])->qpi_qmt;
 259
 260         if (slv_obj != NULL)
 261                 LQUOTA_DEBUG(lqes[0], "declare write for slv "DFID,
 262                              PFID(lu_object_fid(&slv_obj->do_lu)));
 263
 264         /* start transaction */
 265         th = dt_trans_create(env, qmt->qmt_child);
 266         if (IS_ERR(th))
 267                 RETURN(th);
 268
 269         if (sync)
 270                 /* quota settings on master are updated synchronously for the
 271                  * time being */
 272                 th->th_sync = 1;
 273
 274         /* reserve credits for global index update */
 275         for (i = 0; i < lqes_cnt; i++) {
 276                 rc = lquota_disk_declare_write(env, th,
 277                                                LQE_GLB_OBJ(lqes[i]),
 278                                                &lqes[i]->lqe_id);
 279                 if (rc)
 280                         GOTO(out, rc);
 281         }
 282
 283         if (slv_obj != NULL) {
 284                 /* reserve credits for slave index update */
 285                 rc = lquota_disk_declare_write(env, th, slv_obj,
 286                                                &lqes[0]->lqe_id);
 287                 if (rc)
 288                         GOTO(out, rc);
 289         }
 290
 291         /* start transaction */
 292         rc = dt_trans_start_local(env, qmt->qmt_child, th);
 293         if (rc)
 294                 GOTO(out, rc);
 295
 296         EXIT;
 297 out:
 298         if (rc) {
 299                 dt_trans_stop(env, qmt->qmt_child, th);
 300                 th = ERR_PTR(rc);
 301                 LQUOTA_ERROR(lqes[0], "failed to slv declare write for "DFID
 302                              ", rc:%d", PFID(lu_object_fid(&slv_obj->do_lu)),
 303                              rc);
 304         } else {
 305                 for (i = 0; i < lqes_cnt; i++) {
 306                         restore[i].qlr_hardlimit = lqes[i]->lqe_hardlimit;
 307                         restore[i].qlr_softlimit = lqes[i]->lqe_softlimit;
 308                         restore[i].qlr_gracetime = lqes[i]->lqe_gracetime;
 309                         restore[i].qlr_granted   = lqes[i]->lqe_granted;
 310                         restore[i].qlr_qunit     = lqes[i]->lqe_qunit;
 311                 }
 312         }
 313         return th;
 314 }
 315
 316 /*
 317  * Reserve enough credits to update a record in the global index
 318  *
 319  * \param env     - is the environment passed by the caller
 320  * \param lqe     - is the quota entry to be modified in the global index
 321  * \param restore - is a temporary storage for current quota settings which will
 322  *                  be restored if something goes wrong at index update time.
 323  */
 324 struct thandle *qmt_trans_start(const struct lu_env *env,
 325                                 struct lquota_entry *lqe)
 326 {
 327         LQUOTA_DEBUG(lqe, "declare write");
 328         return qmt_trans_start_with_slv(env, lqe, NULL, true);
 329 }
 330
 331 int qmt_glb_write_lqes(const struct lu_env *env, struct thandle *th,
 332                        __u32 flags, __u64 *ver)
 333 {
 334         int i, rc = 0;
 335
 336         for (i = 0; i < qti_lqes_cnt(env); i++) {
 337                 rc = qmt_glb_write(env, th, qti_lqes(env)[i], flags, ver);
 338                 if (rc)
 339                         break;
 340         }
 341         return rc;
 342 }
 343
 344 /*
 345  * Update record associated with a quota entry in the global index.
 346  * If LQUOTA_BUMP_VER is set, then the global index version must also be
 347  * bumped.
 348  * The entry must be at least read locked, dirty and up-to-date.
 349  *
 350  * \param env   - the environment passed by the caller
 351  * \param th    - is the transaction handle to be used for the disk writes
 352  * \param lqe   - is the quota entry to udpate
 353  * \param obj   - is the dt_object associated with the index file
 354  * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
 355  * \param ver   - is used to return the new version of the index.
 356  *
 357  * \retval      - 0 on success and lqe dirty flag cleared,
 358  *                appropriate error on failure and uptodate flag cleared.
 359  */
 360 int qmt_glb_write(const struct lu_env *env, struct thandle *th,
 361                   struct lquota_entry *lqe, __u32 flags, __u64 *ver)
 362 {
 363         struct qmt_thread_info  *qti = qmt_info(env);
 364         struct lquota_glb_rec   *rec;
 365         int                      rc;
 366         ENTRY;
 367
 368         LASSERT(lqe != NULL);
 369         LASSERT(lqe_is_master(lqe));
 370         LASSERT(lqe_is_locked(lqe));
 371         LASSERT(lqe->lqe_uptodate);
 372         LASSERT((flags & ~(LQUOTA_BUMP_VER | LQUOTA_SET_VER)) == 0);
 373
 374         LQUOTA_DEBUG(lqe, "write glb");
 375
 376         /* never delete the entry even when the id isn't enforced and
 377          * no any guota granted, otherwise, this entry will not be
 378          * synced to slave during the reintegration. */
 379         rec = &qti->qti_glb_rec;
 380
 381         /* fill global index with updated quota settings */
 382         rec->qbr_granted   = lqe->lqe_granted;
 383         if (lqe->lqe_is_default) {
 384                 rec->qbr_hardlimit = 0;
 385                 rec->qbr_softlimit = 0;
 386                 rec->qbr_time      = LQUOTA_GRACE_FLAG(lqe->lqe_gracetime,
 387                                                        LQUOTA_FLAG_DEFAULT);
 388         } else if (lqe->lqe_is_reset) {
 389                 rec->qbr_hardlimit = 0;
 390                 rec->qbr_softlimit = 0;
 391                 rec->qbr_granted = 0;
 392                 rec->qbr_time      = LQUOTA_GRACE_FLAG(lqe->lqe_gracetime,
 393                                                        LQUOTA_FLAG_RESET);
 394         } else {
 395                 rec->qbr_hardlimit = lqe->lqe_hardlimit;
 396                 rec->qbr_softlimit = lqe->lqe_softlimit;
 397                 rec->qbr_time      = lqe->lqe_gracetime;
 398         }
 399
 400         /* write new quota settings */
 401         rc = lquota_disk_write(env, th, LQE_GLB_OBJ(lqe), &lqe->lqe_id,
 402                                (struct dt_rec *)rec, flags, ver);
 403         if (rc)
 404                 /* we failed to write the new quota settings to disk, report
 405                  * error to caller who will restore the initial value */
 406                 LQUOTA_ERROR(lqe, "failed to update global index, rc:%d", rc);
 407
 408         RETURN(rc);
 409 }
 410
 411 /*
 412  * Read from disk how much quota space is allocated to a slave.
 413  * This is done by reading records from the dedicated slave index file.
 414  * Return in \granted how much quota space is currently allocated to the
 415  * slave.
 416  * The entry must be at least read locked.
 417  *
 418  * \param env - the environment passed by the caller
 419  * \param lqe_id - is the quota id associated with the identifier to look-up
 420  *              in the slave index
 421  * \param slv_obj - is the dt_object associated with the slave index
 422  * \param granted - is the output parameter where to return how much space
 423  *                  is granted to the slave.
 424  *
 425  * \retval    - 0 on success, appropriate error on failure
 426  */
 427 int qmt_slv_read(const struct lu_env *env, union lquota_id *qid,
 428                  struct dt_object *slv_obj, __u64 *granted)
 429 {
 430         struct qmt_thread_info  *qti = qmt_info(env);
 431         struct lquota_slv_rec   *slv_rec = &qti->qti_slv_rec;
 432         int                      rc;
 433         ENTRY;
 434
 435         CDEBUG(D_QUOTA, "read id:%llu form slv "DFID"\n",
 436                qid->qid_uid, PFID(lu_object_fid(&slv_obj->do_lu)));
 437
 438         /* read slave record from disk */
 439         rc = lquota_disk_read(env, slv_obj, qid,
 440                               (struct dt_rec *)slv_rec);
 441         switch (rc) {
 442         case -ENOENT:
 443                 *granted = 0;
 444                 break;
 445         case 0:
 446                 /* extract granted from on-disk record */
 447                 *granted = slv_rec->qsr_granted;
 448                 break;
 449         default:
 450                 CERROR("Failed to read slave record for %llu from "DFID"\n",
 451                        qid->qid_uid, PFID(lu_object_fid(&slv_obj->do_lu)));
 452                 RETURN(rc);
 453         }
 454
 455         CDEBUG(D_QUOTA, "Successful slv read %llu\n", *granted);
 456
 457         RETURN(0);
 458 }
 459
 460 /*
 461  * Update record in slave index file.
 462  * The entry must be at least read locked.
 463  *
 464  * \param env - the environment passed by the caller
 465  * \param th  - is the transaction handle to be used for the disk writes
 466  * \param lqe - is the dirty quota entry which will be updated at the same time
 467  *              as the slave index
 468  * \param slv_obj - is the dt_object associated with the slave index
 469  * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
 470  * \param ver   - is used to return the new version of the index.
 471  * \param granted - is the new amount of quota space owned by the slave
 472  *
 473  * \retval    - 0 on success, appropriate error on failure
 474  */
 475 int qmt_slv_write(const struct lu_env *env, struct thandle *th,
 476                   struct lquota_entry *lqe, struct dt_object *slv_obj,
 477                   __u32 flags, __u64 *ver, __u64 granted)
 478 {
 479         struct qmt_thread_info  *qti = qmt_info(env);
 480         struct lquota_slv_rec   *rec;
 481         int                      rc;
 482         ENTRY;
 483
 484         LASSERT(lqe != NULL);
 485         LASSERT(lqe_is_master(lqe));
 486         LASSERT(lqe_is_locked(lqe));
 487
 488         LQUOTA_DEBUG(lqe, "write slv "DFID" granted:%llu",
 489                      PFID(lu_object_fid(&slv_obj->do_lu)), granted);
 490
 491         /* never delete the entry, otherwise, it'll not be transferred
 492          * to slave during reintegration. */
 493         rec = &qti->qti_slv_rec;
 494
 495         /* updated space granted to this slave */
 496         rec->qsr_granted = granted;
 497
 498         /* write new granted space */
 499         rc = lquota_disk_write(env, th, slv_obj, &lqe->lqe_id,
 500                                (struct dt_rec *)rec, flags, ver);
 501         if (rc) {
 502                 LQUOTA_ERROR(lqe,
 503                              "failed to update slave index "DFID" granted:%llu",
 504                              PFID(lu_object_fid(&slv_obj->do_lu)),
 505                              granted);
 506                 RETURN(rc);
 507         }
 508
 509         RETURN(0);
 510 }
 511
 512 /*
 513  * Check whether new limits are valid for this pool
 514  *
 515  * \param lqe  - is the quota entry subject to the setquota
 516  * \param hard - is the new hard limit
 517  * \param soft - is the new soft limit
 518  */
 519 int qmt_validate_limits(struct lquota_entry *lqe, __u64 hard, __u64 soft)
 520 {
 521         ENTRY;
 522
 523         if (hard != 0 && soft > hard)
 524                 /* soft limit must be less than hard limit */
 525                 RETURN(-EINVAL);
 526         RETURN(0);
 527 }
 528
 529 /*
 530  * Set/clear edquot flag after quota space allocation/release or settings
 531  * change. Slaves will be notified of changes via glimpse on per-ID lock
 532  *
 533  * \param lqe - is the quota entry to check
 534  * \param now - is the current time in second used for grace time managment
 535  */
 536 bool qmt_adjust_edquot(struct lquota_entry *lqe, __u64 now)
 537 {
 538         struct qmt_pool_info    *pool = lqe2qpi(lqe);
 539         ENTRY;
 540
 541         if (!lqe->lqe_enforced || lqe->lqe_id.qid_uid == 0)
 542                 RETURN(false);
 543
 544         if (!lqe->lqe_edquot) {
 545                 /* space exhausted flag not set, let's check whether it is time
 546                  * to set the flag */
 547
 548                 if (!qmt_space_exhausted(lqe, now))
 549                         /* the qmt still has available space */
 550                         RETURN(false);
 551
 552                 /* See comment in qmt_adjust_qunit(). LU-4139 */
 553                 if (qmt_hard_exhausted(lqe) ||
 554                     pool->qpi_rtype != LQUOTA_RES_DT) {
 555                         time64_t lapse;
 556
 557                         /* we haven't reached the minimal qunit yet so there is
 558                          * still hope that the rebalancing process might free
 559                          * up some quota space */
 560                         if (lqe->lqe_qunit != pool->qpi_least_qunit)
 561                                 RETURN(false);
 562
 563                         /* least qunit value not sent to all slaves yet */
 564                         if (lqe->lqe_revoke_time == 0)
 565                                 RETURN(false);
 566
 567                         /* Let's give more time to slave to release space */
 568                         lapse = ktime_get_seconds() - QMT_REBA_TIMEOUT;
 569                         if (lqe->lqe_may_rel != 0 && lqe->lqe_revoke_time > lapse)
 570                                 RETURN(false);
 571                 } else {
 572                         if (lqe->lqe_qunit > pool->qpi_soft_least_qunit)
 573                                 RETURN(false);
 574                 }
 575
 576                 /* set edquot flag */
 577                 lqe->lqe_edquot = true;
 578         } else {
 579                 /* space exhausted flag set, let's check whether it is time to
 580                  * clear it */
 581
 582                 if (qmt_space_exhausted(lqe, now))
 583                         /* the qmt still has not space */
 584                         RETURN(false);
 585
 586                 if (lqe->lqe_hardlimit != 0 &&
 587                     lqe->lqe_granted + pool->qpi_least_qunit >
 588                                                         lqe->lqe_hardlimit)
 589                         /* we clear the flag only once at least one least qunit
 590                          * is available */
 591                         RETURN(false);
 592
 593                 /* clear edquot flag */
 594                 lqe->lqe_edquot = false;
 595         }
 596
 597         LQUOTA_DEBUG(lqe, "changing edquot flag");
 598
 599         /* let's notify slave by issuing glimpse on per-ID lock.
 600          * the rebalance thread will take care of this */
 601         RETURN(true);
 602 }
 603
 604 /* Using least_qunit when over block softlimit will seriously impact the
 605  * write performance, we need to do some special tweaking on that. */
 606 static __u64 qmt_calc_softlimit(struct lquota_entry *lqe, bool *oversoft)
 607 {
 608         struct qmt_pool_info *pool = lqe2qpi(lqe);
 609
 610         LASSERT(lqe->lqe_softlimit != 0);
 611         *oversoft = false;
 612         /* No need to do special tweaking for inode limit */
 613         if (pool->qpi_rtype != LQUOTA_RES_DT)
 614                 return lqe->lqe_softlimit;
 615
 616         if (lqe->lqe_granted <= lqe->lqe_softlimit +
 617                                 pool->qpi_soft_least_qunit) {
 618                 return lqe->lqe_softlimit;
 619         } else if (lqe->lqe_hardlimit != 0) {
 620                 *oversoft = true;
 621                 return lqe->lqe_hardlimit;
 622         } else {
 623                 *oversoft = true;
 624                 return 0;
 625         }
 626 }
 627
 628 /*
 629  * Try to grant more quota space back to slave.
 630  *
 631  * \param lqe     - is the quota entry for which we would like to allocate more
 632  *                  space
 633  * \param granted - is how much was already granted as part of the request
 634  *                  processing
 635  * \param spare   - is how much unused quota space the slave already owns
 636  *
 637  * \retval return how additional space can be granted to the slave
 638  */
 639 __u64 qmt_alloc_expand(struct lquota_entry *lqe, __u64 granted, __u64 spare)
 640 {
 641         struct qmt_pool_info    *pool = lqe2qpi(lqe);
 642         __u64                    remaining, qunit;
 643         int                      slv_cnt;
 644
 645         LASSERT(lqe->lqe_enforced && lqe->lqe_qunit != 0);
 646
 647         slv_cnt = qpi_slv_nr(lqe2qpi(lqe), lqe_qtype(lqe));
 648         qunit = lqe->lqe_qunit;
 649
 650         /* See comment in qmt_adjust_qunit(). LU-4139. */
 651         if (lqe->lqe_softlimit != 0) {
 652                 bool oversoft;
 653                 remaining = qmt_calc_softlimit(lqe, &oversoft);
 654                 if (remaining == 0)
 655                         remaining = lqe->lqe_granted +
 656                                     pool->qpi_soft_least_qunit;
 657         } else {
 658                 remaining = lqe->lqe_hardlimit;
 659         }
 660
 661         if (lqe->lqe_granted >= remaining)
 662                 RETURN(0);
 663
 664         remaining -= lqe->lqe_granted;
 665
 666         do {
 667                 if (spare >= qunit)
 668                         break;
 669
 670                 granted &= (qunit - 1);
 671
 672                 if (remaining > (slv_cnt * qunit) >> 1) {
 673                         /* enough room to grant more space w/o additional
 674                          * shrinking ... at least for now */
 675                         remaining -= (slv_cnt * qunit) >> 1;
 676                 } else if (qunit != pool->qpi_least_qunit) {
 677                         qunit >>= 2;
 678                         continue;
 679                 }
 680
 681                 granted &= (qunit - 1);
 682                 if (spare > 0)
 683                         RETURN(min_t(__u64, qunit - spare, remaining));
 684                 else
 685                         RETURN(min_t(__u64, qunit - granted, remaining));
 686         } while (qunit >= pool->qpi_least_qunit);
 687
 688         RETURN(0);
 689 }
 690
 691 static inline void
 692 qmt_adjust_qunit_set_revoke(const struct lu_env *env, struct lquota_entry *lqe,
 693                             unsigned long least_qunit)
 694 {
 695         struct lquota_entry *lqe2;
 696         time64_t min = 0;
 697         int i;
 698
 699         if (qti_lqes_cnt(env) <= 1)
 700                 return;
 701
 702         for (i = 0; i < qti_lqes_cnt(env); i++) {
 703                 lqe2 = qti_lqes(env)[i];
 704                 if ((lqe2->lqe_qunit == least_qunit) && lqe2->lqe_revoke_time) {
 705                         if (!min) {
 706                                 min = lqe2->lqe_revoke_time;
 707                                 continue;
 708                         }
 709                         min = lqe2->lqe_revoke_time < min ?
 710                                 lqe2->lqe_revoke_time : min;
 711                 }
 712         }
 713
 714         lqe->lqe_revoke_time = min;
 715 }
 716
 717
 718 /*
 719  * Adjust qunit size according to quota limits and total granted count.
 720  * The caller must have locked the lqe.
 721  *
 722  * \param env - the environment passed by the caller
 723  * \param lqe - is the qid entry to be adjusted
 724  * \retval true - need reseed glbe array
 725  */
 726 bool qmt_adjust_qunit(const struct lu_env *env, struct lquota_entry *lqe)
 727 {
 728         struct qmt_pool_info    *pool = lqe2qpi(lqe);
 729         bool                     need_reseed = false;
 730         int                      slv_cnt;
 731         __u64                    qunit, limit, qunit2 = 0;
 732         ENTRY;
 733
 734         LASSERT(lqe_is_locked(lqe));
 735
 736         if (!lqe->lqe_enforced || lqe->lqe_id.qid_uid == 0)
 737                 /* no quota limits */
 738                 RETURN(need_reseed);
 739
 740         /* record how many slaves have already registered */
 741         slv_cnt = qpi_slv_nr(pool, lqe_qtype(lqe));
 742         if (slv_cnt == 0) {
 743                 /* Pool hasn't slaves anymore. Qunit will be adjusted
 744                  * again when new slaves would be added. */
 745                 if (lqe->lqe_qunit) {
 746                         qunit = 0;
 747                         GOTO(done, qunit);
 748                 }
 749                 /* wait for at least one slave to join */
 750                 RETURN(need_reseed);
 751         }
 752
 753         /* Qunit calculation is based on soft limit, if any, hard limit
 754          * otherwise. This means that qunit is shrunk to the minimum when
 755          * beyond the soft limit. This will impact performance, but that's the
 756          * price of an accurate grace time management. */
 757         if (lqe->lqe_softlimit != 0) {
 758                 bool oversoft;
 759                 /* As a compromise of write performance and the grace time
 760                  * accuracy, the block qunit size will be shrunk to
 761                  * qpi_soft_least_qunit when over softlimit. LU-4139. */
 762                 limit = qmt_calc_softlimit(lqe, &oversoft);
 763                 if (oversoft)
 764                         qunit2 = pool->qpi_soft_least_qunit;
 765                 if (limit == 0)
 766                         GOTO(done, qunit = qunit2);
 767         } else if (lqe->lqe_hardlimit != 0) {
 768                 limit = lqe->lqe_hardlimit;
 769         } else {
 770                 LQUOTA_ERROR(lqe, "enforced bit set, but neither hard nor soft "
 771                              "limit are set");
 772                 RETURN(need_reseed);
 773         }
 774
 775         qunit = lqe->lqe_qunit == 0 ? pool->qpi_least_qunit : lqe->lqe_qunit;
 776
 777         /* The qunit value is computed as follows: limit / (2 * slv_cnt).
 778          * Then 75% of the quota space can be granted with current qunit value.
 779          * The remaining 25% are then used with reduced qunit size (by a factor
 780          * of 4) which is then divided in a similar manner.
 781          *
 782          * |---------------------limit---------------------|
 783          * |-------limit / 2-------|-limit / 4-|-limit / 4-|
 784          * |qunit|qunit|qunit|qunit|           |           |
 785          * |----slv_cnt * qunit----|           |           |
 786          * |-grow limit-|          |           |           |
 787          * |--------------shrink limit---------|           |
 788          * |---space granted in qunit chunks---|-remaining-|
 789          *                                    /             \
 790          *                                   /               \
 791          *                                  /                 \
 792          *                                 /                   \
 793          *                                /                     \
 794          *     qunit >>= 2;            |qunit*slv_cnt|qunit*slv_cnt|
 795          *                             |---space in qunit---|remain|
 796          *                                  ...                               */
 797         if (qunit == pool->qpi_least_qunit ||
 798             limit >= lqe->lqe_granted + ((slv_cnt * qunit) >> 1)) {
 799                 /* current qunit value still fits, let's see if we can afford to
 800                  * increase qunit now ...
 801                  * To increase qunit again, we have to be under 25% */
 802                 while (qunit && limit >= lqe->lqe_granted + 6 * qunit * slv_cnt)
 803                         qunit <<= 2;
 804
 805                 if (!qunit) {
 806                         qunit = limit;
 807                         do_div(qunit, 2 * slv_cnt);
 808                 }
 809
 810         } else {
 811                 /* shrink qunit until we find a suitable value */
 812                 while (qunit > pool->qpi_least_qunit &&
 813                        limit < lqe->lqe_granted + ((slv_cnt * qunit) >> 1))
 814                         qunit >>= 2;
 815         }
 816
 817         if (qunit2 && qunit > qunit2)
 818                 qunit = qunit2;
 819 done:
 820         if (lqe->lqe_qunit == qunit)
 821                 /* keep current qunit */
 822                 RETURN(need_reseed);
 823
 824         LQUOTA_DEBUG(lqe, "%s qunit to %llu",
 825                      lqe->lqe_qunit < qunit ? "increasing" : "decreasing",
 826                      qunit);
 827
 828         /* store new qunit value */
 829         swap(lqe->lqe_qunit, qunit);
 830
 831         /* reseed glbe array and notify
 832          * slave if qunit was shrinked */
 833         need_reseed = true;
 834         /* reset revoke time */
 835         lqe->lqe_revoke_time = 0;
 836
 837         if (lqe->lqe_qunit == pool->qpi_least_qunit) {
 838                 if (lqe->lqe_qunit >= qunit)
 839                         /* initial qunit value is the smallest one */
 840                         lqe->lqe_revoke_time = ktime_get_seconds();
 841                 /* If there are several lqes and lqe_revoke_time is set for
 842                  * some of them, it means appropriate OSTs have been already
 843                  * notified with the least qunit and there is no chance to
 844                  * free more space. Find an lqe with the minimum(earliest)
 845                  * revoke_time and set this time to the current one.
 846                  */
 847                 qmt_adjust_qunit_set_revoke(env, lqe, pool->qpi_least_qunit);
 848         }
 849         RETURN(need_reseed);
 850 }
 851
 852 bool qmt_adjust_edquot_qunit_notify(const struct lu_env *env,
 853                                     struct qmt_device *qmt,
 854                                     __u64 now, bool edquot,
 855                                     bool qunit, __u32 qb_flags,
 856                                     int idx)
 857 {
 858         struct lquota_entry *lqe_gl, *lqe;
 859         bool reseed = false;
 860         bool notify = false;
 861         int i;
 862
 863         lqe_gl = qti_lqes_glbl(env);
 864
 865         for (i = 0; i < qti_lqes_cnt(env); i++) {
 866                 lqe = qti_lqes(env)[i];
 867                 if (qunit)
 868                         reseed |= qmt_adjust_qunit(env, lqe);
 869                 if (edquot)
 870                         reseed |= qmt_adjust_edquot(lqe, now);
 871         }
 872
 873         LASSERT(lqe_gl);
 874         if (!lqe_gl->lqe_glbl_data &&
 875             (req_has_rep(qb_flags) || req_is_rel(qb_flags))) {
 876                 if (reseed)
 877                         CDEBUG(D_QUOTA,
 878                                "%s: can not notify - lge_glbl_data is not set\n",
 879                                qmt->qmt_svname);
 880                 return reseed;
 881         }
 882
 883         if (reseed || idx >= 0) {
 884                 mutex_lock(&lqe_gl->lqe_glbl_data_lock);
 885                 if (lqe_gl->lqe_glbl_data) {
 886                         struct lqe_glbl_data *lgd = lqe_gl->lqe_glbl_data;
 887
 888                         if (reseed) {
 889                                 qmt_seed_glbe_all(env, lgd, qunit, edquot,
 890                                                   false);
 891                         } else if (idx >= 0) {
 892                                 int lge_idx = qmt_map_lge_idx(lgd, idx);
 893
 894                                 /* If there are no locks yet when
 895                                  * lge_qunit/edquot_nu is set, slaves
 896                                  * are still not notified with new
 897                                  * qunit/edquot value. In a such case
 898                                  * we need to notify them with new values to
 899                                  * avoid endless EINPROGRESS if qunit is equal
 900                                  * to the least qunit, but lqe_revoke_time is
 901                                  * still not set.
 902                                  */
 903                                 notify = lgd->lqeg_arr[lge_idx].lge_qunit_nu ||
 904                                          lgd->lqeg_arr[lge_idx].lge_edquot_nu;
 905                         }
 906                 }
 907                 mutex_unlock(&lqe_gl->lqe_glbl_data_lock);
 908         }
 909
 910         if (reseed || notify)
 911                 qmt_id_lock_notify(qmt, lqe_gl);
 912
 913         return reseed;
 914 }
 915
 916
 917 /*
 918  * Adjust qunit & edquot flag in case it wasn't initialized already (e.g.
 919  * limit set while no slaves were connected yet)
 920  */
 921 bool qmt_revalidate(const struct lu_env *env, struct lquota_entry *lqe)
 922 {
 923         bool need_notify = false;
 924
 925         if (lqe->lqe_qunit == 0) {
 926                 /* lqe was read from disk, but neither qunit, nor edquot flag
 927                  * were initialized */
 928                 need_notify = qmt_adjust_qunit(env, lqe);
 929                 if (lqe->lqe_qunit != 0)
 930                         need_notify |= qmt_adjust_edquot(lqe,
 931                                                 ktime_get_real_seconds());
 932         }
 933
 934         return need_notify;
 935 }
 936
 937 void qmt_revalidate_lqes(const struct lu_env *env,
 938                          struct qmt_device *qmt, __u32 qb_flags)
 939 {
 940         struct lquota_entry *lqe_gl = qti_lqes_glbl(env);
 941         bool need_notify = false;
 942         int i;
 943
 944         for (i = 0; i < qti_lqes_cnt(env); i++)
 945                 need_notify |= qmt_revalidate(env, qti_lqes(env)[i]);
 946
 947         if (!need_notify)
 948                 return;
 949
 950         /* There could be no ID lock to the moment of reconciliation.
 951          * As a result lqe global data is not initialised yet. It is ok
 952          * for release and report requests. */
 953         if (!lqe_gl->lqe_glbl_data &&
 954             (req_is_rel(qb_flags) || req_has_rep(qb_flags))) {
 955                 return;
 956         }
 957
 958         mutex_lock(&lqe_gl->lqe_glbl_data_lock);
 959         if (lqe_gl->lqe_glbl_data)
 960                 qmt_seed_glbe(env, lqe_gl->lqe_glbl_data, false);
 961         mutex_unlock(&lqe_gl->lqe_glbl_data_lock);
 962
 963         qmt_id_lock_notify(qmt, lqe_gl);
 964 }
 965
 966 void qti_lqes_init(const struct lu_env *env)
 967 {
 968         struct qmt_thread_info  *qti = qmt_info(env);
 969
 970         qti->qti_lqes_cnt = 0;
 971         qti->qti_glbl_lqe_idx = 0;
 972         qti->qti_lqes_num = QMT_MAX_POOL_NUM;
 973 }
 974
 975 int qti_lqes_add(const struct lu_env *env, struct lquota_entry *lqe)
 976 {
 977         struct qmt_thread_info  *qti = qmt_info(env);
 978
 979         if (qti->qti_lqes_cnt >= qti->qti_lqes_num) {
 980                 struct lquota_entry     **lqes;
 981                 lqes = qti->qti_lqes;
 982                 OBD_ALLOC(lqes, sizeof(lqe) * qti->qti_lqes_num * 2);
 983                 if (!lqes)
 984                         return -ENOMEM;
 985                 memcpy(lqes, qti_lqes(env), qti->qti_lqes_cnt * sizeof(lqe));
 986                 /* Don't need to free, if it is the very 1st allocation */
 987                 if (qti->qti_lqes_num > QMT_MAX_POOL_NUM)
 988                         OBD_FREE(qti->qti_lqes,
 989                                  qti->qti_lqes_num * sizeof(lqe));
 990                 qti->qti_lqes = lqes;
 991                 qti->qti_lqes_num *= 2;
 992         }
 993
 994         if (lqe->lqe_is_global)
 995                 qti->qti_glbl_lqe_idx = qti->qti_lqes_cnt;
 996         qti_lqes(env)[qti->qti_lqes_cnt++] = lqe;
 997
 998         /* The pool could be accessed directly from lqe, so take
 999          * extra reference that is put in qti_lqes_fini */
1000         qpi_getref(lqe2qpi(lqe));
1001
1002         CDEBUG(D_QUOTA, "LQE %px %lu is added, lqe_cnt %d lqes_num %d\n",
1003                          lqe, (long unsigned)lqe->lqe_id.qid_uid,
1004                          qti->qti_lqes_cnt, qti->qti_lqes_num);
1005         LASSERT(qti->qti_lqes_num != 0);
1006
1007         return 0;
1008 }
1009
1010 void qti_lqes_del(const struct lu_env *env, int index)
1011 {
1012         struct lquota_entry     **lqes;
1013         int lqes_cnt = qti_lqes_cnt(env);
1014         int lqep_size = sizeof(struct lquota_entry *);
1015
1016         if (index == 0) {
1017                 /* We can't handle non global lqes correctly without
1018                  * global lqe located at index 0. If we try to do so,
1019                  * something goes wrong. */
1020                 LQUOTA_ERROR(qti_lqes_glbl(env),
1021                              "quota: cannot remove lqe at index 0 as it is global");
1022                 LASSERT(qti_lqes_glbl(env)->lqe_is_global);
1023                 return;
1024         }
1025         lqes = qti_lqes(env);
1026         qpi_putref(env, lqe2qpi(lqes[index]));
1027         lqe_putref(lqes[index]);
1028         memcpy((unsigned char *)lqes + index * lqep_size,
1029                (unsigned char *)lqes + (index + 1) * lqep_size,
1030                (lqes_cnt - index - 1) * lqep_size);
1031         qti_lqes_cnt(env)--;
1032 }
1033
1034 void qti_lqes_fini(const struct lu_env *env)
1035 {
1036         struct qmt_thread_info  *qti = qmt_info(env);
1037         struct lquota_entry     **lqes = qti->qti_lqes;
1038         int i;
1039
1040         lqes = qti_lqes(env);
1041         for (i = 0; i < qti->qti_lqes_cnt; i++) {
1042                 qpi_putref(env, lqe2qpi(lqes[i]));
1043                 lqe_putref(lqes[i]);
1044         }
1045
1046         if (qti->qti_lqes_num > QMT_MAX_POOL_NUM)
1047                 OBD_FREE(qti->qti_lqes,
1048                          qti->qti_lqes_num * sizeof(struct lquota_entry *));
1049
1050         qti->qti_lqes_num = 0;
1051         qti->qti_lqes_cnt = 0;
1052 }
1053
1054 __u64 qti_lqes_min_qunit(const struct lu_env *env)
1055 {
1056         __u64 min, qunit;
1057         int i;
1058
1059         for (i = 1, min = qti_lqe_qunit(env, 0); i < qti_lqes_cnt(env); i++) {
1060                 qunit = qti_lqe_qunit(env, i);
1061                 /* if qunit is 0, lqe is not enforced and we can ignore it */
1062                 if (qunit && qunit < min)
1063                         min = qunit;
1064         }
1065
1066         return min;
1067 }
1068
1069 int qti_lqes_edquot(const struct lu_env *env)
1070 {
1071         int i;
1072
1073         for (i = 0; i < qti_lqes_cnt(env); i++) {
1074                 if (qti_lqes(env)[i]->lqe_edquot)
1075                         return 1;
1076         }
1077
1078         return 0;
1079 }
1080
1081 int qti_lqes_restore_init(const struct lu_env *env)
1082 {
1083         int rc = 0;
1084
1085         if (qti_lqes_inited(env) && qti_lqes_cnt(env) > QMT_MAX_POOL_NUM) {
1086                 OBD_ALLOC(qmt_info(env)->qti_lqes_rstr,
1087                           qti_lqes_cnt(env) * sizeof(struct qmt_lqe_restore));
1088                 if (!qmt_info(env)->qti_lqes_rstr)
1089                         rc = -ENOMEM;
1090         }
1091
1092         return rc;
1093 }
1094
1095 void qti_lqes_restore_fini(const struct lu_env *env)
1096 {
1097         if (qti_lqes_inited(env) && qti_lqes_cnt(env) > QMT_MAX_POOL_NUM)
1098                 OBD_FREE(qmt_info(env)->qti_lqes_rstr,
1099                          qti_lqes_cnt(env) * sizeof(struct qmt_lqe_restore));
1100 }
1101
1102 void qti_lqes_write_lock(const struct lu_env *env)
1103 {
1104         int i;
1105
1106         for (i = 0; i < qti_lqes_cnt(env); i++)
1107                 lqe_write_lock(qti_lqes(env)[i]);
1108 }
1109
1110 void qti_lqes_write_unlock(const struct lu_env *env)
1111 {
1112         int i;
1113
1114         for (i = 0; i < qti_lqes_cnt(env); i++)
1115                 lqe_write_unlock(qti_lqes(env)[i]);
1116 }
1117
1118 #define QMT_INIT_SLV_CNT 64
1119 struct lqe_glbl_data *qmt_alloc_lqe_gd(struct qmt_pool_info *pool, int qtype)
1120 {
1121         struct lqe_glbl_data    *lgd;
1122         struct lqe_glbl_entry   *lqeg_arr;
1123         int                      slv_cnt, glbe_num;
1124
1125         OBD_ALLOC(lgd, sizeof(struct lqe_glbl_data));
1126         if (!lgd)
1127                 RETURN(NULL);
1128
1129         slv_cnt = qpi_slv_nr_by_rtype(pool, qtype);
1130
1131         glbe_num = slv_cnt < QMT_INIT_SLV_CNT ? QMT_INIT_SLV_CNT : slv_cnt;
1132         OBD_ALLOC(lqeg_arr, sizeof(struct lqe_glbl_entry) * glbe_num);
1133         if (!lqeg_arr) {
1134                 OBD_FREE(lgd, sizeof(struct lqe_glbl_data));
1135                 RETURN(NULL);
1136         }
1137
1138         CDEBUG(D_QUOTA, "slv_cnt %d glbe_num %d\n", slv_cnt, glbe_num);
1139
1140         lgd->lqeg_num_used = slv_cnt;
1141         lgd->lqeg_num_alloc = glbe_num;
1142         lgd->lqeg_arr = lqeg_arr;
1143
1144         RETURN(lgd);
1145 }
1146
1147 void qmt_free_lqe_gd(struct lqe_glbl_data *lgd)
1148 {
1149         if (unlikely(!lgd))
1150                 return;
1151
1152         OBD_FREE(lgd->lqeg_arr,
1153                  sizeof(struct lqe_glbl_entry) * lgd->lqeg_num_alloc);
1154         OBD_FREE(lgd, sizeof(struct lqe_glbl_data));
1155 }
1156
1157 int qmt_map_lge_idx(struct lqe_glbl_data *lgd, int ostidx)
1158 {
1159         int k;
1160
1161         /* check common case of sequential OST numbers first */
1162         if (ostidx < lgd->lqeg_num_used &&
1163             lgd->lqeg_arr[ostidx].lge_idx == ostidx)
1164                 return ostidx;
1165
1166         for (k = 0; k < lgd->lqeg_num_used; k++)
1167                 if (lgd->lqeg_arr[k].lge_idx == ostidx)
1168                         break;
1169
1170         LASSERTF(k < lgd->lqeg_num_used, "Cannot map ostidx %d for %p\n",
1171                  ostidx, lgd);
1172         return k;
1173 }
1174
1175 void qmt_seed_glbe_all(const struct lu_env *env, struct lqe_glbl_data *lgd,
1176                        bool qunit, bool edquot, bool pool_locked)
1177 {
1178         struct qmt_pool_info *qpi;
1179         int i, j;
1180         ENTRY;
1181
1182         if (!qti_lqes_cnt(env))
1183                 RETURN_EXIT;
1184         /* lqes array is sorted by qunit - the first entry has minimum qunit.
1185          * Thus start seeding global qunit's array beginning from the 1st lqe
1186          * and appropriate pool. If pools overlapped, slaves from this
1187          * overlapping get minimum qunit value.
1188          * user1: pool1, pool2, pool_glbl;
1189          * pool1: OST1; user1_qunit = 10M;
1190          * pool2: OST0, OST1, OST2; user1_qunit = 30M;
1191          * pool_glbl: OST0, OST1, OST2, OST3; user1_qunit = 160M;
1192          * qunit array after seeding should be:
1193          * OST0: 30M; OST1: 10M; OST2: 30M; OST3: 160M; */
1194
1195         /* edquot resetup algorythm works fine
1196          * with not sorted lqes */
1197         if (qunit)
1198                 qmt_lqes_sort(env);
1199
1200         for (i = 0; i < lgd->lqeg_num_used; i++) {
1201                 lgd->lqeg_arr[i].lge_qunit_set = 0;
1202                 lgd->lqeg_arr[i].lge_qunit_nu = 0;
1203                 lgd->lqeg_arr[i].lge_edquot_nu = 0;
1204         }
1205
1206         for (i = 0; i < qti_lqes_cnt(env); i++) {
1207                 struct lquota_entry *lqe = qti_lqes(env)[i];
1208                 int slaves_cnt;
1209
1210                 CDEBUG(D_QUOTA, "lqes_cnt %d, i %d\n", qti_lqes_cnt(env), i);
1211                 qpi = lqe2qpi(lqe);
1212                 if (!pool_locked)
1213                         qmt_sarr_read_down(qpi);
1214
1215                 slaves_cnt = qmt_sarr_count(qpi);
1216
1217                 for (j = 0; j < slaves_cnt; j++) {
1218                         int idx, tgt_idx;
1219
1220                         tgt_idx = qmt_sarr_get_idx(qpi, j);
1221                         LASSERT(tgt_idx >= 0);
1222                         idx = qmt_map_lge_idx(lgd, tgt_idx);
1223
1224                         if (edquot) {
1225                                 int lge_edquot, new_edquot, edquot_nu;
1226
1227                                 lge_edquot = lgd->lqeg_arr[idx].lge_edquot;
1228                                 edquot_nu = lgd->lqeg_arr[idx].lge_edquot_nu;
1229                                 new_edquot = lqe->lqe_edquot;
1230
1231                                 if (lge_edquot == new_edquot ||
1232                                     (edquot_nu && lge_edquot == 1))
1233                                         goto qunit_lbl;
1234                                 lgd->lqeg_arr[idx].lge_edquot = new_edquot;
1235                                 /* it is needed for the following case:
1236                                  * initial values for idx i -
1237                                  * lqe_edquot = 1, lqe_edquot_nu == 0;
1238                                  * 1: new_edquot == 0 ->
1239                                  *      lqe_edquot = 0, lqe_edquot_nu = 1;
1240                                  * 2: new_edquot == 1 ->
1241                                  *      lqe_edquot = 1, lqe_edquot_nu = 0;
1242                                  * At the 2nd iteration lge_edquot comes back
1243                                  * to 1, so no changes and we don't need
1244                                  * to notify slave. */
1245                                 lgd->lqeg_arr[idx].lge_edquot_nu = !edquot_nu;
1246                         }
1247 qunit_lbl:
1248                         if (qunit) {
1249                                 __u64 lge_qunit, new_qunit;
1250
1251                                 CDEBUG(D_QUOTA,
1252                                        "tgt_idx %d idx %d lge_qunit_set %d lge_qunit %llu new_qunit %llu\n",
1253                                        tgt_idx, idx,
1254                                        lgd->lqeg_arr[idx].lge_qunit_set,
1255                                        lgd->lqeg_arr[idx].lge_qunit,
1256                                        lqe->lqe_qunit);
1257                                 /* lge for this idx is already set
1258                                  * on previous iteration */
1259                                 if (lgd->lqeg_arr[idx].lge_qunit_set)
1260                                         continue;
1261                                 lge_qunit = lgd->lqeg_arr[idx].lge_qunit;
1262                                 new_qunit = lqe->lqe_qunit;
1263                                 /* qunit could be not set,
1264                                  * so use global lqe's qunit */
1265                                 if (!new_qunit)
1266                                         continue;
1267
1268                                 if (lge_qunit != new_qunit)
1269                                         lgd->lqeg_arr[idx].lge_qunit =
1270                                                                 new_qunit;
1271
1272                                 /* TODO: initially slaves notification was done
1273                                  * only for qunit shrinking. Should we always
1274                                  * notify slaves with new qunit ? */
1275                                 if (lge_qunit > new_qunit)
1276                                         lgd->lqeg_arr[idx].lge_qunit_nu = 1;
1277                                 lgd->lqeg_arr[idx].lge_qunit_set = 1;
1278                         }
1279                 }
1280
1281                 if (!pool_locked)
1282                         qmt_sarr_read_up(qpi);
1283         }
1284         /* TODO: only for debug purposes - remove it later */
1285         for (i = 0; i < lgd->lqeg_num_used; i++)
1286                 CDEBUG(D_QUOTA,
1287                         "lgd i %d tgt_idx %d qunit %lu nu %d;  edquot %d nu %d\n",
1288                         i, lgd->lqeg_arr[i].lge_idx,
1289                         (unsigned long)lgd->lqeg_arr[i].lge_qunit,
1290                         lgd->lqeg_arr[i].lge_qunit_nu,
1291                         lgd->lqeg_arr[i].lge_edquot,
1292                         lgd->lqeg_arr[i].lge_edquot_nu);
1293
1294         EXIT;
1295 }
1296
1297 void qmt_setup_lqe_gd(const struct lu_env *env, struct qmt_device *qmt,
1298                       struct lquota_entry *lqe, struct lqe_glbl_data *lgd,
1299                       int pool_type)
1300 {
1301         __u64 qunit;
1302         bool edquot;
1303         int i;
1304
1305         qunit = lqe->lqe_qunit;
1306         edquot = lqe->lqe_edquot;
1307
1308         /* Firstly set all elements in array with
1309          * qunit and edquot of global pool */
1310         qmt_sarr_read_down(lqe2qpi(lqe));
1311         for (i = 0; i < lgd->lqeg_num_used; i++) {
1312                 lgd->lqeg_arr[i].lge_qunit = qunit;
1313                 lgd->lqeg_arr[i].lge_edquot = edquot;
1314                 /* It is the very first lvb setup - qunit and other flags
1315                  * will be sent to slaves during qmt_lvbo_fill. */
1316                 lgd->lqeg_arr[i].lge_qunit_nu = 0;
1317                 lgd->lqeg_arr[i].lge_edquot_nu = 0;
1318                 lgd->lqeg_arr[i].lge_idx = qmt_sarr_get_idx(lqe2qpi(lqe), i);
1319         }
1320         qmt_sarr_read_up(lqe2qpi(lqe));
1321
1322         qmt_pool_lqes_lookup_spec(env, qmt, pool_type,
1323                                   lqe_qtype(lqe), &lqe->lqe_id);
1324         qmt_seed_glbe(env, lgd, false);
1325
1326         mutex_lock(&lqe->lqe_glbl_data_lock);
1327         if (lqe->lqe_glbl_data == NULL) {
1328                 lqe->lqe_glbl_data = lgd;
1329                 lgd = NULL;
1330         }
1331         mutex_unlock(&lqe->lqe_glbl_data_lock);
1332         if (lgd)
1333                 qmt_free_lqe_gd(lgd);
1334
1335         qmt_id_lock_notify(qmt, lqe);
1336
1337         qti_lqes_fini(env);
1338 }