/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * Copyright (C) 2004, 2005, 2012 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; either
 * version 2 of the License, or (at your option) any later version.
 *
 * 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.
 *
 * You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 021110-1307, USA.
 */

#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/configfs.h>

#include "heartbeat.h"
#include "tcp.h"
#include "nodemanager.h"

#include "masklog.h"

/*
 * The first heartbeat pass had one global thread that would serialize all hb
 * callback calls.  This global serializing sem should only be removed once
 * we've made sure that all callees can deal with being called concurrently
 * from multiple hb region threads.
 */
static DECLARE_RWSEM(r2hb_callback_sem);

/*
 * multiple hb threads are watching multiple regions.  A node is live
 * whenever any of the threads sees activity from the node in its region.
 */
static DEFINE_SPINLOCK(r2hb_live_lock);
static unsigned long r2hb_live_node_bitmap[BITS_TO_LONGS(R2NM_MAX_NODES)];

static struct r2hb_callback {
	struct list_head list;
} r2hb_callbacks[R2HB_NUM_CB];

enum r2hb_heartbeat_modes {
	R2HB_HEARTBEAT_LOCAL		= 0,
	R2HB_HEARTBEAT_GLOBAL,
	R2HB_HEARTBEAT_NUM_MODES,
};

char *r2hb_heartbeat_mode_desc[R2HB_HEARTBEAT_NUM_MODES] = {
		"local",	/* R2HB_HEARTBEAT_LOCAL */
		"global",	/* R2HB_HEARTBEAT_GLOBAL */
};

unsigned int r2hb_dead_threshold = R2HB_DEFAULT_DEAD_THRESHOLD;
unsigned int r2hb_heartbeat_mode = R2HB_HEARTBEAT_LOCAL;

/* Only sets a new threshold if there are no active regions.
 *
 * No locking or otherwise interesting code is required for reading
 * r2hb_dead_threshold as it can't change once regions are active and
 * it's not interesting to anyone until then anyway. */
static void r2hb_dead_threshold_set(unsigned int threshold)
{
	if (threshold > R2HB_MIN_DEAD_THRESHOLD) {
		spin_lock(&r2hb_live_lock);
		r2hb_dead_threshold = threshold;
		spin_unlock(&r2hb_live_lock);
	}
}

static int r2hb_global_hearbeat_mode_set(unsigned int hb_mode)
{
	int ret = -1;

	if (hb_mode < R2HB_HEARTBEAT_NUM_MODES) {
		spin_lock(&r2hb_live_lock);
		r2hb_heartbeat_mode = hb_mode;
		ret = 0;
		spin_unlock(&r2hb_live_lock);
	}

	return ret;
}

void r2hb_exit(void)
{
}

int r2hb_init(void)
{
	int i;

	for (i = 0; i < ARRAY_SIZE(r2hb_callbacks); i++)
		INIT_LIST_HEAD(&r2hb_callbacks[i].list);

	memset(r2hb_live_node_bitmap, 0, sizeof(r2hb_live_node_bitmap));

	return 0;
}

/* if we're already in a callback then we're already serialized by the sem */
static void r2hb_fill_node_map_from_callback(unsigned long *map,
					     unsigned bytes)
{
	BUG_ON(bytes < (BITS_TO_LONGS(R2NM_MAX_NODES) * sizeof(unsigned long)));

	memcpy(map, &r2hb_live_node_bitmap, bytes);
}

/*
 * get a map of all nodes that are heartbeating in any regions
 */
void r2hb_fill_node_map(unsigned long *map, unsigned bytes)
{
	/* callers want to serialize this map and callbacks so that they
	 * can trust that they don't miss nodes coming to the party */
	down_read(&r2hb_callback_sem);
	spin_lock(&r2hb_live_lock);
	r2hb_fill_node_map_from_callback(map, bytes);
	spin_unlock(&r2hb_live_lock);
	up_read(&r2hb_callback_sem);
}
EXPORT_SYMBOL_GPL(r2hb_fill_node_map);

/*
 * heartbeat configfs bits.  The heartbeat set is a default set under
 * the cluster set in nodemanager.c.
 */

/* heartbeat set */

struct r2hb_hb_group {
	struct config_group hs_group;
	/* some stuff? */
};

static struct r2hb_hb_group *to_r2hb_hb_group(struct config_group *group)
{
	return group ?
		container_of(group, struct r2hb_hb_group, hs_group)
		: NULL;
}

static struct config_item r2hb_config_item;

static struct config_item *r2hb_hb_group_make_item(struct config_group *group,
							  const char *name)
{
	int ret;

	if (strlen(name) > R2HB_MAX_REGION_NAME_LEN) {
		ret = -ENAMETOOLONG;
		goto free;
	}

	config_item_put(&r2hb_config_item);

	return &r2hb_config_item;
free:
	return ERR_PTR(ret);
}

static void r2hb_hb_group_drop_item(struct config_group *group,
					   struct config_item *item)
{
	if (r2hb_global_heartbeat_active()) {
		printk(KERN_NOTICE "ramster: Heartbeat %s "
			"on region %s (%s)\n",
			"stopped/aborted", config_item_name(item),
			"no region");
	}

	config_item_put(item);
}

struct r2hb_hb_group_attribute {
	struct configfs_attribute attr;
	ssize_t (*show)(struct r2hb_hb_group *, char *);
	ssize_t (*store)(struct r2hb_hb_group *, const char *, size_t);
};

static ssize_t r2hb_hb_group_show(struct config_item *item,
					 struct configfs_attribute *attr,
					 char *page)
{
	struct r2hb_hb_group *reg = to_r2hb_hb_group(to_config_group(item));
	struct r2hb_hb_group_attribute *r2hb_hb_group_attr =
		container_of(attr, struct r2hb_hb_group_attribute, attr);
	ssize_t ret = 0;

	if (r2hb_hb_group_attr->show)
		ret = r2hb_hb_group_attr->show(reg, page);
	return ret;
}

static ssize_t r2hb_hb_group_store(struct config_item *item,
					  struct configfs_attribute *attr,
					  const char *page, size_t count)
{
	struct r2hb_hb_group *reg = to_r2hb_hb_group(to_config_group(item));
	struct r2hb_hb_group_attribute *r2hb_hb_group_attr =
		container_of(attr, struct r2hb_hb_group_attribute, attr);
	ssize_t ret = -EINVAL;

	if (r2hb_hb_group_attr->store)
		ret = r2hb_hb_group_attr->store(reg, page, count);
	return ret;
}

static ssize_t r2hb_hb_group_threshold_show(struct r2hb_hb_group *group,
						     char *page)
{
	return sprintf(page, "%u\n", r2hb_dead_threshold);
}

static ssize_t r2hb_hb_group_threshold_store(struct r2hb_hb_group *group,
						    const char *page,
						    size_t count)
{
	unsigned long tmp;
	char *p = (char *)page;
	int err;

	err = kstrtoul(p, 10, &tmp);
	if (err)
		return err;

	/* this will validate ranges for us. */
	r2hb_dead_threshold_set((unsigned int) tmp);

	return count;
}

static
ssize_t r2hb_hb_group_mode_show(struct r2hb_hb_group *group,
				       char *page)
{
	return sprintf(page, "%s\n",
		       r2hb_heartbeat_mode_desc[r2hb_heartbeat_mode]);
}

static
ssize_t r2hb_hb_group_mode_store(struct r2hb_hb_group *group,
					const char *page, size_t count)
{
	unsigned int i;
	int ret;
	size_t len;

	len = (page[count - 1] == '\n') ? count - 1 : count;
	if (!len)
		return -EINVAL;

	for (i = 0; i < R2HB_HEARTBEAT_NUM_MODES; ++i) {
		if (strnicmp(page, r2hb_heartbeat_mode_desc[i], len))
			continue;

		ret = r2hb_global_hearbeat_mode_set(i);
		if (!ret)
			printk(KERN_NOTICE "ramster: Heartbeat mode "
						"set to %s\n",
			       r2hb_heartbeat_mode_desc[i]);
		return count;
	}

	return -EINVAL;

}

static struct r2hb_hb_group_attribute r2hb_hb_group_attr_threshold = {
	.attr	= { .ca_owner = THIS_MODULE,
		    .ca_name = "dead_threshold",
		    .ca_mode = S_IRUGO | S_IWUSR },
	.show	= r2hb_hb_group_threshold_show,
	.store	= r2hb_hb_group_threshold_store,
};

static struct r2hb_hb_group_attribute r2hb_hb_group_attr_mode = {
	.attr   = { .ca_owner = THIS_MODULE,
		.ca_name = "mode",
		.ca_mode = S_IRUGO | S_IWUSR },
	.show   = r2hb_hb_group_mode_show,
	.store  = r2hb_hb_group_mode_store,
};

static struct configfs_attribute *r2hb_hb_group_attrs[] = {
	&r2hb_hb_group_attr_threshold.attr,
	&r2hb_hb_group_attr_mode.attr,
	NULL,
};

static struct configfs_item_operations r2hb_hearbeat_group_item_ops = {
	.show_attribute		= r2hb_hb_group_show,
	.store_attribute	= r2hb_hb_group_store,
};

static struct configfs_group_operations r2hb_hb_group_group_ops = {
	.make_item	= r2hb_hb_group_make_item,
	.drop_item	= r2hb_hb_group_drop_item,
};

static struct config_item_type r2hb_hb_group_type = {
	.ct_group_ops	= &r2hb_hb_group_group_ops,
	.ct_item_ops	= &r2hb_hearbeat_group_item_ops,
	.ct_attrs	= r2hb_hb_group_attrs,
	.ct_owner	= THIS_MODULE,
};

/* this is just here to avoid touching group in heartbeat.h which the
 * entire damn world #includes */
struct config_group *r2hb_alloc_hb_set(void)
{
	struct r2hb_hb_group *hs = NULL;
	struct config_group *ret = NULL;

	hs = kzalloc(sizeof(struct r2hb_hb_group), GFP_KERNEL);
	if (hs == NULL)
		goto out;

	config_group_init_type_name(&hs->hs_group, "heartbeat",
				    &r2hb_hb_group_type);

	ret = &hs->hs_group;
out:
	if (ret == NULL)
		kfree(hs);
	return ret;
}

void r2hb_free_hb_set(struct config_group *group)
{
	struct r2hb_hb_group *hs = to_r2hb_hb_group(group);
	kfree(hs);
}

/* hb callback registration and issuing */

static struct r2hb_callback *hbcall_from_type(enum r2hb_callback_type type)
{
	if (type == R2HB_NUM_CB)
		return ERR_PTR(-EINVAL);

	return &r2hb_callbacks[type];
}

void r2hb_setup_callback(struct r2hb_callback_func *hc,
			 enum r2hb_callback_type type,
			 r2hb_cb_func *func,
			 void *data,
			 int priority)
{
	INIT_LIST_HEAD(&hc->hc_item);
	hc->hc_func = func;
	hc->hc_data = data;
	hc->hc_priority = priority;
	hc->hc_type = type;
	hc->hc_magic = R2HB_CB_MAGIC;
}
EXPORT_SYMBOL_GPL(r2hb_setup_callback);

int r2hb_register_callback(const char *region_uuid,
			   struct r2hb_callback_func *hc)
{
	struct r2hb_callback_func *tmp;
	struct list_head *iter;
	struct r2hb_callback *hbcall;
	int ret;

	BUG_ON(hc->hc_magic != R2HB_CB_MAGIC);
	BUG_ON(!list_empty(&hc->hc_item));

	hbcall = hbcall_from_type(hc->hc_type);
	if (IS_ERR(hbcall)) {
		ret = PTR_ERR(hbcall);
		goto out;
	}

	down_write(&r2hb_callback_sem);

	list_for_each(iter, &hbcall->list) {
		tmp = list_entry(iter, struct r2hb_callback_func, hc_item);
		if (hc->hc_priority < tmp->hc_priority) {
			list_add_tail(&hc->hc_item, iter);
			break;
		}
	}
	if (list_empty(&hc->hc_item))
		list_add_tail(&hc->hc_item, &hbcall->list);

	up_write(&r2hb_callback_sem);
	ret = 0;
out:
	mlog(ML_CLUSTER, "returning %d on behalf of %p for funcs %p\n",
	     ret, __builtin_return_address(0), hc);
	return ret;
}
EXPORT_SYMBOL_GPL(r2hb_register_callback);

void r2hb_unregister_callback(const char *region_uuid,
			      struct r2hb_callback_func *hc)
{
	BUG_ON(hc->hc_magic != R2HB_CB_MAGIC);

	mlog(ML_CLUSTER, "on behalf of %p for funcs %p\n",
	     __builtin_return_address(0), hc);

	/* XXX Can this happen _with_ a region reference? */
	if (list_empty(&hc->hc_item))
		return;

	down_write(&r2hb_callback_sem);

	list_del_init(&hc->hc_item);

	up_write(&r2hb_callback_sem);
}
EXPORT_SYMBOL_GPL(r2hb_unregister_callback);

int r2hb_check_node_heartbeating_from_callback(u8 node_num)
{
	unsigned long testing_map[BITS_TO_LONGS(R2NM_MAX_NODES)];

	r2hb_fill_node_map_from_callback(testing_map, sizeof(testing_map));
	if (!test_bit(node_num, testing_map)) {
		mlog(ML_HEARTBEAT,
		     "node (%u) does not have heartbeating enabled.\n",
		     node_num);
		return 0;
	}

	return 1;
}
EXPORT_SYMBOL_GPL(r2hb_check_node_heartbeating_from_callback);

void r2hb_stop_all_regions(void)
{
}
EXPORT_SYMBOL_GPL(r2hb_stop_all_regions);

/*
 * this is just a hack until we get the plumbing which flips file systems
 * read only and drops the hb ref instead of killing the node dead.
 */
int r2hb_global_heartbeat_active(void)
{
	return (r2hb_heartbeat_mode == R2HB_HEARTBEAT_GLOBAL);
}
EXPORT_SYMBOL(r2hb_global_heartbeat_active);

/* added for RAMster */
void r2hb_manual_set_node_heartbeating(int node_num)
{
	if (node_num < R2NM_MAX_NODES)
		set_bit(node_num, r2hb_live_node_bitmap);
}
EXPORT_SYMBOL(r2hb_manual_set_node_heartbeating);