root/ompi/mpi/c/comm_join.c

DEFINITIONS

1. MPI_Comm_join
2. ompi_socket_send
3. ompi_socket_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) 2012      Los Alamos National Security, LLC.
  13  *                         All rights reserved.
  14  * Copyright (c) 2015      Research Organization for Information Science
  15  *                         and Technology (RIST). All rights reserved.
  16  * Copyright (c) 2015-2018 Cisco Systems, Inc.  All rights reserved
  17  * $COPYRIGHT$
  18  *
  19  * Additional copyrights may follow
  20  *
  21  * $HEADER$
  22  */
  23 #include "ompi_config.h"
  24 #include <stdio.h>
  25 #include <string.h>
  26 #ifdef HAVE_UNISTD_H
  27 #include <unistd.h>
  28 #endif
  29 #ifdef HAVE_SYS_TYPES_H
  30 #include <sys/types.h>
  31 #endif
  32 #ifdef HAVE_SYS_SOCKET_H
  33 #include <sys/socket.h>
  34 #endif
  35 #include <errno.h>
  36 #ifdef HAVE_NETINET_IN_H
  37 #include <netinet/in.h>
  38 #endif
  39 
  40 #include "opal/util/show_help.h"
  41 
  42 #include "ompi/mpi/c/bindings.h"
  43 #include "ompi/runtime/params.h"
  44 #include "ompi/runtime/mpiruntime.h"
  45 #include "ompi/communicator/communicator.h"
  46 #include "ompi/errhandler/errhandler.h"
  47 #include "ompi/dpm/dpm.h"
  48 
  49 
  50 #if OMPI_BUILD_MPI_PROFILING
  51 #if OPAL_HAVE_WEAK_SYMBOLS
  52 #pragma weak MPI_Comm_join = PMPI_Comm_join
  53 #endif
  54 #define MPI_Comm_join PMPI_Comm_join
  55 #endif
  56 
  57 static const char FUNC_NAME[] = "MPI_Comm_join";
  58 
  59 static int ompi_socket_send (int fd, char *buf, int len );
  60 static int ompi_socket_recv (int fd, char *buf, int len );
  61 
  62 int MPI_Comm_join(int fd, MPI_Comm *intercomm)
  63 {
  64     int rc;
  65     uint32_t len, rlen, llen, lrlen;
  66     int send_first=0;
  67     ompi_process_name_t rname, tmp_name;
  68 
  69     ompi_communicator_t *newcomp;
  70     char port_name[MPI_MAX_PORT_NAME];
  71 
  72     if ( MPI_PARAM_CHECK ) {
  73         OMPI_ERR_INIT_FINALIZE(FUNC_NAME);
  74 
  75         if ( NULL == intercomm ) {
  76             return OMPI_ERRHANDLER_INVOKE(MPI_COMM_WORLD, MPI_ERR_ARG,
  77                                           FUNC_NAME);
  78         }
  79     }
  80 
  81     if (!ompi_mpi_dynamics_is_enabled(FUNC_NAME)) {
  82         return OMPI_ERRHANDLER_INVOKE(MPI_COMM_WORLD, OMPI_ERR_NOT_SUPPORTED,
  83                                       FUNC_NAME);
  84     }
  85 
  86     OPAL_CR_ENTER_LIBRARY();
  87 
  88     /* send my process name */
  89     tmp_name = *OMPI_PROC_MY_NAME;
  90     OMPI_PROCESS_NAME_HTON(tmp_name);
  91     ompi_socket_send(fd, (char*) &tmp_name, sizeof(tmp_name));
  92 
  93     /* recv the remote name */
  94     ompi_socket_recv(fd, (char*) &rname, sizeof(rname));
  95     OMPI_PROCESS_NAME_NTOH(rname);
  96 
  97     /* compare the two to get send_first */
  98     if (OMPI_PROC_MY_NAME->jobid == rname.jobid) {
  99         if (OMPI_PROC_MY_NAME->vpid < rname.vpid) {
 100             send_first = true;
 101         } else if (OMPI_PROC_MY_NAME->vpid == rname.vpid) {
 102             /* joining to myself is not allowed */
 103             *intercomm = MPI_COMM_NULL;
 104             OPAL_CR_EXIT_LIBRARY();
 105             return MPI_ERR_INTERN;
 106         } else {
 107             send_first = false;
 108         }
 109     } else if (OMPI_PROC_MY_NAME->jobid < rname.jobid) {
 110         send_first = true;
 111     }
 112 
 113     /* Assumption: socket_send should not block, even if the socket
 114        is not configured to be non-blocking, because the message length are
 115        so short. */
 116 
 117     /* we will only use the send_first proc's port name,
 118      * so pass it to the recv_first participant */
 119     if (send_first) {
 120         // The port_name that we get back will be \0-terminated.  The
 121         // strlen+\0 will be <= MPI_MAX_PORT_NAME characters.
 122         if (OMPI_SUCCESS != (rc = ompi_dpm_open_port(port_name))) {
 123             goto error;
 124         }
 125         // Send the strlen+1 so that we both send the \0 and the
 126         // receiver receives the \0.
 127         llen   = (uint32_t)(strlen(port_name)+1);
 128         len    = htonl(llen);
 129         ompi_socket_send( fd, (char *) &len, sizeof(uint32_t));
 130         ompi_socket_send (fd, port_name, llen);
 131     } else {
 132         ompi_socket_recv (fd, (char *) &rlen, sizeof(uint32_t));
 133         // The lrlen that we receive will be the strlen+1 (to account
 134         // for \0), and will be <= MPI_MAX_PORT_NAME.
 135         lrlen  = ntohl(rlen);
 136         ompi_socket_recv (fd, port_name, lrlen);
 137     }
 138 
 139     /* use the port to connect/accept */
 140     rc = ompi_dpm_connect_accept (MPI_COMM_SELF, 0, port_name, send_first, &newcomp);
 141 
 142     OPAL_CR_EXIT_LIBRARY();
 143 
 144     *intercomm = newcomp;
 145 
 146  error:
 147     OPAL_CR_EXIT_LIBRARY();
 148 
 149     if (OPAL_ERR_NOT_SUPPORTED == rc) {
 150         opal_show_help("help-mpi-api.txt",
 151                        "MPI function not supported",
 152                        true,
 153                        FUNC_NAME,
 154                        "Underlying runtime environment does not support join functionality");
 155     }
 156 
 157     OMPI_ERRHANDLER_RETURN (rc, MPI_COMM_SELF, rc, FUNC_NAME);
 158 }
 159 
 160 
 161 static int ompi_socket_send (int fd, char *buf, int len )
 162 {
 163     int num;
 164     size_t s_num;
 165     ssize_t a;
 166     char *c_ptr;
 167     int ret = OMPI_SUCCESS;
 168 
 169     num   = len;
 170     c_ptr = buf;
 171 
 172     do {
 173         s_num = (size_t) num;
 174         a = write ( fd, c_ptr, s_num );
 175         if ( a == -1 ) {
 176             if ( errno == EINTR ) {
 177                 /* Catch EINTR on, mainly on IBM RS6000 */
 178                 continue;
 179             }
 180 #ifdef __SR8000
 181             else if ( errno == EWOULDBLOCK ) {
 182                 /*Catch EWOULDBLOCK on Hitachi SR8000 */
 183                 continue;
 184             }
 185             else if ( errno == EAGAIN ) {
 186                 /* Catch EAGAIN on Hitachi SR8000 */
 187                 continue;
 188             }
 189 #endif
 190             else {
 191                 /* Another error occured */
 192                 fprintf (stderr,"ompi_socket_send: error while writing to socket"
 193                          " error:%s", strerror (errno) );
 194                 return MPI_ERR_OTHER;
 195             }
 196         }
 197         num      -= a;
 198         c_ptr    += a;
 199     }   while ( num > 0 );
 200 
 201 
 202     if ( num < 0 )  {
 203         fprintf (stderr, "ompi_socket_send: more data written then available");
 204         ret = MPI_ERR_INTERN;
 205     }
 206 
 207     return ret;
 208 }
 209 
 210 static int ompi_socket_recv (int fd, char *buf, int len )
 211 {
 212     int num;
 213     size_t s_num;
 214     ssize_t a;
 215     char *c_ptr;
 216     int ret = MPI_SUCCESS;
 217 
 218     num      = len;
 219     c_ptr      = buf;
 220 
 221     do {
 222         s_num = (size_t ) num;
 223         a = read ( fd, c_ptr, s_num );
 224         if ( a == -1 ) {
 225             if ( errno == EINTR ) {
 226                 /* Catch EINTR on, mainly on IBM RS6000 */
 227                 continue;
 228             }
 229 #ifdef __SR8000
 230             else if ( errno == EWOULDBLOCK ) {
 231                 /*Catch EWOULDBLOCK on Hitachi SR8000 */
 232                 continue;
 233             }
 234             else if ( errno == EAGAIN ) {
 235                 /* Catch EAGAIN on Hitachi SR8000 */
 236                 continue;
 237             }
 238 #endif
 239             else {
 240                 /* Another error occured */
 241                 fprintf (stderr,"ompi_socket_recv: error while reading from socket"
 242                          " error:%s", strerror (errno) );
 243                 return MPI_ERR_OTHER;
 244             }
 245         }
 246         num    -= a;
 247         c_ptr  += a;
 248     }   while ( num > 0 );
 249 
 250     if ( num < 0 )  {
 251         fprintf (stderr, "ompi_socket_recv: more data read then available");
 252         ret = MPI_ERR_INTERN;
 253     }
 254 
 255     return ret;
 256 }