ocfs2: Periodic quota syncing

This patch creates a work queue for periodic syncing of locally cached quota
information to the global quota files. We constantly queue a delayed work
item, to get the periodic behavior.

Signed-off-by: Mark Fasheh <mfasheh@suse.com>
Acked-by: Jan Kara <jack@suse.cz>

4 files changed, 101 insertions, 0 deletions

diff --git a/fs/ocfs2/quota.h b/fs/ocfs2/quota.h
index 1f1c86311b32..e2233d51507f 100644
--- a/fs/ocfs2/quota.h
+++ b/fs/ocfs2/quota.h
@@ -39,6 +39,7 @@ struct ocfs2_mem_dqinfo {
         unsigned int dqi_chunks;        /* Number of chunks in local quota file */
         unsigned int dqi_blocks;        /* Number of blocks allocated for local quota file */
         unsigned int dqi_syncms;        /* How often should we sync with other nodes */
+        unsigned int dqi_syncjiff;      /* Precomputed dqi_syncms in jiffies */
         struct list_head dqi_chunk;     /* List of chunks */
         struct inode *dqi_gqinode;      /* Global quota file inode */
         struct ocfs2_lock_res dqi_gqlock;       /* Lock protecting quota information structure */
@@ -47,6 +48,7 @@ struct ocfs2_mem_dqinfo {
         struct buffer_head *dqi_lqi_bh; /* Buffer head with local quota file inode */
         struct buffer_head *dqi_ibh;    /* Buffer with information header */
         struct qtree_mem_dqinfo dqi_gi; /* Info about global file */
+        struct delayed_work dqi_sync_work;      /* Work for syncing dquots */
 };
 
 static inline struct ocfs2_dquot *OCFS2_DQUOT(struct dquot *dquot)
@@ -90,4 +92,7 @@ struct buffer_head *ocfs2_read_quota_block(struct inode *inode,
 extern struct dquot_operations ocfs2_quota_operations;
 extern struct quota_format_type ocfs2_quota_format;
 
+int ocfs2_quota_setup(void);
+void ocfs2_quota_shutdown(void);
+
 #endif /* _OCFS2_QUOTA_H */
diff --git a/fs/ocfs2/quota_global.c b/fs/ocfs2/quota_global.c
index af8340c45367..adf53508bdb8 100644
--- a/fs/ocfs2/quota_global.c
+++ b/fs/ocfs2/quota_global.c
@@ -1,10 +1,14 @@
 /*
  *  Implementation of operations over global quota file
  */
+#include <linux/spinlock.h>
 #include <linux/fs.h>
 #include <linux/quota.h>
 #include <linux/quotaops.h>
 #include <linux/dqblk_qtree.h>
+#include <linux/jiffies.h>
+#include <linux/writeback.h>
+#include <linux/workqueue.h>
 
 #define MLOG_MASK_PREFIX ML_QUOTA
 #include <cluster/masklog.h>
@@ -20,6 +24,10 @@
 #include "uptodate.h"
 #include "quota.h"
 
+static struct workqueue_struct *ocfs2_quota_wq = NULL;
+
+static void qsync_work_fn(struct work_struct *work);
+
 static void ocfs2_global_disk2memdqb(struct dquot *dquot, void *dp)
 {
         struct ocfs2_global_disk_dqblk *d = dp;
@@ -313,6 +321,7 @@ int ocfs2_global_read_info(struct super_block *sb, int type)
         info->dqi_bgrace = le32_to_cpu(dinfo.dqi_bgrace);
         info->dqi_igrace = le32_to_cpu(dinfo.dqi_igrace);
         oinfo->dqi_syncms = le32_to_cpu(dinfo.dqi_syncms);
+        oinfo->dqi_syncjiff = msecs_to_jiffies(oinfo->dqi_syncms);
         oinfo->dqi_gi.dqi_blocks = le32_to_cpu(dinfo.dqi_blocks);
         oinfo->dqi_gi.dqi_free_blk = le32_to_cpu(dinfo.dqi_free_blk);
         oinfo->dqi_gi.dqi_free_entry = le32_to_cpu(dinfo.dqi_free_entry);
@@ -320,6 +329,10 @@ int ocfs2_global_read_info(struct super_block *sb, int type)
         oinfo->dqi_gi.dqi_usable_bs = sb->s_blocksize -
                                                 OCFS2_QBLK_RESERVED_SPACE;
         oinfo->dqi_gi.dqi_qtree_depth = qtree_depth(&oinfo->dqi_gi);
+        INIT_DELAYED_WORK(&oinfo->dqi_sync_work, qsync_work_fn);
+        queue_delayed_work(ocfs2_quota_wq, &oinfo->dqi_sync_work,
+                           oinfo->dqi_syncjiff);
+
 out_err:
         mlog_exit(status);
         return status;
@@ -520,6 +533,61 @@ out:
 }
 
 /*
+ *  Functions for periodic syncing of dquots with global file
+ */
+static int ocfs2_sync_dquot_helper(struct dquot *dquot, unsigned long type)
+{
+        handle_t *handle;
+        struct super_block *sb = dquot->dq_sb;
+        struct ocfs2_mem_dqinfo *oinfo = sb_dqinfo(sb, type)->dqi_priv;
+        struct ocfs2_super *osb = OCFS2_SB(sb);
+        int status = 0;
+
+        mlog_entry("id=%u qtype=%u type=%lu device=%s\n", dquot->dq_id,
+                   dquot->dq_type, type, sb->s_id);
+        if (type != dquot->dq_type)
+                goto out;
+        status = ocfs2_lock_global_qf(oinfo, 1);
+        if (status < 0)
+                goto out;
+
+        handle = ocfs2_start_trans(osb, OCFS2_QSYNC_CREDITS);
+        if (IS_ERR(handle)) {
+                status = PTR_ERR(handle);
+                mlog_errno(status);
+                goto out_ilock;
+        }
+        mutex_lock(&sb_dqopt(sb)->dqio_mutex);
+        status = ocfs2_sync_dquot(dquot);
+        mutex_unlock(&sb_dqopt(sb)->dqio_mutex);
+        if (status < 0)
+                mlog_errno(status);
+        /* We have to write local structure as well... */
+        dquot_mark_dquot_dirty(dquot);
+        status = dquot_commit(dquot);
+        if (status < 0)
+                mlog_errno(status);
+        ocfs2_commit_trans(osb, handle);
+out_ilock:
+        ocfs2_unlock_global_qf(oinfo, 1);
+out:
+        mlog_exit(status);
+        return status;
+}
+
+static void qsync_work_fn(struct work_struct *work)
+{
+        struct ocfs2_mem_dqinfo *oinfo = container_of(work,
+                                                      struct ocfs2_mem_dqinfo,
+                                                      dqi_sync_work.work);
+        struct super_block *sb = oinfo->dqi_gqinode->i_sb;
+
+        dquot_scan_active(sb, ocfs2_sync_dquot_helper, oinfo->dqi_type);
+        queue_delayed_work(ocfs2_quota_wq, &oinfo->dqi_sync_work,
+                           oinfo->dqi_syncjiff);
+}
+
+/*
  *  Wrappers for generic quota functions
  */
 
@@ -917,3 +985,20 @@ struct dquot_operations ocfs2_quota_operations = {
         .alloc_dquot    = ocfs2_alloc_dquot,
         .destroy_dquot  = ocfs2_destroy_dquot,
 };
+
+int ocfs2_quota_setup(void)
+{
+        ocfs2_quota_wq = create_workqueue("o2quot");
+        if (!ocfs2_quota_wq)
+                return -ENOMEM;
+        return 0;
+}
+
+void ocfs2_quota_shutdown(void)
+{
+        if (ocfs2_quota_wq) {
+                flush_workqueue(ocfs2_quota_wq);
+                destroy_workqueue(ocfs2_quota_wq);
+                ocfs2_quota_wq = NULL;
+        }
+}
diff --git a/fs/ocfs2/quota_local.c b/fs/ocfs2/quota_local.c
index 55c3f2f98dcd..40e82b483136 100644
--- a/fs/ocfs2/quota_local.c
+++ b/fs/ocfs2/quota_local.c
@@ -368,6 +368,10 @@ static int ocfs2_local_free_info(struct super_block *sb, int type)
         int mark_clean = 1, len;
         int status;
 
+        /* At this point we know there are no more dquots and thus
+         * even if there's some sync in the pdflush queue, it won't
+         * find any dquots and return without doing anything */
+        cancel_delayed_work_sync(&oinfo->dqi_sync_work);
         iput(oinfo->dqi_gqinode);
         ocfs2_simple_drop_lockres(OCFS2_SB(sb), &oinfo->dqi_gqlock);
         ocfs2_lock_res_free(&oinfo->dqi_gqlock);
diff --git a/fs/ocfs2/super.c b/fs/ocfs2/super.c
index 7bb83e41581e..60f1d29421ad 100644
--- a/fs/ocfs2/super.c
+++ b/fs/ocfs2/super.c
@@ -1107,11 +1107,16 @@ static int __init ocfs2_init(void)
                 mlog(ML_ERROR, "Unable to create ocfs2 debugfs root.\n");
         }
 
+        status = ocfs2_quota_setup();
+        if (status)
+                goto leave;
+
         ocfs2_set_locking_protocol();
 
         status = register_quota_format(&ocfs2_quota_format);
 leave:
         if (status < 0) {
+                ocfs2_quota_shutdown();
                 ocfs2_free_mem_caches();
                 exit_ocfs2_uptodate_cache();
         }
@@ -1128,6 +1133,8 @@ static void __exit ocfs2_exit(void)
 {
         mlog_entry_void();
 
+        ocfs2_quota_shutdown();
+
         if (ocfs2_wq) {
                 flush_workqueue(ocfs2_wq);
                 destroy_workqueue(ocfs2_wq);