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);
