root/orte/mca/iof/tool/iof_tool_receive.c

DEFINITIONS

1. orte_iof_tool_recv

   1 /*
   2  * Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
   3  *                         University Research and Technology
   4  *                         Corporation.  All rights reserved.
   5  * Copyright (c) 2004-2005 The University of Tennessee and The University
   6  *                         of Tennessee Research Foundation.  All rights
   7  *                         reserved.
   8  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
   9  *                         University of Stuttgart.  All rights reserved.
  10  * Copyright (c) 2004-2005 The Regents of the University of California.
  11  *                         All rights reserved.
  12  * Copyright (c) 2007      Cisco Systems, Inc.  All rights reserved.
  13  * Copyright (c) 2011      Los Alamos National Security, LLC.  All rights
  14  *                         reserved.
  15  * Copyright (c) 2014-2018 Intel, Inc. All rights reserved.
  16  * $COPYRIGHT$
  17  *
  18  * Additional copyrights may follow
  19  *
  20  * $HEADER$
  21  */
  22 
  23 #include "orte_config.h"
  24 #include "orte/constants.h"
  25 
  26 #include <errno.h>
  27 #ifdef HAVE_UNISTD_H
  28 #include <unistd.h>
  29 #endif  /* HAVE_UNISTD_H */
  30 #include <string.h>
  31 
  32 #include "opal/dss/dss.h"
  33 
  34 #include "orte/mca/rml/rml.h"
  35 #include "orte/mca/rml/rml_types.h"
  36 #include "orte/mca/errmgr/errmgr.h"
  37 #include "orte/util/name_fns.h"
  38 #include "orte/runtime/orte_globals.h"
  39 
  40 #include "orte/mca/iof/iof.h"
  41 #include "orte/mca/iof/base/base.h"
  42 
  43 #include "iof_tool.h"
  44 
  45 
  46 void orte_iof_tool_recv(int status, orte_process_name_t* sender,
  47                         opal_buffer_t* buffer, orte_rml_tag_t tag,
  48                         void* cbdata)
  49 {
  50     orte_process_name_t origin;
  51     unsigned char data[ORTE_IOF_BASE_MSG_MAX];
  52     orte_iof_tag_t stream;
  53     int32_t count, numbytes;
  54     int rc;
  55 
  56 
  57     /* unpack the stream first as this may be flow control info */
  58     count = 1;
  59     if (ORTE_SUCCESS != (rc = opal_dss.unpack(buffer, &stream, &count, ORTE_IOF_TAG))) {
  60         ORTE_ERROR_LOG(rc);
  61         goto CLEAN_RETURN;
  62     }
  63 
  64     /* if this is a CLOSE tag, then ignore the rest - this is just the
  65      * tail end of a handshake to indicate we have closed a stream
  66      */
  67     if (ORTE_IOF_CLOSE & stream) {
  68         OPAL_OUTPUT_VERBOSE((1, orte_iof_base_framework.framework_output,
  69                              "%s received CLOSE handshake from remote hnp %s",
  70                              ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
  71                              ORTE_NAME_PRINT(sender)));
  72         mca_iof_tool_component.closed = true;
  73         goto CLEAN_RETURN;
  74     }
  75 
  76     /* get name of the process whose io we are receiving */
  77     count = 1;
  78     if (ORTE_SUCCESS != (rc = opal_dss.unpack(buffer, &origin, &count, ORTE_NAME))) {
  79         ORTE_ERROR_LOG(rc);
  80         goto CLEAN_RETURN;
  81     }
  82 
  83     /* unpack the data */
  84     numbytes=ORTE_IOF_BASE_MSG_MAX;
  85     if (ORTE_SUCCESS != (rc = opal_dss.unpack(buffer, data, &numbytes, OPAL_BYTE))) {
  86         ORTE_ERROR_LOG(rc);
  87         goto CLEAN_RETURN;
  88     }
  89     /* numbytes will contain the actual #bytes that were sent */
  90 
  91     OPAL_OUTPUT_VERBOSE((1, orte_iof_base_framework.framework_output,
  92                          "%s unpacked %d bytes from remote proc %s",
  93                          ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), numbytes,
  94                          ORTE_NAME_PRINT(&origin)));
  95 
  96     /* if numbytes is zero, it means that the channel was closed on the far end - for
  97      * now, we just ignore this condition
  98      */
  99     if (0 < numbytes) {
 100         /* write the output locally */
 101         if (ORTE_IOF_STDOUT & stream) {
 102             orte_iof_base_write_output(&origin, stream, data, numbytes, orte_iof_base.iof_write_stdout->wev);
 103         } else {
 104             orte_iof_base_write_output(&origin, stream, data, numbytes, orte_iof_base.iof_write_stderr->wev);
 105         }
 106     }
 107 
 108 CLEAN_RETURN:
 109     return;
 110 }