starting integration of i/o forwarding framework

This commit was SVN r3986.

src/Makefile.am

@@ -109,6 +109,7 @@ libmpi_la_LIBADD = \
         $(MCA_common_STATIC_LTLIBS) \
         mca/gpr/base/libmca_gpr_base.la $(MCA_gpr_STATIC_LTLIBS) \
         mca/io/base/libmca_io_base.la $(MCA_io_STATIC_LTLIBS) \
+        mca/iof/base/libmca_iof_base.la $(MCA_iof_STATIC_LTLIBS) \
         mca/llm/base/libmca_llm_base.la $(MCA_llm_STATIC_LTLIBS) \
         mca/mpool/base/libmca_mpool_base.la $(MCA_mpool_STATIC_LTLIBS) \
         mca/ns/base/libmca_ns_base.la $(MCA_ns_STATIC_LTLIBS) \

src/mca/ns/ns.h

@@ -45,6 +45,7 @@
  * useful defines for bit-masks
  */
 
+#define OMPI_NS_CMP_NONE       0x00
 #define OMPI_NS_CMP_CELLID     0x01
 #define OMPI_NS_CMP_JOBID      0x02
 #define OMPI_NS_CMP_VPID       0x04

src/mca/oob/base/base.h

@@ -67,11 +67,12 @@ OMPI_DECLSPEC extern ompi_process_name_t mca_oob_name_self;
 #define MCA_OOB_TAG_RTE         4
 #define MCA_OOB_TAG_EXEC        5
 #define MCA_OOB_TAG_DAEMON      6
-#define MCA_OOB_TAG_STDIO       7
+#define MCA_OOB_TAG_IOF_SVC     7
-#define MCA_OOB_TAG_SCHED       8
+#define MCA_OOB_TAG_IOF_CLT     8
-#define MCA_OOB_TAG_PCM_KILL    9
+#define MCA_OOB_TAG_SCHED       9
 #define MCA_OOB_TAG_XCAST       10
-#define MCA_OOB_TAG_PCM_KILL_ACK 11
+#define MCA_OOB_TAG_PCM_KILL    11
+#define MCA_OOB_TAG_PCM_KILL_ACK 12
 #define MCA_OOB_TAG_USER        1000 /* user defined tags should be assigned above this level */
 
 /*

src/mca/pml/base/base.h

@@ -29,6 +29,7 @@
 extern "C" {
 #endif
 OMPI_DECLSPEC  int mca_pml_base_open(void);
+OMPI_DECLSPEC  int mca_pml_base_progress(void);
 OMPI_DECLSPEC  int mca_pml_base_select(mca_pml_base_module_t *selected, 
                           bool *allow_multi_user_threads, 
                           bool *have_hidden_threads);

src/mca/pml/base/pml_base_close.c

@@ -29,6 +29,7 @@ int mca_pml_base_close(void)
      anyway?  This module is going away, so errors don't matter
      anymore) */
 
+  mca_pml.pml_progress = mca_pml_base_progress;
   if (NULL != mca_pml_base_selected_component.pmlm_finalize) {
     mca_pml_base_selected_component.pmlm_finalize();
   }

src/mca/pml/base/pml_base_open.c

@@ -30,7 +30,7 @@
 
 #include "mca/pml/base/static-components.h"
 
-static int mca_pml_base_progress(void) 
+int mca_pml_base_progress(void) 
 {
     return OMPI_SUCCESS;
 }

src/mpi/runtime/Makefile.am

@@ -32,6 +32,7 @@ libmpiruntime_la_SOURCES = \
 	$(headers) \
 	ompi_mpi_abort.c \
     ompi_mpi_init.c \
+    ompi_mpi_io.c \
     ompi_mpi_finalize.c \
     ompi_mpi_params.c
 

src/mpi/runtime/mpiruntime.h

@@ -68,6 +68,12 @@ extern "C" {
    */
   int ompi_mpi_init(int argc, char **argv, int requested, int *provided);
 
+  /**
+   * Setup I/O forwarding during MPI_Init through the I/O forwarding
+   * framework.
+   */
+  int ompi_mpi_init_io(void);
+
   /**
    * Finalize the Open MPI MPI environment
    *

src/mpi/runtime/ompi_mpi_init.c

@@ -349,13 +349,21 @@ int ompi_mpi_init(int argc, char **argv, int requested, int *provided)
         goto error;
     }
 
-#if OMPI_HAVE_THREADS == 0
+#if OMPI_HAVE_THREADS
-    ompi_progress_events(OMPI_EVLOOP_NONBLOCK);
+    /* setup I/O forwarding */
+    if (OMPI_SUCCESS != (ret = ompi_mpi_init_io())) {
+	error = "ompi_rte_init_io failed";
+	goto error;
+    }
 #endif
 
     /* Wait for everyone to initialize */
     mca_oob_barrier();
 
+#if OMPI_HAVE_THREADS == 0
+    ompi_progress_events(OMPI_EVLOOP_NONBLOCK);
+#endif
+
     /* new very last step: check whether we have been spawned or not.
        We introduce that at the very end, since we need collectives,
        datatypes, ptls etc. up and running here....

src/mpi/runtime/ompi_mpi_io.c

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2004-2005 The Trustees of Indiana University.
+ *                         All rights reserved.
+ * Copyright (c) 2004-2005 The Trustees of the University of Tennessee.
+ *                         All rights reserved.
+ * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart, 
+ *                         University of Stuttgart.  All rights reserved.
+ * $COPYRIGHT$
+ * 
+ * Additional copyrights may follow
+ * 
+ * $HEADER$
+ */
+
+#include "ompi_config.h"
+#include "mca/oob/oob.h"
+#include "mca/iof/iof.h"
+
+
+int ompi_mpi_init_io(void)
+{
+    int fds[2];
+    int rc;
+
+    /* setup stdin */
+    rc = pipe(fds);
+    if(rc < 0) {
+        return OMPI_ERR_OUT_OF_RESOURCE;
+    }
+    dup2(fds[0], 0);
+    close(fds[0]);
+
+    rc = mca_iof.iof_publish(
+        MCA_OOB_NAME_SELF,
+        MCA_IOF_SINK,
+        MCA_IOF_STDIN,
+        fds[1]);
+    if(rc != OMPI_SUCCESS)
+        return rc;
+ 
+    /* setup stdout */
+    rc = pipe(fds);
+    if(rc < 0) {
+        return OMPI_ERR_OUT_OF_RESOURCE;
+    }
+    dup2(fds[1], 1);
+    close(fds[1]);
+
+    rc = mca_iof.iof_publish(
+        MCA_OOB_NAME_SELF,
+        MCA_IOF_SOURCE,
+        MCA_IOF_STDOUT,
+        fds[0]);
+    if(rc != OMPI_SUCCESS)
+        return rc;
+ 
+    /* setup stderr */
+    rc = pipe(fds);
+    if(rc < 0) {
+        return OMPI_ERR_OUT_OF_RESOURCE;
+    }
+    dup2(fds[1], 2);
+    close(fds[1]);
+
+    rc = mca_iof.iof_publish(
+        MCA_OOB_NAME_SELF,
+        MCA_IOF_SOURCE,
+        MCA_IOF_STDERR,
+        fds[0]);
+    if(rc != OMPI_SUCCESS)
+        return rc;
+    return OMPI_SUCCESS;
+}
+
+

src/runtime/ompi_progress.c

@@ -57,7 +57,7 @@ void ompi_progress(void)
         }
     }
 
-#if 0
+#if 1
     /* TSW - disable this until can validate that it doesn't impact SMP
      * performance
     */

src/runtime/ompi_rte_finalize.c

@@ -28,6 +28,7 @@
 #include "mca/pcm/base/base.h"
 #include "mca/pcmclient/base/base.h"
 #include "mca/oob/oob.h"
+#include "mca/iof/base/base.h"
 #include "mca/ns/base/base.h"
 #include "mca/gpr/base/base.h"
 #include "util/session_dir.h"
@@ -45,6 +46,7 @@ int ompi_rte_finalize(void)
   ompi_rte_wait_finalize();
   ompi_rte_internal_fini_spawn();
 
+  mca_iof_base_close();
   mca_pcm_base_close();
   mca_llm_base_close();
   mca_pcmclient_base_close();

src/runtime/ompi_rte_init.c

@@ -31,6 +31,7 @@
 #include "mca/pcm/base/base.h"
 #include "mca/pcmclient/base/base.h"
 #include "mca/llm/base/base.h"
+#include "mca/iof/base/base.h"
 #include "mca/oob/oob.h"
 #include "mca/ns/base/base.h"
 #include "mca/gpr/base/base.h"
@@ -197,6 +198,15 @@ int ompi_rte_init(ompi_cmd_line_t *cmd_line, bool *allow_multi_user_threads, boo
 	return ret;
     }
 
+    /*
+     * Open I/O forwarding components.
+     */
+    if (OMPI_SUCCESS != (ret = mca_iof_base_open())) {
+	/* JMS show_help */
+	printf("show_help: ompi_rte_init failed in mca_iof_base_open\n");
+	return ret;
+    }
+
     printname("component open");
 
     /*
@@ -399,6 +409,15 @@ int ompi_rte_init(ompi_cmd_line_t *cmd_line, bool *allow_multi_user_threads, boo
 	exit(-1);
     }
 
+    /* setup I/O forwarding */
+    if (OMPI_SUCCESS != (ret = mca_iof_base_select(&user_threads, &hidden_threads))) {
+	/* JMS show_help */
+	printf("show_help: ompi_rte_init failed in mca_iof_base_select()\n");
+	return ret;
+    }
+    *allow_multi_user_threads &= user_threads;
+    *have_hidden_threads |= hidden_threads;
+
      /* 
      * All done 
      */

src/runtime/runtime.h

@@ -533,6 +533,11 @@ OMPI_DECLSPEC   int ompi_rte_job_shutdown(mca_ns_base_jobid_t jobid);
      */
 OMPI_DECLSPEC   int ompi_rte_init_cleanup(void);
 
+    /**
+     * Setup I/O forwarding.
+     */
+OMPI_DECLSPEC   int ompi_rte_init_io(void);
+
 #if defined(c_plusplus) || defined(__cplusplus)
 }
 #endif

src/tools/mpirun/mpirun.c

@@ -45,6 +45,7 @@
 #include "mca/pcm/base/base.h"
 #include "mca/oob/oob.h"
 #include "mca/oob/base/base.h"
+#include "mca/iof/iof.h"
 
 #include "runtime/runtime.h"
 #include "runtime/ompi_rte_wait.h"
@@ -103,7 +104,7 @@ main(int argc, char *argv[])
     ompi_rte_process_status_t *proc_status;
     ompi_list_t *status_list;
     ompi_registry_value_t *value;
-        
+    ompi_process_name_t *name;
 
     /*
      * Intialize our Open MPI environment
@@ -353,6 +354,23 @@ main(int argc, char *argv[])
 	     0,
 	     ompi_rte_all_procs_unregistered, NULL);
 
+    /*
+     *  Setup I/O forwarding
+     */
+
+   name = ompi_name_server.create_process_name(0,new_jobid,0);
+   rc_tag = mca_iof.iof_pull(
+         name,
+         OMPI_NS_CMP_JOBID,
+         MCA_IOF_STDOUT,
+         1
+         );
+   rc_tag = mca_iof.iof_pull(
+         name,
+         OMPI_NS_CMP_JOBID,
+         MCA_IOF_STDERR,
+         2
+         );
     /*
      * spawn procs
      */