1 // SPDX-License-Identifier: GPL-2.0-only
5 * Code which interfaces ocfs2 with fs/dlm and a userspace stack.
7 * Copyright (C) 2007 Oracle. All rights reserved.
10 #include <linux/module.h>
12 #include <linux/miscdevice.h>
13 #include <linux/mutex.h>
14 #include <linux/slab.h>
15 #include <linux/reboot.h>
16 #include <linux/sched.h>
17 #include <linux/uaccess.h>
19 #include "stackglue.h"
21 #include <linux/dlm_plock.h>
24 * The control protocol starts with a handshake. Until the handshake
25 * is complete, the control device will fail all write(2)s.
27 * The handshake is simple. First, the client reads until EOF. Each line
28 * of output is a supported protocol tag. All protocol tags are a single
29 * character followed by a two hex digit version number. Currently the
30 * only things supported is T01, for "Text-base version 0x01". Next, the
31 * client writes the version they would like to use, including the newline.
32 * Thus, the protocol tag is 'T01\n'. If the version tag written is
33 * unknown, -EINVAL is returned. Once the negotiation is complete, the
34 * client can start sending messages.
36 * The T01 protocol has three messages. First is the "SETN" message.
37 * It has the following syntax:
39 * SETN<space><8-char-hex-nodenum><newline>
41 * This is 14 characters.
43 * The "SETN" message must be the first message following the protocol.
44 * It tells ocfs2_control the local node number.
46 * Next comes the "SETV" message. It has the following syntax:
48 * SETV<space><2-char-hex-major><space><2-char-hex-minor><newline>
50 * This is 11 characters.
52 * The "SETV" message sets the filesystem locking protocol version as
53 * negotiated by the client. The client negotiates based on the maximum
54 * version advertised in /sys/fs/ocfs2/max_locking_protocol. The major
55 * number from the "SETV" message must match
56 * ocfs2_user_plugin.sp_max_proto.pv_major, and the minor number
57 * must be less than or equal to ...sp_max_version.pv_minor.
59 * Once this information has been set, mounts will be allowed. From this
60 * point on, the "DOWN" message can be sent for node down notification.
61 * It has the following syntax:
63 * DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline>
67 * DOWN 632A924FDD844190BDA93C0DF6B94899 00000001\n
69 * This is 47 characters.
73 * Whether or not the client has done the handshake.
74 * For now, we have just one protocol version.
76 #define OCFS2_CONTROL_PROTO "T01\n"
77 #define OCFS2_CONTROL_PROTO_LEN 4
79 /* Handshake states */
80 #define OCFS2_CONTROL_HANDSHAKE_INVALID (0)
81 #define OCFS2_CONTROL_HANDSHAKE_READ (1)
82 #define OCFS2_CONTROL_HANDSHAKE_PROTOCOL (2)
83 #define OCFS2_CONTROL_HANDSHAKE_VALID (3)
86 #define OCFS2_CONTROL_MESSAGE_OP_LEN 4
87 #define OCFS2_CONTROL_MESSAGE_SETNODE_OP "SETN"
88 #define OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN 14
89 #define OCFS2_CONTROL_MESSAGE_SETVERSION_OP "SETV"
90 #define OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN 11
91 #define OCFS2_CONTROL_MESSAGE_DOWN_OP "DOWN"
92 #define OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN 47
93 #define OCFS2_TEXT_UUID_LEN 32
94 #define OCFS2_CONTROL_MESSAGE_VERNUM_LEN 2
95 #define OCFS2_CONTROL_MESSAGE_NODENUM_LEN 8
96 #define VERSION_LOCK "version_lock"
98 enum ocfs2_connection_type
{
104 * ocfs2_live_connection is refcounted because the filesystem and
105 * miscdevice sides can detach in different order. Let's just be safe.
107 struct ocfs2_live_connection
{
108 struct list_head oc_list
;
109 struct ocfs2_cluster_connection
*oc_conn
;
110 enum ocfs2_connection_type oc_type
;
111 atomic_t oc_this_node
;
113 struct dlm_lksb oc_version_lksb
;
114 char oc_lvb
[DLM_LVB_LEN
];
115 struct completion oc_sync_wait
;
116 wait_queue_head_t oc_wait
;
119 struct ocfs2_control_private
{
120 struct list_head op_list
;
123 struct ocfs2_protocol_version op_proto
;
126 /* SETN<space><8-char-hex-nodenum><newline> */
127 struct ocfs2_control_message_setn
{
128 char tag
[OCFS2_CONTROL_MESSAGE_OP_LEN
];
130 char nodestr
[OCFS2_CONTROL_MESSAGE_NODENUM_LEN
];
134 /* SETV<space><2-char-hex-major><space><2-char-hex-minor><newline> */
135 struct ocfs2_control_message_setv
{
136 char tag
[OCFS2_CONTROL_MESSAGE_OP_LEN
];
138 char major
[OCFS2_CONTROL_MESSAGE_VERNUM_LEN
];
140 char minor
[OCFS2_CONTROL_MESSAGE_VERNUM_LEN
];
144 /* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> */
145 struct ocfs2_control_message_down
{
146 char tag
[OCFS2_CONTROL_MESSAGE_OP_LEN
];
148 char uuid
[OCFS2_TEXT_UUID_LEN
];
150 char nodestr
[OCFS2_CONTROL_MESSAGE_NODENUM_LEN
];
154 union ocfs2_control_message
{
155 char tag
[OCFS2_CONTROL_MESSAGE_OP_LEN
];
156 struct ocfs2_control_message_setn u_setn
;
157 struct ocfs2_control_message_setv u_setv
;
158 struct ocfs2_control_message_down u_down
;
161 static struct ocfs2_stack_plugin ocfs2_user_plugin
;
163 static atomic_t ocfs2_control_opened
;
164 static int ocfs2_control_this_node
= -1;
165 static struct ocfs2_protocol_version running_proto
;
167 static LIST_HEAD(ocfs2_live_connection_list
);
168 static LIST_HEAD(ocfs2_control_private_list
);
169 static DEFINE_MUTEX(ocfs2_control_lock
);
171 static inline void ocfs2_control_set_handshake_state(struct file
*file
,
174 struct ocfs2_control_private
*p
= file
->private_data
;
178 static inline int ocfs2_control_get_handshake_state(struct file
*file
)
180 struct ocfs2_control_private
*p
= file
->private_data
;
184 static struct ocfs2_live_connection
*ocfs2_connection_find(const char *name
)
186 size_t len
= strlen(name
);
187 struct ocfs2_live_connection
*c
;
189 BUG_ON(!mutex_is_locked(&ocfs2_control_lock
));
191 list_for_each_entry(c
, &ocfs2_live_connection_list
, oc_list
) {
192 if ((c
->oc_conn
->cc_namelen
== len
) &&
193 !strncmp(c
->oc_conn
->cc_name
, name
, len
))
201 * ocfs2_live_connection structures are created underneath the ocfs2
202 * mount path. Since the VFS prevents multiple calls to
203 * fill_super(), we can't get dupes here.
205 static int ocfs2_live_connection_attach(struct ocfs2_cluster_connection
*conn
,
206 struct ocfs2_live_connection
*c
)
210 mutex_lock(&ocfs2_control_lock
);
213 if ((c
->oc_type
== NO_CONTROLD
) || atomic_read(&ocfs2_control_opened
))
214 list_add(&c
->oc_list
, &ocfs2_live_connection_list
);
217 "ocfs2: Userspace control daemon is not present\n");
221 mutex_unlock(&ocfs2_control_lock
);
226 * This function disconnects the cluster connection from ocfs2_control.
227 * Afterwards, userspace can't affect the cluster connection.
229 static void ocfs2_live_connection_drop(struct ocfs2_live_connection
*c
)
231 mutex_lock(&ocfs2_control_lock
);
232 list_del_init(&c
->oc_list
);
234 mutex_unlock(&ocfs2_control_lock
);
239 static int ocfs2_control_cfu(void *target
, size_t target_len
,
240 const char __user
*buf
, size_t count
)
242 /* The T01 expects write(2) calls to have exactly one command */
243 if ((count
!= target_len
) ||
244 (count
> sizeof(union ocfs2_control_message
)))
247 if (copy_from_user(target
, buf
, target_len
))
253 static ssize_t
ocfs2_control_validate_protocol(struct file
*file
,
254 const char __user
*buf
,
258 char kbuf
[OCFS2_CONTROL_PROTO_LEN
];
260 ret
= ocfs2_control_cfu(kbuf
, OCFS2_CONTROL_PROTO_LEN
,
265 if (strncmp(kbuf
, OCFS2_CONTROL_PROTO
, OCFS2_CONTROL_PROTO_LEN
))
268 ocfs2_control_set_handshake_state(file
,
269 OCFS2_CONTROL_HANDSHAKE_PROTOCOL
);
274 static void ocfs2_control_send_down(const char *uuid
,
277 struct ocfs2_live_connection
*c
;
279 mutex_lock(&ocfs2_control_lock
);
281 c
= ocfs2_connection_find(uuid
);
283 BUG_ON(c
->oc_conn
== NULL
);
284 c
->oc_conn
->cc_recovery_handler(nodenum
,
285 c
->oc_conn
->cc_recovery_data
);
288 mutex_unlock(&ocfs2_control_lock
);
292 * Called whenever configuration elements are sent to /dev/ocfs2_control.
293 * If all configuration elements are present, try to set the global
294 * values. If there is a problem, return an error. Skip any missing
295 * elements, and only bump ocfs2_control_opened when we have all elements
296 * and are successful.
298 static int ocfs2_control_install_private(struct file
*file
)
302 struct ocfs2_control_private
*p
= file
->private_data
;
304 BUG_ON(p
->op_state
!= OCFS2_CONTROL_HANDSHAKE_PROTOCOL
);
306 mutex_lock(&ocfs2_control_lock
);
308 if (p
->op_this_node
< 0) {
310 } else if ((ocfs2_control_this_node
>= 0) &&
311 (ocfs2_control_this_node
!= p
->op_this_node
)) {
316 if (!p
->op_proto
.pv_major
) {
318 } else if (!list_empty(&ocfs2_live_connection_list
) &&
319 ((running_proto
.pv_major
!= p
->op_proto
.pv_major
) ||
320 (running_proto
.pv_minor
!= p
->op_proto
.pv_minor
))) {
326 ocfs2_control_this_node
= p
->op_this_node
;
327 running_proto
.pv_major
= p
->op_proto
.pv_major
;
328 running_proto
.pv_minor
= p
->op_proto
.pv_minor
;
332 mutex_unlock(&ocfs2_control_lock
);
335 /* We set the global values successfully */
336 atomic_inc(&ocfs2_control_opened
);
337 ocfs2_control_set_handshake_state(file
,
338 OCFS2_CONTROL_HANDSHAKE_VALID
);
344 static int ocfs2_control_get_this_node(void)
348 mutex_lock(&ocfs2_control_lock
);
349 if (ocfs2_control_this_node
< 0)
352 rc
= ocfs2_control_this_node
;
353 mutex_unlock(&ocfs2_control_lock
);
358 static int ocfs2_control_do_setnode_msg(struct file
*file
,
359 struct ocfs2_control_message_setn
*msg
)
363 struct ocfs2_control_private
*p
= file
->private_data
;
365 if (ocfs2_control_get_handshake_state(file
) !=
366 OCFS2_CONTROL_HANDSHAKE_PROTOCOL
)
369 if (strncmp(msg
->tag
, OCFS2_CONTROL_MESSAGE_SETNODE_OP
,
370 OCFS2_CONTROL_MESSAGE_OP_LEN
))
373 if ((msg
->space
!= ' ') || (msg
->newline
!= '\n'))
375 msg
->space
= msg
->newline
= '\0';
377 nodenum
= simple_strtol(msg
->nodestr
, &ptr
, 16);
381 if ((nodenum
== LONG_MIN
) || (nodenum
== LONG_MAX
) ||
382 (nodenum
> INT_MAX
) || (nodenum
< 0))
384 p
->op_this_node
= nodenum
;
386 return ocfs2_control_install_private(file
);
389 static int ocfs2_control_do_setversion_msg(struct file
*file
,
390 struct ocfs2_control_message_setv
*msg
)
394 struct ocfs2_control_private
*p
= file
->private_data
;
395 struct ocfs2_protocol_version
*max
=
396 &ocfs2_user_plugin
.sp_max_proto
;
398 if (ocfs2_control_get_handshake_state(file
) !=
399 OCFS2_CONTROL_HANDSHAKE_PROTOCOL
)
402 if (strncmp(msg
->tag
, OCFS2_CONTROL_MESSAGE_SETVERSION_OP
,
403 OCFS2_CONTROL_MESSAGE_OP_LEN
))
406 if ((msg
->space1
!= ' ') || (msg
->space2
!= ' ') ||
407 (msg
->newline
!= '\n'))
409 msg
->space1
= msg
->space2
= msg
->newline
= '\0';
411 major
= simple_strtol(msg
->major
, &ptr
, 16);
414 minor
= simple_strtol(msg
->minor
, &ptr
, 16);
419 * The major must be between 1 and 255, inclusive. The minor
420 * must be between 0 and 255, inclusive. The version passed in
421 * must be within the maximum version supported by the filesystem.
423 if ((major
== LONG_MIN
) || (major
== LONG_MAX
) ||
424 (major
> (u8
)-1) || (major
< 1))
426 if ((minor
== LONG_MIN
) || (minor
== LONG_MAX
) ||
427 (minor
> (u8
)-1) || (minor
< 0))
429 if ((major
!= max
->pv_major
) ||
430 (minor
> max
->pv_minor
))
433 p
->op_proto
.pv_major
= major
;
434 p
->op_proto
.pv_minor
= minor
;
436 return ocfs2_control_install_private(file
);
439 static int ocfs2_control_do_down_msg(struct file
*file
,
440 struct ocfs2_control_message_down
*msg
)
445 if (ocfs2_control_get_handshake_state(file
) !=
446 OCFS2_CONTROL_HANDSHAKE_VALID
)
449 if (strncmp(msg
->tag
, OCFS2_CONTROL_MESSAGE_DOWN_OP
,
450 OCFS2_CONTROL_MESSAGE_OP_LEN
))
453 if ((msg
->space1
!= ' ') || (msg
->space2
!= ' ') ||
454 (msg
->newline
!= '\n'))
456 msg
->space1
= msg
->space2
= msg
->newline
= '\0';
458 nodenum
= simple_strtol(msg
->nodestr
, &p
, 16);
462 if ((nodenum
== LONG_MIN
) || (nodenum
== LONG_MAX
) ||
463 (nodenum
> INT_MAX
) || (nodenum
< 0))
466 ocfs2_control_send_down(msg
->uuid
, nodenum
);
471 static ssize_t
ocfs2_control_message(struct file
*file
,
472 const char __user
*buf
,
476 union ocfs2_control_message msg
;
478 /* Try to catch padding issues */
479 WARN_ON(offsetof(struct ocfs2_control_message_down
, uuid
) !=
480 (sizeof(msg
.u_down
.tag
) + sizeof(msg
.u_down
.space1
)));
482 memset(&msg
, 0, sizeof(union ocfs2_control_message
));
483 ret
= ocfs2_control_cfu(&msg
, count
, buf
, count
);
487 if ((count
== OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN
) &&
488 !strncmp(msg
.tag
, OCFS2_CONTROL_MESSAGE_SETNODE_OP
,
489 OCFS2_CONTROL_MESSAGE_OP_LEN
))
490 ret
= ocfs2_control_do_setnode_msg(file
, &msg
.u_setn
);
491 else if ((count
== OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN
) &&
492 !strncmp(msg
.tag
, OCFS2_CONTROL_MESSAGE_SETVERSION_OP
,
493 OCFS2_CONTROL_MESSAGE_OP_LEN
))
494 ret
= ocfs2_control_do_setversion_msg(file
, &msg
.u_setv
);
495 else if ((count
== OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN
) &&
496 !strncmp(msg
.tag
, OCFS2_CONTROL_MESSAGE_DOWN_OP
,
497 OCFS2_CONTROL_MESSAGE_OP_LEN
))
498 ret
= ocfs2_control_do_down_msg(file
, &msg
.u_down
);
503 return ret
? ret
: count
;
506 static ssize_t
ocfs2_control_write(struct file
*file
,
507 const char __user
*buf
,
513 switch (ocfs2_control_get_handshake_state(file
)) {
514 case OCFS2_CONTROL_HANDSHAKE_INVALID
:
518 case OCFS2_CONTROL_HANDSHAKE_READ
:
519 ret
= ocfs2_control_validate_protocol(file
, buf
,
523 case OCFS2_CONTROL_HANDSHAKE_PROTOCOL
:
524 case OCFS2_CONTROL_HANDSHAKE_VALID
:
525 ret
= ocfs2_control_message(file
, buf
, count
);
538 * This is a naive version. If we ever have a new protocol, we'll expand
539 * it. Probably using seq_file.
541 static ssize_t
ocfs2_control_read(struct file
*file
,
548 ret
= simple_read_from_buffer(buf
, count
, ppos
,
549 OCFS2_CONTROL_PROTO
, OCFS2_CONTROL_PROTO_LEN
);
551 /* Have we read the whole protocol list? */
552 if (ret
> 0 && *ppos
>= OCFS2_CONTROL_PROTO_LEN
)
553 ocfs2_control_set_handshake_state(file
,
554 OCFS2_CONTROL_HANDSHAKE_READ
);
559 static int ocfs2_control_release(struct inode
*inode
, struct file
*file
)
561 struct ocfs2_control_private
*p
= file
->private_data
;
563 mutex_lock(&ocfs2_control_lock
);
565 if (ocfs2_control_get_handshake_state(file
) !=
566 OCFS2_CONTROL_HANDSHAKE_VALID
)
569 if (atomic_dec_and_test(&ocfs2_control_opened
)) {
570 if (!list_empty(&ocfs2_live_connection_list
)) {
571 /* XXX: Do bad things! */
573 "ocfs2: Unexpected release of ocfs2_control!\n"
574 " Loss of cluster connection requires "
575 "an emergency restart!\n");
579 * Last valid close clears the node number and resets
580 * the locking protocol version
582 ocfs2_control_this_node
= -1;
583 running_proto
.pv_major
= 0;
584 running_proto
.pv_minor
= 0;
588 list_del_init(&p
->op_list
);
589 file
->private_data
= NULL
;
591 mutex_unlock(&ocfs2_control_lock
);
598 static int ocfs2_control_open(struct inode
*inode
, struct file
*file
)
600 struct ocfs2_control_private
*p
;
602 p
= kzalloc(sizeof(struct ocfs2_control_private
), GFP_KERNEL
);
605 p
->op_this_node
= -1;
607 mutex_lock(&ocfs2_control_lock
);
608 file
->private_data
= p
;
609 list_add(&p
->op_list
, &ocfs2_control_private_list
);
610 mutex_unlock(&ocfs2_control_lock
);
615 static const struct file_operations ocfs2_control_fops
= {
616 .open
= ocfs2_control_open
,
617 .release
= ocfs2_control_release
,
618 .read
= ocfs2_control_read
,
619 .write
= ocfs2_control_write
,
620 .owner
= THIS_MODULE
,
621 .llseek
= default_llseek
,
624 static struct miscdevice ocfs2_control_device
= {
625 .minor
= MISC_DYNAMIC_MINOR
,
626 .name
= "ocfs2_control",
627 .fops
= &ocfs2_control_fops
,
630 static int ocfs2_control_init(void)
634 atomic_set(&ocfs2_control_opened
, 0);
636 rc
= misc_register(&ocfs2_control_device
);
639 "ocfs2: Unable to register ocfs2_control device "
646 static void ocfs2_control_exit(void)
648 misc_deregister(&ocfs2_control_device
);
651 static void fsdlm_lock_ast_wrapper(void *astarg
)
653 struct ocfs2_dlm_lksb
*lksb
= astarg
;
654 int status
= lksb
->lksb_fsdlm
.sb_status
;
657 * For now we're punting on the issue of other non-standard errors
658 * where we can't tell if the unlock_ast or lock_ast should be called.
659 * The main "other error" that's possible is EINVAL which means the
660 * function was called with invalid args, which shouldn't be possible
661 * since the caller here is under our control. Other non-standard
662 * errors probably fall into the same category, or otherwise are fatal
663 * which means we can't carry on anyway.
666 if (status
== -DLM_EUNLOCK
|| status
== -DLM_ECANCEL
)
667 lksb
->lksb_conn
->cc_proto
->lp_unlock_ast(lksb
, 0);
669 lksb
->lksb_conn
->cc_proto
->lp_lock_ast(lksb
);
672 static void fsdlm_blocking_ast_wrapper(void *astarg
, int level
)
674 struct ocfs2_dlm_lksb
*lksb
= astarg
;
676 lksb
->lksb_conn
->cc_proto
->lp_blocking_ast(lksb
, level
);
679 static int user_dlm_lock(struct ocfs2_cluster_connection
*conn
,
681 struct ocfs2_dlm_lksb
*lksb
,
684 unsigned int namelen
)
686 if (!lksb
->lksb_fsdlm
.sb_lvbptr
)
687 lksb
->lksb_fsdlm
.sb_lvbptr
= (char *)lksb
+
688 sizeof(struct dlm_lksb
);
690 return dlm_lock(conn
->cc_lockspace
, mode
, &lksb
->lksb_fsdlm
,
691 flags
|DLM_LKF_NODLCKWT
, name
, namelen
, 0,
692 fsdlm_lock_ast_wrapper
, lksb
,
693 fsdlm_blocking_ast_wrapper
);
696 static int user_dlm_unlock(struct ocfs2_cluster_connection
*conn
,
697 struct ocfs2_dlm_lksb
*lksb
,
700 return dlm_unlock(conn
->cc_lockspace
, lksb
->lksb_fsdlm
.sb_lkid
,
701 flags
, &lksb
->lksb_fsdlm
, lksb
);
704 static int user_dlm_lock_status(struct ocfs2_dlm_lksb
*lksb
)
706 return lksb
->lksb_fsdlm
.sb_status
;
709 static int user_dlm_lvb_valid(struct ocfs2_dlm_lksb
*lksb
)
711 int invalid
= lksb
->lksb_fsdlm
.sb_flags
& DLM_SBF_VALNOTVALID
;
716 static void *user_dlm_lvb(struct ocfs2_dlm_lksb
*lksb
)
718 if (!lksb
->lksb_fsdlm
.sb_lvbptr
)
719 lksb
->lksb_fsdlm
.sb_lvbptr
= (char *)lksb
+
720 sizeof(struct dlm_lksb
);
721 return (void *)(lksb
->lksb_fsdlm
.sb_lvbptr
);
724 static void user_dlm_dump_lksb(struct ocfs2_dlm_lksb
*lksb
)
728 static int user_plock(struct ocfs2_cluster_connection
*conn
,
732 struct file_lock
*fl
)
735 * This more or less just demuxes the plock request into any
736 * one of three dlm calls.
738 * Internally, fs/dlm will pass these to a misc device, which
739 * a userspace daemon will read and write to.
741 * For now, cancel requests (which happen internally only),
742 * are turned into unlocks. Most of this function taken from
746 if (cmd
== F_CANCELLK
) {
748 fl
->fl_type
= F_UNLCK
;
752 return dlm_posix_get(conn
->cc_lockspace
, ino
, file
, fl
);
753 else if (fl
->fl_type
== F_UNLCK
)
754 return dlm_posix_unlock(conn
->cc_lockspace
, ino
, file
, fl
);
756 return dlm_posix_lock(conn
->cc_lockspace
, ino
, file
, cmd
, fl
);
760 * Compare a requested locking protocol version against the current one.
762 * If the major numbers are different, they are incompatible.
763 * If the current minor is greater than the request, they are incompatible.
764 * If the current minor is less than or equal to the request, they are
765 * compatible, and the requester should run at the current minor version.
767 static int fs_protocol_compare(struct ocfs2_protocol_version
*existing
,
768 struct ocfs2_protocol_version
*request
)
770 if (existing
->pv_major
!= request
->pv_major
)
773 if (existing
->pv_minor
> request
->pv_minor
)
776 if (existing
->pv_minor
< request
->pv_minor
)
777 request
->pv_minor
= existing
->pv_minor
;
782 static void lvb_to_version(char *lvb
, struct ocfs2_protocol_version
*ver
)
784 struct ocfs2_protocol_version
*pv
=
785 (struct ocfs2_protocol_version
*)lvb
;
787 * ocfs2_protocol_version has two u8 variables, so we don't
788 * need any endian conversion.
790 ver
->pv_major
= pv
->pv_major
;
791 ver
->pv_minor
= pv
->pv_minor
;
794 static void version_to_lvb(struct ocfs2_protocol_version
*ver
, char *lvb
)
796 struct ocfs2_protocol_version
*pv
=
797 (struct ocfs2_protocol_version
*)lvb
;
799 * ocfs2_protocol_version has two u8 variables, so we don't
800 * need any endian conversion.
802 pv
->pv_major
= ver
->pv_major
;
803 pv
->pv_minor
= ver
->pv_minor
;
806 static void sync_wait_cb(void *arg
)
808 struct ocfs2_cluster_connection
*conn
= arg
;
809 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
810 complete(&lc
->oc_sync_wait
);
813 static int sync_unlock(struct ocfs2_cluster_connection
*conn
,
814 struct dlm_lksb
*lksb
, char *name
)
817 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
819 error
= dlm_unlock(conn
->cc_lockspace
, lksb
->sb_lkid
, 0, lksb
, conn
);
821 printk(KERN_ERR
"%s lkid %x error %d\n",
822 name
, lksb
->sb_lkid
, error
);
826 wait_for_completion(&lc
->oc_sync_wait
);
828 if (lksb
->sb_status
!= -DLM_EUNLOCK
) {
829 printk(KERN_ERR
"%s lkid %x status %d\n",
830 name
, lksb
->sb_lkid
, lksb
->sb_status
);
836 static int sync_lock(struct ocfs2_cluster_connection
*conn
,
837 int mode
, uint32_t flags
,
838 struct dlm_lksb
*lksb
, char *name
)
841 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
843 error
= dlm_lock(conn
->cc_lockspace
, mode
, lksb
, flags
,
845 0, sync_wait_cb
, conn
, NULL
);
847 printk(KERN_ERR
"%s lkid %x flags %x mode %d error %d\n",
848 name
, lksb
->sb_lkid
, flags
, mode
, error
);
852 wait_for_completion(&lc
->oc_sync_wait
);
854 status
= lksb
->sb_status
;
856 if (status
&& status
!= -EAGAIN
) {
857 printk(KERN_ERR
"%s lkid %x flags %x mode %d status %d\n",
858 name
, lksb
->sb_lkid
, flags
, mode
, status
);
865 static int version_lock(struct ocfs2_cluster_connection
*conn
, int mode
,
868 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
869 return sync_lock(conn
, mode
, flags
,
870 &lc
->oc_version_lksb
, VERSION_LOCK
);
873 static int version_unlock(struct ocfs2_cluster_connection
*conn
)
875 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
876 return sync_unlock(conn
, &lc
->oc_version_lksb
, VERSION_LOCK
);
879 /* get_protocol_version()
881 * To exchange ocfs2 versioning, we use the LVB of the version dlm lock.
883 * 1. Attempt to take the lock in EX mode (non-blocking).
884 * 2. If successful (which means it is the first mount), write the
885 * version number and downconvert to PR lock.
886 * 3. If unsuccessful (returns -EAGAIN), read the version from the LVB after
887 * taking the PR lock.
890 static int get_protocol_version(struct ocfs2_cluster_connection
*conn
)
893 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
894 struct ocfs2_protocol_version pv
;
896 running_proto
.pv_major
=
897 ocfs2_user_plugin
.sp_max_proto
.pv_major
;
898 running_proto
.pv_minor
=
899 ocfs2_user_plugin
.sp_max_proto
.pv_minor
;
901 lc
->oc_version_lksb
.sb_lvbptr
= lc
->oc_lvb
;
902 ret
= version_lock(conn
, DLM_LOCK_EX
,
903 DLM_LKF_VALBLK
|DLM_LKF_NOQUEUE
);
905 conn
->cc_version
.pv_major
= running_proto
.pv_major
;
906 conn
->cc_version
.pv_minor
= running_proto
.pv_minor
;
907 version_to_lvb(&running_proto
, lc
->oc_lvb
);
908 version_lock(conn
, DLM_LOCK_PR
, DLM_LKF_CONVERT
|DLM_LKF_VALBLK
);
909 } else if (ret
== -EAGAIN
) {
910 ret
= version_lock(conn
, DLM_LOCK_PR
, DLM_LKF_VALBLK
);
913 lvb_to_version(lc
->oc_lvb
, &pv
);
915 if ((pv
.pv_major
!= running_proto
.pv_major
) ||
916 (pv
.pv_minor
> running_proto
.pv_minor
)) {
921 conn
->cc_version
.pv_major
= pv
.pv_major
;
922 conn
->cc_version
.pv_minor
= pv
.pv_minor
;
928 static void user_recover_prep(void *arg
)
932 static void user_recover_slot(void *arg
, struct dlm_slot
*slot
)
934 struct ocfs2_cluster_connection
*conn
= arg
;
935 printk(KERN_INFO
"ocfs2: Node %d/%d down. Initiating recovery.\n",
936 slot
->nodeid
, slot
->slot
);
937 conn
->cc_recovery_handler(slot
->nodeid
, conn
->cc_recovery_data
);
941 static void user_recover_done(void *arg
, struct dlm_slot
*slots
,
942 int num_slots
, int our_slot
,
945 struct ocfs2_cluster_connection
*conn
= arg
;
946 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
949 for (i
= 0; i
< num_slots
; i
++)
950 if (slots
[i
].slot
== our_slot
) {
951 atomic_set(&lc
->oc_this_node
, slots
[i
].nodeid
);
955 lc
->oc_our_slot
= our_slot
;
956 wake_up(&lc
->oc_wait
);
959 static const struct dlm_lockspace_ops ocfs2_ls_ops
= {
960 .recover_prep
= user_recover_prep
,
961 .recover_slot
= user_recover_slot
,
962 .recover_done
= user_recover_done
,
965 static int user_cluster_disconnect(struct ocfs2_cluster_connection
*conn
)
967 version_unlock(conn
);
968 dlm_release_lockspace(conn
->cc_lockspace
, 2);
969 conn
->cc_lockspace
= NULL
;
970 ocfs2_live_connection_drop(conn
->cc_private
);
971 conn
->cc_private
= NULL
;
975 static int user_cluster_connect(struct ocfs2_cluster_connection
*conn
)
977 dlm_lockspace_t
*fsdlm
;
978 struct ocfs2_live_connection
*lc
;
981 BUG_ON(conn
== NULL
);
983 lc
= kzalloc(sizeof(struct ocfs2_live_connection
), GFP_KERNEL
);
987 init_waitqueue_head(&lc
->oc_wait
);
988 init_completion(&lc
->oc_sync_wait
);
989 atomic_set(&lc
->oc_this_node
, 0);
990 conn
->cc_private
= lc
;
991 lc
->oc_type
= NO_CONTROLD
;
993 rc
= dlm_new_lockspace(conn
->cc_name
, conn
->cc_cluster_name
,
994 DLM_LSFL_FS
| DLM_LSFL_NEWEXCL
, DLM_LVB_LEN
,
995 &ocfs2_ls_ops
, conn
, &ops_rv
, &fsdlm
);
997 if (rc
== -EEXIST
|| rc
== -EPROTO
)
998 printk(KERN_ERR
"ocfs2: Unable to create the "
999 "lockspace %s (%d), because a ocfs2-tools "
1000 "program is running on this file system "
1001 "with the same name lockspace\n",
1006 if (ops_rv
== -EOPNOTSUPP
) {
1007 lc
->oc_type
= WITH_CONTROLD
;
1008 printk(KERN_NOTICE
"ocfs2: You seem to be using an older "
1009 "version of dlm_controld and/or ocfs2-tools."
1010 " Please consider upgrading.\n");
1011 } else if (ops_rv
) {
1015 conn
->cc_lockspace
= fsdlm
;
1017 rc
= ocfs2_live_connection_attach(conn
, lc
);
1021 if (lc
->oc_type
== NO_CONTROLD
) {
1022 rc
= get_protocol_version(conn
);
1024 printk(KERN_ERR
"ocfs2: Could not determine"
1025 " locking version\n");
1026 user_cluster_disconnect(conn
);
1029 wait_event(lc
->oc_wait
, (atomic_read(&lc
->oc_this_node
) > 0));
1033 * running_proto must have been set before we allowed any mounts
1036 if (fs_protocol_compare(&running_proto
, &conn
->cc_version
)) {
1038 "Unable to mount with fs locking protocol version "
1039 "%u.%u because negotiated protocol is %u.%u\n",
1040 conn
->cc_version
.pv_major
, conn
->cc_version
.pv_minor
,
1041 running_proto
.pv_major
, running_proto
.pv_minor
);
1043 ocfs2_live_connection_drop(lc
);
1054 static int user_cluster_this_node(struct ocfs2_cluster_connection
*conn
,
1055 unsigned int *this_node
)
1058 struct ocfs2_live_connection
*lc
= conn
->cc_private
;
1060 if (lc
->oc_type
== WITH_CONTROLD
)
1061 rc
= ocfs2_control_get_this_node();
1062 else if (lc
->oc_type
== NO_CONTROLD
)
1063 rc
= atomic_read(&lc
->oc_this_node
);
1074 static struct ocfs2_stack_operations ocfs2_user_plugin_ops
= {
1075 .connect
= user_cluster_connect
,
1076 .disconnect
= user_cluster_disconnect
,
1077 .this_node
= user_cluster_this_node
,
1078 .dlm_lock
= user_dlm_lock
,
1079 .dlm_unlock
= user_dlm_unlock
,
1080 .lock_status
= user_dlm_lock_status
,
1081 .lvb_valid
= user_dlm_lvb_valid
,
1082 .lock_lvb
= user_dlm_lvb
,
1083 .plock
= user_plock
,
1084 .dump_lksb
= user_dlm_dump_lksb
,
1087 static struct ocfs2_stack_plugin ocfs2_user_plugin
= {
1089 .sp_ops
= &ocfs2_user_plugin_ops
,
1090 .sp_owner
= THIS_MODULE
,
1094 static int __init
ocfs2_user_plugin_init(void)
1098 rc
= ocfs2_control_init();
1100 rc
= ocfs2_stack_glue_register(&ocfs2_user_plugin
);
1102 ocfs2_control_exit();
1108 static void __exit
ocfs2_user_plugin_exit(void)
1110 ocfs2_stack_glue_unregister(&ocfs2_user_plugin
);
1111 ocfs2_control_exit();
1114 MODULE_AUTHOR("Oracle");
1115 MODULE_DESCRIPTION("ocfs2 driver for userspace cluster stacks");
1116 MODULE_LICENSE("GPL");
1117 module_init(ocfs2_user_plugin_init
);
1118 module_exit(ocfs2_user_plugin_exit
);