1 // SPDX-License-Identifier: GPL-2.0-only
2 /* -*- mode: c; c-basic-offset: 8; -*-
3 * vim: noexpandtab sw=8 ts=8 sts=0:
7 * Code which interfaces ocfs2 with fs/dlm and a userspace stack.
9 * Copyright (C) 2007 Oracle. All rights reserved.
12 #include <linux/module.h>
14 #include <linux/miscdevice.h>
15 #include <linux/mutex.h>
16 #include <linux/slab.h>
17 #include <linux/reboot.h>
18 #include <linux/sched.h>
19 #include <linux/uaccess.h>
21 #include "stackglue.h"
23 #include <linux/dlm_plock.h>
26 * The control protocol starts with a handshake. Until the handshake
27 * is complete, the control device will fail all write(2)s.
29 * The handshake is simple. First, the client reads until EOF. Each line
30 * of output is a supported protocol tag. All protocol tags are a single
31 * character followed by a two hex digit version number. Currently the
32 * only things supported is T01, for "Text-base version 0x01". Next, the
33 * client writes the version they would like to use, including the newline.
34 * Thus, the protocol tag is 'T01\n'. If the version tag written is
35 * unknown, -EINVAL is returned. Once the negotiation is complete, the
36 * client can start sending messages.
38 * The T01 protocol has three messages. First is the "SETN" message.
39 * It has the following syntax:
41 * SETN<space><8-char-hex-nodenum><newline>
43 * This is 14 characters.
45 * The "SETN" message must be the first message following the protocol.
46 * It tells ocfs2_control the local node number.
48 * Next comes the "SETV" message. It has the following syntax:
50 * SETV<space><2-char-hex-major><space><2-char-hex-minor><newline>
52 * This is 11 characters.
54 * The "SETV" message sets the filesystem locking protocol version as
55 * negotiated by the client. The client negotiates based on the maximum
56 * version advertised in /sys/fs/ocfs2/max_locking_protocol. The major
57 * number from the "SETV" message must match
58 * ocfs2_user_plugin.sp_max_proto.pv_major, and the minor number
59 * must be less than or equal to ...sp_max_version.pv_minor.
61 * Once this information has been set, mounts will be allowed. From this
62 * point on, the "DOWN" message can be sent for node down notification.
63 * It has the following syntax:
65 * DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline>
69 * DOWN 632A924FDD844190BDA93C0DF6B94899 00000001\n
71 * This is 47 characters.
75 * Whether or not the client has done the handshake.
76 * For now, we have just one protocol version.
78 #define OCFS2_CONTROL_PROTO "T01\n"
79 #define OCFS2_CONTROL_PROTO_LEN 4
81 /* Handshake states */
82 #define OCFS2_CONTROL_HANDSHAKE_INVALID (0)
83 #define OCFS2_CONTROL_HANDSHAKE_READ (1)
84 #define OCFS2_CONTROL_HANDSHAKE_PROTOCOL (2)
85 #define OCFS2_CONTROL_HANDSHAKE_VALID (3)
88 #define OCFS2_CONTROL_MESSAGE_OP_LEN 4
89 #define OCFS2_CONTROL_MESSAGE_SETNODE_OP "SETN"
90 #define OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN 14
91 #define OCFS2_CONTROL_MESSAGE_SETVERSION_OP "SETV"
92 #define OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN 11
93 #define OCFS2_CONTROL_MESSAGE_DOWN_OP "DOWN"
94 #define OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN 47
95 #define OCFS2_TEXT_UUID_LEN 32
96 #define OCFS2_CONTROL_MESSAGE_VERNUM_LEN 2
97 #define OCFS2_CONTROL_MESSAGE_NODENUM_LEN 8
98 #define VERSION_LOCK "version_lock"
100 enum ocfs2_connection_type {
106 * ocfs2_live_connection is refcounted because the filesystem and
107 * miscdevice sides can detach in different order. Let's just be safe.
109 struct ocfs2_live_connection {
110 struct list_head oc_list;
111 struct ocfs2_cluster_connection *oc_conn;
112 enum ocfs2_connection_type oc_type;
113 atomic_t oc_this_node;
115 struct dlm_lksb oc_version_lksb;
116 char oc_lvb[DLM_LVB_LEN];
117 struct completion oc_sync_wait;
118 wait_queue_head_t oc_wait;
121 struct ocfs2_control_private {
122 struct list_head op_list;
125 struct ocfs2_protocol_version op_proto;
128 /* SETN<space><8-char-hex-nodenum><newline> */
129 struct ocfs2_control_message_setn {
130 char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
132 char nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
136 /* SETV<space><2-char-hex-major><space><2-char-hex-minor><newline> */
137 struct ocfs2_control_message_setv {
138 char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
140 char major[OCFS2_CONTROL_MESSAGE_VERNUM_LEN];
142 char minor[OCFS2_CONTROL_MESSAGE_VERNUM_LEN];
146 /* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> */
147 struct ocfs2_control_message_down {
148 char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
150 char uuid[OCFS2_TEXT_UUID_LEN];
152 char nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
156 union ocfs2_control_message {
157 char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
158 struct ocfs2_control_message_setn u_setn;
159 struct ocfs2_control_message_setv u_setv;
160 struct ocfs2_control_message_down u_down;
163 static struct ocfs2_stack_plugin ocfs2_user_plugin;
165 static atomic_t ocfs2_control_opened;
166 static int ocfs2_control_this_node = -1;
167 static struct ocfs2_protocol_version running_proto;
169 static LIST_HEAD(ocfs2_live_connection_list);
170 static LIST_HEAD(ocfs2_control_private_list);
171 static DEFINE_MUTEX(ocfs2_control_lock);
173 static inline void ocfs2_control_set_handshake_state(struct file *file,
176 struct ocfs2_control_private *p = file->private_data;
180 static inline int ocfs2_control_get_handshake_state(struct file *file)
182 struct ocfs2_control_private *p = file->private_data;
186 static struct ocfs2_live_connection *ocfs2_connection_find(const char *name)
188 size_t len = strlen(name);
189 struct ocfs2_live_connection *c;
191 BUG_ON(!mutex_is_locked(&ocfs2_control_lock));
193 list_for_each_entry(c, &ocfs2_live_connection_list, oc_list) {
194 if ((c->oc_conn->cc_namelen == len) &&
195 !strncmp(c->oc_conn->cc_name, name, len))
203 * ocfs2_live_connection structures are created underneath the ocfs2
204 * mount path. Since the VFS prevents multiple calls to
205 * fill_super(), we can't get dupes here.
207 static int ocfs2_live_connection_attach(struct ocfs2_cluster_connection *conn,
208 struct ocfs2_live_connection *c)
212 mutex_lock(&ocfs2_control_lock);
215 if ((c->oc_type == NO_CONTROLD) || atomic_read(&ocfs2_control_opened))
216 list_add(&c->oc_list, &ocfs2_live_connection_list);
219 "ocfs2: Userspace control daemon is not present\n");
223 mutex_unlock(&ocfs2_control_lock);
228 * This function disconnects the cluster connection from ocfs2_control.
229 * Afterwards, userspace can't affect the cluster connection.
231 static void ocfs2_live_connection_drop(struct ocfs2_live_connection *c)
233 mutex_lock(&ocfs2_control_lock);
234 list_del_init(&c->oc_list);
236 mutex_unlock(&ocfs2_control_lock);
241 static int ocfs2_control_cfu(void *target, size_t target_len,
242 const char __user *buf, size_t count)
244 /* The T01 expects write(2) calls to have exactly one command */
245 if ((count != target_len) ||
246 (count > sizeof(union ocfs2_control_message)))
249 if (copy_from_user(target, buf, target_len))
255 static ssize_t ocfs2_control_validate_protocol(struct file *file,
256 const char __user *buf,
260 char kbuf[OCFS2_CONTROL_PROTO_LEN];
262 ret = ocfs2_control_cfu(kbuf, OCFS2_CONTROL_PROTO_LEN,
267 if (strncmp(kbuf, OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN))
270 ocfs2_control_set_handshake_state(file,
271 OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
276 static void ocfs2_control_send_down(const char *uuid,
279 struct ocfs2_live_connection *c;
281 mutex_lock(&ocfs2_control_lock);
283 c = ocfs2_connection_find(uuid);
285 BUG_ON(c->oc_conn == NULL);
286 c->oc_conn->cc_recovery_handler(nodenum,
287 c->oc_conn->cc_recovery_data);
290 mutex_unlock(&ocfs2_control_lock);
294 * Called whenever configuration elements are sent to /dev/ocfs2_control.
295 * If all configuration elements are present, try to set the global
296 * values. If there is a problem, return an error. Skip any missing
297 * elements, and only bump ocfs2_control_opened when we have all elements
298 * and are successful.
300 static int ocfs2_control_install_private(struct file *file)
304 struct ocfs2_control_private *p = file->private_data;
306 BUG_ON(p->op_state != OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
308 mutex_lock(&ocfs2_control_lock);
310 if (p->op_this_node < 0) {
312 } else if ((ocfs2_control_this_node >= 0) &&
313 (ocfs2_control_this_node != p->op_this_node)) {
318 if (!p->op_proto.pv_major) {
320 } else if (!list_empty(&ocfs2_live_connection_list) &&
321 ((running_proto.pv_major != p->op_proto.pv_major) ||
322 (running_proto.pv_minor != p->op_proto.pv_minor))) {
328 ocfs2_control_this_node = p->op_this_node;
329 running_proto.pv_major = p->op_proto.pv_major;
330 running_proto.pv_minor = p->op_proto.pv_minor;
334 mutex_unlock(&ocfs2_control_lock);
337 /* We set the global values successfully */
338 atomic_inc(&ocfs2_control_opened);
339 ocfs2_control_set_handshake_state(file,
340 OCFS2_CONTROL_HANDSHAKE_VALID);
346 static int ocfs2_control_get_this_node(void)
350 mutex_lock(&ocfs2_control_lock);
351 if (ocfs2_control_this_node < 0)
354 rc = ocfs2_control_this_node;
355 mutex_unlock(&ocfs2_control_lock);
360 static int ocfs2_control_do_setnode_msg(struct file *file,
361 struct ocfs2_control_message_setn *msg)
365 struct ocfs2_control_private *p = file->private_data;
367 if (ocfs2_control_get_handshake_state(file) !=
368 OCFS2_CONTROL_HANDSHAKE_PROTOCOL)
371 if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
372 OCFS2_CONTROL_MESSAGE_OP_LEN))
375 if ((msg->space != ' ') || (msg->newline != '\n'))
377 msg->space = msg->newline = '\0';
379 nodenum = simple_strtol(msg->nodestr, &ptr, 16);
383 if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) ||
384 (nodenum > INT_MAX) || (nodenum < 0))
386 p->op_this_node = nodenum;
388 return ocfs2_control_install_private(file);
391 static int ocfs2_control_do_setversion_msg(struct file *file,
392 struct ocfs2_control_message_setv *msg)
396 struct ocfs2_control_private *p = file->private_data;
397 struct ocfs2_protocol_version *max =
398 &ocfs2_user_plugin.sp_max_proto;
400 if (ocfs2_control_get_handshake_state(file) !=
401 OCFS2_CONTROL_HANDSHAKE_PROTOCOL)
404 if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETVERSION_OP,
405 OCFS2_CONTROL_MESSAGE_OP_LEN))
408 if ((msg->space1 != ' ') || (msg->space2 != ' ') ||
409 (msg->newline != '\n'))
411 msg->space1 = msg->space2 = msg->newline = '\0';
413 major = simple_strtol(msg->major, &ptr, 16);
416 minor = simple_strtol(msg->minor, &ptr, 16);
421 * The major must be between 1 and 255, inclusive. The minor
422 * must be between 0 and 255, inclusive. The version passed in
423 * must be within the maximum version supported by the filesystem.
425 if ((major == LONG_MIN) || (major == LONG_MAX) ||
426 (major > (u8)-1) || (major < 1))
428 if ((minor == LONG_MIN) || (minor == LONG_MAX) ||
429 (minor > (u8)-1) || (minor < 0))
431 if ((major != max->pv_major) ||
432 (minor > max->pv_minor))
435 p->op_proto.pv_major = major;
436 p->op_proto.pv_minor = minor;
438 return ocfs2_control_install_private(file);
441 static int ocfs2_control_do_down_msg(struct file *file,
442 struct ocfs2_control_message_down *msg)
447 if (ocfs2_control_get_handshake_state(file) !=
448 OCFS2_CONTROL_HANDSHAKE_VALID)
451 if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
452 OCFS2_CONTROL_MESSAGE_OP_LEN))
455 if ((msg->space1 != ' ') || (msg->space2 != ' ') ||
456 (msg->newline != '\n'))
458 msg->space1 = msg->space2 = msg->newline = '\0';
460 nodenum = simple_strtol(msg->nodestr, &p, 16);
464 if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) ||
465 (nodenum > INT_MAX) || (nodenum < 0))
468 ocfs2_control_send_down(msg->uuid, nodenum);
473 static ssize_t ocfs2_control_message(struct file *file,
474 const char __user *buf,
478 union ocfs2_control_message msg;
480 /* Try to catch padding issues */
481 WARN_ON(offsetof(struct ocfs2_control_message_down, uuid) !=
482 (sizeof(msg.u_down.tag) + sizeof(msg.u_down.space1)));
484 memset(&msg, 0, sizeof(union ocfs2_control_message));
485 ret = ocfs2_control_cfu(&msg, count, buf, count);
489 if ((count == OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN) &&
490 !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
491 OCFS2_CONTROL_MESSAGE_OP_LEN))
492 ret = ocfs2_control_do_setnode_msg(file, &msg.u_setn);
493 else if ((count == OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN) &&
494 !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETVERSION_OP,
495 OCFS2_CONTROL_MESSAGE_OP_LEN))
496 ret = ocfs2_control_do_setversion_msg(file, &msg.u_setv);
497 else if ((count == OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN) &&
498 !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
499 OCFS2_CONTROL_MESSAGE_OP_LEN))
500 ret = ocfs2_control_do_down_msg(file, &msg.u_down);
505 return ret ? ret : count;
508 static ssize_t ocfs2_control_write(struct file *file,
509 const char __user *buf,
515 switch (ocfs2_control_get_handshake_state(file)) {
516 case OCFS2_CONTROL_HANDSHAKE_INVALID:
520 case OCFS2_CONTROL_HANDSHAKE_READ:
521 ret = ocfs2_control_validate_protocol(file, buf,
525 case OCFS2_CONTROL_HANDSHAKE_PROTOCOL:
526 case OCFS2_CONTROL_HANDSHAKE_VALID:
527 ret = ocfs2_control_message(file, buf, count);
540 * This is a naive version. If we ever have a new protocol, we'll expand
541 * it. Probably using seq_file.
543 static ssize_t ocfs2_control_read(struct file *file,
550 ret = simple_read_from_buffer(buf, count, ppos,
551 OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN);
553 /* Have we read the whole protocol list? */
554 if (ret > 0 && *ppos >= OCFS2_CONTROL_PROTO_LEN)
555 ocfs2_control_set_handshake_state(file,
556 OCFS2_CONTROL_HANDSHAKE_READ);
561 static int ocfs2_control_release(struct inode *inode, struct file *file)
563 struct ocfs2_control_private *p = file->private_data;
565 mutex_lock(&ocfs2_control_lock);
567 if (ocfs2_control_get_handshake_state(file) !=
568 OCFS2_CONTROL_HANDSHAKE_VALID)
571 if (atomic_dec_and_test(&ocfs2_control_opened)) {
572 if (!list_empty(&ocfs2_live_connection_list)) {
573 /* XXX: Do bad things! */
575 "ocfs2: Unexpected release of ocfs2_control!\n"
576 " Loss of cluster connection requires "
577 "an emergency restart!\n");
581 * Last valid close clears the node number and resets
582 * the locking protocol version
584 ocfs2_control_this_node = -1;
585 running_proto.pv_major = 0;
586 running_proto.pv_minor = 0;
590 list_del_init(&p->op_list);
591 file->private_data = NULL;
593 mutex_unlock(&ocfs2_control_lock);
600 static int ocfs2_control_open(struct inode *inode, struct file *file)
602 struct ocfs2_control_private *p;
604 p = kzalloc(sizeof(struct ocfs2_control_private), GFP_KERNEL);
607 p->op_this_node = -1;
609 mutex_lock(&ocfs2_control_lock);
610 file->private_data = p;
611 list_add(&p->op_list, &ocfs2_control_private_list);
612 mutex_unlock(&ocfs2_control_lock);
617 static const struct file_operations ocfs2_control_fops = {
618 .open = ocfs2_control_open,
619 .release = ocfs2_control_release,
620 .read = ocfs2_control_read,
621 .write = ocfs2_control_write,
622 .owner = THIS_MODULE,
623 .llseek = default_llseek,
626 static struct miscdevice ocfs2_control_device = {
627 .minor = MISC_DYNAMIC_MINOR,
628 .name = "ocfs2_control",
629 .fops = &ocfs2_control_fops,
632 static int ocfs2_control_init(void)
636 atomic_set(&ocfs2_control_opened, 0);
638 rc = misc_register(&ocfs2_control_device);
641 "ocfs2: Unable to register ocfs2_control device "
648 static void ocfs2_control_exit(void)
650 misc_deregister(&ocfs2_control_device);
653 static void fsdlm_lock_ast_wrapper(void *astarg)
655 struct ocfs2_dlm_lksb *lksb = astarg;
656 int status = lksb->lksb_fsdlm.sb_status;
659 * For now we're punting on the issue of other non-standard errors
660 * where we can't tell if the unlock_ast or lock_ast should be called.
661 * The main "other error" that's possible is EINVAL which means the
662 * function was called with invalid args, which shouldn't be possible
663 * since the caller here is under our control. Other non-standard
664 * errors probably fall into the same category, or otherwise are fatal
665 * which means we can't carry on anyway.
668 if (status == -DLM_EUNLOCK || status == -DLM_ECANCEL)
669 lksb->lksb_conn->cc_proto->lp_unlock_ast(lksb, 0);
671 lksb->lksb_conn->cc_proto->lp_lock_ast(lksb);
674 static void fsdlm_blocking_ast_wrapper(void *astarg, int level)
676 struct ocfs2_dlm_lksb *lksb = astarg;
678 lksb->lksb_conn->cc_proto->lp_blocking_ast(lksb, level);
681 static int user_dlm_lock(struct ocfs2_cluster_connection *conn,
683 struct ocfs2_dlm_lksb *lksb,
686 unsigned int namelen)
690 if (!lksb->lksb_fsdlm.sb_lvbptr)
691 lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb +
692 sizeof(struct dlm_lksb);
694 ret = dlm_lock(conn->cc_lockspace, mode, &lksb->lksb_fsdlm,
695 flags|DLM_LKF_NODLCKWT, name, namelen, 0,
696 fsdlm_lock_ast_wrapper, lksb,
697 fsdlm_blocking_ast_wrapper);
701 static int user_dlm_unlock(struct ocfs2_cluster_connection *conn,
702 struct ocfs2_dlm_lksb *lksb,
707 ret = dlm_unlock(conn->cc_lockspace, lksb->lksb_fsdlm.sb_lkid,
708 flags, &lksb->lksb_fsdlm, lksb);
712 static int user_dlm_lock_status(struct ocfs2_dlm_lksb *lksb)
714 return lksb->lksb_fsdlm.sb_status;
717 static int user_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb)
719 int invalid = lksb->lksb_fsdlm.sb_flags & DLM_SBF_VALNOTVALID;
724 static void *user_dlm_lvb(struct ocfs2_dlm_lksb *lksb)
726 if (!lksb->lksb_fsdlm.sb_lvbptr)
727 lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb +
728 sizeof(struct dlm_lksb);
729 return (void *)(lksb->lksb_fsdlm.sb_lvbptr);
732 static void user_dlm_dump_lksb(struct ocfs2_dlm_lksb *lksb)
736 static int user_plock(struct ocfs2_cluster_connection *conn,
740 struct file_lock *fl)
743 * This more or less just demuxes the plock request into any
744 * one of three dlm calls.
746 * Internally, fs/dlm will pass these to a misc device, which
747 * a userspace daemon will read and write to.
749 * For now, cancel requests (which happen internally only),
750 * are turned into unlocks. Most of this function taken from
754 if (cmd == F_CANCELLK) {
756 fl->fl_type = F_UNLCK;
760 return dlm_posix_get(conn->cc_lockspace, ino, file, fl);
761 else if (fl->fl_type == F_UNLCK)
762 return dlm_posix_unlock(conn->cc_lockspace, ino, file, fl);
764 return dlm_posix_lock(conn->cc_lockspace, ino, file, cmd, fl);
768 * Compare a requested locking protocol version against the current one.
770 * If the major numbers are different, they are incompatible.
771 * If the current minor is greater than the request, they are incompatible.
772 * If the current minor is less than or equal to the request, they are
773 * compatible, and the requester should run at the current minor version.
775 static int fs_protocol_compare(struct ocfs2_protocol_version *existing,
776 struct ocfs2_protocol_version *request)
778 if (existing->pv_major != request->pv_major)
781 if (existing->pv_minor > request->pv_minor)
784 if (existing->pv_minor < request->pv_minor)
785 request->pv_minor = existing->pv_minor;
790 static void lvb_to_version(char *lvb, struct ocfs2_protocol_version *ver)
792 struct ocfs2_protocol_version *pv =
793 (struct ocfs2_protocol_version *)lvb;
795 * ocfs2_protocol_version has two u8 variables, so we don't
796 * need any endian conversion.
798 ver->pv_major = pv->pv_major;
799 ver->pv_minor = pv->pv_minor;
802 static void version_to_lvb(struct ocfs2_protocol_version *ver, char *lvb)
804 struct ocfs2_protocol_version *pv =
805 (struct ocfs2_protocol_version *)lvb;
807 * ocfs2_protocol_version has two u8 variables, so we don't
808 * need any endian conversion.
810 pv->pv_major = ver->pv_major;
811 pv->pv_minor = ver->pv_minor;
814 static void sync_wait_cb(void *arg)
816 struct ocfs2_cluster_connection *conn = arg;
817 struct ocfs2_live_connection *lc = conn->cc_private;
818 complete(&lc->oc_sync_wait);
821 static int sync_unlock(struct ocfs2_cluster_connection *conn,
822 struct dlm_lksb *lksb, char *name)
825 struct ocfs2_live_connection *lc = conn->cc_private;
827 error = dlm_unlock(conn->cc_lockspace, lksb->sb_lkid, 0, lksb, conn);
829 printk(KERN_ERR "%s lkid %x error %d\n",
830 name, lksb->sb_lkid, error);
834 wait_for_completion(&lc->oc_sync_wait);
836 if (lksb->sb_status != -DLM_EUNLOCK) {
837 printk(KERN_ERR "%s lkid %x status %d\n",
838 name, lksb->sb_lkid, lksb->sb_status);
844 static int sync_lock(struct ocfs2_cluster_connection *conn,
845 int mode, uint32_t flags,
846 struct dlm_lksb *lksb, char *name)
849 struct ocfs2_live_connection *lc = conn->cc_private;
851 error = dlm_lock(conn->cc_lockspace, mode, lksb, flags,
853 0, sync_wait_cb, conn, NULL);
855 printk(KERN_ERR "%s lkid %x flags %x mode %d error %d\n",
856 name, lksb->sb_lkid, flags, mode, error);
860 wait_for_completion(&lc->oc_sync_wait);
862 status = lksb->sb_status;
864 if (status && status != -EAGAIN) {
865 printk(KERN_ERR "%s lkid %x flags %x mode %d status %d\n",
866 name, lksb->sb_lkid, flags, mode, status);
873 static int version_lock(struct ocfs2_cluster_connection *conn, int mode,
876 struct ocfs2_live_connection *lc = conn->cc_private;
877 return sync_lock(conn, mode, flags,
878 &lc->oc_version_lksb, VERSION_LOCK);
881 static int version_unlock(struct ocfs2_cluster_connection *conn)
883 struct ocfs2_live_connection *lc = conn->cc_private;
884 return sync_unlock(conn, &lc->oc_version_lksb, VERSION_LOCK);
887 /* get_protocol_version()
889 * To exchange ocfs2 versioning, we use the LVB of the version dlm lock.
891 * 1. Attempt to take the lock in EX mode (non-blocking).
892 * 2. If successful (which means it is the first mount), write the
893 * version number and downconvert to PR lock.
894 * 3. If unsuccessful (returns -EAGAIN), read the version from the LVB after
895 * taking the PR lock.
898 static int get_protocol_version(struct ocfs2_cluster_connection *conn)
901 struct ocfs2_live_connection *lc = conn->cc_private;
902 struct ocfs2_protocol_version pv;
904 running_proto.pv_major =
905 ocfs2_user_plugin.sp_max_proto.pv_major;
906 running_proto.pv_minor =
907 ocfs2_user_plugin.sp_max_proto.pv_minor;
909 lc->oc_version_lksb.sb_lvbptr = lc->oc_lvb;
910 ret = version_lock(conn, DLM_LOCK_EX,
911 DLM_LKF_VALBLK|DLM_LKF_NOQUEUE);
913 conn->cc_version.pv_major = running_proto.pv_major;
914 conn->cc_version.pv_minor = running_proto.pv_minor;
915 version_to_lvb(&running_proto, lc->oc_lvb);
916 version_lock(conn, DLM_LOCK_PR, DLM_LKF_CONVERT|DLM_LKF_VALBLK);
917 } else if (ret == -EAGAIN) {
918 ret = version_lock(conn, DLM_LOCK_PR, DLM_LKF_VALBLK);
921 lvb_to_version(lc->oc_lvb, &pv);
923 if ((pv.pv_major != running_proto.pv_major) ||
924 (pv.pv_minor > running_proto.pv_minor)) {
929 conn->cc_version.pv_major = pv.pv_major;
930 conn->cc_version.pv_minor = pv.pv_minor;
936 static void user_recover_prep(void *arg)
940 static void user_recover_slot(void *arg, struct dlm_slot *slot)
942 struct ocfs2_cluster_connection *conn = arg;
943 printk(KERN_INFO "ocfs2: Node %d/%d down. Initiating recovery.\n",
944 slot->nodeid, slot->slot);
945 conn->cc_recovery_handler(slot->nodeid, conn->cc_recovery_data);
949 static void user_recover_done(void *arg, struct dlm_slot *slots,
950 int num_slots, int our_slot,
953 struct ocfs2_cluster_connection *conn = arg;
954 struct ocfs2_live_connection *lc = conn->cc_private;
957 for (i = 0; i < num_slots; i++)
958 if (slots[i].slot == our_slot) {
959 atomic_set(&lc->oc_this_node, slots[i].nodeid);
963 lc->oc_our_slot = our_slot;
964 wake_up(&lc->oc_wait);
967 static const struct dlm_lockspace_ops ocfs2_ls_ops = {
968 .recover_prep = user_recover_prep,
969 .recover_slot = user_recover_slot,
970 .recover_done = user_recover_done,
973 static int user_cluster_disconnect(struct ocfs2_cluster_connection *conn)
975 version_unlock(conn);
976 dlm_release_lockspace(conn->cc_lockspace, 2);
977 conn->cc_lockspace = NULL;
978 ocfs2_live_connection_drop(conn->cc_private);
979 conn->cc_private = NULL;
983 static int user_cluster_connect(struct ocfs2_cluster_connection *conn)
985 dlm_lockspace_t *fsdlm;
986 struct ocfs2_live_connection *lc;
989 BUG_ON(conn == NULL);
991 lc = kzalloc(sizeof(struct ocfs2_live_connection), GFP_KERNEL);
995 init_waitqueue_head(&lc->oc_wait);
996 init_completion(&lc->oc_sync_wait);
997 atomic_set(&lc->oc_this_node, 0);
998 conn->cc_private = lc;
999 lc->oc_type = NO_CONTROLD;
1001 rc = dlm_new_lockspace(conn->cc_name, conn->cc_cluster_name,
1002 DLM_LSFL_FS | DLM_LSFL_NEWEXCL, DLM_LVB_LEN,
1003 &ocfs2_ls_ops, conn, &ops_rv, &fsdlm);
1005 if (rc == -EEXIST || rc == -EPROTO)
1006 printk(KERN_ERR "ocfs2: Unable to create the "
1007 "lockspace %s (%d), because a ocfs2-tools "
1008 "program is running on this file system "
1009 "with the same name lockspace\n",
1014 if (ops_rv == -EOPNOTSUPP) {
1015 lc->oc_type = WITH_CONTROLD;
1016 printk(KERN_NOTICE "ocfs2: You seem to be using an older "
1017 "version of dlm_controld and/or ocfs2-tools."
1018 " Please consider upgrading.\n");
1019 } else if (ops_rv) {
1023 conn->cc_lockspace = fsdlm;
1025 rc = ocfs2_live_connection_attach(conn, lc);
1029 if (lc->oc_type == NO_CONTROLD) {
1030 rc = get_protocol_version(conn);
1032 printk(KERN_ERR "ocfs2: Could not determine"
1033 " locking version\n");
1034 user_cluster_disconnect(conn);
1037 wait_event(lc->oc_wait, (atomic_read(&lc->oc_this_node) > 0));
1041 * running_proto must have been set before we allowed any mounts
1044 if (fs_protocol_compare(&running_proto, &conn->cc_version)) {
1046 "Unable to mount with fs locking protocol version "
1047 "%u.%u because negotiated protocol is %u.%u\n",
1048 conn->cc_version.pv_major, conn->cc_version.pv_minor,
1049 running_proto.pv_major, running_proto.pv_minor);
1051 ocfs2_live_connection_drop(lc);
1062 static int user_cluster_this_node(struct ocfs2_cluster_connection *conn,
1063 unsigned int *this_node)
1066 struct ocfs2_live_connection *lc = conn->cc_private;
1068 if (lc->oc_type == WITH_CONTROLD)
1069 rc = ocfs2_control_get_this_node();
1070 else if (lc->oc_type == NO_CONTROLD)
1071 rc = atomic_read(&lc->oc_this_node);
1082 static struct ocfs2_stack_operations ocfs2_user_plugin_ops = {
1083 .connect = user_cluster_connect,
1084 .disconnect = user_cluster_disconnect,
1085 .this_node = user_cluster_this_node,
1086 .dlm_lock = user_dlm_lock,
1087 .dlm_unlock = user_dlm_unlock,
1088 .lock_status = user_dlm_lock_status,
1089 .lvb_valid = user_dlm_lvb_valid,
1090 .lock_lvb = user_dlm_lvb,
1091 .plock = user_plock,
1092 .dump_lksb = user_dlm_dump_lksb,
1095 static struct ocfs2_stack_plugin ocfs2_user_plugin = {
1097 .sp_ops = &ocfs2_user_plugin_ops,
1098 .sp_owner = THIS_MODULE,
1102 static int __init ocfs2_user_plugin_init(void)
1106 rc = ocfs2_control_init();
1108 rc = ocfs2_stack_glue_register(&ocfs2_user_plugin);
1110 ocfs2_control_exit();
1116 static void __exit ocfs2_user_plugin_exit(void)
1118 ocfs2_stack_glue_unregister(&ocfs2_user_plugin);
1119 ocfs2_control_exit();
1122 MODULE_AUTHOR("Oracle");
1123 MODULE_DESCRIPTION("ocfs2 driver for userspace cluster stacks");
1124 MODULE_LICENSE("GPL");
1125 module_init(ocfs2_user_plugin_init);
1126 module_exit(ocfs2_user_plugin_exit);