diff options
Diffstat (limited to 'lib/clplumbing/replytrack.c')
| -rw-r--r-- | lib/clplumbing/replytrack.c | 643 |
1 files changed, 643 insertions, 0 deletions
diff --git a/lib/clplumbing/replytrack.c b/lib/clplumbing/replytrack.c new file mode 100644 index 0000000..8c7c38e --- /dev/null +++ b/lib/clplumbing/replytrack.c @@ -0,0 +1,643 @@ + +/* + * Reply tracking library. + * + * Copyright (c) 2007 Alan Robertson + * Author: Alan Robertson <alanr@unix.sh> + * + ****************************************************************** + * This library is useful for tracking replies to multicast messages + * sent to cluster members. It tracks incremental membership changes + * according to any desired criteria, and then keeps track of when + * the last expected reply is received according to the dynamically + * updated membership as of when the message was sent out. + ****************************************************************** + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library 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 + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * + */ + +#include <lha_internal.h> +#include <stdlib.h> +#include <errno.h> +#include <sys/wait.h> +#include <sys/types.h> +#include <signal.h> +#include <memory.h> +#include <clplumbing/cl_log.h> +#include <clplumbing/replytrack.h> +#include <clplumbing/Gmain_timeout.h> + +/* + * These are the only data items that go in our GHashTables + */ +struct rt_node_info { + char * nodename; + cl_uuid_t nodeid; +}; + +struct node_tables { + + GHashTable* uuidmap; /* Keyed by uuid */ + int uuidcount; + GHashTable* namemap; /* Keyed by nodename*/ + int namecount; +}; +struct _nodetrack { + struct node_tables nt; + int refcount; + nodetrack_callback_t callback; + gpointer user_data; + nodetrack_callback_t extra_callback; + gpointer ext_data; +}; + +/* + * Things we use to track outstanding replies + * This is the structure used by the replytrack_t typedef + */ +struct _replytrack { + replytrack_callback_t callback; + gpointer user_data; + unsigned timerid; + struct node_tables tables; + gboolean expectingmore; + nodetrack_t* membership; +}; + +struct _nodetrack_intersection { + nodetrack_t** tables; + int ntables; + nodetrack_callback_t callback; + gpointer user_data; + nodetrack_t* intersection; +}; + +static cl_uuid_t nulluuid; +static int nodetrack_t_count = 0; +static int replytrack_t_count = 0; +static int replytrack_intersection_t_count = 0; + +static struct rt_node_info * +rt_node_info_new(const char * nodename, cl_uuid_t nodeid) +{ + struct rt_node_info* ret; + + if (!nodename) { + return NULL; + } + ret = MALLOCT(struct rt_node_info); + + if (!ret) { + return ret; + } + ret->nodename = strdup(nodename); + if (!ret->nodename) { + free(ret); + ret = NULL; + return ret; + } + ret->nodeid = nodeid; + return ret; +} + +static void +rt_node_info_del(struct rt_node_info * ni) +{ + if (ni != NULL) { + if (ni->nodename != NULL) { + free(ni->nodename); + } + memset(ni, 0, sizeof(*ni)); + free(ni); + } +} + +/* + * namehash cannot be NULL, idhash cannot be NULL, and nodename cannot be NULL + * + * 'id' can be a NULL uuid, in which case it goes into only the name table + * 'nodename' can change over time - in which case we update our tables. + * It is possible for one nodename to have more than one uuid. + * We allow for that. + * + * Changing the uuid but keeping the nodename the same is considered to be + * adding a new node with the same nodename. + * Exception: A node with a null uuid is presumed to have acquired a proper + * uuid if it is later seen with a non-null UUID + */ + +static gboolean +del_node_from_hashtables(struct node_tables *t +, const char * nodename, cl_uuid_t id) +{ + struct rt_node_info * entry; + if (cl_uuid_is_null(&id)) { + if ((entry = g_hash_table_lookup(t->namemap,nodename))!=NULL){ + g_hash_table_remove(t->namemap, nodename); + rt_node_info_del(entry); + t->namecount--; + } + return TRUE; + } + if ((entry=g_hash_table_lookup(t->uuidmap, &id)) != NULL) { + g_hash_table_remove(t->uuidmap, &id); + rt_node_info_del(entry); + t->uuidcount--; + } + return TRUE; +} + + +static gboolean +add_node_to_hashtables(struct node_tables * t +, const char * nodename, cl_uuid_t id) +{ + struct rt_node_info* idinfo = NULL; + + if (cl_uuid_is_null(&id)) { + /* Supplied uuid is the null UUID - insert in name table */ + struct rt_node_info* ninfo; + if (g_hash_table_lookup(t->namemap, nodename) == NULL) { + if (NULL == (ninfo = rt_node_info_new(nodename, id))){ + goto outofmem; + } + g_hash_table_insert(t->namemap,ninfo->nodename,ninfo); + t->namecount++; + } + return TRUE; + } + + /* Supplied uuid is not the null UUID */ + + if (g_hash_table_lookup(t->uuidmap,&id) == NULL) { + /* See if a corresponding name is in name map */ + /* If so, delete it - assume uuid was missing before */ + + if (g_hash_table_lookup(t->namemap, nodename) != NULL) { + del_node_from_hashtables(t, nodename, nulluuid); + } + /* Not yet in our uuid hash table */ + idinfo = rt_node_info_new(nodename, id); + if (idinfo == NULL) { + goto outofmem; + } + g_hash_table_insert(t->uuidmap, &idinfo->nodeid, idinfo); + t->uuidcount++; + } + return TRUE; +outofmem: + cl_log(LOG_ERR, "%s: out of memory", __FUNCTION__); + return FALSE; +} + +static gboolean +create_new_hashtables(struct node_tables*t) +{ + t->namemap = g_hash_table_new(g_str_hash, g_str_equal); + if (t->namemap == NULL) { + return FALSE; + } + t->uuidmap = g_hash_table_new(cl_uuid_g_hash, cl_uuid_g_equal); + if (t->uuidmap == NULL) { + g_hash_table_destroy(t->namemap); + t->namemap = NULL; + return FALSE; + } + return TRUE; +} + +static gboolean +hashtable_destroy_rt_node_info(gpointer key, gpointer rti, gpointer unused) +{ + rt_node_info_del(rti); + rti = key = NULL; + return TRUE; +} + +static void +destroy_map_hashtable(GHashTable*t) +{ + g_hash_table_foreach_remove(t, hashtable_destroy_rt_node_info,NULL); + g_hash_table_destroy(t); + t = NULL; +} + +struct tablehelp { + struct node_tables* t; + gboolean ret; +}; + +static void +copy_hashtables_helper(gpointer key_unused, gpointer value +, gpointer user_data) +{ + struct tablehelp * th = user_data; + struct rt_node_info* ni = value; + if (!add_node_to_hashtables(th->t, ni->nodename, ni->nodeid)) { + th->ret = FALSE; + } +} + +static gboolean +copy_hashtables(struct node_tables* tin, struct node_tables* tout) +{ + struct tablehelp newtables; + if (!create_new_hashtables(tout)){ + return FALSE; + } + newtables.t = tout; + newtables.ret = TRUE; + + g_hash_table_foreach(tout->namemap,copy_hashtables_helper,&newtables); + if (!newtables.ret) { + return FALSE; + } + g_hash_table_foreach(tout->uuidmap,copy_hashtables_helper,&newtables); + return newtables.ret; +} + +static gboolean mbr_inityet = FALSE; +static void +init_global_membership(void) +{ + if (mbr_inityet) { + return; + } + mbr_inityet = TRUE; + memset(&nulluuid, 0, sizeof(nulluuid)); +} + +gboolean /* Call us when an expected replier joins / comes up */ +nodetrack_nodeup(nodetrack_t * mbr, const char * node, cl_uuid_t uuid) +{ + gboolean ret; + ret = add_node_to_hashtables(&mbr->nt, node, uuid); + if (ret && mbr->callback) { + mbr->callback(mbr, node, uuid, NODET_UP, mbr->user_data); + } + if (mbr->extra_callback) { + mbr->extra_callback(mbr, node, uuid, NODET_UP,mbr->ext_data); + } + return ret; +} + +gboolean /* Call us when an expected replier goes down / away */ +nodetrack_nodedown(nodetrack_t* mbr, const char* node, cl_uuid_t uuid) +{ + if (mbr->callback) { + mbr->callback(mbr, node, uuid, NODET_DOWN, mbr->user_data); + } + if (mbr->extra_callback) { + mbr->extra_callback(mbr, node,uuid,NODET_DOWN,mbr->ext_data); + } + return del_node_from_hashtables(&mbr->nt, node, uuid); +} + +/* This function calls the user's timeout callback */ +static gboolean +replytrack_timeout_helper(gpointer rldata) +{ + replytrack_t* rl = rldata; + rl->expectingmore = FALSE; + rl->timerid = 0; + if (rl->callback) { + rl->callback(rl, rl->user_data, REPLYT_TIMEOUT); + } + return FALSE; +} + +replytrack_t* /* replytrack_t constructor */ +replytrack_new(nodetrack_t * membership +, replytrack_callback_t callback +, unsigned long timeout_ms +, gpointer user_data) +{ + replytrack_t* ret = MALLOCT(replytrack_t); + if (!ret) { + return ret; + } + if (!copy_hashtables(&membership->nt, &ret->tables)) { + free(ret); + ret = NULL; + return ret; + } + replytrack_t_count++; + ret->membership = membership; + ret->membership->refcount++; + ret->callback = callback; + ret->user_data = user_data; + ret->expectingmore = TRUE; + ret->timerid = 0; + if (timeout_ms != 0 && callback != NULL) { + ret->timerid = Gmain_timeout_add(timeout_ms + , replytrack_timeout_helper, ret); + } + return ret; +} + +void /* replytrack_t destructor */ +replytrack_del(replytrack_t * rl) +{ + rl->membership->refcount--; + replytrack_t_count++; + if (rl->expectingmore && rl->timerid > 0) { + cl_log(LOG_INFO + , "%s: destroying replytrack while still expecting" + " %d replies" + , __FUNCTION__ + , (rl->tables.namecount + rl->tables.uuidcount)); + } + if (rl->timerid > 0) { + g_source_remove(rl->timerid); + rl->timerid = 0; + } + destroy_map_hashtable(rl->tables.namemap); + rl->tables.namemap=NULL; + destroy_map_hashtable(rl->tables.uuidmap); + rl->tables.uuidmap=NULL; + memset(&rl, 0, sizeof(rl)); + free(rl); + rl=NULL; +} + +gboolean /* Call replytrack_gotreply when you receive an expected reply */ +replytrack_gotreply(replytrack_t*rl, const char * node, cl_uuid_t uuid) +{ + gboolean lastone; + del_node_from_hashtables(&rl->tables, node, uuid); + lastone = (rl->tables.namecount + rl->tables.uuidcount) == 0; + if (lastone) { + rl->expectingmore = FALSE; + if (rl->timerid > 0) { + g_source_remove(rl->timerid); + rl->timerid = 0; + } + if (rl->callback){ + rl->callback(rl, rl->user_data, REPLYT_ALLRCVD); + } + } + return lastone; +} + +struct replytrack_iterator_data { + replytrack_t* rlist; + replytrack_iterator_t f; + int count; + gpointer user_data; +}; + + +static void /* g_hash_table user-level iteration helper */ +replytrack_iterator_helper(gpointer key_unused, gpointer entry +, gpointer user_data) +{ + struct replytrack_iterator_data* ri = user_data; + struct rt_node_info* ni = entry; + if (ri && ri->rlist) { + ++ri->count; + if (ri->f) { + ri->f(ri->rlist, ri->user_data + , ni->nodename, ni->nodeid); + } + } +} + + + +int /* iterate through the outstanding expected replies */ +replytrack_outstanding_iterate(replytrack_t* rl +, replytrack_iterator_t i, gpointer user_data) +{ + struct replytrack_iterator_data id; + id.rlist = rl; + id.f = i; + id.count = 0; + id.user_data = user_data; + g_hash_table_foreach(rl->tables.namemap, replytrack_iterator_helper + , &id); + g_hash_table_foreach(rl->tables.uuidmap, replytrack_iterator_helper + , &id); + if (id.count != (rl->tables.namecount + rl->tables.uuidcount)) { + cl_log(LOG_ERR + , "%s: iteration count %d disagrees with" + " (namecount %d+uuidcount %d)" + , __FUNCTION__, id.count + , rl->tables.namecount,rl->tables.uuidcount); + } + return id.count; +} +int /* return count of outstanding expected replies */ +replytrack_outstanding_count(replytrack_t* rl) +{ + return (rl->tables.namecount + rl->tables.uuidcount); +} + +nodetrack_t* +nodetrack_new(nodetrack_callback_t callback, gpointer user_data) +{ + nodetrack_t* ret = MALLOCT(nodetrack_t); + if (!mbr_inityet) { + init_global_membership(); + } + if (!ret) { + return ret; + } + nodetrack_t_count++; + ret->refcount = 0; + if (!create_new_hashtables(&ret->nt)) { + free(ret); + ret = NULL; + } + ret->user_data = user_data; + ret->callback = callback; + ret->extra_callback = NULL; + ret->ext_data = NULL; + return ret; +} +void +nodetrack_del(nodetrack_t * np) +{ + if (np->refcount) { + cl_log(LOG_ERR + , "%s: reply tracking reference count is %d" + , __FUNCTION__, np->refcount); + } + nodetrack_t_count--; + destroy_map_hashtable(np->nt.namemap); + np->nt.namemap=NULL; + destroy_map_hashtable(np->nt.uuidmap); + np->nt.uuidmap=NULL; + memset(np, 0, sizeof(*np)); + free(np); +} + +gboolean +nodetrack_ismember(nodetrack_t* mbr, const char * name, cl_uuid_t u) +{ + if (cl_uuid_is_null(&u)) { + return(g_hash_table_lookup(mbr->nt.namemap, name) != NULL); + } + return (g_hash_table_lookup(mbr->nt.uuidmap, &u) != NULL); +} + +struct nodetrack_iterator_data { + nodetrack_t* rlist; + nodetrack_iterator_t f; + int count; + gpointer user_data; +}; +static void /* g_hash_table user-level iteration helper */ +nodetrack_iterator_helper(gpointer key_unused, gpointer entry +, gpointer user_data) +{ + struct nodetrack_iterator_data* ri = user_data; + struct rt_node_info* ni = entry; + if (ri && ri->rlist) { + ++ri->count; + if (ri->f) { + ri->f(ri->rlist, ri->user_data + , ni->nodename, ni->nodeid); + } + } +} + +int /* iterate through the outstanding expected replies */ +nodetrack_iterate(nodetrack_t* rl +, nodetrack_iterator_t i, gpointer user_data) +{ + struct nodetrack_iterator_data id; + id.rlist = rl; + id.f = i; + id.count = 0; + id.user_data = user_data; + g_hash_table_foreach(rl->nt.namemap, nodetrack_iterator_helper + , &id); + g_hash_table_foreach(rl->nt.uuidmap, nodetrack_iterator_helper + , &id); + if (id.count != (rl->nt.namecount + rl->nt.uuidcount)) { + cl_log(LOG_ERR + , "%s: iteration count %d disagrees with" + " (namecount %d+uuidcount %d)" + , __FUNCTION__, id.count + , rl->nt.namecount,rl->nt.uuidcount); + } + return id.count; +} +static void +intersection_callback +( nodetrack_t * mbr +, const char * node +, cl_uuid_t u +, nodetrack_change_t reason +, gpointer user_data) +{ + nodetrack_intersection_t* it = user_data; + int j; + gboolean allfound = TRUE; + + if (reason == NODET_DOWN) { + if (nodetrack_ismember(it->intersection, node, u)) { + nodetrack_nodedown(it->intersection,node,u); + } + return; + } + for (j=0; j < it->ntables && allfound; ++j) { + if (nodetrack_ismember(it->tables[j], node, u)) { + allfound = FALSE; + } + } + if (allfound) { + nodetrack_nodeup(it->intersection, node, u); + } +} + +struct li_helper { + nodetrack_intersection_t* i; + gboolean result; +}; + +static void +intersection_init_iterator(nodetrack_t* nt +, gpointer ghelp +, const char* node +, cl_uuid_t uuid) +{ + struct li_helper* help = ghelp; + gboolean allfound = TRUE; + int j; + + for (j=1; allfound && j < help->i->ntables; ++j) { + if (!nodetrack_ismember(help->i->tables[j] + , node, uuid)) { + allfound = FALSE; + } + } + if (allfound) { + nodetrack_nodeup(help->i->intersection, node, uuid); + } +} + +nodetrack_intersection_t* +nodetrack_intersection_new(nodetrack_t** tables, int ntables +, nodetrack_callback_t callback, gpointer user_data) +{ + nodetrack_intersection_t* ret; + int j; + ret = MALLOCT(nodetrack_intersection_t); + if (!ret) { + return ret; + } + ret->intersection = nodetrack_new(callback, user_data); + if (!ret->intersection) { + free(ret); + ret = NULL; + return ret; + } + ret->tables = tables; + ret->ntables = ntables; + ret->callback = callback; + ret->user_data = user_data; + for (j=0; j < ntables; ++j) { + tables[j]->refcount ++; + tables[j]->ext_data = ret; + tables[j]->extra_callback = intersection_callback; + } + /* Initialize the intersection membership list */ + nodetrack_iterate(tables[0], intersection_init_iterator, ret); + replytrack_intersection_t_count++; + return ret; +} +void +nodetrack_intersection_del(nodetrack_intersection_t* p) +{ + int j; + + for (j=0; j < p->ntables; ++j) { + p->tables[j]->refcount ++; + } + nodetrack_del(p->intersection); + p->intersection = NULL; + memset(p, 0, sizeof(*p)); + free(p); + p = NULL; + replytrack_intersection_t_count--; +} + +nodetrack_t* +nodetrack_intersection_table(nodetrack_intersection_t*p) +{ + return p->intersection; +} |