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