gds/shmem: Support multiple local clients.

Tested on a single node, multiple clients now appear to share static
job-level information via shared-memory. We still have plenty of work
ahead, but at least it doesn't crash now. We now cache a pointer to a
full-featured gds module and pass certain operations to it.

Signed-off-by: Samuel K. Gutierrez <samuel@lanl.gov>

Author: samuelkgutierrez

src/mca/gds/shmem/gds_shmem.c

@@ -59,6 +59,9 @@
 // TODO(skg) Address FT case at some point. Need to have a broader conversion
 // about how we go about doing this. Ralph has some ideas.
 
+// TODO(skg) Consider using mprotect.
+// TODO(skg) We way need to implement a different hash table.
+
 /**
  * Key names used to find shared-memory segment info.
  */
@@ -270,6 +273,7 @@ job_construct(
 ) {
     job->nspace_id = NULL;
     job->nspace = NULL;
+    job->ffgds = NULL;
     job->shmem = PMIX_NEW(pmix_shmem_t);
     job->smdata = NULL;
 }
@@ -284,6 +288,7 @@ job_destruct(
     if (job->nspace) {
         PMIX_RELEASE(job->nspace);
     }
+    job->ffgds = NULL;
     // This will release the memory for the structures located in the
     // shared-memory segment.
     if (job->shmem) {
@@ -959,6 +964,13 @@ store_job_info(
     // Update the TMA to point to its local function pointers.
     tma_init_function_pointers(&job->smdata->tma);
     pmix_gds_shmem_vout_smdata(job);
+    // Proctect memory: clients can only read from here.
+#if 0
+    mprotect(
+        job->shmem->base_address,
+        job->shmem->size, PROT_READ
+    );
+#endif
     // Done. Before this point the server should have populated the
     // shared-memory segment with the relevant data.
     return rc;
@@ -970,9 +982,18 @@ store(
     pmix_scope_t scope,
     pmix_kval_t *kval
 ) {
-    PMIX_HIDE_UNUSED_PARAMS(proc, scope, kval);
     PMIX_GDS_SHMEM_VOUT_HERE();
-    return PMIX_ERR_NOT_SUPPORTED;
+
+    pmix_status_t rc = PMIX_SUCCESS;
+    // Setup a job tracker for this peer's nspace.
+    pmix_gds_shmem_job_t *job;
+    rc = pmix_gds_shmem_get_job_tracker(proc->nspace, false, &job);
+    if (PMIX_SUCCESS != rc) {
+        PMIX_ERROR_LOG(rc);
+        return rc;
+    }
+    // Let a full-featured gds module handle this.
+    return job->ffgds->store(proc, scope, kval);
 }
 
 /**
@@ -982,13 +1003,23 @@ store(
  */
 static pmix_status_t
 store_modex(
-    struct pmix_namespace_t *nspace,
+    struct pmix_namespace_t *nspace_struct,
     pmix_buffer_t *buff,
     void *cbdata
 ) {
-    PMIX_HIDE_UNUSED_PARAMS(nspace, buff, cbdata);
     PMIX_GDS_SHMEM_VOUT_HERE();
-    return PMIX_ERR_NOT_SUPPORTED;
+
+    pmix_status_t rc = PMIX_SUCCESS;
+    pmix_namespace_t *nspace = (pmix_namespace_t *)nspace_struct;
+    // Setup a job tracker for this peer's nspace.
+    pmix_gds_shmem_job_t *job;
+    rc = pmix_gds_shmem_get_job_tracker(nspace->nspace, false, &job);
+    if (PMIX_SUCCESS != rc) {
+        PMIX_ERROR_LOG(rc);
+        return rc;
+    }
+    // Let a full-featured gds module handle this.
+    return job->ffgds->store_modex(nspace_struct, buff, cbdata);
 }
 
 static pmix_status_t
@@ -998,6 +1029,7 @@ server_setup_fork(
 ) {
     PMIX_HIDE_UNUSED_PARAMS(peer, env);
     PMIX_GDS_SHMEM_VOUT_HERE();
+    // Nothing to do here.
     return PMIX_SUCCESS;
 }
 
@@ -1010,9 +1042,11 @@ server_add_nspace(
 ) {
     PMIX_HIDE_UNUSED_PARAMS(nspace, nlocalprocs, info, ninfo);
     PMIX_GDS_SHMEM_VOUT_HERE();
+    // Nothing to do here.
     return PMIX_SUCCESS;
 }
 
+// TODO(skg) Implement.
 static pmix_status_t
 del_nspace(
     const char *nspace
@@ -1029,18 +1063,37 @@ assemb_kvs_req(
     pmix_buffer_t *buff,
     void *cbdata
 ) {
-    PMIX_HIDE_UNUSED_PARAMS(proc, kvs, buff, cbdata);
     PMIX_GDS_SHMEM_VOUT_HERE();
-    return PMIX_ERR_NOT_SUPPORTED;
+
+    pmix_status_t rc = PMIX_SUCCESS;
+    // Setup a job tracker for this peer's nspace.
+    pmix_gds_shmem_job_t *job;
+    rc = pmix_gds_shmem_get_job_tracker(proc->nspace, false, &job);
+    if (PMIX_SUCCESS != rc) {
+        PMIX_ERROR_LOG(rc);
+        return rc;
+    }
+    // Let a full-featured gds module handle this.
+    return job->ffgds->assemb_kvs_req(proc, kvs, buff, cbdata);
 }
 
 static pmix_status_t
 accept_kvs_resp(
     pmix_buffer_t *buff
 ) {
-    PMIX_HIDE_UNUSED_PARAMS(buff);
     PMIX_GDS_SHMEM_VOUT_HERE();
-    return PMIX_SUCCESS;
+
+    pmix_status_t rc = PMIX_SUCCESS;
+    const char *nspace = pmix_globals.mypeer->nptr->nspace;
+    // Setup a job tracker for this peer's nspace.
+    pmix_gds_shmem_job_t *job;
+    rc = pmix_gds_shmem_get_job_tracker(nspace, false, &job);
+    if (PMIX_SUCCESS != rc) {
+        PMIX_ERROR_LOG(rc);
+        return rc;
+    }
+    // Let a full-featured gds module handle this.
+    return job->ffgds->accept_kvs_resp(buff);
 }
 
 pmix_gds_base_module_t pmix_shmem_module = {

src/mca/gds/shmem/gds_shmem.h

@@ -124,6 +124,8 @@ typedef struct {
     char *nspace_id;
     /** Pointer to the namespace. */
     pmix_namespace_t *nspace;
+    /** Pointer to a full-featured gds module. */
+    pmix_gds_base_module_t *ffgds;
     /** Shared-memory object. */
     pmix_shmem_t *shmem;
     /** Points to shared data located in shared-memory segment. */

src/mca/gds/shmem/gds_shmem_fetch.c

@@ -509,13 +509,6 @@ pmix_gds_shmem_fetch(
     }
     pmix_hash_table2_t *ht = job->smdata->local_hashtab;
 
-    // TODO(skg) Cache during init.
-    pmix_info_t ginfo;
-    PMIX_INFO_LOAD(&ginfo, PMIX_GDS_MODULE, "hash", PMIX_STRING);
-    pmix_gds_base_module_t *hashmod = pmix_gds_base_assign_module(&ginfo, 1);
-    assert(hashmod);
-    PMIX_INFO_DESTRUCT(&ginfo);
-
     if (NULL == key && PMIX_RANK_WILDCARD == proc->rank) {
         assert(false);
         return PMIX_ERR_NOT_SUPPORTED;
@@ -524,7 +517,7 @@ pmix_gds_shmem_fetch(
     for (size_t n = 0; n < nqual; n++) {
         if (PMIX_CHECK_KEY(&qualifiers[n], PMIX_SESSION_INFO)) {
             // We don't handle session info, so pass it along.
-            return hashmod->fetch(
+            return job->ffgds->fetch(
                 proc, scope, copy, key, qualifiers, nqual, kvs
             );
         }
@@ -606,14 +599,14 @@ pmix_gds_shmem_fetch(
             rc = pmix_hash2_fetch(ht, PMIX_RANK_WILDCARD, NULL, NULL, 0, kvs);
         }
         else {
-            return hashmod->fetch(proc, scope, copy, key, qualifiers, nqual, kvs);
+            return job->ffgds->fetch(proc, scope, copy, key, qualifiers, nqual, kvs);
         }
     }
     else {
         rc = pmix_hash2_fetch(ht, proc->rank, key, qualifiers, nqual, kvs);
     }
     if (PMIX_SUCCESS != rc) {
-        rc = hashmod->fetch(proc, scope, copy, key, qualifiers, nqual, kvs);
+        rc = job->ffgds->fetch(proc, scope, copy, key, qualifiers, nqual, kvs);
     }
 
     return rc;

src/mca/gds/shmem/gds_shmem_utils.c

@@ -100,6 +100,8 @@ pmix_gds_shmem_get_job_tracker(
         }
         PMIX_RETAIN(inspace);
         target_tracker->nspace = inspace;
+        // Cache a handle to a full-featured gds module.
+        target_tracker->ffgds = pmix_globals.mypeer->nptr->compat.gds;
         // Add it to the list of jobs I'm supporting.
         pmix_list_append(&component->jobs, &target_tracker->super);
     }