Fix IOF of stdin

Ensure we properly terminate the stream when the input
is complete

Signed-off-by: Ralph Castain <rhc@pmix.org>

Author: rhc54

src/common/pmix_iof.c

@@ -664,7 +664,8 @@ pmix_status_t PMIx_IOF_push(const pmix_proc_t targets[], size_t ntargets, pmix_b
 
     /* if we are not a server, then we send the provided
      * data to our server for processing */
-    if (!PMIX_PEER_IS_SERVER(pmix_globals.mypeer) || PMIX_PEER_IS_LAUNCHER(pmix_globals.mypeer)) {
+    if (!PMIX_PEER_IS_SERVER(pmix_globals.mypeer) ||
+        PMIX_PEER_IS_LAUNCHER(pmix_globals.mypeer)) {
         msg = PMIX_NEW(pmix_buffer_t);
         if (NULL == msg) {
             return PMIX_ERR_NOMEM;
@@ -1803,21 +1804,31 @@ void pmix_iof_read_local_handler(int sd, short args, void *cbdata)
         goto reactivate;
     }
 
+    /* if I am a launcher and connected to a server, then
+     * we want to send things to our server for relay */
+    if (PMIX_PEER_IS_LAUNCHER(pmix_globals.mypeer) &&
+        pmix_globals.connected) {
+        goto forward;
+    }
+
     /* if I am a server, then push this up to my host */
-    if (PMIX_PROC_IS_SERVER(&pmix_globals.mypeer->proc_type)) {
+    if (PMIX_PEER_IS_SERVER(pmix_globals.mypeer)) {
         if (NULL == pmix_host_server.push_stdin) {
             /* nothing we can do with this info - no point in reactivating it */
             return;
         }
         PMIX_BYTE_OBJECT_CREATE(boptr, 1);
-        boptr->bytes = (char*)malloc(bo.size);
-        memcpy(boptr->bytes, bo.bytes, bo.size);
-        boptr->size = bo.size;
+        if (0 < bo.size) {
+            boptr->bytes = (char*)malloc(bo.size);
+            memcpy(boptr->bytes, bo.bytes, bo.size);
+            boptr->size = bo.size;
+        }
         rc = pmix_host_server.push_stdin(&pmix_globals.myid, rev->targets, rev->ntargets,
                                          rev->directives, rev->ndirs, boptr, opcbfn, (void*)boptr);
         goto reactivate;
     }
 
+forward:
     /* pass the data to our PMIx server so it can relay it
      * to the host RM for distribution */
     msg = PMIX_NEW(pmix_buffer_t);
@@ -1911,11 +1922,13 @@ PMIX_CLASS_INSTANCE(pmix_iof_sink_t, pmix_list_item_t, iof_sink_construct, iof_s
 
 static void iof_read_event_construct(pmix_iof_read_event_t *rev)
 {
+    rev->tv.tv_sec = 0;
+    rev->tv.tv_usec = 0;
     rev->fd = -1;
+    rev->channel = PMIX_FWD_NO_CHANNELS;
     rev->active = false;
     rev->childproc = NULL;
-    rev->tv.tv_sec = 0;
-    rev->tv.tv_usec = 0;
+    rev->always_readable = false;
     rev->targets = NULL;
     rev->ntargets = 0;
     rev->directives = NULL;

src/mca/bfrops/base/bfrop_base_pack.c

@@ -43,7 +43,7 @@ pmix_status_t pmix_bfrops_base_pack(pmix_pointer_array_t *regtypes, pmix_buffer_
     pmix_status_t rc;
 
     /* check for error */
-    if (NULL == buffer || NULL == src) {
+    if (NULL == buffer || (NULL == src && 0 < num_vals)) {
         PMIX_ERROR_LOG(PMIX_ERR_BAD_PARAM);
         return PMIX_ERR_BAD_PARAM;
     }

src/server/pmix_server.c

@@ -159,7 +159,7 @@ static void server_iof_handler(struct pmix_peer_t *pr, pmix_ptl_hdr_t *hdr,
 
     PMIX_HIDE_UNUSED_PARAMS(hdr, cbdata);
 
-    pmix_output_verbose(2, pmix_client_globals.iof_output,
+    pmix_output_verbose(2, pmix_server_globals.iof_output,
                         "recvd IOF with %d bytes from %s",
                         (int) buf->bytes_used,
                         PMIX_PNAME_PRINT(&peer->info->pname));

src/server/pmix_server_ops.c

@@ -3621,15 +3621,19 @@ static void stdcbfunc(pmix_status_t status, void *cbdata)
     PMIX_RELEASE(cd);
 }
 
-pmix_status_t pmix_server_iofstdin(pmix_peer_t *peer, pmix_buffer_t *buf, pmix_op_cbfunc_t cbfunc,
+pmix_status_t pmix_server_iofstdin(pmix_peer_t *peer,
+                                   pmix_buffer_t *buf,
+                                   pmix_op_cbfunc_t cbfunc,
                                    void *cbdata)
 {
     int32_t cnt;
     pmix_status_t rc;
     pmix_proc_t source;
     pmix_setup_caddy_t *cd;
 
-    pmix_output_verbose(2, pmix_server_globals.iof_output, "recvd stdin IOF data from tool");
+    pmix_output_verbose(2, pmix_server_globals.iof_output,
+                        "recvd stdin IOF data from tool %s",
+                        PMIX_PEER_PRINT(peer));
 
     if (NULL == pmix_host_server.push_stdin) {
         return PMIX_ERR_NOT_SUPPORTED;