[mirror_ubuntu-artful-kernel.git] / drivers / md / md-cluster.c

/*
 * Copyright (C) 2015, SUSE
 *
 * 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; either version 2, or (at your option)
 * any later version.
 *
 */


#include <linux/module.h>
#include <linux/dlm.h>
#include <linux/sched.h>
#include "md.h"
#include "bitmap.h"
#include "md-cluster.h"

#define LVB_SIZE	64

struct dlm_lock_resource {
	dlm_lockspace_t *ls;
	struct dlm_lksb lksb;
	char *name; /* lock name. */
	uint32_t flags; /* flags to pass to dlm_lock() */
	struct completion completion; /* completion for synchronized locking */
	void (*bast)(void *arg, int mode); /* blocking AST function pointer*/
	struct mddev *mddev; /* pointing back to mddev. */
};

struct suspend_info {
	int slot;
	sector_t lo;
	sector_t hi;
	struct list_head list;
};

struct resync_info {
	__le64 lo;
	__le64 hi;
};

struct md_cluster_info {
	/* dlm lock space and resources for clustered raid. */
	dlm_lockspace_t *lockspace;
	int slot_number;
	struct completion completion;
	struct dlm_lock_resource *sb_lock;
	struct mutex sb_mutex;
	struct dlm_lock_resource *bitmap_lockres;
	struct list_head suspend_list;
	spinlock_t suspend_lock;
	struct md_thread *recovery_thread;
	unsigned long recovery_map;
	/* communication loc resources */
	struct dlm_lock_resource *ack_lockres;
	struct dlm_lock_resource *message_lockres;
	struct dlm_lock_resource *token_lockres;
	struct md_thread *recv_thread;
};

enum msg_type {
	METADATA_UPDATED = 0,
	RESYNCING,
};

struct cluster_msg {
	int type;
	int slot;
	sector_t low;
	sector_t high;
};

static void sync_ast(void *arg)
{
	struct dlm_lock_resource *res;

	res = (struct dlm_lock_resource *) arg;
	complete(&res->completion);
}

static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
{
	int ret = 0;

	init_completion(&res->completion);
	ret = dlm_lock(res->ls, mode, &res->lksb,
			res->flags, res->name, strlen(res->name),
			0, sync_ast, res, res->bast);
	if (ret)
		return ret;
	wait_for_completion(&res->completion);
	return res->lksb.sb_status;
}

static int dlm_unlock_sync(struct dlm_lock_resource *res)
{
	return dlm_lock_sync(res, DLM_LOCK_NL);
}

static struct dlm_lock_resource *lockres_init(struct mddev *mddev,
		char *name, void (*bastfn)(void *arg, int mode), int with_lvb)
{
	struct dlm_lock_resource *res = NULL;
	int ret, namelen;
	struct md_cluster_info *cinfo = mddev->cluster_info;

	res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
	if (!res)
		return NULL;
	res->ls = cinfo->lockspace;
	res->mddev = mddev;
	namelen = strlen(name);
	res->name = kzalloc(namelen + 1, GFP_KERNEL);
	if (!res->name) {
		pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
		goto out_err;
	}
	strlcpy(res->name, name, namelen + 1);
	if (with_lvb) {
		res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
		if (!res->lksb.sb_lvbptr) {
			pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
			goto out_err;
		}
		res->flags = DLM_LKF_VALBLK;
	}

	if (bastfn)
		res->bast = bastfn;

	res->flags |= DLM_LKF_EXPEDITE;

	ret = dlm_lock_sync(res, DLM_LOCK_NL);
	if (ret) {
		pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
		goto out_err;
	}
	res->flags &= ~DLM_LKF_EXPEDITE;
	res->flags |= DLM_LKF_CONVERT;

	return res;
out_err:
	kfree(res->lksb.sb_lvbptr);
	kfree(res->name);
	kfree(res);
	return NULL;
}

static void lockres_free(struct dlm_lock_resource *res)
{
	if (!res)
		return;

	init_completion(&res->completion);
	dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res);
	wait_for_completion(&res->completion);

	kfree(res->name);
	kfree(res->lksb.sb_lvbptr);
	kfree(res);
}

static char *pretty_uuid(char *dest, char *src)
{
	int i, len = 0;

	for (i = 0; i < 16; i++) {
		if (i == 4 || i == 6 || i == 8 || i == 10)
			len += sprintf(dest + len, "-");
		len += sprintf(dest + len, "%02x", (__u8)src[i]);
	}
	return dest;
}

static void add_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres,
		sector_t lo, sector_t hi)
{
	struct resync_info *ri;

	ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
	ri->lo = cpu_to_le64(lo);
	ri->hi = cpu_to_le64(hi);
}

static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres)
{
	struct resync_info ri;
	struct suspend_info *s = NULL;
	sector_t hi = 0;

	dlm_lock_sync(lockres, DLM_LOCK_CR);
	memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
	hi = le64_to_cpu(ri.hi);
	if (ri.hi > 0) {
		s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
		if (!s)
			goto out;
		s->hi = hi;
		s->lo = le64_to_cpu(ri.lo);
	}
	dlm_unlock_sync(lockres);
out:
	return s;
}

void recover_bitmaps(struct md_thread *thread)
{
	struct mddev *mddev = thread->mddev;
	struct md_cluster_info *cinfo = mddev->cluster_info;
	struct dlm_lock_resource *bm_lockres;
	char str[64];
	int slot, ret;
	struct suspend_info *s, *tmp;
	sector_t lo, hi;

	while (cinfo->recovery_map) {
		slot = fls64((u64)cinfo->recovery_map) - 1;

		/* Clear suspend_area associated with the bitmap */
		spin_lock_irq(&cinfo->suspend_lock);
		list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
			if (slot == s->slot) {
				list_del(&s->list);
				kfree(s);
			}
		spin_unlock_irq(&cinfo->suspend_lock);

		snprintf(str, 64, "bitmap%04d", slot);
		bm_lockres = lockres_init(mddev, str, NULL, 1);
		if (!bm_lockres) {
			pr_err("md-cluster: Cannot initialize bitmaps\n");
			goto clear_bit;
		}

		ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
		if (ret) {
			pr_err("md-cluster: Could not DLM lock %s: %d\n",
					str, ret);
			goto clear_bit;
		}
		ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi);
		if (ret) {
			pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
			goto dlm_unlock;
		}
		if (hi > 0) {
			/* TODO:Wait for current resync to get over */
			set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
			if (lo < mddev->recovery_cp)
				mddev->recovery_cp = lo;
			md_check_recovery(mddev);
		}
dlm_unlock:
		dlm_unlock_sync(bm_lockres);
clear_bit:
		clear_bit(slot, &cinfo->recovery_map);
	}
}

static void recover_prep(void *arg)
{
}

static void recover_slot(void *arg, struct dlm_slot *slot)
{
	struct mddev *mddev = arg;
	struct md_cluster_info *cinfo = mddev->cluster_info;

	pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
			mddev->bitmap_info.cluster_name,
			slot->nodeid, slot->slot,
			cinfo->slot_number);
	set_bit(slot->slot - 1, &cinfo->recovery_map);
	if (!cinfo->recovery_thread) {
		cinfo->recovery_thread = md_register_thread(recover_bitmaps,
				mddev, "recover");
		if (!cinfo->recovery_thread) {
			pr_warn("md-cluster: Could not create recovery thread\n");
			return;
		}
	}
	md_wakeup_thread(cinfo->recovery_thread);
}

static void recover_done(void *arg, struct dlm_slot *slots,
		int num_slots, int our_slot,
		uint32_t generation)
{
	struct mddev *mddev = arg;
	struct md_cluster_info *cinfo = mddev->cluster_info;

	cinfo->slot_number = our_slot;
	complete(&cinfo->completion);
}

static const struct dlm_lockspace_ops md_ls_ops = {
	.recover_prep = recover_prep,
	.recover_slot = recover_slot,
	.recover_done = recover_done,
};

/*
 * The BAST function for the ack lock resource
 * This function wakes up the receive thread in
 * order to receive and process the message.
 */
static void ack_bast(void *arg, int mode)
{
	struct dlm_lock_resource *res = (struct dlm_lock_resource *)arg;
	struct md_cluster_info *cinfo = res->mddev->cluster_info;

	if (mode == DLM_LOCK_EX)
		md_wakeup_thread(cinfo->recv_thread);
}

static void process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg)
{
	switch (msg->type) {
	case METADATA_UPDATED:
		pr_info("%s: %d Received message: METADATA_UPDATE from %d\n",
			__func__, __LINE__, msg->slot);
		break;
	case RESYNCING:
		pr_info("%s: %d Received message: RESYNCING from %d\n",
			__func__, __LINE__, msg->slot);
		break;
	};
}

/*
 * thread for receiving message
 */
static void recv_daemon(struct md_thread *thread)
{
	struct md_cluster_info *cinfo = thread->mddev->cluster_info;
	struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
	struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
	struct cluster_msg msg;

	/*get CR on Message*/
	if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
		pr_err("md/raid1:failed to get CR on MESSAGE\n");
		return;
	}

	/* read lvb and wake up thread to process this message_lockres */
	memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
	process_recvd_msg(thread->mddev, &msg);

	/*release CR on ack_lockres*/
	dlm_unlock_sync(ack_lockres);
	/*up-convert to EX on message_lockres*/
	dlm_lock_sync(message_lockres, DLM_LOCK_EX);
	/*get CR on ack_lockres again*/
	dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
	/*release CR on message_lockres*/
	dlm_unlock_sync(message_lockres);
}

/* lock_comm()
 * Takes the lock on the TOKEN lock resource so no other
 * node can communicate while the operation is underway.
 */
static int lock_comm(struct md_cluster_info *cinfo)
{
	int error;

	error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
	if (error)
		pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
				__func__, __LINE__, error);
	return error;
}

static void unlock_comm(struct md_cluster_info *cinfo)
{
	dlm_unlock_sync(cinfo->token_lockres);
}

/* __sendmsg()
 * This function performs the actual sending of the message. This function is
 * usually called after performing the encompassing operation
 * The function:
 * 1. Grabs the message lockresource in EX mode
 * 2. Copies the message to the message LVB
 * 3. Downconverts message lockresource to CR
 * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
 *    and the other nodes read the message. The thread will wait here until all other
 *    nodes have released ack lock resource.
 * 5. Downconvert ack lockresource to CR
 */
static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
{
	int error;
	int slot = cinfo->slot_number - 1;

	cmsg->slot = cpu_to_le32(slot);
	/*get EX on Message*/
	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
	if (error) {
		pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
		goto failed_message;
	}

	memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
			sizeof(struct cluster_msg));
	/*down-convert EX to CR on Message*/
	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CR);
	if (error) {
		pr_err("md-cluster: failed to convert EX to CR on MESSAGE(%d)\n",
				error);
		goto failed_message;
	}

	/*up-convert CR to EX on Ack*/
	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
	if (error) {
		pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
				error);
		goto failed_ack;
	}

	/*down-convert EX to CR on Ack*/
	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
	if (error) {
		pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
				error);
		goto failed_ack;
	}

failed_ack:
	dlm_unlock_sync(cinfo->message_lockres);
failed_message:
	return error;
}

static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
{
	int ret;

	lock_comm(cinfo);
	ret = __sendmsg(cinfo, cmsg);
	unlock_comm(cinfo);
	return ret;
}

static int gather_all_resync_info(struct mddev *mddev, int total_slots)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	int i, ret = 0;
	struct dlm_lock_resource *bm_lockres;
	struct suspend_info *s;
	char str[64];


	for (i = 0; i < total_slots; i++) {
		memset(str, '\0', 64);
		snprintf(str, 64, "bitmap%04d", i);
		bm_lockres = lockres_init(mddev, str, NULL, 1);
		if (!bm_lockres)
			return -ENOMEM;
		if (i == (cinfo->slot_number - 1))
			continue;

		bm_lockres->flags |= DLM_LKF_NOQUEUE;
		ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
		if (ret == -EAGAIN) {
			memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE);
			s = read_resync_info(mddev, bm_lockres);
			if (s) {
				pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
						__func__, __LINE__,
						(unsigned long long) s->lo,
						(unsigned long long) s->hi, i);
				spin_lock_irq(&cinfo->suspend_lock);
				s->slot = i;
				list_add(&s->list, &cinfo->suspend_list);
				spin_unlock_irq(&cinfo->suspend_lock);
			}
			ret = 0;
			lockres_free(bm_lockres);
			continue;
		}
		if (ret)
			goto out;
		/* TODO: Read the disk bitmap sb and check if it needs recovery */
		dlm_unlock_sync(bm_lockres);
		lockres_free(bm_lockres);
	}
out:
	return ret;
}

static int join(struct mddev *mddev, int nodes)
{
	struct md_cluster_info *cinfo;
	int ret, ops_rv;
	char str[64];

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

	cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
	if (!cinfo)
		return -ENOMEM;

	init_completion(&cinfo->completion);

	mutex_init(&cinfo->sb_mutex);
	mddev->cluster_info = cinfo;

	memset(str, 0, 64);
	pretty_uuid(str, mddev->uuid);
	ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
				DLM_LSFL_FS, LVB_SIZE,
				&md_ls_ops, mddev, &ops_rv, &cinfo->lockspace);
	if (ret)
		goto err;
	wait_for_completion(&cinfo->completion);
	if (nodes <= cinfo->slot_number) {
		pr_err("md-cluster: Slot allotted(%d) greater than available slots(%d)", cinfo->slot_number - 1,
			nodes);
		ret = -ERANGE;
		goto err;
	}
	cinfo->sb_lock = lockres_init(mddev, "cmd-super",
					NULL, 0);
	if (!cinfo->sb_lock) {
		ret = -ENOMEM;
		goto err;
	}
	/* Initiate the communication resources */
	ret = -ENOMEM;
	cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv");
	if (!cinfo->recv_thread) {
		pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
		goto err;
	}
	cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
	if (!cinfo->message_lockres)
		goto err;
	cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
	if (!cinfo->token_lockres)
		goto err;
	cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
	if (!cinfo->ack_lockres)
		goto err;
	/* get sync CR lock on ACK. */
	if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
		pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
				ret);

	pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
	snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
	cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
	if (!cinfo->bitmap_lockres)
		goto err;
	if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
		pr_err("Failed to get bitmap lock\n");
		ret = -EINVAL;
		goto err;
	}

	INIT_LIST_HEAD(&cinfo->suspend_list);
	spin_lock_init(&cinfo->suspend_lock);

	ret = gather_all_resync_info(mddev, nodes);
	if (ret)
		goto err;

	return 0;
err:
	lockres_free(cinfo->message_lockres);
	lockres_free(cinfo->token_lockres);
	lockres_free(cinfo->ack_lockres);
	lockres_free(cinfo->bitmap_lockres);
	lockres_free(cinfo->sb_lock);
	if (cinfo->lockspace)
		dlm_release_lockspace(cinfo->lockspace, 2);
	mddev->cluster_info = NULL;
	kfree(cinfo);
	module_put(THIS_MODULE);
	return ret;
}

static int leave(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;

	if (!cinfo)
		return 0;
	md_unregister_thread(&cinfo->recovery_thread);
	md_unregister_thread(&cinfo->recv_thread);
	lockres_free(cinfo->message_lockres);
	lockres_free(cinfo->token_lockres);
	lockres_free(cinfo->ack_lockres);
	lockres_free(cinfo->sb_lock);
	lockres_free(cinfo->bitmap_lockres);
	dlm_release_lockspace(cinfo->lockspace, 2);
	return 0;
}

/* slot_number(): Returns the MD slot number to use
 * DLM starts the slot numbers from 1, wheras cluster-md
 * wants the number to be from zero, so we deduct one
 */
static int slot_number(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;

	return cinfo->slot_number - 1;
}

static void resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;

	add_resync_info(mddev, cinfo->bitmap_lockres, lo, hi);
	/* Re-acquire the lock to refresh LVB */
	dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
}

static struct md_cluster_operations cluster_ops = {
	.join   = join,
	.leave  = leave,
	.slot_number = slot_number,
	.resync_info_update = resync_info_update,
};

static int __init cluster_init(void)
{
	pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n");
	pr_info("Registering Cluster MD functions\n");
	register_md_cluster_operations(&cluster_ops, THIS_MODULE);
	return 0;
}

static void cluster_exit(void)
{
	unregister_md_cluster_operations();
}

module_init(cluster_init);
module_exit(cluster_exit);
MODULE_LICENSE("GPL");
MODULE_DESCRIPTION("Clustering support for MD");
Commit	Line	Data
8e854e9c GR	1	/*
	2	* Copyright (C) 2015, SUSE
	3	*
	4	* This program is free software; you can redistribute it and/or modify
	5	* it under the terms of the GNU General Public License as published by
	6	* the Free Software Foundation; either version 2, or (at your option)
	7	* any later version.
	8	*
	9	*/
	10
	11
	12	#include <linux/module.h>
47741b7c GR	13	#include <linux/dlm.h>
	14	#include <linux/sched.h>
	15	#include "md.h"
e94987db	16	#include "bitmap.h"
edb39c9d	17	#include "md-cluster.h"
47741b7c GR	18
	19	#define LVB_SIZE 64
	20
	21	struct dlm_lock_resource {
	22	dlm_lockspace_t *ls;
	23	struct dlm_lksb lksb;
	24	char name; / lock name. */
	25	uint32_t flags; /* flags to pass to dlm_lock() */
47741b7c	26	struct completion completion; /* completion for synchronized locking */
c4ce867f GR	27	void (bast)(void arg, int mode); /* blocking AST function pointer*/
	28	struct mddev mddev; / pointing back to mddev. */
	29	};
	30
96ae923a GR	31	struct suspend_info {
	32	int slot;
	33	sector_t lo;
	34	sector_t hi;
	35	struct list_head list;
	36	};
	37
	38	struct resync_info {
	39	__le64 lo;
	40	__le64 hi;
	41	};
	42
c4ce867f GR	43	struct md_cluster_info {
	44	/* dlm lock space and resources for clustered raid. */
	45	dlm_lockspace_t *lockspace;
cf921cc1 GR	46	int slot_number;
cf921cc1 GR	47	struct completion completion;
c4ce867f GR	48	struct dlm_lock_resource *sb_lock;
c4ce867f GR	49	struct mutex sb_mutex;
54519c5f	50	struct dlm_lock_resource *bitmap_lockres;
96ae923a GR	51	struct list_head suspend_list;
96ae923a GR	52	spinlock_t suspend_lock;
e94987db GR	53	struct md_thread *recovery_thread;
e94987db GR	54	unsigned long recovery_map;
4664680c GR	55	/* communication loc resources */
	56	struct dlm_lock_resource *ack_lockres;
	57	struct dlm_lock_resource *message_lockres;
	58	struct dlm_lock_resource *token_lockres;
	59	struct md_thread *recv_thread;
	60	};
	61
	62	enum msg_type {
	63	METADATA_UPDATED = 0,
	64	RESYNCING,
	65	};
	66
	67	struct cluster_msg {
	68	int type;
	69	int slot;
	70	sector_t low;
	71	sector_t high;
47741b7c GR	72	};
	73
	74	static void sync_ast(void *arg)
	75	{
	76	struct dlm_lock_resource *res;
	77
	78	res = (struct dlm_lock_resource *) arg;
	79	complete(&res->completion);
	80	}
	81
	82	static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
	83	{
	84	int ret = 0;
	85
	86	init_completion(&res->completion);
	87	ret = dlm_lock(res->ls, mode, &res->lksb,
	88	res->flags, res->name, strlen(res->name),
	89	0, sync_ast, res, res->bast);
	90	if (ret)
	91	return ret;
	92	wait_for_completion(&res->completion);
	93	return res->lksb.sb_status;
	94	}
	95
	96	static int dlm_unlock_sync(struct dlm_lock_resource *res)
	97	{
	98	return dlm_lock_sync(res, DLM_LOCK_NL);
	99	}
	100
c4ce867f	101	static struct dlm_lock_resource lockres_init(struct mddev mddev,
47741b7c GR	102	char name, void (bastfn)(void *arg, int mode), int with_lvb)
	103	{
	104	struct dlm_lock_resource *res = NULL;
	105	int ret, namelen;
c4ce867f	106	struct md_cluster_info *cinfo = mddev->cluster_info;
47741b7c GR	107
	108	res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
	109	if (!res)
	110	return NULL;
c4ce867f GR	111	res->ls = cinfo->lockspace;
c4ce867f GR	112	res->mddev = mddev;
47741b7c GR	113	namelen = strlen(name);
	114	res->name = kzalloc(namelen + 1, GFP_KERNEL);
	115	if (!res->name) {
	116	pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
	117	goto out_err;
	118	}
	119	strlcpy(res->name, name, namelen + 1);
	120	if (with_lvb) {
	121	res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
	122	if (!res->lksb.sb_lvbptr) {
	123	pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
	124	goto out_err;
	125	}
	126	res->flags = DLM_LKF_VALBLK;
	127	}
	128
	129	if (bastfn)
	130	res->bast = bastfn;
	131
	132	res->flags \|= DLM_LKF_EXPEDITE;
	133
	134	ret = dlm_lock_sync(res, DLM_LOCK_NL);
	135	if (ret) {
	136	pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
	137	goto out_err;
	138	}
	139	res->flags &= ~DLM_LKF_EXPEDITE;
	140	res->flags \|= DLM_LKF_CONVERT;
	141
	142	return res;
	143	out_err:
	144	kfree(res->lksb.sb_lvbptr);
	145	kfree(res->name);
	146	kfree(res);
	147	return NULL;
	148	}
	149
	150	static void lockres_free(struct dlm_lock_resource *res)
	151	{
	152	if (!res)
	153	return;
	154
	155	init_completion(&res->completion);
	156	dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res);
	157	wait_for_completion(&res->completion);
	158
	159	kfree(res->name);
	160	kfree(res->lksb.sb_lvbptr);
	161	kfree(res);
	162	}
8e854e9c	163
c4ce867f GR	164	static char pretty_uuid(char dest, char *src)
	165	{
	166	int i, len = 0;
	167
	168	for (i = 0; i < 16; i++) {
	169	if (i == 4 \|\| i == 6 \|\| i == 8 \|\| i == 10)
	170	len += sprintf(dest + len, "-");
	171	len += sprintf(dest + len, "%02x", (__u8)src[i]);
	172	}
	173	return dest;
	174	}
	175
96ae923a GR	176	static void add_resync_info(struct mddev mddev, struct dlm_lock_resource lockres,
	177	sector_t lo, sector_t hi)
	178	{
	179	struct resync_info *ri;
	180
	181	ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
	182	ri->lo = cpu_to_le64(lo);
	183	ri->hi = cpu_to_le64(hi);
	184	}
	185
	186	static struct suspend_info read_resync_info(struct mddev mddev, struct dlm_lock_resource *lockres)
	187	{
	188	struct resync_info ri;
	189	struct suspend_info *s = NULL;
	190	sector_t hi = 0;
	191
	192	dlm_lock_sync(lockres, DLM_LOCK_CR);
	193	memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
	194	hi = le64_to_cpu(ri.hi);
	195	if (ri.hi > 0) {
	196	s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
	197	if (!s)
	198	goto out;
	199	s->hi = hi;
	200	s->lo = le64_to_cpu(ri.lo);
	201	}
	202	dlm_unlock_sync(lockres);
	203	out:
	204	return s;
	205	}
	206
e94987db GR	207	void recover_bitmaps(struct md_thread *thread)
	208	{
	209	struct mddev *mddev = thread->mddev;
	210	struct md_cluster_info *cinfo = mddev->cluster_info;
	211	struct dlm_lock_resource *bm_lockres;
	212	char str[64];
	213	int slot, ret;
	214	struct suspend_info s, tmp;
	215	sector_t lo, hi;
	216
	217	while (cinfo->recovery_map) {
	218	slot = fls64((u64)cinfo->recovery_map) - 1;
	219
	220	/* Clear suspend_area associated with the bitmap */
	221	spin_lock_irq(&cinfo->suspend_lock);
	222	list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
	223	if (slot == s->slot) {
	224	list_del(&s->list);
	225	kfree(s);
	226	}
	227	spin_unlock_irq(&cinfo->suspend_lock);
	228
	229	snprintf(str, 64, "bitmap%04d", slot);
	230	bm_lockres = lockres_init(mddev, str, NULL, 1);
	231	if (!bm_lockres) {
	232	pr_err("md-cluster: Cannot initialize bitmaps\n");
	233	goto clear_bit;
	234	}
	235
	236	ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
	237	if (ret) {
	238	pr_err("md-cluster: Could not DLM lock %s: %d\n",
	239	str, ret);
	240	goto clear_bit;
	241	}
	242	ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi);
4b26a08a	243	if (ret) {
e94987db	244	pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
4b26a08a GR	245	goto dlm_unlock;
	246	}
	247	if (hi > 0) {
	248	/* TODO:Wait for current resync to get over */
	249	set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
	250	if (lo < mddev->recovery_cp)
	251	mddev->recovery_cp = lo;
	252	md_check_recovery(mddev);
	253	}
	254	dlm_unlock:
e94987db GR	255	dlm_unlock_sync(bm_lockres);
	256	clear_bit:
	257	clear_bit(slot, &cinfo->recovery_map);
	258	}
	259	}
	260
cf921cc1 GR	261	static void recover_prep(void *arg)
	262	{
	263	}
	264
	265	static void recover_slot(void arg, struct dlm_slot slot)
	266	{
	267	struct mddev *mddev = arg;
	268	struct md_cluster_info *cinfo = mddev->cluster_info;
	269
	270	pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
	271	mddev->bitmap_info.cluster_name,
	272	slot->nodeid, slot->slot,
	273	cinfo->slot_number);
e94987db GR	274	set_bit(slot->slot - 1, &cinfo->recovery_map);
	275	if (!cinfo->recovery_thread) {
	276	cinfo->recovery_thread = md_register_thread(recover_bitmaps,
	277	mddev, "recover");
	278	if (!cinfo->recovery_thread) {
	279	pr_warn("md-cluster: Could not create recovery thread\n");
	280	return;
	281	}
	282	}
	283	md_wakeup_thread(cinfo->recovery_thread);
cf921cc1 GR	284	}
	285
	286	static void recover_done(void arg, struct dlm_slot slots,
	287	int num_slots, int our_slot,
	288	uint32_t generation)
	289	{
	290	struct mddev *mddev = arg;
	291	struct md_cluster_info *cinfo = mddev->cluster_info;
	292
	293	cinfo->slot_number = our_slot;
	294	complete(&cinfo->completion);
	295	}
	296
	297	static const struct dlm_lockspace_ops md_ls_ops = {
	298	.recover_prep = recover_prep,
	299	.recover_slot = recover_slot,
	300	.recover_done = recover_done,
	301	};
	302
4664680c GR	303	/*
	304	* The BAST function for the ack lock resource
	305	* This function wakes up the receive thread in
	306	* order to receive and process the message.
	307	*/
	308	static void ack_bast(void *arg, int mode)
	309	{
	310	struct dlm_lock_resource res = (struct dlm_lock_resource )arg;
	311	struct md_cluster_info *cinfo = res->mddev->cluster_info;
	312
	313	if (mode == DLM_LOCK_EX)
	314	md_wakeup_thread(cinfo->recv_thread);
	315	}
	316
	317	static void process_recvd_msg(struct mddev mddev, struct cluster_msg msg)
	318	{
	319	switch (msg->type) {
	320	case METADATA_UPDATED:
	321	pr_info("%s: %d Received message: METADATA_UPDATE from %d\n",
	322	__func__, __LINE__, msg->slot);
	323	break;
	324	case RESYNCING:
	325	pr_info("%s: %d Received message: RESYNCING from %d\n",
	326	__func__, __LINE__, msg->slot);
	327	break;
	328	};
	329	}
	330
	331	/*
	332	* thread for receiving message
	333	*/
	334	static void recv_daemon(struct md_thread *thread)
	335	{
	336	struct md_cluster_info *cinfo = thread->mddev->cluster_info;
	337	struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
	338	struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
	339	struct cluster_msg msg;
	340
	341	/get CR on Message/
	342	if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
	343	pr_err("md/raid1:failed to get CR on MESSAGE\n");
	344	return;
	345	}
	346
	347	/* read lvb and wake up thread to process this message_lockres */
	348	memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
	349	process_recvd_msg(thread->mddev, &msg);
	350
	351	/release CR on ack_lockres/
	352	dlm_unlock_sync(ack_lockres);
	353	/up-convert to EX on message_lockres/
	354	dlm_lock_sync(message_lockres, DLM_LOCK_EX);
	355	/get CR on ack_lockres again/
	356	dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
	357	/release CR on message_lockres/
	358	dlm_unlock_sync(message_lockres);
	359	}
	360
601b515c GR	361	/* lock_comm()
	362	* Takes the lock on the TOKEN lock resource so no other
	363	* node can communicate while the operation is underway.
	364	*/
	365	static int lock_comm(struct md_cluster_info *cinfo)
	366	{
	367	int error;
	368
	369	error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
	370	if (error)
	371	pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
	372	__func__, __LINE__, error);
	373	return error;
	374	}
	375
	376	static void unlock_comm(struct md_cluster_info *cinfo)
	377	{
	378	dlm_unlock_sync(cinfo->token_lockres);
	379	}
	380
	381	/* __sendmsg()
	382	* This function performs the actual sending of the message. This function is
	383	* usually called after performing the encompassing operation
	384	* The function:
	385	* 1. Grabs the message lockresource in EX mode
	386	* 2. Copies the message to the message LVB
	387	* 3. Downconverts message lockresource to CR
	388	* 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
	389	* and the other nodes read the message. The thread will wait here until all other
	390	* nodes have released ack lock resource.
	391	* 5. Downconvert ack lockresource to CR
	392	*/
	393	static int __sendmsg(struct md_cluster_info cinfo, struct cluster_msg cmsg)
	394	{
	395	int error;
	396	int slot = cinfo->slot_number - 1;
	397
	398	cmsg->slot = cpu_to_le32(slot);
	399	/get EX on Message/
	400	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
	401	if (error) {
	402	pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
	403	goto failed_message;
	404	}
	405
	406	memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
	407	sizeof(struct cluster_msg));
	408	/down-convert EX to CR on Message/
	409	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CR);
	410	if (error) {
	411	pr_err("md-cluster: failed to convert EX to CR on MESSAGE(%d)\n",
	412	error);
	413	goto failed_message;
	414	}
	415
	416	/up-convert CR to EX on Ack/
	417	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
	418	if (error) {
	419	pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
	420	error);
	421	goto failed_ack;
	422	}
	423
	424	/down-convert EX to CR on Ack/
425	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
426	if (error) {
427	pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
428	error);
429	goto failed_ack;
430	}
431
432	failed_ack:
433	dlm_unlock_sync(cinfo->message_lockres);
434	failed_message:
435	return error;
436	}
437
438	static int sendmsg(struct md_cluster_info cinfo, struct cluster_msg cmsg)
439	{
440	int ret;
441
442	lock_comm(cinfo);
443	ret = __sendmsg(cinfo, cmsg);
444	unlock_comm(cinfo);
445	return ret;
446	}
447
96ae923a GR	448	static int gather_all_resync_info(struct mddev *mddev, int total_slots)
	449	{
	450	struct md_cluster_info *cinfo = mddev->cluster_info;
	451	int i, ret = 0;
	452	struct dlm_lock_resource *bm_lockres;
	453	struct suspend_info *s;
	454	char str[64];
	455
	456
	457	for (i = 0; i < total_slots; i++) {
	458	memset(str, '\0', 64);
	459	snprintf(str, 64, "bitmap%04d", i);
	460	bm_lockres = lockres_init(mddev, str, NULL, 1);
	461	if (!bm_lockres)
	462	return -ENOMEM;
	463	if (i == (cinfo->slot_number - 1))
	464	continue;
	465
	466	bm_lockres->flags \|= DLM_LKF_NOQUEUE;
	467	ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
	468	if (ret == -EAGAIN) {
	469	memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE);
	470	s = read_resync_info(mddev, bm_lockres);
	471	if (s) {
	472	pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
	473	__func__, __LINE__,
	474	(unsigned long long) s->lo,
	475	(unsigned long long) s->hi, i);
	476	spin_lock_irq(&cinfo->suspend_lock);
	477	s->slot = i;
	478	list_add(&s->list, &cinfo->suspend_list);
	479	spin_unlock_irq(&cinfo->suspend_lock);
	480	}
	481	ret = 0;
	482	lockres_free(bm_lockres);
	483	continue;
	484	}
	485	if (ret)
	486	goto out;
	487	/* TODO: Read the disk bitmap sb and check if it needs recovery */
	488	dlm_unlock_sync(bm_lockres);
	489	lockres_free(bm_lockres);
	490	}
	491	out:
	492	return ret;
	493	}
	494
edb39c9d GR	495	static int join(struct mddev *mddev, int nodes)
edb39c9d GR	496	{
c4ce867f	497	struct md_cluster_info *cinfo;
cf921cc1	498	int ret, ops_rv;
c4ce867f GR	499	char str[64];
	500
	501	if (!try_module_get(THIS_MODULE))
	502	return -ENOENT;
	503
	504	cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
	505	if (!cinfo)
	506	return -ENOMEM;
	507
cf921cc1 GR	508	init_completion(&cinfo->completion);
	509
	510	mutex_init(&cinfo->sb_mutex);
	511	mddev->cluster_info = cinfo;
	512
c4ce867f GR	513	memset(str, 0, 64);
c4ce867f GR	514	pretty_uuid(str, mddev->uuid);
cf921cc1 GR	515	ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
	516	DLM_LSFL_FS, LVB_SIZE,
	517	&md_ls_ops, mddev, &ops_rv, &cinfo->lockspace);
c4ce867f GR	518	if (ret)
c4ce867f GR	519	goto err;
cf921cc1	520	wait_for_completion(&cinfo->completion);
b97e9257 GR	521	if (nodes <= cinfo->slot_number) {
	522	pr_err("md-cluster: Slot allotted(%d) greater than available slots(%d)", cinfo->slot_number - 1,
	523	nodes);
	524	ret = -ERANGE;
	525	goto err;
	526	}
c4ce867f GR	527	cinfo->sb_lock = lockres_init(mddev, "cmd-super",
	528	NULL, 0);
	529	if (!cinfo->sb_lock) {
	530	ret = -ENOMEM;
	531	goto err;
	532	}
4664680c GR	533	/* Initiate the communication resources */
	534	ret = -ENOMEM;
	535	cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv");
	536	if (!cinfo->recv_thread) {
	537	pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
	538	goto err;
	539	}
	540	cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
	541	if (!cinfo->message_lockres)
	542	goto err;
	543	cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
	544	if (!cinfo->token_lockres)
	545	goto err;
	546	cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
	547	if (!cinfo->ack_lockres)
	548	goto err;
	549	/* get sync CR lock on ACK. */
	550	if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
	551	pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
	552	ret);
54519c5f GR	553
	554	pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
	555	snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
	556	cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
	557	if (!cinfo->bitmap_lockres)
	558	goto err;
	559	if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
	560	pr_err("Failed to get bitmap lock\n");
	561	ret = -EINVAL;
	562	goto err;
	563	}
	564
96ae923a GR	565	INIT_LIST_HEAD(&cinfo->suspend_list);
	566	spin_lock_init(&cinfo->suspend_lock);
	567
	568	ret = gather_all_resync_info(mddev, nodes);
	569	if (ret)
	570	goto err;
	571
edb39c9d	572	return 0;
c4ce867f	573	err:
4664680c GR	574	lockres_free(cinfo->message_lockres);
	575	lockres_free(cinfo->token_lockres);
	576	lockres_free(cinfo->ack_lockres);
96ae923a GR	577	lockres_free(cinfo->bitmap_lockres);
96ae923a GR	578	lockres_free(cinfo->sb_lock);
c4ce867f GR	579	if (cinfo->lockspace)
c4ce867f GR	580	dlm_release_lockspace(cinfo->lockspace, 2);
cf921cc1	581	mddev->cluster_info = NULL;
c4ce867f GR	582	kfree(cinfo);
	583	module_put(THIS_MODULE);
	584	return ret;
edb39c9d GR	585	}
	586
	587	static int leave(struct mddev *mddev)
	588	{
c4ce867f GR	589	struct md_cluster_info *cinfo = mddev->cluster_info;
	590
	591	if (!cinfo)
	592	return 0;
e94987db	593	md_unregister_thread(&cinfo->recovery_thread);
4664680c GR	594	md_unregister_thread(&cinfo->recv_thread);
	595	lockres_free(cinfo->message_lockres);
	596	lockres_free(cinfo->token_lockres);
	597	lockres_free(cinfo->ack_lockres);
c4ce867f	598	lockres_free(cinfo->sb_lock);
54519c5f	599	lockres_free(cinfo->bitmap_lockres);
c4ce867f	600	dlm_release_lockspace(cinfo->lockspace, 2);
edb39c9d GR	601	return 0;
	602	}
	603
cf921cc1 GR	604	/* slot_number(): Returns the MD slot number to use
	605	* DLM starts the slot numbers from 1, wheras cluster-md
	606	* wants the number to be from zero, so we deduct one
	607	*/
	608	static int slot_number(struct mddev *mddev)
	609	{
	610	struct md_cluster_info *cinfo = mddev->cluster_info;
	611
	612	return cinfo->slot_number - 1;
	613	}
	614
96ae923a GR	615	static void resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
	616	{
	617	struct md_cluster_info *cinfo = mddev->cluster_info;
	618
	619	add_resync_info(mddev, cinfo->bitmap_lockres, lo, hi);
	620	/* Re-acquire the lock to refresh LVB */
	621	dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
	622	}
	623
edb39c9d GR	624	static struct md_cluster_operations cluster_ops = {
	625	.join = join,
	626	.leave = leave,
cf921cc1	627	.slot_number = slot_number,
96ae923a	628	.resync_info_update = resync_info_update,
edb39c9d GR	629	};
edb39c9d GR	630
8e854e9c GR	631	static int __init cluster_init(void)
	632	{
	633	pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n");
	634	pr_info("Registering Cluster MD functions\n");
edb39c9d	635	register_md_cluster_operations(&cluster_ops, THIS_MODULE);
8e854e9c GR	636	return 0;
	637	}
	638
	639	static void cluster_exit(void)
	640	{
edb39c9d	641	unregister_md_cluster_operations();
8e854e9c GR	642	}
	643
	644	module_init(cluster_init);
	645	module_exit(cluster_exit);
	646	MODULE_LICENSE("GPL");
	647	MODULE_DESCRIPTION("Clustering support for MD");