ocfs2: Break out stackglue into modules.

We define the ocfs2_stack_plugin structure to represent a stack driver.
The o2cb stack code is split into stack_o2cb.c.  This becomes the
ocfs2_stack_o2cb.ko module.

The stackglue generic functions are similarly split into the
ocfs2_stackglue.ko module.  This module now provides an interface to
register drivers.  The ocfs2_stack_o2cb driver registers itself.  As
part of this interface, ocfs2_stackglue can load drivers on demand.
This is accomplished in ocfs2_cluster_connect().

ocfs2_cluster_disconnect() is now notified when a _hangup() is pending.
If a hangup is pending, it will not release the driver module and will
let _hangup() do that.

Signed-off-by: Joel Becker <joel.becker@oracle.com>

1 files changed, 214 insertions, 24 deletions

diff --git a/fs/ocfs2/stackglue.c b/fs/ocfs2/stackglue.c
index e197367b6bd6..1978c9cff0e9 100644
--- a/fs/ocfs2/stackglue.c
+++ b/fs/ocfs2/stackglue.c
@@ -18,17 +18,176 @@
  * General Public License for more details.
  */
 
+#include <linux/list.h>
+#include <linux/spinlock.h>
+#include <linux/module.h>
 #include <linux/slab.h>
 #include <linux/kmod.h>
 
-/* Needed for AOP_TRUNCATED_PAGE in mlog_errno() */
-#include <linux/fs.h>
+#include "stackglue.h"
 
-#include "cluster/masklog.h"
+static struct ocfs2_locking_protocol *lproto;
+static DEFINE_SPINLOCK(ocfs2_stack_lock);
+static LIST_HEAD(ocfs2_stack_list);
 
-#include "stackglue.h"
+/*
+ * The stack currently in use.  If not null, active_stack->sp_count > 0,
+ * the module is pinned, and the locking protocol cannot be changed.
+ */
+static struct ocfs2_stack_plugin *active_stack;
+
+static struct ocfs2_stack_plugin *ocfs2_stack_lookup(const char *name)
+{
+	struct ocfs2_stack_plugin *p;
+
+	assert_spin_locked(&ocfs2_stack_lock);
+
+	list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
+		if (!strcmp(p->sp_name, name))
+			return p;
+	}
+
+	return NULL;
+}
+
+static int ocfs2_stack_driver_request(const char *name)
+{
+	int rc;
+	struct ocfs2_stack_plugin *p;
+
+	spin_lock(&ocfs2_stack_lock);
+
+	if (active_stack) {
+		/*
+		 * If the active stack isn't the one we want, it cannot
+		 * be selected right now.
+		 */
+		if (!strcmp(active_stack->sp_name, name))
+			rc = 0;
+		else
+			rc = -EBUSY;
+		goto out;
+	}
+
+	p = ocfs2_stack_lookup(name);
+	if (!p || !try_module_get(p->sp_owner)) {
+		rc = -ENOENT;
+		goto out;
+	}
+
+	/* Ok, the stack is pinned */
+	p->sp_count++;
+	active_stack = p;
+
+	rc = 0;
+
+out:
+	spin_unlock(&ocfs2_stack_lock);
+	return rc;
+}
+
+/*
+ * This function looks up the appropriate stack and makes it active.  If
+ * there is no stack, it tries to load it.  It will fail if the stack still
+ * cannot be found.  It will also fail if a different stack is in use.
+ */
+static int ocfs2_stack_driver_get(const char *name)
+{
+	int rc;
+
+	rc = ocfs2_stack_driver_request(name);
+	if (rc == -ENOENT) {
+		request_module("ocfs2_stack_%s", name);
+		rc = ocfs2_stack_driver_request(name);
+	}
+
+	if (rc == -ENOENT) {
+		printk(KERN_ERR
+		       "ocfs2: Cluster stack driver \"%s\" cannot be found\n",
+		       name);
+	} else if (rc == -EBUSY) {
+		printk(KERN_ERR
+		       "ocfs2: A different cluster stack driver is in use\n");
+	}
+
+	return rc;
+}
 
-struct ocfs2_locking_protocol *stack_glue_lproto;
+static void ocfs2_stack_driver_put(void)
+{
+	spin_lock(&ocfs2_stack_lock);
+	BUG_ON(active_stack == NULL);
+	BUG_ON(active_stack->sp_count == 0);
+
+	active_stack->sp_count--;
+	if (!active_stack->sp_count) {
+		module_put(active_stack->sp_owner);
+		active_stack = NULL;
+	}
+	spin_unlock(&ocfs2_stack_lock);
+}
+
+int ocfs2_stack_glue_register(struct ocfs2_stack_plugin *plugin)
+{
+	int rc;
+
+	spin_lock(&ocfs2_stack_lock);
+	if (!ocfs2_stack_lookup(plugin->sp_name)) {
+		plugin->sp_count = 0;
+		plugin->sp_proto = lproto;
+		list_add(&plugin->sp_list, &ocfs2_stack_list);
+		printk(KERN_INFO "ocfs2: Registered cluster interface %s\n",
+		       plugin->sp_name);
+		rc = 0;
+	} else {
+		printk(KERN_ERR "ocfs2: Stack \"%s\" already registered\n",
+		       plugin->sp_name);
+		rc = -EEXIST;
+	}
+	spin_unlock(&ocfs2_stack_lock);
+
+	return rc;
+}
+EXPORT_SYMBOL_GPL(ocfs2_stack_glue_register);
+
+void ocfs2_stack_glue_unregister(struct ocfs2_stack_plugin *plugin)
+{
+	struct ocfs2_stack_plugin *p;
+
+	spin_lock(&ocfs2_stack_lock);
+	p = ocfs2_stack_lookup(plugin->sp_name);
+	if (p) {
+		BUG_ON(p != plugin);
+		BUG_ON(plugin == active_stack);
+		BUG_ON(plugin->sp_count != 0);
+		list_del_init(&plugin->sp_list);
+		printk(KERN_INFO "ocfs2: Unregistered cluster interface %s\n",
+		       plugin->sp_name);
+	} else {
+		printk(KERN_ERR "Stack \"%s\" is not registered\n",
+		       plugin->sp_name);
+	}
+	spin_unlock(&ocfs2_stack_lock);
+}
+EXPORT_SYMBOL_GPL(ocfs2_stack_glue_unregister);
+
+void ocfs2_stack_glue_set_locking_protocol(struct ocfs2_locking_protocol *proto)
+{
+	struct ocfs2_stack_plugin *p;
+
+	BUG_ON(proto == NULL);
+
+	spin_lock(&ocfs2_stack_lock);
+	BUG_ON(active_stack != NULL);
+
+	lproto = proto;
+	list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
+		p->sp_proto = lproto;
+	}
+
+	spin_unlock(&ocfs2_stack_lock);
+}
+EXPORT_SYMBOL_GPL(ocfs2_stack_glue_set_locking_protocol);
 
 
 int ocfs2_dlm_lock(struct ocfs2_cluster_connection *conn,
@@ -39,26 +198,29 @@ int ocfs2_dlm_lock(struct ocfs2_cluster_connection *conn,
 		   unsigned int namelen,
 		   void *astarg)
 {
-	BUG_ON(stack_glue_lproto == NULL);
+	BUG_ON(lproto == NULL);
 
-	return o2cb_stack_ops.dlm_lock(conn, mode, lksb, flags,
-				       name, namelen, astarg);
+	return active_stack->sp_ops->dlm_lock(conn, mode, lksb, flags,
+					      name, namelen, astarg);
 }
+EXPORT_SYMBOL_GPL(ocfs2_dlm_lock);
 
 int ocfs2_dlm_unlock(struct ocfs2_cluster_connection *conn,
 		     union ocfs2_dlm_lksb *lksb,
 		     u32 flags,
 		     void *astarg)
 {
-	BUG_ON(stack_glue_lproto == NULL);
+	BUG_ON(lproto == NULL);
 
-	return o2cb_stack_ops.dlm_unlock(conn, lksb, flags, astarg);
+	return active_stack->sp_ops->dlm_unlock(conn, lksb, flags, astarg);
 }
+EXPORT_SYMBOL_GPL(ocfs2_dlm_unlock);
 
 int ocfs2_dlm_lock_status(union ocfs2_dlm_lksb *lksb)
 {
-	return o2cb_stack_ops.lock_status(lksb);
+	return active_stack->sp_ops->lock_status(lksb);
 }
+EXPORT_SYMBOL_GPL(ocfs2_dlm_lock_status);
 
 /*
  * Why don't we cast to ocfs2_meta_lvb?  The "clean" answer is that we
@@ -67,13 +229,15 @@ int ocfs2_dlm_lock_status(union ocfs2_dlm_lksb *lksb)
  */
 void *ocfs2_dlm_lvb(union ocfs2_dlm_lksb *lksb)
 {
-	return o2cb_stack_ops.lock_lvb(lksb);
+	return active_stack->sp_ops->lock_lvb(lksb);
 }
+EXPORT_SYMBOL_GPL(ocfs2_dlm_lvb);
 
 void ocfs2_dlm_dump_lksb(union ocfs2_dlm_lksb *lksb)
 {
-	o2cb_stack_ops.dump_lksb(lksb);
+	active_stack->sp_ops->dump_lksb(lksb);
 }
+EXPORT_SYMBOL_GPL(ocfs2_dlm_dump_lksb);
 
 int ocfs2_cluster_connect(const char *group,
 			  int grouplen,
@@ -107,11 +271,16 @@ int ocfs2_cluster_connect(const char *group,
 	new_conn->cc_recovery_data = recovery_data;
 
 	/* Start the new connection at our maximum compatibility level */
-	new_conn->cc_version = stack_glue_lproto->lp_max_version;
+	new_conn->cc_version = lproto->lp_max_version;
+
+	/* This will pin the stack driver if successful */
+	rc = ocfs2_stack_driver_get("o2cb");
+	if (rc)
+		goto out_free;
 
-	rc = o2cb_stack_ops.connect(new_conn);
+	rc = active_stack->sp_ops->connect(new_conn);
 	if (rc) {
-		mlog_errno(rc);
+		ocfs2_stack_driver_put();
 		goto out_free;
 	}
 
@@ -124,39 +293,60 @@ out_free:
 out:
 	return rc;
 }
+EXPORT_SYMBOL_GPL(ocfs2_cluster_connect);
 
-int ocfs2_cluster_disconnect(struct ocfs2_cluster_connection *conn)
+/* If hangup_pending is 0, the stack driver will be dropped */
+int ocfs2_cluster_disconnect(struct ocfs2_cluster_connection *conn,
+			     int hangup_pending)
 {
 	int ret;
 
 	BUG_ON(conn == NULL);
 
-	ret = o2cb_stack_ops.disconnect(conn);
+	ret = active_stack->sp_ops->disconnect(conn, hangup_pending);
 
 	/* XXX Should we free it anyway? */
-	if (!ret)
+	if (!ret) {
 		kfree(conn);
+		if (!hangup_pending)
+			ocfs2_stack_driver_put();
+	}
 
 	return ret;
 }
+EXPORT_SYMBOL_GPL(ocfs2_cluster_disconnect);
 
 void ocfs2_cluster_hangup(const char *group, int grouplen)
 {
 	BUG_ON(group == NULL);
 	BUG_ON(group[grouplen] != '\0');
 
-	o2cb_stack_ops.hangup(group, grouplen);
+	active_stack->sp_ops->hangup(group, grouplen);
+
+	/* cluster_disconnect() was called with hangup_pending==1 */
+	ocfs2_stack_driver_put();
 }
+EXPORT_SYMBOL_GPL(ocfs2_cluster_hangup);
 
 int ocfs2_cluster_this_node(unsigned int *node)
 {
-	return o2cb_stack_ops.this_node(node);
+	return active_stack->sp_ops->this_node(node);
 }
+EXPORT_SYMBOL_GPL(ocfs2_cluster_this_node);
 
-void ocfs2_stack_glue_set_locking_protocol(struct ocfs2_locking_protocol *proto)
+
+static int __init ocfs2_stack_glue_init(void)
 {
-	BUG_ON(proto != NULL);
+	return 0;
+}
 
-	stack_glue_lproto = proto;
+static void __exit ocfs2_stack_glue_exit(void)
+{
+	lproto = NULL;
 }
 
+MODULE_AUTHOR("Oracle");
+MODULE_DESCRIPTION("ocfs2 cluter stack glue layer");
+MODULE_LICENSE("GPL");
+module_init(ocfs2_stack_glue_init);
+module_exit(ocfs2_stack_glue_exit);