text: /*
   * Copyright (c) 2004-2010 The Trustees of Indiana University and Indiana
   *                         University Research and Technology
   *                         Corporation.  All rights reserved.
   * Copyright (c) 2004-2011 The University of Tennessee and The University
   *                         of Tennessee Research Foundation.  All rights
   *                         reserved.
   * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart, 
   *                         University of Stuttgart.  All rights reserved.
   * Copyright (c) 2004-2005 The Regents of the University of California.
   *                         All rights reserved.
   * Copyright (c) 2009      Institut National de Recherche en Informatique
   *                         et Automatique. All rights reserved.
   * Copyright (c) 2011      Cisco Systems, Inc.  All rights reserved.
   * Copyright (c) 2011-2012 Los Alamos National Security, LLC.  All rights
   *                         reserved. 
   * $COPYRIGHT$
   * 
   * Additional copyrights may follow
   * 
   * $HEADER$
   */
  
  #include "orte_config.h"
  #include "orte/constants.h"
  
  #include <sys/types.h>
  #include <stdio.h>
  #ifdef HAVE_FCNTL_H
  #include <fcntl.h>
  #endif
  #ifdef HAVE_UNISTD_H
  #include <unistd.h>
  #endif
  
  #include "opal/dss/dss.h"
  #include "opal/mca/event/event.h"
  #include "opal/runtime/opal.h"
  #include "opal/runtime/opal_cr.h"
  #include "opal/mca/hwloc/base/base.h"
  #include "opal/mca/pstat/base/base.h"
  #include "opal/util/os_path.h"
  
  #include "orte/mca/rml/base/base.h"
  #include "orte/mca/routed/base/base.h"
  #include "orte/mca/routed/routed.h"
  #include "orte/mca/db/base/base.h"
  #include "orte/mca/grpcomm/grpcomm.h"
  #include "orte/mca/grpcomm/base/base.h"
  #include "orte/mca/iof/base/base.h"
  #include "orte/mca/plm/base/base.h"
  #include "orte/mca/odls/base/base.h"
  #include "orte/mca/errmgr/errmgr.h"
  #if OPAL_ENABLE_FT_CR == 1
  #include "orte/mca/snapc/base/base.h"
  #endif
  #include "orte/mca/filem/base/base.h"
  #include "orte/util/proc_info.h"
  #include "orte/util/session_dir.h"
  #include "orte/util/name_fns.h"
  #include "orte/util/nidmap.h"
  #include "orte/util/regex.h"
  #include "orte/util/show_help.h"
  #include "orte/mca/errmgr/base/base.h"
  #include "orte/mca/sensor/base/base.h"
  #include "orte/mca/sensor/sensor.h"
  #include "orte/mca/state/base/base.h"
  #include "orte/mca/state/state.h"
  #include "orte/runtime/orte_cr.h"
  #include "orte/runtime/orte_wait.h"
  #include "orte/runtime/orte_globals.h"
  #include "orte/runtime/orte_quit.h"
  
  #include "orte/mca/ess/base/base.h"
  
  /* local globals */
  static bool plm_in_use=false;
  static bool signals_set=false;
  static opal_event_t term_handler;
  static opal_event_t int_handler;
  static opal_event_t epipe_handler;
  #ifndef __WINDOWS__
  static opal_event_t sigusr1_handler;
  static opal_event_t sigusr2_handler;
  #endif  /* __WINDOWS__ */
  char *log_path = NULL;
  static void shutdown_signal(int fd, short flags, void *arg);
  static void signal_callback(int fd, short flags, void *arg);
  static void epipe_signal_callback(int fd, short flags, void *arg);
  
  static void setup_sighandler(int signal, opal_event_t *ev,
                               opal_event_cbfunc_t cbfunc)
  {
      opal_event_signal_set(orte_event_base, ev, signal, cbfunc, ev);
      opal_event_set_priority(ev, ORTE_ERROR_PRI);
      opal_event_signal_add(ev, NULL);
  }
  
  
  int orte_ess_base_orted_setup(char **hosts)
  {
      int ret = ORTE_ERROR;
      int fd;
      char log_file[PATH_MAX];
      char *jobidstring;
      char *error = NULL;
      char *plm_to_use;
      orte_job_t *jdata;
      orte_proc_t *proc;
      orte_app_context_t *app;
      orte_node_t *node;
  
  #ifndef __WINDOWS__
      /* setup callback for SIGPIPE */
      setup_sighandler(SIGPIPE, &epipe_handler, epipe_signal_callback);
      /* Set signal handlers to catch kill signals so we can properly clean up
       * after ourselves. 
       */
      setup_sighandler(SIGTERM, &term_handler, shutdown_signal);
      setup_sighandler(SIGINT, &int_handler, shutdown_signal);
      
      /** setup callbacks for signals we should ignore */
      setup_sighandler(SIGUSR1, &sigusr1_handler, signal_callback);
      setup_sighandler(SIGUSR2, &sigusr2_handler, signal_callback);
  #endif  /* __WINDOWS__ */
      
      signals_set = true;
      
  #if OPAL_HAVE_HWLOC
      {
          hwloc_obj_t obj;
          unsigned i, j;
          
          /* get the local topology */
          if (NULL == opal_hwloc_topology) {
              if (OPAL_SUCCESS != opal_hwloc_base_get_topology()) {
                  error = "topology discovery";
                  goto error;
              }
          }
          
          /* remove the hostname from the topology. Unfortunately, hwloc
           * decided to add the source hostname to the "topology", thus
           * rendering it unusable as a pure topological description. So
           * we remove that information here.
           */
          obj = hwloc_get_root_obj(opal_hwloc_topology);
          for (i=0; i < obj->infos_count; i++) {
              if (NULL == obj->infos[i].name ||
                  NULL == obj->infos[i].value) {
                  continue;
              }
              if (0 == strncmp(obj->infos[i].name, "HostName", strlen("HostName"))) {
                  free(obj->infos[i].name);
                  free(obj->infos[i].value);
                  /* left justify the array */
                  for (j=i; j < obj->infos_count-1; j++) {
                      obj->infos[j] = obj->infos[j+1];
                  }
                  obj->infos[obj->infos_count-1].name = NULL;
                  obj->infos[obj->infos_count-1].value = NULL;
                  obj->infos_count--;
                  break;
              }
          }
          
          if (4 < opal_output_get_verbosity(orte_ess_base_output)) {
              opal_output(0, "%s Topology Info:", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME));
              opal_dss.dump(0, opal_hwloc_topology, OPAL_HWLOC_TOPO);
          }
      }
  #endif
      
      /* open and setup the opal_pstat framework so we can provide
       * process stats if requested
       */
      if (ORTE_SUCCESS != (ret = opal_pstat_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "opal_pstat_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = opal_pstat_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "opal_pstat_base_select";
          goto error;
      }
      
      /* open and setup the state machine */
      if (ORTE_SUCCESS != (ret = orte_state_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_state_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_state_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_state_base_select";
          goto error;
      }
      
      /* open the errmgr */
      if (ORTE_SUCCESS != (ret = orte_errmgr_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_errmgr_base_open";
          goto error;
      }
      
      /* some environments allow remote launches - e.g., ssh - so
       * open and select something -only- if we are given
       * a specific module to use
       */
      mca_base_param_reg_string_name("plm", NULL,
                                     "Which plm component to use (empty = none)",
                                     false, false,
                                     NULL, &plm_to_use);
      
      if (NULL == plm_to_use) {
          plm_in_use = false;
      } else {
          plm_in_use = true;
          
          if (ORTE_SUCCESS != (ret = orte_plm_base_open())) {
              ORTE_ERROR_LOG(ret);
              error = "orte_plm_base_open";
              goto error;
          }
          
          if (ORTE_SUCCESS != (ret = orte_plm_base_select())) {
              ORTE_ERROR_LOG(ret);
              error = "orte_plm_base_select";
              goto error;
          }
      }
      
      /* Setup the communication infrastructure */
      
      /* Runtime Messaging Layer - this opens/selects the OOB as well */
      if (ORTE_SUCCESS != (ret = orte_rml_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_rml_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_rml_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_rml_base_select";
          goto error;
      }
      
      /* select the errmgr */
      if (ORTE_SUCCESS != (ret = orte_errmgr_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_errmgr_base_select";
          goto error;
      }
      
      /* Routed system */
      if (ORTE_SUCCESS != (ret = orte_routed_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_routed_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_routed_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_routed_base_select";
          goto error;
      }
      
      /* database */
      if (ORTE_SUCCESS != (ret = orte_db_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_db_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_db_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_db_base_select";
          goto error;
      }
  
      /*
       * Group communications
       */
      if (ORTE_SUCCESS != (ret = orte_grpcomm_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_grpcomm_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_grpcomm_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_grpcomm_base_select";
          goto error;
      }
      
      /* Open/select the odls */
      if (ORTE_SUCCESS != (ret = orte_odls_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_odls_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_odls_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_odls_base_select";
          goto error;
      }
      
      /* enable communication with the rml */
      if (ORTE_SUCCESS != (ret = orte_rml.enable_comm())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_rml.enable_comm";
          goto error;
      }
      
      /* initialize the nidmaps */
      if (ORTE_SUCCESS != (ret = orte_util_nidmap_init(NULL))) {
          ORTE_ERROR_LOG(ret);
          error = "orte_util_nidmap_init";
          goto error;
      }
  #if ORTE_ENABLE_STATIC_PORTS
      /* if we are using static ports, then we need to setup
       * the daemon info so the RML can function properly
       * without requiring a wireup stage. This must be done
       * after we enable_comm as that function determines our
       * own port, which we need in order to construct the nidmap
       */
      if (orte_static_ports) {
          /* define the routing tree so we know the pattern
           * if we are trying to setup common or static ports
           */
          orte_routed.update_routing_plan();
  
          /* extract the node info from the environment and
           * build a nidmap from it
           */
          if (ORTE_SUCCESS != (ret = orte_util_build_daemon_nidmap(hosts))) {
              ORTE_ERROR_LOG(ret);
              error = "construct daemon map from static ports";
              goto error;
          }
      }
  #endif
      /* be sure to update the routing tree so the initial "phone home"
       * to mpirun goes through the tree if static ports were enabled - still
       * need to do it anyway just to initialize things
       */
      orte_routed.update_routing_plan();
      
      /* Now provide a chance for the PLM
       * to perform any module-specific init functions. This
       * needs to occur AFTER the communications are setup
       * as it may involve starting a non-blocking recv
       * Do this only if a specific PLM was given to us - the
       * orted has no need of the proxy PLM at all
       */
      if (plm_in_use) {
          if (ORTE_SUCCESS != (ret = orte_plm.init())) {
              ORTE_ERROR_LOG(ret);
              error = "orte_plm_init";
              goto error;
          }
      }
      
      /* setup my session directory */
      if (orte_create_session_dirs) {
          OPAL_OUTPUT_VERBOSE((2, orte_ess_base_output,
                               "%s setting up session dir with\n\ttmpdir: %s\n\thost %s",
                               ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
                               (NULL == orte_process_info.tmpdir_base) ? "UNDEF" : orte_process_info.tmpdir_base,
                               orte_process_info.nodename));
          
          if (ORTE_SUCCESS != (ret = orte_session_dir(true,
                                                      orte_process_info.tmpdir_base,
                                                      orte_process_info.nodename, NULL,
                                                      ORTE_PROC_MY_NAME))) {
              ORTE_ERROR_LOG(ret);
              error = "orte_session_dir";
              goto error;
          }
          /* Once the session directory location has been established, set
             the opal_output env file location to be in the
             proc-specific session directory. */
          opal_output_set_output_file_info(orte_process_info.proc_session_dir,
                                           "output-", NULL, NULL);
          
          /* setup stdout/stderr */
          if (orte_debug_daemons_file_flag) {
              /* if we are debugging to a file, then send stdout/stderr to
               * the orted log file
               */
              
              /* get my jobid */
              if (ORTE_SUCCESS != (ret = orte_util_convert_jobid_to_string(&jobidstring,
                                                                           ORTE_PROC_MY_NAME->jobid))) {
                  ORTE_ERROR_LOG(ret);
                  error = "convert_jobid";
                  goto error;
              }
              
              /* define a log file name in the session directory */
              snprintf(log_file, PATH_MAX, "output-orted-%s-%s.log",
                       jobidstring, orte_process_info.nodename);
              log_path = opal_os_path(false,
                                      orte_process_info.tmpdir_base,
                                      orte_process_info.top_session_dir,
                                      log_file,
                                      NULL);
              
              fd = open(log_path, O_RDWR|O_CREAT|O_TRUNC, 0640);
              if (fd < 0) {
                  /* couldn't open the file for some reason, so
                   * just connect everything to /dev/null
                   */
                  fd = open("/dev/null", O_RDWR|O_CREAT|O_TRUNC, 0666);
              } else {
                  dup2(fd, STDOUT_FILENO);
                  dup2(fd, STDERR_FILENO);
                  if(fd != STDOUT_FILENO && fd != STDERR_FILENO) {
                      close(fd);
                  }
              }
          }
      }
      
      /* setup the global job and node arrays */
      orte_job_data = OBJ_NEW(opal_pointer_array_t);
      if (ORTE_SUCCESS != (ret = opal_pointer_array_init(orte_job_data,
                                                         1,
                                                         ORTE_GLOBAL_ARRAY_MAX_SIZE,
                                                         1))) {
          ORTE_ERROR_LOG(ret);
          error = "setup job array";
          goto error;
      }
      
      orte_node_pool = OBJ_NEW(opal_pointer_array_t);
      if (ORTE_SUCCESS != (ret = opal_pointer_array_init(orte_node_pool,
                                                         ORTE_GLOBAL_ARRAY_BLOCK_SIZE,
                                                         ORTE_GLOBAL_ARRAY_MAX_SIZE,
                                                         ORTE_GLOBAL_ARRAY_BLOCK_SIZE))) {
          ORTE_ERROR_LOG(ret);
          error = "setup node array";
          goto error;
      }
      orte_node_topologies = OBJ_NEW(opal_pointer_array_t);
      if (ORTE_SUCCESS != (ret = opal_pointer_array_init(orte_node_topologies,
                                                         ORTE_GLOBAL_ARRAY_BLOCK_SIZE,
                                                         ORTE_GLOBAL_ARRAY_MAX_SIZE,
                                                         ORTE_GLOBAL_ARRAY_BLOCK_SIZE))) {
          ORTE_ERROR_LOG(ret);
          error = "setup node topologies array";
          goto error;
      }
  
      /* Setup the job data object for the daemons */        
      /* create and store the job data object */
      jdata = OBJ_NEW(orte_job_t);
      jdata->jobid = ORTE_PROC_MY_NAME->jobid;
      opal_pointer_array_set_item(orte_job_data, 0, jdata);
      
      /* every job requires at least one app */
      app = OBJ_NEW(orte_app_context_t);
      opal_pointer_array_set_item(jdata->apps, 0, app);
      jdata->num_apps++;
      
      /* create and store a node object where we are */
      node = OBJ_NEW(orte_node_t);
      node->name = strdup(orte_process_info.nodename);
      node->index = opal_pointer_array_set_item(orte_node_pool, ORTE_PROC_MY_NAME->vpid, node);
  #if OPAL_HAVE_HWLOC
      /* point our topology to the one detected locally */
      node->topology = opal_hwloc_topology;
  #endif
  
      /* create and store a proc object for us */
      proc = OBJ_NEW(orte_proc_t);
      proc->name.jobid = ORTE_PROC_MY_NAME->jobid;
      proc->name.vpid = ORTE_PROC_MY_NAME->vpid;
      
      proc->pid = orte_process_info.pid;
      proc->rml_uri = orte_rml.get_contact_info();
      proc->state = ORTE_PROC_STATE_RUNNING;
      opal_pointer_array_set_item(jdata->procs, proc->name.vpid, proc);
      
      /* record that the daemon (i.e., us) is on this node 
       * NOTE: we do not add the proc object to the node's
       * proc array because we are not an application proc.
       * Instead, we record it in the daemon field of the
       * node object
       */
      OBJ_RETAIN(proc);   /* keep accounting straight */
      node->daemon = proc;
      node->daemon_launched = true;
      node->state = ORTE_NODE_STATE_UP;
      
      /* record that the daemon job is running */
      jdata->num_procs = 1;
      jdata->state = ORTE_JOB_STATE_RUNNING;
      /* obviously, we have "reported" */
      jdata->num_reported = 1;
      
      /* setup the routed info - the selected routed component
       * will know what to do. 
       */
      if (ORTE_SUCCESS != (ret = orte_routed.init_routes(ORTE_PROC_MY_NAME->jobid, NULL))) {
          ORTE_ERROR_LOG(ret);
          error = "orte_routed.init_routes";
          goto error;
      }
      
      /* setup I/O forwarding system - must come after we init routes */
      if (ORTE_SUCCESS != (ret = orte_iof_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_iof_base_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_iof_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_iof_base_select";
          goto error;
      }
      
      /* setup the FileM */
      if (ORTE_SUCCESS != (ret = orte_filem_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_filem_base_open";
          goto error;
      }
      
      if (ORTE_SUCCESS != (ret = orte_filem_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_filem_base_select";
          goto error;
      }
      
  #if OPAL_ENABLE_FT_CR == 1
      /*
       * Setup the SnapC
       */
      if (ORTE_SUCCESS != (ret = orte_snapc_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_snapc_base_open";
          goto error;
      }
      
      if (ORTE_SUCCESS != (ret = orte_snapc_base_select(ORTE_PROC_IS_HNP, !ORTE_PROC_IS_DAEMON))) {
          ORTE_ERROR_LOG(ret);
          error = "orte_snapc_base_select";
          goto error;
      }
      
      /* For daemons, ORTE doesn't need the OPAL CR stuff */
      opal_cr_set_enabled(false);
  #else
      opal_cr_set_enabled(false);
  #endif
      
      /*
       * Initalize the CR setup
       * Note: Always do this, even in non-FT builds.
       * If we don't some user level tools may hang.
       */
      if (ORTE_SUCCESS != (ret = orte_cr_init())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_cr_init";
          goto error;
      }
      
      /* setup the SENSOR framework */
      if (ORTE_SUCCESS != (ret = orte_sensor_base_open())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_sensor_open";
          goto error;
      }
      if (ORTE_SUCCESS != (ret = orte_sensor_base_select())) {
          ORTE_ERROR_LOG(ret);
          error = "orte_sensor_select";
          goto error;
      }
      /* start the local sensors */
      orte_sensor.start(ORTE_PROC_MY_NAME->jobid);
      
      return ORTE_SUCCESS;
      
   error:
      orte_show_help("help-orte-runtime.txt",
                     "orte_init:startup:internal-failure",
                     true, error, ORTE_ERROR_NAME(ret), ret);
      
      return ORTE_ERR_SILENT;
  }
  
  int orte_ess_base_orted_finalize(void)
  {
      /* stop the local sensors */
      orte_sensor.stop(ORTE_PROC_MY_NAME->jobid);
      
      if (signals_set) {
          /* Release all local signal handlers */
          opal_event_del(&epipe_handler);
          opal_event_del(&term_handler);
          opal_event_del(&int_handler);
  #ifndef __WINDOWS__
          opal_event_signal_del(&sigusr1_handler);
          opal_event_signal_del(&sigusr2_handler);
  #endif  /* __WINDOWS__ */
      }
      
      /* cleanup */
      if (NULL != log_path) {
          unlink(log_path);
      }
      
      /* make sure our local procs are dead */
      orte_odls.kill_local_procs(NULL);
      
      /* cleanup any lingering session directories */
      orte_session_dir_cleanup(ORTE_JOBID_WILDCARD);
       
      return ORTE_SUCCESS;    
  }
  
  static void shutdown_signal(int fd, short flags, void *arg)
  {
      /* trigger the call to shutdown callback to protect
       * against race conditions - the trigger event will
       * check the one-time lock
       */
      ORTE_UPDATE_EXIT_STATUS(ORTE_ERROR_DEFAULT_EXIT_CODE);
      ORTE_ACTIVATE_JOB_STATE(NULL, ORTE_JOB_STATE_FORCED_EXIT);
  }
  
  /**
   * Deal with sigpipe errors
   */
  static void epipe_signal_callback(int fd, short flags, void *arg)
  {
      /* for now, we just announce and ignore them */
      opal_output(0, "%s reports a SIGPIPE error on fd %d",
                  ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), fd);
      return;
  }
  
  static void signal_callback(int fd, short event, void *arg)
  {
      /* just ignore these signals */
  }
  