server.c

Path MPICH-V for MPICH the MPI Implementation

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
extern int h_errno;
#include <unistd.h>
#include <signal.h>
#include <sys/ioctl.h>

#include "cmdline_server.h"
#include "debug.h"
#include "redirect.h"

#define __USE_GNU
#include <fcntl.h>
#include <stdio.h>

static int recv_line(int s, char *line, int maxlen)
{
  char c;
  int i;

  for(i = 0; i < maxlen; i++)
    {
      if(recv(s, &c, 1, 0) != 1)
	return -1;
      line[i] = c;
      if(c == '\n')
	{
	  line[i] = 0;
	  return 0;
	}
    }
  line[i] = 0;
  return 1;
}

static int recv_intparam(int s, char *name)
{
  char line[1024];
  if( recv_line(s, line, 1023) < 0 )
    {
      perror("receive intparam");
      close(s);
      exit(1);
    }

  if( !strncmp(line, name, strlen(name)) )
    return strtol( line + strlen(name) + 1, NULL, 0);

  fprintf(stderr, "unexpected parameter: %s\n", line);
  exit(2);
}

static char *recv_strparam(int s, char *name)
{
  char line[1024];
  char *value;

  if( recv_line(s, line, 1023) < 0 )
    {
      perror("receive strparam");
      close(s);
      exit(1);
    }

  if( !strncmp(line, name, strlen(name)) )
    {
      value = (char*)calloc(1, strlen(line) - strlen(name) );
      memcpy(value, line+strlen(name)+1, strlen(line)-strlen(name)-1);
      return value;
    }

  exit(2);
}

struct gengetopt_args_info *initialize_mpichv_service(char *myid,
						      int argc, char *argv[], 
						      int *sock_to_dispatcher,
						      int *main_listen_sock)
{
  struct gengetopt_args_info orig, *result;
  struct sockaddr_in remote;
  struct hostent *h;
  socklen_t rlen;
  int fd;

  cmdline_parser_init (&orig);
  if(cmdline_parser(argc, argv, &orig) != 0)
    {
      cmdline_parser_print_help();
      exit(0);
    }

  if(orig.help_given)
    {
      cmdline_parser_print_help();
      exit(0);
    }
  
  *sock_to_dispatcher = socket(AF_INET, SOCK_STREAM, 0);
  if(*sock_to_dispatcher < 0)
    {
      perror("socket");
      exit(1);
    }

  h = gethostbyname( orig.runtime_ip_arg );
  if(h == NULL)
    {
      fprintf(stderr, "gethostbyname error : %s is not solvable (%s)\n", orig.runtime_ip_arg, hstrerror(h_errno));
      exit(1);
    }
  
  remote.sin_family = AF_INET;
  remote.sin_port = htons(orig.runtime_port_arg);
  memcpy(&remote.sin_addr, h->h_addr, h->h_length);

  if(connect(*sock_to_dispatcher, (struct sockaddr*)&remote, sizeof(struct sockaddr_in)) < 0)
    {
      fprintf(stderr, "connect to %s:%d error: %s\n", 
	      inet_ntoa(remote.sin_addr), ntohs(remote.sin_port), strerror(errno));
      exit(1);
    }

  daemon(1, 0);

  result = (struct gengetopt_args_info*)malloc(sizeof(struct gengetopt_args_info));

#define set_intparam(name) do { \
    if(orig.name##_given) \
       {\
         result->name##_arg = orig.name##_arg; \
         recv_intparam(*sock_to_dispatcher, #name); \
       }\
    else\
         result->name##_arg = recv_intparam(*sock_to_dispatcher, #name);\
  } while(0)
#define set_strparam(name) do { \
    if(orig.name##_given) \
      {\
         result->name##_arg = strdup(orig.name##_arg); \
         free(recv_strparam(*sock_to_dispatcher, #name)); \
      }\
    else\
         result->name##_arg = recv_strparam(*sock_to_dispatcher, #name);\
  } while(0)

  result->runtime_ip_arg = strdup(orig.runtime_ip_arg);
  result->runtime_port_arg = orig.runtime_port_arg;

  /* !! PLEASE !!
   * for sanity reasons, keep this always synchronized with 
   *     server.ggo -- self explanatory
   * and vrun.c   -- search for the !! PLEASE !! commentary
   */
  
  set_intparam(n_procs);
  set_intparam(jobid);
  set_intparam(port);
  set_strparam(working_dir);
  set_strparam(tmp_dir);
  set_strparam(helpers_dir);
  set_strparam(master_sched);
  set_intparam(ckpt_timeout);
  set_strparam(ckpt_strategy);
  set_strparam(el_strategy);
  set_intparam(stdio_port);
  set_intparam(stder_port);
  set_strparam(debug);
  free(recv_strparam(*sock_to_dispatcher, "EOF"));

  redirect_stdio(orig.runtime_ip_arg, result->stdio_port_arg, result->stder_port_arg);

  cmdline_parser_free(&orig);

  chdir(result->working_dir_arg);
  initDebug(myid, result->debug_arg);

  remote.sin_family = AF_INET;
  remote.sin_port = htons(result->port_arg);
  remote.sin_addr.s_addr = INADDR_ANY;

  *main_listen_sock = socket(AF_INET, SOCK_STREAM, 0);
  if(*main_listen_sock < 0)
    qerror("socket creation");

  if(bind(*main_listen_sock, (struct sockaddr*)&remote, sizeof(struct sockaddr_in)) < 0)
    qerror("socket binding");
  
  if(listen(*main_listen_sock, 5) < 0)
    qerror("socket listening");

  rlen = sizeof(struct sockaddr_in);
  if(getsockname(*main_listen_sock, (struct sockaddr*)&remote, &rlen) < 0)
    qerror("get socket name");
  
  result->port_arg = ntohs(remote.sin_port);
  if( send(*sock_to_dispatcher, &remote.sin_port, sizeof(in_port_t), 0) != sizeof(in_port_t) )
    qerror("sending port to dispatcher");

  return result;  
}