[deliverable/linux.git] / fs / dlm / lockspace.c

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2004-2005 Red Hat, Inc.  All rights reserved.
**
**  This copyrighted material is made available to anyone wishing to use,
**  modify, copy, or redistribute it subject to the terms and conditions
**  of the GNU General Public License v.2.
**
*******************************************************************************
******************************************************************************/

#include "dlm_internal.h"
#include "lockspace.h"
#include "member.h"
#include "recoverd.h"
#include "ast.h"
#include "dir.h"
#include "lowcomms.h"
#include "config.h"
#include "memory.h"
#include "lock.h"
#include "recover.h"

#ifdef CONFIG_DLM_DEBUG
int dlm_create_debug_file(struct dlm_ls *ls);
void dlm_delete_debug_file(struct dlm_ls *ls);
#else
static inline int dlm_create_debug_file(struct dlm_ls *ls) { return 0; }
static inline void dlm_delete_debug_file(struct dlm_ls *ls) { }
#endif

static int			ls_count;
static struct mutex		ls_lock;
static struct list_head		lslist;
static spinlock_t		lslist_lock;
static struct task_struct *	scand_task;


static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
{
	ssize_t ret = len;
	int n = simple_strtol(buf, NULL, 0);

	switch (n) {
	case 0:
		dlm_ls_stop(ls);
		break;
	case 1:
		dlm_ls_start(ls);
		break;
	default:
		ret = -EINVAL;
	}
	return ret;
}

static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
{
	ls->ls_uevent_result = simple_strtol(buf, NULL, 0);
	set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
	wake_up(&ls->ls_uevent_wait);
	return len;
}

static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
{
	return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
}

static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
{
	ls->ls_global_id = simple_strtoul(buf, NULL, 0);
	return len;
}

static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
{
	uint32_t status = dlm_recover_status(ls);
	return snprintf(buf, PAGE_SIZE, "%x\n", status);
}

static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
{
	return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
}

struct dlm_attr {
	struct attribute attr;
	ssize_t (*show)(struct dlm_ls *, char *);
	ssize_t (*store)(struct dlm_ls *, const char *, size_t);
};

static struct dlm_attr dlm_attr_control = {
	.attr  = {.name = "control", .mode = S_IWUSR},
	.store = dlm_control_store
};

static struct dlm_attr dlm_attr_event = {
	.attr  = {.name = "event_done", .mode = S_IWUSR},
	.store = dlm_event_store
};

static struct dlm_attr dlm_attr_id = {
	.attr  = {.name = "id", .mode = S_IRUGO | S_IWUSR},
	.show  = dlm_id_show,
	.store = dlm_id_store
};

static struct dlm_attr dlm_attr_recover_status = {
	.attr  = {.name = "recover_status", .mode = S_IRUGO},
	.show  = dlm_recover_status_show
};

static struct dlm_attr dlm_attr_recover_nodeid = {
	.attr  = {.name = "recover_nodeid", .mode = S_IRUGO},
	.show  = dlm_recover_nodeid_show
};

static struct attribute *dlm_attrs[] = {
	&dlm_attr_control.attr,
	&dlm_attr_event.attr,
	&dlm_attr_id.attr,
	&dlm_attr_recover_status.attr,
	&dlm_attr_recover_nodeid.attr,
	NULL,
};

static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
			     char *buf)
{
	struct dlm_ls *ls  = container_of(kobj, struct dlm_ls, ls_kobj);
	struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
	return a->show ? a->show(ls, buf) : 0;
}

static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
			      const char *buf, size_t len)
{
	struct dlm_ls *ls  = container_of(kobj, struct dlm_ls, ls_kobj);
	struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
	return a->store ? a->store(ls, buf, len) : len;
}

static void lockspace_kobj_release(struct kobject *k)
{
	struct dlm_ls *ls  = container_of(k, struct dlm_ls, ls_kobj);
	kfree(ls);
}

static struct sysfs_ops dlm_attr_ops = {
	.show  = dlm_attr_show,
	.store = dlm_attr_store,
};

static struct kobj_type dlm_ktype = {
	.default_attrs = dlm_attrs,
	.sysfs_ops     = &dlm_attr_ops,
	.release       = lockspace_kobj_release,
};

static struct kset dlm_kset = {
	.subsys = &kernel_subsys,
	.kobj   = {.name = "dlm",},
	.ktype  = &dlm_ktype,
};

static int kobject_setup(struct dlm_ls *ls)
{
	char lsname[DLM_LOCKSPACE_LEN];
	int error;

	memset(lsname, 0, DLM_LOCKSPACE_LEN);
	snprintf(lsname, DLM_LOCKSPACE_LEN, "%s", ls->ls_name);

	error = kobject_set_name(&ls->ls_kobj, "%s", lsname);
	if (error)
		return error;

	ls->ls_kobj.kset = &dlm_kset;
	ls->ls_kobj.ktype = &dlm_ktype;
	return 0;
}

static int do_uevent(struct dlm_ls *ls, int in)
{
	int error;

	if (in)
		kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
	else
		kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);

	error = wait_event_interruptible(ls->ls_uevent_wait,
			test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
	if (error)
		goto out;

	error = ls->ls_uevent_result;
 out:
	return error;
}


int dlm_lockspace_init(void)
{
	int error;

	ls_count = 0;
	mutex_init(&ls_lock);
	INIT_LIST_HEAD(&lslist);
	spin_lock_init(&lslist_lock);

	error = kset_register(&dlm_kset);
	if (error)
		printk("dlm_lockspace_init: cannot register kset %d\n", error);
	return error;
}

void dlm_lockspace_exit(void)
{
	kset_unregister(&dlm_kset);
}

static int dlm_scand(void *data)
{
	struct dlm_ls *ls;

	while (!kthread_should_stop()) {
		list_for_each_entry(ls, &lslist, ls_list)
			dlm_scan_rsbs(ls);
		schedule_timeout_interruptible(dlm_config.scan_secs * HZ);
	}
	return 0;
}

static int dlm_scand_start(void)
{
	struct task_struct *p;
	int error = 0;

	p = kthread_run(dlm_scand, NULL, "dlm_scand");
	if (IS_ERR(p))
		error = PTR_ERR(p);
	else
		scand_task = p;
	return error;
}

static void dlm_scand_stop(void)
{
	kthread_stop(scand_task);
}

static struct dlm_ls *dlm_find_lockspace_name(char *name, int namelen)
{
	struct dlm_ls *ls;

	spin_lock(&lslist_lock);

	list_for_each_entry(ls, &lslist, ls_list) {
		if (ls->ls_namelen == namelen &&
		    memcmp(ls->ls_name, name, namelen) == 0)
			goto out;
	}
	ls = NULL;
 out:
	spin_unlock(&lslist_lock);
	return ls;
}

struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
{
	struct dlm_ls *ls;

	spin_lock(&lslist_lock);

	list_for_each_entry(ls, &lslist, ls_list) {
		if (ls->ls_global_id == id) {
			ls->ls_count++;
			goto out;
		}
	}
	ls = NULL;
 out:
	spin_unlock(&lslist_lock);
	return ls;
}

struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
{
	struct dlm_ls *ls;

	spin_lock(&lslist_lock);
	list_for_each_entry(ls, &lslist, ls_list) {
		if (ls->ls_local_handle == lockspace) {
			ls->ls_count++;
			goto out;
		}
	}
	ls = NULL;
 out:
	spin_unlock(&lslist_lock);
	return ls;
}

struct dlm_ls *dlm_find_lockspace_device(int minor)
{
	struct dlm_ls *ls;

	spin_lock(&lslist_lock);
	list_for_each_entry(ls, &lslist, ls_list) {
		if (ls->ls_device.minor == minor) {
			ls->ls_count++;
			goto out;
		}
	}
	ls = NULL;
 out:
	spin_unlock(&lslist_lock);
	return ls;
}

void dlm_put_lockspace(struct dlm_ls *ls)
{
	spin_lock(&lslist_lock);
	ls->ls_count--;
	spin_unlock(&lslist_lock);
}

static void remove_lockspace(struct dlm_ls *ls)
{
	for (;;) {
		spin_lock(&lslist_lock);
		if (ls->ls_count == 0) {
			list_del(&ls->ls_list);
			spin_unlock(&lslist_lock);
			return;
		}
		spin_unlock(&lslist_lock);
		ssleep(1);
	}
}

static int threads_start(void)
{
	int error;

	/* Thread which process lock requests for all lockspace's */
	error = dlm_astd_start();
	if (error) {
		log_print("cannot start dlm_astd thread %d", error);
		goto fail;
	}

	error = dlm_scand_start();
	if (error) {
		log_print("cannot start dlm_scand thread %d", error);
		goto astd_fail;
	}

	/* Thread for sending/receiving messages for all lockspace's */
	error = dlm_lowcomms_start();
	if (error) {
		log_print("cannot start dlm lowcomms %d", error);
		goto scand_fail;
	}

	return 0;

 scand_fail:
	dlm_scand_stop();
 astd_fail:
	dlm_astd_stop();
 fail:
	return error;
}

static void threads_stop(void)
{
	dlm_scand_stop();
	dlm_lowcomms_stop();
	dlm_astd_stop();
}

static int new_lockspace(char *name, int namelen, void **lockspace,
			 uint32_t flags, int lvblen)
{
	struct dlm_ls *ls;
	int i, size, error = -ENOMEM;

	if (namelen > DLM_LOCKSPACE_LEN)
		return -EINVAL;

	if (!lvblen || (lvblen % 8))
		return -EINVAL;

	if (!try_module_get(THIS_MODULE))
		return -EINVAL;

	ls = dlm_find_lockspace_name(name, namelen);
	if (ls) {
		*lockspace = ls;
		module_put(THIS_MODULE);
		return -EEXIST;
	}

	ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_KERNEL);
	if (!ls)
		goto out;
	memcpy(ls->ls_name, name, namelen);
	ls->ls_namelen = namelen;
	ls->ls_exflags = flags;
	ls->ls_lvblen = lvblen;
	ls->ls_count = 0;
	ls->ls_flags = 0;

	size = dlm_config.rsbtbl_size;
	ls->ls_rsbtbl_size = size;

	ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_KERNEL);
	if (!ls->ls_rsbtbl)
		goto out_lsfree;
	for (i = 0; i < size; i++) {
		INIT_LIST_HEAD(&ls->ls_rsbtbl[i].list);
		INIT_LIST_HEAD(&ls->ls_rsbtbl[i].toss);
		rwlock_init(&ls->ls_rsbtbl[i].lock);
	}

	size = dlm_config.lkbtbl_size;
	ls->ls_lkbtbl_size = size;

	ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_KERNEL);
	if (!ls->ls_lkbtbl)
		goto out_rsbfree;
	for (i = 0; i < size; i++) {
		INIT_LIST_HEAD(&ls->ls_lkbtbl[i].list);
		rwlock_init(&ls->ls_lkbtbl[i].lock);
		ls->ls_lkbtbl[i].counter = 1;
	}

	size = dlm_config.dirtbl_size;
	ls->ls_dirtbl_size = size;

	ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_KERNEL);
	if (!ls->ls_dirtbl)
		goto out_lkbfree;
	for (i = 0; i < size; i++) {
		INIT_LIST_HEAD(&ls->ls_dirtbl[i].list);
		rwlock_init(&ls->ls_dirtbl[i].lock);
	}

	INIT_LIST_HEAD(&ls->ls_waiters);
	mutex_init(&ls->ls_waiters_mutex);

	INIT_LIST_HEAD(&ls->ls_nodes);
	INIT_LIST_HEAD(&ls->ls_nodes_gone);
	ls->ls_num_nodes = 0;
	ls->ls_low_nodeid = 0;
	ls->ls_total_weight = 0;
	ls->ls_node_array = NULL;

	memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
	ls->ls_stub_rsb.res_ls = ls;

	ls->ls_debug_rsb_dentry = NULL;
	ls->ls_debug_waiters_dentry = NULL;

	init_waitqueue_head(&ls->ls_uevent_wait);
	ls->ls_uevent_result = 0;

	ls->ls_recoverd_task = NULL;
	mutex_init(&ls->ls_recoverd_active);
	spin_lock_init(&ls->ls_recover_lock);
	ls->ls_recover_status = 0;
	ls->ls_recover_seq = 0;
	ls->ls_recover_args = NULL;
	init_rwsem(&ls->ls_in_recovery);
	INIT_LIST_HEAD(&ls->ls_requestqueue);
	mutex_init(&ls->ls_requestqueue_mutex);
	mutex_init(&ls->ls_clear_proc_locks);

	ls->ls_recover_buf = kmalloc(dlm_config.buffer_size, GFP_KERNEL);
	if (!ls->ls_recover_buf)
		goto out_dirfree;

	INIT_LIST_HEAD(&ls->ls_recover_list);
	spin_lock_init(&ls->ls_recover_list_lock);
	ls->ls_recover_list_count = 0;
	ls->ls_local_handle = ls;
	init_waitqueue_head(&ls->ls_wait_general);
	INIT_LIST_HEAD(&ls->ls_root_list);
	init_rwsem(&ls->ls_root_sem);

	down_write(&ls->ls_in_recovery);

	spin_lock(&lslist_lock);
	list_add(&ls->ls_list, &lslist);
	spin_unlock(&lslist_lock);

	/* needs to find ls in lslist */
	error = dlm_recoverd_start(ls);
	if (error) {
		log_error(ls, "can't start dlm_recoverd %d", error);
		goto out_rcomfree;
	}

	dlm_create_debug_file(ls);

	error = kobject_setup(ls);
	if (error)
		goto out_del;

	error = kobject_register(&ls->ls_kobj);
	if (error)
		goto out_del;

	error = do_uevent(ls, 1);
	if (error)
		goto out_unreg;

	*lockspace = ls;
	return 0;

 out_unreg:
	kobject_unregister(&ls->ls_kobj);
 out_del:
	dlm_delete_debug_file(ls);
	dlm_recoverd_stop(ls);
 out_rcomfree:
	spin_lock(&lslist_lock);
	list_del(&ls->ls_list);
	spin_unlock(&lslist_lock);
	kfree(ls->ls_recover_buf);
 out_dirfree:
	kfree(ls->ls_dirtbl);
 out_lkbfree:
	kfree(ls->ls_lkbtbl);
 out_rsbfree:
	kfree(ls->ls_rsbtbl);
 out_lsfree:
	kfree(ls);
 out:
	module_put(THIS_MODULE);
	return error;
}

int dlm_new_lockspace(char *name, int namelen, void **lockspace,
		      uint32_t flags, int lvblen)
{
	int error = 0;

	mutex_lock(&ls_lock);
	if (!ls_count)
		error = threads_start();
	if (error)
		goto out;

	error = new_lockspace(name, namelen, lockspace, flags, lvblen);
	if (!error)
		ls_count++;
 out:
	mutex_unlock(&ls_lock);
	return error;
}

/* Return 1 if the lockspace still has active remote locks,
 *        2 if the lockspace still has active local locks.
 */
static int lockspace_busy(struct dlm_ls *ls)
{
	int i, lkb_found = 0;
	struct dlm_lkb *lkb;

	/* NOTE: We check the lockidtbl here rather than the resource table.
	   This is because there may be LKBs queued as ASTs that have been
	   unlinked from their RSBs and are pending deletion once the AST has
	   been delivered */

	for (i = 0; i < ls->ls_lkbtbl_size; i++) {
		read_lock(&ls->ls_lkbtbl[i].lock);
		if (!list_empty(&ls->ls_lkbtbl[i].list)) {
			lkb_found = 1;
			list_for_each_entry(lkb, &ls->ls_lkbtbl[i].list,
					    lkb_idtbl_list) {
				if (!lkb->lkb_nodeid) {
					read_unlock(&ls->ls_lkbtbl[i].lock);
					return 2;
				}
			}
		}
		read_unlock(&ls->ls_lkbtbl[i].lock);
	}
	return lkb_found;
}

static int release_lockspace(struct dlm_ls *ls, int force)
{
	struct dlm_lkb *lkb;
	struct dlm_rsb *rsb;
	struct list_head *head;
	int i;
	int busy = lockspace_busy(ls);

	if (busy > force)
		return -EBUSY;

	if (force < 3)
		do_uevent(ls, 0);

	dlm_recoverd_stop(ls);

	remove_lockspace(ls);

	dlm_delete_debug_file(ls);

	dlm_astd_suspend();

	kfree(ls->ls_recover_buf);

	/*
	 * Free direntry structs.
	 */

	dlm_dir_clear(ls);
	kfree(ls->ls_dirtbl);

	/*
	 * Free all lkb's on lkbtbl[] lists.
	 */

	for (i = 0; i < ls->ls_lkbtbl_size; i++) {
		head = &ls->ls_lkbtbl[i].list;
		while (!list_empty(head)) {
			lkb = list_entry(head->next, struct dlm_lkb,
					 lkb_idtbl_list);

			list_del(&lkb->lkb_idtbl_list);

			dlm_del_ast(lkb);

			if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
				free_lvb(lkb->lkb_lvbptr);

			free_lkb(lkb);
		}
	}
	dlm_astd_resume();

	kfree(ls->ls_lkbtbl);

	/*
	 * Free all rsb's on rsbtbl[] lists
	 */

	for (i = 0; i < ls->ls_rsbtbl_size; i++) {
		head = &ls->ls_rsbtbl[i].list;
		while (!list_empty(head)) {
			rsb = list_entry(head->next, struct dlm_rsb,
					 res_hashchain);

			list_del(&rsb->res_hashchain);
			free_rsb(rsb);
		}

		head = &ls->ls_rsbtbl[i].toss;
		while (!list_empty(head)) {
			rsb = list_entry(head->next, struct dlm_rsb,
					 res_hashchain);
			list_del(&rsb->res_hashchain);
			free_rsb(rsb);
		}
	}

	kfree(ls->ls_rsbtbl);

	/*
	 * Free structures on any other lists
	 */

	kfree(ls->ls_recover_args);
	dlm_clear_free_entries(ls);
	dlm_clear_members(ls);
	dlm_clear_members_gone(ls);
	kfree(ls->ls_node_array);
	kobject_unregister(&ls->ls_kobj);
        /* The ls structure will be freed when the kobject is done with */

	mutex_lock(&ls_lock);
	ls_count--;
	if (!ls_count)
		threads_stop();
	mutex_unlock(&ls_lock);

	module_put(THIS_MODULE);
	return 0;
}

/*
 * Called when a system has released all its locks and is not going to use the
 * lockspace any longer.  We free everything we're managing for this lockspace.
 * Remaining nodes will go through the recovery process as if we'd died.  The
 * lockspace must continue to function as usual, participating in recoveries,
 * until this returns.
 *
 * Force has 4 possible values:
 * 0 - don't destroy locksapce if it has any LKBs
 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
 * 2 - destroy lockspace regardless of LKBs
 * 3 - destroy lockspace as part of a forced shutdown
 */

int dlm_release_lockspace(void *lockspace, int force)
{
	struct dlm_ls *ls;

	ls = dlm_find_lockspace_local(lockspace);
	if (!ls)
		return -EINVAL;
	dlm_put_lockspace(ls);
	return release_lockspace(ls, force);
}
Commit	Line	Data
e7fd4179 DT	1	/******************************************************************************
	2	*******************************************************************************
	3	**
	4	** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
	5	** Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved.
	6	**
	7	** This copyrighted material is made available to anyone wishing to use,
	8	** modify, copy, or redistribute it subject to the terms and conditions
	9	** of the GNU General Public License v.2.
	10	**
	11	*******************************************************************************
	12	******************************************************************************/
	13
	14	#include "dlm_internal.h"
	15	#include "lockspace.h"
	16	#include "member.h"
	17	#include "recoverd.h"
	18	#include "ast.h"
	19	#include "dir.h"
	20	#include "lowcomms.h"
	21	#include "config.h"
	22	#include "memory.h"
	23	#include "lock.h"
c56b39cd	24	#include "recover.h"
e7fd4179 DT	25
	26	#ifdef CONFIG_DLM_DEBUG
	27	int dlm_create_debug_file(struct dlm_ls *ls);
	28	void dlm_delete_debug_file(struct dlm_ls *ls);
	29	#else
	30	static inline int dlm_create_debug_file(struct dlm_ls *ls) { return 0; }
	31	static inline void dlm_delete_debug_file(struct dlm_ls *ls) { }
	32	#endif
	33
	34	static int ls_count;
90135925	35	static struct mutex ls_lock;
e7fd4179 DT	36	static struct list_head lslist;
	37	static spinlock_t lslist_lock;
	38	static struct task_struct * scand_task;
	39
	40
	41	static ssize_t dlm_control_store(struct dlm_ls ls, const char buf, size_t len)
	42	{
	43	ssize_t ret = len;
	44	int n = simple_strtol(buf, NULL, 0);
	45
	46	switch (n) {
	47	case 0:
	48	dlm_ls_stop(ls);
	49	break;
	50	case 1:
	51	dlm_ls_start(ls);
	52	break;
	53	default:
	54	ret = -EINVAL;
	55	}
	56	return ret;
	57	}
	58
	59	static ssize_t dlm_event_store(struct dlm_ls ls, const char buf, size_t len)
	60	{
	61	ls->ls_uevent_result = simple_strtol(buf, NULL, 0);
	62	set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
	63	wake_up(&ls->ls_uevent_wait);
	64	return len;
	65	}
	66
	67	static ssize_t dlm_id_show(struct dlm_ls ls, char buf)
	68	{
a1d144c7	69	return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
e7fd4179 DT	70	}
	71
	72	static ssize_t dlm_id_store(struct dlm_ls ls, const char buf, size_t len)
	73	{
	74	ls->ls_global_id = simple_strtoul(buf, NULL, 0);
	75	return len;
	76	}
	77
c56b39cd DT	78	static ssize_t dlm_recover_status_show(struct dlm_ls ls, char buf)
	79	{
	80	uint32_t status = dlm_recover_status(ls);
a1d144c7	81	return snprintf(buf, PAGE_SIZE, "%x\n", status);
c56b39cd DT	82	}
c56b39cd DT	83
faa0f267 DT	84	static ssize_t dlm_recover_nodeid_show(struct dlm_ls ls, char buf)
faa0f267 DT	85	{
a1d144c7	86	return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
faa0f267 DT	87	}
faa0f267 DT	88
e7fd4179 DT	89	struct dlm_attr {
	90	struct attribute attr;
	91	ssize_t (show)(struct dlm_ls , char *);
	92	ssize_t (store)(struct dlm_ls , const char *, size_t);
	93	};
	94
	95	static struct dlm_attr dlm_attr_control = {
	96	.attr = {.name = "control", .mode = S_IWUSR},
	97	.store = dlm_control_store
	98	};
	99
	100	static struct dlm_attr dlm_attr_event = {
	101	.attr = {.name = "event_done", .mode = S_IWUSR},
	102	.store = dlm_event_store
	103	};
	104
	105	static struct dlm_attr dlm_attr_id = {
	106	.attr = {.name = "id", .mode = S_IRUGO \| S_IWUSR},
	107	.show = dlm_id_show,
	108	.store = dlm_id_store
	109	};
	110
c56b39cd DT	111	static struct dlm_attr dlm_attr_recover_status = {
	112	.attr = {.name = "recover_status", .mode = S_IRUGO},
	113	.show = dlm_recover_status_show
	114	};
	115
faa0f267 DT	116	static struct dlm_attr dlm_attr_recover_nodeid = {
	117	.attr = {.name = "recover_nodeid", .mode = S_IRUGO},
	118	.show = dlm_recover_nodeid_show
	119	};
	120
e7fd4179 DT	121	static struct attribute *dlm_attrs[] = {
	122	&dlm_attr_control.attr,
	123	&dlm_attr_event.attr,
	124	&dlm_attr_id.attr,
c56b39cd	125	&dlm_attr_recover_status.attr,
faa0f267	126	&dlm_attr_recover_nodeid.attr,
e7fd4179 DT	127	NULL,
	128	};
	129
	130	static ssize_t dlm_attr_show(struct kobject kobj, struct attribute attr,
	131	char *buf)
	132	{
	133	struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
	134	struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
	135	return a->show ? a->show(ls, buf) : 0;
	136	}
	137
	138	static ssize_t dlm_attr_store(struct kobject kobj, struct attribute attr,
	139	const char *buf, size_t len)
	140	{
	141	struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
	142	struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
	143	return a->store ? a->store(ls, buf, len) : len;
	144	}
	145
ba542e3b PC	146	static void lockspace_kobj_release(struct kobject *k)
	147	{
	148	struct dlm_ls *ls = container_of(k, struct dlm_ls, ls_kobj);
	149	kfree(ls);
	150	}
	151
e7fd4179 DT	152	static struct sysfs_ops dlm_attr_ops = {
	153	.show = dlm_attr_show,
	154	.store = dlm_attr_store,
	155	};
	156
	157	static struct kobj_type dlm_ktype = {
	158	.default_attrs = dlm_attrs,
	159	.sysfs_ops = &dlm_attr_ops,
ba542e3b	160	.release = lockspace_kobj_release,
e7fd4179 DT	161	};
	162
	163	static struct kset dlm_kset = {
	164	.subsys = &kernel_subsys,
	165	.kobj = {.name = "dlm",},
	166	.ktype = &dlm_ktype,
	167	};
	168
	169	static int kobject_setup(struct dlm_ls *ls)
	170	{
	171	char lsname[DLM_LOCKSPACE_LEN];
	172	int error;
	173
	174	memset(lsname, 0, DLM_LOCKSPACE_LEN);
	175	snprintf(lsname, DLM_LOCKSPACE_LEN, "%s", ls->ls_name);
	176
	177	error = kobject_set_name(&ls->ls_kobj, "%s", lsname);
	178	if (error)
	179	return error;
	180
	181	ls->ls_kobj.kset = &dlm_kset;
	182	ls->ls_kobj.ktype = &dlm_ktype;
	183	return 0;
	184	}
	185
	186	static int do_uevent(struct dlm_ls *ls, int in)
	187	{
	188	int error;
	189
	190	if (in)
	191	kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
	192	else
	193	kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
	194
	195	error = wait_event_interruptible(ls->ls_uevent_wait,
	196	test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
	197	if (error)
	198	goto out;
	199
	200	error = ls->ls_uevent_result;
	201	out:
	202	return error;
	203	}
	204
	205
	206	int dlm_lockspace_init(void)
	207	{
	208	int error;
	209
	210	ls_count = 0;
90135925	211	mutex_init(&ls_lock);
e7fd4179 DT	212	INIT_LIST_HEAD(&lslist);
	213	spin_lock_init(&lslist_lock);
	214
	215	error = kset_register(&dlm_kset);
	216	if (error)
	217	printk("dlm_lockspace_init: cannot register kset %d\n", error);
	218	return error;
	219	}
	220
	221	void dlm_lockspace_exit(void)
	222	{
	223	kset_unregister(&dlm_kset);
	224	}
	225
	226	static int dlm_scand(void *data)
	227	{
	228	struct dlm_ls *ls;
	229
	230	while (!kthread_should_stop()) {
	231	list_for_each_entry(ls, &lslist, ls_list)
	232	dlm_scan_rsbs(ls);
	233	schedule_timeout_interruptible(dlm_config.scan_secs * HZ);
	234	}
	235	return 0;
	236	}
	237
	238	static int dlm_scand_start(void)
	239	{
	240	struct task_struct *p;
	241	int error = 0;
	242
	243	p = kthread_run(dlm_scand, NULL, "dlm_scand");
	244	if (IS_ERR(p))
	245	error = PTR_ERR(p);
	246	else
	247	scand_task = p;
	248	return error;
	249	}
	250
	251	static void dlm_scand_stop(void)
	252	{
	253	kthread_stop(scand_task);
	254	}
	255
	256	static struct dlm_ls dlm_find_lockspace_name(char name, int namelen)
	257	{
	258	struct dlm_ls *ls;
	259
	260	spin_lock(&lslist_lock);
	261
	262	list_for_each_entry(ls, &lslist, ls_list) {
	263	if (ls->ls_namelen == namelen &&
	264	memcmp(ls->ls_name, name, namelen) == 0)
	265	goto out;
	266	}
	267	ls = NULL;
	268	out:
	269	spin_unlock(&lslist_lock);
	270	return ls;
	271	}
	272
	273	struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
	274	{
	275	struct dlm_ls *ls;
276
277	spin_lock(&lslist_lock);
278
279	list_for_each_entry(ls, &lslist, ls_list) {
280	if (ls->ls_global_id == id) {
281	ls->ls_count++;
282	goto out;
283	}
284	}
285	ls = NULL;
286	out:
287	spin_unlock(&lslist_lock);
288	return ls;
289	}
290
597d0cae	291	struct dlm_ls dlm_find_lockspace_local(dlm_lockspace_t lockspace)
e7fd4179	292	{
597d0cae	293	struct dlm_ls *ls;
e7fd4179 DT	294
e7fd4179 DT	295	spin_lock(&lslist_lock);
597d0cae DT	296	list_for_each_entry(ls, &lslist, ls_list) {
	297	if (ls->ls_local_handle == lockspace) {
	298	ls->ls_count++;
	299	goto out;
	300	}
	301	}
	302	ls = NULL;
	303	out:
	304	spin_unlock(&lslist_lock);
	305	return ls;
	306	}
	307
	308	struct dlm_ls *dlm_find_lockspace_device(int minor)
	309	{
	310	struct dlm_ls *ls;
	311
	312	spin_lock(&lslist_lock);
	313	list_for_each_entry(ls, &lslist, ls_list) {
	314	if (ls->ls_device.minor == minor) {
	315	ls->ls_count++;
	316	goto out;
	317	}
	318	}
	319	ls = NULL;
	320	out:
e7fd4179 DT	321	spin_unlock(&lslist_lock);
	322	return ls;
	323	}
	324
	325	void dlm_put_lockspace(struct dlm_ls *ls)
	326	{
	327	spin_lock(&lslist_lock);
	328	ls->ls_count--;
	329	spin_unlock(&lslist_lock);
	330	}
	331
	332	static void remove_lockspace(struct dlm_ls *ls)
	333	{
	334	for (;;) {
	335	spin_lock(&lslist_lock);
	336	if (ls->ls_count == 0) {
	337	list_del(&ls->ls_list);
	338	spin_unlock(&lslist_lock);
	339	return;
	340	}
	341	spin_unlock(&lslist_lock);
	342	ssleep(1);
	343	}
	344	}
	345
	346	static int threads_start(void)
	347	{
	348	int error;
	349
	350	/* Thread which process lock requests for all lockspace's */
	351	error = dlm_astd_start();
	352	if (error) {
	353	log_print("cannot start dlm_astd thread %d", error);
	354	goto fail;
	355	}
	356
	357	error = dlm_scand_start();
	358	if (error) {
	359	log_print("cannot start dlm_scand thread %d", error);
	360	goto astd_fail;
	361	}
	362
	363	/* Thread for sending/receiving messages for all lockspace's */
	364	error = dlm_lowcomms_start();
	365	if (error) {
	366	log_print("cannot start dlm lowcomms %d", error);
	367	goto scand_fail;
	368	}
	369
	370	return 0;
	371
	372	scand_fail:
	373	dlm_scand_stop();
	374	astd_fail:
	375	dlm_astd_stop();
	376	fail:
	377	return error;
	378	}
	379
	380	static void threads_stop(void)
	381	{
	382	dlm_scand_stop();
	383	dlm_lowcomms_stop();
	384	dlm_astd_stop();
385	}
386
387	static int new_lockspace(char name, int namelen, void *lockspace,
388	uint32_t flags, int lvblen)
389	{
390	struct dlm_ls *ls;
391	int i, size, error = -ENOMEM;
392
393	if (namelen > DLM_LOCKSPACE_LEN)
394	return -EINVAL;
395
396	if (!lvblen \|\| (lvblen % 8))
397	return -EINVAL;
398
399	if (!try_module_get(THIS_MODULE))
400	return -EINVAL;
401
402	ls = dlm_find_lockspace_name(name, namelen);
403	if (ls) {
404	*lockspace = ls;
405	module_put(THIS_MODULE);
406	return -EEXIST;
407	}
408
90135925	409	ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_KERNEL);
e7fd4179 DT	410	if (!ls)
e7fd4179 DT	411	goto out;
e7fd4179 DT	412	memcpy(ls->ls_name, name, namelen);
	413	ls->ls_namelen = namelen;
	414	ls->ls_exflags = flags;
	415	ls->ls_lvblen = lvblen;
	416	ls->ls_count = 0;
	417	ls->ls_flags = 0;
	418
	419	size = dlm_config.rsbtbl_size;
	420	ls->ls_rsbtbl_size = size;
	421
	422	ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_KERNEL);
	423	if (!ls->ls_rsbtbl)
	424	goto out_lsfree;
	425	for (i = 0; i < size; i++) {
	426	INIT_LIST_HEAD(&ls->ls_rsbtbl[i].list);
	427	INIT_LIST_HEAD(&ls->ls_rsbtbl[i].toss);
	428	rwlock_init(&ls->ls_rsbtbl[i].lock);
	429	}
	430
	431	size = dlm_config.lkbtbl_size;
	432	ls->ls_lkbtbl_size = size;
	433
	434	ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_KERNEL);
	435	if (!ls->ls_lkbtbl)
	436	goto out_rsbfree;
	437	for (i = 0; i < size; i++) {
	438	INIT_LIST_HEAD(&ls->ls_lkbtbl[i].list);
	439	rwlock_init(&ls->ls_lkbtbl[i].lock);
	440	ls->ls_lkbtbl[i].counter = 1;
	441	}
	442
	443	size = dlm_config.dirtbl_size;
	444	ls->ls_dirtbl_size = size;
	445
	446	ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_KERNEL);
	447	if (!ls->ls_dirtbl)
	448	goto out_lkbfree;
	449	for (i = 0; i < size; i++) {
	450	INIT_LIST_HEAD(&ls->ls_dirtbl[i].list);
	451	rwlock_init(&ls->ls_dirtbl[i].lock);
	452	}
	453
	454	INIT_LIST_HEAD(&ls->ls_waiters);
90135925	455	mutex_init(&ls->ls_waiters_mutex);
e7fd4179 DT	456
	457	INIT_LIST_HEAD(&ls->ls_nodes);
	458	INIT_LIST_HEAD(&ls->ls_nodes_gone);
	459	ls->ls_num_nodes = 0;
	460	ls->ls_low_nodeid = 0;
	461	ls->ls_total_weight = 0;
	462	ls->ls_node_array = NULL;
	463
	464	memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
	465	ls->ls_stub_rsb.res_ls = ls;
	466
5de6319b DT	467	ls->ls_debug_rsb_dentry = NULL;
5de6319b DT	468	ls->ls_debug_waiters_dentry = NULL;
e7fd4179 DT	469
	470	init_waitqueue_head(&ls->ls_uevent_wait);
	471	ls->ls_uevent_result = 0;
	472
	473	ls->ls_recoverd_task = NULL;
90135925	474	mutex_init(&ls->ls_recoverd_active);
e7fd4179 DT	475	spin_lock_init(&ls->ls_recover_lock);
	476	ls->ls_recover_status = 0;
	477	ls->ls_recover_seq = 0;
	478	ls->ls_recover_args = NULL;
	479	init_rwsem(&ls->ls_in_recovery);
	480	INIT_LIST_HEAD(&ls->ls_requestqueue);
90135925	481	mutex_init(&ls->ls_requestqueue_mutex);
597d0cae	482	mutex_init(&ls->ls_clear_proc_locks);
e7fd4179 DT	483
	484	ls->ls_recover_buf = kmalloc(dlm_config.buffer_size, GFP_KERNEL);
	485	if (!ls->ls_recover_buf)
	486	goto out_dirfree;
	487
	488	INIT_LIST_HEAD(&ls->ls_recover_list);
	489	spin_lock_init(&ls->ls_recover_list_lock);
	490	ls->ls_recover_list_count = 0;
597d0cae	491	ls->ls_local_handle = ls;
e7fd4179 DT	492	init_waitqueue_head(&ls->ls_wait_general);
	493	INIT_LIST_HEAD(&ls->ls_root_list);
	494	init_rwsem(&ls->ls_root_sem);
	495
	496	down_write(&ls->ls_in_recovery);
	497
5f88f1ea DT	498	spin_lock(&lslist_lock);
	499	list_add(&ls->ls_list, &lslist);
	500	spin_unlock(&lslist_lock);
	501
	502	/* needs to find ls in lslist */
e7fd4179 DT	503	error = dlm_recoverd_start(ls);
	504	if (error) {
	505	log_error(ls, "can't start dlm_recoverd %d", error);
	506	goto out_rcomfree;
	507	}
	508
e7fd4179 DT	509	dlm_create_debug_file(ls);
	510
	511	error = kobject_setup(ls);
	512	if (error)
	513	goto out_del;
	514
	515	error = kobject_register(&ls->ls_kobj);
	516	if (error)
	517	goto out_del;
	518
	519	error = do_uevent(ls, 1);
	520	if (error)
	521	goto out_unreg;
	522
	523	*lockspace = ls;
	524	return 0;
	525
	526	out_unreg:
	527	kobject_unregister(&ls->ls_kobj);
	528	out_del:
	529	dlm_delete_debug_file(ls);
5f88f1ea DT	530	dlm_recoverd_stop(ls);
5f88f1ea DT	531	out_rcomfree:
e7fd4179 DT	532	spin_lock(&lslist_lock);
	533	list_del(&ls->ls_list);
	534	spin_unlock(&lslist_lock);
e7fd4179 DT	535	kfree(ls->ls_recover_buf);
	536	out_dirfree:
	537	kfree(ls->ls_dirtbl);
	538	out_lkbfree:
	539	kfree(ls->ls_lkbtbl);
	540	out_rsbfree:
	541	kfree(ls->ls_rsbtbl);
	542	out_lsfree:
	543	kfree(ls);
	544	out:
	545	module_put(THIS_MODULE);
	546	return error;
	547	}
	548
	549	int dlm_new_lockspace(char name, int namelen, void *lockspace,
	550	uint32_t flags, int lvblen)
	551	{
	552	int error = 0;
	553
90135925	554	mutex_lock(&ls_lock);
e7fd4179 DT	555	if (!ls_count)
	556	error = threads_start();
	557	if (error)
	558	goto out;
	559
	560	error = new_lockspace(name, namelen, lockspace, flags, lvblen);
	561	if (!error)
	562	ls_count++;
	563	out:
90135925	564	mutex_unlock(&ls_lock);
e7fd4179 DT	565	return error;
	566	}
	567
	568	/* Return 1 if the lockspace still has active remote locks,
	569	* 2 if the lockspace still has active local locks.
	570	*/
	571	static int lockspace_busy(struct dlm_ls *ls)
	572	{
	573	int i, lkb_found = 0;
	574	struct dlm_lkb *lkb;
	575
	576	/* NOTE: We check the lockidtbl here rather than the resource table.
	577	This is because there may be LKBs queued as ASTs that have been
	578	unlinked from their RSBs and are pending deletion once the AST has
	579	been delivered */
	580
	581	for (i = 0; i < ls->ls_lkbtbl_size; i++) {
	582	read_lock(&ls->ls_lkbtbl[i].lock);
	583	if (!list_empty(&ls->ls_lkbtbl[i].list)) {
	584	lkb_found = 1;
	585	list_for_each_entry(lkb, &ls->ls_lkbtbl[i].list,
	586	lkb_idtbl_list) {
	587	if (!lkb->lkb_nodeid) {
	588	read_unlock(&ls->ls_lkbtbl[i].lock);
	589	return 2;
	590	}
	591	}
	592	}
	593	read_unlock(&ls->ls_lkbtbl[i].lock);
	594	}
	595	return lkb_found;
	596	}
	597
	598	static int release_lockspace(struct dlm_ls *ls, int force)
	599	{
	600	struct dlm_lkb *lkb;
	601	struct dlm_rsb *rsb;
	602	struct list_head *head;
	603	int i;
	604	int busy = lockspace_busy(ls);
	605
	606	if (busy > force)
	607	return -EBUSY;
	608
	609	if (force < 3)
	610	do_uevent(ls, 0);
	611
	612	dlm_recoverd_stop(ls);
	613
	614	remove_lockspace(ls);
	615
	616	dlm_delete_debug_file(ls);
	617
	618	dlm_astd_suspend();
	619
	620	kfree(ls->ls_recover_buf);
	621
	622	/*
	623	* Free direntry structs.
	624	*/
	625
	626	dlm_dir_clear(ls);
	627	kfree(ls->ls_dirtbl);
	628
629	/*
630	* Free all lkb's on lkbtbl[] lists.
631	*/
632
633	for (i = 0; i < ls->ls_lkbtbl_size; i++) {
634	head = &ls->ls_lkbtbl[i].list;
635	while (!list_empty(head)) {
636	lkb = list_entry(head->next, struct dlm_lkb,
637	lkb_idtbl_list);
638
639	list_del(&lkb->lkb_idtbl_list);
640
641	dlm_del_ast(lkb);
642
643	if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
644	free_lvb(lkb->lkb_lvbptr);
645
646	free_lkb(lkb);
647	}
648	}
649	dlm_astd_resume();
650
651	kfree(ls->ls_lkbtbl);
652
653	/*
654	* Free all rsb's on rsbtbl[] lists
655	*/
656
657	for (i = 0; i < ls->ls_rsbtbl_size; i++) {
658	head = &ls->ls_rsbtbl[i].list;
659	while (!list_empty(head)) {
660	rsb = list_entry(head->next, struct dlm_rsb,
661	res_hashchain);
662
663	list_del(&rsb->res_hashchain);
664	free_rsb(rsb);
665	}
666
667	head = &ls->ls_rsbtbl[i].toss;
668	while (!list_empty(head)) {
669	rsb = list_entry(head->next, struct dlm_rsb,
670	res_hashchain);
671	list_del(&rsb->res_hashchain);
672	free_rsb(rsb);
673	}
674	}
675
676	kfree(ls->ls_rsbtbl);
677
678	/*
679	* Free structures on any other lists
680	*/
681
682	kfree(ls->ls_recover_args);
683	dlm_clear_free_entries(ls);
684	dlm_clear_members(ls);
685	dlm_clear_members_gone(ls);
686	kfree(ls->ls_node_array);
687	kobject_unregister(&ls->ls_kobj);
ba542e3b	688	/* The ls structure will be freed when the kobject is done with */
e7fd4179	689
90135925	690	mutex_lock(&ls_lock);
e7fd4179 DT	691	ls_count--;
	692	if (!ls_count)
	693	threads_stop();
90135925	694	mutex_unlock(&ls_lock);
e7fd4179 DT	695
	696	module_put(THIS_MODULE);
	697	return 0;
	698	}
	699
	700	/*
	701	* Called when a system has released all its locks and is not going to use the
	702	* lockspace any longer. We free everything we're managing for this lockspace.
	703	* Remaining nodes will go through the recovery process as if we'd died. The
	704	* lockspace must continue to function as usual, participating in recoveries,
	705	* until this returns.
	706	*
	707	* Force has 4 possible values:
	708	* 0 - don't destroy locksapce if it has any LKBs
	709	* 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
	710	* 2 - destroy lockspace regardless of LKBs
	711	* 3 - destroy lockspace as part of a forced shutdown
	712	*/
	713
	714	int dlm_release_lockspace(void *lockspace, int force)
	715	{
	716	struct dlm_ls *ls;
	717
	718	ls = dlm_find_lockspace_local(lockspace);
	719	if (!ls)
	720	return -EINVAL;
	721	dlm_put_lockspace(ls);
	722	return release_lockspace(ls, force);
	723	}
	724