iof: do not release a sink before all read data is written.

When too much data is available on stdin, it might not be
forwarded immediatly to the task (write() might fail with -EAGAIN),
so when stdin is terminated, there might be some remaining data
to be pushed to the task. In this case, delay the release of the sink
so no data is discarded.

Refs open-mpi/ompi#4744

Signed-off-by: Gilles Gouaillardet <gilles@rist.or.jp>

orte/mca/iof/base/base.h

@@ -105,6 +105,7 @@ typedef struct {
     orte_iof_write_event_t *wev;
     bool xoff;
     bool exclusive;
+    bool closed;
 } orte_iof_sink_t;
 ORTE_DECLSPEC OBJ_CLASS_DECLARATION(orte_iof_sink_t);
 

orte/mca/iof/base/iof_base_frame.c

@@ -241,6 +241,7 @@ static void orte_iof_base_sink_construct(orte_iof_sink_t* ptr)
     ptr->wev = OBJ_NEW(orte_iof_write_event_t);
     ptr->xoff = false;
     ptr->exclusive = false;
+    ptr->closed = false;
 }
 static void orte_iof_base_sink_destruct(orte_iof_sink_t* ptr)
 {

orte/mca/iof/hnp/iof_hnp.c

@@ -13,7 +13,7 @@
  * Copyright (c) 2007      Cisco Systems, Inc.  All rights reserved.
  * Copyright (c) 2011-2013 Los Alamos National Security, LLC.  All rights
  *                         reserved.
- * Copyright (c) 2014-2017 Research Organization for Information Science
+ * Copyright (c) 2014-2018 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2016-2017 Intel, Inc. All rights reserved.
  * Copyright (c) 2017      Mellanox Technologies. All rights reserved.
@@ -629,6 +629,10 @@ static void stdin_write_handler(int fd, short event, void *cbdata)
             ORTE_IOF_READ_ACTIVATE(mca_iof_hnp_component.stdinev);
         }
     }
+    if (sink->closed && 0 == opal_list_get_size(&wev->outputs)) {
+        /* the sink has already been closed and everything was written, time to release it */
+        OBJ_RELEASE(sink);
+    }
     return;
   finish:
     OBJ_RELEASE(wev);

orte/mca/iof/hnp/iof_hnp_read.c

@@ -14,6 +14,8 @@
  *                         reserved.
  * Copyright (c) 2014-2017 Intel, Inc.  All rights reserved.
  * Copyright (c) 2017      Mellanox Technologies. All rights reserved.
+ * Copyright (c) 2018      Research Organization for Information Science
+ *                         and Technology (RIST).  All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -202,8 +204,13 @@ void orte_iof_hnp_read_local_handler(int fd, short event, void *cbdata)
 
         /* if num_bytes was zero, or we read the last piece of the file, then we need to terminate the event */
         if (0 == numbytes) {
-            /* this will also close our stdin file descriptor */
-            OBJ_RELEASE(proct->stdinev);
+            if (0 != opal_list_get_size(&proct->stdinev->wev->outputs)) {
+                /* some stuff has yet to be written, so delay the release of proct->stdinev */
+                proct->stdinev->closed = true;
+            } else {
+                /* this will also close our stdin file descriptor */
+                OBJ_RELEASE(proct->stdinev);
+            }
         } else {
             /* if we are looking at a tty, then we just go ahead and restart the
              * read event assuming we are not backgrounded