Whamcloud - gitweb
LU-10973 lnet: infrastructure to build the LUTF
[fs/lustre-release.git] / lustre / quota / qsd_writeback.c
index 8352e59..6584df5 100644 (file)
@@ -21,7 +21,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2012, 2014, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.
  * Use is subject to license terms.
  *
  * Author: Johann Lombardi <johann.lombardi@intel.com>
@@ -33,8 +33,6 @@
 #include <linux/kthread.h>
 #include "qsd_internal.h"
 
-extern struct kmem_cache *upd_kmem;
-
 /*
  * Allocate and fill an qsd_upd_rec structure to be processed by the writeback
  * thread.
@@ -87,7 +85,8 @@ static void qsd_upd_add(struct qsd_instance *qsd, struct qsd_upd_rec *upd)
        if (!qsd->qsd_stopping) {
                list_add_tail(&upd->qur_link, &qsd->qsd_upd_list);
                /* wake up the upd thread */
-               wake_up(&qsd->qsd_upd_thread.t_ctl_waitq);
+               if (qsd->qsd_upd_task)
+                       wake_up_process(qsd->qsd_upd_task);
        } else {
                CWARN("%s: discard update.\n", qsd->qsd_svname);
                if (upd->qur_lqe)
@@ -121,9 +120,14 @@ static void qsd_add_deferred(struct qsd_instance *qsd, struct list_head *list,
                 * updates. We should just delete the legacy record in such
                 * case. */
                if (upd->qur_ver == tmp->qur_ver) {
-                       LASSERT(tmp->qur_lqe);
-                       LQUOTA_ERROR(tmp->qur_lqe, "Found a conflict record "
-                                    "with ver:%llu", tmp->qur_ver);
+                       if (tmp->qur_lqe)
+                               LQUOTA_WARN(tmp->qur_lqe, "Found a conflict "
+                                           "record with ver:%llu",
+                                           tmp->qur_ver);
+                       else
+                               CWARN("%s: Found a conflict record with ver: "
+                                     "%llu\n", qsd->qsd_svname, tmp->qur_ver);
+
                        list_del_init(&tmp->qur_link);
                        qsd_upd_free(tmp);
                } else if (upd->qur_ver < tmp->qur_ver) {
@@ -293,6 +297,11 @@ static int qsd_process_upd(const struct lu_env *env, struct qsd_upd_rec *upd)
                        GOTO(out, rc);
                /* refresh usage */
                qsd_refresh_usage(env, lqe);
+
+               spin_lock(&qqi->qqi_qsd->qsd_adjust_lock);
+               lqe->lqe_adjust_time = 0;
+               spin_unlock(&qqi->qqi_qsd->qsd_adjust_lock);
+
                /* Report usage asynchronously */
                rc = qsd_adjust(env, lqe);
                if (rc)
@@ -302,6 +311,21 @@ static int qsd_process_upd(const struct lu_env *env, struct qsd_upd_rec *upd)
        rc = qsd_update_index(env, qqi, &upd->qur_qid, upd->qur_global,
                              upd->qur_ver, &upd->qur_rec);
 out:
+       if (upd->qur_global && rc == 0 &&
+           upd->qur_rec.lqr_glb_rec.qbr_softlimit == 0 &&
+           upd->qur_rec.lqr_glb_rec.qbr_hardlimit == 0 &&
+           (LQUOTA_FLAG(upd->qur_rec.lqr_glb_rec.qbr_time) &
+                                                       LQUOTA_FLAG_DEFAULT)) {
+               lqe->lqe_is_default = true;
+               if (qqi->qqi_default_softlimit == 0 &&
+                   qqi->qqi_default_hardlimit == 0)
+                       lqe->lqe_enforced = false;
+               else
+                       lqe->lqe_enforced = true;
+
+               LQUOTA_DEBUG(lqe, "update to use default quota");
+       }
+
        if (lqe && !IS_ERR(lqe)) {
                lqe_putref(lqe);
                upd->qur_lqe = NULL;
@@ -333,12 +357,14 @@ void qsd_adjust_schedule(struct lquota_entry *lqe, bool defer, bool cancel)
        }
 
        if (list_empty(&lqe->lqe_link)) {
-               if (cancel)
+               if (!cancel) {
+                       lqe->lqe_adjust_time = ktime_get_seconds();
+                       if (defer)
+                               lqe->lqe_adjust_time += QSD_WB_INTERVAL;
+               } else {
                        lqe->lqe_adjust_time = 0;
-               else
-                       lqe->lqe_adjust_time = defer ?
-                               cfs_time_shift_64(QSD_WB_INTERVAL) :
-                               cfs_time_current_64();
+               }
+
                /* lqe reference transferred to list */
                if (defer)
                        list_add_tail(&lqe->lqe_link,
@@ -349,10 +375,14 @@ void qsd_adjust_schedule(struct lquota_entry *lqe, bool defer, bool cancel)
        }
        spin_unlock(&qsd->qsd_adjust_lock);
 
-       if (added)
-               wake_up(&qsd->qsd_upd_thread.t_ctl_waitq);
-       else
+       if (!added)
                lqe_putref(lqe);
+       else {
+               read_lock(&qsd->qsd_lock);
+               if (qsd->qsd_upd_task)
+                       wake_up_process(qsd->qsd_upd_task);
+               read_unlock(&qsd->qsd_lock);
+       }
 }
 
 /* return true if there is pending writeback records or the pending
@@ -371,8 +401,7 @@ static bool qsd_job_pending(struct qsd_instance *qsd, struct list_head *upd,
                struct lquota_entry *lqe;
                lqe = list_entry(qsd->qsd_adjust_list.next,
                                     struct lquota_entry, lqe_link);
-               if (cfs_time_beforeq_64(lqe->lqe_adjust_time,
-                                       cfs_time_current_64()))
+               if (ktime_get_seconds() >= lqe->lqe_adjust_time)
                        job_pending = true;
        }
        spin_unlock(&qsd->qsd_adjust_lock);
@@ -383,16 +412,14 @@ static bool qsd_job_pending(struct qsd_instance *qsd, struct list_head *upd,
                job_pending = true;
        }
 
-       if (qsd->qsd_acct_failed) {
-               /* don't bother kicking off reintegration if space accounting
-                * failed to be enabled */
-               write_unlock(&qsd->qsd_lock);
-               return job_pending;
-       }
-
        for (qtype = USRQUOTA; qtype < LL_MAXQUOTAS; qtype++) {
                struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype];
 
+               /* don't bother kicking off reintegration if space accounting
+                * failed to be enabled */
+               if (qqi->qqi_acct_failed)
+                       continue;
+
                if (!qsd_type_enabled(qsd, qtype))
                        continue;
 
@@ -407,40 +434,42 @@ static bool qsd_job_pending(struct qsd_instance *qsd, struct list_head *upd,
        return job_pending;
 }
 
-static int qsd_upd_thread(void *arg)
+struct qsd_upd_args {
+       struct qsd_instance     *qua_inst;
+       struct lu_env            qua_env;
+       struct completion       *qua_started;
+};
+
+#ifndef TASK_IDLE
+/* This identity is only safe inside kernel threads, or other places where
+ * all signals are disabled.  So it is placed here rather than in an include
+ * file.
+ * TASK_IDLE was added in v4.1-rc4-43-g80ed87c8a9ca so this can be removed
+ * when we no longer support kernels older than that.
+ */
+#define TASK_IDLE TASK_INTERRUPTIBLE
+#endif
+
+static int qsd_upd_thread(void *_args)
 {
-       struct qsd_instance     *qsd = (struct qsd_instance *)arg;
-       struct ptlrpc_thread    *thread = &qsd->qsd_upd_thread;
-       struct l_wait_info       lwi;
-       struct list_head         queue;
+       struct qsd_upd_args     *args = _args;
+       struct qsd_instance     *qsd = args->qua_inst;
+       LIST_HEAD(queue);
        struct qsd_upd_rec      *upd, *n;
-       struct lu_env           *env;
+       struct lu_env           *env = &args->qua_env;
        int                      qtype, rc = 0;
        bool                     uptodate;
        struct lquota_entry     *lqe;
-       __u64                    cur_time;
+       time64_t cur_time;
        ENTRY;
 
-       OBD_ALLOC_PTR(env);
-       if (env == NULL)
-               RETURN(-ENOMEM);
-
-       rc = lu_env_init(env, LCT_DT_THREAD);
-       if (rc) {
-               CERROR("%s: cannot init env: rc = %d\n", qsd->qsd_svname, rc);
-               OBD_FREE_PTR(env);
-               RETURN(rc);
-       }
-
-       thread_set_flags(thread, SVC_RUNNING);
-       wake_up(&thread->t_ctl_waitq);
+       complete(args->qua_started);
+       while (({set_current_state(TASK_IDLE);
+                !kthread_should_stop(); })) {
 
-       INIT_LIST_HEAD(&queue);
-       lwi = LWI_TIMEOUT(cfs_time_seconds(QSD_WB_INTERVAL), NULL, NULL);
-       while (1) {
-               l_wait_event(thread->t_ctl_waitq,
-                            qsd_job_pending(qsd, &queue, &uptodate) ||
-                            !thread_is_running(thread), &lwi);
+               if (!qsd_job_pending(qsd, &queue, &uptodate))
+                       schedule_timeout(cfs_time_seconds(QSD_WB_INTERVAL));
+               __set_current_state(TASK_RUNNING);
 
                list_for_each_entry_safe(upd, n, &queue, qur_link) {
                        list_del_init(&upd->qur_link);
@@ -449,19 +478,18 @@ static int qsd_upd_thread(void *arg)
                }
 
                spin_lock(&qsd->qsd_adjust_lock);
-               cur_time = cfs_time_current_64();
+               cur_time = ktime_get_seconds();
                while (!list_empty(&qsd->qsd_adjust_list)) {
                        lqe = list_entry(qsd->qsd_adjust_list.next,
                                         struct lquota_entry, lqe_link);
                        /* deferred items are sorted by time */
-                       if (!cfs_time_beforeq_64(lqe->lqe_adjust_time,
-                                                cur_time))
+                       if (lqe->lqe_adjust_time > cur_time)
                                break;
 
                        list_del_init(&lqe->lqe_link);
                        spin_unlock(&qsd->qsd_adjust_lock);
 
-                       if (thread_is_running(thread) && uptodate) {
+                       if (!kthread_should_stop() && uptodate) {
                                qsd_refresh_usage(env, lqe);
                                if (lqe->lqe_adjust_time == 0)
                                        qsd_id_lock_cancel(env, lqe);
@@ -474,42 +502,58 @@ static int qsd_upd_thread(void *arg)
                }
                spin_unlock(&qsd->qsd_adjust_lock);
 
-               if (!thread_is_running(thread))
-                       break;
-
-               if (uptodate)
+               if (uptodate || kthread_should_stop())
                        continue;
 
                for (qtype = USRQUOTA; qtype < LL_MAXQUOTAS; qtype++)
                        qsd_start_reint_thread(qsd->qsd_type_array[qtype]);
        }
+       __set_current_state(TASK_RUNNING);
+
        lu_env_fini(env);
-       OBD_FREE_PTR(env);
-       thread_set_flags(thread, SVC_STOPPED);
-       wake_up(&thread->t_ctl_waitq);
+       OBD_FREE_PTR(args);
+
        RETURN(rc);
 }
 
 int qsd_start_upd_thread(struct qsd_instance *qsd)
 {
-       struct ptlrpc_thread    *thread = &qsd->qsd_upd_thread;
-       struct l_wait_info       lwi = { 0 };
-       struct task_struct              *task;
+       struct qsd_upd_args *args;
+       struct task_struct *task;
+       DECLARE_COMPLETION_ONSTACK(started);
+       int rc;
        ENTRY;
 
-       task = kthread_run(qsd_upd_thread, (void *)qsd,
-                          "lquota_wb_%s", qsd->qsd_svname);
+       OBD_ALLOC_PTR(args);
+       if (args == NULL)
+               RETURN(-ENOMEM);
+
+       rc = lu_env_init(&args->qua_env, LCT_DT_THREAD);
+       if (rc) {
+               CERROR("%s: cannot init env: rc = %d\n", qsd->qsd_svname, rc);
+               goto out_free;
+       }
+       args->qua_inst = qsd;
+       args->qua_started = &started;
+
+       task = kthread_create(qsd_upd_thread, args,
+                             "lquota_wb_%s", qsd->qsd_svname);
        if (IS_ERR(task)) {
-               CERROR("fail to start quota update thread: rc = %ld\n",
-                       PTR_ERR(task));
-               thread_set_flags(thread, SVC_STOPPED);
-               RETURN(PTR_ERR(task));
+               rc = PTR_ERR(task);
+               CERROR("fail to start quota update thread: rc = %d\n", rc);
+               goto out_fini;
        }
+       qsd->qsd_upd_task = task;
+       wake_up_process(task);
+       wait_for_completion(&started);
 
-       l_wait_event(thread->t_ctl_waitq,
-                    thread_is_running(thread) || thread_is_stopped(thread),
-                    &lwi);
        RETURN(0);
+
+out_fini:
+       lu_env_fini(&args->qua_env);
+out_free:
+       OBD_FREE_PTR(args);
+       RETURN(rc);
 }
 
 static void qsd_cleanup_deferred(struct qsd_instance *qsd)
@@ -562,16 +606,15 @@ static void qsd_cleanup_adjust(struct qsd_instance *qsd)
 
 void qsd_stop_upd_thread(struct qsd_instance *qsd)
 {
-       struct ptlrpc_thread    *thread = &qsd->qsd_upd_thread;
-       struct l_wait_info       lwi    = { 0 };
+       struct task_struct *task;
 
-       if (!thread_is_stopped(thread)) {
-               thread_set_flags(thread, SVC_STOPPING);
-               wake_up(&thread->t_ctl_waitq);
+       write_lock(&qsd->qsd_lock);
+       task = qsd->qsd_upd_task;
+       qsd->qsd_upd_task = NULL;
+       write_unlock(&qsd->qsd_lock);
+       if (task)
+               kthread_stop(task);
 
-               l_wait_event(thread->t_ctl_waitq, thread_is_stopped(thread),
-                            &lwi);
-       }
        qsd_cleanup_deferred(qsd);
        qsd_cleanup_adjust(qsd);
 }