[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]

Re: [Linux-cluster] [RFC] Generic Kernel API



Ok this should make things a little clearer...working code!

The files should be self-explanatory but just in case:


cluster.h    generic header file for clients to include
cluster.c    implementation of the registry
clcman.c     CMAN implementation of the API
test.c       a test program showing calls of some of the APIs


-- 

patrick
#define MAX_ADDR_LEN 128
#define MAX_NAME_LEN  64

struct cluster_instance;

struct membership_node_address
{
	int32_t mna_len;
	char    mna_address[MAX_ADDR_LEN];
};

struct membership_node
{
	int32_t				mn_nodeid;
        struct membership_node_address  mn_address;
	char				mn_name[MAX_NAME_LEN];
	uint32_t			mn_member;
	struct timeval			mn_boottime;
};

struct membership_notify_info
{
	void *		mni_context;
	uint32_t	mni_viewnumber;
	uint32_t	mni_numitems;
	uint32_t	mni_nummembers;
	char *		mni_buffer;
};

/* This is what is called by membership services as a callback */
typedef int (*membership_callback_t) (struct membership_notify_info *info);

struct membership_ops
{
        int (*start_notify) (struct cluster_instance *ci,
			     void *context, uint32_t flags,
			     membership_callback_t callback,
			     char *buffer, int max_items);
#define	MEMBERSHIP_FLAGS_NOTIFY_CHANGES  1 /* Notify of membership changes */
#define	MEMBERSHIP_FLAGS_NOTIFY_NODES    2 /* Send me a full node list now */

        int (*stop_notify)  (struct cluster_instance *ci);
	int (*get_name)     (struct cluster_instance *ci, char *name, int maxlen);
	int (*get_node)     (struct cluster_instance *ci,int32_t nodeid, struct membership_node *node);
#define MEMBERSHIP_NODE_THISNODE        -1 /* Get info about local node */

};


struct quorum_info
{
	uint32_t qi_total_votes;
	uint32_t qi_expected_votes;
	uint32_t qi_quorum;
};

struct quorum_ops
{

	int (*get_quorate) (struct cluster_instance *ci);
	int (*get_votes)   (struct cluster_instance *ci, int32_t nodeid);
	int (*get_info)    (struct cluster_instance *ci, struct quorum_info *info);
};

/* Bottom interface */


struct cluster_instance
{
    struct cm_proto        *ci_proto;
    struct cluster_manager *ci_mgr;
    void                   *ci_private;
};

/* A CM sets up one of these structs with the functions it can provide and
 * registers it, along with its name (type) using cm_register() */

struct cm_proto {
        char co_proto_name[MAX_NAME_LEN];

        /* These are required */

        int (*co_attach) (struct cluster_instance *ci);
        int (*co_detach) (struct cluster_instance *ci);

        /* These are optional, a CM may provide some or all */

        struct comms_ops      *co_cops;
        struct membership_ops *co_mops;
        struct quorum_ops     *co_qops;

        struct module         *co_owner;
};


/* When a CM module is loaded it calls cm_register()
 * which adds its proto_name/ops pair to a global list. */

int cm_register(struct cm_proto *proto);
int cm_unregister(struct cm_proto *proto);

/* Attach an instance of a cluster manager */
int cm_detach(struct cluster_instance *ci);
int cm_attach(char *name, struct cluster_instance **retci);
#define EXPORT_SYMTAB
#include <linux/list.h>
#include <linux/module.h>
#include <asm/errno.h>
#include <asm/atomic.h>
#include <linux/spinlock.h>
#include <linux/string.h>
#include <linux/slab.h>

#include "cluster.h"

struct cluster_manager
{
    struct list_head cm_list;
    struct cm_proto *cm_proto;
    atomic_t         cm_refcnt;
};


static LIST_HEAD(cluster_mgr_list);
static spinlock_t cluster_mgr_lock = SPIN_LOCK_UNLOCKED;


static struct cluster_manager *cmgr_by_name(char *name)
{
    struct cluster_manager *cmgr = NULL;

    spin_lock(&cluster_mgr_lock);
    list_for_each_entry(cmgr, &cluster_mgr_list, cm_list) {
	if (strcmp(name, cmgr->cm_proto->co_proto_name) == 0)
	    goto finish;
    }

    cmgr = NULL;

 finish:
    spin_unlock(&cluster_mgr_lock);
    return cmgr;
}

int cm_register(struct cm_proto *proto)
{
    struct cluster_manager *cmgr;
    struct cluster_manager *lcmgr;
    int ret = -EAGAIN;

    if (!try_module_get(proto->co_owner))
	goto finish;

    ret = -EINVAL;
    if (!proto->co_attach || !proto->co_detach)
	goto finish;

    ret = -ENOMEM;
    cmgr = kmalloc(sizeof(struct cluster_manager), GFP_KERNEL);
    if (!cmgr)
	goto finish_mod;

    ret = -EEXIST;

    spin_lock(&cluster_mgr_lock);
    list_for_each_entry(lcmgr, &cluster_mgr_list, cm_list) {
	if (proto == lcmgr->cm_proto ||
	    strcmp(proto->co_proto_name, lcmgr->cm_proto->co_proto_name) == 0)
	    goto finish_free;
    }

    ret = 0;
    cmgr->cm_proto = proto;
    atomic_set(&cmgr->cm_refcnt, 0);

    list_add(&cmgr->cm_list, &cluster_mgr_list);

    spin_unlock(&cluster_mgr_lock);
    return ret;

 finish_free:
    spin_unlock(&cluster_mgr_lock);
    kfree(cmgr);

 finish_mod:
    module_put(proto->co_owner);

 finish:
    return ret;
}

int cm_unregister(struct cm_proto *proto)
{
    struct cluster_manager *cmgr;
    int ret = -EBUSY;

    spin_lock(&cluster_mgr_lock);
    list_for_each_entry(cmgr, &cluster_mgr_list, cm_list) {
	if (proto == cmgr->cm_proto)
	    goto got_it;
    }

    ret = -ENOENT;
    goto finish;

 got_it:
    ret = -EBUSY;
    if (atomic_read(&cmgr->cm_refcnt) != 0)
	goto finish;

    list_del(&cmgr->cm_list);
    module_put(proto->co_owner);
    kfree(cmgr);

    ret = 0;

 finish:
    spin_unlock(&cluster_mgr_lock);

    return ret;
}

int cm_attach(char *name, struct cluster_instance **retci)
{
    struct cluster_manager *cmgr;
    struct cluster_instance *ci;
    int ret;

    cmgr = cmgr_by_name(name);
    if (!cmgr)
	return -ENOENT;

    atomic_inc(&cmgr->cm_refcnt);

    ret = -ENOMEM;
    ci = kmalloc(sizeof(struct cluster_instance), GFP_KERNEL);
    if (!ci)
	goto dec_ret;

    ci->ci_proto = cmgr->cm_proto;
    ci->ci_mgr = cmgr;
    ret = cmgr->cm_proto->co_attach(ci);

    *retci = ci;

    return 0;

 dec_ret:
    atomic_dec(&cmgr->cm_refcnt);

    return ret;
}


int cm_detach(struct cluster_instance *ci)
{
    int ret;

    ret = ci->ci_mgr->cm_proto->co_detach(ci);

    if (ret) {
	atomic_dec(&ci->ci_mgr->cm_refcnt);
	kfree(ci);
    }

    return ret;

}

EXPORT_SYMBOL(cm_attach);
EXPORT_SYMBOL(cm_detach);
EXPORT_SYMBOL(cm_register);
EXPORT_SYMBOL(cm_unregister);

#ifdef MODULE
MODULE_DESCRIPTION("Generic Cluster API");
MODULE_AUTHOR("Red Hat, Inc");
MODULE_LICENSE("GPL");
#endif
/* Example cman interface via cluster ops */

#include <linux/init.h>
#include <linux/socket.h>
#include <linux/kernel.h>
#include <linux/sched.h>
#include <linux/file.h>
#include <linux/utsname.h>
#include <net/sock.h>
#include <linux/proc_fs.h>
#include <linux/poll.h>
#include <linux/module.h>
#include <linux/list.h>
#include <cluster/cnxman.h>
#include <cluster/service.h>

#include "cnxman-private.h"

#include "cluster.h"

/* Variables in main part of cnxman */
extern int we_are_a_cluster_member;
extern int cluster_is_quorate;
extern struct cluster_node *us;
extern int cluster_generation;
extern struct list_head cluster_members_list;
extern struct semaphore cluster_members_lock;
extern struct cluster_node *quorum_device;
extern int address_length;

/* How many times we have been attached, max 1 :-) */
static long refcnt;

/* callback parameters. Currently only one callback allowed */
static membership_callback_t notify_callback;
static char *notify_buffer;
static int  notify_max_items;
static void *notify_context;


/* copy internal node structure to external one */
static void copy_node(struct membership_node *m, struct cluster_node *c)
{
    struct cluster_node_addr *first_addr;

    m->mn_nodeid = c->node_id;
    strcpy(m->mn_name, c->name); /* TODO CHECK SIZE */
    if (c->state == NODESTATE_MEMBER)
	m->mn_member = 1;
    else
	m->mn_member = 0;
    m->mn_boottime = c->join_time;

    /* TODO: Can only supply the first node address, and I'm not sure
       how much use this is anyway as it doesn't say whether it's
       IP4 or IP6 or what.... */
    first_addr = list_entry(c->addr_list.next, struct cluster_node_addr, list);
    memcpy(m->mn_address.mna_address, first_addr, address_length);
    m->mn_address.mna_len = address_length;
}

static void call_notification(void)
{
    struct list_head *nodelist;
    struct cluster_node *cnode;
    struct membership_node *mnode;
    struct membership_notify_info info;
    int nodecount = 0;
    int members = 0;

    mnode = (struct membership_node *)notify_buffer;
    down(&cluster_members_lock);
    list_for_each(nodelist, &cluster_members_list) {
	cnode = list_entry(nodelist, struct cluster_node, list);
	if (nodecount < notify_max_items) {
	    copy_node(mnode, cnode);
	    mnode++;
	}

	nodecount++;
	if (cnode->state == NODESTATE_MEMBER)
	    members++;
    }

    up(&cluster_members_lock);

    info.mni_context = notify_context;
    info.mni_viewnumber = cluster_generation;
    info.mni_numitems = nodecount;
    info.mni_nummembers = members;
    info.mni_buffer = notify_buffer;

    notify_callback(&info);
}


/* Called by cman when the nodelist changes */
static void event_callback(kcl_callback_reason reason, long arg)
{
    if (reason == DIED || reason == NEWNODE) {
	call_notification();
    }
}



/* Enforce single instance */
static int clcman_attach(struct cluster_instance *ci)
{
    if (test_and_set_bit(1, &refcnt))
	return -EBUSY;

    return 0;
}

static int clcman_detach(struct cluster_instance *ci)
{
    if (we_are_a_cluster_member)
	return -EBUSY;

    if (test_and_clear_bit(1, &refcnt))
	return 0;

    return -EINVAL;
}

/* QUORUM function */
static int clcman_get_quorate(struct cluster_instance *ci)
{
    return cluster_is_quorate;
}

static int clcman_get_votes(struct cluster_instance *ci, int32_t nodeid)
{
    struct cluster_node *node = find_node_by_nodeid(nodeid);

    if (node)
	return node->votes;
    else
	return -EINVAL;
}

static int clcman_get_info(struct cluster_instance *ci, struct quorum_info *info)
{
    struct cluster_node *node;
    unsigned int total_votes = 0;
    unsigned int max_expected = 0;
    struct list_head *nodelist;

    if (!info)
	return -EINVAL;

    if (!we_are_a_cluster_member)
	return -ENOTCONN;

    /* Total the votes */
    down(&cluster_members_lock);
    list_for_each(nodelist, &cluster_members_list) {
	node = list_entry(nodelist, struct cluster_node, list);
	if (node->state == NODESTATE_MEMBER) {
	    total_votes += node->votes;
	    max_expected =
		max(max_expected, node->expected_votes);
	}
    }
    up(&cluster_members_lock);

    if (quorum_device && quorum_device->state == NODESTATE_MEMBER)
	total_votes += quorum_device->votes;

    info->qi_total_votes = total_votes;
    info->qi_expected_votes = max_expected;
    info->qi_quorum = get_quorum();

    return 0;
}

/* MEMBERSHIP functions */

static int clcman_start_notify(struct cluster_instance *ci,
			       void *context, uint32_t flags,
			       membership_callback_t callback,
			       char *buffer, int max_items)
{
    if (!callback)
	return -EINVAL;

    if (!buffer)
	return -EINVAL;

    if (notify_callback)
	return -EBUSY;

    notify_callback  = callback;
    notify_buffer    = buffer;
    notify_max_items = max_items;

    /* Call now if node list wanted */
    if (flags & MEMBERSHIP_FLAGS_NOTIFY_NODES)
	call_notification();

    /* Enable cman notifications */
    if (flags & MEMBERSHIP_FLAGS_NOTIFY_CHANGES)
	kcl_add_callback(event_callback);

    return 0;
}

static int clcman_stop_notify(struct cluster_instance *ci)
{
    notify_callback = NULL;
    return kcl_remove_callback(event_callback);
}


static int clcman_get_name(struct cluster_instance *ci, char *name, int maxlen)
{
    if (!name)
	return -EINVAL;

    if (strlen(ci->ci_proto->co_proto_name) > maxlen)
	return -ENOSPC;

    strcpy(name, ci->ci_proto->co_proto_name);
    return 0;
}

static int clcman_get_node(struct cluster_instance *ci, int32_t nodeid,
			   struct membership_node *node)
{
    struct cluster_node *cman_node;

    if (!node)
	return -EINVAL;

    if (nodeid == MEMBERSHIP_NODE_THISNODE) {
	cman_node = us;
    }
    else {
	cman_node = find_node_by_nodeid(nodeid);
    }
    if (!cman_node)
	return -EINVAL;


    copy_node(node, cman_node);
    return 0;
}


/* OPS structures for registration */
static struct membership_ops m_ops =
{
    .start_notify = clcman_start_notify,
    .stop_notify  = clcman_stop_notify,
    .get_name     = clcman_get_name,
    .get_node     = clcman_get_node,
};


static struct quorum_ops q_ops =
{
    .get_quorate = clcman_get_quorate,
    .get_votes   = clcman_get_votes,
    .get_info    = clcman_get_info,
};

static struct cm_proto cman_proto =
{
    .co_proto_name = "cman",
    .co_attach = clcman_attach,
    .co_detach = clcman_detach,

    .co_mops  = &m_ops,
    .co_qops  = &q_ops,
    .co_owner = THIS_MODULE,
};

int clcman_init(void)
{
    int ret;

    ret = cm_register(&cman_proto);
    if (!ret)
	printk("register cluster manager %s failed (ret=%d)\n", cman_proto.co_proto_name, ret);

    return ret;
}

void clcman_exit(void)
{
    cm_unregister(&cman_proto);
}
/* Test the cman interface....a bit */

#include <linux/init.h>
#include <linux/socket.h>
#include <linux/kernel.h>
#include <linux/sched.h>
#include <linux/file.h>
#include <linux/utsname.h>
#include <net/sock.h>
#include <linux/proc_fs.h>
#include <linux/poll.h>
#include <linux/module.h>
#include <linux/list.h>

#include "cluster.h"
struct cluster_instance *ci;
char members_buffer[4096];

static int member_callback(struct membership_notify_info *info)
{
    int i;
    struct membership_node *node = (struct membership_node *)info->mni_buffer;

    printk("Member callback:\n");
    printk("  view number    = %d\n", info->mni_viewnumber);
    printk("  num of items   = %d\n", info->mni_numitems);
    printk("  num of members = %d\n", info->mni_nummembers);

    for (i=0; i<info->mni_numitems; i++) {
	printk("     node id   = %d\n", node->mn_nodeid);
	printk("     node name = %s\n", node->mn_name);
	printk("     member    = %d\n", node->mn_member);
	node++;
    }

    return 0;
}



static int __init test_init(void)
{
    int ret;
    struct quorum_info qinfo;

    ret = cm_attach("cman", &ci);
    if (ret) {
	printk("Unable to attach to cluster manager: %d\n", ret);
    }
    else {
	if ( !(ret = ci->ci_proto->co_qops->get_info(ci, &qinfo)) ) {
	    printk("Quorum info: \n");
	    printk("   total votes    = %d\n", qinfo.qi_total_votes);
	    printk("   expected votes = %d\n", qinfo.qi_expected_votes);
	    printk("   quorum         = %d\n", qinfo.qi_quorum);

	    ret = ci->ci_proto->co_mops->start_notify(ci, NULL,
						      MEMBERSHIP_FLAGS_NOTIFY_CHANGES | MEMBERSHIP_FLAGS_NOTIFY_NODES,
						      member_callback, members_buffer,
						      sizeof(members_buffer)/sizeof(struct membership_node));
	    if (ret)
		printk("start notify failed: %d\n", ret);
	}
	else {
	    printk("Get info failed: %d\n", ret);
	}
    }
    return ret;
}

static void __exit test_exit(void)
{
    if (ci) {
	ci->ci_proto->co_mops->stop_notify(ci);
	cm_detach(ci);
    }
}


module_init(test_init);
module_exit(test_exit);

[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]