Add a channel for reporting errors, fix a bug in the tcp module

This commit was SVN r23610.

orte/mca/rmcast/rmcast_types.h

@@ -33,8 +33,9 @@ typedef uint32_t orte_rmcast_channel_t;
 #define ORTE_RMCAST_SYS_CHANNEL             4
 #define ORTE_RMCAST_APP_PUBLIC_CHANNEL      5
 #define ORTE_RMCAST_DATA_SERVER_CHANNEL     6
+#define ORTE_RMCAST_ERROR_CHANNEL           7
 
-#define ORTE_RMCAST_DYNAMIC_CHANNELS        7
+#define ORTE_RMCAST_DYNAMIC_CHANNELS        8
 
 
 /* define channel directions */

orte/mca/rmcast/tcp/rmcast_tcp.c

@@ -180,7 +180,13 @@ static int init(void)
             ORTE_ERROR_LOG(rc);
             return rc;
         }
-        /* activate a recv to catch relays */
+        /* open the error reporting channel */
+         if (ORTE_SUCCESS != (rc = open_channel(ORTE_RMCAST_ERROR_CHANNEL, "error",
+                                               NULL, -1, NULL, ORTE_RMCAST_BIDIR))) {
+            ORTE_ERROR_LOG(rc);
+            return rc;
+        }
+       /* activate a recv to catch relays */
         if (ORTE_SUCCESS != (rc = orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD,
                                                           ORTE_RML_TAG_MULTICAST_RELAY,
                                                           ORTE_RML_NON_PERSISTENT,
@@ -201,7 +207,13 @@ static int init(void)
             ORTE_ERROR_LOG(rc);
             return rc;
         }
-        /* finally, if we are an app, setup our grp xmit/recv channels, if given */
+         /* open the error reporting channel */
+         if (ORTE_SUCCESS != (rc = open_channel(ORTE_RMCAST_ERROR_CHANNEL, "error",
+                                               NULL, -1, NULL, ORTE_RMCAST_BIDIR))) {
+            ORTE_ERROR_LOG(rc);
+            return rc;
+        }
+       /* finally, if we are an app, setup our grp xmit/recv channels, if given */
         if (ORTE_PROC_IS_APP && NULL != orte_rmcast_base.my_group_name) {
             if (ORTE_SUCCESS != (rc = open_channel(orte_rmcast_base.my_group_number,
                                                    "recv", NULL, -1, NULL, ORTE_RMCAST_RECV))) {
@@ -332,7 +344,7 @@ static int queue_xmit(rmcast_base_send_t *snd,
         return ORTE_ERR_NOT_FOUND;
     }
     
-process:    
+ process:    
     /* setup the message for xmission */
     if (ORTE_SUCCESS != (rc = orte_rmcast_base_build_msg(ch, &buf, snd))) {
         ORTE_ERROR_LOG(rc);
@@ -381,6 +393,27 @@ process:
         }
         rc = ORTE_SUCCESS;
     } else {
+        /* if I am a daemon, I need to relay this to my children first */
+        if (ORTE_PROC_IS_DAEMON) {
+            for (item = opal_list_get_first(&orte_local_children);
+                 item != opal_list_get_end(&orte_local_children);
+                 item = opal_list_get_next(item)) {
+                child = (orte_odls_child_t*)item;
+                if (NULL == child->rml_uri) {
+                    /* race condition */
+                    continue;
+                }
+                OPAL_OUTPUT_VERBOSE((2, orte_rmcast_base.rmcast_output,
+                                     "%s relaying multicast to %s",
+                                     ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
+                                     ORTE_NAME_PRINT(child->name)));
+                if (0 > (rc = orte_rml.send_buffer(child->name, buf, ORTE_RML_TAG_MULTICAST, 0))) {
+                    ORTE_ERROR_LOG(rc);
+                    goto cleanup;
+                }
+            }
+        }
+
         /* send it to the HNP */
         OPAL_OUTPUT_VERBOSE((2, orte_rmcast_base.rmcast_output,
                              "%s rmcast:tcp sending multicast to HNP %s",
@@ -415,7 +448,7 @@ process:
     /* roll to next message sequence number */
     ORTE_MULTICAST_NEXT_SEQUENCE_NUM(ch->seq_num);
     
-cleanup:
+ cleanup:
     OBJ_RELEASE(buf);
 
     OPAL_THREAD_UNLOCK(&ch->send_lock);

orte/mca/rmcast/udp/rmcast_udp.c

@@ -177,6 +177,12 @@ static int init(void)
             ORTE_ERROR_LOG(rc);
             return rc;
         }
+        /* open the error reporting channel */
+         if (ORTE_SUCCESS != (rc = open_channel(ORTE_RMCAST_ERROR_CHANNEL, "error",
+                                               NULL, -1, NULL, ORTE_RMCAST_BIDIR))) {
+            ORTE_ERROR_LOG(rc);
+            return rc;
+        }
     } else if (ORTE_PROC_IS_APP) {
         /* apps open the app public and data server channels */
         if (ORTE_SUCCESS != (rc = open_channel(ORTE_RMCAST_APP_PUBLIC_CHANNEL, "app-announce",
@@ -188,6 +194,12 @@ static int init(void)
                                                NULL, -1, NULL, ORTE_RMCAST_BIDIR))) {
             ORTE_ERROR_LOG(rc);
             return rc;
+        }
+        /* open the error reporting channel */
+         if (ORTE_SUCCESS != (rc = open_channel(ORTE_RMCAST_ERROR_CHANNEL, "error",
+                                               NULL, -1, NULL, ORTE_RMCAST_BIDIR))) {
+            ORTE_ERROR_LOG(rc);
+            return rc;
         }
          /* finally, if we are an app, setup our grp xmit/recv channels, if given */
         if (ORTE_PROC_IS_APP && NULL != orte_rmcast_base.my_group_name) {