gpr proxy i/f (put) seems to be working w/ tcp oob!

This commit was SVN r2235.

configure.ac

@@ -946,6 +946,7 @@ AC_CONFIG_FILES([
     test/mca/gpr/Makefile
     test/mca/oob/Makefile
     test/mca/ns/Makefile
+    test/mca/gpr/Makefile
     test/mca/llm/Makefile
     test/mca/llm/base/Makefile
     test/mca/pcm/Makefile

src/mca/oob/base/oob_base_close.c

@@ -22,7 +22,7 @@ int mca_oob_base_close(void)
        item != NULL;
        item =  ompi_list_remove_first(&mca_oob_base_modules)) {
     mca_oob_base_info_t* base = (mca_oob_base_info_t *) item;
-    base->oob_module->oob_finalize(base->oob_module);
+    base->oob_module->oob_fini();
   }
 
   /* Close all remaining available modules (may be one if this is a

src/mca/oob/base/oob_base_init.c

@@ -215,3 +215,24 @@ char* mca_oob_get_contact_info()
     return contact_info;
 }
 
+/**
+* Called to request the selected oob components to
+* register their address with the seed deamon.
+*/
+
+int mca_oob_base_register(void)
+{
+  ompi_list_item_t* item;
+
+  /* Initialize all modules after oob/gpr/ns have initialized */
+  for (item =  ompi_list_get_first(&mca_oob_base_modules);
+       item != ompi_list_get_end(&mca_oob_base_modules);
+       item =  ompi_list_get_next(item)) {
+    mca_oob_base_info_t* base = (mca_oob_base_info_t *) item;
+    if (NULL != base->oob_module->oob_init)
+        base->oob_module->oob_init();
+  }
+  return OMPI_SUCCESS;
+}
+                                                                                                                                           
+

src/mca/oob/base/oob_base_open.c

@@ -50,7 +50,7 @@ int mca_oob_base_open(void)
   mca_base_param_lookup_string(
       mca_base_param_register_string("oob","base","include",NULL,NULL), &mca_oob_base_include);
   mca_base_param_lookup_string(
-      mca_base_param_register_string("oob","base","exclude",NULL,NULL), &mca_oob_base_exclude);
+      mca_base_param_register_string("oob","base","exclude",NULL,"tcp"), &mca_oob_base_exclude);
 
   /* All done */
   return OMPI_SUCCESS;

src/mca/oob/cofs/src/oob_cofs.h

@@ -22,7 +22,8 @@ int mca_oob_cofs_close(void);
  * Startup / Shutdown
  */
 mca_oob_t* mca_oob_cofs_init(bool *allow_multi_user_threads, bool *have_hidden_threads);
-int mca_oob_cofs_finalize(mca_oob_t*);
+int mca_oob_cofs_module_init(void);
+int mca_oob_cofs_module_fini(void);
 
 /* stubs */
 char* mca_oob_cofs_get_addr(void);

src/mca/oob/cofs/src/oob_cofs_component.c

@@ -47,7 +47,9 @@ mca_oob_t mca_oob_cofs = {
   mca_oob_cofs_recv,
   mca_oob_cofs_send_nb,
   mca_oob_cofs_recv_nb,
-  mca_oob_cofs_finalize
+  NULL,
+  mca_oob_cofs_module_init,
+  mca_oob_cofs_module_fini
 };
 
 char mca_oob_cofs_comm_loc[OMPI_PATH_MAX];
@@ -109,7 +111,13 @@ mca_oob_t* mca_oob_cofs_init(bool *allow_multi_user_threads, bool *have_hidden_t
 }
 
 
-int mca_oob_cofs_finalize(mca_oob_t* oob)
+int mca_oob_cofs_module_init(void)
 {
-  return OMPI_SUCCESS;
+    return OMPI_SUCCESS;
 }
+
+int mca_oob_cofs_module_fini(void)
+{
+    return OMPI_SUCCESS;
+}
+

src/mca/oob/oob.h

@@ -136,9 +136,15 @@ typedef int (*mca_oob_base_module_recv_nb_fn_t)(
     void* cbdata);
 
 /**
- * OOB finalize function
+ * Hook function called by mca_oob_base_register to allow
+ * the oob component a chance to register contact information
  */
-typedef int (*mca_oob_base_module_finalize_fn_t)(mca_oob_t*);
+typedef int (*mca_oob_base_module_init_fn_t)(void);
+
+/**
+ * Cleanup during finalize.
+ */
+typedef int (*mca_oob_base_module_fini_fn_t)(void);
 
 /**
  * OOB Module
@@ -150,7 +156,8 @@ struct mca_oob_1_0_0_t {
     mca_oob_base_module_recv_fn_t      oob_recv;
     mca_oob_base_module_send_nb_fn_t   oob_send_nb;
     mca_oob_base_module_recv_nb_fn_t   oob_recv_nb;
-    mca_oob_base_module_finalize_fn_t  oob_finalize;
+    mca_oob_base_module_init_fn_t      oob_init;
+    mca_oob_base_module_fini_fn_t      oob_fini;
 };
 
 /**
@@ -221,6 +228,7 @@ extern "C" {
 #endif
     int mca_oob_base_open(void);
     int mca_oob_base_init(bool *allow_multi_user_threads, bool *have_hidden_threads);
+    int mca_oob_base_register(void);
     int mca_oob_base_close(void);
 #if defined(c_plusplus) || defined(__cplusplus)
 }

src/mca/oob/tcp/oob_tcp.c

@@ -12,6 +12,9 @@
 #include "util/output.h"
 #include "util/if.h"
 #include "mca/oob/tcp/oob_tcp.h"
+#include "mca/ns/ns.h"
+#include "mca/gpr/base/base.h"
+#include "mca/gpr/gpr.h"
 
 
 static int  mca_oob_tcp_create_listen(void);
@@ -27,16 +30,16 @@ mca_oob_tcp_component_t mca_oob_tcp_component = {
     {
         MCA_OOB_BASE_VERSION_1_0_0,
         "tcp", /* MCA module name */
-        1,  /* MCA module major version */
+        1,  /* MCA component major version */
-        0,  /* MCA module minor version */
+        0,  /* MCA component minor version */
-        0,  /* MCA module release version */
+        0,  /* MCA component release version */
-        mca_oob_tcp_open,  /* module open */
+        mca_oob_tcp_component_open,  /* component open */
-        mca_oob_tcp_close /* module close */
+        mca_oob_tcp_component_close /* component close */
     },
     {
         false /* checkpoint / restart */
     },
-    mca_oob_tcp_init    /* module init */
+    mca_oob_tcp_component_init 
   }
 };
 
@@ -47,7 +50,8 @@ static mca_oob_t mca_oob_tcp = {
     mca_oob_tcp_recv,
     mca_oob_tcp_send_nb,
     mca_oob_tcp_recv_nb,
-    mca_oob_tcp_finalize
+    mca_oob_tcp_init,
+    mca_oob_tcp_fini,
 };
 
 
@@ -80,7 +84,7 @@ static inline char* mca_oob_tcp_param_register_str(
 /*
  * Initialize global variables used w/in this module.
  */
-int mca_oob_tcp_open(void)
+int mca_oob_tcp_component_open(void)
 {
     OBJ_CONSTRUCT(&mca_oob_tcp_component.tcp_peer_list,  ompi_list_t);
     OBJ_CONSTRUCT(&mca_oob_tcp_component.tcp_peer_tree,  ompi_rb_tree_t);
@@ -108,7 +112,7 @@ int mca_oob_tcp_open(void)
  * Cleanup of global variables used by this module.
  */
 
-int mca_oob_tcp_close(void)
+int mca_oob_tcp_component_close(void)
 {
     OBJ_DESTRUCT(&mca_oob_tcp_component.tcp_peer_list);
     OBJ_DESTRUCT(&mca_oob_tcp_component.tcp_peer_tree);
@@ -174,11 +178,10 @@ static int mca_oob_tcp_create_listen(void)
             errno);
     }
 
-    /* bind to all addresses and dynamically assigned port */
     memset(&inaddr, 0, sizeof(inaddr));
     inaddr.sin_family = AF_INET;
     inaddr.sin_addr.s_addr = INADDR_ANY;
-    inaddr.sin_port = htons(5000+mca_oob_name_self.vpid);
+    inaddr.sin_port = 0;
 
     if(bind(mca_oob_tcp_component.tcp_listen_sd, (struct sockaddr*)&inaddr, sizeof(inaddr)) < 0) {
         ompi_output(0,"mca_oob_tcp_create_listen: bind() failed with errno=%d", errno);
@@ -278,11 +281,11 @@ static void mca_oob_tcp_recv_handler(int sd, short flags, void* user)
 }
 
 /*
- * Module initialization.
+ * Component initialization - create a module.
  * (1) initialize static resources
  * (2) create listen socket
  */
-mca_oob_t* mca_oob_tcp_init(bool *allow_multi_user_threads, bool *have_hidden_threads)
+mca_oob_t* mca_oob_tcp_component_init(bool *allow_multi_user_threads, bool *have_hidden_threads)
 {
     /* initialize data structures */
     ompi_rb_tree_init(&mca_oob_tcp_component.tcp_peer_tree, (ompi_rb_tree_comp_fn_t)mca_oob_tcp_process_name_compare);
@@ -303,7 +306,6 @@ mca_oob_t* mca_oob_tcp_init(bool *allow_multi_user_threads, bool *have_hidden_th
         8,  /* increment to grow by */
         NULL); /* use default allocator */
 
-#if 0
     /* intialize event library */
     memset(&mca_oob_tcp_component.tcp_recv_event, 0, sizeof(ompi_event_t));
     memset(&mca_oob_tcp_component.tcp_send_event, 0, sizeof(ompi_event_t));
@@ -318,16 +320,36 @@ mca_oob_t* mca_oob_tcp_init(bool *allow_multi_user_threads, bool *have_hidden_th
         return NULL;
     }
     return &mca_oob_tcp;
-#else
-    return NULL;
-#endif
 }
 
+/*
+ * Setup contact information in the registry.
+ */
+int mca_oob_tcp_init(void)
+{
+    char *keys[3];
+    char *addr;
+    int rc;
+
+    /* put contact info in registry */
+    keys[0] = "tcp";
+    keys[1] = ompi_name_server.get_proc_name_string(&mca_oob_name_self);
+    keys[2] = NULL;
+
+    addr = mca_oob_tcp_get_addr();
+    rc = ompi_registry.put(OMPI_REGISTRY_OVERWRITE, "oob", keys, addr, strlen(addr)+1);
+    free(addr);
+    if(rc != OMPI_SUCCESS) {
+        ompi_output(0, "mca_oob_tcp_init: unable to contact registry.");
+        return rc;
+    }
+    return OMPI_SUCCESS;
+}
 
 /*
  * Module cleanup.
  */
-int mca_oob_tcp_finalize(mca_oob_t* oob)
+int mca_oob_tcp_fini(void)
 {
     mca_oob_tcp_peer_t * peer;
  
@@ -407,7 +429,7 @@ char* mca_oob_tcp_get_addr(void)
 * Parse a URI string into an IP address and port number.
 */
 
-static int mca_oob_tcp_parse_uri(const char* uri, struct sockaddr_in* inaddr)
+int mca_oob_tcp_parse_uri(const char* uri, struct sockaddr_in* inaddr)
 {
     char* tmp = strdup(uri);
     char* ptr = tmp + 6;

src/mca/oob/tcp/oob_tcp.h

@@ -26,12 +26,23 @@ extern "C" {
 #endif
 
 /*
- * standard module functions
+ * standard component functions
  */
-int mca_oob_tcp_open(void);
+int        mca_oob_tcp_component_open(void);
-int mca_oob_tcp_close(void);
+int        mca_oob_tcp_component_close(void);
-mca_oob_t* mca_oob_tcp_init(bool *allow_multi_user_threads, bool *have_hidden_threads);
+mca_oob_t* mca_oob_tcp_component_init(bool *allow_multi_user_threads, bool *have_hidden_threads);
-int mca_oob_tcp_finalize(mca_oob_t*);
+
+/**
+ * Hook function to allow the selected oob components
+ * to register their contact info with the registry
+*/
+
+int mca_oob_tcp_init(void);
+
+/**
+ * Cleanup resources during shutdown.
+ */
+int mca_oob_tcp_fini(void);
 
 /**
  * Convert process name from network to host byte order.
@@ -166,6 +177,14 @@ int mca_oob_tcp_recv_nb(
     mca_oob_callback_fn_t cbfunc, 
     void* cbdata);
 
+/**
+ *  Parse a URI string into an IP address and port number.
+ */
+int mca_oob_tcp_parse_uri(
+    const char* uri, 
+    struct sockaddr_in* inaddr
+);
+
 
 /**
  *  OOB TCP Component

src/mca/oob/tcp/oob_tcp_peer.c

@@ -8,6 +8,8 @@
 #include <arpa/inet.h>
 #include "util/output.h"
 #include "mca/oob/tcp/oob_tcp_peer.h"
+#include "mca/gpr/base/base.h"
+#include "mca/gpr/gpr.h"
 
 
 static int  mca_oob_tcp_peer_start_connect(mca_oob_tcp_peer_t* peer);
@@ -163,12 +165,6 @@ mca_oob_tcp_peer_t * mca_oob_tcp_peer_lookup(ompi_process_name_t* name, bool get
     peer->peer_recv_msg = NULL;
     peer->peer_send_msg = NULL;
     peer->peer_retries = 0;
-    /******
-     * need to add the peer's address to the structure
-     ******/
-    peer->peer_addr.sin_family = AF_INET;
-    peer->peer_addr.sin_port = htons(5000+peer->peer_name.vpid);
-    peer->peer_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
 
     if(OMPI_SUCCESS != ompi_rb_tree_insert(&mca_oob_tcp_component.tcp_peer_tree, &peer->peer_name, peer)) {
         MCA_OOB_TCP_PEER_RETURN(peer);
@@ -778,12 +774,32 @@ int mca_oob_tcp_peer_name_lookup(mca_oob_tcp_peer_t* peer)
         peer->peer_addr = mca_oob_tcp_component.tcp_seed_addr;
         return OMPI_SUCCESS;
     } else {
-        peer->peer_addr.sin_family = AF_INET;
+        ompi_registry_value_t *item;
-        peer->peer_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
+        ompi_list_t* items;
-        peer->peer_addr.sin_port = htons(5000+peer->peer_name.vpid);
+        char *keys[3];
-    }
+        char *uri = NULL;
 
-    /* insert code to resolve name via registry */
+        /* lookup the name in the registry */
-    return OMPI_ERROR;
+        keys[0] = "tcp";
+        keys[1] = ompi_name_server.get_proc_name_string(&peer->peer_name);
+        keys[2] = NULL;
+        items = ompi_registry.get(OMPI_REGISTRY_AND, "oob", keys);
+        if(items == NULL || ompi_list_get_size(items) == 0)
+            return OMPI_ERR_UNREACH;
+
+        /* unpack the results into a uri string */
+        item = (ompi_registry_value_t*)ompi_list_remove_first(items);
+        if((uri = item->object) == NULL) 
+            return OMPI_ERR_UNREACH;
+
+        /* validate the result */
+        if(mca_oob_tcp_parse_uri(uri, &peer->peer_addr) != OMPI_SUCCESS) {
+            OBJ_RELEASE(item);
+            return OMPI_ERR_UNREACH;
+        }
+        OBJ_RELEASE(item);
+        return OMPI_SUCCESS;
+    }
 }
 
+

src/runtime/ompi_rte_init.c

@@ -17,6 +17,7 @@
 #include "mca/llm/base/base.h"
 #include "mca/oob/oob.h"
 #include "mca/ns/base/base.h"
+#include "mca/gpr/base/base.h"
 #include "util/proc_info.h"
 #include "util/session_dir.h"
 #include "util/sys_info.h"
@@ -162,12 +163,28 @@ int ompi_rte_init(bool *allow_multi_user_threads, bool *have_hidden_threads)
   if (OMPI_SUCCESS != (ret = mca_oob_base_init(&user_threads, 
                                                &hidden_threads))) {
     /* JMS show_help */
-    printf("show_help: ompi_rte_init failed in oob_base_init\n");
+    printf("show_help: ompi_rte_init failed in mca_oob_base_init()\n");
     return ret;
   }
   *allow_multi_user_threads &= user_threads;
   *have_hidden_threads |= hidden_threads;
 
+  /*
+   * Registry 
+   */
+  if (OMPI_SUCCESS != (ret = mca_gpr_base_open())) {
+    /* JMS show_help */
+    printf("show_help: ompi_rte_init failed in mca_gpr_base_open()\n");
+    return ret;
+  }
+  if (OMPI_SUCCESS != (ret = mca_gpr_base_select(&user_threads, 
+                                               &hidden_threads))) {
+    /* JMS show_help */
+    printf("show_help: ompi_rte_init failed in mca_gpr_base_select()\n");
+    return ret;
+  }
+  *allow_multi_user_threads &= user_threads;
+  *have_hidden_threads |= hidden_threads;
 
   /*
    * Fill in the various important structures
@@ -198,14 +215,20 @@ int ompi_rte_init(bool *allow_multi_user_threads, bool *have_hidden_threads)
 
 
   /*
-   * Call back into NS/GPR to allow them to do any final initialization
+   * Call back into NS/GPR/OOB to allow them to do any final initialization
-   * (e.g. register callbacks w/ OOB).
+   * (e.g. register callbacks w/ OOB, put contact info in register).
   */
   if (OMPI_SUCCESS != (ret = ompi_name_server.init())) {
       printf("show_help: ompi_rte_init failed in ompi_name_server.init()\n");
       return ret;
   }
 
+  if (OMPI_SUCCESS != (ret = mca_oob_base_register())) {
+      printf("show_help: ompi_rte_init failed in mca_oob_base_register()\n");
+      return ret;
+  }
+
+
   /* 
    * All done 
    */

test/mca/Makefile.am

@@ -5,6 +5,6 @@
 
 include $(top_srcdir)/config/Makefile.options
 
-SUBDIRS		= oob llm pcm
+SUBDIRS		= oob llm pcm ns gpr
 
 DIST_SUBDIRS = $(SUBDIRS) ns 

test/mca/gpr/Makefile.am

@@ -7,8 +7,8 @@ include $(top_srcdir)/config/Makefile.options
 AM_CPPFLAGS = -I$(top_srcdir)/test/support -DOMPI_ENABLE_DEBUG_OVERRIDE=1
 
 noinst_PROGRAMS = \
-	test_gpr_replica \
+    test_gpr_proxy \
-        test_gpr_proxy
+	test_gpr_replica 
 
 test_gpr_replica_SOURCES = test_gpr_replica.c
 test_gpr_replica_LDADD = \
@@ -16,9 +16,9 @@ test_gpr_replica_LDADD = \
         $(top_builddir)/test/support/libsupport.la 
 test_gpr_replica_DEPENDENCIES = $(test_gpr_replica_LDADD)
 
-
 test_gpr_proxy_SOURCES = test_gpr_proxy.c
 test_gpr_proxy_LDADD = \
         $(top_builddir)/src/libmpi.la \
         $(top_builddir)/test/support/libsupport.la 
 test_gpr_proxy_DEPENDENCIES = $(test_gpr_proxy_LDADD)
+

test/mca/gpr/test_gpr_proxy.c

@@ -0,0 +1,282 @@
+/*
+ * unit test for gpr proxy.
+
+ --------------------------------------------------------------------------
+
+ Authors:	Ralph H. Castain <rhc@lanl.gov>
+            Tim S. Woodall <twoodall@lanl.gov>
+
+ --------------------------------------------------------------------------
+
+*/
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+
+#include "ompi_config.h"
+#include "support.h"
+
+#include "include/constants.h"
+#include "util/sys_info.h"
+#include "util/proc_info.h"
+#include "event/event.h"
+#include "runtime/runtime.h"
+#include "mca/mca.h"
+#include "mca/base/base.h"
+#include "mca/ns/base/base.h"
+#include "mca/oob/base/base.h"
+#include "mca/gpr/gpr.h"
+#include "mca/gpr/base/base.h"
+#include "mca/gpr/replica/gpr_replica.h"
+#include "mca/gpr/replica/gpr_replica_internals.h"
+
+
+/* output files needed by the test */
+static FILE *test_out=NULL;
+
+static char *cmd_str="diff ./test_gpr_proxy_out ./test_gpr_replica_out_std";
+static int run_test(void);
+static int exec_client(int argc, char** argv);
+
+
+int main(int argc, char **argv)
+{
+    bool multi, hidden;
+    char *seed;
+
+    test_init("test_gpr_proxy");
+    test_out = fopen( "test_ns_proxy_out", "w+" );
+    if( test_out == NULL ) {
+      test_failure("test_ns_proxy couldn't open test file failed");
+      test_finalize();
+      exit(1);
+    } 
+
+    /* setup environment for rte */
+    seed = getenv("OMPI_DAEMON_SEED");
+    setenv("OMPI_MCA_pcmclient_env_num_procs", "2", 1);
+    setenv("OMPI_MCA_pcmclient_env_vpid_start", "0", 1);
+    setenv("OMPI_MCA_pcmclient_env_cellid", "0", 1);
+    setenv("OMPI_MCA_pcmclient_env_jobid", "0", 1);
+    if(seed == NULL || atoi(seed) != 0) {
+        ompi_process_info.seed = true;
+        setenv("OMPI_MCA_pcmclient_env_procid", "0", 1);
+    } else {
+        ompi_process_info.seed = false;
+        setenv("OMPI_MCA_pcmclient_env_procid", "1", 1);
+    }
+    /* require tcp oob */
+    setenv("OMPI_MCA_oob_base_include", "tcp", 1);
+
+    /* basic ompi init */
+    if (OMPI_SUCCESS != ompi_init(argc, argv)) {
+	    fprintf(test_out, "ompi_init failed - please report error to bugs@open-mpi.org\n");
+	    exit (1);
+    }
+
+    /* initialize event library */
+    if (OMPI_SUCCESS != ompi_event_init()) {
+	    fprintf(test_out, "ompi_event_init failed - please report error to bugs@open-mpi.org\n");
+	    exit (1);
+    }
+
+    /* startup the MCA */
+    if (OMPI_SUCCESS == mca_base_open()) {
+	    fprintf(test_out, "MCA started\n");
+    } else {
+	    fprintf(test_out, "MCA could not start - please report error to bugs@open-mpi.org\n");
+	    exit (1);
+    }
+
+    /* initialize the rte - including ns and oob */
+    if(OMPI_SUCCESS == ompi_rte_init(&multi, &hidden)) {
+	    fprintf(test_out, "NS opened\n");
+	    fprintf(test_out, "NS started\n");
+    } else {
+	    fprintf(test_out, "RTE could not start - please report error to bugs@open-mpi.org\n");
+	    exit (1);
+    }
+
+    /* if daemon seed - just wait for requests */
+    if(ompi_process_info.seed) {
+#if 0
+        /* wait on child to exit */
+        int pid = exec_client(argc, argv);
+        while(true) {
+            int status;
+            if(waitpid(pid, &status, WNOHANG) != 0)
+                break;
+            ompi_event_loop(OMPI_EVLOOP_NONBLOCK);
+        }
+#else
+        fprintf(stderr, "OMPI_MCA_oob_base_seed=%s", mca_oob_get_contact_info());
+        ompi_event_loop(0);
+#endif
+        return(0);
+    } else {
+        return run_test();
+    }
+}
+ 
+
+int exec_client(int argc, char** argv)
+{
+    int pid;
+    pid = fork();
+    if(pid < 0) {
+	    fprintf(test_out, "fork() failed with errno=%d\n", errno);
+        exit(1);
+    } else if (pid == 0) {
+        /* child process should not be a daemon */
+        static char seed_flag[] = "OMPI_DAEMON_SEED=0";
+        static char seed_addr[128];
+        char* argp[] = {
+            seed_flag,
+            seed_addr,
+            NULL
+        };
+        /* setup seed address for child process */
+        sprintf(seed_addr, "OMPI_MCA_oob_base_seed=%s", mca_oob_get_contact_info());
+        execve(argv[0], argv, argp);
+    } 
+    return pid;
+}
+
+
+int run_test()
+{
+    mca_gpr_replica_key_t test_key, test_key2;
+    ompi_list_t *test_list, *internal_tests;
+    ompi_registry_index_value_t *ptr;
+    ompi_registry_internal_test_results_t *ptri;
+    ompi_registry_object_t *test_buffer;
+    uint8_t *test_buf;
+    ompi_registry_object_size_t input_size;
+    ompi_registry_mode_t mode;
+    bool multi, hidden;
+    int i, j;
+    bool success;
+    char *tmp;
+    char name[30], *name2[30];
+    int result, put_test; /* result from system call */
+
+    /* check index */
+    test_list = ompi_registry.index(NULL);
+    if (0 == ompi_list_get_size(test_list)) { /* should have been something in dictionary */
+	fprintf(test_out, "GPR replica: index function failed\n");
+	test_failure("test_gpr_replica index_global_dictionary failed\n");
+	test_finalize();
+	exit(1);
+    } else {
+	fprintf(test_out, "GPR index returned list\n");
+	for (ptr = (ompi_registry_index_value_t*)ompi_list_get_first(test_list);
+	     ptr != (ompi_registry_index_value_t*)ompi_list_get_end(test_list);
+	     ptr = (ompi_registry_index_value_t*)ompi_list_get_next(ptr)) {
+	    fprintf(test_out, "\t%s\n", ptr->token);
+	}
+	test_success();
+    }
+
+
+    /* check internals */
+    internal_tests = ompi_registry.test_internals(1);
+    if (0 == ompi_list_get_size(internal_tests)) { /* should have been something in list */
+	fprintf(test_out, "internal tests failed\n");
+	test_failure("test_gpr_replica internal_tests failed\n");
+	test_finalize();
+	exit(1);
+    } else {
+	fprintf(test_out, "internal test results list\n");
+	for (ptri = (ompi_registry_internal_test_results_t*)ompi_list_get_first(internal_tests);
+	     ptri != (ompi_registry_internal_test_results_t*)ompi_list_get_end(internal_tests);
+	     ptri = (ompi_registry_internal_test_results_t*)ompi_list_get_next(ptri)) {
+	    fprintf(test_out, "\t%s\n", ptri->test);
+	    fprintf(test_out, "\t%s\n", ptri->message);
+	}
+	test_success();
+    }
+
+    /* test the put function */
+    success = true;
+    input_size = 10000;
+    test_buffer = (ompi_registry_object_t*)malloc(input_size);
+    test_buf = (uint8_t*)test_buffer;
+    for (i=0; i<input_size; i++) {
+	*test_buf = i % 256;
+	test_buf++;
+    }
+    for (i=0; i<5 && success; i++) {
+	sprintf(name, "test-def-seg%d", i);
+	for (j=0; j<10 && success; j++) {
+ 	    asprintf(&name2[j], "test-key%d", j);
+	}
+	name2[j] = NULL;
+	if (OMPI_SUCCESS != ompi_registry.put(OMPI_REGISTRY_NONE, name,
+					      name2, test_buffer, input_size)) {
+	    fprintf(test_out, "put test failed for segment %s\n", name);
+	    for (j=0; j<10; j++) {
+		fprintf(test_out, "\t%s\n", name2[j]);
+	    }
+	    success = false;
+	}
+    }
+    if (success) {
+	fprintf(test_out, "put test: success\n");
+	test_success();
+    } else {
+	fprintf(test_out, "put test failed\n");
+	test_failure("test_gpr_replica put_test failed\n");
+	test_finalize();
+	exit(1);
+    }
+
+    /* test the put overwrite function */
+    for (i=0; i<5 && success; i++) {
+	sprintf(name, "test-def-seg%d", i);
+	for (j=0; j<10 && success; j++) {
+ 	    asprintf(&name2[j], "test-key%d", j);
+	}
+	name2[j] = NULL;
+	if (10 %  i) {
+	    mode = OMPI_REGISTRY_OVERWRITE;
+	} else {
+	    mode = OMPI_REGISTRY_NONE;
+	}
+	put_test = ompi_registry.put(mode, name, name2, test_buffer, input_size);
+	if ((OMPI_REGISTRY_OVERWRITE == mode && OMPI_SUCCESS != put_test) ||
+	    (OMPI_REGISTRY_NONE == mode && OMPI_SUCCESS == put_test)) {
+	    fprintf(test_out, "put test failed for segment %s\n", name);
+	    for (j=0; j<10; j++) {
+		fprintf(test_out, "\t%s\n", name2[j]);
+	    }
+	    success = false;
+	}
+    }
+    if (success) {
+	fprintf(test_out, "put overwrite test: success\n");
+	test_success();
+    } else {
+	fprintf(test_out, "put overwrite test failed\n");
+	test_failure("test_gpr_replica put_test failed\n");
+	test_finalize();
+	exit(1);
+    }
+
+    /* check the universe segment - should have a key value of "1" */
+
+    fclose( test_out );
+    /*    result = system( cmd_str );
+    if( result == 0 ) {
+        test_success();
+    }
+    else {
+      test_failure( "test_gpr_replica ompi_registry init, etc failed");
+    }
+    */
+    test_finalize();
+    return(0);
+}

test/mca/ns/test_ns_proxy.c

@@ -51,22 +51,19 @@ int main(int argc, char **argv)
       exit(1);
     } 
 
-    /* check wether or not we are seed */
+    /* setup environment for rte */
     seed = getenv("OMPI_DAEMON_SEED");
+    setenv("OMPI_MCA_pcmclient_env_num_procs", "2", 1);
+    setenv("OMPI_MCA_pcmclient_env_vpid_start", "0", 1);
+    setenv("OMPI_MCA_pcmclient_env_cellid", "0", 1);
+    setenv("OMPI_MCA_pcmclient_env_jobid", "0", 1);
     if(seed == NULL || atoi(seed) != 0) {
         ompi_process_info.seed = true;
-        /* setup the environment for cofs pcm */
+        setenv("OMPI_MCA_pcmclient_env_procid", "0", 1);
-        setenv("OMPI_MCA_pcm_cofs_cellid", "0", 1);
-        setenv("OMPI_MCA_pcm_cofs_jobid", "0", 1);
-        setenv("OMPI_MCA_pcm_cofs_procid", "0", 1);
     } else {
         ompi_process_info.seed = false;
-        /* setup the environment for cofs pcm */
+        setenv("OMPI_MCA_pcmclient_env_procid", "1", 1);
-        setenv("OMPI_MCA_pcm_cofs_cellid", "0", 1);
-        setenv("OMPI_MCA_pcm_cofs_jobid", "0", 1);
-        setenv("OMPI_MCA_pcm_cofs_procid", "1", 1);
     }
-    setenv("OMPI_MCA_pcm_cofs_num_procs", "2", 1);
     /* require tcp oob */
     setenv("OMPI_MCA_oob_base_include", "tcp", 1);
 
@@ -101,6 +98,7 @@ int main(int argc, char **argv)
 
     /* if daemon seed - just wait for requests */
     if(ompi_process_info.seed) {
+#if 0
         /* wait on child to exit */
         int pid = exec_client(argc, argv);
         while(true) {
@@ -109,6 +107,10 @@ int main(int argc, char **argv)
                 break;
             ompi_event_loop(OMPI_EVLOOP_NONBLOCK);
         }
+#else
+        fprintf(stderr, "OMPI_MCA_oob_base_seed=%s", mca_oob_get_contact_info());
+        ompi_event_loop(0);
+#endif
     } else {
         run_test();
     }