1 files changed, 63 insertions, 35 deletions

diff --git a/fs/ocfs2/stackglue.c b/fs/ocfs2/stackglue.c
index f3df0baa9a48..39abf89697ed 100644
--- a/fs/ocfs2/stackglue.c
+++ b/fs/ocfs2/stackglue.c
@@ -36,7 +36,7 @@
 #define OCFS2_STACK_PLUGIN_USER         "user"
 #define OCFS2_MAX_HB_CTL_PATH           256
 
-static struct ocfs2_locking_protocol *lproto;
+static struct ocfs2_protocol_version locking_max_version;
 static DEFINE_SPINLOCK(ocfs2_stack_lock);
 static LIST_HEAD(ocfs2_stack_list);
 static char cluster_stack_name[OCFS2_STACK_LABEL_LEN + 1];
@@ -176,7 +176,7 @@ int ocfs2_stack_glue_register(struct ocfs2_stack_plugin *plugin)
         spin_lock(&ocfs2_stack_lock);
         if (!ocfs2_stack_lookup(plugin->sp_name)) {
                 plugin->sp_count = 0;
-                plugin->sp_proto = lproto;
+                plugin->sp_max_proto = locking_max_version;
                 list_add(&plugin->sp_list, &ocfs2_stack_list);
                 printk(KERN_INFO "ocfs2: Registered cluster interface %s\n",
                        plugin->sp_name);
@@ -213,77 +213,76 @@ void ocfs2_stack_glue_unregister(struct ocfs2_stack_plugin *plugin)
 }
 EXPORT_SYMBOL_GPL(ocfs2_stack_glue_unregister);
 
-void ocfs2_stack_glue_set_locking_protocol(struct ocfs2_locking_protocol *proto)
+void ocfs2_stack_glue_set_max_proto_version(struct ocfs2_protocol_version *max_proto)
 {
         struct ocfs2_stack_plugin *p;
 
-        BUG_ON(proto == NULL);
-
         spin_lock(&ocfs2_stack_lock);
-        BUG_ON(active_stack != NULL);
+        if (memcmp(max_proto, &locking_max_version,
+                   sizeof(struct ocfs2_protocol_version))) {
+                BUG_ON(locking_max_version.pv_major != 0);
 
-        lproto = proto;
-        list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
-                p->sp_proto = lproto;
+                locking_max_version = *max_proto;
+                list_for_each_entry(p, &ocfs2_stack_list, sp_list) {
+                        p->sp_max_proto = locking_max_version;
+                }
         }
-
         spin_unlock(&ocfs2_stack_lock);
 }
-EXPORT_SYMBOL_GPL(ocfs2_stack_glue_set_locking_protocol);
+EXPORT_SYMBOL_GPL(ocfs2_stack_glue_set_max_proto_version);
 
 
 /*
- * The ocfs2_dlm_lock() and ocfs2_dlm_unlock() functions take
- * "struct ocfs2_lock_res *astarg" instead of "void *astarg" because the
- * underlying stack plugins need to pilfer the lksb off of the lock_res.
- * If some other structure needs to be passed as an astarg, the plugins
- * will need to be given a different avenue to the lksb.
+ * The ocfs2_dlm_lock() and ocfs2_dlm_unlock() functions take no argument
+ * for the ast and bast functions.  They will pass the lksb to the ast
+ * and bast.  The caller can wrap the lksb with their own structure to
+ * get more information.
  */
 int ocfs2_dlm_lock(struct ocfs2_cluster_connection *conn,
                    int mode,
-                   union ocfs2_dlm_lksb *lksb,
+                   struct ocfs2_dlm_lksb *lksb,
                    u32 flags,
                    void *name,
-                   unsigned int namelen,
-                   struct ocfs2_lock_res *astarg)
+                   unsigned int namelen)
 {
-        BUG_ON(lproto == NULL);
-
+        if (!lksb->lksb_conn)
+                lksb->lksb_conn = conn;
+        else
+                BUG_ON(lksb->lksb_conn != conn);
         return active_stack->sp_ops->dlm_lock(conn, mode, lksb, flags,
-                                              name, namelen, astarg);
+                                              name, namelen);
 }
 EXPORT_SYMBOL_GPL(ocfs2_dlm_lock);
 
 int ocfs2_dlm_unlock(struct ocfs2_cluster_connection *conn,
-                     union ocfs2_dlm_lksb *lksb,
-                     u32 flags,
-                     struct ocfs2_lock_res *astarg)
+                     struct ocfs2_dlm_lksb *lksb,
+                     u32 flags)
 {
-        BUG_ON(lproto == NULL);
+        BUG_ON(lksb->lksb_conn == NULL);
 
-        return active_stack->sp_ops->dlm_unlock(conn, lksb, flags, astarg);
+        return active_stack->sp_ops->dlm_unlock(conn, lksb, flags);
 }
 EXPORT_SYMBOL_GPL(ocfs2_dlm_unlock);
 
-int ocfs2_dlm_lock_status(union ocfs2_dlm_lksb *lksb)
+int ocfs2_dlm_lock_status(struct ocfs2_dlm_lksb *lksb)
 {
         return active_stack->sp_ops->lock_status(lksb);
 }
 EXPORT_SYMBOL_GPL(ocfs2_dlm_lock_status);
 
-int ocfs2_dlm_lvb_valid(union ocfs2_dlm_lksb *lksb)
+int ocfs2_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb)
 {
         return active_stack->sp_ops->lvb_valid(lksb);
 }
 EXPORT_SYMBOL_GPL(ocfs2_dlm_lvb_valid);
 
-void *ocfs2_dlm_lvb(union ocfs2_dlm_lksb *lksb)
+void *ocfs2_dlm_lvb(struct ocfs2_dlm_lksb *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)
+void ocfs2_dlm_dump_lksb(struct ocfs2_dlm_lksb *lksb)
 {
         active_stack->sp_ops->dump_lksb(lksb);
 }
@@ -312,6 +311,7 @@ EXPORT_SYMBOL_GPL(ocfs2_plock);
 int ocfs2_cluster_connect(const char *stack_name,
                           const char *group,
                           int grouplen,
+                          struct ocfs2_locking_protocol *lproto,
                           void (*recovery_handler)(int node_num,
                                                    void *recovery_data),
                           void *recovery_data,
@@ -329,6 +329,12 @@ int ocfs2_cluster_connect(const char *stack_name,
                 goto out;
         }
 
+        if (memcmp(&lproto->lp_max_version, &locking_max_version,
+                   sizeof(struct ocfs2_protocol_version))) {
+                rc = -EINVAL;
+                goto out;
+        }
+
         new_conn = kzalloc(sizeof(struct ocfs2_cluster_connection),
                            GFP_KERNEL);
         if (!new_conn) {
@@ -341,6 +347,7 @@ int ocfs2_cluster_connect(const char *stack_name,
         new_conn->cc_recovery_handler = recovery_handler;
         new_conn->cc_recovery_data = recovery_data;
 
+        new_conn->cc_proto = lproto;
         /* Start the new connection at our maximum compatibility level */
         new_conn->cc_version = lproto->lp_max_version;
 
@@ -366,6 +373,24 @@ out:
 }
 EXPORT_SYMBOL_GPL(ocfs2_cluster_connect);
 
+/* The caller will ensure all nodes have the same cluster stack */
+int ocfs2_cluster_connect_agnostic(const char *group,
+                                   int grouplen,
+                                   struct ocfs2_locking_protocol *lproto,
+                                   void (*recovery_handler)(int node_num,
+                                                            void *recovery_data),
+                                   void *recovery_data,
+                                   struct ocfs2_cluster_connection **conn)
+{
+        char *stack_name = NULL;
+
+        if (cluster_stack_name[0])
+                stack_name = cluster_stack_name;
+        return ocfs2_cluster_connect(stack_name, group, grouplen, lproto,
+                                     recovery_handler, recovery_data, conn);
+}
+EXPORT_SYMBOL_GPL(ocfs2_cluster_connect_agnostic);
+
 /* If hangup_pending is 0, the stack driver will be dropped */
 int ocfs2_cluster_disconnect(struct ocfs2_cluster_connection *conn,
                              int hangup_pending)
@@ -453,10 +478,10 @@ static ssize_t ocfs2_max_locking_protocol_show(struct kobject *kobj,
         ssize_t ret = 0;
 
         spin_lock(&ocfs2_stack_lock);
-        if (lproto)
+        if (locking_max_version.pv_major)
                 ret = snprintf(buf, PAGE_SIZE, "%u.%u\n",
-                               lproto->lp_max_version.pv_major,
-                               lproto->lp_max_version.pv_minor);
+                               locking_max_version.pv_major,
+                               locking_max_version.pv_minor);
         spin_unlock(&ocfs2_stack_lock);
 
         return ret;
@@ -685,7 +710,10 @@ static int __init ocfs2_stack_glue_init(void)
 
 static void __exit ocfs2_stack_glue_exit(void)
 {
-        lproto = NULL;
+        memset(&locking_max_version, 0,
+               sizeof(struct ocfs2_protocol_version));
+        locking_max_version.pv_major = 0;
+        locking_max_version.pv_minor = 0;
         ocfs2_sysfs_exit();
         if (ocfs2_table_header)
                 unregister_sysctl_table(ocfs2_table_header);