Changeset 4747

Timestamp:

06/16/09 16:39:07 (5 months ago)

Author:

goodell

Message:

Merge process mapping fixes from trunk -> mpich2-1.1 branch.

This commit significantly reduces the amount of time that MPI_Init takes
when using mpd on systems with many nodes.

This is the merge of 4 commits:
r4666 - Merging process mapping code into trunk
r4671 - Fixed srcdir typo in the mpd makefile
r4705 - Prevent mpds from crashing when jobs are launched that don't use all hosts.
r4706 - Multiple fixes for mpd's new process mapping capability.

Location:

mpich2/branches/release/mpich2-1.1/src

Files:

• mpid/ch3/src/mpid_vc.c (modified) (6 diffs)
• pm/mpd/Makefile.in (modified) (1 diff)
• pm/mpd/mpd.py (modified) (11 diffs)
• pm/mpd/mpdlib.py (modified) (1 diff)
• pm/mpd/mpdman.py (modified) (2 diffs)

mpich2/branches/release/mpich2-1.1/src/mpid/ch3/src/mpid_vc.c

@@ -21 +21 @@
 #include <errno.h>
 #endif
+#include <ctype.h>
 
 
@@ -797 +798 @@
 #endif
 
+
+#define parse_error() MPIU_ERR_SETANDJUMP1(mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "parse error")
+/* advance _c until we find a non whitespace character */
+#define skip_space(_c) while (isspace(*(_c))) ++(_c)
+/* return true iff _c points to a character valid as an indentifier, i.e., [-_a-zA-Z0-9] */
+#define isident(_c) (isalnum(_c) || (_c) == '-' || (_c) == '_')
+
+/* give an error iff *_c != _e */
+#define expect_c(_c, _e) do { if (*(_c) != _e) parse_error(); } while (0)
+#define expect_and_skip_c(_c, _e) do { expect_c(_c, _e); ++c; } while (0)
+/* give an error iff the first |_m| characters of the string _s are equal to _e */
+#define expect_s(_s, _e) (strncmp(_s, _e, strlen(_e)) == 0 && !isident((_s)[strlen(_e)]))
+
+typedef enum {
+    NULL_MAPPING = 0,
+    VECTOR_MAPPING
+} mapping_type_t;
+
+#define VECTOR "vector"
+
+typedef struct map_block
+{
+    int start_id;
+    int count;
+    int size;
+} map_block_t;
+
+#undef FUNCNAME
+#define FUNCNAME parse_mapping
+#undef FCNAME
+#define FCNAME MPIDI_QUOTE(FUNCNAME)
+static int parse_mapping(char *map_str, mapping_type_t *type, map_block_t **map, int *nblocks)
+{
+    int mpi_errno = MPI_SUCCESS;
+    char *c = map_str, *d;
+    int num_blocks = 0;
+    int i;
+    MPIU_CHKPMEM_DECL(1);
+    MPIDI_STATE_DECL(MPID_STATE_PARSE_MAPPING);
+
+    MPIDI_FUNC_ENTER(MPID_STATE_PARSE_MAPPING);
+
+    /* parse string of the form:
+       '(' <format> ',' '(' <num> ',' <num> ',' <num> ')' {',' '(' <num> ',' <num> ',' <num> ')'} ')'
+
+       the values of each 3-tuple have the following meaning (X,Y,Z):
+         X - node id start value
+         Y - number of nodes with size Z
+         Z - number of processes assigned to each node
+     */
+    MPIU_DBG_MSG_S(CH3_OTHER,VERBOSE,"parsing mapping string '%s'", map_str);
+
+    if (!strlen(map_str)) {
+        /* An empty-string indicates an inability to determine or express the
+         * process layout on the part of the process manager.  Consider this a
+         * non-fatal error case. */
+        *type = NULL_MAPPING;
+        *map = NULL;
+        *nblocks = 0;
+        goto fn_exit;
+    }
+
+    skip_space(c);
+    expect_and_skip_c(c, '(');
+    skip_space(c);
+
+    d = c;
+    if (expect_s(d, VECTOR))
+        *type = VECTOR_MAPPING;
+    else
+        parse_error();
+    c += strlen(VECTOR);
+    skip_space(c);
+
+    /* first count the number of block descriptors */
+    d = c;
+    while (*d) {
+        if (*d == '(')
+            ++num_blocks;
+        ++d;
+    }
+
+    MPIU_CHKPMEM_MALLOC(*map, map_block_t *, sizeof(map_block_t) * num_blocks, mpi_errno, "map");
+
+    /* parse block descriptors */
+    for (i = 0; i < num_blocks; ++i) {
+        expect_and_skip_c(c, ',');
+        skip_space(c);
+
+        expect_and_skip_c(c, '(');
+        skip_space(c);
+
+        if (!isdigit(*c))
+            parse_error();
+        (*map)[i].start_id = strtol(c, &c, 0);
+        skip_space(c);
+
+        expect_and_skip_c(c, ',');
+        skip_space(c);
+
+        if (!isdigit(*c))
+            parse_error();
+        (*map)[i].count = strtol(c, &c, 0);
+        skip_space(c);
+
+        expect_and_skip_c(c, ',');
+        skip_space(c);
+
+        if (!isdigit(*c))
+            parse_error();
+        (*map)[i].size = strtol(c, &c, 0);
+
+        expect_and_skip_c(c, ')');
+        skip_space(c);
+    }
+
+    expect_and_skip_c(c, ')');
+
+    *nblocks = num_blocks;
+    MPIU_CHKPMEM_COMMIT();
+fn_exit:
+    MPIDI_FUNC_EXIT(MPID_STATE_PARSE_MAPPING);
+    return mpi_errno;
+fn_fail:
+    MPIU_CHKPMEM_REAP();
+    goto fn_exit;
+}
+
+#if 0
+static void t(const char *s, int nprocs)
+{
+    int ret;
+    map_block_t *mb;
+    int nblocks=0;
+    int i;
+    mapping_type_t mt = -1;
+    int rank;
+    int block, block_node, node_proc;
+
+    ret = parse_mapping(strdup(s), &mt, &mb, &nblocks);
+    printf("str=\"%s\" type=%d ret=%d\n", s, mt, ret);
+    if (ret) return;
+    for (i = 0; i < nblocks; ++i)
+        printf("    %d: start=%d size=%d count=%d\n", i, mb[i].start_id, mb[i].size, mb[i].count);
+    printf("\n");
+
+
+    rank = 0;
+    while (rank < nprocs) {
+        int node_id;
+        for (block = 0; block < nblocks; ++block) {
+            node_id = mb[block].start_id;
+            for (block_node = 0; block_node < mb[block].count; ++block_node) {
+                for (node_proc = 0; node_proc < mb[block].size; ++node_proc) {
+                    printf("    %d  %d\n", rank, node_id);
+                    ++rank;
+                    if (rank == nprocs)
+                        goto done;
+                }
+                ++node_id;
+            }
+        }
+    }
+done:
+    return;
+
+}
+
+
+ void test_parse_mapping(void)
+{
+    t("(vector, (0,1,1))", 5);
+    t("(vector, (0,1,1), (1,5,3), (6,2, 5))", 100);
+    t("(vector, (1,1,1), (0,2,2))", 5);
+    
+    t("(vector, (1,1,1), (0,2,2),)", 5);
+    t("XXX, (1,1))", 1);
+    t("vector, (1,1))", 1);
+    t("(vector, (1.11, 2,2))", 1);
+    t("", 1);
+
+}
+
+
+#endif
+
+#undef FUNCNAME
+#define FUNCNAME populate_ids_from_mapping
+#undef FCNAME
+#define FCNAME MPIDI_QUOTE(FUNCNAME)
+static int populate_ids_from_mapping(char *mapping, int *num_nodes, MPIDI_PG_t *pg, int *did_map)
+{
+    int mpi_errno = MPI_SUCCESS;
+    /* process-mapping is available */
+    mapping_type_t mt = -1;
+    map_block_t *mb = NULL;
+    int nblocks = 0;
+    int rank;
+    int block, block_node, node_proc;
+
+    *did_map = 1; /* reset upon failure */
+
+    mpi_errno = parse_mapping(mapping, &mt, &mb, &nblocks);
+    if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+    if (NULL_MAPPING == mt) goto fn_fail;
+    MPIU_ERR_CHKANDJUMP1(mt != VECTOR_MAPPING, mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "unsupported mapping type");
+
+    rank = 0;
+    /* for a representation like (block,N,(1,1)) this while loop causes us to
+     * re-use that sole map block over and over until we have assigned node
+     * ids to every process */
+    while (rank < pg->size) {
+        for (block = 0; block < nblocks; ++block) {
+            int node_id = mb[block].start_id;
+            for (block_node = 0; block_node < mb[block].count; ++block_node) {
+                if (node_id > *num_nodes)
+                    *num_nodes = node_id;
+
+                for (node_proc = 0; node_proc < mb[block].size; ++node_proc) {
+                    pg->vct[rank].node_id = node_id;
+                    ++rank;
+                    if (rank == pg->size)
+                        goto fn_exit;
+                }
+                ++node_id;
+            }
+        }
+    }
+
+fn_exit:
+    ++(*num_nodes); /* add one to get the num instead of the max */
+    MPIU_Free(mb);
+    return mpi_errno;
+fn_fail:
+    *did_map = 0;
+    goto fn_exit;
+}
+
 /* Fills in the node_id info from PMI info.  Adapted from MPIU_Get_local_procs.
    This function is collective over the entire PG because PMI_Barrier is called.
-   
+
    our_pg_rank should be set to -1 if this is not the current process' PG.  This
    is currently not supported due to PMI limitations.
 
-   Algorithm:
-  
+   Fallback Algorithm:
+
    Each process kvs_puts its hostname and stores the total number of
    processes (g_num_global).  Each process determines the number of nodes
    (g_num_nodes) and assigns a node id to each process (g_node_ids[]):
-   
+
      For each hostname the process seaches the list of unique nodes
      names (node_names[]) for a match.  If a match is found, the node id
@@ -814 +1054 @@
      added to the list of node names.
 */
+#undef FUNCNAME
+#define FUNCNAME MPIDI_Populate_vc_node_ids
+#undef FCNAME
+#define FCNAME MPIDI_QUOTE(FUNCNAME)
 int MPIDI_Populate_vc_node_ids(MPIDI_PG_t *pg, int our_pg_rank)
 {
@@ -822 +1066 @@
     int i, j;
     char *key;
+    char *value;
     int key_max_sz;
+    int val_max_sz;
     char *kvs_name;
     char **node_names;
@@ -828 +1074 @@
     int no_local = 0;
     int odd_even_cliques = 0;
-    MPIU_CHKLMEM_DECL(3);
-
-#ifdef USE_PMI2_API
-    {
-        int *node_ids;
-        int outlen;
-        int found = FALSE;
-        MPIU_CHKLMEM_MALLOC(node_ids, int *, pg->size * sizeof(int), mpi_errno, "node_ids");
-
-        mpi_errno = PMI_Info_GetJobAttrIntArray("nodeIDs", node_ids, pg->size, &outlen, &found);
-        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
-        MPIU_ERR_CHKANDJUMP1(outlen != pg->size, mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "did not receive enough nodeids");
-        g_num_nodes = 0;
-        for (i = 0; i < pg->size; ++i) {
-            pg->vct[i].node_id = node_ids[i];
-            if (g_num_nodes < node_ids[i])
-                g_num_nodes = node_ids[i];
-        }
-        
-        ++g_num_nodes;
-        
-        /* FIXME: need to handle oddeven cliques DARIUS */
-    }
-#else
-    if (our_pg_rank == -1) {
-        /* FIXME this routine can't handle the dynamic process case at this
-           time.  This will require more support from the process manager. */
-        MPIU_Assert(0);
-    }
-
-    mpi_errno = publish_node_id(pg, our_pg_rank);
-    if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+    MPIU_CHKLMEM_DECL(4);
+
+    if (pg->size == 1) {
+        pg->vct[0].node_id = g_num_nodes++;
+        goto fn_exit;
+    }
 
     /* Used for debugging only.  This disables communication over shared memory */
@@ -889 +1109 @@
     }
 
-    /* Allocate space for pmi key */
+#ifdef USE_PMI2_API
+#if 0 /* use nodeid list */
+    {
+        int *node_ids;
+        int outlen;
+        int found = FALSE;
+        MPIU_CHKLMEM_MALLOC(node_ids, int *, pg->size * sizeof(int), mpi_errno, "node_ids");
+
+        mpi_errno = PMI_Info_GetJobAttrIntArray("nodeIDs", node_ids, pg->size, &outlen, &found);
+        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+        MPIU_ERR_CHKANDJUMP1(!found, mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "nodeIDs attribute not found");
+        MPIU_ERR_CHKANDJUMP1(outlen != pg->size, mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "did not receive enough nodeids");
+        g_num_nodes = 0;
+        for (i = 0; i < pg->size; ++i) {
+            pg->vct[i].node_id = node_ids[i];
+            if (g_num_nodes < node_ids[i])
+                g_num_nodes = node_ids[i];
+        }
+
+        ++g_num_nodes;
+
+        /* FIXME: need to handle oddeven cliques DARIUS */
+    }
+#else
+    {
+        char process_mapping[PMI_MAX_VALLEN];
+        int outlen;
+        int found = FALSE;
+        int i;
+        map_block_t *mb;
+        int nblocks;
+        int rank;
+        int block, block_node, node_proc;
+        int did_map = 0;
+        int num_nodes = 0;
+
+        mpi_errno = PMI_Info_GetJobAttr("process-mapping", process_mapping, sizeof(process_mapping), &found);
+        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+        MPIU_ERR_CHKANDJUMP1(!found, mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "process-mapping attribute not found");
+        /* this code currently assumes pg is comm_world */
+        mpi_errno = populate_ids_from_mapping(process_mapping, &num_nodes, pg, &did_map);
+        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+        MPIU_ERR_CHKANDJUMP1(!did_map, mpi_errno, MPI_ERR_OTHER, "**intern", "**intern %s", "unable to populate node ids from process-mapping");
+        g_num_nodes = num_nodes;
+    }
+#endif
+#else /* USE_PMI2_API */
+    if (our_pg_rank == -1) {
+        /* FIXME this routine can't handle the dynamic process case at this
+           time.  This will require more support from the process manager. */
+        MPIU_Assert(0);
+    }
+
+    /* Allocate space for pmi key and value */
     pmi_errno = PMI_KVS_Get_key_length_max(&key_max_sz);
     MPIU_ERR_CHKANDJUMP1(pmi_errno, mpi_errno, MPI_ERR_OTHER, "**fail", "**fail %d", pmi_errno);
-
     MPIU_CHKLMEM_MALLOC(key, char *, key_max_sz, mpi_errno, "key");
 
+    pmi_errno = PMI_KVS_Get_value_length_max(&val_max_sz);
+    MPIU_ERR_CHKANDJUMP1(pmi_errno, mpi_errno, MPI_ERR_OTHER, "**fail", "**fail %d", pmi_errno);
+    MPIU_CHKLMEM_MALLOC(value, char *, val_max_sz, mpi_errno, "value");
+
     mpi_errno = MPIDI_PG_GetConnKVSname(&kvs_name);
+    if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+    /* See if process manager supports process-mapping keyval */
+
+    /* FIXME 'process-mapping' only applies for the original PG (MPI_COMM_WORLD) */
+    pmi_errno = PMI_KVS_Get(kvs_name, "process-mapping", value, val_max_sz);
+    if (pmi_errno == 0) {
+        int did_map = 0;
+        int num_nodes = 0;
+        /* this code currently assumes pg is comm_world */
+        mpi_errno = populate_ids_from_mapping(value, &num_nodes, pg, &did_map);
+        if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+        g_num_nodes = num_nodes;
+        if (did_map) {
+            goto fn_exit;
+        }
+        else {
+            MPIU_DBG_MSG_S(CH3_OTHER,TERSE,"did_map==0, unable to populate node ids from mapping=%s",value);
+        }
+        /* else fall through to O(N^2) PMI_KVS_Gets version */
+    }
+    else {
+        MPIU_DBG_MSG(CH3_OTHER,TERSE,"unable to obtain the 'process-mapping' PMI key");
+    }
+
+    mpi_errno = publish_node_id(pg, our_pg_rank);
     if (mpi_errno) MPIU_ERR_POP(mpi_errno);
 

mpich2/branches/release/mpich2-1.1/src/pm/mpd/Makefile.in

@@ -86 +86 @@
         @-rm -f *.o *.dep *~ ${EXAMPLES} core* *.1 *.pyc
         @-rm -f *.gcno *.gcda *.bb *.bbg
-        @-rm -f ${srcdir}/*.gcno ${scrdir}/*.gcda
+        @-rm -f ${srcdir}/*.gcno ${srcdir}/*.gcda
 
 # echo "some of the following might be executables you want to remove"

mpich2/branches/release/mpich2-1.1/src/pm/mpd/mpd.py

@@ -631 +631 @@
             msg['ringsize'] = 0
             msg['ring_ncpus'] = 0
+            # maps rank => hostname
+            msg['process_mapping'] = {}
             if msg.has_key('try_1st_locally'):
                 self.do_mpdrun(msg)
@@ -798 +800 @@
             msg['totalview'] = 0
             msg['ifhns'] = {}
+            # maps rank => hostname
+            msg['process_mapping'] = {}
             self.spawnQ.append(msg)
         elif msg['cmd'] == 'publish_name':
@@ -827 +831 @@
             sock.send_dict_msg(msgToSend)
 
+    def calculate_process_mapping(self,mapping_dict):
+        # mapping_dict maps ranks => hostnames
+        ranks = list(mapping_dict.keys())
+        ranks.sort()
+
+        # assign node ids based in first-come-first-serve order when iterating
+        # over the ranks in increasing order
+        next_id = 0
+        node_ids = {}
+        for rank in ranks:
+            host = mapping_dict[rank]
+            if not node_ids.has_key(host):
+                node_ids[host] = next_id
+                next_id += 1
+
+
+        # maps {node_id_A: set([rankX,rankY,...]), node_id_B:...}
+        node_to_ranks = {}
+        for rank in ranks:
+            node_id = node_ids[mapping_dict[rank]]
+            if not node_to_ranks.has_key(node_id):
+                node_to_ranks[node_id] = set([])
+            node_to_ranks[node_id].add(rank)
+
+        # we only handle two cases for now:
+        # 1. block regular
+        # 2. round-robin regular
+        # we do handle a "remainder node" that might not be full
+        delta = -1
+        max_ranks_per_node = 0
+        for node_id in node_to_ranks.keys():
+            last_rank = -1
+            if len(node_to_ranks[node_id]) > max_ranks_per_node:
+                max_ranks_per_node = len(node_to_ranks[node_id])
+            ranks = list(node_to_ranks[node_id])
+            ranks.sort()
+            for rank in ranks:
+                if last_rank != -1:
+                    if delta == -1:
+                        if node_id == 0:
+                            delta = rank - last_rank
+                        else:
+                            # irregular case detected such as {0:A,1:B,2:B}
+                            mpd_print(1, "irregular case A detected")
+                            return ''
+                    elif (rank - last_rank) != delta:
+                        # irregular such as {0:A,1:B,2:A,3:A,4:B}
+                        mpd_print(1, "irregular case B detected")
+                        return ''
+                last_rank = rank
+
+        num_nodes = len(node_to_ranks.keys())
+        if delta == 1:
+            return '(vector,(%d,%d,%d))' % (0,num_nodes,max_ranks_per_node)
+        else:
+            # either we are round-robin-regular (delta > 1) or there is only one
+            # process per node (delta == -1), either way results in the same
+            # mapping spec
+            return '(vector,(%d,%d,%d))' % (0,num_nodes,1)
+
     def handle_lhs_input(self,sock):
         msg = self.ring.lhsSock.recv_dict_msg()
@@ -842 +906 @@
                     self.currRingNCPUs = msg['ring_ncpus']
                 if msg['nstarted'] == msg['nprocs']:
+                    # we have started all processes in the job, tell the
+                    # requester this and stop forwarding the mpdrun/spawn
+                    # message around the loop
                     if msg['cmd'] == 'spawn':
                         self.spawnInProgress = 0
@@ -849 +916 @@
                                       'ring_ncpus' : self.currRingNCPUs}
                         self.conSock.send_dict_msg(msgToSend)
+                    # Tell all MPDs in the ring the final process mapping.  In
+                    # turn, they will inform all of their child mpdmans.
+                    # Only do this in the case of a regular mpdrun.  The spawn
+                    # case it too complicated to handle this way right now.
+                    if msg['cmd'] == 'mpdrun':
+                        process_mapping_str = self.calculate_process_mapping(msg['process_mapping'])
+                        msgToSend = { 'cmd' : 'process_mapping',
+                                      'jobid' : msg['jobid'],
+                                      'mpdid_mpdrun_start' : self.myId,
+                                      'process_mapping' : process_mapping_str }
+                        self.ring.rhsSock.send_dict_msg(msgToSend)
                     return
                 if not msg['first_loop']  and  msg['nstarted_on_this_loop'] == 0:
@@ -873 +951 @@
                 msg['nstarted_on_this_loop'] = 0
             self.do_mpdrun(msg)
+        elif msg['cmd'] == 'process_mapping':
+            # message transmission terminates once the message has made it all
+            # the way around the loop once
+            if msg['mpdid_mpdrun_start'] != self.myId:
+                self.ring.rhsSock.send_dict_msg(msg) # forward it on around
+
+            # send to all mpdman's for the jobid embedded in the msg
+            jobid = msg['jobid']
+
+            # there may be no entry for jobid in the activeJobs table if there
+            # weren't any processes from that job actually launched on our host
+            if self.activeJobs.has_key(jobid):
+                for manPid in self.activeJobs[jobid].keys():
+                    manSock = self.activeJobs[jobid][manPid]['socktoman']
+                    manSock.send_dict_msg(msg)
         elif msg['cmd'] == 'mpdtrace_info':
             if msg['dest'] == self.myId:
@@ -1175 +1268 @@
             except:
                 pass
+
         if msg.has_key('jobid'):
             jobid = msg['jobid']
@@ -1193 +1287 @@
                     for rank in range(lorank,hirank+1):
                         self.run_one_cli(rank,msg)
+                        # we use myHost under the assumption that there is only
+                        # one mpd per user on a given host.  The ifhn only
+                        # affects how the MPDs communicate with each other, not
+                        # which host they are on
+                        msg['process_mapping'][rank] = self.myHost
                         msg['nstarted'] += 1
                         msg['nstarted_on_this_loop'] += 1
@@ -1205 +1304 @@
                     if self.myIfhn in hostSpecPool  or  self.myHost in hostSpecPool:
                         self.run_one_cli(lorank,msg)
+                        msg['process_mapping'][lorank] = self.myHost
                         msg['nstarted'] += 1
                         msg['nstarted_on_this_loop'] += 1
@@ -1220 +1320 @@
                         (lorank,hirank) = ranks
                         self.run_one_cli(lorank,msg)
+                        msg['process_mapping'][lorank] = self.myHost
                         msg['nstarted'] += 1
                         msg['nstarted_on_this_loop'] += 1
@@ -1225 +1326 @@
                         if lorank < hirank:
                             msg['hosts'][(lorank+1,hirank)] = '_any_'
+                        # self.activeJobs maps:
+                        # { jobid => { mpdman_pid => {...} } }
                         procsHereForJob = len(self.activeJobs[jobid].keys())
                         if procsHereForJob >= self.parmdb['MPD_NCPUS']:

mpich2/branches/release/mpich2-1.1/src/pm/mpd/mpdlib.py

@@ -480 +480 @@
                     try:
                         mpd_signum = 0
-                        (readyToRecv,unused1,unused2) = select.select([self.sock],[],[],timeout)
+                        if timeout == -1:
+                            # use -1 to indicate indefinite timeout
+                            (readyToRecv,unused1,unused2) = select.select([self.sock],[],[])
+                        else:
+                            (readyToRecv,unused1,unused2) = select.select([self.sock],[],[],timeout)
                         break;
                     except os.error, errinfo:

mpich2/branches/release/mpich2-1.1/src/pm/mpd/mpdman.py

@@ -374 +374 @@
                       'spawner_mpd' : os.environ['MPDMAN_SPAWNER_MPD'] }
         self.mpdSock.send_dict_msg(msgToSend)
+
         if not self.subproc:
             self.streamHandler.set_handler(self.fd_read_cli_stdout,
@@ -424 +425 @@
             # rshipSock.close()
             self.waitPids.append(rshipPid)
+
+        if not self.spawned:
+            # receive the final process mapping from our MPD overlords
+            msg = self.mpdSock.recv_dict_msg(timeout=-1)
+
+            # a few defensive checks now to make sure that the various parts of the
+            # code are all on the same page
+            if not msg.has_key('cmd') or msg['cmd'] != 'process_mapping':
+                mpd_print(1,'expected cmd="process_mapping", got cmd="%s" instead' % (msg.get('cmd','**not_present**')))
+                sys.exit(-1)
+            if msg['jobid'] != self.jobid:
+                mpd_print(1,'expected jobid="%s", got jobid="%s" instead' % (self.jobid,msg['jobid']))
+                sys.exit(-1)
+            if not msg.has_key('process_mapping'):
+                mpd_print(1,'expected msg to contain a process_mapping key')
+                sys.exit(-1)
+            self.KVSs[self.default_kvsname]['process-mapping'] = msg['process_mapping']
+
 
         self.tvReady = 0