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                             !lqe->lqe_gl && list_empty(&lqe->lqe_link)) {
 566                                 /* LU-16736: the revoke_time should be set when
 567                                  * the qunit reachs the least qunit, the quota
 568                                  * LDLM lock could encounter some issue, setting
 569                                  * it to avoid endless wait in QSD. */
 570                                 LQUOTA_ERROR(lqe, "set revoke_time explicitly");
 571
 572                                 lqe->lqe_revoke_time = ktime_get_seconds();
 573                                 RETURN(false);
 574                         }
 575
 576                         /* Let's give more time to slave to release space */
 577                         lapse = ktime_get_seconds() - QMT_REBA_TIMEOUT;
 578                         if (lqe->lqe_may_rel != 0 && lqe->lqe_revoke_time > lapse)
 579                                 RETURN(false);
 580                 } else {
 581                         if (lqe->lqe_qunit > pool->qpi_soft_least_qunit)
 582                                 RETURN(false);
 583                 }
 584
 585                 /* set edquot flag */
 586                 lqe->lqe_edquot = true;
 587         } else {
 588                 /* space exhausted flag set, let's check whether it is time to
 589                  * clear it */
 590
 591                 if (qmt_space_exhausted(lqe, now))
 592                         /* the qmt still has not space */
 593                         RETURN(false);
 594
 595                 if (lqe->lqe_hardlimit != 0 &&
 596                     lqe->lqe_granted + pool->qpi_least_qunit >
 597                                                         lqe->lqe_hardlimit)
 598                         /* we clear the flag only once at least one least qunit
 599                          * is available */
 600                         RETURN(false);
 601
 602                 /* clear edquot flag */
 603                 lqe->lqe_edquot = false;
 604         }
 605
 606         LQUOTA_DEBUG(lqe, "changing edquot flag");
 607
 608         /* let's notify slave by issuing glimpse on per-ID lock.
 609          * the rebalance thread will take care of this */
 610         RETURN(true);
 611 }
 612
 613 /* Using least_qunit when over block softlimit will seriously impact the
 614  * write performance, we need to do some special tweaking on that. */
 615 static __u64 qmt_calc_softlimit(struct lquota_entry *lqe, bool *oversoft)
 616 {
 617         struct qmt_pool_info *pool = lqe2qpi(lqe);
 618
 619         LASSERT(lqe->lqe_softlimit != 0);
 620         *oversoft = false;
 621         /* No need to do special tweaking for inode limit */
 622         if (pool->qpi_rtype != LQUOTA_RES_DT)
 623                 return lqe->lqe_softlimit;
 624
 625         if (lqe->lqe_granted <= lqe->lqe_softlimit +
 626                                 pool->qpi_soft_least_qunit) {
 627                 return lqe->lqe_softlimit;
 628         } else if (lqe->lqe_hardlimit != 0) {
 629                 *oversoft = true;
 630                 return lqe->lqe_hardlimit;
 631         } else {
 632                 *oversoft = true;
 633                 return 0;
 634         }
 635 }
 636
 637 /*
 638  * Try to grant more quota space back to slave.
 639  *
 640  * \param lqe     - is the quota entry for which we would like to allocate more
 641  *                  space
 642  * \param granted - is how much was already granted as part of the request
 643  *                  processing
 644  * \param spare   - is how much unused quota space the slave already owns
 645  *
 646  * \retval return how additional space can be granted to the slave
 647  */
 648 __u64 qmt_alloc_expand(struct lquota_entry *lqe, __u64 granted, __u64 spare)
 649 {
 650         struct qmt_pool_info    *pool = lqe2qpi(lqe);
 651         __u64                    remaining, qunit;
 652         int                      slv_cnt;
 653
 654         LASSERT(lqe->lqe_enforced && lqe->lqe_qunit != 0);
 655
 656         slv_cnt = qpi_slv_nr(lqe2qpi(lqe), lqe_qtype(lqe));
 657         qunit = lqe->lqe_qunit;
 658
 659         /* See comment in qmt_adjust_qunit(). LU-4139. */
 660         if (lqe->lqe_softlimit != 0) {
 661                 bool oversoft;
 662                 remaining = qmt_calc_softlimit(lqe, &oversoft);
 663                 if (remaining == 0)
 664                         remaining = lqe->lqe_granted +
 665                                     pool->qpi_soft_least_qunit;
 666         } else {
 667                 remaining = lqe->lqe_hardlimit;
 668         }
 669
 670         if (lqe->lqe_granted >= remaining)
 671                 RETURN(0);
 672
 673         remaining -= lqe->lqe_granted;
 674
 675         do {
 676                 if (spare >= qunit)
 677                         break;
 678
 679                 granted &= (qunit - 1);
 680
 681                 if (remaining > (slv_cnt * qunit) >> 1) {
 682                         /* enough room to grant more space w/o additional
 683                          * shrinking ... at least for now */
 684                         remaining -= (slv_cnt * qunit) >> 1;
 685                 } else if (qunit != pool->qpi_least_qunit) {
 686                         qunit >>= 2;
 687                         continue;
 688                 }
 689
 690                 granted &= (qunit - 1);
 691                 if (spare > 0)
 692                         RETURN(min_t(__u64, qunit - spare, remaining));
 693                 else
 694                         RETURN(min_t(__u64, qunit - granted, remaining));
 695         } while (qunit >= pool->qpi_least_qunit);
 696
 697         RETURN(0);
 698 }
 699
 700 static inline void
 701 qmt_adjust_qunit_set_revoke(const struct lu_env *env, struct lquota_entry *lqe,
 702                             unsigned long least_qunit)
 703 {
 704         struct lquota_entry *lqe2;
 705         time64_t min = 0;
 706         int i;
 707
 708         if (qti_lqes_cnt(env) <= 1)
 709                 return;
 710
 711         for (i = 0; i < qti_lqes_cnt(env); i++) {
 712                 lqe2 = qti_lqes(env)[i];
 713                 if ((lqe2->lqe_qunit == least_qunit) && lqe2->lqe_revoke_time) {
 714                         if (!min) {
 715                                 min = lqe2->lqe_revoke_time;
 716                                 continue;
 717                         }
 718                         min = lqe2->lqe_revoke_time < min ?
 719                                 lqe2->lqe_revoke_time : min;
 720                 }
 721         }
 722
 723         lqe->lqe_revoke_time = min;
 724 }
 725
 726
 727 /*
 728  * Adjust qunit size according to quota limits and total granted count.
 729  * The caller must have locked the lqe.
 730  *
 731  * \param env - the environment passed by the caller
 732  * \param lqe - is the qid entry to be adjusted
 733  * \retval true - need reseed glbe array
 734  */
 735 bool qmt_adjust_qunit(const struct lu_env *env, struct lquota_entry *lqe)
 736 {
 737         struct qmt_pool_info    *pool = lqe2qpi(lqe);
 738         bool                     need_reseed = false;
 739         int                      slv_cnt;
 740         __u64                    qunit, limit, qunit2 = 0;
 741         ENTRY;
 742
 743         LASSERT(lqe_is_locked(lqe));
 744
 745         if (!lqe->lqe_enforced || lqe->lqe_id.qid_uid == 0)
 746                 /* no quota limits */
 747                 RETURN(need_reseed);
 748
 749         /* record how many slaves have already registered */
 750         slv_cnt = qpi_slv_nr(pool, lqe_qtype(lqe));
 751         if (slv_cnt == 0) {
 752                 /* Pool hasn't slaves anymore. Qunit will be adjusted
 753                  * again when new slaves would be added. */
 754                 if (lqe->lqe_qunit) {
 755                         qunit = 0;
 756                         GOTO(done, qunit);
 757                 }
 758                 /* wait for at least one slave to join */
 759                 RETURN(need_reseed);
 760         }
 761
 762         /* Qunit calculation is based on soft limit, if any, hard limit
 763          * otherwise. This means that qunit is shrunk to the minimum when
 764          * beyond the soft limit. This will impact performance, but that's the
 765          * price of an accurate grace time management. */
 766         if (lqe->lqe_softlimit != 0) {
 767                 bool oversoft;
 768                 /* As a compromise of write performance and the grace time
 769                  * accuracy, the block qunit size will be shrunk to
 770                  * qpi_soft_least_qunit when over softlimit. LU-4139. */
 771                 limit = qmt_calc_softlimit(lqe, &oversoft);
 772                 if (oversoft)
 773                         qunit2 = pool->qpi_soft_least_qunit;
 774                 if (limit == 0)
 775                         GOTO(done, qunit = qunit2);
 776         } else if (lqe->lqe_hardlimit != 0) {
 777                 limit = lqe->lqe_hardlimit;
 778         } else {
 779                 LQUOTA_ERROR(lqe, "enforced bit set, but neither hard nor soft "
 780                              "limit are set");
 781                 RETURN(need_reseed);
 782         }
 783
 784         qunit = lqe->lqe_qunit == 0 ? pool->qpi_least_qunit : lqe->lqe_qunit;
 785
 786         /* The qunit value is computed as follows: limit / (2 * slv_cnt).
 787          * Then 75% of the quota space can be granted with current qunit value.
 788          * The remaining 25% are then used with reduced qunit size (by a factor
 789          * of 4) which is then divided in a similar manner.
 790          *
 791          * |---------------------limit---------------------|
 792          * |-------limit / 2-------|-limit / 4-|-limit / 4-|
 793          * |qunit|qunit|qunit|qunit|           |           |
 794          * |----slv_cnt * qunit----|           |           |
 795          * |-grow limit-|          |           |           |
 796          * |--------------shrink limit---------|           |
 797          * |---space granted in qunit chunks---|-remaining-|
 798          *                                    /             \
 799          *                                   /               \
 800          *                                  /                 \
 801          *                                 /                   \
 802          *                                /                     \
 803          *     qunit >>= 2;            |qunit*slv_cnt|qunit*slv_cnt|
 804          *                             |---space in qunit---|remain|
 805          *                                  ...                               */
 806         if (qunit == pool->qpi_least_qunit ||
 807             limit >= lqe->lqe_granted + ((slv_cnt * qunit) >> 1)) {
 808                 /* current qunit value still fits, let's see if we can afford to
 809                  * increase qunit now ...
 810                  * To increase qunit again, we have to be under 25% */
 811                 while (qunit && limit >= lqe->lqe_granted + 6 * qunit * slv_cnt)
 812                         qunit <<= 2;
 813
 814                 if (!qunit) {
 815                         qunit = limit;
 816                         do_div(qunit, 2 * slv_cnt);
 817                 }
 818
 819         } else {
 820                 /* shrink qunit until we find a suitable value */
 821                 while (qunit > pool->qpi_least_qunit &&
 822                        limit < lqe->lqe_granted + ((slv_cnt * qunit) >> 1))
 823                         qunit >>= 2;
 824         }
 825
 826         if (qunit2 && qunit > qunit2)
 827                 qunit = qunit2;
 828 done:
 829         if (lqe->lqe_qunit == qunit)
 830                 /* keep current qunit */
 831                 RETURN(need_reseed);
 832
 833         LQUOTA_DEBUG(lqe, "%s qunit to %llu",
 834                      lqe->lqe_qunit < qunit ? "increasing" : "decreasing",
 835                      qunit);
 836
 837         /* store new qunit value */
 838         swap(lqe->lqe_qunit, qunit);
 839
 840         /* reseed glbe array and notify
 841          * slave if qunit was shrinked */
 842         need_reseed = true;
 843         /* reset revoke time */
 844         lqe->lqe_revoke_time = 0;
 845
 846         if (lqe->lqe_qunit == pool->qpi_least_qunit) {
 847                 if (lqe->lqe_qunit >= qunit)
 848                         /* initial qunit value is the smallest one */
 849                         lqe->lqe_revoke_time = ktime_get_seconds();
 850                 /* If there are several lqes and lqe_revoke_time is set for
 851                  * some of them, it means appropriate OSTs have been already
 852                  * notified with the least qunit and there is no chance to
 853                  * free more space. Find an lqe with the minimum(earliest)
 854                  * revoke_time and set this time to the current one.
 855                  */
 856                 qmt_adjust_qunit_set_revoke(env, lqe, pool->qpi_least_qunit);
 857         }
 858         RETURN(need_reseed);
 859 }
 860
 861 bool qmt_adjust_edquot_qunit_notify(const struct lu_env *env,
 862                                     struct qmt_device *qmt,
 863                                     __u64 now, bool edquot,
 864                                     bool qunit, __u32 qb_flags,
 865                                     int idx)
 866 {
 867         struct lquota_entry *lqe_gl, *lqe;
 868         bool reseed = false;
 869         bool notify = false;
 870         int i;
 871
 872         lqe_gl = qti_lqes_glbl(env);
 873
 874         for (i = 0; i < qti_lqes_cnt(env); i++) {
 875                 lqe = qti_lqes(env)[i];
 876                 if (qunit)
 877                         reseed |= qmt_adjust_qunit(env, lqe);
 878                 if (edquot)
 879                         reseed |= qmt_adjust_edquot(lqe, now);
 880         }
 881
 882         LASSERT(lqe_gl);
 883         if (!lqe_gl->lqe_glbl_data &&
 884             (req_has_rep(qb_flags) || req_is_rel(qb_flags))) {
 885                 if (reseed)
 886                         CDEBUG(D_QUOTA,
 887                                "%s: can not notify - lge_glbl_data is not set\n",
 888                                qmt->qmt_svname);
 889                 return reseed;
 890         }
 891
 892         if (reseed || idx >= 0) {
 893                 mutex_lock(&lqe_gl->lqe_glbl_data_lock);
 894                 if (lqe_gl->lqe_glbl_data) {
 895                         struct lqe_glbl_data *lgd = lqe_gl->lqe_glbl_data;
 896
 897                         if (reseed) {
 898                                 qmt_seed_glbe_all(env, lgd, qunit, edquot,
 899                                                   false);
 900                         } else if (idx >= 0) {
 901                                 int lge_idx = qmt_map_lge_idx(lgd, idx);
 902
 903                                 /* If there are no locks yet when
 904                                  * lge_qunit/edquot_nu is set, slaves
 905                                  * are still not notified with new
 906                                  * qunit/edquot value. In a such case
 907                                  * we need to notify them with new values to
 908                                  * avoid endless EINPROGRESS if qunit is equal
 909                                  * to the least qunit, but lqe_revoke_time is
 910                                  * still not set.
 911                                  */
 912                                 notify = lgd->lqeg_arr[lge_idx].lge_qunit_nu ||
 913                                          lgd->lqeg_arr[lge_idx].lge_edquot_nu;
 914                         }
 915                 }
 916                 mutex_unlock(&lqe_gl->lqe_glbl_data_lock);
 917         }
 918
 919         if (reseed || notify)
 920                 qmt_id_lock_notify(qmt, lqe_gl);
 921
 922         return reseed;
 923 }
 924
 925
 926 /*
 927  * Adjust qunit & edquot flag in case it wasn't initialized already (e.g.
 928  * limit set while no slaves were connected yet)
 929  */
 930 bool qmt_revalidate(const struct lu_env *env, struct lquota_entry *lqe)
 931 {
 932         bool need_notify = false;
 933
 934         if (lqe->lqe_qunit == 0) {
 935                 /* lqe was read from disk, but neither qunit, nor edquot flag
 936                  * were initialized */
 937                 need_notify = qmt_adjust_qunit(env, lqe);
 938                 if (lqe->lqe_qunit != 0)
 939                         need_notify |= qmt_adjust_edquot(lqe,
 940                                                 ktime_get_real_seconds());
 941         }
 942
 943         return need_notify;
 944 }
 945
 946 void qmt_revalidate_lqes(const struct lu_env *env,
 947                          struct qmt_device *qmt, __u32 qb_flags)
 948 {
 949         struct lquota_entry *lqe_gl = qti_lqes_glbl(env);
 950         bool need_notify = false;
 951         int i;
 952
 953         for (i = 0; i < qti_lqes_cnt(env); i++)
 954                 need_notify |= qmt_revalidate(env, qti_lqes(env)[i]);
 955
 956         if (!need_notify)
 957                 return;
 958
 959         /* There could be no ID lock to the moment of reconciliation.
 960          * As a result lqe global data is not initialised yet. It is ok
 961          * for release and report requests. */
 962         if (!lqe_gl->lqe_glbl_data &&
 963             (req_is_rel(qb_flags) || req_has_rep(qb_flags))) {
 964                 return;
 965         }
 966
 967         mutex_lock(&lqe_gl->lqe_glbl_data_lock);
 968         if (lqe_gl->lqe_glbl_data)
 969                 qmt_seed_glbe(env, lqe_gl->lqe_glbl_data, false);
 970         mutex_unlock(&lqe_gl->lqe_glbl_data_lock);
 971
 972         qmt_id_lock_notify(qmt, lqe_gl);
 973 }
 974
 975 void qti_lqes_init(const struct lu_env *env)
 976 {
 977         struct qmt_thread_info  *qti = qmt_info(env);
 978
 979         qti->qti_lqes_cnt = 0;
 980         qti->qti_glbl_lqe_idx = 0;
 981         qti->qti_lqes_num = QMT_MAX_POOL_NUM;
 982 }
 983
 984 int qti_lqes_add(const struct lu_env *env, struct lquota_entry *lqe)
 985 {
 986         struct qmt_thread_info  *qti = qmt_info(env);
 987
 988         if (qti->qti_lqes_cnt >= qti->qti_lqes_num) {
 989                 struct lquota_entry     **lqes;
 990                 lqes = qti->qti_lqes;
 991                 OBD_ALLOC(lqes, sizeof(lqe) * qti->qti_lqes_num * 2);
 992                 if (!lqes)
 993                         return -ENOMEM;
 994                 memcpy(lqes, qti_lqes(env), qti->qti_lqes_cnt * sizeof(lqe));
 995                 /* Don't need to free, if it is the very 1st allocation */
 996                 if (qti->qti_lqes_num > QMT_MAX_POOL_NUM)
 997                         OBD_FREE(qti->qti_lqes,
 998                                  qti->qti_lqes_num * sizeof(lqe));
 999                 qti->qti_lqes = lqes;
1000                 qti->qti_lqes_num *= 2;
1001         }
1002
1003         if (lqe->lqe_is_global)
1004                 qti->qti_glbl_lqe_idx = qti->qti_lqes_cnt;
1005         qti_lqes(env)[qti->qti_lqes_cnt++] = lqe;
1006
1007         /* The pool could be accessed directly from lqe, so take
1008          * extra reference that is put in qti_lqes_fini */
1009         qpi_getref(lqe2qpi(lqe));
1010
1011         CDEBUG(D_QUOTA, "LQE %px %lu is added, lqe_cnt %d lqes_num %d\n",
1012                          lqe, (long unsigned)lqe->lqe_id.qid_uid,
1013                          qti->qti_lqes_cnt, qti->qti_lqes_num);
1014         LASSERT(qti->qti_lqes_num != 0);
1015
1016         return 0;
1017 }
1018
1019 void qti_lqes_del(const struct lu_env *env, int index)
1020 {
1021         struct lquota_entry     **lqes;
1022         int lqes_cnt = qti_lqes_cnt(env);
1023         int lqep_size = sizeof(struct lquota_entry *);
1024
1025         if (index == 0) {
1026                 /* We can't handle non global lqes correctly without
1027                  * global lqe located at index 0. If we try to do so,
1028                  * something goes wrong. */
1029                 LQUOTA_ERROR(qti_lqes_glbl(env),
1030                              "quota: cannot remove lqe at index 0 as it is global");
1031                 LASSERT(qti_lqes_glbl(env)->lqe_is_global);
1032                 return;
1033         }
1034         lqes = qti_lqes(env);
1035         qpi_putref(env, lqe2qpi(lqes[index]));
1036         lqe_putref(lqes[index]);
1037         memcpy((unsigned char *)lqes + index * lqep_size,
1038                (unsigned char *)lqes + (index + 1) * lqep_size,
1039                (lqes_cnt - index - 1) * lqep_size);
1040         qti_lqes_cnt(env)--;
1041 }
1042
1043 void qti_lqes_fini(const struct lu_env *env)
1044 {
1045         struct qmt_thread_info  *qti = qmt_info(env);
1046         struct lquota_entry     **lqes = qti->qti_lqes;
1047         int i;
1048
1049         lqes = qti_lqes(env);
1050         for (i = 0; i < qti->qti_lqes_cnt; i++) {
1051                 qpi_putref(env, lqe2qpi(lqes[i]));
1052                 lqe_putref(lqes[i]);
1053         }
1054
1055         if (qti->qti_lqes_num > QMT_MAX_POOL_NUM)
1056                 OBD_FREE(qti->qti_lqes,
1057                          qti->qti_lqes_num * sizeof(struct lquota_entry *));
1058
1059         qti->qti_lqes_num = 0;
1060         qti->qti_lqes_cnt = 0;
1061 }
1062
1063 __u64 qti_lqes_min_qunit(const struct lu_env *env)
1064 {
1065         __u64 min, qunit;
1066         int i;
1067
1068         for (i = 1, min = qti_lqe_qunit(env, 0); i < qti_lqes_cnt(env); i++) {
1069                 qunit = qti_lqe_qunit(env, i);
1070                 /* if qunit is 0, lqe is not enforced and we can ignore it */
1071                 if (qunit && qunit < min)
1072                         min = qunit;
1073         }
1074
1075         return min;
1076 }
1077
1078 int qti_lqes_edquot(const struct lu_env *env)
1079 {
1080         int i;
1081
1082         for (i = 0; i < qti_lqes_cnt(env); i++) {
1083                 if (qti_lqes(env)[i]->lqe_edquot)
1084                         return 1;
1085         }
1086
1087         return 0;
1088 }
1089
1090 int qti_lqes_restore_init(const struct lu_env *env)
1091 {
1092         int rc = 0;
1093
1094         if (qti_lqes_inited(env) && qti_lqes_cnt(env) > QMT_MAX_POOL_NUM) {
1095                 OBD_ALLOC(qmt_info(env)->qti_lqes_rstr,
1096                           qti_lqes_cnt(env) * sizeof(struct qmt_lqe_restore));
1097                 if (!qmt_info(env)->qti_lqes_rstr)
1098                         rc = -ENOMEM;
1099         }
1100
1101         return rc;
1102 }
1103
1104 void qti_lqes_restore_fini(const struct lu_env *env)
1105 {
1106         if (qti_lqes_inited(env) && qti_lqes_cnt(env) > QMT_MAX_POOL_NUM)
1107                 OBD_FREE(qmt_info(env)->qti_lqes_rstr,
1108                          qti_lqes_cnt(env) * sizeof(struct qmt_lqe_restore));
1109 }
1110
1111 void qti_lqes_write_lock(const struct lu_env *env)
1112 {
1113         int i;
1114
1115         for (i = 0; i < qti_lqes_cnt(env); i++)
1116                 lqe_write_lock(qti_lqes(env)[i]);
1117 }
1118
1119 void qti_lqes_write_unlock(const struct lu_env *env)
1120 {
1121         int i;
1122
1123         for (i = 0; i < qti_lqes_cnt(env); i++)
1124                 lqe_write_unlock(qti_lqes(env)[i]);
1125 }
1126
1127 #define QMT_INIT_SLV_CNT 64
1128 struct lqe_glbl_data *qmt_alloc_lqe_gd(struct qmt_pool_info *pool, int qtype)
1129 {
1130         struct lqe_glbl_data    *lgd;
1131         struct lqe_glbl_entry   *lqeg_arr;
1132         int                      slv_cnt, glbe_num;
1133
1134         OBD_ALLOC(lgd, sizeof(struct lqe_glbl_data));
1135         if (!lgd)
1136                 RETURN(NULL);
1137
1138         slv_cnt = qpi_slv_nr_by_rtype(pool, qtype);
1139
1140         glbe_num = slv_cnt < QMT_INIT_SLV_CNT ? QMT_INIT_SLV_CNT : slv_cnt;
1141         OBD_ALLOC(lqeg_arr, sizeof(struct lqe_glbl_entry) * glbe_num);
1142         if (!lqeg_arr) {
1143                 OBD_FREE(lgd, sizeof(struct lqe_glbl_data));
1144                 RETURN(NULL);
1145         }
1146
1147         CDEBUG(D_QUOTA, "slv_cnt %d glbe_num %d\n", slv_cnt, glbe_num);
1148
1149         lgd->lqeg_num_used = slv_cnt;
1150         lgd->lqeg_num_alloc = glbe_num;
1151         lgd->lqeg_arr = lqeg_arr;
1152
1153         RETURN(lgd);
1154 }
1155
1156 void qmt_free_lqe_gd(struct lqe_glbl_data *lgd)
1157 {
1158         if (unlikely(!lgd))
1159                 return;
1160
1161         OBD_FREE(lgd->lqeg_arr,
1162                  sizeof(struct lqe_glbl_entry) * lgd->lqeg_num_alloc);
1163         OBD_FREE(lgd, sizeof(struct lqe_glbl_data));
1164 }
1165
1166 int qmt_map_lge_idx(struct lqe_glbl_data *lgd, int ostidx)
1167 {
1168         int k;
1169
1170         /* check common case of sequential OST numbers first */
1171         if (ostidx < lgd->lqeg_num_used &&
1172             lgd->lqeg_arr[ostidx].lge_idx == ostidx)
1173                 return ostidx;
1174
1175         for (k = 0; k < lgd->lqeg_num_used; k++)
1176                 if (lgd->lqeg_arr[k].lge_idx == ostidx)
1177                         break;
1178
1179         LASSERTF(k < lgd->lqeg_num_used, "Cannot map ostidx %d for %p\n",
1180                  ostidx, lgd);
1181         return k;
1182 }
1183
1184 void qmt_seed_glbe_all(const struct lu_env *env, struct lqe_glbl_data *lgd,
1185                        bool qunit, bool edquot, bool pool_locked)
1186 {
1187         struct qmt_pool_info *qpi;
1188         int i, j;
1189         ENTRY;
1190
1191         if (!qti_lqes_cnt(env))
1192                 RETURN_EXIT;
1193         /* lqes array is sorted by qunit - the first entry has minimum qunit.
1194          * Thus start seeding global qunit's array beginning from the 1st lqe
1195          * and appropriate pool. If pools overlapped, slaves from this
1196          * overlapping get minimum qunit value.
1197          * user1: pool1, pool2, pool_glbl;
1198          * pool1: OST1; user1_qunit = 10M;
1199          * pool2: OST0, OST1, OST2; user1_qunit = 30M;
1200          * pool_glbl: OST0, OST1, OST2, OST3; user1_qunit = 160M;
1201          * qunit array after seeding should be:
1202          * OST0: 30M; OST1: 10M; OST2: 30M; OST3: 160M; */
1203
1204         /* edquot resetup algorythm works fine
1205          * with not sorted lqes */
1206         if (qunit)
1207                 qmt_lqes_sort(env);
1208
1209         for (i = 0; i < lgd->lqeg_num_used; i++) {
1210                 lgd->lqeg_arr[i].lge_qunit_set = 0;
1211                 lgd->lqeg_arr[i].lge_qunit_nu = 0;
1212                 lgd->lqeg_arr[i].lge_edquot_nu = 0;
1213         }
1214
1215         for (i = 0; i < qti_lqes_cnt(env); i++) {
1216                 struct lquota_entry *lqe = qti_lqes(env)[i];
1217                 int slaves_cnt;
1218
1219                 CDEBUG(D_QUOTA, "lqes_cnt %d, i %d\n", qti_lqes_cnt(env), i);
1220                 qpi = lqe2qpi(lqe);
1221                 if (!pool_locked)
1222                         qmt_sarr_read_down(qpi);
1223
1224                 slaves_cnt = qmt_sarr_count(qpi);
1225
1226                 for (j = 0; j < slaves_cnt; j++) {
1227                         int idx, tgt_idx;
1228
1229                         tgt_idx = qmt_sarr_get_idx(qpi, j);
1230                         LASSERT(tgt_idx >= 0);
1231                         idx = qmt_map_lge_idx(lgd, tgt_idx);
1232
1233                         if (edquot) {
1234                                 int lge_edquot, new_edquot, edquot_nu;
1235
1236                                 lge_edquot = lgd->lqeg_arr[idx].lge_edquot;
1237                                 edquot_nu = lgd->lqeg_arr[idx].lge_edquot_nu;
1238                                 new_edquot = lqe->lqe_edquot;
1239
1240                                 if (lge_edquot == new_edquot ||
1241                                     (edquot_nu && lge_edquot == 1))
1242                                         goto qunit_lbl;
1243                                 lgd->lqeg_arr[idx].lge_edquot = new_edquot;
1244                                 /* it is needed for the following case:
1245                                  * initial values for idx i -
1246                                  * lqe_edquot = 1, lqe_edquot_nu == 0;
1247                                  * 1: new_edquot == 0 ->
1248                                  *      lqe_edquot = 0, lqe_edquot_nu = 1;
1249                                  * 2: new_edquot == 1 ->
1250                                  *      lqe_edquot = 1, lqe_edquot_nu = 0;
1251                                  * At the 2nd iteration lge_edquot comes back
1252                                  * to 1, so no changes and we don't need
1253                                  * to notify slave. */
1254                                 lgd->lqeg_arr[idx].lge_edquot_nu = !edquot_nu;
1255                         }
1256 qunit_lbl:
1257                         if (qunit) {
1258                                 __u64 lge_qunit, new_qunit;
1259
1260                                 CDEBUG(D_QUOTA,
1261                                        "tgt_idx %d idx %d lge_qunit_set %d lge_qunit %llu new_qunit %llu\n",
1262                                        tgt_idx, idx,
1263                                        lgd->lqeg_arr[idx].lge_qunit_set,
1264                                        lgd->lqeg_arr[idx].lge_qunit,
1265                                        lqe->lqe_qunit);
1266                                 /* lge for this idx is already set
1267                                  * on previous iteration */
1268                                 if (lgd->lqeg_arr[idx].lge_qunit_set)
1269                                         continue;
1270                                 lge_qunit = lgd->lqeg_arr[idx].lge_qunit;
1271                                 new_qunit = lqe->lqe_qunit;
1272                                 /* qunit could be not set,
1273                                  * so use global lqe's qunit */
1274                                 if (!new_qunit)
1275                                         continue;
1276
1277                                 if (lge_qunit != new_qunit)
1278                                         lgd->lqeg_arr[idx].lge_qunit =
1279                                                                 new_qunit;
1280
1281                                 /* TODO: initially slaves notification was done
1282                                  * only for qunit shrinking. Should we always
1283                                  * notify slaves with new qunit ? */
1284                                 if (lge_qunit > new_qunit)
1285                                         lgd->lqeg_arr[idx].lge_qunit_nu = 1;
1286                                 lgd->lqeg_arr[idx].lge_qunit_set = 1;
1287                         }
1288                 }
1289
1290                 if (!pool_locked)
1291                         qmt_sarr_read_up(qpi);
1292         }
1293         /* TODO: only for debug purposes - remove it later */
1294         for (i = 0; i < lgd->lqeg_num_used; i++)
1295                 CDEBUG(D_QUOTA,
1296                         "lgd i %d tgt_idx %d qunit %lu nu %d;  edquot %d nu %d\n",
1297                         i, lgd->lqeg_arr[i].lge_idx,
1298                         (unsigned long)lgd->lqeg_arr[i].lge_qunit,
1299                         lgd->lqeg_arr[i].lge_qunit_nu,
1300                         lgd->lqeg_arr[i].lge_edquot,
1301                         lgd->lqeg_arr[i].lge_edquot_nu);
1302
1303         EXIT;
1304 }
1305
1306 void qmt_setup_lqe_gd(const struct lu_env *env, struct qmt_device *qmt,
1307                       struct lquota_entry *lqe, struct lqe_glbl_data *lgd,
1308                       int pool_type)
1309 {
1310         __u64 qunit;
1311         bool edquot;
1312         int i;
1313
1314         qunit = lqe->lqe_qunit;
1315         edquot = lqe->lqe_edquot;
1316
1317         /* Firstly set all elements in array with
1318          * qunit and edquot of global pool */
1319         qmt_sarr_read_down(lqe2qpi(lqe));
1320         for (i = 0; i < lgd->lqeg_num_used; i++) {
1321                 lgd->lqeg_arr[i].lge_qunit = qunit;
1322                 lgd->lqeg_arr[i].lge_edquot = edquot;
1323                 /* It is the very first lvb setup - qunit and other flags
1324                  * will be sent to slaves during qmt_lvbo_fill. */
1325                 lgd->lqeg_arr[i].lge_qunit_nu = 0;
1326                 lgd->lqeg_arr[i].lge_edquot_nu = 0;
1327                 lgd->lqeg_arr[i].lge_idx = qmt_sarr_get_idx(lqe2qpi(lqe), i);
1328         }
1329         qmt_sarr_read_up(lqe2qpi(lqe));
1330
1331         qmt_pool_lqes_lookup_spec(env, qmt, pool_type,
1332                                   lqe_qtype(lqe), &lqe->lqe_id);
1333         qmt_seed_glbe(env, lgd, false);
1334
1335         mutex_lock(&lqe->lqe_glbl_data_lock);
1336         if (lqe->lqe_glbl_data == NULL) {
1337                 lqe->lqe_glbl_data = lgd;
1338                 lgd = NULL;
1339         }
1340         mutex_unlock(&lqe->lqe_glbl_data_lock);
1341         if (lgd)
1342                 qmt_free_lqe_gd(lgd);
1343
1344         qmt_id_lock_notify(qmt, lqe);
1345
1346         qti_lqes_fini(env);
1347 }