[net-next-2.6.git] / fs / ocfs2 / stack_user.c

/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * stack_user.c
 *
 * Code which interfaces ocfs2 with fs/dlm and a userspace stack.
 *
 * Copyright (C) 2007 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 as published by the Free Software Foundation, version 2.
 *
 * 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.
 */

#include <linux/module.h>
#include <linux/fs.h>
#include <linux/miscdevice.h>
#include <linux/mutex.h>
#include <linux/reboot.h>
#include <asm/uaccess.h>

#include "stackglue.h"


/*
 * The control protocol starts with a handshake.  Until the handshake
 * is complete, the control device will fail all write(2)s.
 *
 * The handshake is simple.  First, the client reads until EOF.  Each line
 * of output is a supported protocol tag.  All protocol tags are a single
 * character followed by a two hex digit version number.  Currently the
 * only things supported is T01, for "Text-base version 0x01".  Next, the
 * client writes the version they would like to use, including the newline.
 * Thus, the protocol tag is 'T01\n'.  If the version tag written is
 * unknown, -EINVAL is returned.  Once the negotiation is complete, the
 * client can start sending messages.
 *
 * The T01 protocol only has two messages.  First is the "SETN" message.
 * It has the following syntax:
 *
 *  SETN<space><8-char-hex-nodenum><newline>
 *
 * This is 14 characters.
 *
 * The "SETN" message must be the first message following the protocol.
 * It tells ocfs2_control the local node number.
 *
 * Once the local node number has been set, the "DOWN" message can be
 * sent for node down notification.  It has the following syntax:
 *
 *  DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline>
 *
 * eg:
 *
 *  DOWN 632A924FDD844190BDA93C0DF6B94899 00000001\n
 *
 * This is 47 characters.
 */

/*
 * Whether or not the client has done the handshake.
 * For now, we have just one protocol version.
 */
#define OCFS2_CONTROL_PROTO			"T01\n"
#define OCFS2_CONTROL_PROTO_LEN			4

/* Handshake states */
#define OCFS2_CONTROL_HANDSHAKE_INVALID		(0)
#define OCFS2_CONTROL_HANDSHAKE_READ		(1)
#define OCFS2_CONTROL_HANDSHAKE_PROTOCOL	(2)
#define OCFS2_CONTROL_HANDSHAKE_VALID		(3)

/* Messages */
#define OCFS2_CONTROL_MESSAGE_OP_LEN		4
#define OCFS2_CONTROL_MESSAGE_SETNODE_OP	"SETN"
#define OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN	14
#define OCFS2_CONTROL_MESSAGE_DOWN_OP		"DOWN"
#define OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN	47
#define OCFS2_TEXT_UUID_LEN			32
#define OCFS2_CONTROL_MESSAGE_NODENUM_LEN	8

/*
 * ocfs2_live_connection is refcounted because the filesystem and
 * miscdevice sides can detach in different order.  Let's just be safe.
 */
struct ocfs2_live_connection {
	struct list_head		oc_list;
	struct ocfs2_cluster_connection	*oc_conn;
};

struct ocfs2_control_private {
	struct list_head op_list;
	int op_state;
	int op_this_node;
};

/* SETN<space><8-char-hex-nodenum><newline> */
struct ocfs2_control_message_setn {
	char	tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
	char	space;
	char	nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
	char	newline;
};

/* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> */
struct ocfs2_control_message_down {
	char	tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
	char	space1;
	char	uuid[OCFS2_TEXT_UUID_LEN];
	char	space2;
	char	nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
	char	newline;
};

union ocfs2_control_message {
	char					tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
	struct ocfs2_control_message_setn	u_setn;
	struct ocfs2_control_message_down	u_down;
};

static atomic_t ocfs2_control_opened;
static int ocfs2_control_this_node = -1;

static LIST_HEAD(ocfs2_live_connection_list);
static LIST_HEAD(ocfs2_control_private_list);
static DEFINE_MUTEX(ocfs2_control_lock);

static inline void ocfs2_control_set_handshake_state(struct file *file,
						     int state)
{
	struct ocfs2_control_private *p = file->private_data;
	p->op_state = state;
}

static inline int ocfs2_control_get_handshake_state(struct file *file)
{
	struct ocfs2_control_private *p = file->private_data;
	return p->op_state;
}

static struct ocfs2_live_connection *ocfs2_connection_find(const char *name)
{
	size_t len = strlen(name);
	struct ocfs2_live_connection *c;

	BUG_ON(!mutex_is_locked(&ocfs2_control_lock));

	list_for_each_entry(c, &ocfs2_live_connection_list, oc_list) {
		if ((c->oc_conn->cc_namelen == len) &&
		    !strncmp(c->oc_conn->cc_name, name, len))
			return c;
	}

	return c;
}

/*
 * ocfs2_live_connection structures are created underneath the ocfs2
 * mount path.  Since the VFS prevents multiple calls to
 * fill_super(), we can't get dupes here.
 */
static int ocfs2_live_connection_new(struct ocfs2_cluster_connection *conn,
				     struct ocfs2_live_connection **c_ret)
{
	int rc = 0;
	struct ocfs2_live_connection *c;

	c = kzalloc(sizeof(struct ocfs2_live_connection), GFP_KERNEL);
	if (!c)
		return -ENOMEM;

	mutex_lock(&ocfs2_control_lock);
	c->oc_conn = conn;

	if (atomic_read(&ocfs2_control_opened))
		list_add(&c->oc_list, &ocfs2_live_connection_list);
	else {
		printk(KERN_ERR
		       "ocfs2: Userspace control daemon is not present\n");
		rc = -ESRCH;
	}

	mutex_unlock(&ocfs2_control_lock);

	if (!rc)
		*c_ret = c;
	else
		kfree(c);

	return rc;
}

/*
 * This function disconnects the cluster connection from ocfs2_control.
 * Afterwards, userspace can't affect the cluster connection.
 */
static void ocfs2_live_connection_drop(struct ocfs2_live_connection *c)
{
	mutex_lock(&ocfs2_control_lock);
	list_del_init(&c->oc_list);
	c->oc_conn = NULL;
	mutex_unlock(&ocfs2_control_lock);

	kfree(c);
}

static int ocfs2_control_cfu(void *target, size_t target_len,
			     const char __user *buf, size_t count)
{
	/* The T01 expects write(2) calls to have exactly one command */
	if ((count != target_len) ||
	    (count > sizeof(union ocfs2_control_message)))
		return -EINVAL;

	if (copy_from_user(target, buf, target_len))
		return -EFAULT;

	return 0;
}

static ssize_t ocfs2_control_validate_protocol(struct file *file,
					       const char __user *buf,
					       size_t count)
{
	ssize_t ret;
	char kbuf[OCFS2_CONTROL_PROTO_LEN];

	ret = ocfs2_control_cfu(kbuf, OCFS2_CONTROL_PROTO_LEN,
				buf, count);
	if (ret)
		return ret;

	if (strncmp(kbuf, OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN))
		return -EINVAL;

	ocfs2_control_set_handshake_state(file,
					  OCFS2_CONTROL_HANDSHAKE_PROTOCOL);

	return count;
}

static void ocfs2_control_send_down(const char *uuid,
				    int nodenum)
{
	struct ocfs2_live_connection *c;

	mutex_lock(&ocfs2_control_lock);

	c = ocfs2_connection_find(uuid);
	if (c) {
		BUG_ON(c->oc_conn == NULL);
		c->oc_conn->cc_recovery_handler(nodenum,
						c->oc_conn->cc_recovery_data);
	}

	mutex_unlock(&ocfs2_control_lock);
}

/*
 * Called whenever configuration elements are sent to /dev/ocfs2_control.
 * If all configuration elements are present, try to set the global
 * values.  If not, return -EAGAIN.  If there is a problem, return a
 * different error.
 */
static int ocfs2_control_install_private(struct file *file)
{
	int rc = 0;
	int set_p = 1;
	struct ocfs2_control_private *p = file->private_data;

	BUG_ON(p->op_state != OCFS2_CONTROL_HANDSHAKE_PROTOCOL);

	if (p->op_this_node < 0)
		set_p = 0;

	mutex_lock(&ocfs2_control_lock);
	if (ocfs2_control_this_node < 0) {
		if (set_p)
			ocfs2_control_this_node = p->op_this_node;
	} else if (ocfs2_control_this_node != p->op_this_node)
		rc = -EINVAL;
	mutex_unlock(&ocfs2_control_lock);

	if (!rc && set_p) {
		/* We set the global values successfully */
		atomic_inc(&ocfs2_control_opened);
		ocfs2_control_set_handshake_state(file,
					OCFS2_CONTROL_HANDSHAKE_VALID);
	}

	return rc;
}

static int ocfs2_control_do_setnode_msg(struct file *file,
					struct ocfs2_control_message_setn *msg)
{
	long nodenum;
	char *ptr = NULL;
	struct ocfs2_control_private *p = file->private_data;

	if (ocfs2_control_get_handshake_state(file) !=
	    OCFS2_CONTROL_HANDSHAKE_PROTOCOL)
		return -EINVAL;

	if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
		    OCFS2_CONTROL_MESSAGE_OP_LEN))
		return -EINVAL;

	if ((msg->space != ' ') || (msg->newline != '\n'))
		return -EINVAL;
	msg->space = msg->newline = '\0';

	nodenum = simple_strtol(msg->nodestr, &ptr, 16);
	if (!ptr || *ptr)
		return -EINVAL;

	if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) ||
	    (nodenum > INT_MAX) || (nodenum < 0))
		return -ERANGE;
	p->op_this_node = nodenum;

	return ocfs2_control_install_private(file);
}

static int ocfs2_control_do_down_msg(struct file *file,
				     struct ocfs2_control_message_down *msg)
{
	long nodenum;
	char *p = NULL;

	if (ocfs2_control_get_handshake_state(file) !=
	    OCFS2_CONTROL_HANDSHAKE_VALID)
		return -EINVAL;

	if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
		    OCFS2_CONTROL_MESSAGE_OP_LEN))
		return -EINVAL;

	if ((msg->space1 != ' ') || (msg->space2 != ' ') ||
	    (msg->newline != '\n'))
		return -EINVAL;
	msg->space1 = msg->space2 = msg->newline = '\0';

	nodenum = simple_strtol(msg->nodestr, &p, 16);
	if (!p || *p)
		return -EINVAL;

	if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) ||
	    (nodenum > INT_MAX) || (nodenum < 0))
		return -ERANGE;

	ocfs2_control_send_down(msg->uuid, nodenum);

	return 0;
}

static ssize_t ocfs2_control_message(struct file *file,
				     const char __user *buf,
				     size_t count)
{
	ssize_t ret;
	union ocfs2_control_message msg;

	/* Try to catch padding issues */
	WARN_ON(offsetof(struct ocfs2_control_message_down, uuid) !=
		(sizeof(msg.u_down.tag) + sizeof(msg.u_down.space1)));

	memset(&msg, 0, sizeof(union ocfs2_control_message));
	ret = ocfs2_control_cfu(&msg, count, buf, count);
	if (ret)
		goto out;

	if ((count == OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN) &&
	    !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
		     OCFS2_CONTROL_MESSAGE_OP_LEN))
		ret = ocfs2_control_do_setnode_msg(file, &msg.u_setn);
	else if ((count == OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN) &&
		 !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
			  OCFS2_CONTROL_MESSAGE_OP_LEN))
		ret = ocfs2_control_do_down_msg(file, &msg.u_down);
	else
		ret = -EINVAL;

out:
	return ret ? ret : count;
}

static ssize_t ocfs2_control_write(struct file *file,
				   const char __user *buf,
				   size_t count,
				   loff_t *ppos)
{
	ssize_t ret;

	switch (ocfs2_control_get_handshake_state(file)) {
		case OCFS2_CONTROL_HANDSHAKE_INVALID:
			ret = -EINVAL;
			break;

		case OCFS2_CONTROL_HANDSHAKE_READ:
			ret = ocfs2_control_validate_protocol(file, buf,
							      count);
			break;

		case OCFS2_CONTROL_HANDSHAKE_PROTOCOL:
		case OCFS2_CONTROL_HANDSHAKE_VALID:
			ret = ocfs2_control_message(file, buf, count);
			break;

		default:
			BUG();
			ret = -EIO;
			break;
	}

	return ret;
}

/*
 * This is a naive version.  If we ever have a new protocol, we'll expand
 * it.  Probably using seq_file.
 */
static ssize_t ocfs2_control_read(struct file *file,
				  char __user *buf,
				  size_t count,
				  loff_t *ppos)
{
	char *proto_string = OCFS2_CONTROL_PROTO;
	size_t to_write = 0;

	if (*ppos >= OCFS2_CONTROL_PROTO_LEN)
		return 0;

	to_write = OCFS2_CONTROL_PROTO_LEN - *ppos;
	if (to_write > count)
		to_write = count;
	if (copy_to_user(buf, proto_string + *ppos, to_write))
		return -EFAULT;

	*ppos += to_write;

	/* Have we read the whole protocol list? */
	if (*ppos >= OCFS2_CONTROL_PROTO_LEN)
		ocfs2_control_set_handshake_state(file,
						  OCFS2_CONTROL_HANDSHAKE_READ);

	return to_write;
}

static int ocfs2_control_release(struct inode *inode, struct file *file)
{
	struct ocfs2_control_private *p = file->private_data;

	mutex_lock(&ocfs2_control_lock);

	if (ocfs2_control_get_handshake_state(file) !=
	    OCFS2_CONTROL_HANDSHAKE_VALID)
		goto out;

	if (atomic_dec_and_test(&ocfs2_control_opened)) {
		if (!list_empty(&ocfs2_live_connection_list)) {
			/* XXX: Do bad things! */
			printk(KERN_ERR
			       "ocfs2: Unexpected release of ocfs2_control!\n"
			       "       Loss of cluster connection requires "
			       "an emergency restart!\n");
			emergency_restart();
		}
		/* Last valid close clears the node number */
		ocfs2_control_this_node = -1;
	}

out:
	list_del_init(&p->op_list);
	file->private_data = NULL;

	mutex_unlock(&ocfs2_control_lock);

	kfree(p);

	return 0;
}

static int ocfs2_control_open(struct inode *inode, struct file *file)
{
	struct ocfs2_control_private *p;

	p = kzalloc(sizeof(struct ocfs2_control_private), GFP_KERNEL);
	if (!p)
		return -ENOMEM;
	p->op_this_node = -1;

	mutex_lock(&ocfs2_control_lock);
	file->private_data = p;
	list_add(&p->op_list, &ocfs2_control_private_list);
	mutex_unlock(&ocfs2_control_lock);

	return 0;
}

static const struct file_operations ocfs2_control_fops = {
	.open    = ocfs2_control_open,
	.release = ocfs2_control_release,
	.read    = ocfs2_control_read,
	.write   = ocfs2_control_write,
	.owner   = THIS_MODULE,
};

struct miscdevice ocfs2_control_device = {
	.minor		= MISC_DYNAMIC_MINOR,
	.name		= "ocfs2_control",
	.fops		= &ocfs2_control_fops,
};

static int ocfs2_control_init(void)
{
	int rc;

	atomic_set(&ocfs2_control_opened, 0);

	rc = misc_register(&ocfs2_control_device);
	if (rc)
		printk(KERN_ERR
		       "ocfs2: Unable to register ocfs2_control device "
		       "(errno %d)\n",
		       -rc);

	return rc;
}

static void ocfs2_control_exit(void)
{
	int rc;

	rc = misc_deregister(&ocfs2_control_device);
	if (rc)
		printk(KERN_ERR
		       "ocfs2: Unable to deregister ocfs2_control device "
		       "(errno %d)\n",
		       -rc);
}

static int __init user_stack_init(void)
{
	return ocfs2_control_init();
}

static void __exit user_stack_exit(void)
{
	ocfs2_control_exit();
}

MODULE_AUTHOR("Oracle");
MODULE_DESCRIPTION("ocfs2 driver for userspace cluster stacks");
MODULE_LICENSE("GPL");
module_init(user_stack_init);
module_exit(user_stack_exit);
Commit	Line	Data
8adf0536 JB	1	/* -- mode: c; c-basic-offset: 8; --
	2	* vim: noexpandtab sw=8 ts=8 sts=0:
	3	*
	4	* stack_user.c
	5	*
	6	* Code which interfaces ocfs2 with fs/dlm and a userspace stack.
	7	*
	8	* Copyright (C) 2007 Oracle. All rights reserved.
	9	*
	10	* This program is free software; you can redistribute it and/or
	11	* modify it under the terms of the GNU General Public
	12	* License as published by the Free Software Foundation, version 2.
	13	*
	14	* This program is distributed in the hope that it will be useful,
	15	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	16	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	17	* General Public License for more details.
	18	*/
	19
	20	#include <linux/module.h>
6427a727 JB	21	#include <linux/fs.h>
	22	#include <linux/miscdevice.h>
	23	#include <linux/mutex.h>
	24	#include <linux/reboot.h>
462c7e6a	25	#include <asm/uaccess.h>
8adf0536 JB	26
	27	#include "stackglue.h"
	28
	29
6427a727 JB	30	/*
	31	* The control protocol starts with a handshake. Until the handshake
	32	* is complete, the control device will fail all write(2)s.
	33	*
	34	* The handshake is simple. First, the client reads until EOF. Each line
	35	* of output is a supported protocol tag. All protocol tags are a single
	36	* character followed by a two hex digit version number. Currently the
	37	* only things supported is T01, for "Text-base version 0x01". Next, the
de870ef0 JB	38	* client writes the version they would like to use, including the newline.
	39	* Thus, the protocol tag is 'T01\n'. If the version tag written is
	40	* unknown, -EINVAL is returned. Once the negotiation is complete, the
	41	* client can start sending messages.
	42	*
3cfd4ab6 JB	43	* The T01 protocol only has two messages. First is the "SETN" message.
	44	* It has the following syntax:
	45	*
	46	* SETN<space><8-char-hex-nodenum><newline>
	47	*
	48	* This is 14 characters.
	49	*
	50	* The "SETN" message must be the first message following the protocol.
	51	* It tells ocfs2_control the local node number.
	52	*
	53	* Once the local node number has been set, the "DOWN" message can be
	54	* sent for node down notification. It has the following syntax:
de870ef0 JB	55	*
	56	* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline>
	57	*
	58	* eg:
	59	*
	60	* DOWN 632A924FDD844190BDA93C0DF6B94899 00000001\n
	61	*
	62	* This is 47 characters.
6427a727 JB	63	*/
6427a727 JB	64
462c7e6a JB	65	/*
	66	* Whether or not the client has done the handshake.
	67	* For now, we have just one protocol version.
	68	*/
	69	#define OCFS2_CONTROL_PROTO "T01\n"
	70	#define OCFS2_CONTROL_PROTO_LEN 4
3cfd4ab6 JB	71
3cfd4ab6 JB	72	/* Handshake states */
462c7e6a JB	73	#define OCFS2_CONTROL_HANDSHAKE_INVALID (0)
462c7e6a JB	74	#define OCFS2_CONTROL_HANDSHAKE_READ (1)
3cfd4ab6 JB	75	#define OCFS2_CONTROL_HANDSHAKE_PROTOCOL (2)
	76	#define OCFS2_CONTROL_HANDSHAKE_VALID (3)
	77
	78	/* Messages */
	79	#define OCFS2_CONTROL_MESSAGE_OP_LEN 4
	80	#define OCFS2_CONTROL_MESSAGE_SETNODE_OP "SETN"
	81	#define OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN 14
	82	#define OCFS2_CONTROL_MESSAGE_DOWN_OP "DOWN"
de870ef0 JB	83	#define OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN 47
	84	#define OCFS2_TEXT_UUID_LEN 32
	85	#define OCFS2_CONTROL_MESSAGE_NODENUM_LEN 8
462c7e6a	86
6427a727 JB	87	/*
	88	* ocfs2_live_connection is refcounted because the filesystem and
	89	* miscdevice sides can detach in different order. Let's just be safe.
	90	*/
	91	struct ocfs2_live_connection {
	92	struct list_head oc_list;
	93	struct ocfs2_cluster_connection *oc_conn;
	94	};
	95
462c7e6a JB	96	struct ocfs2_control_private {
	97	struct list_head op_list;
	98	int op_state;
3cfd4ab6 JB	99	int op_this_node;
	100	};
	101
	102	/* SETN<space><8-char-hex-nodenum><newline> */
	103	struct ocfs2_control_message_setn {
	104	char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
	105	char space;
	106	char nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
	107	char newline;
	108	};
	109
	110	/* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> */
	111	struct ocfs2_control_message_down {
	112	char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
	113	char space1;
	114	char uuid[OCFS2_TEXT_UUID_LEN];
	115	char space2;
	116	char nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
	117	char newline;
	118	};
	119
	120	union ocfs2_control_message {
	121	char tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
	122	struct ocfs2_control_message_setn u_setn;
	123	struct ocfs2_control_message_down u_down;
462c7e6a JB	124	};
462c7e6a JB	125
6427a727	126	static atomic_t ocfs2_control_opened;
3cfd4ab6	127	static int ocfs2_control_this_node = -1;
6427a727 JB	128
6427a727 JB	129	static LIST_HEAD(ocfs2_live_connection_list);
462c7e6a	130	static LIST_HEAD(ocfs2_control_private_list);
6427a727 JB	131	static DEFINE_MUTEX(ocfs2_control_lock);
6427a727 JB	132
462c7e6a JB	133	static inline void ocfs2_control_set_handshake_state(struct file *file,
	134	int state)
	135	{
	136	struct ocfs2_control_private *p = file->private_data;
	137	p->op_state = state;
	138	}
	139
	140	static inline int ocfs2_control_get_handshake_state(struct file *file)
	141	{
	142	struct ocfs2_control_private *p = file->private_data;
	143	return p->op_state;
	144	}
	145
6427a727 JB	146	static struct ocfs2_live_connection ocfs2_connection_find(const char name)
	147	{
	148	size_t len = strlen(name);
	149	struct ocfs2_live_connection *c;
	150
	151	BUG_ON(!mutex_is_locked(&ocfs2_control_lock));
	152
	153	list_for_each_entry(c, &ocfs2_live_connection_list, oc_list) {
	154	if ((c->oc_conn->cc_namelen == len) &&
	155	!strncmp(c->oc_conn->cc_name, name, len))
	156	return c;
	157	}
	158
	159	return c;
	160	}
	161
	162	/*
	163	* ocfs2_live_connection structures are created underneath the ocfs2
	164	* mount path. Since the VFS prevents multiple calls to
	165	* fill_super(), we can't get dupes here.
	166	*/
	167	static int ocfs2_live_connection_new(struct ocfs2_cluster_connection *conn,
	168	struct ocfs2_live_connection **c_ret)
	169	{
	170	int rc = 0;
	171	struct ocfs2_live_connection *c;
	172
	173	c = kzalloc(sizeof(struct ocfs2_live_connection), GFP_KERNEL);
	174	if (!c)
	175	return -ENOMEM;
	176
	177	mutex_lock(&ocfs2_control_lock);
	178	c->oc_conn = conn;
	179
	180	if (atomic_read(&ocfs2_control_opened))
	181	list_add(&c->oc_list, &ocfs2_live_connection_list);
	182	else {
	183	printk(KERN_ERR
	184	"ocfs2: Userspace control daemon is not present\n");
	185	rc = -ESRCH;
	186	}
	187
	188	mutex_unlock(&ocfs2_control_lock);
	189
	190	if (!rc)
	191	*c_ret = c;
	192	else
	193	kfree(c);
	194
	195	return rc;
	196	}
	197
	198	/*
	199	* This function disconnects the cluster connection from ocfs2_control.
	200	* Afterwards, userspace can't affect the cluster connection.
	201	*/
	202	static void ocfs2_live_connection_drop(struct ocfs2_live_connection *c)
	203	{
	204	mutex_lock(&ocfs2_control_lock);
	205	list_del_init(&c->oc_list);
	206	c->oc_conn = NULL;
	207	mutex_unlock(&ocfs2_control_lock);
	208
	209	kfree(c);
210	}
211
3cfd4ab6 JB	212	static int ocfs2_control_cfu(void *target, size_t target_len,
3cfd4ab6 JB	213	const char __user *buf, size_t count)
462c7e6a JB	214	{
462c7e6a JB	215	/* The T01 expects write(2) calls to have exactly one command */
3cfd4ab6 JB	216	if ((count != target_len) \|\|
3cfd4ab6 JB	217	(count > sizeof(union ocfs2_control_message)))
462c7e6a JB	218	return -EINVAL;
	219
	220	if (copy_from_user(target, buf, target_len))
	221	return -EFAULT;
	222
3cfd4ab6	223	return 0;
462c7e6a JB	224	}
462c7e6a JB	225
3cfd4ab6 JB	226	static ssize_t ocfs2_control_validate_protocol(struct file *file,
	227	const char __user *buf,
	228	size_t count)
462c7e6a JB	229	{
	230	ssize_t ret;
	231	char kbuf[OCFS2_CONTROL_PROTO_LEN];
	232
	233	ret = ocfs2_control_cfu(kbuf, OCFS2_CONTROL_PROTO_LEN,
	234	buf, count);
3cfd4ab6	235	if (ret)
462c7e6a JB	236	return ret;
	237
	238	if (strncmp(kbuf, OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN))
	239	return -EINVAL;
	240
462c7e6a	241	ocfs2_control_set_handshake_state(file,
3cfd4ab6	242	OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
462c7e6a JB	243
	244	return count;
	245	}
	246
de870ef0 JB	247	static void ocfs2_control_send_down(const char *uuid,
	248	int nodenum)
	249	{
	250	struct ocfs2_live_connection *c;
	251
	252	mutex_lock(&ocfs2_control_lock);
	253
	254	c = ocfs2_connection_find(uuid);
	255	if (c) {
	256	BUG_ON(c->oc_conn == NULL);
	257	c->oc_conn->cc_recovery_handler(nodenum,
	258	c->oc_conn->cc_recovery_data);
	259	}
	260
	261	mutex_unlock(&ocfs2_control_lock);
	262	}
	263
3cfd4ab6 JB	264	/*
	265	* Called whenever configuration elements are sent to /dev/ocfs2_control.
	266	* If all configuration elements are present, try to set the global
	267	* values. If not, return -EAGAIN. If there is a problem, return a
	268	* different error.
	269	*/
	270	static int ocfs2_control_install_private(struct file *file)
	271	{
	272	int rc = 0;
	273	int set_p = 1;
	274	struct ocfs2_control_private *p = file->private_data;
de870ef0	275
3cfd4ab6 JB	276	BUG_ON(p->op_state != OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
	277
	278	if (p->op_this_node < 0)
	279	set_p = 0;
	280
	281	mutex_lock(&ocfs2_control_lock);
	282	if (ocfs2_control_this_node < 0) {
	283	if (set_p)
	284	ocfs2_control_this_node = p->op_this_node;
	285	} else if (ocfs2_control_this_node != p->op_this_node)
	286	rc = -EINVAL;
	287	mutex_unlock(&ocfs2_control_lock);
	288
	289	if (!rc && set_p) {
	290	/* We set the global values successfully */
	291	atomic_inc(&ocfs2_control_opened);
	292	ocfs2_control_set_handshake_state(file,
	293	OCFS2_CONTROL_HANDSHAKE_VALID);
	294	}
	295
	296	return rc;
	297	}
	298
	299	static int ocfs2_control_do_setnode_msg(struct file *file,
	300	struct ocfs2_control_message_setn *msg)
de870ef0	301	{
de870ef0	302	long nodenum;
3cfd4ab6 JB	303	char *ptr = NULL;
3cfd4ab6 JB	304	struct ocfs2_control_private *p = file->private_data;
de870ef0	305
3cfd4ab6 JB	306	if (ocfs2_control_get_handshake_state(file) !=
	307	OCFS2_CONTROL_HANDSHAKE_PROTOCOL)
	308	return -EINVAL;
de870ef0	309
3cfd4ab6 JB	310	if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
	311	OCFS2_CONTROL_MESSAGE_OP_LEN))
	312	return -EINVAL;
	313
	314	if ((msg->space != ' ') \|\| (msg->newline != '\n'))
	315	return -EINVAL;
	316	msg->space = msg->newline = '\0';
	317
	318	nodenum = simple_strtol(msg->nodestr, &ptr, 16);
	319	if (!ptr \|\| *ptr)
	320	return -EINVAL;
	321
	322	if ((nodenum == LONG_MIN) \|\| (nodenum == LONG_MAX) \|\|
	323	(nodenum > INT_MAX) \|\| (nodenum < 0))
	324	return -ERANGE;
	325	p->op_this_node = nodenum;
	326
	327	return ocfs2_control_install_private(file);
	328	}
	329
	330	static int ocfs2_control_do_down_msg(struct file *file,
	331	struct ocfs2_control_message_down *msg)
	332	{
	333	long nodenum;
	334	char *p = NULL;
	335
	336	if (ocfs2_control_get_handshake_state(file) !=
	337	OCFS2_CONTROL_HANDSHAKE_VALID)
	338	return -EINVAL;
de870ef0	339
3cfd4ab6 JB	340	if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
3cfd4ab6 JB	341	OCFS2_CONTROL_MESSAGE_OP_LEN))
de870ef0 JB	342	return -EINVAL;
de870ef0 JB	343
3cfd4ab6 JB	344	if ((msg->space1 != ' ') \|\| (msg->space2 != ' ') \|\|
3cfd4ab6 JB	345	(msg->newline != '\n'))
de870ef0	346	return -EINVAL;
3cfd4ab6	347	msg->space1 = msg->space2 = msg->newline = '\0';
de870ef0	348
3cfd4ab6	349	nodenum = simple_strtol(msg->nodestr, &p, 16);
de870ef0 JB	350	if (!p \|\| *p)
	351	return -EINVAL;
	352
	353	if ((nodenum == LONG_MIN) \|\| (nodenum == LONG_MAX) \|\|
	354	(nodenum > INT_MAX) \|\| (nodenum < 0))
	355	return -ERANGE;
	356
3cfd4ab6	357	ocfs2_control_send_down(msg->uuid, nodenum);
de870ef0	358
3cfd4ab6 JB	359	return 0;
	360	}
	361
	362	static ssize_t ocfs2_control_message(struct file *file,
	363	const char __user *buf,
	364	size_t count)
	365	{
	366	ssize_t ret;
	367	union ocfs2_control_message msg;
	368
	369	/* Try to catch padding issues */
	370	WARN_ON(offsetof(struct ocfs2_control_message_down, uuid) !=
	371	(sizeof(msg.u_down.tag) + sizeof(msg.u_down.space1)));
	372
	373	memset(&msg, 0, sizeof(union ocfs2_control_message));
	374	ret = ocfs2_control_cfu(&msg, count, buf, count);
	375	if (ret)
	376	goto out;
	377
	378	if ((count == OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN) &&
	379	!strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
	380	OCFS2_CONTROL_MESSAGE_OP_LEN))
	381	ret = ocfs2_control_do_setnode_msg(file, &msg.u_setn);
	382	else if ((count == OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN) &&
	383	!strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
	384	OCFS2_CONTROL_MESSAGE_OP_LEN))
	385	ret = ocfs2_control_do_down_msg(file, &msg.u_down);
	386	else
	387	ret = -EINVAL;
	388
	389	out:
	390	return ret ? ret : count;
de870ef0	391	}
6427a727 JB	392
	393	static ssize_t ocfs2_control_write(struct file *file,
	394	const char __user *buf,
	395	size_t count,
	396	loff_t *ppos)
8adf0536	397	{
462c7e6a JB	398	ssize_t ret;
	399
	400	switch (ocfs2_control_get_handshake_state(file)) {
	401	case OCFS2_CONTROL_HANDSHAKE_INVALID:
	402	ret = -EINVAL;
	403	break;
	404
	405	case OCFS2_CONTROL_HANDSHAKE_READ:
3cfd4ab6 JB	406	ret = ocfs2_control_validate_protocol(file, buf,
3cfd4ab6 JB	407	count);
462c7e6a JB	408	break;
462c7e6a JB	409
3cfd4ab6	410	case OCFS2_CONTROL_HANDSHAKE_PROTOCOL:
462c7e6a	411	case OCFS2_CONTROL_HANDSHAKE_VALID:
de870ef0	412	ret = ocfs2_control_message(file, buf, count);
462c7e6a JB	413	break;
	414
	415	default:
	416	BUG();
	417	ret = -EIO;
	418	break;
	419	}
	420
	421	return ret;
8adf0536 JB	422	}
8adf0536 JB	423
462c7e6a JB	424	/*
	425	* This is a naive version. If we ever have a new protocol, we'll expand
	426	* it. Probably using seq_file.
	427	*/
6427a727 JB	428	static ssize_t ocfs2_control_read(struct file *file,
	429	char __user *buf,
	430	size_t count,
	431	loff_t *ppos)
	432	{
462c7e6a JB	433	char *proto_string = OCFS2_CONTROL_PROTO;
	434	size_t to_write = 0;
	435
	436	if (*ppos >= OCFS2_CONTROL_PROTO_LEN)
	437	return 0;
	438
	439	to_write = OCFS2_CONTROL_PROTO_LEN - *ppos;
	440	if (to_write > count)
	441	to_write = count;
	442	if (copy_to_user(buf, proto_string + *ppos, to_write))
	443	return -EFAULT;
	444
	445	*ppos += to_write;
	446
	447	/* Have we read the whole protocol list? */
	448	if (*ppos >= OCFS2_CONTROL_PROTO_LEN)
	449	ocfs2_control_set_handshake_state(file,
	450	OCFS2_CONTROL_HANDSHAKE_READ);
	451
	452	return to_write;
6427a727 JB	453	}
	454
	455	static int ocfs2_control_release(struct inode inode, struct file file)
	456	{
462c7e6a JB	457	struct ocfs2_control_private *p = file->private_data;
	458
	459	mutex_lock(&ocfs2_control_lock);
	460
	461	if (ocfs2_control_get_handshake_state(file) !=
	462	OCFS2_CONTROL_HANDSHAKE_VALID)
	463	goto out;
	464
6427a727	465	if (atomic_dec_and_test(&ocfs2_control_opened)) {
6427a727 JB	466	if (!list_empty(&ocfs2_live_connection_list)) {
	467	/* XXX: Do bad things! */
	468	printk(KERN_ERR
	469	"ocfs2: Unexpected release of ocfs2_control!\n"
	470	" Loss of cluster connection requires "
	471	"an emergency restart!\n");
	472	emergency_restart();
	473	}
3cfd4ab6 JB	474	/* Last valid close clears the node number */
3cfd4ab6 JB	475	ocfs2_control_this_node = -1;
6427a727 JB	476	}
6427a727 JB	477
462c7e6a JB	478	out:
	479	list_del_init(&p->op_list);
	480	file->private_data = NULL;
	481
	482	mutex_unlock(&ocfs2_control_lock);
	483
	484	kfree(p);
	485
6427a727 JB	486	return 0;
	487	}
	488
	489	static int ocfs2_control_open(struct inode inode, struct file file)
	490	{
462c7e6a JB	491	struct ocfs2_control_private *p;
	492
	493	p = kzalloc(sizeof(struct ocfs2_control_private), GFP_KERNEL);
	494	if (!p)
	495	return -ENOMEM;
3cfd4ab6	496	p->op_this_node = -1;
462c7e6a JB	497
	498	mutex_lock(&ocfs2_control_lock);
	499	file->private_data = p;
	500	list_add(&p->op_list, &ocfs2_control_private_list);
	501	mutex_unlock(&ocfs2_control_lock);
6427a727 JB	502
	503	return 0;
	504	}
	505
	506	static const struct file_operations ocfs2_control_fops = {
	507	.open = ocfs2_control_open,
	508	.release = ocfs2_control_release,
	509	.read = ocfs2_control_read,
	510	.write = ocfs2_control_write,
	511	.owner = THIS_MODULE,
	512	};
	513
	514	struct miscdevice ocfs2_control_device = {
	515	.minor = MISC_DYNAMIC_MINOR,
	516	.name = "ocfs2_control",
	517	.fops = &ocfs2_control_fops,
	518	};
	519
	520	static int ocfs2_control_init(void)
	521	{
	522	int rc;
	523
	524	atomic_set(&ocfs2_control_opened, 0);
	525
	526	rc = misc_register(&ocfs2_control_device);
	527	if (rc)
	528	printk(KERN_ERR
	529	"ocfs2: Unable to register ocfs2_control device "
	530	"(errno %d)\n",
	531	-rc);
	532
	533	return rc;
	534	}
	535
	536	static void ocfs2_control_exit(void)
	537	{
	538	int rc;
	539
	540	rc = misc_deregister(&ocfs2_control_device);
	541	if (rc)
	542	printk(KERN_ERR
	543	"ocfs2: Unable to deregister ocfs2_control device "
	544	"(errno %d)\n",
	545	-rc);
	546	}
	547
	548	static int __init user_stack_init(void)
	549	{
	550	return ocfs2_control_init();
	551	}
	552
8adf0536 JB	553	static void __exit user_stack_exit(void)
8adf0536 JB	554	{
6427a727	555	ocfs2_control_exit();
8adf0536 JB	556	}
	557
	558	MODULE_AUTHOR("Oracle");
	559	MODULE_DESCRIPTION("ocfs2 driver for userspace cluster stacks");
	560	MODULE_LICENSE("GPL");
	561	module_init(user_stack_init);
	562	module_exit(user_stack_exit);