Whamcloud - gitweb
LU-9859 libcfs: deprecate libcfs_debug_vmsg2
[fs/lustre-release.git] / 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 /*
36  * Initialize qmt-specific fields of quota entry.
37  *
38  * \param lqe - is the quota entry to initialize
39  * \param arg - is the pointer to the qmt_pool_info structure
40  */
41 static void qmt_lqe_init(struct lquota_entry *lqe, void *arg)
42 {
43         LASSERT(lqe_is_master(lqe));
44
45         lqe->lqe_revoke_time = 0;
46         init_rwsem(&lqe->lqe_sem);
47 }
48
49 /* Apply the default quota setting to the specified quota entry
50  *
51  * \param env           - is the environment passed by the caller
52  * \param pool          - is the quota pool of the quota entry
53  * \param lqe           - is the lquota_entry object to apply default quota on
54  * \param create_record - if true, an global quota record will be created and
55  *                        write to the disk.
56  *
57  * \retval 0            : success
58  * \retval -ve          : other appropriate errors
59  */
60 int qmt_lqe_set_default(const struct lu_env *env, struct qmt_pool_info *pool,
61                         struct lquota_entry *lqe, bool create_record)
62 {
63         struct lquota_entry     *lqe_def;
64         int                     rc = 0;
65
66         ENTRY;
67
68         if (lqe->lqe_id.qid_uid == 0)
69                 RETURN(0);
70
71         lqe_def = pool->qpi_grace_lqe[lqe->lqe_site->lqs_qtype];
72
73         LQUOTA_DEBUG(lqe, "inherit default quota");
74
75         lqe->lqe_is_default = true;
76         lqe->lqe_hardlimit = lqe_def->lqe_hardlimit;
77         lqe->lqe_softlimit = lqe_def->lqe_softlimit;
78
79         if (create_record) {
80                 lqe->lqe_uptodate = true;
81                 rc = qmt_set_with_lqe(env, pool->qpi_qmt, lqe, 0, 0,
82                                       LQUOTA_GRACE_FLAG(0, LQUOTA_FLAG_DEFAULT),
83                                       QIF_TIMES, true, false);
84
85                 if (rc != 0)
86                         LQUOTA_ERROR(lqe, "failed to create the global quota"
87                                      " record: %d", rc);
88         }
89
90         if (lqe->lqe_hardlimit == 0 && lqe->lqe_softlimit == 0)
91                 lqe->lqe_enforced = false;
92         else
93                 lqe->lqe_enforced = true;
94
95         RETURN(rc);
96 }
97
98 /*
99  * Update a lquota entry. This is done by reading quota settings from the global
100  * index. The lquota entry must be write locked.
101  *
102  * \param env - the environment passed by the caller
103  * \param lqe - is the quota entry to refresh
104  * \param arg - is the pointer to the qmt_pool_info structure
105  */
106 static int qmt_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
107                         void *arg)
108 {
109         struct qmt_thread_info  *qti = qmt_info(env);
110         struct qmt_pool_info    *pool = (struct qmt_pool_info *)arg;
111         int                      rc;
112         ENTRY;
113
114         LASSERT(lqe_is_master(lqe));
115
116         /* read record from disk */
117         rc = lquota_disk_read(env, pool->qpi_glb_obj[lqe->lqe_site->lqs_qtype],
118                               &lqe->lqe_id, (struct dt_rec *)&qti->qti_glb_rec);
119
120         switch (rc) {
121         case -ENOENT:
122                 qmt_lqe_set_default(env, pool, lqe, true);
123                 break;
124         case 0:
125                 /* copy quota settings from on-disk record */
126                 lqe->lqe_granted   = qti->qti_glb_rec.qbr_granted;
127                 lqe->lqe_hardlimit = qti->qti_glb_rec.qbr_hardlimit;
128                 lqe->lqe_softlimit = qti->qti_glb_rec.qbr_softlimit;
129                 lqe->lqe_gracetime = LQUOTA_GRACE(qti->qti_glb_rec.qbr_time);
130
131                 if (lqe->lqe_hardlimit == 0 && lqe->lqe_softlimit == 0 &&
132                     (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) &
133                      LQUOTA_FLAG_DEFAULT))
134                         qmt_lqe_set_default(env, pool, lqe, false);
135                 break;
136         default:
137                 LQUOTA_ERROR(lqe, "failed to read quota entry from disk, rc:%d",
138                              rc);
139                 RETURN(rc);
140         }
141
142         if (lqe->lqe_id.qid_uid == 0 ||
143             (lqe->lqe_hardlimit == 0 && lqe->lqe_softlimit == 0))
144                 /* {hard,soft}limit=0 means no quota enforced */
145                 lqe->lqe_enforced = false;
146         else
147                 lqe->lqe_enforced  = true;
148
149         LQUOTA_DEBUG(lqe, "read");
150         RETURN(0);
151 }
152
153 /*
154  * Print lqe information for debugging.
155  *
156  * \param lqe - is the quota entry to debug
157  * \param arg - is the pointer to the qmt_pool_info structure
158  * \param msgdata - debug message
159  * \param fmt     - format of debug message
160  */
161 static void qmt_lqe_debug(struct lquota_entry *lqe, void *arg,
162                           struct libcfs_debug_msg_data *msgdata,
163                           struct va_format *vaf)
164 {
165         struct qmt_pool_info    *pool = (struct qmt_pool_info *)arg;
166
167         libcfs_debug_msg(msgdata,
168                          "%pV qmt:%s pool:%d-%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",
169                          vaf, pool->qpi_qmt->qmt_svname,
170                          pool->qpi_key & 0x0000ffff,
171                          RES_NAME(pool->qpi_key >> 16),
172                          lqe->lqe_id.qid_uid, lqe->lqe_enforced,
173                          lqe->lqe_hardlimit, lqe->lqe_softlimit,
174                          lqe->lqe_granted, lqe->lqe_gracetime,
175                          lqe->lqe_qunit, lqe->lqe_edquot, lqe->lqe_may_rel,
176                          lqe->lqe_revoke_time,
177                          lqe->lqe_is_default ? "yes" : "no");
178 }
179
180 /*
181  * Vector of quota entry operations supported on the master
182  */
183 struct lquota_entry_operations qmt_lqe_ops = {
184         .lqe_init       = qmt_lqe_init,
185         .lqe_read       = qmt_lqe_read,
186         .lqe_debug      = qmt_lqe_debug,
187 };
188
189 /*
190  * Reserve enough credits to update records in both the global index and
191  * the slave index identified by \slv_obj
192  *
193  * \param env     - is the environment passed by the caller
194  * \param lqe     - is the quota entry associated with the identifier
195  *                  subject to the change
196  * \param slv_obj - is the dt_object associated with the index file
197  * \param restore - is a temporary storage for current quota settings which will
198  *                  be restored if something goes wrong at index update time.
199  */
200 struct thandle *qmt_trans_start_with_slv(const struct lu_env *env,
201                                          struct lquota_entry *lqe,
202                                          struct dt_object *slv_obj,
203                                          struct qmt_lqe_restore *restore)
204 {
205         struct qmt_device       *qmt;
206         struct thandle          *th;
207         int                      rc;
208         ENTRY;
209
210         LASSERT(lqe != NULL);
211         LASSERT(lqe_is_master(lqe));
212
213         qmt = lqe2qpi(lqe)->qpi_qmt;
214
215         if (slv_obj != NULL)
216                 LQUOTA_DEBUG(lqe, "declare write for slv "DFID,
217                              PFID(lu_object_fid(&slv_obj->do_lu)));
218
219         /* start transaction */
220         th = dt_trans_create(env, qmt->qmt_child);
221         if (IS_ERR(th))
222                 RETURN(th);
223
224         if (slv_obj == NULL)
225                 /* quota settings on master are updated synchronously for the
226                  * time being */
227                 th->th_sync = 1;
228
229         /* reserve credits for global index update */
230         rc = lquota_disk_declare_write(env, th, LQE_GLB_OBJ(lqe), &lqe->lqe_id);
231         if (rc)
232                 GOTO(out, rc);
233
234         if (slv_obj != NULL) {
235                 /* reserve credits for slave index update */
236                 rc = lquota_disk_declare_write(env, th, slv_obj, &lqe->lqe_id);
237                 if (rc)
238                         GOTO(out, rc);
239         }
240
241         /* start transaction */
242         rc = dt_trans_start_local(env, qmt->qmt_child, th);
243         if (rc)
244                 GOTO(out, rc);
245
246         EXIT;
247 out:
248         if (rc) {
249                 dt_trans_stop(env, qmt->qmt_child, th);
250                 th = ERR_PTR(rc);
251                 LQUOTA_ERROR(lqe, "failed to slv declare write for "DFID
252                              ", rc:%d", PFID(lu_object_fid(&slv_obj->do_lu)),
253                              rc);
254         } else {
255                 restore->qlr_hardlimit = lqe->lqe_hardlimit;
256                 restore->qlr_softlimit = lqe->lqe_softlimit;
257                 restore->qlr_gracetime = lqe->lqe_gracetime;
258                 restore->qlr_granted   = lqe->lqe_granted;
259                 restore->qlr_qunit     = lqe->lqe_qunit;
260         }
261         return th;
262 }
263
264 /*
265  * Reserve enough credits to update a record in the global index
266  *
267  * \param env     - is the environment passed by the caller
268  * \param lqe     - is the quota entry to be modified in the global index
269  * \param restore - is a temporary storage for current quota settings which will
270  *                  be restored if something goes wrong at index update time.
271  */
272 struct thandle *qmt_trans_start(const struct lu_env *env,
273                                 struct lquota_entry *lqe,
274                                 struct qmt_lqe_restore *restore)
275 {
276         LQUOTA_DEBUG(lqe, "declare write");
277         return qmt_trans_start_with_slv(env, lqe, NULL, restore);
278 }
279
280 /*
281  * Update record associated with a quota entry in the global index.
282  * If LQUOTA_BUMP_VER is set, then the global index version must also be
283  * bumped.
284  * The entry must be at least read locked, dirty and up-to-date.
285  *
286  * \param env   - the environment passed by the caller
287  * \param th    - is the transaction handle to be used for the disk writes
288  * \param lqe   - is the quota entry to udpate
289  * \param obj   - is the dt_object associated with the index file
290  * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
291  * \param ver   - is used to return the new version of the index.
292  *
293  * \retval      - 0 on success and lqe dirty flag cleared,
294  *                appropriate error on failure and uptodate flag cleared.
295  */
296 int qmt_glb_write(const struct lu_env *env, struct thandle *th,
297                   struct lquota_entry *lqe, __u32 flags, __u64 *ver)
298 {
299         struct qmt_thread_info  *qti = qmt_info(env);
300         struct lquota_glb_rec   *rec;
301         int                      rc;
302         ENTRY;
303
304         LASSERT(lqe != NULL);
305         LASSERT(lqe_is_master(lqe));
306         LASSERT(lqe_is_locked(lqe));
307         LASSERT(lqe->lqe_uptodate);
308         LASSERT((flags & ~(LQUOTA_BUMP_VER | LQUOTA_SET_VER)) == 0);
309
310         LQUOTA_DEBUG(lqe, "write glb");
311
312         /* never delete the entry even when the id isn't enforced and
313          * no any guota granted, otherwise, this entry will not be
314          * synced to slave during the reintegration. */
315         rec = &qti->qti_glb_rec;
316
317         /* fill global index with updated quota settings */
318         rec->qbr_granted   = lqe->lqe_granted;
319         if (lqe->lqe_is_default) {
320                 rec->qbr_hardlimit = 0;
321                 rec->qbr_softlimit = 0;
322                 rec->qbr_time      = LQUOTA_GRACE_FLAG(0, LQUOTA_FLAG_DEFAULT);
323         } else {
324                 rec->qbr_hardlimit = lqe->lqe_hardlimit;
325                 rec->qbr_softlimit = lqe->lqe_softlimit;
326                 rec->qbr_time      = lqe->lqe_gracetime;
327         }
328
329         /* write new quota settings */
330         rc = lquota_disk_write(env, th, LQE_GLB_OBJ(lqe), &lqe->lqe_id,
331                                (struct dt_rec *)rec, flags, ver);
332         if (rc)
333                 /* we failed to write the new quota settings to disk, report
334                  * error to caller who will restore the initial value */
335                 LQUOTA_ERROR(lqe, "failed to update global index, rc:%d", rc);
336
337         RETURN(rc);
338 }
339
340 /*
341  * Read from disk how much quota space is allocated to a slave.
342  * This is done by reading records from the dedicated slave index file.
343  * Return in \granted how much quota space is currently allocated to the
344  * slave.
345  * The entry must be at least read locked.
346  *
347  * \param env - the environment passed by the caller
348  * \param lqe - is the quota entry associated with the identifier to look-up
349  *              in the slave index
350  * \param slv_obj - is the dt_object associated with the slave index
351  * \param granted - is the output parameter where to return how much space
352  *                  is granted to the slave.
353  *
354  * \retval    - 0 on success, appropriate error on failure
355  */
356 int qmt_slv_read(const struct lu_env *env, struct lquota_entry *lqe,
357                  struct dt_object *slv_obj, __u64 *granted)
358 {
359         struct qmt_thread_info  *qti = qmt_info(env);
360         struct lquota_slv_rec   *slv_rec = &qti->qti_slv_rec;
361         int                      rc;
362         ENTRY;
363
364         LASSERT(lqe != NULL);
365         LASSERT(lqe_is_master(lqe));
366         LASSERT(lqe_is_locked(lqe));
367
368         LQUOTA_DEBUG(lqe, "read slv "DFID,
369                      PFID(lu_object_fid(&slv_obj->do_lu)));
370
371         /* read slave record from disk */
372         rc = lquota_disk_read(env, slv_obj, &lqe->lqe_id,
373                               (struct dt_rec *)slv_rec);
374         switch (rc) {
375         case -ENOENT:
376                 *granted = 0;
377                 break;
378         case 0:
379                 /* extract granted from on-disk record */
380                 *granted = slv_rec->qsr_granted;
381                 break;
382         default:
383                 LQUOTA_ERROR(lqe, "failed to read slave record "DFID,
384                              PFID(lu_object_fid(&slv_obj->do_lu)));
385                 RETURN(rc);
386         }
387
388         LQUOTA_DEBUG(lqe, "successful slv read %llu", *granted);
389
390         RETURN(0);
391 }
392
393 /*
394  * Update record in slave index file.
395  * The entry must be at least read locked.
396  *
397  * \param env - the environment passed by the caller
398  * \param th  - is the transaction handle to be used for the disk writes
399  * \param lqe - is the dirty quota entry which will be updated at the same time
400  *              as the slave index
401  * \param slv_obj - is the dt_object associated with the slave index
402  * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
403  * \param ver   - is used to return the new version of the index.
404  * \param granted - is the new amount of quota space owned by the slave
405  *
406  * \retval    - 0 on success, appropriate error on failure
407  */
408 int qmt_slv_write(const struct lu_env *env, struct thandle *th,
409                   struct lquota_entry *lqe, struct dt_object *slv_obj,
410                   __u32 flags, __u64 *ver, __u64 granted)
411 {
412         struct qmt_thread_info  *qti = qmt_info(env);
413         struct lquota_slv_rec   *rec;
414         int                      rc;
415         ENTRY;
416
417         LASSERT(lqe != NULL);
418         LASSERT(lqe_is_master(lqe));
419         LASSERT(lqe_is_locked(lqe));
420
421         LQUOTA_DEBUG(lqe, "write slv "DFID" granted:%llu",
422                      PFID(lu_object_fid(&slv_obj->do_lu)), granted);
423
424         /* never delete the entry, otherwise, it'll not be transferred
425          * to slave during reintegration. */
426         rec = &qti->qti_slv_rec;
427
428         /* updated space granted to this slave */
429         rec->qsr_granted = granted;
430
431         /* write new granted space */
432         rc = lquota_disk_write(env, th, slv_obj, &lqe->lqe_id,
433                                (struct dt_rec *)rec, flags, ver);
434         if (rc) {
435                 LQUOTA_ERROR(lqe, "failed to update slave index "DFID" granted:"
436                              "%llu", PFID(lu_object_fid(&slv_obj->do_lu)),
437                              granted);
438                 RETURN(rc);
439         }
440
441         RETURN(0);
442 }
443
444 /*
445  * Check whether new limits are valid for this pool
446  *
447  * \param lqe  - is the quota entry subject to the setquota
448  * \param hard - is the new hard limit
449  * \param soft - is the new soft limit
450  */
451 int qmt_validate_limits(struct lquota_entry *lqe, __u64 hard, __u64 soft)
452 {
453         ENTRY;
454
455         if (hard != 0 && soft > hard)
456                 /* soft limit must be less than hard limit */
457                 RETURN(-EINVAL);
458         RETURN(0);
459 }
460
461 /*
462  * Set/clear edquot flag after quota space allocation/release or settings
463  * change. Slaves will be notified of changes via glimpse on per-ID lock
464  *
465  * \param lqe - is the quota entry to check
466  * \param now - is the current time in second used for grace time managment
467  */
468 void qmt_adjust_edquot(struct lquota_entry *lqe, __u64 now)
469 {
470         struct qmt_pool_info    *pool = lqe2qpi(lqe);
471         ENTRY;
472
473         if (!lqe->lqe_enforced || lqe->lqe_id.qid_uid == 0)
474                 RETURN_EXIT;
475
476         if (!lqe->lqe_edquot) {
477                 /* space exhausted flag not set, let's check whether it is time
478                  * to set the flag */
479
480                 if (!qmt_space_exhausted(lqe, now))
481                         /* the qmt still has available space */
482                         RETURN_EXIT;
483
484                 /* See comment in qmt_adjust_qunit(). LU-4139 */
485                 if (qmt_hard_exhausted(lqe) ||
486                     pool->qpi_key >> 16 != LQUOTA_RES_DT) {
487                         time64_t lapse;
488
489                         /* we haven't reached the minimal qunit yet so there is
490                          * still hope that the rebalancing process might free
491                          * up some quota space */
492                         if (lqe->lqe_qunit != pool->qpi_least_qunit)
493                                 RETURN_EXIT;
494
495                         /* least qunit value not sent to all slaves yet */
496                         if (lqe->lqe_revoke_time == 0)
497                                 RETURN_EXIT;
498
499                         /* Let's give more time to slave to release space */
500                         lapse = ktime_get_seconds() - QMT_REBA_TIMEOUT;
501                         if (lqe->lqe_may_rel != 0 && lqe->lqe_revoke_time > lapse)
502                                 RETURN_EXIT;
503                 } else {
504                         if (lqe->lqe_qunit > pool->qpi_soft_least_qunit)
505                                 RETURN_EXIT;
506                 }
507
508                 /* set edquot flag */
509                 lqe->lqe_edquot = true;
510         } else {
511                 /* space exhausted flag set, let's check whether it is time to
512                  * clear it */
513
514                 if (qmt_space_exhausted(lqe, now))
515                         /* the qmt still has not space */
516                         RETURN_EXIT;
517
518                 if (lqe->lqe_hardlimit != 0 &&
519                     lqe->lqe_granted + pool->qpi_least_qunit >
520                                                         lqe->lqe_hardlimit)
521                         /* we clear the flag only once at least one least qunit
522                          * is available */
523                         RETURN_EXIT;
524
525                 /* clear edquot flag */
526                 lqe->lqe_edquot = false;
527         }
528
529         LQUOTA_DEBUG(lqe, "changing edquot flag");
530
531         /* let's notify slave by issuing glimpse on per-ID lock.
532          * the rebalance thread will take care of this */
533         qmt_id_lock_notify(pool->qpi_qmt, lqe);
534         EXIT;
535 }
536
537 /* Using least_qunit when over block softlimit will seriously impact the
538  * write performance, we need to do some special tweaking on that. */
539 static __u64 qmt_calc_softlimit(struct lquota_entry *lqe, bool *oversoft)
540 {
541         struct qmt_pool_info *pool = lqe2qpi(lqe);
542
543         LASSERT(lqe->lqe_softlimit != 0);
544         *oversoft = false;
545         /* No need to do special tweaking for inode limit */
546         if (pool->qpi_key >> 16 != LQUOTA_RES_DT)
547                 return lqe->lqe_softlimit;
548
549         if (lqe->lqe_granted <= lqe->lqe_softlimit +
550                                 pool->qpi_soft_least_qunit) {
551                 return lqe->lqe_softlimit;
552         } else if (lqe->lqe_hardlimit != 0) {
553                 *oversoft = true;
554                 return lqe->lqe_hardlimit;
555         } else {
556                 *oversoft = true;
557                 return 0;
558         }
559 }
560
561 /*
562  * Try to grant more quota space back to slave.
563  *
564  * \param lqe     - is the quota entry for which we would like to allocate more
565  *                  space
566  * \param granted - is how much was already granted as part of the request
567  *                  processing
568  * \param spare   - is how much unused quota space the slave already owns
569  *
570  * \retval return how additional space can be granted to the slave
571  */
572 __u64 qmt_alloc_expand(struct lquota_entry *lqe, __u64 granted, __u64 spare)
573 {
574         struct qmt_pool_info    *pool = lqe2qpi(lqe);
575         __u64                    remaining, qunit;
576         int                      slv_cnt;
577
578         LASSERT(lqe->lqe_enforced && lqe->lqe_qunit != 0);
579
580         slv_cnt = lqe2qpi(lqe)->qpi_slv_nr[lqe->lqe_site->lqs_qtype];
581         qunit   = lqe->lqe_qunit;
582
583         /* See comment in qmt_adjust_qunit(). LU-4139. */
584         if (lqe->lqe_softlimit != 0) {
585                 bool oversoft;
586                 remaining = qmt_calc_softlimit(lqe, &oversoft);
587                 if (remaining == 0)
588                         remaining = lqe->lqe_granted +
589                                     pool->qpi_soft_least_qunit;
590         } else {
591                 remaining = lqe->lqe_hardlimit;
592         }
593
594         if (lqe->lqe_granted >= remaining)
595                 RETURN(0);
596
597         remaining -= lqe->lqe_granted;
598
599         do {
600                 if (spare >= qunit)
601                         break;
602
603                 granted &= (qunit - 1);
604
605                 if (remaining > (slv_cnt * qunit) >> 1) {
606                         /* enough room to grant more space w/o additional
607                          * shrinking ... at least for now */
608                         remaining -= (slv_cnt * qunit) >> 1;
609                 } else if (qunit != pool->qpi_least_qunit) {
610                         qunit >>= 2;
611                         continue;
612                 }
613
614                 granted &= (qunit - 1);
615                 if (spare > 0)
616                         RETURN(min_t(__u64, qunit - spare, remaining));
617                 else
618                         RETURN(min_t(__u64, qunit - granted, remaining));
619         } while (qunit >= pool->qpi_least_qunit);
620
621         RETURN(0);
622 }
623
624 /*
625  * Adjust qunit size according to quota limits and total granted count.
626  * The caller must have locked the lqe.
627  *
628  * \param env - the environment passed by the caller
629  * \param lqe - is the qid entry to be adjusted
630  */
631 void qmt_adjust_qunit(const struct lu_env *env, struct lquota_entry *lqe)
632 {
633         struct qmt_pool_info    *pool = lqe2qpi(lqe);
634         int                      slv_cnt;
635         __u64                    qunit, limit, qunit2 = 0;
636         ENTRY;
637
638         LASSERT(lqe_is_locked(lqe));
639
640         if (!lqe->lqe_enforced || lqe->lqe_id.qid_uid == 0)
641                 /* no quota limits */
642                 RETURN_EXIT;
643
644         /* record how many slaves have already registered */
645         slv_cnt = pool->qpi_slv_nr[lqe->lqe_site->lqs_qtype];
646         if (slv_cnt == 0)
647                 /* wait for at least one slave to join */
648                 RETURN_EXIT;
649
650         /* Qunit calculation is based on soft limit, if any, hard limit
651          * otherwise. This means that qunit is shrunk to the minimum when
652          * beyond the soft limit. This will impact performance, but that's the
653          * price of an accurate grace time management. */
654         if (lqe->lqe_softlimit != 0) {
655                 bool oversoft;
656                 /* As a compromise of write performance and the grace time
657                  * accuracy, the block qunit size will be shrunk to
658                  * qpi_soft_least_qunit when over softlimit. LU-4139. */
659                 limit = qmt_calc_softlimit(lqe, &oversoft);
660                 if (oversoft)
661                         qunit2 = pool->qpi_soft_least_qunit;
662                 if (limit == 0)
663                         GOTO(done, qunit = qunit2);
664         } else if (lqe->lqe_hardlimit != 0) {
665                 limit = lqe->lqe_hardlimit;
666         } else {
667                 LQUOTA_ERROR(lqe, "enforced bit set, but neither hard nor soft "
668                              "limit are set");
669                 RETURN_EXIT;
670         }
671
672         qunit = lqe->lqe_qunit == 0 ? pool->qpi_least_qunit : lqe->lqe_qunit;
673
674         /* The qunit value is computed as follows: limit / (2 * slv_cnt).
675          * Then 75% of the quota space can be granted with current qunit value.
676          * The remaining 25% are then used with reduced qunit size (by a factor
677          * of 4) which is then divided in a similar manner.
678          *
679          * |---------------------limit---------------------|
680          * |-------limit / 2-------|-limit / 4-|-limit / 4-|
681          * |qunit|qunit|qunit|qunit|           |           |
682          * |----slv_cnt * qunit----|           |           |
683          * |-grow limit-|          |           |           |
684          * |--------------shrink limit---------|           |
685          * |---space granted in qunit chunks---|-remaining-|
686          *                                    /             \
687          *                                   /               \
688          *                                  /                 \
689          *                                 /                   \
690          *                                /                     \
691          *     qunit >>= 2;            |qunit*slv_cnt|qunit*slv_cnt|
692          *                             |---space in qunit---|remain|
693          *                                  ...                               */
694         if (qunit == pool->qpi_least_qunit ||
695             limit >= lqe->lqe_granted + ((slv_cnt * qunit) >> 1)) {
696                 /* current qunit value still fits, let's see if we can afford to
697                  * increase qunit now ...
698                  * To increase qunit again, we have to be under 25% */
699                 while (qunit && limit >= lqe->lqe_granted + 6 * qunit * slv_cnt)
700                         qunit <<= 2;
701
702                 if (!qunit) {
703                         qunit = limit;
704                         do_div(qunit, 2 * slv_cnt);
705                 }
706
707         } else {
708                 /* shrink qunit until we find a suitable value */
709                 while (qunit > pool->qpi_least_qunit &&
710                        limit < lqe->lqe_granted + ((slv_cnt * qunit) >> 1))
711                         qunit >>= 2;
712         }
713
714         if (qunit2 && qunit > qunit2)
715                 qunit = qunit2;
716 done:
717         if (lqe->lqe_qunit == qunit)
718                 /* keep current qunit */
719                 RETURN_EXIT;
720
721         LQUOTA_DEBUG(lqe, "%s qunit to %llu",
722                      lqe->lqe_qunit < qunit ? "increasing" : "decreasing",
723                      qunit);
724
725         /* store new qunit value */
726         swap(lqe->lqe_qunit, qunit);
727
728         /* reset revoke time */
729         lqe->lqe_revoke_time = 0;
730
731         if (lqe->lqe_qunit < qunit)
732                 /* let's notify slave of qunit shrinking */
733                 qmt_id_lock_notify(pool->qpi_qmt, lqe);
734         else if (lqe->lqe_qunit == pool->qpi_least_qunit)
735                 /* initial qunit value is the smallest one */
736                 lqe->lqe_revoke_time = ktime_get_seconds();
737         EXIT;
738 }
739
740 /*
741  * Adjust qunit & edquot flag in case it wasn't initialized already (e.g.
742  * limit set while no slaves were connected yet)
743  */
744 void qmt_revalidate(const struct lu_env *env, struct lquota_entry *lqe)
745 {
746         if (lqe->lqe_qunit == 0) {
747                 /* lqe was read from disk, but neither qunit, nor edquot flag
748                  * were initialized */
749                 qmt_adjust_qunit(env, lqe);
750                 if (lqe->lqe_qunit != 0)
751                         qmt_adjust_edquot(lqe, ktime_get_real_seconds());
752         }
753 }