root/orte/test/mpi/singleton_client_server.c

DEFINITIONS

1. main

   1 #include "orte_config.h"
   2 
   3 #include <stdio.h>
   4 #include <stdlib.h>
   5 #include <string.h>
   6 #include <errno.h>
   7 #include <unistd.h>
   8 #include <mpi.h>
   9 
  10 /*
  11 
  12    LOGIC:
  13 
  14      - the 'server' opens a port and write the info to a file
  15      - the 'clients' open the file and connect to the port
  16      - after each accept, the server and client do a merge to
  17        convert the intercomm to an intracomm
  18 
  19    DETAIL STEPS:
  20 
  21      - server open port
  22      - server does accept
  23      - client #1 does connect
  24      - server and client #1 do merge
  25      - server does accept
  26      - client #2 does connect
  27      - server, client #1 and client #2 do merge
  28      - server does accept
  29      - client #3 does connect
  30      - server, client #1, client #2 and client #3 do merge
  31 
  32 */
  33 
  34 #define TAG 0
  35 
  36 #define CHK(code) do                            \
  37   {                                             \
  38     int retval = code ;                         \
  39     if (retval != MPI_SUCCESS)                  \
  40     {                                           \
  41       fprintf(stderr, "Error: " #code "\n") ;   \
  42       exit(1) ;                                 \
  43     }                                           \
  44   } while(0)
  45 
  46 int main(int argc, char *argv[])
  47 {
  48   char hostname[OPAL_MAXHOSTNAMELEN] ;
  49   char buff[255] ;
  50 
  51   int role ;
  52   int num_clients ;
  53   int size, rank ;
  54 
  55   FILE *fp ;
  56   char server_port_name[MPI_MAX_PORT_NAME] ;
  57 
  58   MPI_Comm intercomm, intracomm ;
  59   MPI_Status status ;
  60   int msg_count ;
  61   int i ;
  62 
  63   /* sanity check the args */
  64   if(argc != 3)
  65   {
  66     fprintf(stderr, "usage %s <num clients> <1:server | 0:client>\n", argv[0]) ;
  67     exit(1) ;
  68   }
  69 
  70   num_clients = atoi(argv[1]) ;
  71   role = atoi(argv[2]) ;
  72 
  73   if (num_clients <= 0 || (role != 0 && role != 1))
  74   {
  75     fprintf(stderr, "usage %s <num clients> <1:server | 0:client>\n", argv[0]) ;
  76     exit(1) ;
  77   }
  78 
  79   /* initialize MPI  */
  80   CHK(MPI_Init(&argc, &argv)) ;
  81 
  82   /* get the node name */
  83   {
  84     int retval = gethostname(hostname, sizeof(hostname));
  85     if(retval == -1)
  86     {
  87       fprintf(stderr, "gethostname failed: %s\n", strerror(errno)) ;
  88       exit(1) ;
  89     }
  90   }
  91 
  92   /* server */
  93   if(role == 1)
  94   {
  95     printf("SERVER: on node '%s'\n", hostname) ;
  96 
  97     /* open port to establish connections */
  98     CHK(MPI_Open_port(MPI_INFO_NULL, server_port_name)) ;
  99 
 100     printf("SERVER: opened port=%s\n", server_port_name) ;
 101 
 102     /* store the port name */
 103     fp = fopen("server_port_name.txt", "w") ;
 104     if(fp == NULL)
 105     {
 106       fprintf(stderr, "fopen failed: %s\n", strerror(errno)) ;
 107       exit(1) ;
 108     }
 109     fprintf(fp, "%s", server_port_name) ;
 110     fclose(fp) ;
 111 
 112     /* the server accepts connections from all the clients */
 113     for(i = 0 ; i < num_clients ; i++ )
 114     {
 115       /* accept connections at this port */
 116       CHK(MPI_Comm_accept(server_port_name, MPI_INFO_NULL, 0,
 117                           i == 0 ? MPI_COMM_WORLD : intracomm,
 118                           &intercomm)) ;
 119 
 120       printf("SERVER: accepted connection from client %d\n", i+1) ;
 121 
 122       /* merge, to form one intra communicator */
 123       CHK(MPI_Intercomm_merge(intercomm, 0, &intracomm)) ;
 124 
 125       printf("SERVER: merged with client %d\n", i+1) ;
 126 
 127       CHK(MPI_Comm_size(intracomm, &size)) ;
 128       CHK(MPI_Comm_rank(intracomm, &rank)) ;
 129 
 130       printf("SERVER: after merging with client %d: size=%d rank=%d\n", i+1, size, rank) ;
 131     }
 132   } /* end server */
 133 
 134   /* client */
 135   if(role == 0)
 136   {
 137     printf("CLIENT: on node '%s'\n", hostname) ;
 138 
 139     fp = fopen("server_port_name.txt", "r") ;
 140     if(fp == NULL)
 141     {
 142       fprintf(stderr, "fopen failed: %s\n", strerror(errno)) ;
 143       exit(1) ;
 144     }
 145     fscanf(fp, "%s", server_port_name) ;
 146     fclose(fp) ;
 147 
 148     printf("CLIENT: attempting to connect to server on port=%s\n", server_port_name) ;
 149 
 150     /* connect to the server */
 151     CHK(MPI_Comm_connect (server_port_name, MPI_INFO_NULL, 0, MPI_COMM_WORLD, &intercomm)) ;
 152 
 153     printf("CLIENT: connected to server on port\n") ;
 154 
 155     /* merge the server and client to one intra communicator */
 156     CHK(MPI_Intercomm_merge(intercomm, 1, &intracomm)) ;
 157 
 158     printf("CLIENT: merged with existing intracomm\n") ;
 159 
 160     CHK(MPI_Comm_size(intracomm, &size)) ;
 161     CHK(MPI_Comm_rank(intracomm, &rank)) ;
 162 
 163     printf("CLIENT: after merging, new comm: size=%d rank=%d\n", size, rank) ;
 164 
 165     for (i = rank ; i < num_clients ; i++)
 166     {
 167       /* client performs a collective accept */
 168       CHK(MPI_Comm_accept(server_port_name, MPI_INFO_NULL, 0, intracomm, &intercomm)) ;
 169 
 170       printf("CLIENT: connected to server on port\n") ;
 171 
 172       /* merge the two intra comms back to one communicator */
 173       CHK(MPI_Intercomm_merge(intercomm, 0, &intracomm)) ;
 174 
 175       printf("CLIENT: merged with existing members\n") ;
 176 
 177       CHK(MPI_Comm_size(intracomm, &size)) ;
 178       CHK(MPI_Comm_rank(intracomm, &rank)) ;
 179 
 180       printf("CLIENT: new size after merging with existing members: size=%d rank=%d\n", size, rank) ;
 181     }
 182 
 183   } /* end client */
 184 
 185   CHK(MPI_Comm_size(intracomm, &size)) ;
 186   CHK(MPI_Comm_rank(intracomm, &rank)) ;
 187 
 188   printf("After fusion: size=%d rank=%d\n", size, rank) ;
 189 
 190   if(rank == 0)
 191   {
 192     msg_count = num_clients ;
 193 
 194     while(msg_count)
 195     {
 196       CHK(MPI_Recv(buff, 255, MPI_CHAR, MPI_ANY_SOURCE,
 197                    MPI_ANY_TAG, intracomm, &status)) ;
 198 
 199       printf("Received hello msg from '%s'\n", buff) ;
 200       msg_count-- ;
 201     }
 202   }
 203   else
 204   {
 205     /* all ranks > 0 */
 206 
 207     CHK(MPI_Send(hostname, strlen(hostname) + 1, MPI_CHAR, 0, TAG, intracomm)) ;
 208   }
 209 
 210   CHK(MPI_Finalize()) ;
 211 
 212   fprintf(stderr, "Rank %d is exiting\n", rank);
 213   return 0 ;
 214 }