more attach code changes

include/ssg-mpi.h

@@ -24,13 +24,11 @@ extern "C" {
  *
  * @param[in]  group_name   Name of the SSG group
  * @param[in]  comm         MPI communicator containing group members
- * @param[out] group_id     Pointer to output SSG group ID
- * @returns SSG_SUCCESS on success, SSG error code otherwise
+ * @returns SSG group identifier on success, SSG_GROUP_ID_NULL otherwise
  */
-int ssg_group_create_mpi(
+ssg_group_id_t ssg_group_create_mpi(
     const char * group_name,
-    MPI_Comm comm,
-    ssg_group_id_t * group_id);
+    MPI_Comm comm);
 
 #ifdef __cplusplus
 }

include/ssg.h

@@ -24,21 +24,13 @@ extern "C" {
 #define SSG_SUCCESS 0
 #define SSG_FAILURE (-1)
 
-#define SSG_MEMBER_ID_INVALID UINT64_MAX
 typedef uint64_t ssg_member_id_t;
+#define SSG_MEMBER_ID_NULL UINT64_MAX
 
-/* SSG group identifier datatype */
-/* TODO: this shouldn't be visible ... we can't use a typical
- * opaque pointer since we want to be able to xmit these to
- * other processes.
- */
-#define SSG_GROUP_ID_MAX_ADDR_LEN 64
-typedef struct ssg_group_id
-{
-    uint64_t magic_nr;
-    uint64_t name_hash;
-    char addr_str[SSG_GROUP_ID_MAX_ADDR_LEN];
-} ssg_group_id_t;
+/* opaque SSG group identifier type */
+struct ssg_group_descriptor;
+typedef struct ssg_group_descriptor *ssg_group_id_t;
+#define SSG_GROUP_ID_NULL ((ssg_group_id_t)NULL)
 
 /***************************************************
  *** SSG runtime intialization/shutdown routines ***
@@ -71,18 +63,16 @@ int ssg_finalize(
  * @param[in]  group_name       Name of the SSG group
  * @param[in]  group_addr_strs  Array of HG address strings for each group member
  * @param[in]  group_size       Number of group members
- * @param[out] group_id         Pointer to output SSG group ID
- * @returns SSG_SUCCESS on success, SSG error code otherwise
+ * @returns SSG group identifier on success, SSG_GROUP_ID_NULL otherwise
  *
  * NOTE: The HG address string of the caller of this function must be present in
  * the list of address strings given in 'group_addr_strs'. That is, the caller
  * of this function is required to be a member of the SSG group that is created.
  */
-int ssg_group_create(
+ssg_group_id_t ssg_group_create(
     const char * group_name,
     const char * const group_addr_strs[],
-    int group_size,
-    ssg_group_id_t * group_id);
+    int group_size);
 
 /**
  * Creates an SSG group from a given config file containing the HG address strings
@@ -92,16 +82,15 @@ int ssg_group_create(
  * @param[in]  file_name    Name of the config file containing the corresponding
  *                          HG address strings for this group
  * @param[out] group_id     Pointer to output SSG group ID
- * @returns SSG_SUCCESS on success, SSG error code otherwise
+ * @returns SSG group identifier on success, SSG_GROUP_ID_NULL otherwise
  * 
  * NOTE: The HG address string of the caller of this function must be present in
  * the list of address strings given in the config file. That is, the caller of
  * this function is required to be a member of the SSG group that is created.
  */
-int ssg_group_create_config(
+ssg_group_id_t ssg_group_create_config(
     const char * group_name,
-    const char * file_name,
-    ssg_group_id_t * group_id);
+    const char * file_name);
 
 /**
  * Destroys data structures associated with a given SSG group ID.
@@ -142,7 +131,7 @@ int ssg_group_detach(
  * Obtains the caller's member ID in the given SSG group.
  *
  * @param[in] group_id SSG group ID
- * @returns caller's group ID on success, SSG_MEMBER_ID_INVALID otherwise
+ * @returns caller's group ID on success, SSG_MEMBER_ID_NULL otherwise
  */
 ssg_member_id_t ssg_get_group_self_id(
     ssg_group_id_t group_id);

src/ssg-internal.h

@@ -10,6 +10,7 @@
 #include <inttypes.h>
 
 #include <mercury.h>
+#include <mercury_proc_string.h>
 #include <abt.h>
 #include <margo.h>
 
@@ -39,9 +40,9 @@ extern "C" {
 
 /* SSG internal dataypes */
 
-typedef struct ssg_group ssg_group_t;
-typedef struct ssg_view ssg_view_t;
 typedef struct ssg_member_state ssg_member_state_t;
+typedef struct ssg_view ssg_view_t;
+typedef struct ssg_group ssg_group_t;
 typedef struct ssg_instance ssg_instance_t;
 
 struct ssg_member_state
@@ -52,16 +53,28 @@ struct ssg_member_state
 
 struct ssg_view
 {
-    int size;
+    uint32_t size;
     ssg_member_state_t *member_states;
 };
 
+MERCURY_GEN_PROC(ssg_group_descriptor_t, \
+    ((uint64_t) (magic_nr)) \
+    ((uint64_t) (name_hash)) \
+    ((hg_string_t) (addr_str)));
+
+struct ssg_group_descriptor
+{
+    uint64_t magic_nr;
+    uint64_t name_hash;
+    char *addr_str;
+};
+
 struct ssg_group
 {
     char *group_name;
-    ssg_group_id_t group_id;
-    ssg_member_id_t self_id;
+    ssg_group_descriptor_t *group_descriptor;
     ssg_view_t group_view;
+    ssg_member_id_t self_id;
     void *fd_ctx; /* failure detector context (currently just SWIM) */
     UT_hash_handle hh;
 };
@@ -83,7 +96,7 @@ hg_return_t ssg_group_lookup(
     ssg_group_t * g,
     const char * const addr_strs[]);
 hg_return_t ssg_group_attach_send(
-    const char *member_addr_str);
+    ssg_group_descriptor_t *group_descriptor);
 
 /* XXX: is this right? can this be a global? */
 extern ssg_instance_t *ssg_inst; 

src/ssg-rpc.c

@@ -15,10 +15,13 @@
 #include "ssg.h"
 #include "ssg-internal.h"
 
-/* SSG RPCS handlers */
+/* SSG RPCS handler prototypes */
 static void ssg_lookup_ult(void * arg);
 DECLARE_MARGO_RPC_HANDLER(ssg_group_attach_recv_ult)
 
+/* SSG RPC (de)serialization routine prototypes */
+static hg_return_t hg_proc_ssg_group_id_t(hg_proc_t proc, void *data);
+
 /* SSG RPC ids */
 static hg_id_t ssg_group_attach_rpc_id;
 
@@ -34,8 +37,8 @@ void ssg_register_rpcs()
     if (!hgcl) return;
 
     /* register HG RPCs for SSG */
-    ssg_group_attach_rpc_id = MERCURY_REGISTER(hgcl, "ssg_attach", void, void,
-        ssg_group_attach_recv_ult_handler);
+    ssg_group_attach_rpc_id = MERCURY_REGISTER(hgcl, "ssg_group_attach",
+        ssg_group_descriptor_t, void, ssg_group_attach_recv_ult_handler);
 
     return;
 }
@@ -58,7 +61,7 @@ hg_return_t ssg_group_lookup(
 {
     ABT_thread *ults;
     struct lookup_ult_args *args;
-    int i, r;
+    unsigned int i, r;
     int aret;
     hg_return_t hret = HG_SUCCESS;
 
@@ -142,7 +145,7 @@ static void ssg_lookup_ult(
  *
  *
  */
-hg_return_t ssg_group_attach_send(const char *member_addr_str)
+hg_return_t ssg_group_attach_send(ssg_group_descriptor_t * group_descriptor)
 {
     hg_class_t *hgcl = NULL;
     hg_addr_t member_addr = HG_ADDR_NULL;
@@ -153,7 +156,8 @@ hg_return_t ssg_group_attach_send(const char *member_addr_str)
     if (!hgcl) goto fini;
 
     /* lookup the address of the given group member */
-    hret = margo_addr_lookup(ssg_inst->mid, member_addr_str, &member_addr);
+    hret = margo_addr_lookup(ssg_inst->mid, group_descriptor->addr_str,
+        &member_addr);
     if (hret != HG_SUCCESS) goto fini;
 
     hret = HG_Create(margo_get_context(ssg_inst->mid), member_addr,
@@ -161,11 +165,9 @@ hg_return_t ssg_group_attach_send(const char *member_addr_str)
     if (hret != HG_SUCCESS) goto fini;
 
     /* send an attach request to the given group member address */
-    hret = margo_forward(ssg_inst->mid, handle, NULL);
+    hret = margo_forward(ssg_inst->mid, handle, group_descriptor);
     if (hret != HG_SUCCESS) goto fini;
 
-    /* XXX: store the obtained view locally to refer to */
-
     /* TODO: hold on to leader addr so we don't have to look it up again? */
 fini:
     if (hgcl && member_addr != HG_ADDR_NULL) HG_Addr_free(hgcl, member_addr);
@@ -176,7 +178,55 @@ fini:
 
 static void ssg_group_attach_recv_ult(hg_handle_t handle)
 {
+    ssg_group_t *g = NULL;
+    ssg_group_descriptor_t group_descriptor;
+    hg_return_t hret;
+
+    /* TODO: how to handle errors */
+    if (!ssg_inst) goto fini;
+
+    hret = HG_Get_input(handle, &group_descriptor);
+    if (hret != HG_SUCCESS) goto fini;
+
+    /* look for the given group in my local table of groups */
+    HASH_FIND(hh, ssg_inst->group_table, &group_descriptor.name_hash,
+        sizeof(uint64_t), g);
+    if (!g)
+    {
+        HG_Free_input(handle, &group_descriptor);
+        goto fini;
+    }
+
+    margo_respond(ssg_inst->mid, handle, NULL);
+
+    HG_Free_input(handle, &group_descriptor);
+
+fini:
     HG_Destroy(handle);
     return;
 }
 DEFINE_MARGO_RPC_HANDLER(ssg_group_attach_recv_ult)
+
+/* SSG RPC (de)serialization routines */
+
+#if 0
+static hg_return_t hg_proc_ssg_group_id_t(hg_proc_t proc, void *data)
+{
+    ssg_group_descriptor_t *group_descriptor = (ssg_group_descriptor_t *)data;
+    hg_return_t hret = HG_PROTOCOL_ERROR;
+
+    switch(hg_proc_get_op(proc))
+    {
+        case HG_ENCODE:
+            break;
+        case HG_DECODE:
+            break;
+        case HG_FREE:
+            break;
+        default:
+            break;
+    }
+
+    return hret;
+}
+#endif

src/ssg.c

@@ -33,9 +33,8 @@
 #include "uthash.h"
 
 /* SSG helper routine prototypes */
-static void ssg_generate_group_id(
-    const char * name, const char * leader_addr_str,
-    ssg_group_id_t *group_id);
+static ssg_group_descriptor_t * ssg_generate_group_descriptor(
+    const char * name, const char * leader_addr_str);
 static const char ** ssg_setup_addr_str_list(
     char * buf, int num_addrs);
 static int ssg_group_destroy_internal(
@@ -90,11 +89,10 @@ int ssg_finalize()
  *** SSG group management routines ***
  *************************************/
 
-int ssg_group_create(
+ssg_group_id_t ssg_group_create(
     const char * group_name,
     const char * const group_addr_strs[],
-    int group_size,
-    ssg_group_id_t * group_id)
+    int group_size)
 {
     hg_class_t *hgcl = NULL;
     hg_addr_t self_addr = HG_ADDR_NULL;
@@ -103,10 +101,10 @@ int ssg_group_create(
     const char *self_addr_substr = NULL;
     const char *addr_substr = NULL;
     int i;
+    ssg_group_descriptor_t *group_descriptor = NULL;
     ssg_group_t *g = NULL;
-    ssg_group_id_t new_gid;
     hg_return_t hret;
-    int sret = SSG_FAILURE;
+    ssg_group_id_t group_id = SSG_GROUP_ID_NULL;
 
     if (!ssg_inst) goto fini;
 
@@ -114,10 +112,12 @@ int ssg_group_create(
     if (!hgcl) goto fini;
 
     /* generate a unique ID for this group  */
-    ssg_generate_group_id(group_name, group_addr_strs[0], &new_gid);
+    group_descriptor = ssg_generate_group_descriptor(group_name, group_addr_strs[0]);
+    if (group_descriptor == NULL) goto fini;
 
-    /* make sure we aren't re-adding an existing group */
-    HASH_FIND(hh, ssg_inst->group_table, &new_gid.name_hash, sizeof(uint64_t), g);
+    /* make sure we aren't re-creating an existing group */
+    HASH_FIND(hh, ssg_inst->group_table, &group_descriptor->name_hash,
+        sizeof(uint64_t), g);
     if (g) goto fini;
 
     /* allocate an SSG group data structure and initialize some of it */
@@ -126,7 +126,7 @@ int ssg_group_create(
     memset(g, 0, sizeof(*g));
     g->group_name = strdup(group_name);
     if (!g->group_name) goto fini;
-    memcpy(&g->group_id, &new_gid, sizeof(new_gid));
+    g->group_descriptor = group_descriptor;
     // TODO? g->self_id = -1;
     g->group_view.size = group_size;
     g->group_view.member_states = malloc(
@@ -194,7 +194,6 @@ int ssg_group_create(
             group_name);
         goto fini;
     }
-    SSG_DEBUG(g, "group lookup successful (size=%d)\n", group_size);
 
 #ifdef SSG_USE_SWIM_FD
     int swim_active = 1;
@@ -211,18 +210,24 @@ int ssg_group_create(
 #endif
 
     /* add this group reference to our group table */
-    HASH_ADD(hh, ssg_inst->group_table, group_id.name_hash, sizeof(uint64_t), g);
+    HASH_ADD(hh, ssg_inst->group_table, group_descriptor->name_hash,
+        sizeof(uint64_t), g);
 
-    /* everything successful -- set the output for this call */
-    memcpy(group_id, &new_gid, sizeof(new_gid));
-    sret = SSG_SUCCESS;
+    /* everything successful -- set the output group identifier, which is just
+     * an opaque pointer to the group descriptor structure
+     */
+    group_id = (ssg_group_id_t)group_descriptor;
+
+    SSG_DEBUG(g, "group create successful (size=%d)\n", group_size);
 
     /* don't free these pointers on success */
+    group_descriptor = NULL;
     self_addr = HG_ADDR_NULL;
     g = NULL;
 fini:
     if (hgcl && self_addr != HG_ADDR_NULL) HG_Addr_free(hgcl, self_addr);
     free(self_addr_str);
+    free(group_descriptor);
     if (g)
     {
         free(g->group_name);
@@ -230,13 +235,12 @@ fini:
         free(g);
     }
 
-    return sret;
+    return group_id;
 }
 
-int ssg_group_create_config(
+ssg_group_id_t ssg_group_create_config(
     const char * group_name,
-    const char * file_name,
-    ssg_group_id_t * group_id)
+    const char * file_name)
 {
     int fd;
     struct stat st;
@@ -247,7 +251,7 @@ int ssg_group_create_config(
     int addr_buf_len = 0, num_addrs = 0;
     int ret;
     const char **addr_strs = NULL;
-    int sret = SSG_FAILURE;
+    ssg_group_id_t group_id = SSG_GROUP_ID_NULL;
 
     /* open config file for reading */
     fd = open(file_name, O_RDONLY);
@@ -309,7 +313,7 @@ int ssg_group_create_config(
     if (!addr_strs) goto fini;
 
     /* invoke the generic group create routine using our list of addrs */
-    sret = ssg_group_create(group_name, addr_strs, num_addrs, group_id);
+    group_id = ssg_group_create(group_name, addr_strs, num_addrs);
 
 fini:
     /* cleanup before returning */
@@ -318,14 +322,13 @@ fini:
     free(addr_buf);
     free(addr_strs);
 
-    return sret;
+    return group_id;
 }
 
 #ifdef SSG_HAVE_MPI
-int ssg_group_create_mpi(
+ssg_group_id_t ssg_group_create_mpi(
     const char * group_name,
-    MPI_Comm comm,
-    ssg_group_id_t * group_id)
+    MPI_Comm comm)
 {
     hg_class_t *hgcl = NULL;
     hg_addr_t self_addr = HG_ADDR_NULL;
@@ -338,7 +341,7 @@ int ssg_group_create_mpi(
     int comm_size = 0, comm_rank = 0;
     const char **addr_strs = NULL;
     hg_return_t hret;
-    int sret = SSG_FAILURE;
+    ssg_group_id_t group_id = SSG_GROUP_ID_NULL;
 
     if (!ssg_inst) goto fini;
 
@@ -384,7 +387,7 @@ int ssg_group_create_mpi(
     if (!addr_strs) goto fini;
 
     /* invoke the generic group create routine using our list of addrs */
-    sret = ssg_group_create(group_name, addr_strs, comm_size, group_id);
+    group_id = ssg_group_create(group_name, addr_strs, comm_size);
 
 fini:
     /* cleanup before returning */
@@ -395,21 +398,23 @@ fini:
     free(self_addr_str);
     free(addr_strs);
 
-    return sret;
+    return group_id;
 }
 #endif
 
 int ssg_group_destroy(
     ssg_group_id_t group_id)
 {
+    ssg_group_descriptor_t *group_descriptor = (ssg_group_descriptor_t *)group_id;
     ssg_group_t *g;
     int sret;
 
-    if (!ssg_inst)
+    if (!ssg_inst || group_id == SSG_GROUP_ID_NULL)
         return SSG_FAILURE;
 
     /* find the group structure and destroy it */
-    HASH_FIND(hh, ssg_inst->group_table, &group_id.name_hash, sizeof(uint64_t), g);
+    HASH_FIND(hh, ssg_inst->group_table, &group_descriptor->name_hash,
+        sizeof(uint64_t), g);
     HASH_DELETE(hh, ssg_inst->group_table, g);
     sret = ssg_group_destroy_internal(g);
 
@@ -419,13 +424,13 @@ int ssg_group_destroy(
 int ssg_group_attach(
     ssg_group_id_t group_id)
 {
+    ssg_group_descriptor_t *group_descriptor = (ssg_group_descriptor_t *)group_id;
     hg_return_t hret;
 
-    if (!ssg_inst)
+    if (!ssg_inst || group_id == SSG_GROUP_ID_NULL)
         return SSG_FAILURE;
 
-    /* XXX: for now just send to the 1 member addr in the group id */
-    hret = ssg_group_attach_send(group_id.addr_str);
+    hret = ssg_group_attach_send(group_descriptor);
     if (hret != HG_SUCCESS)
         return SSG_FAILURE;
 
@@ -445,14 +450,16 @@ int ssg_group_detach(
 ssg_member_id_t ssg_get_group_self_id(
     ssg_group_id_t group_id)
 {
+    ssg_group_descriptor_t *group_descriptor = (ssg_group_descriptor_t *)group_id;
     ssg_group_t *g;
 
-    if (!ssg_inst)
-        return SSG_MEMBER_ID_INVALID;
+    if (!ssg_inst || group_id == SSG_GROUP_ID_NULL)
+        return SSG_MEMBER_ID_NULL;
 
-    HASH_FIND(hh, ssg_inst->group_table, &group_id.name_hash, sizeof(uint64_t), g);
+    HASH_FIND(hh, ssg_inst->group_table, &group_descriptor->name_hash,
+        sizeof(uint64_t), g);
     if (!g)
-        return SSG_MEMBER_ID_INVALID;
+        return SSG_MEMBER_ID_NULL;
 
     return g->self_id;
 }
@@ -460,12 +467,14 @@ ssg_member_id_t ssg_get_group_self_id(
 int ssg_get_group_size(
     ssg_group_id_t group_id)
 {
+    ssg_group_descriptor_t *group_descriptor = (ssg_group_descriptor_t *)group_id;
     ssg_group_t *g;
 
-    if (!ssg_inst)
+    if (!ssg_inst || group_id == SSG_GROUP_ID_NULL)
         return 0;
 
-    HASH_FIND(hh, ssg_inst->group_table, &group_id.name_hash, sizeof(uint64_t), g);
+    HASH_FIND(hh, ssg_inst->group_table, &group_descriptor->name_hash,
+        sizeof(uint64_t), g);
     if (!g)
         return 0;
 
@@ -476,12 +485,14 @@ hg_addr_t ssg_get_addr(
     ssg_group_id_t group_id,
     ssg_member_id_t member_id)
 {
+    ssg_group_descriptor_t *group_descriptor = (ssg_group_descriptor_t *)group_id;
     ssg_group_t *g;
 
-    if (!ssg_inst)
+    if (!ssg_inst || group_id == SSG_GROUP_ID_NULL)
         return HG_ADDR_NULL;
 
-    HASH_FIND(hh, ssg_inst->group_table, &group_id.name_hash, sizeof(uint64_t), g);
+    HASH_FIND(hh, ssg_inst->group_table, &group_descriptor->name_hash,
+        sizeof(uint64_t), g);
     if (!g)
         return HG_ADDR_NULL;
 
@@ -492,20 +503,28 @@ hg_addr_t ssg_get_addr(
  *** SSG internal helper routines ***
  ************************************/
 
-static void ssg_generate_group_id(
-    const char * name, const char * leader_addr_str,
-    ssg_group_id_t *group_id)
+static ssg_group_descriptor_t * ssg_generate_group_descriptor(
+    const char * name, const char * leader_addr_str)
 {
+    ssg_group_descriptor_t *group_descriptor;
     uint32_t upper, lower;
 
+    group_descriptor = malloc(sizeof(*group_descriptor));
+    if (!group_descriptor) return NULL;
+
     /* hash the group name to obtain an 64-bit unique ID */
     ssg_hashlittle2(name, strlen(name), &lower, &upper);
 
-    group_id->magic_nr = SSG_MAGIC_NR;
-    group_id->name_hash = lower + (((uint64_t)upper)<<32);
-    strcpy(group_id->addr_str, leader_addr_str);
+    group_descriptor->magic_nr = SSG_MAGIC_NR;
+    group_descriptor->name_hash = lower + (((uint64_t)upper)<<32);
+    group_descriptor->addr_str = strdup(leader_addr_str);
+    if (!group_descriptor->addr_str)
+    {
+        free(group_descriptor);
+        return NULL;
+    }
 
-    return;
+    return group_descriptor;
 }
 
 static const char ** ssg_setup_addr_str_list(
@@ -525,7 +544,7 @@ static const char ** ssg_setup_addr_str_list(
 
 static int ssg_group_destroy_internal(ssg_group_t *g)
 {
-    int i;
+    unsigned int i;
 
     /* TODO: send a leave message to the group ? */
 

tests/Makefile.subdir

 check_PROGRAMS += \
- tests/ssg-test-simple
+ tests/ssg-test-simple \
+ tests/ssg-test-attach
 
 tests_ssg_test_simple_SOURCES = tests/ssg-test-simple.c
 tests_ssg_test_simple_LDADD = src/libssg.la
+
+tests_ssg_test_attach_SOURCES = tests/ssg-test-attach.c
+tests_ssg_test_attach_LDADD = src/libssg.la

tests/ssg-test-attach.c

@@ -10,12 +10,17 @@
 #include <stdio.h>
 #include <string.h>
 #include <assert.h>
+#ifdef SSG_HAVE_MPI
 #include <mpi.h>
+#endif
 
 #include <margo.h>
 #include <mercury.h>
 #include <abt.h>
 #include <ssg.h>
+#ifdef SSG_HAVE_MPI
+#include <ssg-mpi.h>
+#endif
 
 #define DIE_IF(cond_expr, err_fmt, ...) \
     do { \
@@ -30,14 +35,19 @@ static void usage()
 {
     fprintf(stderr,
         "Usage: "
-        "ssg-test-simple [-s <time>] <addr> \n"
-        "\t-s <time> - time to sleep after SSG group creation\n");
+        "ssg-test-attach [-s <time>] <addr> \n"
+        "\t-s <time> - time to sleep between SSG group operations\n");
 }
 
 static void parse_args(int argc, char *argv[], int *sleep_time, const char **addr_str)
 {
     int ndx = 1;
 
+#ifndef SSG_HAVE_MPI
+    fprintf(stderr, "Error: ssg-test-attach currently requries MPI support\n");
+    exit(1);
+#endif
+
     if (argc < 2)
     {
         usage();
@@ -71,16 +81,15 @@ int main(int argc, char *argv[])
     const char *addr_str;
     const char *group_name = "simple_group";
     ssg_group_id_t g_id;
-    int my_world_rank;
-    int my_ssg_rank;
-    int color;
-    MPI_Comm ssg_comm;
+    int is_attacher = 0;
     int sret;
 
     parse_args(argc, argv, &sleep_time, &addr_str);