scoutfs: use dlmglue for lockspace bringup/shutdown

Ultimataly the direct dlm lock calls will go away. For now though we
grab the lockspace off our cluster connection object. In order to get
this going, I stubbed out our recovery callbacks which now gets us a
print when a node goes down.

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

kmod/src/lock.c

@@ -22,18 +22,22 @@
 #include "scoutfs_trace.h"
 #include "msg.h"
 #include "cmp.h"
+#include "dlmglue.h"
 
 #define LN_FMT "%u.%u.%llu.%llu"
 #define LN_ARG(name) \
 	(name)->zone, (name)->type, le64_to_cpu((name)->first), \
 	le64_to_cpu((name)->second)
 
+typedef struct ocfs2_super dlmglue_ctxt;
+
 /*
  * allocated per-super, freed on unmount.
  */
 struct lock_info {
 	struct super_block *sb;
-	dlm_lockspace_t *ls;
+	dlmglue_ctxt dlmglue;
+	bool dlmglue_online;
 	char ls_name[DLM_LOCKSPACE_LEN];
 	bool shutdown;
 	struct list_head id_head;
@@ -394,8 +398,9 @@ check_lock_state:
 	lock->holders++;
 	spin_unlock(&linfo->lock);
 
-	ret = dlm_lock(linfo->ls, mode, &lock->lksb, DLM_LKF_NOORDER,
-		       &lock->lock_name, sizeof(struct scoutfs_lock_name),
+	ret = dlm_lock(linfo->dlmglue.cconn->cc_lockspace, mode, &lock->lksb,
+		       DLM_LKF_NOORDER, &lock->lock_name,
+		       sizeof(struct scoutfs_lock_name),
 		       0, scoutfs_ast, lock, scoutfs_bast);
 	if (ret) {
 		scoutfs_err(sb, "Error %d locking "LN_FMT, ret,
@@ -533,7 +538,8 @@ static void unlock_range(struct super_block *sb, struct scoutfs_lock *lock)
 	spin_lock(&linfo->lock);
 	lock->rqmode = DLM_LOCK_IV;
 	spin_unlock(&linfo->lock);
-	ret = dlm_unlock(linfo->ls, lock->lksb.sb_lkid, 0, &lock->lksb, lock);
+	ret = dlm_unlock(linfo->dlmglue.cconn->cc_lockspace, lock->lksb.sb_lkid,
+			 0, &lock->lksb, lock);
 	if (ret) {
 		scoutfs_err(sb, "Error %d unlocking "LN_FMT, ret,
 			    LN_ARG(&lock->lock_name));
@@ -601,11 +607,16 @@ static int init_lock_info(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct lock_info *linfo;
+	int ret;
 
 	linfo = kzalloc(sizeof(struct lock_info), GFP_KERNEL);
 	if (!linfo)
 		return -ENOMEM;
 
+	ret = ocfs2_init_super(&linfo->dlmglue, 0);
+	if (ret)
+		goto out;
+
 	spin_lock_init(&linfo->lock);
 	init_waitqueue_head(&linfo->waitq);
 	INIT_LIST_HEAD(&linfo->lru_list);
@@ -615,7 +626,6 @@ static int init_lock_info(struct super_block *sb)
 	linfo->sb = sb;
 	linfo->shutdown = false;
 	INIT_LIST_HEAD(&linfo->id_head);
-	linfo->ls = NULL;
 
 	snprintf(linfo->ls_name, DLM_LOCKSPACE_LEN, "%llx",
 		 le64_to_cpu(sbi->super.hdr.fsid));
@@ -624,6 +634,9 @@ static int init_lock_info(struct super_block *sb)
 
 	trace_printk("sb %p id %016llx allocated linfo %p held %p\n",
 		     sb, le64_to_cpu(sbi->super.id), linfo, linfo);
+out:
+	if (ret)
+		kfree(linfo);
 
 	return 0;
 }
@@ -651,17 +664,14 @@ void scoutfs_lock_destroy(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	DECLARE_LOCK_INFO(sb, linfo);
-	int ret;
 
 	if (linfo) {
 		if (linfo->downconvert_wq)
 			destroy_workqueue(linfo->downconvert_wq);
 		unregister_shrinker(&linfo->shrinker);
-		if (linfo->ls) {
-			ret = dlm_release_lockspace(linfo->ls, 2);
-			if (ret)
-				scoutfs_info(sb, "Error %d releasing lockspace %s\n",
-					     ret, linfo->ls_name);
+		if (linfo->dlmglue_online) {
+			ocfs2_dlm_shutdown(&linfo->dlmglue, 0);
+			ocfs2_uninit_super(&linfo->dlmglue);
 		}
 
 		free_lock_tree(sb);
@@ -684,23 +694,22 @@ int scoutfs_lock_setup(struct super_block *sb)
 	ret = init_lock_info(sb);
 	if (ret)
 		return ret;
-
 	linfo = sbi->lock_info;
+
 	linfo->downconvert_wq = alloc_workqueue("scoutfs_dc",
 					       WQ_UNBOUND|WQ_HIGHPRI, 0);
 	if (!linfo->downconvert_wq) {
-		kfree(linfo);
-		return -ENOMEM;
+		ret = -ENOMEM;
+		goto out;
 	}
 
-	/*
-	 * Open coded '64' here is for lvb_len. We never use the LVB
-	 * flag so this doesn't matter, but the dlm needs a non-zero
-	 * multiple of 8
-	 */
-	ret = dlm_new_lockspace(linfo->ls_name, sbi->opts.cluster_name,
-				DLM_LSFL_FS|DLM_LSFL_NEWEXCL, 64, NULL,
-				NULL, NULL, &linfo->ls);
+	ret = ocfs2_dlm_init(&linfo->dlmglue, "null", sbi->opts.cluster_name,
+			     linfo->ls_name, sbi->debug_root);
+	if (ret)
+		goto out;
+	linfo->dlmglue_online = true;
+
+out:
 	if (ret)
 		scoutfs_lock_destroy(sb);
 

kmod/src/stackglue.c

@@ -194,7 +194,29 @@ int ocfs2_plock(struct ocfs2_cluster_connection *conn, u64 ino,
 	return user_plock(conn, ino, file, cmd, fl);
 }
 
-static struct dlm_lockspace_ops *ocfs2_ls_ops = NULL;
+static void user_recover_prep(void *arg)
+{
+	/* XXX: Set FS in recovery here */
+}
+
+static void user_recover_slot(void *arg, struct dlm_slot *slot)
+{
+	printk(KERN_INFO "scoutfs: Node %d/%d down. Initiating recovery.\n",
+			slot->nodeid, slot->slot);
+}
+
+static void user_recover_done(void *arg, struct dlm_slot *slots,
+		int num_slots, int our_slot,
+		uint32_t generation)
+{
+	/* XXX: Do actual fs recovery here */
+}
+
+static const struct dlm_lockspace_ops ocfs2_ls_ops = {
+	.recover_prep = user_recover_prep,
+	.recover_slot = user_recover_slot,
+	.recover_done = user_recover_done,
+};
 
 static int user_cluster_connect(struct ocfs2_cluster_connection *conn)
 {
@@ -218,7 +240,7 @@ static int user_cluster_connect(struct ocfs2_cluster_connection *conn)
 
 	rc = dlm_new_lockspace(conn->cc_name, conn->cc_cluster_name,
 			       DLM_LSFL_FS | DLM_LSFL_NEWEXCL, DLM_LVB_LEN,
-			       ocfs2_ls_ops, conn, &ops_rv, &fsdlm);
+			       &ocfs2_ls_ops, conn, &ops_rv, &fsdlm);
 	if (rc) {
 		if (rc == -EEXIST || rc == -EPROTO)
 			printk(KERN_ERR "scoutfs: Unable to create the "

kmod/src/super.c

@@ -19,6 +19,7 @@
 #include <linux/random.h>
 #include <linux/statfs.h>
 #include <linux/sched.h>
+#include <linux/debugfs.h>
 
 #include "super.h"
 #include "format.h"
@@ -42,6 +43,7 @@
 #include "scoutfs_trace.h"
 
 static struct kset *scoutfs_kset;
+static struct dentry *scoutfs_debugfs_root;
 
 /*
  * Ask the server for the current statfs fields.  The message is very
@@ -202,6 +204,24 @@ int scoutfs_read_supers(struct super_block *sb,
 	return 0;
 }
 
+static int scoutfs_debugfs_setup(struct super_block *sb)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	char name[32];
+
+	/*
+	 * XXX: Move the name variable to sbi and use it in
+	 * init_lock_info as well.
+	 */
+	snprintf(name, 32, "%llx", le64_to_cpu(sbi->super.hdr.fsid));
+
+	sbi->debug_root = debugfs_create_dir(name, scoutfs_debugfs_root);
+	if (!sbi->debug_root)
+		return -ENOMEM;
+
+	return 0;
+}
+
 static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 {
 	struct scoutfs_sb_info *sbi;
@@ -247,6 +267,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 
 	ret = scoutfs_setup_counters(sb) ?:
 	      scoutfs_read_supers(sb, &SCOUTFS_SB(sb)->super) ?:
+	      scoutfs_debugfs_setup(sb) ?:
 	      scoutfs_seg_setup(sb) ?:
 	      scoutfs_item_setup(sb) ?:
 	      scoutfs_inode_setup(sb) ?:
@@ -330,6 +351,7 @@ static void scoutfs_kill_sb(struct super_block *sb)
 		scoutfs_inode_destroy(sb);
 		scoutfs_item_destroy(sb);
 		scoutfs_seg_destroy(sb);
+		debugfs_remove(sbi->debug_root);
 		scoutfs_destroy_counters(sb);
 		if (sbi->kset)
 			kset_unregister(sbi->kset);
@@ -349,6 +371,7 @@ MODULE_ALIAS_FS("scoutfs");
 /* safe to call at any failure point in _init */
 static void teardown_module(void)
 {
+	debugfs_remove(scoutfs_debugfs_root);
 	scoutfs_dir_exit();
 	scoutfs_inode_exit();
 	if (scoutfs_kset)
@@ -374,10 +397,16 @@ static int __init scoutfs_module_init(void)
 	if (!scoutfs_kset)
 		return -ENOMEM;
 
+	scoutfs_debugfs_root = debugfs_create_dir("scoutfs", NULL);
+	if (!scoutfs_debugfs_root) {
+		ret = -ENOMEM;
+		goto out;
+	}
 	ret = scoutfs_inode_init() ?:
 	      scoutfs_dir_init() ?:
 	      scoutfs_xattr_init() ?:
 	      register_filesystem(&scoutfs_fs_type);
+out:
 	if (ret)
 		teardown_module();
 	return ret;

kmod/src/super.h

@@ -61,6 +61,8 @@ struct scoutfs_sb_info {
 	struct scoutfs_counters *counters;
 
 	struct mount_options opts;
+
+	struct dentry *debug_root;
 };
 
 static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb)