ocfs2: Negotiate locking protocol versions.

Currently, when ocfs2 nodes connect via TCP, they advertise their compatibility level. If the versions do not match, two nodes cannot speak to each other and they disconnect. As a result, this provides no forward or backwards compatibility. This patch implements a simple protocol negotiation at the dlm level by introducing a major/minor version number scheme for entities that communicate. Specifically, o2dlm has a major/minor version for interaction with o2dlm on other nodes, and ocfs2 itself has a major/minor version for interacting with the filesystem on other nodes. This will allow rolling upgrades of ocfs2 clusters when changes to the locking or network protocols can be done in a backwards compatible manner. In those cases, only the minor number is changed and the negotatied protocol minor is returned from dlm join. In the far less likely event that a required protocol change makes backwards compatibility impossible, we simply bump the major number. Signed-off-by: Joel Becker <joel.becker@oracle.com> Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>

ocfs2: Negotiate locking protocol versions.
Currently, when ocfs2 nodes connect via TCP, they advertise their compatibility level. If the versions do not match, two nodes cannot speak to each other and they disconnect. As a result, this provides no forward or backwards compatibility. This patch implements a simple protocol negotiation at the dlm level by introducing a major/minor version number scheme for entities that communicate. Specifically, o2dlm has a major/minor version for interaction with o2dlm on other nodes, and ocfs2 itself has a major/minor version for interacting with the filesystem on other nodes. This will allow rolling upgrades of ocfs2 clusters when changes to the locking or network protocols can be done in a backwards compatible manner. In those cases, only the minor number is changed and the negotatied protocol minor is returned from dlm join. In the far less likely event that a required protocol change makes backwards compatibility impossible, we simply bump the major number. Signed-off-by: Joel Becker <joel.becker@oracle.com> Signed-off-by: Mark Fasheh <mark.fasheh@oracle.com>
Joel Becker · Mark Fasheh
1 parent 3e6bdf473f
Showing 12 changed files with 288 additions and 34 deletions Side-by-side Diff
fs/ocfs2/cluster/tcp_internal.h
fs/ocfs2/dlm/dlmapi.h
fs/ocfs2/dlm/dlmcommon.h
fs/ocfs2/dlm/dlmdomain.c
fs/ocfs2/dlm/dlmfs.c
fs/ocfs2/dlm/userdlm.c
fs/ocfs2/dlm/userdlm.h
fs/ocfs2/dlmglue.c
fs/ocfs2/dlmglue.h
fs/ocfs2/ocfs2.h
fs/ocfs2/ocfs2_lockingver.h
fs/ocfs2/super.c
@@ -38,6 +38,15 @@
  * locking semantics of the file system using the protocol.  It should 
  * be somewhere else, I'm sure, but right now it isn't.
  *
+ * With version 11, we separate out the filesystem locking portion.  The
+ * filesystem now has a major.minor version it negotiates.  Version 11
+ * introduces this negotiation to the o2dlm protocol, and as such the
+ * version here in tcp_internal.h should not need to be bumped for
+ * filesystem locking changes.
+ *
+ * New in version 11
+ * 	- Negotiation of filesystem locking in the dlm join.
+ *
  * New in version 10:
  * 	- Meta/data locks combined
  *
@@ -66,7 +75,7 @@
  * 	- full 64 bit i_size in the metadata lock lvbs
  * 	- introduction of "rw" lock and pushing meta/data locking down
  */
-#define O2NET_PROTOCOL_VERSION 10ULL
+#define O2NET_PROTOCOL_VERSION 11ULL
 struct o2net_handshake {
 	__be64	protocol_version;
 	__be64	connector_id;
@@ -193,7 +193,12 @@
 			  dlm_astunlockfunc_t *unlockast,
 			  void *data);
  
-struct dlm_ctxt * dlm_register_domain(const char *domain, u32 key);
+struct dlm_protocol_version {
+	u8 pv_major;
+	u8 pv_minor;
+};
+struct dlm_ctxt * dlm_register_domain(const char *domain, u32 key,
+				      struct dlm_protocol_version *fs_proto);
  
 void dlm_unregister_domain(struct dlm_ctxt *dlm);
  
@@ -142,6 +142,12 @@
 	spinlock_t work_lock;
 	struct list_head dlm_domain_handlers;
 	struct list_head	dlm_eviction_callbacks;
+
+	/* The filesystem specifies this at domain registration.  We
+	 * cache it here to know what to tell other nodes. */
+	struct dlm_protocol_version fs_locking_proto;
+	/* This is the inter-dlm communication version */
+	struct dlm_protocol_version dlm_locking_proto;
 };
  
 static inline struct hlist_head *dlm_lockres_hash(struct dlm_ctxt *dlm, unsigned i)
  
  
@@ -589,12 +595,26 @@
 #define DLM_PROXY_AST_MAX_LEN  (sizeof(struct dlm_proxy_ast)+DLM_LVB_LEN)
  
 #define DLM_MOD_KEY (0x666c6172)
-enum dlm_query_join_response {
+enum dlm_query_join_response_code {
 	JOIN_DISALLOW = 0,
 	JOIN_OK,
 	JOIN_OK_NO_MAP,
+	JOIN_PROTOCOL_MISMATCH,
 };
  
+union dlm_query_join_response {
+	u32 intval;
+	struct {
+		u8 code;	/* Response code.  dlm_minor and fs_minor
+				   are only valid if this is JOIN_OK */
+		u8 dlm_minor;	/* The minor version of the protocol the
+				   dlm is speaking. */
+		u8 fs_minor;	/* The minor version of the protocol the
+				   filesystem is speaking. */
+		u8 reserved;
+	} packet;
+};
+
 struct dlm_lock_request
 {
 	u8 node_idx;
@@ -633,6 +653,8 @@
 	u8 node_idx;
 	u8 pad1[2];
 	u8 name_len;
+	struct dlm_protocol_version dlm_proto;
+	struct dlm_protocol_version fs_proto;
 	u8 domain[O2NM_MAX_NAME_LEN];
 	u8 node_map[BITS_TO_BYTES(O2NM_MAX_NODES)];
 };
@@ -123,6 +123,17 @@
 LIST_HEAD(dlm_domains);
 static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events);
  
+/*
+ * The supported protocol version for DLM communication.  Running domains
+ * will have a negotiated version with the same major number and a minor
+ * number equal or smaller.  The dlm_ctxt->dlm_locking_proto field should
+ * be used to determine what a running domain is actually using.
+ */
+static const struct dlm_protocol_version dlm_protocol = {
+	.pv_major = 1,
+	.pv_minor = 0,
+};
+
 #define DLM_DOMAIN_BACKOFF_MS 200
  
 static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
@@ -133,6 +144,8 @@
 				   void **ret_data);
 static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
 				   void **ret_data);
+static int dlm_protocol_compare(struct dlm_protocol_version *existing,
+				struct dlm_protocol_version *request);
  
 static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm);
  
  
@@ -668,11 +681,45 @@
 }
 EXPORT_SYMBOL_GPL(dlm_unregister_domain);
  
+static int dlm_query_join_proto_check(char *proto_type, int node,
+				      struct dlm_protocol_version *ours,
+				      struct dlm_protocol_version *request)
+{
+	int rc;
+	struct dlm_protocol_version proto = *request;
+
+	if (!dlm_protocol_compare(ours, &proto)) {
+		mlog(0,
+		     "node %u wanted to join with %s locking protocol "
+		     "%u.%u, we respond with %u.%u\n",
+		     node, proto_type,
+		     request->pv_major,
+		     request->pv_minor,
+		     proto.pv_major, proto.pv_minor);
+		request->pv_minor = proto.pv_minor;
+		rc = 0;
+	} else {
+		mlog(ML_NOTICE,
+		     "Node %u wanted to join with %s locking "
+		     "protocol %u.%u, but we have %u.%u, disallowing\n",
+		     node, proto_type,
+		     request->pv_major,
+		     request->pv_minor,
+		     ours->pv_major,
+		     ours->pv_minor);
+		rc = 1;
+	}
+
+	return rc;
+}
+
 static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
 				  void **ret_data)
 {
 	struct dlm_query_join_request *query;
-	enum dlm_query_join_response response;
+	union dlm_query_join_response response = {
+		.packet.code = JOIN_DISALLOW,
+	};
 	struct dlm_ctxt *dlm = NULL;
 	u8 nodenum;
  
  
@@ -690,11 +737,11 @@
 		mlog(0, "node %u is not in our live map yet\n",
 		     query->node_idx);
  
-		response = JOIN_DISALLOW;
+		response.packet.code = JOIN_DISALLOW;
 		goto respond;
 	}
  
-	response = JOIN_OK_NO_MAP;
+	response.packet.code = JOIN_OK_NO_MAP;
  
 	spin_lock(&dlm_domain_lock);
 	dlm = __dlm_lookup_domain_full(query->domain, query->name_len);
@@ -713,7 +760,7 @@
 				mlog(0, "disallow join as node %u does not "
 				     "have node %u in its nodemap\n",
 				     query->node_idx, nodenum);
-				response = JOIN_DISALLOW;
+				response.packet.code = JOIN_DISALLOW;
 				goto unlock_respond;
 			}
 		}
  
  
  
  
  
@@ -733,30 +780,48 @@
 			/*If this is a brand new context and we
 			 * haven't started our join process yet, then
 			 * the other node won the race. */
-			response = JOIN_OK_NO_MAP;
+			response.packet.code = JOIN_OK_NO_MAP;
 		} else if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) {
 			/* Disallow parallel joins. */
-			response = JOIN_DISALLOW;
+			response.packet.code = JOIN_DISALLOW;
 		} else if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) {
 			mlog(0, "node %u trying to join, but recovery "
 			     "is ongoing.\n", bit);
-			response = JOIN_DISALLOW;
+			response.packet.code = JOIN_DISALLOW;
 		} else if (test_bit(bit, dlm->recovery_map)) {
 			mlog(0, "node %u trying to join, but it "
 			     "still needs recovery.\n", bit);
-			response = JOIN_DISALLOW;
+			response.packet.code = JOIN_DISALLOW;
 		} else if (test_bit(bit, dlm->domain_map)) {
 			mlog(0, "node %u trying to join, but it "
 			     "is still in the domain! needs recovery?\n",
 			     bit);
-			response = JOIN_DISALLOW;
+			response.packet.code = JOIN_DISALLOW;
 		} else {
 			/* Alright we're fully a part of this domain
 			 * so we keep some state as to who's joining
 			 * and indicate to him that needs to be fixed
 			 * up. */
-			response = JOIN_OK;
-			__dlm_set_joining_node(dlm, query->node_idx);
+
+			/* Make sure we speak compatible locking protocols.  */
+			if (dlm_query_join_proto_check("DLM", bit,
+						       &dlm->dlm_locking_proto,
+						       &query->dlm_proto)) {
+				response.packet.code =
+					JOIN_PROTOCOL_MISMATCH;
+			} else if (dlm_query_join_proto_check("fs", bit,
+							      &dlm->fs_locking_proto,
+							      &query->fs_proto)) {
+				response.packet.code =
+					JOIN_PROTOCOL_MISMATCH;
+			} else {
+				response.packet.dlm_minor =
+					query->dlm_proto.pv_minor;
+				response.packet.fs_minor =
+					query->fs_proto.pv_minor;
+				response.packet.code = JOIN_OK;
+				__dlm_set_joining_node(dlm, query->node_idx);
+			}
 		}
  
 		spin_unlock(&dlm->spinlock);
  
@@ -765,9 +830,9 @@
 	spin_unlock(&dlm_domain_lock);
  
 respond:
-	mlog(0, "We respond with %u\n", response);
+	mlog(0, "We respond with %u\n", response.packet.code);
  
-	return response;
+	return response.intval;
 }
  
 static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
  
  
@@ -899,10 +964,11 @@
  
 static int dlm_request_join(struct dlm_ctxt *dlm,
 			    int node,
-			    enum dlm_query_join_response *response)
+			    enum dlm_query_join_response_code *response)
 {
-	int status, retval;
+	int status;
 	struct dlm_query_join_request join_msg;
+	union dlm_query_join_response join_resp;
  
 	mlog(0, "querying node %d\n", node);
  
  
@@ -910,12 +976,15 @@
 	join_msg.node_idx = dlm->node_num;
 	join_msg.name_len = strlen(dlm->name);
 	memcpy(join_msg.domain, dlm->name, join_msg.name_len);
+	join_msg.dlm_proto = dlm->dlm_locking_proto;
+	join_msg.fs_proto = dlm->fs_locking_proto;
  
 	/* copy live node map to join message */
 	byte_copymap(join_msg.node_map, dlm->live_nodes_map, O2NM_MAX_NODES);
  
 	status = o2net_send_message(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, &join_msg,
-				    sizeof(join_msg), node, &retval);
+				    sizeof(join_msg), node,
+				    &join_resp.intval);
 	if (status < 0 && status != -ENOPROTOOPT) {
 		mlog_errno(status);
 		goto bail;
  
@@ -928,14 +997,41 @@
 	if (status == -ENOPROTOOPT) {
 		status = 0;
 		*response = JOIN_OK_NO_MAP;
-	} else if (retval == JOIN_DISALLOW ||
-		   retval == JOIN_OK ||
-		   retval == JOIN_OK_NO_MAP) {
-		*response = retval;
+	} else if (join_resp.packet.code == JOIN_DISALLOW ||
+		   join_resp.packet.code == JOIN_OK_NO_MAP) {
+		*response = join_resp.packet.code;
+	} else if (join_resp.packet.code == JOIN_PROTOCOL_MISMATCH) {
+		mlog(ML_NOTICE,
+		     "This node requested DLM locking protocol %u.%u and "
+		     "filesystem locking protocol %u.%u.  At least one of "
+		     "the protocol versions on node %d is not compatible, "
+		     "disconnecting\n",
+		     dlm->dlm_locking_proto.pv_major,
+		     dlm->dlm_locking_proto.pv_minor,
+		     dlm->fs_locking_proto.pv_major,
+		     dlm->fs_locking_proto.pv_minor,
+		     node);
+		status = -EPROTO;
+		*response = join_resp.packet.code;
+	} else if (join_resp.packet.code == JOIN_OK) {
+		*response = join_resp.packet.code;
+		/* Use the same locking protocol as the remote node */
+		dlm->dlm_locking_proto.pv_minor =
+			join_resp.packet.dlm_minor;
+		dlm->fs_locking_proto.pv_minor =
+			join_resp.packet.fs_minor;
+		mlog(0,
+		     "Node %d responds JOIN_OK with DLM locking protocol "
+		     "%u.%u and fs locking protocol %u.%u\n",
+		     node,
+		     dlm->dlm_locking_proto.pv_major,
+		     dlm->dlm_locking_proto.pv_minor,
+		     dlm->fs_locking_proto.pv_major,
+		     dlm->fs_locking_proto.pv_minor);
 	} else {
 		status = -EINVAL;
-		mlog(ML_ERROR, "invalid response %d from node %u\n", retval,
-		     node);
+		mlog(ML_ERROR, "invalid response %d from node %u\n",
+		     join_resp.packet.code, node);
 	}
  
 	mlog(0, "status %d, node %d response is %d\n", status, node,
@@ -1008,7 +1104,7 @@
  
 static int dlm_should_restart_join(struct dlm_ctxt *dlm,
 				   struct domain_join_ctxt *ctxt,
-				   enum dlm_query_join_response response)
+				   enum dlm_query_join_response_code response)
 {
 	int ret;
  
@@ -1034,7 +1130,7 @@
 {
 	int status = 0, tmpstat, node;
 	struct domain_join_ctxt *ctxt;
-	enum dlm_query_join_response response = JOIN_DISALLOW;
+	enum dlm_query_join_response_code response = JOIN_DISALLOW;
  
 	mlog_entry("%p", dlm);
  
  
  
@@ -1450,10 +1546,38 @@
 }
  
 /*
- * dlm_register_domain: one-time setup per "domain"
+ * Compare a requested locking protocol version against the current one.
+ *
+ * If the major numbers are different, they are incompatible.
+ * If the current minor is greater than the request, they are incompatible.
+ * If the current minor is less than or equal to the request, they are
+ * compatible, and the requester should run at the current minor version.
  */
+static int dlm_protocol_compare(struct dlm_protocol_version *existing,
+				struct dlm_protocol_version *request)
+{
+	if (existing->pv_major != request->pv_major)
+		return 1;
+
+	if (existing->pv_minor > request->pv_minor)
+		return 1;
+
+	if (existing->pv_minor < request->pv_minor)
+		request->pv_minor = existing->pv_minor;
+
+	return 0;
+}
+
+/*
+ * dlm_register_domain: one-time setup per "domain".
+ *
+ * The filesystem passes in the requested locking version via proto.
+ * If registration was successful, proto will contain the negotiated
+ * locking protocol.
+ */
 struct dlm_ctxt * dlm_register_domain(const char *domain,
-			       u32 key)
+			       u32 key,
+			       struct dlm_protocol_version *fs_proto)
 {
 	int ret;
 	struct dlm_ctxt *dlm = NULL;
@@ -1496,6 +1620,15 @@
 			goto retry;
 		}
  
+		if (dlm_protocol_compare(&dlm->fs_locking_proto, fs_proto)) {
+			mlog(ML_ERROR,
+			     "Requested locking protocol version is not "
+			     "compatible with already registered domain "
+			     "\"%s\"\n", domain);
+			ret = -EPROTO;
+			goto leave;
+		}
+
 		__dlm_get(dlm);
 		dlm->num_joins++;
  
  
@@ -1526,12 +1659,22 @@
 	list_add_tail(&dlm->list, &dlm_domains);
 	spin_unlock(&dlm_domain_lock);
  
+	/*
+	 * Pass the locking protocol version into the join.  If the join
+	 * succeeds, it will have the negotiated protocol set.
+	 */
+	dlm->dlm_locking_proto = dlm_protocol;
+	dlm->fs_locking_proto = *fs_proto;
+
 	ret = dlm_join_domain(dlm);
 	if (ret) {
 		mlog_errno(ret);
 		dlm_put(dlm);
 		goto leave;
 	}
+
+	/* Tell the caller what locking protocol we negotiated */
+	*fs_proto = dlm->fs_locking_proto;
  
 	ret = 0;
 leave:
@@ -60,6 +60,8 @@
 #define MLOG_MASK_PREFIX ML_DLMFS
 #include "cluster/masklog.h"
  
+#include "ocfs2_lockingver.h"
+
 static const struct super_operations dlmfs_ops;
 static const struct file_operations dlmfs_file_operations;
 static const struct inode_operations dlmfs_dir_inode_operations;
@@ -70,6 +72,16 @@
 struct workqueue_struct *user_dlm_worker;
  
 /*
+ * This is the userdlmfs locking protocol version.
+ *
+ * See fs/ocfs2/dlmglue.c for more details on locking versions.
+ */
+static const struct dlm_protocol_version user_locking_protocol = {
+	.pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
+	.pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
+};
+
+/*
  * decodes a set of open flags into a valid lock level and a set of flags.
  * returns < 0 if we have invalid flags
  * flags which mean something to us:
@@ -416,6 +428,7 @@
 	struct qstr *domain = &dentry->d_name;
 	struct dlmfs_inode_private *ip;
 	struct dlm_ctxt *dlm;
+	struct dlm_protocol_version proto = user_locking_protocol;
  
 	mlog(0, "mkdir %.*s\n", domain->len, domain->name);
  
@@ -435,7 +448,7 @@
  
 	ip = DLMFS_I(inode);
  
-	dlm = user_dlm_register_context(domain);
+	dlm = user_dlm_register_context(domain, &proto);
 	if (IS_ERR(dlm)) {
 		status = PTR_ERR(dlm);
 		mlog(ML_ERROR, "Error %d could not register domain \"%.*s\"\n",
@@ -645,7 +645,8 @@
 	return status;
 }
  
-struct dlm_ctxt *user_dlm_register_context(struct qstr *name)
+struct dlm_ctxt *user_dlm_register_context(struct qstr *name,
+					   struct dlm_protocol_version *proto)
 {
 	struct dlm_ctxt *dlm;
 	u32 dlm_key;
@@ -661,7 +662,7 @@
  
 	snprintf(domain, name->len + 1, "%.*s", name->len, name->name);
  
-	dlm = dlm_register_domain(domain, dlm_key);
+	dlm = dlm_register_domain(domain, dlm_key, proto);
 	if (IS_ERR(dlm))
 		mlog_errno(PTR_ERR(dlm));
  
@@ -83,7 +83,8 @@
 void user_dlm_read_lvb(struct inode *inode,
 		       char *val,
 		       unsigned int len);
-struct dlm_ctxt *user_dlm_register_context(struct qstr *name);
+struct dlm_ctxt *user_dlm_register_context(struct qstr *name,
+					   struct dlm_protocol_version *proto);
 void user_dlm_unregister_context(struct dlm_ctxt *dlm);
  
 struct dlmfs_inode_private {
@@ -43,6 +43,7 @@
 #include <cluster/masklog.h>
  
 #include "ocfs2.h"
+#include "ocfs2_lockingver.h"
  
 #include "alloc.h"
 #include "dcache.h"
@@ -258,6 +259,31 @@
 	.flags		= 0,
 };
  
+/*
+ * This is the filesystem locking protocol version.
+ *
+ * Whenever the filesystem does new things with locks (adds or removes a
+ * lock, orders them differently, does different things underneath a lock),
+ * the version must be changed.  The protocol is negotiated when joining
+ * the dlm domain.  A node may join the domain if its major version is
+ * identical to all other nodes and its minor version is greater than
+ * or equal to all other nodes.  When its minor version is greater than
+ * the other nodes, it will run at the minor version specified by the
+ * other nodes.
+ *
+ * If a locking change is made that will not be compatible with older
+ * versions, the major number must be increased and the minor version set
+ * to zero.  If a change merely adds a behavior that can be disabled when
+ * speaking to older versions, the minor version must be increased.  If a
+ * change adds a fully backwards compatible change (eg, LVB changes that
+ * are just ignored by older versions), the version does not need to be
+ * updated.
+ */
+const struct dlm_protocol_version ocfs2_locking_protocol = {
+	.pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
+	.pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
+};
+
 static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
 {
 	return lockres->l_type == OCFS2_LOCK_TYPE_META ||
@@ -2506,7 +2532,8 @@
 	dlm_key = crc32_le(0, osb->uuid_str, strlen(osb->uuid_str));
  
 	/* for now, uuid == domain */
-	dlm = dlm_register_domain(osb->uuid_str, dlm_key);
+	dlm = dlm_register_domain(osb->uuid_str, dlm_key,
+				  &osb->osb_locking_proto);
 	if (IS_ERR(dlm)) {
 		status = PTR_ERR(dlm);
 		mlog_errno(status);
@@ -116,5 +116,6 @@
 struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void);
 void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug);
  
+extern const struct dlm_protocol_version ocfs2_locking_protocol;
 #endif	/* DLMGLUE_H */
@@ -251,6 +251,7 @@
 	struct ocfs2_lock_res osb_rename_lockres;
 	struct dlm_eviction_cb osb_eviction_cb;
 	struct ocfs2_dlm_debug *osb_dlm_debug;
+	struct dlm_protocol_version osb_locking_proto;
  
 	struct dentry *osb_debug_root;
  
+/* -*- mode: c; c-basic-offset: 8; -*-
+ * vim: noexpandtab sw=8 ts=8 sts=0:
+ *
+ * ocfs2_lockingver.h
+ *
+ * Defines OCFS2 Locking version values.
+ *
+ * Copyright (C) 2008 Oracle.  All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License, version 2, as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ */
+
+#ifndef OCFS2_LOCKINGVER_H
+#define OCFS2_LOCKINGVER_H
+
+/*
+ * The protocol version for ocfs2 cluster locking.  See dlmglue.c for
+ * more details.
+ */
+#define OCFS2_LOCKING_PROTOCOL_MAJOR 1
+#define OCFS2_LOCKING_PROTOCOL_MINOR 0
+
+#endif  /* OCFS2_LOCKINGVER_H */
@@ -1355,6 +1355,7 @@
 	sb->s_fs_info = osb;
 	sb->s_op = &ocfs2_sops;
 	sb->s_export_op = &ocfs2_export_ops;
+	osb->osb_locking_proto = ocfs2_locking_protocol;
 	sb->s_time_gran = 1;
 	sb->s_flags |= MS_NOATIME;
 	/* this is needed to support O_LARGEFILE */
...	...	@@ -38,6 +38,15 @@
38	38	* locking semantics of the file system using the protocol. It should
39	39	* be somewhere else, I'm sure, but right now it isn't.
40	40	*
	41	+ * With version 11, we separate out the filesystem locking portion. The
	42	+ * filesystem now has a major.minor version it negotiates. Version 11
	43	+ * introduces this negotiation to the o2dlm protocol, and as such the
	44	+ * version here in tcp_internal.h should not need to be bumped for
	45	+ * filesystem locking changes.
	46	+ *
	47	+ * New in version 11
	48	+ * - Negotiation of filesystem locking in the dlm join.
	49	+ *
41	50	* New in version 10:
42	51	* - Meta/data locks combined
43	52	*
...	...	@@ -66,7 +75,7 @@
66	75	* - full 64 bit i_size in the metadata lock lvbs
67	76	* - introduction of "rw" lock and pushing meta/data locking down
68	77	*/
69		-#define O2NET_PROTOCOL_VERSION 10ULL
	78	+#define O2NET_PROTOCOL_VERSION 11ULL
70	79	struct o2net_handshake {
71	80	__be64 protocol_version;
72	81	__be64 connector_id;
...	...	@@ -193,7 +193,12 @@
193	193	dlm_astunlockfunc_t *unlockast,
194	194	void *data);
195	195
196		-struct dlm_ctxt * dlm_register_domain(const char *domain, u32 key);
	196	+struct dlm_protocol_version {
	197	+ u8 pv_major;
	198	+ u8 pv_minor;
	199	+};
	200	+struct dlm_ctxt * dlm_register_domain(const char *domain, u32 key,
	201	+ struct dlm_protocol_version *fs_proto);
197	202
198	203	void dlm_unregister_domain(struct dlm_ctxt *dlm);
199	204
...	...	@@ -142,6 +142,12 @@
142	142	spinlock_t work_lock;
143	143	struct list_head dlm_domain_handlers;
144	144	struct list_head dlm_eviction_callbacks;
	145	+
	146	+ /* The filesystem specifies this at domain registration. We
	147	+ * cache it here to know what to tell other nodes. */
	148	+ struct dlm_protocol_version fs_locking_proto;
	149	+ /* This is the inter-dlm communication version */
	150	+ struct dlm_protocol_version dlm_locking_proto;
145	151	};
146	152
147	153	static inline struct hlist_head dlm_lockres_hash(struct dlm_ctxt dlm, unsigned i)
148	154
149	155
...	...	@@ -589,12 +595,26 @@
589	595	#define DLM_PROXY_AST_MAX_LEN (sizeof(struct dlm_proxy_ast)+DLM_LVB_LEN)
590	596
591	597	#define DLM_MOD_KEY (0x666c6172)
592		-enum dlm_query_join_response {
	598	+enum dlm_query_join_response_code {
593	599	JOIN_DISALLOW = 0,
594	600	JOIN_OK,
595	601	JOIN_OK_NO_MAP,
	602	+ JOIN_PROTOCOL_MISMATCH,
596	603	};
597	604
	605	+union dlm_query_join_response {
	606	+ u32 intval;
	607	+ struct {
	608	+ u8 code; /* Response code. dlm_minor and fs_minor
	609	+ are only valid if this is JOIN_OK */
	610	+ u8 dlm_minor; /* The minor version of the protocol the
	611	+ dlm is speaking. */
	612	+ u8 fs_minor; /* The minor version of the protocol the
	613	+ filesystem is speaking. */
	614	+ u8 reserved;
	615	+ } packet;
	616	+};
	617	+
598	618	struct dlm_lock_request
599	619	{
600	620	u8 node_idx;
...	...	@@ -633,6 +653,8 @@
633	653	u8 node_idx;
634	654	u8 pad1[2];
635	655	u8 name_len;
	656	+ struct dlm_protocol_version dlm_proto;
	657	+ struct dlm_protocol_version fs_proto;
636	658	u8 domain[O2NM_MAX_NAME_LEN];
637	659	u8 node_map[BITS_TO_BYTES(O2NM_MAX_NODES)];
638	660	};
...	...	@@ -123,6 +123,17 @@
123	123	LIST_HEAD(dlm_domains);
124	124	static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events);
125	125
	126	+/*
	127	+ * The supported protocol version for DLM communication. Running domains
	128	+ * will have a negotiated version with the same major number and a minor
	129	+ * number equal or smaller. The dlm_ctxt->dlm_locking_proto field should
	130	+ * be used to determine what a running domain is actually using.
	131	+ */
	132	+static const struct dlm_protocol_version dlm_protocol = {
	133	+ .pv_major = 1,
	134	+ .pv_minor = 0,
	135	+};
	136	+
126	137	#define DLM_DOMAIN_BACKOFF_MS 200
127	138
128	139	static int dlm_query_join_handler(struct o2net_msg msg, u32 len, void data,
...	...	@@ -133,6 +144,8 @@
133	144	void **ret_data);
134	145	static int dlm_exit_domain_handler(struct o2net_msg msg, u32 len, void data,
135	146	void **ret_data);
	147	+static int dlm_protocol_compare(struct dlm_protocol_version *existing,
	148	+ struct dlm_protocol_version *request);
136	149
137	150	static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm);
138	151
139	152
...	...	@@ -668,11 +681,45 @@
668	681	}
669	682	EXPORT_SYMBOL_GPL(dlm_unregister_domain);
670	683
	684	+static int dlm_query_join_proto_check(char *proto_type, int node,
	685	+ struct dlm_protocol_version *ours,
	686	+ struct dlm_protocol_version *request)
	687	+{
	688	+ int rc;
	689	+ struct dlm_protocol_version proto = *request;
	690	+
	691	+ if (!dlm_protocol_compare(ours, &proto)) {
	692	+ mlog(0,
	693	+ "node %u wanted to join with %s locking protocol "
	694	+ "%u.%u, we respond with %u.%u\n",
	695	+ node, proto_type,
	696	+ request->pv_major,
	697	+ request->pv_minor,
	698	+ proto.pv_major, proto.pv_minor);
	699	+ request->pv_minor = proto.pv_minor;
	700	+ rc = 0;
	701	+ } else {
	702	+ mlog(ML_NOTICE,
	703	+ "Node %u wanted to join with %s locking "
	704	+ "protocol %u.%u, but we have %u.%u, disallowing\n",
	705	+ node, proto_type,
	706	+ request->pv_major,
	707	+ request->pv_minor,
	708	+ ours->pv_major,
	709	+ ours->pv_minor);
	710	+ rc = 1;
	711	+ }
	712	+
	713	+ return rc;
	714	+}
	715	+
671	716	static int dlm_query_join_handler(struct o2net_msg msg, u32 len, void data,
672	717	void **ret_data)
673	718	{
674	719	struct dlm_query_join_request *query;
675		- enum dlm_query_join_response response;
	720	+ union dlm_query_join_response response = {
	721	+ .packet.code = JOIN_DISALLOW,
	722	+ };
676	723	struct dlm_ctxt *dlm = NULL;
677	724	u8 nodenum;
678	725
679	726
...	...	@@ -690,11 +737,11 @@
690	737	mlog(0, "node %u is not in our live map yet\n",
691	738	query->node_idx);
692	739
693		- response = JOIN_DISALLOW;
	740	+ response.packet.code = JOIN_DISALLOW;
694	741	goto respond;
695	742	}
696	743
697		- response = JOIN_OK_NO_MAP;
	744	+ response.packet.code = JOIN_OK_NO_MAP;
698	745
699	746	spin_lock(&dlm_domain_lock);
700	747	dlm = __dlm_lookup_domain_full(query->domain, query->name_len);
...	...	@@ -713,7 +760,7 @@
713	760	mlog(0, "disallow join as node %u does not "
714	761	"have node %u in its nodemap\n",
715	762	query->node_idx, nodenum);
716		- response = JOIN_DISALLOW;
	763	+ response.packet.code = JOIN_DISALLOW;
717	764	goto unlock_respond;
718	765	}
719	766	}
720	767
721	768
722	769
723	770
724	771
...	...	@@ -733,30 +780,48 @@
733	780	/*If this is a brand new context and we
734	781	* haven't started our join process yet, then
735	782	* the other node won the race. */
736		- response = JOIN_OK_NO_MAP;
	783	+ response.packet.code = JOIN_OK_NO_MAP;
737	784	} else if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) {
738	785	/* Disallow parallel joins. */
739		- response = JOIN_DISALLOW;
	786	+ response.packet.code = JOIN_DISALLOW;
740	787	} else if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) {
741	788	mlog(0, "node %u trying to join, but recovery "
742	789	"is ongoing.\n", bit);
743		- response = JOIN_DISALLOW;
	790	+ response.packet.code = JOIN_DISALLOW;
744	791	} else if (test_bit(bit, dlm->recovery_map)) {
745	792	mlog(0, "node %u trying to join, but it "
746	793	"still needs recovery.\n", bit);
747		- response = JOIN_DISALLOW;
	794	+ response.packet.code = JOIN_DISALLOW;
748	795	} else if (test_bit(bit, dlm->domain_map)) {
749	796	mlog(0, "node %u trying to join, but it "
750	797	"is still in the domain! needs recovery?\n",
751	798	bit);
752		- response = JOIN_DISALLOW;
	799	+ response.packet.code = JOIN_DISALLOW;
753	800	} else {
754	801	/* Alright we're fully a part of this domain
755	802	* so we keep some state as to who's joining
756	803	* and indicate to him that needs to be fixed
757	804	* up. */
758		- response = JOIN_OK;
759		- __dlm_set_joining_node(dlm, query->node_idx);
	805	+
	806	+ /* Make sure we speak compatible locking protocols. */
	807	+ if (dlm_query_join_proto_check("DLM", bit,
	808	+ &dlm->dlm_locking_proto,
	809	+ &query->dlm_proto)) {
	810	+ response.packet.code =
	811	+ JOIN_PROTOCOL_MISMATCH;
	812	+ } else if (dlm_query_join_proto_check("fs", bit,
	813	+ &dlm->fs_locking_proto,
	814	+ &query->fs_proto)) {
	815	+ response.packet.code =
	816	+ JOIN_PROTOCOL_MISMATCH;
	817	+ } else {
	818	+ response.packet.dlm_minor =
	819	+ query->dlm_proto.pv_minor;
	820	+ response.packet.fs_minor =
	821	+ query->fs_proto.pv_minor;
	822	+ response.packet.code = JOIN_OK;
	823	+ __dlm_set_joining_node(dlm, query->node_idx);
	824	+ }
760	825	}
761	826
762	827	spin_unlock(&dlm->spinlock);
763	828
...	...	@@ -765,9 +830,9 @@
765	830	spin_unlock(&dlm_domain_lock);
766	831
767	832	respond:
768		- mlog(0, "We respond with %u\n", response);
	833	+ mlog(0, "We respond with %u\n", response.packet.code);
769	834
770		- return response;
	835	+ return response.intval;
771	836	}
772	837
773	838	static int dlm_assert_joined_handler(struct o2net_msg msg, u32 len, void data,
774	839
775	840
...	...	@@ -899,10 +964,11 @@
899	964
900	965	static int dlm_request_join(struct dlm_ctxt *dlm,
901	966	int node,
902		- enum dlm_query_join_response *response)
	967	+ enum dlm_query_join_response_code *response)
903	968	{
904		- int status, retval;
	969	+ int status;
905	970	struct dlm_query_join_request join_msg;
	971	+ union dlm_query_join_response join_resp;
906	972
907	973	mlog(0, "querying node %d\n", node);
908	974
909	975
...	...	@@ -910,12 +976,15 @@
910	976	join_msg.node_idx = dlm->node_num;
911	977	join_msg.name_len = strlen(dlm->name);
912	978	memcpy(join_msg.domain, dlm->name, join_msg.name_len);
	979	+ join_msg.dlm_proto = dlm->dlm_locking_proto;
	980	+ join_msg.fs_proto = dlm->fs_locking_proto;
913	981
914	982	/* copy live node map to join message */
915	983	byte_copymap(join_msg.node_map, dlm->live_nodes_map, O2NM_MAX_NODES);
916	984
917	985	status = o2net_send_message(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, &join_msg,
918		- sizeof(join_msg), node, &retval);
	986	+ sizeof(join_msg), node,
	987	+ &join_resp.intval);
919	988	if (status < 0 && status != -ENOPROTOOPT) {
920	989	mlog_errno(status);
921	990	goto bail;
922	991
...	...	@@ -928,14 +997,41 @@
928	997	if (status == -ENOPROTOOPT) {
929	998	status = 0;
930	999	*response = JOIN_OK_NO_MAP;
931		- } else if (retval == JOIN_DISALLOW \|\|
932		- retval == JOIN_OK \|\|
933		- retval == JOIN_OK_NO_MAP) {
934		- *response = retval;
	1000	+ } else if (join_resp.packet.code == JOIN_DISALLOW \|\|
	1001	+ join_resp.packet.code == JOIN_OK_NO_MAP) {
	1002	+ *response = join_resp.packet.code;
	1003	+ } else if (join_resp.packet.code == JOIN_PROTOCOL_MISMATCH) {
	1004	+ mlog(ML_NOTICE,
	1005	+ "This node requested DLM locking protocol %u.%u and "
	1006	+ "filesystem locking protocol %u.%u. At least one of "
	1007	+ "the protocol versions on node %d is not compatible, "
	1008	+ "disconnecting\n",
	1009	+ dlm->dlm_locking_proto.pv_major,
	1010	+ dlm->dlm_locking_proto.pv_minor,
	1011	+ dlm->fs_locking_proto.pv_major,
	1012	+ dlm->fs_locking_proto.pv_minor,
	1013	+ node);
	1014	+ status = -EPROTO;
	1015	+ *response = join_resp.packet.code;
	1016	+ } else if (join_resp.packet.code == JOIN_OK) {
	1017	+ *response = join_resp.packet.code;
	1018	+ /* Use the same locking protocol as the remote node */
	1019	+ dlm->dlm_locking_proto.pv_minor =
	1020	+ join_resp.packet.dlm_minor;
	1021	+ dlm->fs_locking_proto.pv_minor =
	1022	+ join_resp.packet.fs_minor;
	1023	+ mlog(0,
	1024	+ "Node %d responds JOIN_OK with DLM locking protocol "
	1025	+ "%u.%u and fs locking protocol %u.%u\n",
	1026	+ node,
	1027	+ dlm->dlm_locking_proto.pv_major,
	1028	+ dlm->dlm_locking_proto.pv_minor,
	1029	+ dlm->fs_locking_proto.pv_major,
	1030	+ dlm->fs_locking_proto.pv_minor);
935	1031	} else {
936	1032	status = -EINVAL;
937		- mlog(ML_ERROR, "invalid response %d from node %u\n", retval,
938		- node);
	1033	+ mlog(ML_ERROR, "invalid response %d from node %u\n",
	1034	+ join_resp.packet.code, node);
939	1035	}
940	1036
941	1037	mlog(0, "status %d, node %d response is %d\n", status, node,
...	...	@@ -1008,7 +1104,7 @@
1008	1104
1009	1105	static int dlm_should_restart_join(struct dlm_ctxt *dlm,
1010	1106	struct domain_join_ctxt *ctxt,
1011		- enum dlm_query_join_response response)
	1107	+ enum dlm_query_join_response_code response)
1012	1108	{
1013	1109	int ret;
1014	1110
...	...	@@ -1034,7 +1130,7 @@
1034	1130	{
1035	1131	int status = 0, tmpstat, node;
1036	1132	struct domain_join_ctxt *ctxt;
1037		- enum dlm_query_join_response response = JOIN_DISALLOW;
	1133	+ enum dlm_query_join_response_code response = JOIN_DISALLOW;
1038	1134
1039	1135	mlog_entry("%p", dlm);
1040	1136
1041	1137
1042	1138
...	...	@@ -1450,10 +1546,38 @@
1450	1546	}
1451	1547
1452	1548	/*
1453		- * dlm_register_domain: one-time setup per "domain"
	1549	+ * Compare a requested locking protocol version against the current one.
	1550	+ *
	1551	+ * If the major numbers are different, they are incompatible.
	1552	+ * If the current minor is greater than the request, they are incompatible.
	1553	+ * If the current minor is less than or equal to the request, they are
	1554	+ * compatible, and the requester should run at the current minor version.
1454	1555	*/
	1556	+static int dlm_protocol_compare(struct dlm_protocol_version *existing,
	1557	+ struct dlm_protocol_version *request)
	1558	+{
	1559	+ if (existing->pv_major != request->pv_major)
	1560	+ return 1;
	1561	+
	1562	+ if (existing->pv_minor > request->pv_minor)
	1563	+ return 1;
	1564	+
	1565	+ if (existing->pv_minor < request->pv_minor)
	1566	+ request->pv_minor = existing->pv_minor;
	1567	+
	1568	+ return 0;
	1569	+}
	1570	+
	1571	+/*
	1572	+ * dlm_register_domain: one-time setup per "domain".
	1573	+ *
	1574	+ * The filesystem passes in the requested locking version via proto.
	1575	+ * If registration was successful, proto will contain the negotiated
	1576	+ * locking protocol.
	1577	+ */
1455	1578	struct dlm_ctxt * dlm_register_domain(const char *domain,
1456		- u32 key)
	1579	+ u32 key,
	1580	+ struct dlm_protocol_version *fs_proto)
1457	1581	{
1458	1582	int ret;
1459	1583	struct dlm_ctxt *dlm = NULL;
...	...	@@ -1496,6 +1620,15 @@
1496	1620	goto retry;
1497	1621	}
1498	1622
	1623	+ if (dlm_protocol_compare(&dlm->fs_locking_proto, fs_proto)) {
	1624	+ mlog(ML_ERROR,
	1625	+ "Requested locking protocol version is not "
	1626	+ "compatible with already registered domain "
	1627	+ "\"%s\"\n", domain);
	1628	+ ret = -EPROTO;
	1629	+ goto leave;
	1630	+ }
	1631	+
1499	1632	__dlm_get(dlm);
1500	1633	dlm->num_joins++;
1501	1634
1502	1635
...	...	@@ -1526,12 +1659,22 @@
1526	1659	list_add_tail(&dlm->list, &dlm_domains);
1527	1660	spin_unlock(&dlm_domain_lock);
1528	1661
	1662	+ /*
	1663	+ * Pass the locking protocol version into the join. If the join
	1664	+ * succeeds, it will have the negotiated protocol set.
	1665	+ */
	1666	+ dlm->dlm_locking_proto = dlm_protocol;
	1667	+ dlm->fs_locking_proto = *fs_proto;
	1668	+
1529	1669	ret = dlm_join_domain(dlm);
1530	1670	if (ret) {
1531	1671	mlog_errno(ret);
1532	1672	dlm_put(dlm);
1533	1673	goto leave;
1534	1674	}
	1675	+
	1676	+ /* Tell the caller what locking protocol we negotiated */
	1677	+ *fs_proto = dlm->fs_locking_proto;
1535	1678
1536	1679	ret = 0;
1537	1680	leave:
...	...	@@ -60,6 +60,8 @@
60	60	#define MLOG_MASK_PREFIX ML_DLMFS
61	61	#include "cluster/masklog.h"
62	62
	63	+#include "ocfs2_lockingver.h"
	64	+
63	65	static const struct super_operations dlmfs_ops;
64	66	static const struct file_operations dlmfs_file_operations;
65	67	static const struct inode_operations dlmfs_dir_inode_operations;
...	...	@@ -70,6 +72,16 @@
70	72	struct workqueue_struct *user_dlm_worker;
71	73
72	74	/*
	75	+ * This is the userdlmfs locking protocol version.
	76	+ *
	77	+ * See fs/ocfs2/dlmglue.c for more details on locking versions.
	78	+ */
	79	+static const struct dlm_protocol_version user_locking_protocol = {
	80	+ .pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
	81	+ .pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
	82	+};
	83	+
	84	+/*
73	85	* decodes a set of open flags into a valid lock level and a set of flags.
74	86	* returns < 0 if we have invalid flags
75	87	* flags which mean something to us:
...	...	@@ -416,6 +428,7 @@
416	428	struct qstr *domain = &dentry->d_name;
417	429	struct dlmfs_inode_private *ip;
418	430	struct dlm_ctxt *dlm;
	431	+ struct dlm_protocol_version proto = user_locking_protocol;
419	432
420	433	mlog(0, "mkdir %.*s\n", domain->len, domain->name);
421	434
...	...	@@ -435,7 +448,7 @@
435	448
436	449	ip = DLMFS_I(inode);
437	450
438		- dlm = user_dlm_register_context(domain);
	451	+ dlm = user_dlm_register_context(domain, &proto);
439	452	if (IS_ERR(dlm)) {
440	453	status = PTR_ERR(dlm);
441	454	mlog(ML_ERROR, "Error %d could not register domain \"%.*s\"\n",
...	...	@@ -645,7 +645,8 @@
645	645	return status;
646	646	}
647	647
648		-struct dlm_ctxt user_dlm_register_context(struct qstr name)
	648	+struct dlm_ctxt user_dlm_register_context(struct qstr name,
	649	+ struct dlm_protocol_version *proto)
649	650	{
650	651	struct dlm_ctxt *dlm;
651	652	u32 dlm_key;
...	...	@@ -661,7 +662,7 @@
661	662
662	663	snprintf(domain, name->len + 1, "%.*s", name->len, name->name);
663	664
664		- dlm = dlm_register_domain(domain, dlm_key);
	665	+ dlm = dlm_register_domain(domain, dlm_key, proto);
665	666	if (IS_ERR(dlm))
666	667	mlog_errno(PTR_ERR(dlm));
667	668
...	...	@@ -83,7 +83,8 @@
83	83	void user_dlm_read_lvb(struct inode *inode,
84	84	char *val,
85	85	unsigned int len);
86		-struct dlm_ctxt user_dlm_register_context(struct qstr name);
	86	+struct dlm_ctxt user_dlm_register_context(struct qstr name,
	87	+ struct dlm_protocol_version *proto);
87	88	void user_dlm_unregister_context(struct dlm_ctxt *dlm);
88	89
89	90	struct dlmfs_inode_private {
...	...	@@ -43,6 +43,7 @@
43	43	#include <cluster/masklog.h>
44	44
45	45	#include "ocfs2.h"
	46	+#include "ocfs2_lockingver.h"
46	47
47	48	#include "alloc.h"
48	49	#include "dcache.h"
...	...	@@ -258,6 +259,31 @@
258	259	.flags = 0,
259	260	};
260	261
	262	+/*
	263	+ * This is the filesystem locking protocol version.
	264	+ *
	265	+ * Whenever the filesystem does new things with locks (adds or removes a
	266	+ * lock, orders them differently, does different things underneath a lock),
	267	+ * the version must be changed. The protocol is negotiated when joining
	268	+ * the dlm domain. A node may join the domain if its major version is
	269	+ * identical to all other nodes and its minor version is greater than
	270	+ * or equal to all other nodes. When its minor version is greater than
	271	+ * the other nodes, it will run at the minor version specified by the
	272	+ * other nodes.
	273	+ *
	274	+ * If a locking change is made that will not be compatible with older
	275	+ * versions, the major number must be increased and the minor version set
	276	+ * to zero. If a change merely adds a behavior that can be disabled when
	277	+ * speaking to older versions, the minor version must be increased. If a
	278	+ * change adds a fully backwards compatible change (eg, LVB changes that
	279	+ * are just ignored by older versions), the version does not need to be
	280	+ * updated.
	281	+ */
	282	+const struct dlm_protocol_version ocfs2_locking_protocol = {
	283	+ .pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
	284	+ .pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
	285	+};
	286	+
261	287	static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
262	288	{
263	289	return lockres->l_type == OCFS2_LOCK_TYPE_META \|\|
...	...	@@ -2506,7 +2532,8 @@
2506	2532	dlm_key = crc32_le(0, osb->uuid_str, strlen(osb->uuid_str));
2507	2533
2508	2534	/* for now, uuid == domain */
2509		- dlm = dlm_register_domain(osb->uuid_str, dlm_key);
	2535	+ dlm = dlm_register_domain(osb->uuid_str, dlm_key,
	2536	+ &osb->osb_locking_proto);
2510	2537	if (IS_ERR(dlm)) {
2511	2538	status = PTR_ERR(dlm);
2512	2539	mlog_errno(status);
...	...	@@ -116,5 +116,6 @@
116	116	struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void);
117	117	void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug);
118	118
	119	+extern const struct dlm_protocol_version ocfs2_locking_protocol;
119	120	#endif /* DLMGLUE_H */
...	...	@@ -251,6 +251,7 @@
251	251	struct ocfs2_lock_res osb_rename_lockres;
252	252	struct dlm_eviction_cb osb_eviction_cb;
253	253	struct ocfs2_dlm_debug *osb_dlm_debug;
	254	+ struct dlm_protocol_version osb_locking_proto;
254	255
255	256	struct dentry *osb_debug_root;
256	257