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) 2011, 2012, Intel, Inc.
25 * Use is subject to license terms.
27 * Author: Johann Lombardi <johann.lombardi@intel.com>
28 * Author: Niu Yawei <yawei.niu@intel.com>
32 # define EXPORT_SYMTAB
35 #define DEBUG_SUBSYSTEM S_LQUOTA
37 #include "qsd_internal.h"
39 extern cfs_mem_cache_t *upd_kmem;
42 * Allocate and fill an qsd_upd_rec structure to be processed by the writeback
45 * \param qqi - is the qsd_qtype_info structure relevant to the update
46 * \param lqe - is the lquota entry subject to the update
47 * \param qid - is the identifier subject to the update
48 * \param rec - is the record storing the new quota settings
49 * \param ver - is the version associated with the update
50 * \param global - is a boolean set to true if this is an update of the global
51 * index and false for a slave index.
53 static struct qsd_upd_rec *qsd_upd_alloc(struct qsd_qtype_info *qqi,
54 struct lquota_entry *lqe,
56 union lquota_rec *rec, __u64 ver,
59 struct qsd_upd_rec *upd;
61 OBD_SLAB_ALLOC_PTR_GFP(upd, upd_kmem, CFS_ALLOC_IO);
63 CERROR("Failed to allocate upd");
68 CFS_INIT_LIST_HEAD(&upd->qur_link);
76 upd->qur_global = global;
81 static void qsd_upd_free(struct qsd_upd_rec *upd)
84 lqe_putref(upd->qur_lqe);
85 OBD_SLAB_FREE_PTR(upd, upd_kmem);
88 /* must hold the qsd_lock */
89 static void qsd_upd_add(struct qsd_instance *qsd, struct qsd_upd_rec *upd)
91 if (!qsd->qsd_stopping) {
92 list_add_tail(&upd->qur_link, &qsd->qsd_upd_list);
93 /* wake up the upd thread */
94 cfs_waitq_signal(&qsd->qsd_upd_thread.t_ctl_waitq);
96 CWARN("%s: discard deferred update.\n", qsd->qsd_svname);
98 LQUOTA_WARN(upd->qur_lqe, "discard deferred update.");
103 /* must hold the qsd_lock */
104 static void qsd_add_deferred(cfs_list_t *list, struct qsd_upd_rec *upd)
106 struct qsd_upd_rec *tmp;
108 /* Sort the updates in ascending order */
109 cfs_list_for_each_entry_reverse(tmp, list, qur_link) {
111 LASSERTF(upd->qur_ver != tmp->qur_ver, "ver:"LPU64"\n",
114 if (upd->qur_ver < tmp->qur_ver) {
117 cfs_list_add_tail(&upd->qur_link, &tmp->qur_link);
121 cfs_list_add(&upd->qur_link, list);
124 /* must hold the qsd_lock */
125 static void qsd_kickoff_deferred(struct qsd_qtype_info *qqi, cfs_list_t *list,
128 struct qsd_upd_rec *upd, *tmp;
131 /* Get the first update record in the list, which has the smallest
132 * version, discard all records with versions smaller than the current
134 cfs_list_for_each_entry_safe(upd, tmp, list, qur_link) {
135 if (upd->qur_ver <= ver) {
136 /* drop this update */
137 cfs_list_del_init(&upd->qur_link);
138 CDEBUG(D_QUOTA, "%s: skipping deferred update ver:"
139 LPU64"/"LPU64", global:%d, qid:"LPU64"\n",
140 qqi->qqi_qsd->qsd_svname, upd->qur_ver, ver,
141 upd->qur_global, upd->qur_qid.qid_uid);
148 /* No remaining deferred update */
149 if (cfs_list_empty(list))
152 CDEBUG(D_QUOTA, "%s: found deferred update record. "
153 "version:"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);
157 LASSERTF(upd->qur_ver > ver, "lur_ver:"LPU64", cur_ver:"LPU64"\n",
160 /* Kick off the deferred udpate */
161 if (upd->qur_ver == ver + 1) {
162 list_del_init(&upd->qur_link);
163 qsd_upd_add(qqi->qqi_qsd, upd);
168 /* Bump version of global or slave index copy
170 * \param qqi - qsd_qtype_info
171 * \param ver - version to be bumped to
172 * \param global - global or slave index copy?
174 void qsd_bump_version(struct qsd_qtype_info *qqi, __u64 ver, bool global)
179 idx_ver = global ? &qqi->qqi_glb_ver : &qqi->qqi_slv_ver;
180 list = global ? &qqi->qqi_deferred_glb : &qqi->qqi_deferred_slv;
182 cfs_write_lock(&qqi->qqi_qsd->qsd_lock);
185 qqi->qqi_glb_uptodate = 1;
187 qqi->qqi_slv_uptodate = 1;
188 qsd_kickoff_deferred(qqi, list, ver);
189 cfs_write_unlock(&qqi->qqi_qsd->qsd_lock);
193 * Schedule a commit of a lquota entry
195 * \param qqi - qsd_qtype_info
196 * \param lqe - lquota_entry
197 * \param qid - quota id
198 * \param rec - global or slave record to be updated to disk
199 * \param ver - new index file version
200 * \param global- ture : master record; false : slave record
202 void qsd_upd_schedule(struct qsd_qtype_info *qqi, struct lquota_entry *lqe,
203 union lquota_id *qid, union lquota_rec *rec, __u64 ver,
206 struct qsd_upd_rec *upd;
207 struct qsd_instance *qsd = qqi->qqi_qsd;
211 CDEBUG(D_QUOTA, "%s: schedule update. global:%s, version:"LPU64"\n",
212 qsd->qsd_svname, global ? "true" : "false", ver);
214 upd = qsd_upd_alloc(qqi, lqe, qid, rec, ver, global);
218 /* If we don't want update index version, no need to sort the
219 * records in version order, just schedule the updates instantly. */
221 cfs_write_lock(&qsd->qsd_lock);
222 qsd_upd_add(qsd, upd);
223 cfs_write_unlock(&qsd->qsd_lock);
227 cfs_write_lock(&qsd->qsd_lock);
229 cur_ver = global ? qqi->qqi_glb_ver : qqi->qqi_slv_ver;
231 if (ver <= cur_ver) {
233 /* legitimate race between glimpse AST and
235 CDEBUG(D_QUOTA, "%s: discarding glb update from glimpse"
236 " ver:"LPU64" local ver:"LPU64"\n",
237 qsd->qsd_svname, ver, cur_ver);
239 CERROR("%s: discard slv update, ver:"LPU64" local ver:"
240 LPU64"\n", qsd->qsd_svname, ver, cur_ver);
242 } else if ((ver == cur_ver + 1) && qqi->qqi_glb_uptodate &&
243 qqi->qqi_slv_uptodate) {
244 /* In order update, and reintegration has been done. */
245 qsd_upd_add(qsd, upd);
247 /* Out of order update (the one with smaller version hasn't
248 * reached slave or hasn't been flushed to disk yet), or
249 * the reintegration is in progress. Defer the update. */
250 cfs_list_t *list = global ? &qqi->qqi_deferred_glb :
251 &qqi->qqi_deferred_slv;
252 qsd_add_deferred(list, upd);
255 cfs_write_unlock(&qsd->qsd_lock);
260 static int qsd_process_upd(const struct lu_env *env, struct qsd_upd_rec *upd)
262 struct lquota_entry *lqe = upd->qur_lqe;
263 struct qsd_qtype_info *qqi = upd->qur_qqi;
268 lqe = lqe_locate(env, qqi->qqi_site, &upd->qur_qid);
270 GOTO(out, rc = PTR_ERR(lqe));
273 /* The in-memory lqe update for slave index copy isn't deferred,
274 * we shouldn't touch it here. */
275 if (upd->qur_global) {
276 rc = qsd_update_lqe(env, lqe, upd->qur_global, &upd->qur_rec);
279 /* Report usage asynchronously */
280 if (lqe->lqe_enforced &&
281 !qsd_refresh_usage(env, lqe)) {
282 rc = qsd_dqacq(env, lqe, QSD_REP);
283 LQUOTA_DEBUG(lqe, "Report usage. rc:%d", rc);
287 rc = qsd_update_index(env, qqi, &upd->qur_qid, upd->qur_global,
288 upd->qur_ver, &upd->qur_rec);
290 if (lqe && !IS_ERR(lqe)) {
297 void qsd_adjust_schedule(struct lquota_entry *lqe, bool defer, bool cancel)
299 struct qsd_instance *qsd = lqe2qqi(lqe)->qqi_qsd;
303 cfs_spin_lock(&qsd->qsd_adjust_lock);
305 /* the lqe is being queued for the per-ID lock cancel, we should
306 * cancel the lock cancel and re-add it for quota adjust */
307 if (!cfs_list_empty(&lqe->lqe_link) &&
308 lqe->lqe_adjust_time == 0) {
309 cfs_list_del_init(&lqe->lqe_link);
313 if (cfs_list_empty(&lqe->lqe_link)) {
315 lqe->lqe_adjust_time = 0;
317 lqe->lqe_adjust_time = defer ?
318 cfs_time_shift_64(QSD_WB_INTERVAL) :
319 cfs_time_current_64();
320 /* lqe reference transfered to list */
322 cfs_list_add_tail(&lqe->lqe_link,
323 &qsd->qsd_adjust_list);
325 cfs_list_add(&lqe->lqe_link, &qsd->qsd_adjust_list);
328 cfs_spin_unlock(&qsd->qsd_adjust_lock);
331 cfs_waitq_signal(&qsd->qsd_upd_thread.t_ctl_waitq);
336 /* return true if there is pending writeback records or the pending
338 static bool qsd_job_pending(struct qsd_instance *qsd, cfs_list_t *upd,
341 bool job_pending = false;
344 LASSERT(cfs_list_empty(upd));
347 cfs_spin_lock(&qsd->qsd_adjust_lock);
348 if (!cfs_list_empty(&qsd->qsd_adjust_list)) {
349 struct lquota_entry *lqe;
350 lqe = cfs_list_entry(qsd->qsd_adjust_list.next,
351 struct lquota_entry, lqe_link);
352 if (cfs_time_beforeq_64(lqe->lqe_adjust_time,
353 cfs_time_current_64()))
356 cfs_spin_unlock(&qsd->qsd_adjust_lock);
358 cfs_write_lock(&qsd->qsd_lock);
359 if (!cfs_list_empty(&qsd->qsd_upd_list)) {
360 cfs_list_splice_init(&qsd->qsd_upd_list, upd);
364 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
365 struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype];
367 if (!qsd_type_enabled(qsd, qtype))
370 if ((!qqi->qqi_glb_uptodate || !qqi->qqi_slv_uptodate) &&
372 /* global or slave index not up to date and reint
373 * thread not running */
377 cfs_write_unlock(&qsd->qsd_lock);
381 static int qsd_upd_thread(void *arg)
383 struct qsd_instance *qsd = (struct qsd_instance *)arg;
384 struct ptlrpc_thread *thread = &qsd->qsd_upd_thread;
385 struct l_wait_info lwi;
387 struct qsd_upd_rec *upd, *n;
388 char pname[MTI_NAME_MAXLEN];
392 struct lquota_entry *lqe, *tmp;
400 rc = lu_env_init(env, LCT_DT_THREAD);
402 CERROR("%s: Fail to init env.", qsd->qsd_svname);
407 snprintf(pname, MTI_NAME_MAXLEN, "lquota_wb_%s", qsd->qsd_svname);
408 cfs_daemonize(pname);
410 thread_set_flags(thread, SVC_RUNNING);
411 cfs_waitq_signal(&thread->t_ctl_waitq);
413 CFS_INIT_LIST_HEAD(&queue);
414 lwi = LWI_TIMEOUT(cfs_time_seconds(QSD_WB_INTERVAL), NULL, NULL);
416 l_wait_event(thread->t_ctl_waitq,
417 qsd_job_pending(qsd, &queue, &uptodate) ||
418 !thread_is_running(thread), &lwi);
420 cfs_list_for_each_entry_safe(upd, n, &queue, qur_link) {
421 cfs_list_del_init(&upd->qur_link);
422 qsd_process_upd(env, upd);
426 cfs_spin_lock(&qsd->qsd_adjust_lock);
427 cur_time = cfs_time_current_64();
428 cfs_list_for_each_entry_safe(lqe, tmp, &qsd->qsd_adjust_list,
430 /* deferred items are sorted by time */
431 if (!cfs_time_beforeq_64(lqe->lqe_adjust_time,
435 cfs_list_del_init(&lqe->lqe_link);
436 cfs_spin_unlock(&qsd->qsd_adjust_lock);
438 if (thread_is_running(thread) && uptodate) {
439 qsd_refresh_usage(env, lqe);
440 if (lqe->lqe_adjust_time == 0)
441 qsd_id_lock_cancel(env, lqe);
443 qsd_dqacq(env, lqe, QSD_ADJ);
447 cfs_spin_lock(&qsd->qsd_adjust_lock);
449 cfs_spin_unlock(&qsd->qsd_adjust_lock);
451 if (!thread_is_running(thread))
457 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++)
458 qsd_start_reint_thread(qsd->qsd_type_array[qtype]);
462 thread_set_flags(thread, SVC_STOPPED);
463 cfs_waitq_signal(&thread->t_ctl_waitq);
467 int qsd_start_upd_thread(struct qsd_instance *qsd)
469 struct ptlrpc_thread *thread = &qsd->qsd_upd_thread;
470 struct l_wait_info lwi = { 0 };
474 rc = cfs_create_thread(qsd_upd_thread, (void *)qsd, 0);
476 CERROR("Fail to start quota update thread. rc: %d\n", rc);
477 thread_set_flags(thread, SVC_STOPPED);
481 l_wait_event(thread->t_ctl_waitq,
482 thread_is_running(thread) || thread_is_stopped(thread),
487 static void qsd_cleanup_deferred(struct qsd_instance *qsd)
491 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
492 struct qsd_upd_rec *upd, *tmp;
493 struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype];
498 cfs_write_lock(&qsd->qsd_lock);
499 cfs_list_for_each_entry_safe(upd, tmp, &qqi->qqi_deferred_glb,
501 CWARN("%s: Free global deferred upd: ID:"LPU64", "
502 "ver:"LPU64"/"LPU64"\n", qsd->qsd_svname,
503 upd->qur_qid.qid_uid, upd->qur_ver,
505 list_del_init(&upd->qur_link);
508 cfs_list_for_each_entry_safe(upd, tmp, &qqi->qqi_deferred_slv,
510 CWARN("%s: Free slave deferred upd: ID:"LPU64", "
511 "ver:"LPU64"/"LPU64"\n", qsd->qsd_svname,
512 upd->qur_qid.qid_uid, upd->qur_ver,
514 list_del_init(&upd->qur_link);
517 cfs_write_unlock(&qsd->qsd_lock);
521 static void qsd_cleanup_adjust(struct qsd_instance *qsd)
523 struct lquota_entry *lqe;
525 cfs_spin_lock(&qsd->qsd_adjust_lock);
526 while (!cfs_list_empty(&qsd->qsd_adjust_list)) {
527 lqe = cfs_list_entry(qsd->qsd_adjust_list.next,
528 struct lquota_entry, lqe_link);
529 cfs_list_del_init(&lqe->lqe_link);
532 cfs_spin_unlock(&qsd->qsd_adjust_lock);
535 void qsd_stop_upd_thread(struct qsd_instance *qsd)
537 struct ptlrpc_thread *thread = &qsd->qsd_upd_thread;
538 struct l_wait_info lwi = { 0 };
540 if (!thread_is_stopped(thread)) {
541 thread_set_flags(thread, SVC_STOPPING);
542 cfs_waitq_signal(&thread->t_ctl_waitq);
544 l_wait_event(thread->t_ctl_waitq, thread_is_stopped(thread),
547 qsd_cleanup_deferred(qsd);
548 qsd_cleanup_adjust(qsd);