Avoid conflicts when multiple collectives are underway in ORTE by giving each grpcomm component its own RML tag and posting persistent receives. We use the signature anyway to determine which collective the received message is addressing, so there is no need to post non-persistent receives.

This commit was SVN r32703.
2014-09-10 17:36:16 +00:00 · 2014-09-10 17:36:16 +00:00 · 4eb6291334
--- a/orte/mca/grpcomm/brks/grpcomm_brks.c
+++ b/orte/mca/grpcomm/brks/grpcomm_brks.c
@ -58,6 +58,11 @@ orte_grpcomm_base_module_t orte_grpcomm_brks_module = {
 */
 static int init(void)
 {
+    /* setup recv for distance data */
+    orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD,
+                            ORTE_RML_TAG_ALLGATHER_BRKS,
+                            ORTE_RML_PERSISTENT,
+                            brks_allgather_recv_dist, NULL);
    return OPAL_SUCCESS;
 }

@ -66,6 +71,9 @@ static int init(void)
 */
 static void finalize(void)
 {
+    /* cancel the recv */
+    orte_rml.recv_cancel(ORTE_NAME_WILDCARD, ORTE_RML_TAG_ALLGATHER_BRKS);
+
    return;
 }

@ -175,7 +183,7 @@ static int brks_allgather_send_dist(orte_grpcomm_coll_t *coll, orte_vpid_t dista
                             ORTE_NAME_PRINT(&peer_send)));

    if (0 > (rc = orte_rml.send_buffer_nb(&peer_send, send_buf,
-                                          -ORTE_RML_TAG_ALLGATHER,
+                                          ORTE_RML_TAG_ALLGATHER_BRKS,
                                          orte_rml_send_callback, NULL))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(send_buf);
@ -187,12 +195,6 @@ static int brks_allgather_send_dist(orte_grpcomm_coll_t *coll, orte_vpid_t dista
                             ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
                             ORTE_NAME_PRINT(&peer_recv)));

-    /* setup recv for distance data */
-    orte_rml.recv_buffer_nb(&peer_recv,
-                            -ORTE_RML_TAG_ALLGATHER,
-                            ORTE_RML_NON_PERSISTENT,
-                            brks_allgather_recv_dist, NULL);
-
    return ORTE_SUCCESS;
 }

--- a/orte/mca/grpcomm/direct/grpcomm_direct.c
+++ b/orte/mca/grpcomm/direct/grpcomm_direct.c
@ -78,7 +78,7 @@ static int init(void)
                            ORTE_RML_PERSISTENT,
                            xcast_recv, NULL);
    orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD,
-                            ORTE_RML_TAG_ALLGATHER,
+                            ORTE_RML_TAG_ALLGATHER_DIRECT,
                            ORTE_RML_PERSISTENT,
                            allgather_recv, NULL);
    /* setup recv for barrier release */
@ -190,7 +190,7 @@ static int allgather(orte_grpcomm_coll_t *coll,

    /* send the info to the HNP for tracking */
    rc = orte_rml.send_buffer_nb(ORTE_PROC_MY_HNP, relay,
-                                 ORTE_RML_TAG_ALLGATHER,
+                                 ORTE_RML_TAG_ALLGATHER_DIRECT,
                                 orte_rml_send_callback, NULL);
    return rc;
 }
--- a/orte/mca/grpcomm/rcd/grpcomm_rcd.c
+++ b/orte/mca/grpcomm/rcd/grpcomm_rcd.c
@ -57,6 +57,11 @@ orte_grpcomm_base_module_t orte_grpcomm_rcd_module = {
 */
 static int init(void)
 {
+    /* setup recv for distance data */
+    orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD,
+                            ORTE_RML_TAG_ALLGATHER_RCD,
+                            ORTE_RML_PERSISTENT,
+                            rcd_allgather_recv_dist, NULL);
    return OPAL_SUCCESS;
 }

@ -65,6 +70,8 @@ static int init(void)
 */
 static void finalize(void)
 {
+    /* cancel the recv */
+    orte_rml.recv_cancel(ORTE_NAME_WILDCARD, ORTE_RML_TAG_ALLGATHER_RCD);
    return;
 }

@ -164,7 +171,7 @@ static int rcd_allgather_send_dist(orte_grpcomm_coll_t *coll, orte_vpid_t distan


    if (0 > (rc = orte_rml.send_buffer_nb(&peer, send_buf,
-                                          -ORTE_RML_TAG_ALLGATHER,
+                                          ORTE_RML_TAG_ALLGATHER_RCD,
                                          orte_rml_send_callback, NULL))) {
        ORTE_ERROR_LOG(rc);
        OBJ_RELEASE(send_buf);
@ -176,12 +183,6 @@ static int rcd_allgather_send_dist(orte_grpcomm_coll_t *coll, orte_vpid_t distan
                             ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
                             ORTE_NAME_PRINT(&peer)));

-    /* setup recv for distance data */
-    orte_rml.recv_buffer_nb(&peer,
-                            -ORTE_RML_TAG_ALLGATHER,
-                            ORTE_RML_NON_PERSISTENT,
-                            rcd_allgather_recv_dist, NULL);
-
    return ORTE_SUCCESS;
 }

--- a/orte/mca/rml/rml_types.h
+++ b/orte/mca/rml/rml_types.h
@ -102,55 +102,54 @@ BEGIN_C_DECLS
 #define ORTE_RML_TAG_COLLECTIVE             30
 #define ORTE_RML_TAG_COLL_RELEASE           31
 #define ORTE_RML_TAG_DAEMON_COLL            32
-#define ORTE_RML_TAG_ALLGATHER              33
+#define ORTE_RML_TAG_ALLGATHER_DIRECT       33
+#define ORTE_RML_TAG_ALLGATHER_BRKS         34
+#define ORTE_RML_TAG_ALLGATHER_RCD          35

 /* show help */
-#define ORTE_RML_TAG_SHOW_HELP              34
+#define ORTE_RML_TAG_SHOW_HELP              36

 /* debugger release */
-#define ORTE_RML_TAG_DEBUGGER_RELEASE       35
+#define ORTE_RML_TAG_DEBUGGER_RELEASE       37

 /* bootstrap */
-#define ORTE_RML_TAG_BOOTSTRAP              36
+#define ORTE_RML_TAG_BOOTSTRAP              38

 /* report a missed msg */
-#define ORTE_RML_TAG_MISSED_MSG             37
+#define ORTE_RML_TAG_MISSED_MSG             39

 /* tag for receiving ack of abort msg */
-#define ORTE_RML_TAG_ABORT                  38
+#define ORTE_RML_TAG_ABORT                  40

 /* tag for receiving heartbeats */
-#define ORTE_RML_TAG_HEARTBEAT              39
+#define ORTE_RML_TAG_HEARTBEAT              41

 /* Process Migration Tool Tag */
-#define ORTE_RML_TAG_MIGRATE                40
+#define ORTE_RML_TAG_MIGRATE                42

 /* For SStore Framework */
-#define ORTE_RML_TAG_SSTORE                 41
-#define ORTE_RML_TAG_SSTORE_INTERNAL        42
+#define ORTE_RML_TAG_SSTORE                 43
+#define ORTE_RML_TAG_SSTORE_INTERNAL        44

-#define ORTE_RML_TAG_SUBSCRIBE              43
+#define ORTE_RML_TAG_SUBSCRIBE              45


 /* Notify of failed processes */
-#define ORTE_RML_TAG_FAILURE_NOTICE         44
+#define ORTE_RML_TAG_FAILURE_NOTICE         46

 /* distributed file system */
-#define ORTE_RML_TAG_DFS_CMD                45
-#define ORTE_RML_TAG_DFS_DATA               46
+#define ORTE_RML_TAG_DFS_CMD                47
+#define ORTE_RML_TAG_DFS_DATA               48

 /* sensor data */
-#define ORTE_RML_TAG_SENSOR_DATA            47
+#define ORTE_RML_TAG_SENSOR_DATA            49

 /* direct modex support */
-#define ORTE_RML_TAG_DIRECT_MODEX           48
-#define ORTE_RML_TAG_DIRECT_MODEX_RESP      49
+#define ORTE_RML_TAG_DIRECT_MODEX           50
+#define ORTE_RML_TAG_DIRECT_MODEX_RESP      51

 /* notifier support */
-#define ORTE_RML_TAG_NOTIFIER_HNP           50
-
-/* global collective ID request */
-#define ORTE_RML_TAG_FULL_COLL_ID           51
+#define ORTE_RML_TAG_NOTIFIER_HNP           52

 #define ORTE_RML_TAG_MAX                   100