Whamcloud - gitweb
a4086f6b52bcf1aff8b04ee70ebde50071ab688d
[fs/lustre-release.git] / lustre / quota / qmt_lock.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, 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 #ifndef EXPORT_SYMTAB
32 # define EXPORT_SYMTAB
33 #endif
34
35 #define DEBUG_SUBSYSTEM S_LQUOTA
36
37 #include <lustre_dlm.h>
38 #include <obd_class.h>
39
40 #include "qmt_internal.h"
41
42 /* intent policy function called from mdt_intent_opc() when the intent is of
43  * quota type */
44 int qmt_intent_policy(const struct lu_env *env, struct lu_device *ld,
45                       struct ptlrpc_request *req, struct ldlm_lock **lockp,
46                       int flags)
47 {
48         struct qmt_device       *qmt = lu2qmt_dev(ld);
49         struct ldlm_intent      *it;
50         struct quota_body       *reqbody;
51         struct quota_body       *repbody;
52         struct obd_uuid         *uuid;
53         struct lquota_lvb       *lvb;
54         struct ldlm_resource    *res = (*lockp)->l_resource;
55         int                      rc;
56         ENTRY;
57
58         req_capsule_extend(&req->rq_pill, &RQF_LDLM_INTENT_QUOTA);
59         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
60                              ldlm_lvbo_size(*lockp));
61
62         /* extract quota body and intent opc */
63         it = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
64         if (it == NULL)
65                 RETURN(err_serious(-EFAULT));
66
67         reqbody = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_BODY);
68         if (reqbody == NULL)
69                 RETURN(err_serious(-EFAULT));
70
71         /* prepare reply */
72         rc = req_capsule_server_pack(&req->rq_pill);
73         if (rc != 0) {
74                 CERROR("Can't pack response, rc %d\n", rc);
75                 RETURN(err_serious(rc));
76         }
77
78         repbody = req_capsule_server_get(&req->rq_pill, &RMF_QUOTA_BODY);
79         if (repbody == NULL)
80                 RETURN(err_serious(-EFAULT));
81
82         uuid = &(*lockp)->l_export->exp_client_uuid;
83         switch (it->opc) {
84
85         case IT_QUOTA_DQACQ: {
86                 struct lquota_entry     *lqe;
87                 struct ldlm_lock        *lock;
88
89                 if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] == 0)
90                         /* acquire on global lock? something is wrong ... */
91                         GOTO(out, rc = -EPROTO);
92
93                 /* verify global lock isn't stale */
94                 if (!lustre_handle_is_used(&reqbody->qb_glb_lockh))
95                         GOTO(out, rc = -ENOLCK);
96
97                 lock = ldlm_handle2lock(&reqbody->qb_glb_lockh);
98                 if (lock == NULL)
99                         GOTO(out, rc = -ENOLCK);
100                 LDLM_LOCK_PUT(lock);
101
102                 lqe = res->lr_lvb_data;
103                 LASSERT(lqe != NULL);
104                 lqe_getref(lqe);
105
106                 /* acquire quota space */
107                 rc = qmt_dqacq0(env, lqe, qmt, uuid, reqbody->qb_flags,
108                                 reqbody->qb_count, reqbody->qb_usage,
109                                 repbody);
110                 lqe_putref(lqe);
111                 if (rc)
112                         GOTO(out, rc);
113                 break;
114         }
115
116         case IT_QUOTA_CONN:
117                 /* new connection from slave */
118
119                 if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] != 0)
120                         /* connection on per-ID lock? something is wrong ... */
121                         GOTO(out, rc = -EPROTO);
122
123                 rc = qmt_pool_new_conn(env, qmt, &reqbody->qb_fid,
124                                        &repbody->qb_slv_fid,
125                                        &repbody->qb_slv_ver, uuid);
126                 if (rc)
127                         GOTO(out, rc);
128                 break;
129
130         default:
131                 CERROR("%s: invalid intent opcode: "LPU64"\n", qmt->qmt_svname,
132                        it->opc);
133                 GOTO(out, rc = err_serious(-EINVAL));
134         }
135
136         /* on success, pack lvb in reply */
137         lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB);
138         ldlm_lvbo_fill(*lockp, lvb, ldlm_lvbo_size(*lockp));
139         EXIT;
140 out:
141         return rc;
142 }
143
144 /*
145  * Initialize quota LVB associated with quota indexes.
146  * Called with res->lr_lvb_sem held
147  */
148 int qmt_lvbo_init(struct lu_device *ld, struct ldlm_resource *res)
149 {
150         struct lu_env           *env;
151         struct qmt_thread_info  *qti;
152         struct qmt_device       *qmt = lu2qmt_dev(ld);
153         int                      pool_id, pool_type, qtype;
154         int                      rc;
155         ENTRY;
156
157         LASSERT(res != NULL);
158
159         if (res->lr_type != LDLM_PLAIN)
160                 RETURN(-ENOTSUPP);
161
162         if (res->lr_lvb_data ||
163             res->lr_name.name[LUSTRE_RES_ID_SEQ_OFF] != FID_SEQ_QUOTA_GLB)
164                 RETURN(0);
165
166         OBD_ALLOC_PTR(env);
167         if (env == NULL)
168                 RETURN(-ENOMEM);
169
170         /* initialize environment */
171         rc = lu_env_init(env, LCT_MD_THREAD);
172         if (rc) {
173                 OBD_FREE_PTR(env);
174                 RETURN(rc);
175         }
176         qti = qmt_info(env);
177
178         /* extract global index FID and quota identifier */
179         fid_extract_quota_resid(&res->lr_name, &qti->qti_fid, &qti->qti_id);
180
181         /* sanity check the global index FID */
182         rc = lquota_extract_fid(&qti->qti_fid, &pool_id, &pool_type, &qtype);
183         if (rc) {
184                 CERROR("can't extract pool information from FID "DFID"\n",
185                        PFID(&qti->qti_fid));
186                 GOTO(out, rc);
187         }
188
189         if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] != 0) {
190                 /* no ID quota lock associated with UID/GID 0 or with a seq 0,
191                  * we are thus dealing with an ID lock. */
192                 struct lquota_entry     *lqe;
193
194                 /* Find the quota entry associated with the quota id */
195                 lqe = qmt_pool_lqe_lookup(env, qmt, pool_id, pool_type, qtype,
196                                           &qti->qti_id);
197                 if (IS_ERR(lqe))
198                         GOTO(out, rc = PTR_ERR(lqe));
199
200                 /* store reference to lqe in lr_lvb_data */
201                 res->lr_lvb_data = lqe;
202                 LQUOTA_DEBUG(lqe, "initialized res lvb");
203         } else {
204                 struct dt_object        *obj;
205
206                 /* lookup global index */
207                 obj = dt_locate(env, qmt->qmt_child, &qti->qti_fid);
208                 if (IS_ERR(obj))
209                         GOTO(out, rc = PTR_ERR(obj));
210                 if (!dt_object_exists(obj)) {
211                         lu_object_put(env, &obj->do_lu);
212                         GOTO(out, rc = -ENOENT);
213                 }
214
215                 /* store reference to global index object in lr_lvb_data */
216                 res->lr_lvb_data = obj;
217                 CDEBUG(D_QUOTA, DFID" initialized lvb\n", PFID(&qti->qti_fid));
218         }
219
220         res->lr_lvb_len  = sizeof(struct lquota_lvb);
221         EXIT;
222 out:
223         lu_env_fini(env);
224         OBD_FREE_PTR(env);
225         return rc;
226 }
227
228 /*
229  * Update LVB associated with the global quota index.
230  * This function is called from the DLM itself after a glimpse callback, in this
231  * case valid ptlrpc request is passed.
232  */
233 int qmt_lvbo_update(struct lu_device *ld, struct ldlm_resource *res,
234                     struct ptlrpc_request *req, int increase_only)
235 {
236         struct lu_env           *env;
237         struct qmt_thread_info  *qti;
238         struct qmt_device       *qmt = lu2qmt_dev(ld);
239         struct lquota_entry     *lqe;
240         struct lquota_lvb       *lvb;
241         struct ldlm_lock        *lock;
242         struct obd_export       *exp;
243         int                      rc = 0;
244         ENTRY;
245
246         LASSERT(res != NULL);
247
248         if (req == NULL)
249                 RETURN(0);
250
251         if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] == 0)
252                 /* no need to update lvb for global quota locks */
253                 RETURN(0);
254
255         lvb = req_capsule_server_swab_get(&req->rq_pill, &RMF_DLM_LVB,
256                                           lustre_swab_lquota_lvb);
257         if (lvb == NULL) {
258                 CERROR("%s: failed to extract lvb from request\n",
259                        qmt->qmt_svname);
260                 RETURN(-EFAULT);
261         }
262
263         lqe = res->lr_lvb_data;
264         LASSERT(lqe != NULL);
265         lqe_getref(lqe);
266
267         LQUOTA_DEBUG(lqe, "releasing:"LPU64" may release:"LPU64,
268                      lvb->lvb_id_rel, lvb->lvb_id_may_rel);
269
270         if (lvb->lvb_id_rel == 0) {
271                 /* nothing to release */
272                 if (lvb->lvb_id_may_rel != 0)
273                         /* but might still release later ... */
274                         lqe->lqe_may_rel += lvb->lvb_id_may_rel;
275                 GOTO(out_lqe, rc = 0);
276         }
277
278         /* allocate environement */
279         OBD_ALLOC_PTR(env);
280         if (env == NULL)
281                 GOTO(out_lqe, rc = -ENOMEM);
282
283         /* initialize environment */
284         rc = lu_env_init(env, LCT_MD_THREAD);
285         if (rc)
286                 GOTO(out_env, rc);
287         qti = qmt_info(env);
288
289         /* The request is a glimpse callback which was sent via the
290          * reverse import to the slave. What we care about here is the
291          * export associated with the slave and req->rq_export is
292          * definitely not what we are looking for (it is actually set to
293          * NULL here).
294          * Therefore we extract the lock from the request argument
295          * and use lock->l_export. */
296         lock = ldlm_request_lock(req);
297         if (IS_ERR(lock)) {
298                 CERROR("%s: failed to get lock from request!\n",
299                        qmt->qmt_svname);
300                 GOTO(out_env_init, rc = PTR_ERR(lock));
301         }
302
303         exp = class_export_get(lock->l_export);
304         if (exp == NULL) {
305                 CERROR("%s: failed to get export from lock!\n",
306                        qmt->qmt_svname);
307                 GOTO(out_env_init, rc = -EFAULT);
308         }
309
310         /* release quota space */
311         rc = qmt_dqacq0(env, lqe, qmt, &exp->exp_client_uuid,
312                         QUOTA_DQACQ_FL_REL, lvb->lvb_id_rel, 0, &qti->qti_body);
313         if (rc || qti->qti_body.qb_count != lvb->lvb_id_rel)
314                 LQUOTA_ERROR(lqe, "failed to release quota space on glimpse "
315                              LPU64"!="LPU64" rc:%d\n", qti->qti_body.qb_count,
316                              lvb->lvb_id_rel, rc);
317         class_export_put(exp);
318         if (rc)
319                 GOTO(out_env_init, rc);
320         EXIT;
321 out_env_init:
322         lu_env_fini(env);
323 out_env:
324         OBD_FREE_PTR(env);
325 out_lqe:
326         lqe_putref(lqe);
327         return rc;
328 }
329
330 /*
331  * Report size of lvb to ldlm layer in order to allocate lvb buffer
332  * As far as quota locks are concerned, the size is static and is the same
333  * for both global and per-ID locks which shares the same lvb format.
334  */
335 int qmt_lvbo_size(struct lu_device *ld, struct ldlm_lock *lock)
336 {
337         return sizeof(struct lquota_lvb);
338 }
339
340 /*
341  * Fill request buffer with quota lvb
342  */
343 int qmt_lvbo_fill(struct lu_device *ld, struct ldlm_lock *lock, void *lvb,
344                   int lvblen)
345 {
346         struct ldlm_resource    *res = lock->l_resource;
347         struct lquota_lvb       *qlvb = lvb;
348         ENTRY;
349
350         LASSERT(res != NULL);
351
352         if (res->lr_type != LDLM_PLAIN || res->lr_lvb_data == NULL ||
353             res->lr_name.name[LUSTRE_RES_ID_SEQ_OFF] != FID_SEQ_QUOTA_GLB)
354                 RETURN(-EINVAL);
355
356         if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] != 0) {
357                 /* no ID quota lock associated with UID/GID 0 or with a seq 0,
358                  * we are thus dealing with an ID lock. */
359                 struct lquota_entry     *lqe = res->lr_lvb_data;
360
361                 /* return current qunit value & edquot flags in lvb */
362                 lqe_getref(lqe);
363                 qlvb->lvb_id_qunit = lqe->lqe_qunit;
364                 qlvb->lvb_flags = 0;
365                 if (lqe->lqe_edquot)
366                         qlvb->lvb_flags = LQUOTA_FL_EDQUOT;
367                 lqe_putref(lqe);
368         } else {
369                 /* global quota lock */
370                 struct lu_env           *env;
371                 int                      rc;
372                 struct dt_object        *obj = res->lr_lvb_data;
373
374                 OBD_ALLOC_PTR(env);
375                 if (env == NULL)
376                         RETURN(-ENOMEM);
377
378                 /* initialize environment */
379                 rc = lu_env_init(env, LCT_LOCAL);
380                 if (rc) {
381                         OBD_FREE_PTR(env);
382                         RETURN(rc);
383                 }
384
385                 /* return current version of global index */
386                 qlvb->lvb_glb_ver = dt_version_get(env, obj);
387
388                 lu_env_fini(env);
389                 OBD_FREE_PTR(env);
390         }
391
392         RETURN(sizeof(struct lquota_lvb));
393 }
394
395 /*
396  * Free lvb associated with a given ldlm resource
397  * we don't really allocate a lvb, lr_lvb_data just points to
398  * the appropriate backend structures.
399  */
400 int qmt_lvbo_free(struct lu_device *ld, struct ldlm_resource *res)
401 {
402         ENTRY;
403
404         if (res->lr_lvb_data == NULL)
405                 RETURN(0);
406
407         if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] != 0) {
408                 struct lquota_entry     *lqe = res->lr_lvb_data;
409
410                 /* release lqe reference */
411                 lqe_putref(lqe);
412         } else {
413                 struct dt_object        *obj = res->lr_lvb_data;
414                 struct lu_env           *env;
415                 int                      rc;
416
417                 OBD_ALLOC_PTR(env);
418                 if (env == NULL)
419                         RETURN(-ENOMEM);
420
421                 /* initialize environment */
422                 rc = lu_env_init(env, LCT_LOCAL);
423                 if (rc) {
424                         OBD_FREE_PTR(env);
425                         RETURN(rc);
426                 }
427
428                 /* release object reference */
429                 lu_object_put(env, &obj->do_lu);
430                 lu_env_fini(env);
431                 OBD_FREE_PTR(env);
432         }
433
434         res->lr_lvb_data = NULL;
435         res->lr_lvb_len  = 0;
436
437         RETURN(0);
438 }
439
440 typedef int (*qmt_glimpse_cb_t)(const struct lu_env *, struct qmt_device *,
441                                 struct obd_uuid *, union ldlm_gl_desc *,
442                                 void *);
443 /*
444  * Send glimpse callback to slaves holding a lock on resource \res.
445  * This is used to notify slaves of new quota settings or to claim quota space
446  * back.
447  *
448  * \param env  - is the environment passed by the caller
449  * \param qmt  - is the quota master target
450  * \param res  - is the dlm resource associated with the quota object
451  * \param desc - is the glimpse descriptor to pack in glimpse callback
452  * \param cb   - is the callback function called on every lock and determine
453  *               whether a glimpse should be issued
454  * \param arg  - is an opaq parameter passed to the callback function
455  */
456 static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt,
457                             struct ldlm_resource *res, union ldlm_gl_desc *desc,
458                             qmt_glimpse_cb_t cb, void *arg)
459 {
460         cfs_list_t      *tmp, *pos;
461         CFS_LIST_HEAD(gl_list);
462         int              rc = 0;
463         ENTRY;
464
465         lock_res(res);
466         /* scan list of granted locks */
467         cfs_list_for_each(pos, &res->lr_granted) {
468                 struct ldlm_glimpse_work        *work;
469                 struct ldlm_lock                *lock;
470                 struct obd_uuid                 *uuid;
471
472                 lock = cfs_list_entry(pos, struct ldlm_lock, l_res_link);
473                 LASSERT(lock->l_export);
474                 uuid = &lock->l_export->exp_client_uuid;
475
476                 if (cb != NULL) {
477                         rc = cb(env, qmt, uuid, desc, arg);
478                         if (rc == 0)
479                                 /* slave should not be notified */
480                                 continue;
481                         if (rc < 0)
482                                 /* something wrong happened, we still notify */
483                                 CERROR("%s: callback function failed to "
484                                        "determine whether slave %s should be "
485                                        "notified (%d)\n", qmt->qmt_svname,
486                                        obd_uuid2str(uuid), rc);
487                 }
488
489                 OBD_ALLOC_PTR(work);
490                 if (work == NULL) {
491                         CERROR("%s: failed to notify %s\n", qmt->qmt_svname,
492                                obd_uuid2str(uuid));
493                         continue;
494                 }
495
496                 cfs_list_add_tail(&work->gl_list, &gl_list);
497                 work->gl_lock  = LDLM_LOCK_GET(lock);
498                 work->gl_flags = 0;
499                 work->gl_desc  = desc;
500
501         }
502         unlock_res(res);
503
504         if (cfs_list_empty(&gl_list)) {
505                 CDEBUG(D_QUOTA, "%s: nobody to notify\n", qmt->qmt_svname);
506                 RETURN(0);
507         }
508
509         /* issue glimpse callbacks to all connected slaves */
510         rc = ldlm_glimpse_locks(res, &gl_list);
511
512         cfs_list_for_each_safe(pos, tmp, &gl_list) {
513                 struct ldlm_glimpse_work *work;
514
515                 work = cfs_list_entry(pos, struct ldlm_glimpse_work, gl_list);
516
517                 cfs_list_del(&work->gl_list);
518                 CERROR("%s: failed to notify %s of new quota settings\n",
519                        qmt->qmt_svname,
520                        obd_uuid2str(&work->gl_lock->l_export->exp_client_uuid));
521                 LDLM_LOCK_RELEASE(work->gl_lock);
522                 OBD_FREE_PTR(work);
523         }
524
525         RETURN(rc);
526 }
527
528 /*
529  * Send glimpse request to all global quota locks to push new quota setting to
530  * slaves.
531  *
532  * \param env - is the environment passed by the caller
533  * \param lqe - is the lquota entry which has new settings
534  * \param ver - is the version associated with the setting change
535  */
536 void qmt_glb_lock_notify(const struct lu_env *env, struct lquota_entry *lqe,
537                          __u64 ver)
538 {
539         struct qmt_thread_info  *qti = qmt_info(env);
540         struct qmt_pool_info    *pool = lqe2qpi(lqe);
541         struct ldlm_resource    *res = NULL;
542         int                      rc;
543         ENTRY;
544
545         lquota_generate_fid(&qti->qti_fid, pool->qpi_key & 0x0000ffff,
546                             pool->qpi_key >> 16, lqe->lqe_site->lqs_qtype);
547
548         /* send glimpse callback to notify slaves of new quota settings */
549         qti->qti_gl_desc.lquota_desc.gl_id        = lqe->lqe_id;
550         qti->qti_gl_desc.lquota_desc.gl_flags     = 0;
551         qti->qti_gl_desc.lquota_desc.gl_hardlimit = lqe->lqe_hardlimit;
552         qti->qti_gl_desc.lquota_desc.gl_softlimit = lqe->lqe_softlimit;
553         qti->qti_gl_desc.lquota_desc.gl_time      = lqe->lqe_gracetime;
554         qti->qti_gl_desc.lquota_desc.gl_ver       = ver;
555
556         /* look up ldlm resource associated with global index */
557         fid_build_reg_res_name(&qti->qti_fid, &qti->qti_resid);
558         res = ldlm_resource_get(pool->qpi_qmt->qmt_ns, NULL, &qti->qti_resid,
559                                 LDLM_PLAIN, 0);
560         if (res == NULL) {
561                 /* this might happen if no slaves have enqueued global quota
562                  * locks yet */
563                 LQUOTA_DEBUG(lqe, "failed to lookup ldlm resource associated "
564                              "with "DFID, PFID(&qti->qti_fid));
565                 RETURN_EXIT;
566         }
567
568         rc = qmt_glimpse_lock(env, pool->qpi_qmt, res, &qti->qti_gl_desc,
569                               NULL, NULL);
570         ldlm_resource_putref(res);
571         EXIT;
572 }
573
574 /* Callback function used to select locks that should be glimpsed when
575  * broadcasting the new qunit value */
576 static int qmt_id_lock_cb(const struct lu_env *env, struct qmt_device *qmt,
577                           struct obd_uuid *uuid, union ldlm_gl_desc *desc,
578                           void *arg)
579 {
580         struct obd_uuid *slv_uuid = arg;
581
582         if (slv_uuid != NULL && obd_uuid_equals(uuid, slv_uuid))
583                 RETURN(0);
584         RETURN(+1);
585 }
586
587 /*
588  * Send glimpse request on per-ID lock to push new qunit value to slave.
589  *
590  * \param env  - is the environment passed by the caller
591  * \param qmt  - is the quota master target device
592  * \param lqe  - is the lquota entry with the new qunit value
593  * \param uuid - is the uuid of the slave acquiring space, if any
594  */
595 static void qmt_id_lock_glimpse(const struct lu_env *env,
596                                 struct qmt_device *qmt,
597                                 struct lquota_entry *lqe, struct obd_uuid *uuid)
598 {
599         struct qmt_thread_info  *qti = qmt_info(env);
600         struct qmt_pool_info    *pool = lqe2qpi(lqe);
601         struct ldlm_resource    *res = NULL;
602         int                      rc;
603         ENTRY;
604
605         if (!lqe->lqe_enforced)
606                 RETURN_EXIT;
607
608         lquota_generate_fid(&qti->qti_fid, pool->qpi_key & 0x0000ffff,
609                             pool->qpi_key >> 16, lqe->lqe_site->lqs_qtype);
610         fid_build_quota_resid(&qti->qti_fid, &lqe->lqe_id, &qti->qti_resid);
611         res = ldlm_resource_get(qmt->qmt_ns, NULL, &qti->qti_resid, LDLM_PLAIN,
612                                 0);
613         if (res == NULL) {
614                 /* this might legitimately happens if slaves haven't had the
615                  * opportunity to enqueue quota lock yet. */
616                 LQUOTA_DEBUG(lqe, "failed to lookup ldlm resource for per-ID "
617                              "lock "DFID, PFID(&qti->qti_fid));
618                 lqe_write_lock(lqe);
619                 if (lqe->lqe_revoke_time == 0 &&
620                     lqe->lqe_qunit == pool->qpi_least_qunit)
621                         lqe->lqe_revoke_time = cfs_time_current_64();
622                 lqe_write_unlock(lqe);
623                 RETURN_EXIT;
624         }
625
626         lqe_write_lock(lqe);
627         /* The purpose of glimpse callback on per-ID lock is twofold:
628          * - notify slaves of new qunit value and hope they will release some
629          *   spare quota space in return
630          * - notify slaves that master ran out of quota space and there is no
631          *   need to send acquire request any more until further notice */
632
633         /* fill glimpse descriptor with lqe settings */
634         if (lqe->lqe_edquot)
635                 qti->qti_gl_desc.lquota_desc.gl_flags = LQUOTA_FL_EDQUOT;
636         else
637                 qti->qti_gl_desc.lquota_desc.gl_flags = 0;
638         qti->qti_gl_desc.lquota_desc.gl_qunit = lqe->lqe_qunit;
639
640         if (lqe->lqe_revoke_time == 0 &&
641             qti->qti_gl_desc.lquota_desc.gl_qunit == pool->qpi_least_qunit)
642                 /* reset lqe_may_rel, it will be updated on glimpse callback
643                  * replies if needed */
644                 lqe->lqe_may_rel = 0;
645
646         /* The rebalance thread is the only thread which can issue glimpses */
647         LASSERT(!lqe->lqe_gl);
648         lqe->lqe_gl = true;
649         lqe_write_unlock(lqe);
650
651         /* issue glimpse callback to slaves */
652         rc = qmt_glimpse_lock(env, qmt, res, &qti->qti_gl_desc,
653                               uuid ? qmt_id_lock_cb : NULL, (void *)uuid);
654
655         lqe_write_lock(lqe);
656         if (lqe->lqe_revoke_time == 0 &&
657             qti->qti_gl_desc.lquota_desc.gl_qunit == pool->qpi_least_qunit &&
658             lqe->lqe_qunit == pool->qpi_least_qunit) {
659                 lqe->lqe_revoke_time = cfs_time_current_64();
660                 qmt_adjust_edquot(lqe, cfs_time_current_sec());
661         }
662         LASSERT(lqe->lqe_gl);
663         lqe->lqe_gl = false;
664         lqe_write_unlock(lqe);
665
666         ldlm_resource_putref(res);
667         EXIT;
668 }
669
670 /*
671  * Schedule a glimpse request on per-ID locks to push new qunit value or
672  * edquot flag to quota slaves.
673  *
674  * \param qmt  - is the quota master target device
675  * \param lqe  - is the lquota entry with the new qunit value
676  */
677 void qmt_id_lock_notify(struct qmt_device *qmt, struct lquota_entry *lqe)
678 {
679         bool    added = false;
680         ENTRY;
681
682         lqe_getref(lqe);
683         spin_lock(&qmt->qmt_reba_lock);
684         if (!qmt->qmt_stopping && cfs_list_empty(&lqe->lqe_link)) {
685                 cfs_list_add_tail(&lqe->lqe_link, &qmt->qmt_reba_list);
686                 added = true;
687         }
688         spin_unlock(&qmt->qmt_reba_lock);
689
690         if (added)
691                 cfs_waitq_signal(&qmt->qmt_reba_thread.t_ctl_waitq);
692         else
693                 lqe_putref(lqe);
694         EXIT;
695 }
696
697 /*
698  * The rebalance thread is in charge of sending glimpse callbacks on per-ID
699  * quota locks owned by slaves in order to notify them of:
700  * - a qunit shrink in which case slaves might release quota space back in
701  *   glimpse reply.
702  * - set/clear edquot flag used to cache the "quota exhausted" state of the
703  *   master. When the flag is set, slaves know that there is no need to
704  *   try to acquire quota from the master since this latter has already
705  *   distributed all the space.
706  */
707 static int qmt_reba_thread(void *arg)
708 {
709         struct qmt_device       *qmt = (struct qmt_device *)arg;
710         struct ptlrpc_thread    *thread = &qmt->qmt_reba_thread;
711         struct l_wait_info       lwi = { 0 };
712         struct lu_env           *env;
713         struct lquota_entry     *lqe, *tmp;
714         char                     pname[MTI_NAME_MAXLEN];
715         int                      rc;
716         ENTRY;
717
718         OBD_ALLOC_PTR(env);
719         if (env == NULL)
720                 RETURN(-ENOMEM);
721
722         rc = lu_env_init(env, LCT_MD_THREAD);
723         if (rc) {
724                 CERROR("%s: failed to init env.", qmt->qmt_svname);
725                 OBD_FREE_PTR(env);
726                 RETURN(rc);
727         }
728
729         snprintf(pname, MTI_NAME_MAXLEN, "qmt_reba_%s", qmt->qmt_svname);
730         cfs_daemonize(pname);
731
732         thread_set_flags(thread, SVC_RUNNING);
733         cfs_waitq_signal(&thread->t_ctl_waitq);
734
735         while (1) {
736                 l_wait_event(thread->t_ctl_waitq,
737                              !cfs_list_empty(&qmt->qmt_reba_list) ||
738                              !thread_is_running(thread), &lwi);
739
740                 spin_lock(&qmt->qmt_reba_lock);
741                 cfs_list_for_each_entry_safe(lqe, tmp, &qmt->qmt_reba_list,
742                                              lqe_link) {
743                         cfs_list_del_init(&lqe->lqe_link);
744                         spin_unlock(&qmt->qmt_reba_lock);
745
746                         if (thread_is_running(thread))
747                                 qmt_id_lock_glimpse(env, qmt, lqe, NULL);
748
749                         lqe_putref(lqe);
750                         spin_lock(&qmt->qmt_reba_lock);
751                 }
752                 spin_unlock(&qmt->qmt_reba_lock);
753
754                 if (!thread_is_running(thread))
755                         break;
756         }
757         lu_env_fini(env);
758         OBD_FREE_PTR(env);
759         thread_set_flags(thread, SVC_STOPPED);
760         cfs_waitq_signal(&thread->t_ctl_waitq);
761         RETURN(rc);
762 }
763
764 /*
765  * Start rebalance thread. Called when the QMT is being setup
766  */
767 int qmt_start_reba_thread(struct qmt_device *qmt)
768 {
769         struct ptlrpc_thread    *thread = &qmt->qmt_reba_thread;
770         struct l_wait_info       lwi    = { 0 };
771         int                      rc;
772         ENTRY;
773
774         rc = cfs_create_thread(qmt_reba_thread, (void *)qmt, 0);
775         if (rc < 0) {
776                 CERROR("%s: failed to start rebalance thread (%d)\n",
777                        qmt->qmt_svname, rc);
778                 thread_set_flags(thread, SVC_STOPPED);
779                 RETURN(rc);
780         }
781
782         l_wait_event(thread->t_ctl_waitq,
783                      thread_is_running(thread) || thread_is_stopped(thread),
784                      &lwi);
785
786         RETURN(0);
787 }
788
789 /*
790  * Stop rebalance thread. Called when the QMT is about to shutdown.
791  */
792 void qmt_stop_reba_thread(struct qmt_device *qmt)
793 {
794         struct ptlrpc_thread *thread = &qmt->qmt_reba_thread;
795
796         if (!thread_is_stopped(thread)) {
797                 struct l_wait_info lwi = { 0 };
798
799                 thread_set_flags(thread, SVC_STOPPING);
800                 cfs_waitq_signal(&thread->t_ctl_waitq);
801
802                 l_wait_event(thread->t_ctl_waitq, thread_is_stopped(thread),
803                              &lwi);
804         }
805         LASSERT(cfs_list_empty(&qmt->qmt_reba_list));
806 }