ref count group descriptors

src/ssg-internal.h

@@ -47,21 +47,19 @@ typedef struct ssg_member_state
     uint8_t is_member;
 } ssg_member_state_t;
 
-/* TODO: these really need to be ref-counted, else I don't think
- * duplicated references can be kept in sync...
- */
 /* TODO: associate a version number with a descriptor */
 typedef struct ssg_group_descriptor
 {
     uint64_t magic_nr;
     uint64_t name_hash;
     char *addr_str;
-    uint8_t owner_status;
+    int owner_status;
+    int ref_count;
 } ssg_group_descriptor_t;
 
 typedef struct ssg_group_view
 {
-    uint32_t size;
+    unsigned int size;
     ssg_member_state_t *member_states;
 } ssg_group_view_t;
 

src/ssg-rpc.c

@@ -42,7 +42,7 @@ DECLARE_MARGO_RPC_HANDLER(ssg_group_attach_recv_ult)
 static int ssg_group_view_serialize(
     ssg_group_view_t *view, void **buf, hg_size_t *buf_size);
 
-/* SSG RPC ids */
+/* SSG RPC IDs */
 static hg_id_t ssg_group_attach_rpc_id;
 
 /* ssg_register_rpcs
@@ -323,15 +323,18 @@ hg_return_t hg_proc_ssg_group_id_t(
                 hret = HG_PROTOCOL_ERROR;
                 return hret;
             }
+            (*group_descriptor)->ref_count = 1;
             break;
         case HG_FREE:
-            hret = hg_proc_ssg_group_descriptor_t(proc, *group_descriptor);
-            if (hret != HG_SUCCESS)
+            if((*group_descriptor)->ref_count == 1)
             {
-                hret = HG_PROTOCOL_ERROR;
-                return hret;
+                free((*group_descriptor)->addr_str);
+                free(*group_descriptor);
+            }
+            else
+            {
+                (*group_descriptor)->ref_count--;
             }
-            free(*group_descriptor);
             hret = HG_SUCCESS;
             break;
         default:

src/ssg.c

@@ -479,9 +479,6 @@ int ssg_group_attach(
 
     sret = SSG_SUCCESS;
 
-    /* XXX if group descriptors are ref counted, we don't need this awkwardness */
-    group_descriptor->owner_status = SSG_OWNER_IS_ATTACHER;
-
     /* don't free on success */
     ag = NULL;
 fini:
@@ -656,6 +653,7 @@ void ssg_group_id_free(
     ssg_group_descriptor_t *descriptor = (ssg_group_descriptor_t *)group_id;
 
     ssg_group_descriptor_free(descriptor);
+    descriptor = SSG_GROUP_ID_NULL;
     return;
 }
 
@@ -710,7 +708,7 @@ void ssg_group_dump(
         printf("\trole: '%s'\n", group_role);
         if (strcmp(group_role, "member") == 0)
             printf("\tself_id: %s\n", group_self_id);
-        printf("\tsize: %"PRIu32"\n", group_view->size);
+        printf("\tsize: %d\n", group_view->size);
         printf("\tview:\n");
         for (i = 0; i < group_view->size; i++)
             printf("\t\tid: %d\taddr: %s\n", i, group_view->member_states[i].addr_str);
@@ -746,21 +744,15 @@ static ssg_group_descriptor_t * ssg_group_descriptor_create(
         return NULL;
     }
     descriptor->owner_status = SSG_OWNER_IS_MEMBER;
+    descriptor->ref_count = 0;
     return descriptor;
 }
 
 static ssg_group_descriptor_t * ssg_group_descriptor_dup(
     ssg_group_descriptor_t * descriptor)
 {
-    ssg_group_descriptor_t *dup;
-
-    dup = malloc(sizeof(*dup));
-    if(!dup) return NULL;
-    dup->magic_nr = SSG_MAGIC_NR;
-    dup->name_hash = descriptor->name_hash;
-    dup->addr_str = strdup(descriptor->addr_str);
-    dup->owner_status = descriptor->owner_status;
-    return dup;
+    descriptor->ref_count++;
+    return descriptor;
 }
 
 static void ssg_group_descriptor_free(
@@ -768,8 +760,15 @@ static void ssg_group_descriptor_free(
 {
     if (descriptor)
     {
-        free(descriptor->addr_str);
-        free(descriptor);
+        if(descriptor->ref_count == 1)
+        {
+            free(descriptor->addr_str);
+            free(descriptor);
+        }
+        else
+        {
+            descriptor->ref_count--;
+        }
     }
     return;
 }
@@ -807,6 +806,7 @@ static int ssg_group_view_create(
         free(lookup_ult_args);
         return SSG_FAILURE;
     }
+
     for (i = 0; i < group_size; i++)
     {
         view->member_states[i].addr_str = NULL;
@@ -955,6 +955,7 @@ static void ssg_group_destroy_internal(
 
     /* destroy group state */
     ssg_group_view_destroy(&g->view);
+    g->descriptor->owner_status = SSG_OWNER_IS_UNASSOCIATED;
     ssg_group_descriptor_free(g->descriptor);
     free(g->name);
     free(g);
@@ -965,7 +966,8 @@ static void ssg_group_destroy_internal(
 static void ssg_attached_group_destroy(
     ssg_attached_group_t * ag)
 {
-    ssg_group_view_destroy(&ag->view);   
+    ssg_group_view_destroy(&ag->view);
+    ag->descriptor->owner_status = SSG_OWNER_IS_UNASSOCIATED;
     ssg_group_descriptor_free(ag->descriptor);
     free(ag->name);
     free(ag);