4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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
24 * Copyright (c) 2012, 2013, Intel Corporation.
25 * Use is subject to license terms.
27 * Author: Johann Lombardi <johann.lombardi@intel.com>
28 * Author: Niu Yawei <yawei.niu@intel.com>
31 #define DEBUG_SUBSYSTEM S_LQUOTA
33 #include "qsd_internal.h"
35 extern struct kmem_cache *upd_kmem;
38 * Allocate and fill an qsd_upd_rec structure to be processed by the writeback
41 * \param qqi - is the qsd_qtype_info structure relevant to the update
42 * \param lqe - is the lquota entry subject to the update
43 * \param qid - is the identifier subject to the update
44 * \param rec - is the record storing the new quota settings
45 * \param ver - is the version associated with the update
46 * \param global - is a boolean set to true if this is an update of the global
47 * index and false for a slave index.
49 static struct qsd_upd_rec *qsd_upd_alloc(struct qsd_qtype_info *qqi,
50 struct lquota_entry *lqe,
52 union lquota_rec *rec, __u64 ver,
55 struct qsd_upd_rec *upd;
57 OBD_SLAB_ALLOC_PTR_GFP(upd, upd_kmem, GFP_NOFS);
63 INIT_LIST_HEAD(&upd->qur_link);
71 upd->qur_global = global;
76 static void qsd_upd_free(struct qsd_upd_rec *upd)
79 lqe_putref(upd->qur_lqe);
80 OBD_SLAB_FREE_PTR(upd, upd_kmem);
83 /* must hold the qsd_lock */
84 static void qsd_upd_add(struct qsd_instance *qsd, struct qsd_upd_rec *upd)
86 if (!qsd->qsd_stopping) {
87 list_add_tail(&upd->qur_link, &qsd->qsd_upd_list);
88 /* wake up the upd thread */
89 wake_up(&qsd->qsd_upd_thread.t_ctl_waitq);
91 CWARN("%s: discard update.\n", qsd->qsd_svname);
93 LQUOTA_WARN(upd->qur_lqe, "discard update.");
98 /* must hold the qsd_lock */
99 static void qsd_add_deferred(struct qsd_instance *qsd, struct list_head *list,
100 struct qsd_upd_rec *upd)
102 struct qsd_upd_rec *tmp, *n;
104 if (qsd->qsd_stopping) {
105 CWARN("%s: discard deferred udpate.\n", qsd->qsd_svname);
107 LQUOTA_WARN(upd->qur_lqe, "discard deferred update.");
112 /* Sort the updates in ascending order */
113 list_for_each_entry_safe_reverse(tmp, n, list, qur_link) {
115 /* There could be some legacy records which have duplicated
116 * version. Imagine following scenario: slave received global
117 * glimpse and queued a record in the deferred list, then
118 * master crash and rollback to an ealier version, then the
119 * version of queued record will be conflicting with later
120 * updates. We should just delete the legacy record in such
122 if (upd->qur_ver == tmp->qur_ver) {
123 LASSERT(tmp->qur_lqe);
124 LQUOTA_ERROR(tmp->qur_lqe, "Found a conflict record "
125 "with ver:"LPU64"", tmp->qur_ver);
126 list_del_init(&tmp->qur_link);
128 } else if (upd->qur_ver < tmp->qur_ver) {
131 list_add_tail(&upd->qur_link, &tmp->qur_link);
135 list_add(&upd->qur_link, list);
138 /* must hold the qsd_lock */
139 static void qsd_kickoff_deferred(struct qsd_qtype_info *qqi,
140 struct list_head *list, __u64 ver)
142 struct qsd_upd_rec *upd, *tmp;
145 /* Get the first update record in the list, which has the smallest
146 * version, discard all records with versions smaller than the current
148 list_for_each_entry_safe(upd, tmp, list, qur_link) {
149 if (upd->qur_ver <= ver) {
150 /* drop this update */
151 list_del_init(&upd->qur_link);
152 CDEBUG(D_QUOTA, "%s: skipping deferred update ver:"
153 LPU64"/"LPU64", global:%d, qid:"LPU64"\n",
154 qqi->qqi_qsd->qsd_svname, upd->qur_ver, ver,
155 upd->qur_global, upd->qur_qid.qid_uid);
162 /* No remaining deferred update */
163 if (list_empty(list))
166 CDEBUG(D_QUOTA, "%s: found deferred update record. "
167 "version:"LPU64"/"LPU64", global:%d, qid:"LPU64"\n",
168 qqi->qqi_qsd->qsd_svname, upd->qur_ver, ver,
169 upd->qur_global, upd->qur_qid.qid_uid);
171 LASSERTF(upd->qur_ver > ver, "lur_ver:"LPU64", cur_ver:"LPU64"\n",
174 /* Kick off the deferred udpate */
175 if (upd->qur_ver == ver + 1) {
176 list_del_init(&upd->qur_link);
177 qsd_upd_add(qqi->qqi_qsd, upd);
182 /* Bump version of global or slave index copy
184 * \param qqi - qsd_qtype_info
185 * \param ver - version to be bumped to
186 * \param global - global or slave index copy?
188 void qsd_bump_version(struct qsd_qtype_info *qqi, __u64 ver, bool global)
190 struct list_head *list;
193 idx_ver = global ? &qqi->qqi_glb_ver : &qqi->qqi_slv_ver;
194 list = global ? &qqi->qqi_deferred_glb : &qqi->qqi_deferred_slv;
196 write_lock(&qqi->qqi_qsd->qsd_lock);
199 qqi->qqi_glb_uptodate = 1;
201 qqi->qqi_slv_uptodate = 1;
202 qsd_kickoff_deferred(qqi, list, ver);
203 write_unlock(&qqi->qqi_qsd->qsd_lock);
207 * Schedule a commit of a lquota entry
209 * \param qqi - qsd_qtype_info
210 * \param lqe - lquota_entry
211 * \param qid - quota id
212 * \param rec - global or slave record to be updated to disk
213 * \param ver - new index file version
214 * \param global- true: master record; false: slave record
216 void qsd_upd_schedule(struct qsd_qtype_info *qqi, struct lquota_entry *lqe,
217 union lquota_id *qid, union lquota_rec *rec, __u64 ver,
220 struct qsd_upd_rec *upd;
221 struct qsd_instance *qsd = qqi->qqi_qsd;
225 CDEBUG(D_QUOTA, "%s: schedule update. global:%s, version:"LPU64"\n",
226 qsd->qsd_svname, global ? "true" : "false", ver);
228 upd = qsd_upd_alloc(qqi, lqe, qid, rec, ver, global);
232 /* If we don't want update index version, no need to sort the
233 * records in version order, just schedule the updates instantly. */
235 write_lock(&qsd->qsd_lock);
236 qsd_upd_add(qsd, upd);
237 write_unlock(&qsd->qsd_lock);
241 write_lock(&qsd->qsd_lock);
243 cur_ver = global ? qqi->qqi_glb_ver : qqi->qqi_slv_ver;
245 if (ver <= cur_ver) {
247 /* legitimate race between glimpse AST and
249 CDEBUG(D_QUOTA, "%s: discarding glb update from glimpse"
250 " ver:"LPU64" local ver:"LPU64"\n",
251 qsd->qsd_svname, ver, cur_ver);
253 CERROR("%s: discard slv update, ver:"LPU64" local ver:"
254 LPU64"\n", qsd->qsd_svname, ver, cur_ver);
256 } else if ((ver == cur_ver + 1) && qqi->qqi_glb_uptodate &&
257 qqi->qqi_slv_uptodate) {
258 /* In order update, and reintegration has been done. */
259 qsd_upd_add(qsd, upd);
261 /* Out of order update (the one with smaller version hasn't
262 * reached slave or hasn't been flushed to disk yet), or
263 * the reintegration is in progress. Defer the update. */
264 struct list_head *list = global ? &qqi->qqi_deferred_glb :
265 &qqi->qqi_deferred_slv;
266 qsd_add_deferred(qsd, list, upd);
269 write_unlock(&qsd->qsd_lock);
274 static int qsd_process_upd(const struct lu_env *env, struct qsd_upd_rec *upd)
276 struct lquota_entry *lqe = upd->qur_lqe;
277 struct qsd_qtype_info *qqi = upd->qur_qqi;
282 lqe = lqe_locate(env, qqi->qqi_site, &upd->qur_qid);
284 GOTO(out, rc = PTR_ERR(lqe));
287 /* The in-memory lqe update for slave index copy isn't deferred,
288 * we shouldn't touch it here. */
289 if (upd->qur_global) {
290 rc = qsd_update_lqe(env, lqe, upd->qur_global, &upd->qur_rec);
294 qsd_refresh_usage(env, lqe);
295 /* Report usage asynchronously */
296 rc = qsd_adjust(env, lqe);
298 LQUOTA_ERROR(lqe, "failed to report usage, rc:%d", rc);
301 rc = qsd_update_index(env, qqi, &upd->qur_qid, upd->qur_global,
302 upd->qur_ver, &upd->qur_rec);
304 if (lqe && !IS_ERR(lqe)) {
311 void qsd_adjust_schedule(struct lquota_entry *lqe, bool defer, bool cancel)
313 struct qsd_instance *qsd = lqe2qqi(lqe)->qqi_qsd;
316 read_lock(&qsd->qsd_lock);
317 if (qsd->qsd_stopping) {
318 read_unlock(&qsd->qsd_lock);
321 read_unlock(&qsd->qsd_lock);
324 spin_lock(&qsd->qsd_adjust_lock);
326 /* the lqe is being queued for the per-ID lock cancel, we should
327 * cancel the lock cancel and re-add it for quota adjust */
328 if (!list_empty(&lqe->lqe_link) &&
329 lqe->lqe_adjust_time == 0) {
330 list_del_init(&lqe->lqe_link);
334 if (list_empty(&lqe->lqe_link)) {
336 lqe->lqe_adjust_time = 0;
338 lqe->lqe_adjust_time = defer ?
339 cfs_time_shift_64(QSD_WB_INTERVAL) :
340 cfs_time_current_64();
341 /* lqe reference transfered to list */
343 list_add_tail(&lqe->lqe_link,
344 &qsd->qsd_adjust_list);
346 list_add(&lqe->lqe_link, &qsd->qsd_adjust_list);
349 spin_unlock(&qsd->qsd_adjust_lock);
352 wake_up(&qsd->qsd_upd_thread.t_ctl_waitq);
357 /* return true if there is pending writeback records or the pending
359 static bool qsd_job_pending(struct qsd_instance *qsd, struct list_head *upd,
362 bool job_pending = false;
365 LASSERT(list_empty(upd));
368 spin_lock(&qsd->qsd_adjust_lock);
369 if (!list_empty(&qsd->qsd_adjust_list)) {
370 struct lquota_entry *lqe;
371 lqe = list_entry(qsd->qsd_adjust_list.next,
372 struct lquota_entry, lqe_link);
373 if (cfs_time_beforeq_64(lqe->lqe_adjust_time,
374 cfs_time_current_64()))
377 spin_unlock(&qsd->qsd_adjust_lock);
379 write_lock(&qsd->qsd_lock);
380 if (!list_empty(&qsd->qsd_upd_list)) {
381 list_splice_init(&qsd->qsd_upd_list, upd);
385 if (qsd->qsd_acct_failed) {
386 /* don't bother kicking off reintegration if space accounting
387 * failed to be enabled */
388 write_unlock(&qsd->qsd_lock);
392 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
393 struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype];
395 if (!qsd_type_enabled(qsd, qtype))
398 if ((!qqi->qqi_glb_uptodate || !qqi->qqi_slv_uptodate) &&
400 /* global or slave index not up to date and reint
401 * thread not running */
405 write_unlock(&qsd->qsd_lock);
409 static int qsd_upd_thread(void *arg)
411 struct qsd_instance *qsd = (struct qsd_instance *)arg;
412 struct ptlrpc_thread *thread = &qsd->qsd_upd_thread;
413 struct l_wait_info lwi;
414 struct list_head queue;
415 struct qsd_upd_rec *upd, *n;
419 struct lquota_entry *lqe;
427 rc = lu_env_init(env, LCT_DT_THREAD);
429 CERROR("%s: cannot init env: rc = %d\n", qsd->qsd_svname, rc);
434 thread_set_flags(thread, SVC_RUNNING);
435 wake_up(&thread->t_ctl_waitq);
437 INIT_LIST_HEAD(&queue);
438 lwi = LWI_TIMEOUT(cfs_time_seconds(QSD_WB_INTERVAL), NULL, NULL);
440 l_wait_event(thread->t_ctl_waitq,
441 qsd_job_pending(qsd, &queue, &uptodate) ||
442 !thread_is_running(thread), &lwi);
444 list_for_each_entry_safe(upd, n, &queue, qur_link) {
445 list_del_init(&upd->qur_link);
446 qsd_process_upd(env, upd);
450 spin_lock(&qsd->qsd_adjust_lock);
451 cur_time = cfs_time_current_64();
452 while (!list_empty(&qsd->qsd_adjust_list)) {
453 lqe = list_entry(qsd->qsd_adjust_list.next,
454 struct lquota_entry, lqe_link);
455 /* deferred items are sorted by time */
456 if (!cfs_time_beforeq_64(lqe->lqe_adjust_time,
460 list_del_init(&lqe->lqe_link);
461 spin_unlock(&qsd->qsd_adjust_lock);
463 if (thread_is_running(thread) && uptodate) {
464 qsd_refresh_usage(env, lqe);
465 if (lqe->lqe_adjust_time == 0)
466 qsd_id_lock_cancel(env, lqe);
468 qsd_adjust(env, lqe);
472 spin_lock(&qsd->qsd_adjust_lock);
474 spin_unlock(&qsd->qsd_adjust_lock);
476 if (!thread_is_running(thread))
482 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++)
483 qsd_start_reint_thread(qsd->qsd_type_array[qtype]);
487 thread_set_flags(thread, SVC_STOPPED);
488 wake_up(&thread->t_ctl_waitq);
492 int qsd_start_upd_thread(struct qsd_instance *qsd)
494 struct ptlrpc_thread *thread = &qsd->qsd_upd_thread;
495 struct l_wait_info lwi = { 0 };
496 struct task_struct *task;
499 task = kthread_run(qsd_upd_thread, (void *)qsd,
500 "lquota_wb_%s", qsd->qsd_svname);
502 CERROR("fail to start quota update thread: rc = %ld\n",
504 thread_set_flags(thread, SVC_STOPPED);
505 RETURN(PTR_ERR(task));
508 l_wait_event(thread->t_ctl_waitq,
509 thread_is_running(thread) || thread_is_stopped(thread),
514 static void qsd_cleanup_deferred(struct qsd_instance *qsd)
518 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
519 struct qsd_upd_rec *upd, *tmp;
520 struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype];
525 write_lock(&qsd->qsd_lock);
526 list_for_each_entry_safe(upd, tmp, &qqi->qqi_deferred_glb,
528 CWARN("%s: Free global deferred upd: ID:"LPU64", "
529 "ver:"LPU64"/"LPU64"\n", qsd->qsd_svname,
530 upd->qur_qid.qid_uid, upd->qur_ver,
532 list_del_init(&upd->qur_link);
535 list_for_each_entry_safe(upd, tmp, &qqi->qqi_deferred_slv,
537 CWARN("%s: Free slave deferred upd: ID:"LPU64", "
538 "ver:"LPU64"/"LPU64"\n", qsd->qsd_svname,
539 upd->qur_qid.qid_uid, upd->qur_ver,
541 list_del_init(&upd->qur_link);
544 write_unlock(&qsd->qsd_lock);
548 static void qsd_cleanup_adjust(struct qsd_instance *qsd)
550 struct lquota_entry *lqe;
552 spin_lock(&qsd->qsd_adjust_lock);
553 while (!list_empty(&qsd->qsd_adjust_list)) {
554 lqe = list_entry(qsd->qsd_adjust_list.next,
555 struct lquota_entry, lqe_link);
556 list_del_init(&lqe->lqe_link);
559 spin_unlock(&qsd->qsd_adjust_lock);
562 void qsd_stop_upd_thread(struct qsd_instance *qsd)
564 struct ptlrpc_thread *thread = &qsd->qsd_upd_thread;
565 struct l_wait_info lwi = { 0 };
567 if (!thread_is_stopped(thread)) {
568 thread_set_flags(thread, SVC_STOPPING);
569 wake_up(&thread->t_ctl_waitq);
571 l_wait_event(thread->t_ctl_waitq, thread_is_stopped(thread),
574 qsd_cleanup_deferred(qsd);
575 qsd_cleanup_adjust(qsd);