scoutfs: plug the rest of our locking into dlmglue

We move struct ocfs2_lock_res_ops and flags to dlmglue.c so that
locks.c can get access to it. Similarly, we export
ocfs2_lock_res_init_common() for locks.c can initialize each lockres
before use. Also, free_lock_tree() now has to happen before we shut
down the dlm - this gives dlmglue the opportunity to unlock their
underlying dlm locks before we go off freeing the structures.

Signed-off-by: Mark Fasheh <mfasheh@versity.com>

kmod/src/data.c

@@ -1238,7 +1238,7 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
 		blk_off = ext.blk_off + ext.blocks;
 	}
 
-	scoutfs_unlock(sb, inode_lock);
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
 out:
 	mutex_unlock(&inode->i_mutex);
 

kmod/src/dir.c

@@ -283,7 +283,7 @@ out:
 	else
 		inode = scoutfs_iget(sb, ino);
 
-	scoutfs_unlock(sb, dir_lock);
+	scoutfs_unlock(sb, dir_lock, DLM_LOCK_PR);
 
 	scoutfs_key_free(sb, key);
 
@@ -390,7 +390,7 @@ static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir)
 	}
 
 out:
-	scoutfs_unlock(sb, dir_lock);
+	scoutfs_unlock(sb, dir_lock, DLM_LOCK_PR);
 
 	kfree(dent);
 	return ret;
@@ -545,8 +545,8 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 out:
 	scoutfs_release_trans(sb);
 out_unlock:
-	scoutfs_unlock(sb, dir_lock);
-	scoutfs_unlock(sb, inode_lock);
+	scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX);
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX);
 	/* XXX delete the inode item here */
 	if (ret && !IS_ERR_OR_NULL(inode))
 		iput(inode);
@@ -614,8 +614,8 @@ static int scoutfs_link(struct dentry *old_dentry,
 out:
 	scoutfs_release_trans(sb);
 out_unlock:
-	scoutfs_unlock(sb, dir_lock);
-	scoutfs_unlock(sb, inode_lock);
+	scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX);
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX);
 	return ret;
 }
 
@@ -714,8 +714,8 @@ out_trans:
 out:
 	scoutfs_key_free(sb, keys[0]);
 	scoutfs_key_free(sb, keys[2]);
-	scoutfs_unlock(sb, dir_lock);
-	scoutfs_unlock(sb, inode_lock);
+	scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX);
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX);
 	return ret;
 }
 
@@ -837,7 +837,7 @@ static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
 		nd_set_link(nd, path);
 	}
 out:
-	scoutfs_unlock(sb, inode_lock);
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
 	return path;
 }
 
@@ -934,8 +934,8 @@ out:
 
 	scoutfs_release_trans(sb);
 out_unlock:
-	scoutfs_unlock(sb, dir_lock);
-	scoutfs_unlock(sb, inode_lock);
+	scoutfs_unlock(sb, dir_lock, DLM_LOCK_EX);
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX);
 	return ret;
 }
 

kmod/src/dlmglue.c

@@ -97,6 +97,7 @@ static struct ocfs2_super *ocfs2_get_file_osb(struct ocfs2_lock_res *lockres);
 static struct ocfs2_super *ocfs2_get_qinfo_osb(struct ocfs2_lock_res *lockres);
 #endif
 
+#if 0
 /*
  * Return value from ->downconvert_worker functions.
  *
@@ -111,7 +112,7 @@ enum ocfs2_unblock_action {
 	UNBLOCK_STOP_POST	= 2, /* Do not downconvert, fire
 				      * ->post_unlock() callback. */
 };
-
+#endif
 struct ocfs2_unblock_ctl {
 	int requeue;
 	enum ocfs2_unblock_action unblock_action;
@@ -169,7 +170,6 @@ static void ocfs2_dump_meta_lvb_info(u64 level,
 	     (long long)be64_to_cpu(lvb->lvb_imtime_packed),
 	     be32_to_cpu(lvb->lvb_iattr));
 }
-#endif
 
 
 /*
@@ -260,7 +260,6 @@ struct ocfs2_lock_res_ops {
  */
 #define LOCK_TYPE_USES_LVB		0x2
 
-#if 0
 static struct ocfs2_lock_res_ops ocfs2_inode_rw_lops = {
 	.get_osb	= ocfs2_get_inode_osb,
 	.flags		= 0,
@@ -532,10 +531,10 @@ static inline void ocfs2_init_start_time(struct ocfs2_mask_waiter *mw)
 }
 #endif
 
-static void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
-				       struct ocfs2_lock_res *res,
-				       struct ocfs2_lock_res_ops *ops,
-				       void *priv)
+void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
+				struct ocfs2_lock_res *res,
+				struct ocfs2_lock_res_ops *ops,
+				void *priv)
 {
 	res->l_ops           = ops;
 	res->l_priv          = priv;

kmod/src/dlmglue.h

@@ -172,6 +172,110 @@ struct ocfs2_super
 /* For s_mount_opt */
 #define OCFS2_MOUNT_NOINTR (1 << 2)
 
+/*
+ * Return value from ->downconvert_worker functions.
+ *
+ * These control the precise actions of ocfs2_unblock_lock()
+ * and ocfs2_process_blocked_lock()
+ *
+ */
+enum ocfs2_unblock_action {
+	UNBLOCK_CONTINUE	= 0, /* Continue downconvert */
+	UNBLOCK_CONTINUE_POST	= 1, /* Continue downconvert, fire
+				      * ->post_unlock callback */
+	UNBLOCK_STOP_POST	= 2, /* Do not downconvert, fire
+				      * ->post_unlock() callback. */
+};
+
+/*
+ * OCFS2 Lock Resource Operations
+ *
+ * These fine tune the behavior of the generic dlmglue locking infrastructure.
+ *
+ * The most basic of lock types can point ->l_priv to their respective
+ * struct ocfs2_super and allow the default actions to manage things.
+ *
+ * Right now, each lock type also needs to implement an init function,
+ * and trivial lock/unlock wrappers. ocfs2_simple_drop_lockres()
+ * should be called when the lock is no longer needed (i.e., object
+ * destruction time).
+ */
+struct ocfs2_lock_res_ops {
+	/*
+	 * Translate an ocfs2_lock_res * into an ocfs2_super *. Define
+	 * this callback if ->l_priv is not an ocfs2_super pointer
+	 */
+	struct ocfs2_super * (*get_osb)(struct ocfs2_lock_res *);
+
+	/*
+	 * Optionally called in the downconvert thread after a
+	 * successful downconvert. The lockres will not be referenced
+	 * after this callback is called, so it is safe to free
+	 * memory, etc.
+	 *
+	 * The exact semantics of when this is called are controlled
+	 * by ->downconvert_worker()
+	 */
+	void (*post_unlock)(struct ocfs2_super *, struct ocfs2_lock_res *);
+
+	/*
+	 * Allow a lock type to add checks to determine whether it is
+	 * safe to downconvert a lock. Return 0 to re-queue the
+	 * downconvert at a later time, nonzero to continue.
+	 *
+	 * For most locks, the default checks that there are no
+	 * incompatible holders are sufficient.
+	 *
+	 * Called with the lockres spinlock held.
+	 */
+	int (*check_downconvert)(struct ocfs2_lock_res *, int);
+
+	/*
+	 * Allows a lock type to populate the lock value block. This
+	 * is called on downconvert, and when we drop a lock.
+	 *
+	 * Locks that want to use this should set LOCK_TYPE_USES_LVB
+	 * in the flags field.
+	 *
+	 * Called with the lockres spinlock held.
+	 */
+	void (*set_lvb)(struct ocfs2_lock_res *);
+
+	/*
+	 * Called from the downconvert thread when it is determined
+	 * that a lock will be downconverted. This is called without
+	 * any locks held so the function can do work that might
+	 * schedule (syncing out data, etc).
+	 *
+	 * This should return any one of the ocfs2_unblock_action
+	 * values, depending on what it wants the thread to do.
+	 */
+	int (*downconvert_worker)(struct ocfs2_lock_res *, int);
+
+	/*
+	 * LOCK_TYPE_* flags which describe the specific requirements
+	 * of a lock type. Descriptions of each individual flag follow.
+	 */
+	int flags;
+};
+
+/*
+ * Some locks want to "refresh" potentially stale data when a
+ * meaningful (PRMODE or EXMODE) lock level is first obtained. If this
+ * flag is set, the OCFS2_LOCK_NEEDS_REFRESH flag will be set on the
+ * individual lockres l_flags member from the ast function. It is
+ * expected that the locking wrapper will clear the
+ * OCFS2_LOCK_NEEDS_REFRESH flag when done.
+ */
+#define LOCK_TYPE_REQUIRES_REFRESH 0x1
+
+/*
+ * Indicate that a lock type makes use of the lock value block. The
+ * ->set_lvb lock type callback must be defined.
+ */
+#define LOCK_TYPE_USES_LVB		0x2
+
+
 #if 0
 #include "dcache.h"
 
@@ -253,6 +357,10 @@ int ocfs2_dlm_init(struct ocfs2_super *osb, char *cluster_stack,
 		   char *cluster_name, char *ls_name, struct dentry *debug_root);
 void ocfs2_dlm_shutdown(struct ocfs2_super *osb, int hangup_pending);
 void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res);
+void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
+				struct ocfs2_lock_res *res,
+				struct ocfs2_lock_res_ops *ops,
+				void *priv);
 #if 0
 void ocfs2_inode_lock_res_init(struct ocfs2_lock_res *res,
 			       enum ocfs2_lock_type type,

kmod/src/inode.c

@@ -284,7 +284,7 @@ static int scoutfs_getattr(struct vfsmount *mnt, struct dentry *dentry,
 	if (ret == 0)
 		generic_fillattr(inode, stat);
 
-	scoutfs_unlock(sb, lock);
+	scoutfs_unlock(sb, lock, DLM_LOCK_PR);
 	return ret;
 }
 
@@ -416,7 +416,7 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino)
 	}
 
 out:
-	scoutfs_unlock(sb, lock);
+	scoutfs_unlock(sb, lock, DLM_LOCK_PR);
 	return inode;
 }
 

kmod/src/ioctl.c

@@ -132,7 +132,7 @@ static long scoutfs_ioc_walk_inodes(struct file *file, unsigned long arg)
 
 		if (ret == -ENOENT) {
 
-			scoutfs_unlock(sb, lock);
+			scoutfs_unlock(sb, lock, DLM_LOCK_PR);
 			/*
 			 * XXX This will miss dirty items.  We'd need to
 			 * force writeouts of dirty items in our
@@ -181,7 +181,7 @@ static long scoutfs_ioc_walk_inodes(struct file *file, unsigned long arg)
 		scoutfs_key_inc_cur_len(&key);
 	}
 
-	scoutfs_unlock(sb, lock);
+	scoutfs_unlock(sb, lock, DLM_LOCK_PR);
 
 out:
 	scoutfs_key_free(sb, next_key);

kmod/src/lock.c

@@ -83,7 +83,12 @@ static int invalidate_caches(struct super_block *sb, int mode,
 
 static void free_scoutfs_lock(struct scoutfs_lock *lock)
 {
+	struct lock_info *linfo;
+
 	if (lock) {
+		linfo = SCOUTFS_SB(lock->sb)->lock_info;
+
+		ocfs2_simple_drop_lockres(&linfo->dlmglue, &lock->lockres);
 		scoutfs_key_free(lock->sb, lock->start);
 		scoutfs_key_free(lock->sb, lock->end);
 		kfree(lock);
@@ -113,12 +118,50 @@ static void put_scoutfs_lock(struct super_block *sb, struct scoutfs_lock *lock)
 	}
 }
 
+static struct ocfs2_super *get_ino_lock_osb(struct ocfs2_lock_res *lockres)
+{
+	struct scoutfs_lock *lock = lockres->l_priv;
+	struct super_block *sb = lock->sb;
+	DECLARE_LOCK_INFO(sb, linfo);
+
+	return &linfo->dlmglue;
+}
+
+static int ino_lock_downconvert(struct ocfs2_lock_res *lockres, int blocking)
+{
+	struct scoutfs_lock *lock = lockres->l_priv;
+	struct super_block *sb = lock->sb;
+
+	invalidate_caches(sb, blocking, lock->start, lock->end);
+
+	return UNBLOCK_CONTINUE;
+}
+
+static struct ocfs2_lock_res_ops scoufs_ino_lops = {
+	.get_osb 		= get_ino_lock_osb,
+	.downconvert_worker 	= ino_lock_downconvert,
+	/* XXX: .post_unlock for lru */
+	/* XXX: .check_downconvert that queries the item cache for dirty items */
+	.flags			= LOCK_TYPE_REQUIRES_REFRESH,
+};
+
+static struct ocfs2_lock_res_ops scoufs_ino_index_lops = {
+	.get_osb 		= get_ino_lock_osb,
+	.downconvert_worker 	= ino_lock_downconvert,
+	/* XXX: .post_unlock for lru */
+	/* XXX: .check_downconvert that queries the item cache for dirty items */
+	.flags			= 0,
+};
+
 static struct scoutfs_lock *alloc_scoutfs_lock(struct super_block *sb,
 					       struct scoutfs_lock_name *lock_name,
+					       struct ocfs2_lock_res_ops *type,
 					       struct scoutfs_key_buf *start,
 					       struct scoutfs_key_buf *end)
 
 {
+	DECLARE_LOCK_INFO(sb, linfo);
+//	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_lock *lock;
 
 	lock = kzalloc(sizeof(struct scoutfs_lock), GFP_NOFS);
@@ -135,6 +178,14 @@ static struct scoutfs_lock *alloc_scoutfs_lock(struct super_block *sb,
 			lock->mode = DLM_LOCK_IV;
 			INIT_WORK(&lock->dc_work, scoutfs_downconvert_func);
 			INIT_LIST_HEAD(&lock->lru_entry);
+			ocfs2_lock_res_init_once(&lock->lockres);
+			BUG_ON(sizeof(struct scoutfs_lock_name) >=
+			       OCFS2_LOCK_ID_MAX_LEN);
+			/* kzalloc above ensures that l_name is NULL terminated */
+			memcpy(&lock->lockres.l_name[0], &lock->lock_name,
+			       sizeof(struct scoutfs_lock_name));
+			ocfs2_lock_res_init_common(&linfo->dlmglue,
+						   &lock->lockres, type, lock);
 		}
 	}
 
@@ -152,6 +203,7 @@ static int cmp_lock_names(struct scoutfs_lock_name *a,
 
 static struct scoutfs_lock *find_alloc_scoutfs_lock(struct super_block *sb,
 					struct scoutfs_lock_name *lock_name,
+					struct ocfs2_lock_res_ops *type,
 					struct scoutfs_key_buf *start,
 					struct scoutfs_key_buf *end)
 {
@@ -187,7 +239,8 @@ search:
 	if (!found) {
 		if (!new) {
 			spin_unlock(&linfo->lock);
-			new = alloc_scoutfs_lock(sb, lock_name, start, end);
+			new = alloc_scoutfs_lock(sb, lock_name, type, start,
+						 end);
 			if (!new)
 				return NULL;
 
@@ -349,6 +402,7 @@ static int lock_blocking(struct lock_info *linfo, struct scoutfs_lock *lock)
  */
 static int lock_name_keys(struct super_block *sb, int mode,
 			 struct scoutfs_lock_name *lock_name,
+			 struct ocfs2_lock_res_ops *type,
 			 struct scoutfs_key_buf *start,
 			 struct scoutfs_key_buf *end,
 			 struct scoutfs_lock **ret_lock)
@@ -357,12 +411,20 @@ static int lock_name_keys(struct super_block *sb, int mode,
 	struct scoutfs_lock *lock;
 	int ret;
 
-	lock = find_alloc_scoutfs_lock(sb, lock_name, start, end);
+	lock = find_alloc_scoutfs_lock(sb, lock_name, type, start, end);
 	if (!lock)
 		return -ENOMEM;
 
 	trace_scoutfs_lock_resource(sb, lock);
 
+	ret = ocfs2_cluster_lock(&linfo->dlmglue, &lock->lockres, mode,
+				 DLM_LKF_NOORDER, 0);
+	if (ret)
+		return ret;
+
+	*ret_lock = lock;
+	return 0;
+#if 0
 check_lock_state:
 	spin_lock(&linfo->lock);
 	if (linfo->shutdown) {
@@ -413,6 +475,7 @@ check_lock_state:
 out:
 	*ret_lock = lock;
 	return 0;
+#endif
 }
 
 int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino,
@@ -441,7 +504,8 @@ int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino,
 	end_ikey.type = ~0;
 	scoutfs_key_init(&end, &end_ikey, sizeof(end_ikey));
 
-	return lock_name_keys(sb, mode, &lock_name, &start, &end, ret_lock);
+	return lock_name_keys(sb, mode, &lock_name, &scoufs_ino_lops, &start,
+			      &end, ret_lock);
 }
 
 /*
@@ -505,10 +569,12 @@ int scoutfs_lock_inode_index(struct super_block *sb, int mode,
 	end_ikey.ino = cpu_to_be64(ino | ino_mask);
 	scoutfs_key_init(&end, &end_ikey, sizeof(end_ikey));
 
-	return lock_name_keys(sb, mode, &lock_name, &start, &end, ret_lock);
+	return lock_name_keys(sb, mode, &lock_name, &scoufs_ino_index_lops,
+			      &start, &end, ret_lock);
 }
 
-void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock)
+void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
+		    int level)
 {
 	DECLARE_LOCK_INFO(sb, linfo);
 
@@ -517,12 +583,15 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock)
 
 	trace_scoutfs_unlock(sb, lock);
 
+	ocfs2_cluster_unlock(&linfo->dlmglue, &lock->lockres, level);
+
+#if 0
 	spin_lock(&linfo->lock);
 	lock->holders--;
 	if (lock->holders == 0 && (lock->flags & SCOUTFS_LOCK_BLOCKING))
 		queue_blocking_work(linfo, lock);
 	spin_unlock(&linfo->lock);
-
+#endif
 	put_scoutfs_lock(sb, lock);
 }
 
@@ -666,6 +735,8 @@ void scoutfs_lock_destroy(struct super_block *sb)
 	DECLARE_LOCK_INFO(sb, linfo);
 
 	if (linfo) {
+		free_lock_tree(sb); /* Do this before uninitializing the dlm. */
+
 		if (linfo->downconvert_wq)
 			destroy_workqueue(linfo->downconvert_wq);
 		unregister_shrinker(&linfo->shrinker);
@@ -674,8 +745,6 @@ void scoutfs_lock_destroy(struct super_block *sb)
 			ocfs2_uninit_super(&linfo->dlmglue);
 		}
 
-		free_lock_tree(sb);
-
 		sbi->lock_info = NULL;
 
 		trace_printk("sb %p id %016llx freeing linfo %p linfo %p\n",

kmod/src/lock.h

@@ -3,6 +3,7 @@
 
 #include <linux/dlm.h>
 #include "key.h"
+#include "dlmglue.h"
 
 #define	SCOUTFS_LOCK_BLOCKING	0x01 /* Blocking another lock request */
 #define	SCOUTFS_LOCK_QUEUED	0x02 /* Put on drop workqueue */
@@ -22,6 +23,7 @@ struct scoutfs_lock {
 	unsigned int holders; /* Tracks active users of this lock */
 	unsigned int flags;
 	struct work_struct dc_work;
+	struct ocfs2_lock_res lockres;
 };
 
 int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino,
@@ -29,7 +31,8 @@ int scoutfs_lock_ino_group(struct super_block *sb, int mode, u64 ino,
 int scoutfs_lock_inode_index(struct super_block *sb, int mode,
 			     u8 type, u64 major, u64 ino,
 			     struct scoutfs_lock **ret_lock);
-void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock);
+void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
+		    int level);
 
 int scoutfs_lock_setup(struct super_block *sb);
 void scoutfs_lock_shutdown(struct super_block *sb);

kmod/src/xattr.c

@@ -229,7 +229,7 @@ ssize_t scoutfs_getxattr(struct dentry *dentry, const char *name, void *buffer,
 		ret = -ERANGE;
 
 	up_read(&si->xattr_rwsem);
-	scoutfs_unlock(sb, lck);
+	scoutfs_unlock(sb, lck, DLM_LOCK_PR);
 
 out:
 	scoutfs_key_free(sb, key);
@@ -336,7 +336,7 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	scoutfs_release_trans(sb);
 
 unlock:
-	scoutfs_unlock(sb, lck);
+	scoutfs_unlock(sb, lck, DLM_LOCK_EX);
 
 out:
 	scoutfs_item_free_batch(sb, &list);
@@ -436,7 +436,7 @@ ssize_t scoutfs_listxattr(struct dentry *dentry, char *buffer, size_t size)
 	}
 
 	up_read(&si->xattr_rwsem);
-	scoutfs_unlock(sb, lck);
+	scoutfs_unlock(sb, lck, DLM_LOCK_PR);
 out:
 	scoutfs_key_free(sb, key);
 	scoutfs_key_free(sb, last);
@@ -490,7 +490,7 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino)
 		/* don't need to increment past deleted key */
 	}
 
-	scoutfs_unlock(sb, lck);
+	scoutfs_unlock(sb, lck, DLM_LOCK_EX);
 
 out:
 	scoutfs_key_free(sb, key);