Home Research Papers Posters Manuals Talks Download People Help Internal

arch/net/charmrun/charmrun.c

Go to the documentation of this file.

#include "converse.h"

#include "../sockRoutines.h"
#include "../sockRoutines.c"
#include "../ccs-auth.h"
#include "../ccs-auth.c"
#include "../ccs-server.h"
#include "../ccs-server.c"

#include <stdio.h>
#include <string.h>
#include <ctype.h>
#include <errno.h>
#include <setjmp.h>
#include <stdlib.h>
#include <signal.h>
#include <fcntl.h>
#include <time.h>
#include <assert.h>
#include <math.h>
#if CMK_BPROC
#include <sys/bproc.h>
#endif
#if CMK_USE_POLL
#include <poll.h>
#endif
#include <sys/stat.h>


#if defined(_WIN32) && !defined(__CYGWIN__)
/*Win32 has screwy names for the standard UNIX calls:*/
#define getcwd _getcwd
#define strdup _strdup
#define unlink _unlink
#define open _open
#define fdopen _fdopen
#define ftruncate _chsize
#include <winbase.h>
#include <direct.h>
#include <io.h>
#include <sys/timeb.h>
#include <process.h>
#define DIRSEP "\\"
#define SIGBUS -1  /*These signals don't exist in Win32*/
#define SIGKILL -1
#define SIGQUIT -1


#else /*UNIX*/
#include <pwd.h>   /*getcwd*/
#include <unistd.h>
#define DIRSEP "/"
#endif

#if CMK_RSH_NOT_NEEDED /*No RSH-- use daemon to start node-programs*/
#  define CMK_USE_RSH 0

#else /*Use RSH to start node-programs*/
#  define CMK_USE_RSH 1
#ifdef __MINGW_H
#  include <rpc.h>
#elif !defined(__CYGWIN__)
#  include <rpc/rpc.h>
#else
#  include <w32api/rpc.h>
#endif
#  if CMK_RSH_IS_A_COMMAND
#    define RSH_CMD "rsh"
#  endif

#  if CMK_RSH_USE_REMSH
#    define RSH_CMD "remsh"
#  endif
#endif

#include "daemon.h"

/*#define DEBUGF(x) printf x*/
#define DEBUGF(x) 

#ifndef MAXPATHLEN
#define MAXPATHLEN 1024
#endif


//#define HSTART
#ifdef HSTART
/*Hierarchical-start routines*/
int mynodes_start ;     /* To keep a global node numbering */

#endif

static double ftTimer;

double start_timer;

int *rsh_pids=NULL;
                                                                                
double GetClock(void)
{
#if defined(_WIN32) && !defined(__CYGWIN__)
  struct _timeb tv;
  _ftime(&tv);
  return (tv.time * 1.0 + tv.millitm * 1.0E-3);
#else
  struct timeval tv; int ok;
  ok = gettimeofday(&tv, NULL);
  if (ok<0) { perror("gettimeofday"); exit(1); }
  return (tv.tv_sec * 1.0 + tv.tv_usec * 1.0E-6);
#endif
}


int probefile(path)
    char *path;
{
        FILE *f=fopen(path,"r");
        if (f==NULL) return 0;
        fclose(f);
        return 1;
}

char *mylogin(void)
{
#if defined(_WIN32) && !defined(__CYGWIN__)
        static char name[100]={'d','u','n','n','o',0};
        int len=100;
        GetUserName(name,&len);
        return name;
#else /*UNIX*/
  struct passwd *self;

  self = getpwuid(getuid());
  if (self==0) { 
#if CMK_HAS_POPEN
    char cmd[16];
    char uname[64];
    FILE *p;
    sprintf(cmd, "id -u -n");
    p = popen(cmd, "r");
    if (p){
        fscanf(p, "%s", uname);
        pclose(p);
        return strdup(uname);
    }
    else
        return "unknown"; 
#else
    return "unknown"; 
#endif
  }
  return self->pw_name;
#endif
} 

/**************************************************************************
 *
 * ping_developers
 *
 * Sends a single UDP packet to the charm developers notifying them
 * that charm is in use.
 *
 **************************************************************************/

void ping_developers()
{
#ifdef NOTIFY
  char               info[1000];
  /*This is the resolved IP address of elegance.cs.uiuc.edu */
  skt_ip_t destination_ip=skt_lookup_ip("128.174.241.211");
  unsigned int destination_port=6571;
  struct sockaddr_in addr=skt_build_addr(destination_ip,destination_port);
  SOCKET             skt;
  
  skt = socket(AF_INET, SOCK_DGRAM, 0);
  if (skt == INVALID_SOCKET) return;

  sprintf(info,"%s",mylogin());
  
  sendto(skt, info, strlen(info), 0, (struct sockaddr *)&addr, sizeof(addr));
  skt_close(skt);
#endif /* NOTIFY */
}

/**************************************************************************
 *
 * Pathfix : alters a path according to a set of rewrite rules
 *
 *************************************************************************/

typedef struct pathfixlist {
  char *s1;
  char *s2;
  struct pathfixlist *next;
} *pathfixlist;

pathfixlist pathfix_append(char *s1, char *s2, pathfixlist l)
{
  pathfixlist pf = (pathfixlist)malloc(sizeof(struct pathfixlist));
  pf->s1 = s1;
  pf->s2 = s2;
  pf->next = l;
  return pf;
}

char *pathfix(char *path, pathfixlist fixes)
{
  char buffer[MAXPATHLEN]; pathfixlist l; 
  char buf2[MAXPATHLEN]; 
  char *offs; int mod, len;
  strcpy(buffer,path);
  mod = 1;
  while (mod) {
    mod = 0;
    for (l=fixes; l; l=l->next) {
      len = strlen(l->s1);
      offs = strstr(buffer, l->s1);
      if (offs) {
        offs[0]=0;
        sprintf(buf2,"%s%s%s",buffer,l->s2,offs+len);
        strcpy(buffer,buf2);
        mod = 1;
      }
    }
  }
  return strdup(buffer);
}

char *pathextfix(char *path, pathfixlist fixes, char *ext)
{
  char *newpath = pathfix(path, fixes);
  char *ret;
  if (ext == NULL) return newpath;
  ret = (char *)malloc(strlen(newpath)+strlen(ext)+2);
  strcpy(ret, newpath);
  strcat(ret, ext);
  return ret;
}

/****************************************************************************
 *
 * Miscellaneous minor routines.
 *
 ****************************************************************************/

int is_quote(char c)
{
  return (c=='\'' || c == '"');
}

void zap_newline(char *s)
{
  char *p;
  p = s + strlen(s)-1;
  if (*p == '\n') *p = '\0';
  /* in case of DOS ^m */
  p--;
  if (*p == '\15') *p = '\0';
}

/* get substring from lo to hi, remove quote chars */
char *substr(char *lo, char *hi)
{
  int len;
  char *res;
  if (is_quote(*lo)) lo++;
  if (is_quote(*(hi-1))) hi--;
  len = hi-lo;
  res = (char *)malloc(1+len);
  memcpy(res, lo, len);
  res[len]=0;
  return res;
}

int subeqs(char *lo, char *hi, char *str)
{
  int len = strlen(str);
  if (hi-lo != len) return 0;
  if (memcmp(lo, str, len)) return 0;
  return 1;
}

/* advance pointer over blank characters */
char *skipblanks(char *p)
{
  while ((*p==' ')||(*p=='\t')) p++;
  return p;
}

/* advance pointer over nonblank characters and a quoted string */
char *skipstuff(char *p)
{
  char quote = 0;
  if (*p && (*p=='\'' || *p=='"')) { quote=*p; p++; }
  if (quote != 0) {
    while (*p&&*p!=quote) p++;
    if (*p!=quote) {
      fprintf(stderr, "ERROR> Unmatched quote in nodelist file.\n");
      exit(1);
    }
    p++;
  }
  else
    while ((*p)&&(*p!=' ')&&(*p!='\t')) p++;
  return p;
}

#if CMK_USE_RSH
char *getenv_rsh()
{
  char *e;

  e = getenv("CONV_RSH");
  return e ? e : RSH_CMD;
}
#endif

#if !defined(_WIN32) || defined(__CYGWIN__)
char *getenv_display()
{
  static char result[100],ipBuf[200];
  char *e, *p;
  
  e = getenv("DISPLAY");
  if (e==0) return NULL;
  p = strrchr(e, ':');
  if (p==0) return NULL;
  if ((e[0]==':')||(strncmp(e,"unix:",5)==0)) {
    sprintf(result,"%s:%s",skt_print_ip(ipBuf,skt_my_ip()),p+1);
  }
  else strcpy(result, e);
  return result;
}
char *getenv_display_no_tamper()
{
  static char result[100],ipBuf[200];
  char *e, *p;
  
  e = getenv("DISPLAY");
  if (e==0) return NULL;
  p = strrchr(e, ':');
  if (p==0) return NULL;
  strcpy(result, e);
  return result;
}

#endif

/*****************************************************************************
 *                                                                           *
 * PPARAM - obtaining "program parameters" from the user.                    *
 *                                                                           *
 *****************************************************************************/

typedef struct ppdef
{
  union {
      int *i;
      double *r;
      char **s;
      int *f;
    } where;/*Where to store result*/
  const char *lname; /*Argument name on command line*/
  const char *doc;
  char  type; /*One of i, r, s, f.*/
  struct ppdef *next;
}
*ppdef;

static ppdef ppdefs;

static int     pparam_pos;
static char  **pparam_argv;
static char    pparam_optc='-';
char           pparam_error[100];

static ppdef pparam_find(lname)
    const char *lname;
{
  ppdef def;
  for (def=ppdefs; def; def=def->next)
    if (strcmp(def->lname, lname)==0)
      return def;
  return 0;
}

static ppdef pparam_cell(lname)
    const char *lname;
{
  ppdef def = pparam_find(lname);
  if (def) return def;
  def = (ppdef)malloc(sizeof(struct ppdef));
  def->lname = lname;
  def->type  = 's';
  def->doc   = "(undocumented)";
  def->next  = ppdefs;
  ppdefs = def;
  return def;
}



void pparam_int(int *where,int defValue,
                                   const char *arg,const char *doc)
{
  ppdef def = pparam_cell(arg);
  def->type  = 'i';
  def->where.i = where; *where=defValue;
  def->lname=arg;
  def->doc=doc;
}

void pparam_flag(int *where,int defValue,
                                   const char *arg,const char *doc)
{
  ppdef def = pparam_cell(arg);
  def->type  = 'f';
  def->where.f = where; *where=defValue;
  def->lname=arg;
  def->doc=doc;
}

void pparam_real(double *where,double defValue,
                                   const char *arg,const char *doc)
{
  ppdef def = pparam_cell(arg);
  def->type  = 'r';
  def->where.r = where; *where=defValue;
  def->lname=arg;
  def->doc=doc;
}
void pparam_str(char **where,char *defValue,
                                   const char *arg,const char *doc)
{
  ppdef def = pparam_cell(arg);
  def->type  = 's';
  def->where.s = where; *where=defValue;
  def->lname=arg;
  def->doc=doc;
}

static int pparam_setdef(def, value)
    ppdef def; char *value;
{
  char *p;
  switch(def->type)
    {
    case 'i' :
      *def->where.i = strtol(value, &p, 10);
      if (*p) return -1;
      return 0;
    case 'r' :
      *def->where.r = strtod(value, &p);
      if (*p) return -1;
      return 0;
    case 's' :
      *def->where.s = strdup(value);
      return 0;
    case 'f' :
      *def->where.f = strtol(value, &p, 10);
      if (*p) return -1;
      return 0;
    }
  return -1;
}

int pparam_set(lname, value)
    char *lname; char *value;
{
  ppdef def = pparam_cell(lname);
  return pparam_setdef(def, value);
}

char *pparam_getdef(def)
    ppdef def;
{
  static char result[100];
  switch(def->type)
    {
    case 'i': sprintf(result,"%d", *def->where.i); return result;
    case 'r': sprintf(result,"%f",*def->where.r); return result;
    case 's': return *def->where.s?*def->where.s:"";
    case 'f': sprintf(result,"%d", *def->where.f); return result;
    }
  return NULL;
}

void pparam_printdocs()
{
  ppdef def; int len, maxname, maxdoc;
  maxname = 0;
  maxdoc = 0;
  for (def=ppdefs; def; def=def->next)
    {
      len = strlen(def->lname);
      if (len>maxname) maxname=len;
      len = strlen(def->doc);
      if (len>maxdoc) maxdoc=len;
    }
  fprintf(stderr,"\n");
  fprintf(stderr,"Charmrun Command-line Parameters:\n");
  for (def=ppdefs; def; def=def->next)
    {
      fprintf(stderr,"  %c%c%-*s ",pparam_optc,pparam_optc,maxname,def->lname);
      fprintf(stderr,"  %-*s [%s]\n",maxdoc,def->doc,pparam_getdef(def));
    }
  fprintf(stderr,"\n");
}

void pparam_delarg(i)
    int i;
{
  int j;
  for (j=i; pparam_argv[j]; j++)
    pparam_argv[j]=pparam_argv[j+1];
}

int pparam_countargs(argv)
    char **argv;
{
  int argc;
  for (argc=0; argv[argc]; argc++);
  return argc;
}

int pparam_parseopt()
{
  int ok; ppdef def=NULL;
  char *opt = pparam_argv[pparam_pos];
  /* handle ++ by skipping to end */
  if ((opt[1]=='+')&&(opt[2]==0))
    {
      pparam_delarg(pparam_pos);
      while (pparam_argv[pparam_pos]) pparam_pos++;
      return 0;
    }
  /* handle + by itself - an error */
  if (opt[1]==0) 
    {
      sprintf(pparam_error,"Illegal option +\n");
      return -1;
    }
  /* look up option definition */
  if (opt[1]=='+') def = pparam_find(opt+2);
  else
    {
      char name[2];
      name[0]=opt[1];
      if (strlen(opt)<=2 || !isalpha(opt[2]))
      {
      name[1]=0;
      def = pparam_find(name);
      }
    }
  if (def==NULL)
  {
    if (opt[1]=='+')
    {
       sprintf(pparam_error,"Option %s not recognized.",opt);
       return -1;
    } else {
           /*Unrecognized + option-- skip it.*/
           pparam_pos++;
           return 0;
        }
  }
  /* handle flag-options */
  if ((def->type=='f')&&(opt[1]!='+')&&(opt[2]))
    {
      sprintf(pparam_error,"Option %s should not include a value",opt);
      return -1;
    }
  if (def->type=='f')
    {
      *def->where.f = 1;
      pparam_delarg(pparam_pos);
      return 0;
    }
  /* handle non-flag options */
  if ((opt[1]=='+')||(opt[2]==0))
    {
      pparam_delarg(pparam_pos);
      opt = pparam_argv[pparam_pos];
    }
  else opt+=2;
  if ((opt == 0)||(opt[0] == 0))
    {
      sprintf(pparam_error,"%s must be followed by a value.",opt);
      return -1;
    }
  ok = pparam_setdef(def, opt);
  pparam_delarg(pparam_pos);
  if (ok<0)
    {
      sprintf(pparam_error,"Illegal value for %s",opt);
      return -1;
    }
  return 0;
}

int pparam_parsecmd(optchr, argv)
    char optchr; char **argv;
{
  pparam_error[0]=0;
  pparam_argv = argv;
  pparam_optc = optchr;
  pparam_pos  = 0;
  while(1)
    {
      char *opt = pparam_argv[pparam_pos];
      if (opt==0) break;
      if (opt[0]!=optchr) pparam_pos++;
      else if (pparam_parseopt()<0) return -1;
    }
  return 0;
}

#ifdef HSTART
char **
dupargv (argv)
     char **argv;
{
  int argc;
  char **copy;
  
  if (argv == NULL)
    return NULL;
  
  /* the vector */
  for (argc = 0; argv[argc] != NULL; argc++);
  copy = (char **) malloc ((argc +2) * sizeof (char *));
  if (copy == NULL)
    return NULL;
  
  /* the strings */
  for (argc = 0; argv[argc] != NULL; argc++)
    {
      int len = strlen (argv[argc]);
      copy[argc] = malloc (sizeof (char ) * (len + 1));
      strcpy (copy[argc], argv[argc]);
    }
  copy[argc] = NULL;
  return copy;
}

#endif

/****************************************************************************
 * 
 * ARG
 *
 * The following module computes a whole bunch of miscellaneous values, which
 * are all constant throughout the program.  Naturally, this includes the
 * value of the command-line arguments.
 *
 *****************************************************************************/


#define MAX_NODES 1000
#define MAX_LINE_LENGTH 1000

char **arg_argv;
int    arg_argc;

int   arg_requested_pes;
int   arg_timeout;
int   arg_verbose;
char *arg_nodelist;
char *arg_nodegroup;
char *arg_runscript; /* script to run the node-program with */
char *arg_charmrunip;
#if CONVERSE_VERSION_VMI
char *arg_vmispecfile;
#endif

int   arg_debug;
int   arg_debug_no_pause;
int   arg_debug_no_xrdb;
int   arg_charmdebug;
char *arg_debug_commands; /* commands that are provided by a ++debug-commands flag. These are passed into gdb. */

int   arg_local;        /* start node programs directly by exec on localhost */
int   arg_batch_spawn;  /* control starting node programs, several at a time */
int   arg_scalable_start;

#ifdef HSTART
int       arg_hierarchical_start;
int       arg_child_charmrun;
#endif
int   arg_help;         /* print help message */
int   arg_ppn;          /* pes per node */
int   arg_usehostname;

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
int     arg_read_pes=0;
#endif

#if CMK_USE_RSH
int   arg_maxrsh;
char *arg_shell;
int   arg_in_xterm;
char *arg_debugger;
char *arg_xterm;
char *arg_display;
int arg_ssh_display;
char *arg_mylogin;
#endif
int   arg_mpiexec;
int   arg_no_va_rand;

char *arg_nodeprog_a;
char *arg_nodeprog_r;
char *arg_currdir_a;
char *arg_currdir_r;

int   arg_server;
int   arg_server_port=0;
char *arg_server_auth=NULL;
int   replay_single=0;

#if CMK_BPROC
int   arg_startpe;
int   arg_endpe;
int   arg_singlemaster;
int   arg_skipmaster;
#endif

void arg_init(int argc, char **argv)
{
  static char buf[1024];
  
  int i, local_def=0;
#if CMK_CHARMRUN_LOCAL
  local_def=1; /*++local is the default*/
#endif
  
  pparam_int(&arg_requested_pes, 1, "p",             "number of processes to create");
  pparam_int(&arg_timeout,      60, "timeout",       "seconds to wait per host connection");
  pparam_flag(&arg_verbose,      0, "verbose",       "Print diagnostic messages");
  pparam_str(&arg_nodelist,      0, "nodelist",      "file containing list of nodes");
  pparam_str(&arg_nodegroup,"main", "nodegroup",     "which group of nodes to use");
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
        pparam_int(&arg_read_pes, 0, "readpe",             "number of host names to read into the host table");
#endif

#if CMK_CCS_AVAILABLE
  pparam_flag(&arg_server,       0, "server",        "Enable client-server (CCS) mode");
  pparam_int(&arg_server_port,   0, "server-port",   "Port to listen for CCS requests");
  pparam_str(&arg_server_auth,   0, "server-auth",   "CCS Authentication file");
#endif
  pparam_flag(&arg_local,       local_def, "local", "Start node programs locally without daemon");
  pparam_int(&arg_batch_spawn,   0, "batch", "Rsh several node programs at a time, avoiding overloading charmrun pe");
  pparam_flag(&arg_scalable_start, 0, "scalable-start", "scalable start");
#ifdef HSTART
  pparam_flag(&arg_hierarchical_start, 0, "hierarchical-start", "hierarchical start");
  pparam_flag(&arg_child_charmrun, 0, "child-charmrun", "child charmrun");
#endif
  pparam_flag(&arg_usehostname,  0, "usehostname", "Send nodes our symbolic hostname instead of IP address");
  pparam_str(&arg_charmrunip,    0, "useip",      "Use IP address provided for charmrun IP");
  pparam_flag(&arg_mpiexec,          0, "mpiexec",   "use mpiexec to start jobs");
#if CMK_USE_RSH
  pparam_flag(&arg_debug,         0, "debug",         "Run each node under gdb in an xterm window");
  pparam_flag(&arg_debug_no_pause,0, "debug-no-pause","Like debug, except doesn't pause at beginning");
  pparam_str(&arg_debug_commands,   0, "debug-commands",   "Commands to be run inside gdb at startup");
  pparam_flag(&arg_debug_no_xrdb,0, "no-xrdb","Don't check xrdb");

  /* When the ++charmdebug flag is used, charmrun listens from its stdin for
     commands, and forwards them to the gdb info program (a child), or to the
     processor gdbs. The stderr is redirected to the stdout, so the two streams
     are mixed together. The channel for stderr is reused to forward the replies
     of gdb back to the java debugger. */
#if !defined(_WIN32)
  pparam_flag(&arg_charmdebug,    0, "charmdebug",    "Used only when charmrun is started by charmdebug");
#endif

  pparam_int(&arg_maxrsh,        16, "maxrsh",        "Maximum number of rsh's to run at a time");
  pparam_str(&arg_shell,          0, "remote-shell",  "which remote shell to use");
  pparam_str(&arg_debugger,       0, "debugger",      "which debugger to use");
  pparam_str(&arg_display,        0, "display",       "X Display for xterm");
  pparam_flag(&arg_ssh_display,   0, "ssh-display",   "use own X Display for each ssh session");
  pparam_flag(&arg_in_xterm,      0, "in-xterm",      "Run each node in an xterm window");
  pparam_str(&arg_xterm,          0, "xterm",         "which xterm to use");
#endif
#ifdef CMK_BPROC
  /* options for Scyld */
  pparam_int(&arg_startpe,   0, "startpe",   "first pe to start job(SCYLD)");
  pparam_int(&arg_endpe,  1000000, "endpe",   "last pe to start job(SCYLD)");
  pparam_flag(&arg_singlemaster, 0, "singlemaster", "Only assign one process to master node(SCYLD)");
  pparam_flag(&arg_skipmaster, 0, "skipmaster", "Donot assign any process to master node(SCYLD)");
  if (arg_skipmaster && arg_singlemaster) {
    printf("Charmrun> 'singlemaster' is ignored due to 'skipmaster'. \n");
    arg_singlemaster = 0;
  }
  pparam_flag(&arg_debug,         0, "debug",         "turn on more verbose debug print");
#endif
#ifdef CONVERSE_VERSION_VMI
  pparam_str (&arg_vmispecfile, 0, "specfile", "device specfile to load (VMI)");
#endif
  pparam_str(&arg_runscript,    0, "runscript", "script to run node-program with");
  pparam_flag(&arg_help,        0, "help", "print help messages");
  pparam_int(&arg_ppn,          0, "ppn",             "number of pes per node");
  pparam_flag(&arg_no_va_rand,   0, "no-va-randomization",   "Disables randomization of the virtual address  space");
#ifdef HSTART
  arg_argv = dupargv(argv);
#endif

  if (pparam_parsecmd('+', argv) < 0) {
    fprintf(stderr,"ERROR> syntax: %s\n",pparam_error);
    pparam_printdocs();
    exit(1);
  }
  
  /* Check for (but do *not* remove) the "-?", "-h", or "--help" flags */
  for (i=0;argv[i];i++) {
        if (0==strcmp(argv[i],"-?") ||
            0==strcmp(argv[i],"-h") ||
            0==strcmp(argv[i],"--help")) 
                arg_help=1;
  }
  if (arg_help) {
    pparam_printdocs();
    /*exit(0);*/
  }

#ifdef HSTART
  if (!arg_hierarchical_start || arg_child_charmrun)
#endif
  arg_argv = argv+1; /*Skip over charmrun (0) here and program name (1) later*/
  arg_argc = pparam_countargs(arg_argv);
  if (arg_argc<1) {
    fprintf(stderr,"ERROR> You must specify a node-program.\n");
    pparam_printdocs();
    exit(1);
  }

#ifdef HSTART
        if (!arg_hierarchical_start || arg_child_charmrun){
                //Removing nodeprogram from the list
                arg_argv++; arg_argc--;
        }
        else{
                //Removing charmrun from parameters     
                arg_argv++;arg_argc--;
                
                arg_argv[arg_argc]=malloc(sizeof(char) * strlen("++child-charmrun"));
                strcpy(arg_argv[arg_argc++],"++child-charmrun");
                arg_argv[arg_argc] = NULL;
        }
#else
  arg_argv++; arg_argc--;
#endif
                                        
  if (arg_server_port || arg_server_auth) arg_server=1;

  if (arg_debug || arg_debug_no_pause) {
        arg_verbose=1;
        /*Pass ++debug along to program (used by machine.c)*/
        arg_argv[arg_argc++]="++debug";
  }

  /* Check for +replay-detail to know we have to load only one single processor */
  for (i=0;argv[i];i++) {
    if (0==strcmp(argv[i],"+replay-detail")) {
      replay_single = 1;
      arg_requested_pes = 1;
    }
  }
  
#ifdef CMK_BPROC
  if (arg_local) {
    fprintf(stderr,"Warning> ++local cannot be used in bproc version, ignored!\n");
    arg_local = 0;
  }
#endif

#if CMK_USE_RSH
  /* Find the current value of the CONV_RSH variable */
  if(!arg_shell) {
    if (arg_mpiexec)
      arg_shell = "mpiexec";
    else
      arg_shell = getenv_rsh();
  }

  /* Find the current value of the DISPLAY variable */
  if(!arg_display)
    arg_display = getenv_display_no_tamper();
  if ((arg_debug || arg_debug_no_pause || arg_in_xterm) && (arg_display==0)) {
    fprintf(stderr,"ERROR> DISPLAY must be set to use debugging mode\n");
    exit(1);
  }
  if (arg_debug || arg_debug_no_pause) 
    arg_timeout=8*60*60; /* Wait 8 hours for ++debug */

  /* default debugger is gdb */
  if(!arg_debugger)
    arg_debugger = "gdb" ;
  /* default xterm is xterm */
  if(!arg_xterm)
    arg_xterm = "xterm" ;

  arg_mylogin = mylogin();
#endif

#if CONVERSE_VERSION_VMI
  if (!arg_vmispecfile) {
    arg_vmispecfile = getenv ("VMI_SPECFILE");
  }

  if (!arg_vmispecfile) {
    fprintf (stderr, "ERROR> ++specfile not specified and VMI_SPECFILE not given in environment\n");
    exit (1);
  }
#endif

  /* find the current directory, absolute version */
  getcwd(buf, 1023);
  arg_currdir_a = strdup(buf);
    
  /* find the node-program, absolute version */
  arg_nodeprog_r = argv[1];

  if (arg_nodeprog_r[0]=='-' || arg_nodeprog_r[0]=='+') 
  { /*If it starts with - or +, it ain't a node program.
      Chances are, the user screwed up and passed some unknown flag to charmrun*/
     printf("Charmrun does not recognize the flag '%s'.\n",arg_nodeprog_r);
     if (arg_nodeprog_r[0]=='+')
       printf("Charm++'s flags need to be placed *after* the program name.\n");
     pparam_printdocs();
     exit(1);
  }


#if defined(_WIN32) && !defined(__CYGWIN__)
  if (argv[1][1]==':' || argv[1][0]=='\\' && argv[1][1]=='\\') { /*E.g.: "C:\foo\bar.exe*/
#else
  if (argv[1][0]=='/') { /*E.g.: "\foo\bar"*/
#endif
          /*Absolute path to node-program*/
    arg_nodeprog_a = argv[1];
  } else {
    sprintf(buf,"%s%s%s",arg_currdir_a,DIRSEP,arg_nodeprog_r);
    arg_nodeprog_a = strdup(buf);
  }
  if (arg_scalable_start) {
    printf("Charmrun> scalable start enabled. \n");
    if (arg_debug || arg_debug_no_pause) {
      fprintf(stderr, "Charmrun> Error: ++scalable-start does not support debugging mode. \n");
      exit(1);
    }
}

#ifdef HSTART
  if (arg_hierarchical_start) {
    printf("Charmrun> Hierarchical scalable start enabled. \n");
    if (arg_debug || arg_debug_no_pause) {
      fprintf(stderr, "Charmrun> Error: ++hierarchial-start does not support debugging mode. \n");
      exit(1);
    }
   if (arg_verbose) {
      fprintf(stderr, "Charmrun> Warning: you have enabled verbose output with Hierarchical startup, you may get inconsistent verbose outputs. \n++hierarchial-start does not support verbose mode. \n");
    }

        }
  else if(arg_child_charmrun) {
      fprintf(stderr, "Charmrun> Error: ++child-charmrun is not a user-specified flag. \n");
      exit(1);
    }
#endif
if(arg_debug && arg_local){
        printf("++debug cannot be used with ++local.\n");
        exit(0);
        }
}

/****************************************************************************
 *                                                                           
 * NODETAB:  The nodes file and nodes table.
 *
 ****************************************************************************/

static int portOk = 1;
static const char *nodetab_tempName=NULL;
char *nodetab_file_find()
{
  char buffer[MAXPATHLEN];

  /* Find a nodes-file as specified by ++nodelist */
  if (arg_nodelist) {
    char *path = arg_nodelist;
    if (probefile(path)) return strdup(path);
    fprintf(stderr,"ERROR> No such nodelist file %s\n",path);
    exit(1);
  }
  /* Find a nodes-file as specified by getenv("NODELIST") */
  if (getenv("NODELIST")) {
    char *path = getenv("NODELIST");        
    if (path && probefile(path)) return strdup(path);
    fprintf(stderr,"ERROR> Cannot find nodelist file %s\n",path);
    exit(1);
  }
  /* Find a nodes-file by looking under 'nodelist' in the current directory */
  if (probefile("./nodelist")) return strdup("./nodelist");
#if defined(_WIN32) && !defined(__CYGWIN__)
  tmpnam(buffer);
  nodetab_tempName=strdup(buffer);
#else /*UNIX*/
  if (getenv("HOME")) {
    sprintf(buffer,"%s/.nodelist",getenv("HOME"));
  }
#endif
  if (!probefile(buffer)) 
  {
    /*Create a simple nodelist in the user's home*/
    FILE *f=fopen(buffer,"w");
    if (f==NULL) {
      fprintf(stderr,"ERROR> Cannot create a 'nodelist' file.\n");
      exit(1);
    }
    fprintf(f,"group main\nhost localhost\n");
    fclose(f);
  }
  return strdup(buffer);
}

typedef struct nodetab_host {
  char    *name;  /*Host DNS name*/
  skt_ip_t ip; /*IP address of host*/
  pathfixlist pathfixes;
  char    *ext;  /*FIXME: What the heck is this?  OSL 9/8/00*/
  int      cpus;  /* # of physical CPUs*/
  int      rank;  /*Rank of this CPU*/
  double   speed; /*Relative speed of each CPU*/
  int      nice;  /* process priority */
  int      forks; /* number of processes to fork on remote node */
  /*These fields are set during node-startup*/
  int     dataport;/*UDP port number*/
  SOCKET  ctrlfd;/*Connection to control port*/
#if CMK_USE_RSH
  char    *shell;  /*Rsh to use*/
  char    *debugger ; /*Debugger to use*/
  char    *xterm ;  /*Xterm to use*/
  char    *login;  /*User login name to use*/
  char    *passwd;  /*User login password*/
  char    *setup;  /*Commands to execute on login*/
#endif

#if CMK_USE_IBVERBS
  ChInfiAddr *qpData;
#endif
#if CMK_USE_IBUD
  ChInfiAddr qp;
#endif


} nodetab_host;

nodetab_host **nodetab_table;
int           nodetab_max;
int           nodetab_size;
int          *nodetab_rank0_table;
int           nodetab_rank0_size;

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
int                     loaded_max_pe;
#endif

void nodetab_reset(nodetab_host *h)
{
  h->name="SET_H->NAME";
  h->ip=_skt_invalid_ip;
  h->pathfixes = 0;
  h->ext = NULL;
  h->speed = 1.0;
  h->cpus = 1;
  h->rank = 0;
  h->nice=-100;
  h->forks = 0;
  h->dataport=-1;
  h->ctrlfd=-1;
#if CMK_USE_RSH
  h->shell = arg_shell;
  h->debugger = arg_debugger;
  h->xterm = arg_xterm;
  h->login = arg_mylogin;
  h->passwd = "*";
  h->setup = "*";
#endif
}

void nodetab_add(nodetab_host *h)
{
  if (h->rank == 0)
    nodetab_rank0_table[nodetab_rank0_size++] = nodetab_size;
  nodetab_table[nodetab_size] = (nodetab_host *) malloc(sizeof(nodetab_host));

  if (arg_verbose) {
    char ips[200];
    skt_print_ip(ips,h->ip);
    printf("Charmrun> adding client %d: \"%s\", IP:%s\n", nodetab_size, h->name, ips);
  }

  *nodetab_table[nodetab_size++] = *h;
}

void nodetab_makehost(char *name,nodetab_host *h)
{
  h->name=strdup(name);
  h->ip = skt_innode_lookup_ip(name);
  if (skt_ip_match(h->ip,_skt_invalid_ip)) {
#ifdef CMK_BPROC
    /* only the master node is used */
    if (!(1 == arg_requested_pes && atoi(name)==-1))
#endif
    {
    fprintf(stderr,"ERROR> Cannot obtain IP address of %s\n", name);
    exit(1);
    }
  }
  if (nodetab_size == nodetab_max) return;
  nodetab_add(h);
}

char *nodetab_args(char *args,nodetab_host *h)
{
  if (arg_ppn>0) h->cpus = arg_ppn;
  while(*args != 0) {
    char *b1 = skipblanks(args), *e1 = skipstuff(b1);
    char *b2 = skipblanks(e1), *e2 = skipstuff(b2);
    while (*b1=='+') b1++;/*Skip over "++" on parameters*/
#if CMK_USE_RSH
    if (subeqs(b1,e1,"login")) h->login = substr(b2,e2);
    else if (subeqs(b1,e1,"passwd")) h->passwd = substr(b2,e2);
    else if (subeqs(b1,e1,"setup")) h->setup = strdup(b2);
    else if (subeqs(b1,e1,"shell")) h->shell = substr(b2,e2);
    else if (subeqs(b1,e1,"debugger")) h->debugger = substr(b2,e2);
    else if (subeqs(b1,e1,"xterm")) h->xterm = substr(b2,e2);
    else 
#endif
    if (subeqs(b1,e1,"speed")) h->speed = atof(b2);
    else if (subeqs(b1,e1,"cpus")) {
      if (arg_ppn==0) h->cpus = atol(b2);       /* ignore if there is ++ppn */
    }
    else if (subeqs(b1,e1,"pathfix")) {
      char *b3 = skipblanks(e2), *e3 = skipstuff(b3);
      args = skipblanks(e3);
      h->pathfixes=pathfix_append(substr(b2,e2),substr(b3,e3),h->pathfixes);
      e2 = e3;       /* for the skipblanks at the end */
    } 
    else if (subeqs(b1,e1,"ext"))  h->ext = substr(b2,e2);
    else if (subeqs(b1,e1,"nice"))  h->nice = atoi(b2);
    else return args;
    args = skipblanks(e2);
  }
#if CMK_SHARED_VARS_UNAVAILABLE
  if (h->cpus != 1) {
    fprintf(stderr,"Warning> Invalid cpus %d in nodelist ignored.\n", h->cpus);
    h->cpus = 1;
  }
#endif
  return args;
}

/*  setup nodetab as localhost only */
void nodetab_init_for_local()
{
  int tablesize, i, done=0;
  nodetab_host group;

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
    if(arg_read_pes == 0){
        arg_read_pes = arg_requested_pes;
    }
#endif

  tablesize = arg_requested_pes;
  nodetab_table=(nodetab_host**)malloc(tablesize*sizeof(nodetab_host*));
  nodetab_rank0_table=(int*)malloc(tablesize*sizeof(int));
  nodetab_max=tablesize;

  nodetab_reset(&group);
  if (arg_ppn==0) arg_ppn=1;
#if CMK_SHARED_VARS_UNAVAILABLE
  if (arg_ppn > 1) {
    fprintf(stderr,"Warning> Invalid ppn %d in nodelist ignored.\n", arg_ppn);
    arg_ppn=1;
  }
#endif
  group.cpus = arg_ppn;
  i = 0;
  while (!done) {
    char *hostname = "127.0.0.1";
    for (group.rank = 0; group.rank<arg_ppn; group.rank++) {
      nodetab_makehost(hostname, &group);
      if (++i == arg_requested_pes) { done = 1; break; }
    }
  }
}


#ifdef HSTART
/* Sets the parent field of hosts to point to their parent charmrun. The root charmrun will create children for all hosts which are parent of at least one other host*/
int branchfactor;
int nodes_per_child;
int * nodetab_unique_table;
int nodetab_unique_size;
char  *nodetab_name(int i);
void nodetab_init_hierarchical_start(void)
{
        int node_start = 0;
        char * node_name;
        nodetab_unique_size = 0;
        nodetab_unique_table = (int *)malloc(nodetab_rank0_size * sizeof(int));
        while(node_start<nodetab_rank0_size)
        {
                        nodetab_unique_table[nodetab_unique_size++] = node_start;
                        node_name = nodetab_name(node_start);
                        do{
                                        node_start++;
            }
                        while(node_start<nodetab_rank0_size&&(!strcmp(nodetab_name(node_start),node_name)));
                                
        }
        branchfactor = ceil(sqrt(nodetab_unique_size));
        nodes_per_child = round(nodetab_unique_size*1.0/branchfactor);
}
#endif

void nodetab_init()
{
  FILE *f,*fopen();
  char *nodesfile; 
  nodetab_host global,group,host;
  char input_line[MAX_LINE_LENGTH];
  int rightgroup, basicsize, i, remain;
  
  /* if arg_local is set, ignore the nodelist file */
  if (arg_local || arg_mpiexec) {
    nodetab_init_for_local();
    goto fin;
  }

  /* Open the NODES_FILE. */
  nodesfile = nodetab_file_find();
  if(arg_verbose)
    fprintf(stderr, "Charmrun> using %s as nodesfile\n", nodesfile);
  if (!(f = fopen(nodesfile,"r"))) {
    fprintf(stderr,"ERROR> Cannot read %s: %s\n",nodesfile,strerror(errno));
    exit(1);
  }
 
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
        if(arg_read_pes == 0){
        arg_read_pes = arg_requested_pes;
    }
        nodetab_table=(nodetab_host**)malloc(arg_read_pes*sizeof(nodetab_host*));
        nodetab_rank0_table=(int*)malloc(arg_read_pes*sizeof(int));
        nodetab_max=arg_read_pes;
    fprintf(stderr,"arg_read_pes %d arg_requested_pes %d\n",arg_read_pes,arg_requested_pes);
#else
        nodetab_table=(nodetab_host**)malloc(arg_requested_pes*sizeof(nodetab_host*));
        nodetab_rank0_table=(int*)malloc(arg_requested_pes*sizeof(int));
        nodetab_max=arg_requested_pes;
#endif
 
  
  nodetab_reset(&global);
  group=global;
  rightgroup = (strcmp(arg_nodegroup,"main")==0);
  
  while(fgets(input_line,sizeof(input_line)-1,f)!=0) {
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
        if (nodetab_size == arg_read_pes) break;
#else
    if (nodetab_size == arg_requested_pes) break;
#endif
    if (input_line[0]=='#') continue;
    zap_newline(input_line);
        if (!nodetab_args(input_line,&global)) {
                /*An option line-- also add options to current group*/
                nodetab_args(input_line,&group);
        }
        else {/*Not an option line*/
                char *b1 = skipblanks(input_line), *e1 = skipstuff(b1);
                char *b2 = skipblanks(e1), *e2 = skipstuff(b2);
                char *b3 = skipblanks(e2);
                if (subeqs(b1,e1,"host")) {
                        if (rightgroup) {
                                host=group;
                                nodetab_args(b3,&host);
                                for (host.rank=0; host.rank<host.cpus; host.rank++)
                                        nodetab_makehost(substr(b2,e2),&host);
                        }
                } else if (subeqs(b1,e1, "group")) {
                        group=global;
                        nodetab_args(b3,&group);
                        rightgroup = subeqs(b2,e2,arg_nodegroup);
                } else if (b1!=b3) {
                        fprintf(stderr,"ERROR> unrecognized command in nodesfile:\n");
                        fprintf(stderr,"ERROR> %s\n", input_line);
                        exit(1);
                }
        }
  }
  fclose(f);
  if (nodetab_tempName!=NULL) unlink(nodetab_tempName);

  /*Wrap nodes in table around if there aren't enough yet*/
  basicsize = nodetab_size;
  if (basicsize==0) {
    fprintf(stderr,"ERROR> No hosts in group %s\n", arg_nodegroup);
    exit(1);
  }
  while ((nodetab_size < arg_requested_pes)&&(arg_requested_pes!=MAX_NODES))
     nodetab_add(nodetab_table[nodetab_size%basicsize]);
  
fin:
  /*Clip off excess CPUs at end*/
  for (i=0; i<nodetab_size; i++) {
    if (nodetab_table[i]->rank == 0)
      remain = nodetab_size - i;
    if (nodetab_table[i]->cpus > remain)
      nodetab_table[i]->cpus = remain;
  }

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
        loaded_max_pe = arg_requested_pes-1;
#endif
#ifdef HSTART
        if(arg_hierarchical_start) 
                nodetab_init_hierarchical_start();              
#endif

}

/* Given a processor number, look up the nodetab info: */
nodetab_host *nodetab_getinfo(int i)
{
  if (nodetab_table==0) {
    fprintf(stderr,"ERROR> Node table not initialized.\n");
    exit(1);
  }
  return nodetab_table[i];
}

/* Given a node number, look up the nodetab info: */
nodetab_host *nodetab_getnodeinfo(int i)
{
  return nodetab_getinfo(nodetab_rank0_table[i]);
}

/*These routines all take *PE* numbers (NOT node numbers!)*/
char        *nodetab_name(int i)     { return nodetab_getinfo(i)->name; }
pathfixlist  nodetab_pathfixes(int i){ return nodetab_getinfo(i)->pathfixes; }
char        *nodetab_ext(int i)      { return nodetab_getinfo(i)->ext; }
skt_ip_t     nodetab_ip(int i)       { return nodetab_getinfo(i)->ip; }
unsigned int nodetab_cpus(int i)     { return nodetab_getinfo(i)->cpus; }
unsigned int nodetab_rank(int i)     { return nodetab_getinfo(i)->rank; }
int          nodetab_dataport(int i) { return nodetab_getinfo(i)->dataport; }
int          nodetab_nice(int i)     { return nodetab_getinfo(i)->nice; }
SOCKET      nodetab_ctrlfd(int i)    { return nodetab_getinfo(i)->ctrlfd;}
#if CMK_USE_RSH
char        *nodetab_setup(int i)    { return nodetab_getinfo(i)->setup; }
char        *nodetab_shell(int i)    { return nodetab_getinfo(i)->shell; }
char        *nodetab_debugger(int i) { return nodetab_getinfo(i)->debugger; }
char        *nodetab_xterm(int i)    { return nodetab_getinfo(i)->xterm; }
char        *nodetab_login(int i)    { return nodetab_getinfo(i)->login; }
char        *nodetab_passwd(int i)   { return nodetab_getinfo(i)->passwd; }
#endif

/****************************************************************************
 *
 * Nodeinfo
 *
 * The global list of node PEs, IPs, and port numbers.
 * Stored in ChMachineInt_t format so the table can easily be sent
 * back to the nodes.
 *
 ****************************************************************************/

static ChNodeinfo *nodeinfo_arr;/*Indexed by node number.*/

void nodeinfo_allocate(void)
{
        nodeinfo_arr=(ChNodeinfo *)malloc(nodetab_rank0_size*sizeof(ChNodeinfo));
}
void nodeinfo_add(const ChSingleNodeinfo *in,SOCKET ctrlfd)
{
        int node=ChMessageInt(in->nodeNo);
        ChNodeinfo i=in->info;
        unsigned int nt;
        unsigned int pe;
        unsigned int dataport;
        int lid,qpn,psn;
        if (node<0 || node>=nodetab_rank0_size)
                {fprintf(stderr,"Unexpected node %d registered!\n",node);exit(1);}
        nt=nodetab_rank0_table[node];/*Nodetable index for this node*/  
        i.nPE=ChMessageInt_new(nodetab_cpus(nt));
        if (arg_mpiexec)
           nodetab_getinfo(nt)->ip = i.IP;   /* get IP */
        i.IP=nodetab_ip(nt);
#if CMK_USE_IBVERBS 
        nodeinfo_arr[node] = i;
        for (pe=0;pe<nodetab_cpus(nt);pe++){
                nodetab_table[nt+pe]->ctrlfd=ctrlfd;
        }
        /* printf("Charmrun> client %d connected\n", nt); */
#else
        dataport = ChMessageInt(i.dataport);
        if (0==dataport)
                {fprintf(stderr,"Node %d could not initialize network!\n",node);exit(1);}
        nodeinfo_arr[node]=i;
        for (pe=0;pe<nodetab_cpus(nt);pe++)
          {
            nodetab_table[nt+pe]->dataport=dataport;
            nodetab_table[nt+pe]->ctrlfd=ctrlfd;
#if CMK_USE_IBUD
            nodetab_table[nt+pe]->qp=i.qp;
#endif
          }
        if (arg_verbose) {
          char ips[200];
          skt_print_ip(ips,nodetab_ip(nt));
          printf("Charmrun> client %d connected (IP=%s data_port=%d)\n", nt, ips, dataport);
#if CMK_USE_IBUD
          printf("Charmrun> client %d lid=%d qpn=%i psn=%i\n",nt,ChMessageInt(i.qp.lid),ChMessageInt(i.qp.qpn),ChMessageInt(i.qp.psn));
#endif
        }
#endif
}

/****************************************************************************
 *
 * input handling
 *
 * You can use this module to read the standard input.  It supports
 * one odd function, input_scanf_chars, which is what makes it useful.
 * if you use this module, you may not read stdin yourself.
 *
 * void input_init(void)
 * char *input_gets(void)
 * char *input_scanf_chars(char *fmt)
 *
 ****************************************************************************/

char *input_buffer;

void input_extend()
{
  char line[1024];
  int len = input_buffer?strlen(input_buffer):0;
  fflush(stdout);
  if (fgets(line, 1023, stdin)==0) { 
    fprintf(stderr,"end-of-file on stdin");
    exit(1);
  }
  input_buffer = realloc(input_buffer, len + strlen(line) + 1);
  strcpy(input_buffer+len, line);
}

void input_init()
{
  input_buffer = strdup("");
}

char *input_extract(nchars)
    int nchars;
{
  char *res = substr(input_buffer, input_buffer+nchars);
  char *tmp = substr(input_buffer+nchars, input_buffer+strlen(input_buffer));
  free(input_buffer);
  input_buffer = tmp;
  return res;
}

char *input_gets()
{
  char *p, *res; int len;
  while(1) {
    p = strchr(input_buffer,'\n');
    if (p) break;
    input_extend();
  }
  len = p-input_buffer;
  res = input_extract(len+1);
  res[len]=0;
  return res;
}

/*FIXME: I am terrified by this routine. OSL 9/8/00*/
char *input_scanf_chars(fmt)
    char *fmt;
{
  char buf[8192]; int len, pos;
  static int fd; static FILE *file;
  fflush(stdout);
  if (file==0) {
#if CMK_USE_MKSTEMP
    char tmp[128];
    strcpy(tmp, "/tmp/fnordXXXXXX");
    mkstemp(tmp);
#else
    char *tmp=tmpnam(NULL);/*This was once /tmp/fnord*/
#endif
    unlink(tmp);
    fd = open(tmp,O_RDWR | O_CREAT | O_TRUNC, 0664);
    if (fd<0) { 
      fprintf(stderr,"cannot open temp file /tmp/fnord");
      exit(1);
    }
    file = fdopen(fd, "r+");
    unlink(tmp);
  }
  while (1) {
    len = strlen(input_buffer);
    rewind(file);
    fwrite(input_buffer, len, 1, file);
    fflush(file);
    rewind(file);
    ftruncate(fd, len);
    fscanf(file, fmt, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf, buf);
    pos = ftell(file);
    if (pos<len) break;
    input_extend();
  }
  return input_extract(pos);
}

/***************************************************************************
CCS Interface:
  Charmrun forwards CCS requests on to the node-programs' control
sockets.
***************************************************************************/

#if CMK_CCS_AVAILABLE

/*The Ccs Server socket became active-- 
rec'v the message and respond to the request,
by forwarding the request to the appropriate node.
 */
void req_ccs_connect(void)
{
  const void *bufs[3]; int lens[3];
  struct {
   ChMessageHeader ch;/*Make a charmrun header*/
   CcsImplHeader hdr;/*Ccs internal header*/
  } h;
  void *reqData;/*CCS request data*/
  int pe,reqBytes;
  if (0==CcsServer_recvRequest(&h.hdr,&reqData))
    return;/*Malformed request*/
  pe=ChMessageInt(h.hdr.pe);
  reqBytes=ChMessageInt(h.hdr.len);

  if (pe == -1) {
    /*Treat -1 as broadcast and sent to 0 as root of the spanning tree*/
    pe = 0;
  }
  if ((pe<=-nodetab_size || pe>=nodetab_size) && 0==replay_single) {
    /*Treat out of bound values as errors. Helps detecting bugs*/
    /* But when virtualized with Bigemulator, we can have more pes than nodetabs */
    /* TODO: We should somehow check boundaries also for bigemulator... */
#if ! CMK_BIGSIM_CHARM
    if (pe==-nodetab_size) fprintf(stderr,"Invalid processor index in CCS request: are you trying to do a broadcast instead?");
    else fprintf(stderr,"Invalid processor index in CCS request.");
    CcsServer_sendReply(&h.hdr,0,0);
    free(reqData);
    return;
#endif
  }
  else if (pe < -1) {
    /*Treat negative values as multicast to a number of processors specified by -pe.
      The pes to multicast to follows sits at the beginning of reqData*/
    reqBytes -= pe * sizeof(ChMessageInt_t);
    pe = ChMessageInt(*(ChMessageInt_t*)reqData);
  }
  
  if (! check_stdio_header(&h.hdr)) {

#define LOOPBACK 0
#if LOOPBACK /*Immediately reply "there's nothing!" (for performance testing)*/
  CcsServer_sendReply(&h.hdr,0,0);
#else
    int destpe = pe;
#if CMK_BIGSIM_CHARM
    destpe = destpe % nodetab_size;
#endif
    if (replay_single) destpe = 0;
    /*Fill out the charmrun header & forward the CCS request*/
    ChMessageHeader_new("req_fw",sizeof(h.hdr)+reqBytes,&h.ch);  

    bufs[0]=&h; lens[0]=sizeof(h);
    bufs[1]=reqData; lens[1]=reqBytes;
    skt_sendV(nodetab_ctrlfd(destpe),2,bufs,lens);

#endif
  }
  free(reqData);
}

/*
Forward the CCS reply (if any) from this client back to the 
original network requestor, on the original request socket.
 */
int req_ccs_reply_fw(ChMessage *msg,SOCKET srcFd) {
  int len=msg->len; /* bytes of data remaining to receive */
  
  /* First pull down the CCS header sent by the client. */
  CcsImplHeader hdr;
  skt_recvN(srcFd,&hdr,sizeof(hdr)); len-=sizeof(hdr);
  
#define m (4*1024) /* packets of message to recv/send at once */
  if (len<m || hdr.attr.auth) 
  { /* short or authenticated message: grab the whole thing first */
     void *data=malloc(len);
     skt_recvN(srcFd,data,len);
     CcsServer_sendReply(&hdr,len,data);
     free(data);
  } 
  else 
  { /* long messages: packetize (for pipelined sending; a 2x bandwidth improvement!) */
    ChMessageInt_t outLen;
    int destFd; /* destination for data */
    skt_abortFn old=skt_set_abort(reply_abortFn);
    int destErrs=0;
    
    destFd=ChMessageInt(hdr.replyFd);
    outLen=ChMessageInt_new(len);
    skt_sendN(destFd,&outLen,sizeof(outLen)); /* first comes the length */
    while(len>0) {
       char buf[m];
       int r=m; if (r>len) r=len;
       skt_recvN(srcFd,buf,r);
       if (0==destErrs) /* don't keep sending to dead clients, but *do* clean out srcFd */
          destErrs|=skt_sendN(destFd,buf,r);
       len-=m;
#undef m
    }
    skt_close(destFd);
  
    skt_set_abort(old); 
  }
  return 0;
}

#else
int req_ccs_reply_fw(ChMessage *msg,SOCKET srcFd) {
  
}
#endif /*CMK_CCS_AVAILABLE*/

/****************************************************************************
 *
 * REQUEST SERVICER
 *
 * The request servicer accepts connections on a TCP port.  The client
 * sends a sequence of commands (each is one line).  It then closes the
 * connection.  The server must then contact the client, sending replies.
 *
 ****************************************************************************/
/*#define __FAULT__*/

SOCKET *req_clients; /*TCP request sockets for each node*/
#ifdef HSTART
SOCKET *charmrun_fds;
#endif
int  req_nClients;/*Number of entries in above list (==nodetab_rank0_size)*/
int             req_ending=0;

/* socket and std streams for the gdb info program */
int gdb_info_pid=0;
int gdb_info_std[3];
FILE *gdb_stream=NULL;

#define REQ_OK 0
#define REQ_FAILED -1

#ifdef HSTART
int req_reply_child(SOCKET fd, char *type, 
              const char *data, int dataLen)
{

                int status = req_reply(fd, type, data, dataLen);
                if(status != REQ_OK) return status;
            SOCKET clientFd ;
                skt_recvN(fd, (const char *)&clientFd, sizeof(SOCKET));
            skt_sendN(fd, (const char *)&clientFd, sizeof(fd)); 
                return status;
}
#endif

int req_reply(SOCKET fd, char *type, 
              const char *data, int dataLen)
{
        ChMessageHeader msg;
        if (fd == INVALID_SOCKET) return REQ_FAILED;
        ChMessageHeader_new(type,dataLen,&msg);
        skt_sendN(fd,(const char *)&msg,sizeof(msg));
        skt_sendN(fd,data,dataLen);
        return REQ_OK;
}

/* Request handlers:
When a client asks us to do something, these are the
routines that actually respond to the request.
*/
/*Stash this new node's control and data ports.
 */
int req_handle_initnode(ChMessage *msg,SOCKET fd)
{
#if CMK_USE_IBVERBS
  int i;
  ChSingleNodeinfo *nodeInfo = (ChSingleNodeinfo *)msg->data;
//      printf("Charmrun> msg->len %d sizeof(ChSingleNodeinfo) %d sizeof(ChInfiAddr) %d \n",msg->len,sizeof(ChSingleNodeinfo),sizeof(ChInfiAddr));
  if(msg->len != sizeof(ChSingleNodeinfo) + (nodetab_rank0_size-1)*sizeof(ChInfiAddr)){
    fprintf(stderr,"Charmrun: Bad initnode data length. Aborting\n");
        exit(1);
  }
  nodeInfo->info.qpList = malloc(sizeof(ChInfiAddr)*(nodetab_rank0_size-1));
  memcpy((char *)nodeInfo->info.qpList,&msg->data[sizeof(ChSingleNodeinfo)],sizeof(ChInfiAddr)*(nodetab_rank0_size-1));
/*      for(i=0;i<nodetab_rank0_size-1;i++){
                printf("i %d  0x%0x 0x%0x 0x%0x\n",i,ChMessageInt(nodeInfo->info.qpList[i].lid),ChMessageInt(nodeInfo->info.qpList[i].qpn),ChMessageInt(nodeInfo->info.qpList[i].psn));
        }*/
#else
  if (msg->len!=sizeof(ChSingleNodeinfo)) {
    fprintf(stderr,"Charmrun: Bad initnode data length. Aborting\n");
    fprintf(stderr,"Charmrun: possibly because: %s.\n", msg->data);
    exit(1);
  }
#endif  
  nodeinfo_add((ChSingleNodeinfo *)msg->data,fd);
  return REQ_OK;
}

int req_handle_initnodetab(ChMessage *msg,SOCKET fd)
{
        ChMessageHeader hdr;
        ChMessageInt_t nNodes=ChMessageInt_new(nodetab_rank0_size);
        ChMessageHeader_new("initnodetab",sizeof(ChMessageInt_t)+
                            sizeof(ChNodeinfo)*nodetab_rank0_size,&hdr);
        skt_sendN(fd,(const char *)&hdr,sizeof(hdr));
        skt_sendN(fd,(const char *)&nNodes,sizeof(nNodes));
        skt_sendN(fd,(const char *)nodeinfo_arr,
                  sizeof(ChNodeinfo)*nodetab_rank0_size);
                        
        return REQ_OK;
}

#ifdef HSTART
/* Used for fault tolerance with hierarchical start */
int req_handle_initnodetab1(ChMessage *msg,SOCKET fd)
{
        ChMessageHeader hdr;
        ChMessageInt_t nNodes=ChMessageInt_new(nodetab_rank0_size);
        ChMessageHeader_new("initnttab",sizeof(ChMessageInt_t)+
                            sizeof(ChNodeinfo)*nodetab_rank0_size,&hdr);
        skt_sendN(fd,(const char *)&hdr,sizeof(hdr));
        skt_sendN(fd,(const char *)&nNodes,sizeof(nNodes));
        skt_sendN(fd,(const char *)nodeinfo_arr,
                  sizeof(ChNodeinfo)*nodetab_rank0_size);
        
        return REQ_OK;
}
/*Get the array of node numbers, IPs, and ports.
This is used by the node-programs to talk to one another.
*/
static int parent_charmrun_fd = -1;
int req_handle_initnodedistribution(ChMessage *msg,SOCKET fd, int client)
{
        int nodes_to_fork = nodes_per_child; /* rounding should help in better load distribution*/
        int rank0_start = nodetab_unique_table[client*nodes_per_child];
        int rank0_finish;
        if(client == branchfactor -1)
        {
                nodes_to_fork = nodetab_unique_size- client*nodes_per_child;
                rank0_finish = nodetab_rank0_size;
        }
        else
                        rank0_finish = nodetab_unique_table[client*nodes_per_child + nodes_to_fork];
        int k;
        ChMessageInt_t * nodemsg = (ChMessageInt_t *)malloc((rank0_finish - rank0_start)*sizeof(ChMessageInt_t)); 
        for(k =0; k <rank0_finish- rank0_start; k++)
                        nodemsg[k] = ChMessageInt_new(nodetab_rank0_table[rank0_start+k]);
        ChMessageHeader hdr;
        ChMessageInt_t nNodes=ChMessageInt_new(rank0_finish- rank0_start);
        ChMessageInt_t nTotalNodes=ChMessageInt_new(nodetab_rank0_size);
        ChMessageHeader_new("initnodetab",sizeof(ChMessageInt_t)*2+
                            sizeof(ChMessageInt_t)*(rank0_finish- rank0_start),&hdr);
        skt_sendN(fd,(const char *)&hdr,sizeof(hdr));
        skt_sendN(fd,(const char *)&nNodes,sizeof(nNodes));
        skt_sendN(fd,(const char *)&nTotalNodes,sizeof(nTotalNodes));
        skt_sendN(fd,(const char *)nodemsg,(rank0_finish- rank0_start)*sizeof(ChMessageInt_t));
        free(nodemsg);          
        return REQ_OK;
}

ChSingleNodeinfo * myNodesInfo;
int     send_myNodeInfo_to_parent()
{
        ChMessageHeader hdr;
        ChMessageInt_t nNodes=ChMessageInt_new(nodetab_rank0_size);
        ChMessageHeader_new("initnodetab",sizeof(ChMessageInt_t)+
                            sizeof(ChSingleNodeinfo)*nodetab_rank0_size,&hdr);
        skt_sendN(parent_charmrun_fd,(const char *)&hdr,sizeof(hdr));
        skt_sendN(parent_charmrun_fd,(const char *)&nNodes,sizeof(nNodes));
        skt_sendN(parent_charmrun_fd,(const char *)myNodesInfo,
                  sizeof(ChSingleNodeinfo)*nodetab_rank0_size);

        return REQ_OK;
}
void forward_nodetab_to_children()
{
        /*it just needs to receive and copy the nodetab info if required and send it as it is to its nodes */   
if (!skt_select1(parent_charmrun_fd,1200*1000)){
        exit(0);
        }
ChMessage msg;
ChMessage_recv(parent_charmrun_fd,&msg);

ChMessageInt_t * nodelistmsg = (ChMessageInt_t *)msg.data;                      
int nodetab_Nodes = ChMessageInt(nodelistmsg[0]);
int client;
for (client=0;client<nodetab_rank0_size;client++)       {
        SOCKET fd = req_clients[client];
        ChMessageHeader hdr;
        ChMessageInt_t nNodes=ChMessageInt_new(nodetab_Nodes);
        ChMessageHeader_new("initnodetab",sizeof(ChMessageInt_t)+
                            sizeof(ChNodeinfo)*nodetab_Nodes,&hdr);
        skt_sendN(fd,(const char *)&hdr,sizeof(hdr));
        skt_sendN(fd,(const char *)&nNodes,sizeof(nNodes));
        skt_sendN(fd,(const char *)(nodelistmsg+1),
                  sizeof(ChNodeinfo)*nodetab_Nodes);
}
}
/*Parent Charmrun receives the nodetab from child and processes it. msg contain array of ChSingleNodeInfo*/
void receive_nodeset_from_child(ChMessage *msg, SOCKET fd)
{
        ChMessageInt_t * n32 = (ChMessageInt_t *)msg->data;     
    int numOfNodes =ChMessageInt(n32[0]); 
    ChSingleNodeinfo *childNodeInfo = (ChSingleNodeinfo*) (n32+1);
        int k;
        for(k = 0; k<numOfNodes; k++)
                        nodeinfo_add(childNodeInfo+k,fd);
}

void set_sockets_list(ChMessage *msg, SOCKET fd)
{
        ChMessageInt_t * n32 = (ChMessageInt_t *)msg->data;     
    int node_start =ChMessageInt(n32[0]); 
        charmrun_fds[node_start/nodes_per_child] = fd;
}
#endif
/* Check this return code from "printf". */
static void checkPrintfError(int err) {
  if (err<0) {
    static int warned=0;
    if (!warned) {
      perror("charmrun WARNING> error in printf");
      warned=1;
    }
  }
}

int req_handle_print(ChMessage *msg,SOCKET fd)
{
  checkPrintfError(printf("%s",msg->data));
  checkPrintfError(fflush(stdout));
  write_stdio_duplicate(msg->data);
  return REQ_OK;
}


int req_handle_printerr(ChMessage *msg,SOCKET fd)
{
  fprintf(stderr,"%s",msg->data);
  fflush(stderr);
  write_stdio_duplicate(msg->data);
  return REQ_OK;
}


int req_handle_printsyn(ChMessage *msg,SOCKET fd)
{
  checkPrintfError(printf("%s",msg->data));
  checkPrintfError(fflush(stdout));
  write_stdio_duplicate(msg->data);
#ifdef HSTART
  if(arg_hierarchical_start) 
                  req_reply_child(fd, "printdone", "", 1);
  else
#endif
  req_reply(fd, "printdone", "", 1);
  return REQ_OK;
}


int req_handle_printerrsyn(ChMessage *msg,SOCKET fd)
{
  fprintf(stderr,"%s",msg->data);
  fflush(stderr);
  write_stdio_duplicate(msg->data);
#ifdef HSTART
 if(arg_hierarchical_start) 
                  req_reply_child(fd, "printdone", "", 1);
else
#endif
  req_reply(fd, "printdone", "", 1);
  return REQ_OK;
}


int req_handle_ending(ChMessage *msg,SOCKET fd)
{  
  int i;
  req_ending++;

#if (!defined(_FAULT_MLOG_) && !defined(_FAULT_CAUSAL_))    
        if (req_ending == nodetab_size)
#else
        if(req_ending == arg_requested_pes)
#endif
  {
    for (i=0;i<req_nClients;i++)
      skt_close(req_clients[i]);
    if (arg_verbose) printf("Charmrun> Graceful exit.\n");
    exit(0);
  }
  return REQ_OK;
}

int req_handle_barrier(ChMessage *msg,SOCKET fd)
{
  int i;
  static int barrier_count = 0;
  static int barrier_phase = 0;
  barrier_count ++;
#ifdef HSTART
  if (barrier_count == arg_requested_pes) 
#else
  if (barrier_count == req_nClients) 
#endif
  {
    barrier_count = 0;
        barrier_phase ++;
        for (i=0;i<req_nClients;i++)
          if (REQ_OK != req_reply(req_clients[i], "barrier", "", 1))
          {
                fprintf(stderr, "req_handle_barrier socket error: %d\n", i);
        abort();
          }
  }
  return REQ_OK;
}

int req_handle_barrier0(ChMessage *msg,SOCKET fd)
{
  int i;
  static int count = 0;
  static SOCKET fd0;
  int pe = atoi(msg->data);
  if (pe == 0) fd0 = fd;
  count ++;
#ifdef HSTART
  if (count == arg_requested_pes) 
#else
  if (count == req_nClients) 
#endif
  {
        req_reply(fd0, "barrier0", "", 1);     /* only send to node 0 */
        count = 0;
  }
  return REQ_OK;
}


int req_handle_abort(ChMessage *msg,SOCKET fd)
{
  /*fprintf(stderr,"req_handle_abort called \n");*/
  if (msg->len==0) 
        fprintf(stderr,"Aborting!\n");
  else
        fprintf(stderr, "%s\n", msg->data);
  exit(1);
}

int req_handle_scanf(ChMessage *msg,SOCKET fd)
{
  char *fmt, *res, *p;

  fmt = msg->data;
  fmt[msg->len-1]=0;
  res = input_scanf_chars(fmt);
  p = res; while (*p) { if (*p=='\n') *p=' '; p++; }
#ifdef HSTART
  if(arg_hierarchical_start) 
          req_reply_child(fd, "scanf-data", res, strlen(res)+1);
  else
#endif
  req_reply(fd, "scanf-data", res, strlen(res)+1);
  free(res);
  return REQ_OK;
}

#ifdef __FAULT__        
void restart_node(int crashed_node);
void reconnect_crashed_client(int socket_index,int crashed_node);
void announce_crash(int socket_index,int crashed_node);

static int _last_crash = 0;                     /* last crashed pe number */
static int _crash_socket_index = 0;             /* last restart socket */
#ifdef HSTART
static int _crash_socket_charmrun_index = 0;            /* last restart socket */
int crashed_pe_id;
int restarted_pe_id;
#endif
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
static int numCrashes=0;  /*number of crashes*/
static SOCKET last_crashed_fd=-1;
#endif

int req_handle_crashack(ChMessage *msg,SOCKET fd)
{
        static int count = 0;
        count ++;
#ifdef HSTART
        if(arg_hierarchical_start)
                        {
                  if (count == nodetab_rank0_size-1) {
                        /* only after everybody else update its nodetab, can this
                           restarted process continue */
                        printf("Charmrun> continue node: %d\n", _last_crash);
                        req_handle_initnodetab1(NULL,req_clients[_crash_socket_charmrun_index]);
                        _last_crash = 0;
                        count = 0;
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
        last_crashed_fd=-1;
#endif
            }
                }
                
    else
                        
#endif
        if (count == req_nClients-1) {
                // only after everybody else update its nodetab, can this restarted process continue 
                printf("Charmrun> continue node: %d\n", _last_crash);
                req_handle_initnodetab(NULL,req_clients[_crash_socket_index]);  
                _last_crash = 0;
                count = 0;
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
                        last_crashed_fd=-1;
#endif
        }
}

#ifdef HSTART
/* send initnode to root*/
                int set_crashed_socket_id(ChMessage *msg,SOCKET fd)
                {
                        ChSingleNodeinfo *nodeInfo = (ChSingleNodeinfo *)msg->data;
                        int nt=nodetab_rank0_table[ChMessageInt(nodeInfo->nodeNo)-mynodes_start];
                        nodeInfo->nodeNo = ChMessageInt_new(nt);
                        /* Required for CCS */
                /*Nodetable index for this node*/
                        int pe;
                        for (pe=0;pe<nodetab_cpus(nt);pe++)
                          {
                                nodetab_table[nt+pe]->ctrlfd=fd;
                          }
                }

/* Receives new dataport of restarted prcoess   and resends nodetable to everyone*/
                int req_handle_crash(ChMessage *msg,SOCKET fd)
                {

                        ChMessageInt_t oldpe, newpe;
                        skt_recvN(fd, (const char *)&oldpe, sizeof(oldpe));
                        skt_recvN(fd, (const char *)&newpe, sizeof(newpe));
                        *nodetab_table[ChMessageInt(oldpe)] = *nodetab_table[ChMessageInt(newpe)]; 

                        int status = req_handle_initnode(msg,fd);
                        int i;
                        for(i=0;i<req_nClients;i++){
                                if(req_clients[i] == fd){
                                break;
                                }
                        }       
                        _crash_socket_charmrun_index = i;

                        fprintf(stdout,"Root charmrun : Socket %d failed %d\n",fd, _crash_socket_charmrun_index);
                        fflush(stdout);
                        ChSingleNodeinfo *nodeInfo = (ChSingleNodeinfo *)msg->data;
                        int crashed_node = ChMessageInt(nodeInfo->nodeNo);
                        _last_crash = crashed_node;
                        switch (status) 
                        {
        case REQ_OK: break;
                case REQ_FAILED: 
            return REQ_FAILED;
                         }

                        /* Already processed, so send*/
                        int client;
                        for (client=0;client<req_nClients;client++)     {                       
                                req_handle_initnodetab(NULL,req_clients[client]);
                        }
                        
                        /*Anounce crash to all child charmruns*/
                        announce_crash(nodetab_rank0_size+1,crashed_node );

                }

#endif
#endif

#ifdef __FAULT__
void error_in_req_serve_client(SOCKET fd){
        SOCKET * new_req_clients=(SOCKET *)malloc((req_nClients-1)*sizeof(SOCKET));
        int count=0,i;
        int crashed_node,crashed_pe,node_index,socket_index;
        fprintf(stdout,"Socket %d failed \n",fd);

        
#ifdef HSTART
        if(arg_hierarchical_start)
                        {
                        for(i=mynodes_start;i<mynodes_start+nodetab_rank0_size;i++){
                                if(nodetab_ctrlfd(i) == fd){
                                        break;
                                }
                        }
                        }

                        else
#endif
                        for(i=0;i<nodetab_max;i++){
                                if(nodetab_ctrlfd(i) == fd){
                                        break;
                                }
                        }
                
        fflush(stdout);
#if (!defined(_FAULT_MLOG_) && !defined(_FAULT_CAUSAL_))
        skt_close(fd);
#endif
        crashed_pe = i;
        node_index = i-nodetab_rank(crashed_pe);
        for(i=0;i<nodetab_rank0_size;i++){
                if(node_index == nodetab_rank0_table[i]){
                        break;
                }
        }
        crashed_node = i;
        
        /*announce_crash(socket_index,crashed_node);*/
        restart_node(crashed_node);

        fprintf(stdout,"charmrun says Processor %d failed on Node %d\n",crashed_pe,crashed_node);
        for(i=0;i<req_nClients;i++){
                if(req_clients[i] == fd){
                        break;
                }
        }       
        socket_index = i;
        reconnect_crashed_client(socket_index,crashed_node);
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
        skt_close(fd);
#endif
}
#endif

int req_handler_dispatch(ChMessage *msg,SOCKET replyFd)
{
  char *cmd=msg->header.type;
  int recv_status;
  DEBUGF(("Got request '%s'\n",cmd,replyFd));
#if CMK_CCS_AVAILABLE   /* CCS *doesn't* want data yet, for faster forwarding */
  if (strcmp(cmd,"reply_fw")==0)   return req_ccs_reply_fw(msg,replyFd);
#endif

  /* grab request data */
  recv_status = ChMessageData_recv(replyFd,msg);
#ifdef __FAULT__
#ifdef HSTART
        if(!arg_hierarchical_start)
#endif
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
 if(recv_status < 0){
        if(replyFd == last_crashed_fd){
            return REQ_OK;
        }
        DEBUGF(("recv_status %d on socket %d \n",recv_status,replyFd));
        error_in_req_serve_client(replyFd);
    }
#else   
  if(recv_status < 0)  error_in_req_serve_client(replyFd);
#endif
#endif

       if (strcmp(cmd,"ping")==0)       return REQ_OK;
  else if (strcmp(cmd,"print")==0)      return req_handle_print(msg,replyFd);
  else if (strcmp(cmd,"printerr")==0)   return req_handle_printerr(msg,replyFd);
  else if (strcmp(cmd,"printsyn")==0)  return req_handle_printsyn(msg,replyFd);
  else if (strcmp(cmd,"printerrsyn")==0) return req_handle_printerrsyn(msg,replyFd);
  else if (strcmp(cmd,"scanf")==0)      return req_handle_scanf(msg,replyFd);
  else if (strcmp(cmd,"barrier")==0)    return req_handle_barrier(msg,replyFd);
  else if (strcmp(cmd,"barrier0")==0)   return req_handle_barrier0(msg,replyFd);
  else if (strcmp(cmd,"ending")==0)     return req_handle_ending(msg,replyFd);
  else if (strcmp(cmd,"abort")==0)      return req_handle_abort(msg,replyFd);
#ifdef __FAULT__        
  else if (strcmp(cmd,"crash_ack")==0)   return req_handle_crashack(msg,replyFd);
#ifdef HSTART
  else if (strcmp(cmd,"initnode")==0)   return req_handle_crash(msg,replyFd);
#endif
#endif
  else {
#ifndef __FAULT__       
        fprintf(stderr,"Charmrun> Bad control socket request '%s'\n",cmd); 
        abort();
        return REQ_OK;
#endif                          
  }
  return REQ_OK;
}

void req_serve_client(SOCKET fd)
{
  int recv_status;
  int status;
  ChMessage msg;
  DEBUGF(("Getting message from client...\n"));
  recv_status = ChMessageHeader_recv(fd,&msg);
#ifdef __FAULT__        
#ifdef HSTART
  if(!arg_hierarchical_start && recv_status < 0) error_in_req_serve_client(fd);
#else
  if(recv_status < 0) error_in_req_serve_client(fd);
#endif
#endif
        
  DEBUGF(("Message is '%s'\n",msg.header.type));
  status = req_handler_dispatch(&msg,fd);
  switch (status) 
  {
    case REQ_OK: break;
    case REQ_FAILED: 
        fprintf(stderr,"Charmrun> Error processing control socket request %s\n",msg.header.type); 
        abort();
        break;
  }
  ChMessage_free(&msg);
}

#ifdef HSTART
void req_forward_root(SOCKET fd)
{
  int recv_status;
  int status;
  ChMessage msg;
  recv_status = ChMessage_recv(fd,&msg);

  char *cmd=msg.header.type;

#ifdef __FAULT__
        if(recv_status < 0)
  {
                  error_in_req_serve_client(fd);
                  return;
  }

        /*called from reconnect_crashed_client */
        if (strcmp(cmd,"initnode")==0)
  {
                set_crashed_socket_id(&msg,fd);
        }
#endif

  if (strcmp(cmd,"ping")!=0)
  {
                  status = req_reply(parent_charmrun_fd, cmd, msg.data,ChMessageInt(msg.header.len));

  if (strcmp(cmd,"scanf")==0 || strcmp(cmd,"printsyn")==0 || strcmp(cmd,"printerrsyn")==0)
                  skt_sendN(parent_charmrun_fd,(const char *)&fd, sizeof(fd)); 
  
#ifdef __FAULT__
  if (strcmp(cmd,"initnode")==0)
  {
        ChMessageInt_t oldpe=ChMessageInt_new(crashed_pe_id);
        ChMessageInt_t newpe=ChMessageInt_new(restarted_pe_id);
        skt_sendN(parent_charmrun_fd,(const char *)&oldpe, sizeof(oldpe)); 
    skt_sendN(parent_charmrun_fd,(const char *)&newpe, sizeof(newpe)); 
  }
#endif
  }


  switch (status) 
  {
    case REQ_OK: break;
    case REQ_FAILED:
        abort();
        break;
  }
  ChMessage_free(&msg);
}

void req_forward_client()
{
  int recv_status;
  int status;
  ChMessage msg;
  recv_status = ChMessage_recv(parent_charmrun_fd,&msg);
  if(recv_status < 0)
      {
          int i;
          for (i=0;i<req_nClients;i++)
              skt_close(req_clients[i]);
          exit(0);
      }
    
  char *cmd=msg.header.type;

    if(strcmp(cmd, "barrier") ==0){
                        int i;
                for (i=0;i<req_nClients;i++)
      if (REQ_OK != req_reply(req_clients[i],cmd,msg.data,ChMessageInt(msg.header.len)))
           {
                abort();
      }
                return;
        }
#ifdef __FAULT__
        if(strcmp(cmd, "initnodetab") ==0){
                if(_last_crash ==0 ) 
                        current_restart_phase++;
                int i;
                for (i=0;i<req_nClients;i++)
                        if(_last_crash==0 || i !=_crash_socket_index)
                                if (REQ_OK != req_reply(req_clients[i],cmd,msg.data,ChMessageInt(msg.header.len))){
                                        abort();
                                }
                return;
        }

        if(strcmp(cmd, "crashnode") ==0){

                int i;
                for (i=0;i<req_nClients;i++)
                        if(_last_crash==0 || i !=_crash_socket_index)
      if (REQ_OK != req_reply(req_clients[i],cmd,msg.data,ChMessageInt(msg.header.len)))
           {
                abort();
      }
                return;
        }
        if(strcmp(cmd, "initnttab") ==0){
                        _last_crash = 0;
            if (REQ_OK != req_reply(req_clients[_crash_socket_index],"initnodetab",msg.data,ChMessageInt(msg.header.len)))
           {
                abort();
      }
                return;
        }

#endif

  SOCKET fd;

  /* CCS forward request */
  if(strcmp(cmd,"req_fw") ==0)
                  {
                         CcsImplHeader * hdr =(CcsImplHeader *)msg.data;
                         int pe=ChMessageInt(hdr->pe);
                         fd = nodetab_table[pe]->ctrlfd;
                  }     
                  else if(strcmp(cmd, "barrier0") ==0)
                  {
                        fd = nodetab_table[0]->ctrlfd;
                  }
    else
                  skt_recvN(parent_charmrun_fd, (const char *)&fd,sizeof(SOCKET));
                  
                  status = req_reply(fd, cmd, msg.data,ChMessageInt(msg.header.len));

                  switch (status) 
                  {
                        case REQ_OK: break;
                        case REQ_FAILED:
                                abort();
                                break;
                  }
                  ChMessage_free(&msg);
}

#endif  

int ignore_socket_errors(int c,const char *m)
{  /*Abandon on further socket errors during error shutdown*/
                  
#ifndef __FAULT__       
                  exit(2);
#endif  
                        return -1;
}

/*A socket went bad somewhere!  Immediately disconnect,
which kills everybody.
*/
int socket_error_in_poll(int code,const char *msg)
{
                /*commenting it for fault tolerance*/
                /*ifdef it*/

        int i;
        skt_set_abort(ignore_socket_errors);
        fprintf(stderr,"Charmrun: error on request socket--\n"
                        "%s\n",msg);
#ifndef __FAULT__                       
        for (i=0;i<req_nClients;i++)
                skt_close(req_clients[i]);
        exit(1);
#endif  
        ftTimer = GetClock();
        return -1;
}

#if CMK_USE_POLL /*poll() version*/
# define CMK_PIPE_DECL(maxn,delayMs) \
        static struct pollfd *fds = NULL; \
        int nFds_sto=0; int *nFds=&nFds_sto; \
        int pollDelayMs=delayMs; \
        if (fds == NULL) fds = (struct pollfd *)malloc((maxn) * sizeof(struct pollfd));
# define CMK_PIPE_SUB fds,nFds
# define CMK_PIPE_CALL() poll(fds, *nFds, pollDelayMs); *nFds=0

# define CMK_PIPE_PARAM struct pollfd *fds,int *nFds
# define CMK_PIPE_ADDREAD(rd_fd) \
        do {fds[*nFds].fd=rd_fd; fds[*nFds].events=POLLIN; (*nFds)++;} while(0)
# define CMK_PIPE_ADDWRITE(wr_fd) \
        do {fds[*nFds].fd=wr_fd; fds[*nFds].events=POLLOUT; (*nFds)++;} while(0)
# define CMK_PIPE_CHECKREAD(rd_fd) fds[(*nFds)++].revents&POLLIN
# define CMK_PIPE_CHECKWRITE(wr_fd) fds[(*nFds)++].revents&POLLOUT

#else /*select() version*/

# define CMK_PIPE_DECL(maxn, delayMs) \
        fd_set rfds_sto,wfds_sto;\
        int nFds=0;  \
        fd_set *rfds=&rfds_sto,*wfds=&wfds_sto; struct timeval tmo; \
        FD_ZERO(rfds); FD_ZERO(wfds); \
        tmo.tv_sec=delayMs/1000; tmo.tv_usec=1000*(delayMs%1000);
# define CMK_PIPE_SUB rfds,wfds
# define CMK_PIPE_CALL() select(FD_SETSIZE, rfds, 0, 0, &tmo)

# define CMK_PIPE_PARAM fd_set *rfds,fd_set *wfds
# define CMK_PIPE_ADDREAD(rd_fd) { assert(nFds<FD_SETSIZE);FD_SET(rd_fd,rfds); nFds++; }
# define CMK_PIPE_ADDWRITE(wr_fd) FD_SET(wr_fd,wfds)
# define CMK_PIPE_CHECKREAD(rd_fd) FD_ISSET(rd_fd,rfds)
# define CMK_PIPE_CHECKWRITE(wr_fd) FD_ISSET(wr_fd,wfds)
#endif

/*
Wait for incoming requests on all client sockets,
and the CCS socket (if present).
*/
void req_poll()
{
  int status,i;
  int readcount;

  CMK_PIPE_DECL(req_nClients+5, 1000);
  for (i=0;i<req_nClients;i++)
        CMK_PIPE_ADDREAD(req_clients[i]);
  if (CcsServer_fd()!=INVALID_SOCKET) CMK_PIPE_ADDREAD(CcsServer_fd());
  if (arg_charmdebug) {
    CMK_PIPE_ADDREAD(0);
    CMK_PIPE_ADDREAD(gdb_info_std[1]);
    CMK_PIPE_ADDREAD(gdb_info_std[2]);
  }

  skt_set_abort(socket_error_in_poll);

  DEBUGF(("Req_poll: Calling select...\n"));
  status=CMK_PIPE_CALL();
  DEBUGF(("Req_poll: Select returned %d...\n",status));

  if (status==0) return;/*Nothing to do-- timeout*/

  if (status<0){ 
                if (errno == EINTR || errno == EAGAIN) return;
                fflush(stdout);
                fflush(stderr);
                socket_error_in_poll(1359,"Node program terminated unexpectedly!\n");
        }
  for (i=0;i<req_nClients;i++)
        if (CMK_PIPE_CHECKREAD(req_clients[i]))
          {
            readcount=10;   /*number of successive reads we serve per socket*/
            /*This client is ready to read*/
            do { req_serve_client(req_clients[i]); readcount--;}
            while (1==skt_select1(req_clients[i],0) && readcount>0);
          }

  if (CcsServer_fd()!=INVALID_SOCKET)
         if (CMK_PIPE_CHECKREAD(CcsServer_fd())) {
                  DEBUGF(("Activity on CCS server port...\n"));
                  req_ccs_connect();
         }

  if (arg_charmdebug) {
    char buf[2048];
    if (CMK_PIPE_CHECKREAD(0)) {
      int indata = read(0, buf, 5);
      buf[indata] = 0;
      if (indata < 5) fprintf(stderr,"Error reading command (%s)\n",buf);
      if (strncmp(buf,"info:",5)==0) {
        /* Found info command, forward data to gdb info program */
        char c;
        int num=0;
        //printf("Command to be forwarded\n");
        while (read(0, &c, 1)!=-1) {
          buf[num++]=c;
          if (c=='\n' || num >= 2045) {
            write(gdb_info_std[0], buf, num);
            if (c=='\n') break;
          }
        }
      }
      //printf("Command from charmdebug: %d(%s)\n",indata,buf);
    }
    /* All streams from gdb are forwarded to the stderr stream through the FILE
       gdb_stream which has been duplicated from stderr */
    /* NOTE: gdb_info_std[2] must be flushed before gdb_info_std[1] because the
       latter contains the string "(gdb) " ending the synchronization. Also the
       std[1] should be read with the else statement. It will not work without. */
    if (CMK_PIPE_CHECKREAD(gdb_info_std[2])) {
      int indata = read(gdb_info_std[2], buf, 100);
      /*printf("read data from gdb info stderr %d\n",indata);*/
      if (indata > 0) {
        buf[indata] = 0;
        //printf("printing %s\n",buf);
        //fflush(stdout);
        //fprintf(gdb_stream,"%s",buf);
        fflush(gdb_stream);
      }
    } else if (CMK_PIPE_CHECKREAD(gdb_info_std[1])) {
      int indata = read(gdb_info_std[1], buf, 100);
      /*printf("read data from gdb info stdout %d\n",indata);*/
      if (indata > 0) {
        buf[indata] = 0;
        //printf("printing %s\n",buf);
        //fflush(stdout);
        fprintf(gdb_stream,"%s",buf);
        fflush(gdb_stream);
      }
    }
  }
}

#ifdef HSTART
                void req_poll_hierarchical()
                {
                  int status,i;
                  fd_set  rfds;
                  struct timeval tmo;
                  int readcount;

                  skt_set_abort(socket_error_in_poll);

                  tmo.tv_sec = 1;
                  tmo.tv_usec = 0;
                  FD_ZERO(&rfds); /* clears set of file descriptor */
                  for (i=0;i<req_nClients;i++)
                        FD_SET(req_clients[i],&rfds); /* adds client sockets to rfds set*/
                  if (CcsServer_fd()!=INVALID_SOCKET) FD_SET(CcsServer_fd(),&rfds);
                  if (arg_charmdebug) {
                        FD_SET(0, &rfds);
                        FD_SET(gdb_info_std[1], &rfds);
                        FD_SET(gdb_info_std[2], &rfds);
                  }

                  if(arg_child_charmrun) 
                        FD_SET(parent_charmrun_fd,&rfds); /* adds client sockets to rfds set*/
                  DEBUGF(("Req_poll: Calling select...\n"));
                  status=select(FD_SETSIZE, &rfds, 0, 0, &tmo); /* FD_SETSIZE is the maximum number of file descriptors that a fd_set object can hold information about, select returns number of polls gathered */ 
                  DEBUGF(("Req_poll: Select returned %d...\n",status));

                  if (status==0) return;/*Nothing to do-- timeout*/
                  if (status<0){ 
                                fflush(stdout);
                                fflush(stderr);
                                socket_error_in_poll(1359,"Node program terminated unexpectedly!\n");
                        }
                  for (i=0;i<req_nClients;i++)
                        if (FD_ISSET(req_clients[i],&rfds))
                          {
                                readcount=10;   /*number of successive reads we serve per socket*/
                                /*This client is ready to read*/
                                do {
                                                if(arg_child_charmrun)
                                                        req_forward_root(req_clients[i]);
                                                else
                                                        req_serve_client(req_clients[i]); 
                                                readcount--;
                                }
                                while (1==skt_select1(req_clients[i],0) && readcount>0);
                          }


                if(arg_child_charmrun)
                  //Forward from root to clients
                        if (FD_ISSET(parent_charmrun_fd,&rfds))
                          {
                                readcount=10;   /*number of successive reads we serve per socket*/
                                do{ 
                                                req_forward_client();
                                                readcount--;
                                }
                                while (1==skt_select1(parent_charmrun_fd,0) && readcount>0);
                          }

                  /*Wait to receive responses and Forward responses */
                  if (CcsServer_fd()!=INVALID_SOCKET)
                         if (FD_ISSET(CcsServer_fd(),&rfds)) {
                                  DEBUGF(("Activity on CCS server port...\n"));
                                  req_ccs_connect();
                         }

                  if (arg_charmdebug) {
                        char buf[2048];
                        if (FD_ISSET(0, &rfds)) {
                          int indata = read(0, buf, 5);
                          buf[indata] = 0;
                          if (indata < 5) fprintf(stderr,"Error reading command (%s)\n",buf);
                          if (strncmp(buf,"info:",5)==0) {
                        /* Found info command, forward data to gdb info program */
                        char c;
                        int num=0;
                        //printf("Command to be forwarded\n");
                        while (read(0, &c, 1)!=-1) {
                          buf[num++]=c;
                          if (c=='\n' || num >= 2045) {
                                write(gdb_info_std[0], buf, num);
                                if (c=='\n') break;
                          }
                        }
                          }
                          //printf("Command from charmdebug: %d(%s)\n",indata,buf);
                        }
                        /* All streams from gdb are forwarded to the stderr stream through the FILE
                           gdb_stream which has been duplicated from stderr */
                        /* NOTE: gdb_info_std[2] must be flushed before gdb_info_std[1] because the
                           latter contains the string "(gdb) " ending the synchronization. Also the
                           std[1] should be read with the else statement. It will not work without. */
                        if (FD_ISSET(gdb_info_std[2], &rfds)) {
                          int indata = read(gdb_info_std[2], buf, 100);
                          /*printf("read data from gdb info stderr %d\n",indata);*/
                          if (indata > 0) {
                        buf[indata] = 0;
                                //printf("printing %s\n",buf);
                                //fflush(stdout);
                        //fprintf(gdb_stream,"%s",buf);
                        fflush(gdb_stream);
                          }
                        } else if (FD_ISSET(gdb_info_std[1], &rfds)) {
                          int indata = read(gdb_info_std[1], buf, 100);
                          /*printf("read data from gdb info stdout %d\n",indata);*/
                          if (indata > 0) {
                        buf[indata] = 0;
                                //printf("printing %s\n",buf);
                                //fflush(stdout);
                        fprintf(gdb_stream,"%s",buf);
                        fflush(gdb_stream);
                          }
                        }
                  }
                }
#endif 
        
static unsigned int server_port;
static char server_addr[1024];/* IP address or hostname of charmrun*/
static SOCKET server_fd;

#ifdef HSTART
static skt_ip_t parent_charmrun_IP;
static int parent_charmrun_port;
static int parent_charmrun_pid;
static int dataport;
static SOCKET dataskt;
int charmrun_phase =0;
#endif 

int client_connect_problem(int code,const char *msg)
{/*Called when something goes wrong during a client connect*/

        fprintf(stderr,"Charmrun> error %d attaching to node:\n"
                "%s\n",code,msg);
        exit(1);
        return -1;
}

int errorcheck_one_client_connect(int client){
#ifdef HSTART
        /* Child charmruns are already connected - Do we need to conect again*/ 
        if(arg_hierarchical_start && !arg_child_charmrun && charmrun_phase ==1) 
            return 1;   
#endif 
        unsigned int clientPort;/*These are actually ignored*/
        skt_ip_t clientIP;
        if (arg_verbose) printf("Charmrun> Waiting for %d-th client to connect.\n",client);
        if (0==skt_select1(server_fd,arg_timeout*1000))
                client_connect_problem(client,"Timeout waiting for node-program to connect");

                
        req_clients[client]=skt_accept(server_fd,&clientIP,&clientPort);

        if (req_clients[client]==SOCKET_ERROR) 
                client_connect_problem(client,"Failure in node accept");

        skt_tcp_no_nagle(req_clients[client]);

        return 1;
};


#if CMK_C_INLINE
inline static
#endif
void read_initnode_one_client(int client){
                ChMessage msg;
                if (!skt_select1(req_clients[client],arg_timeout*1000))
                   client_connect_problem(client,"Timeout on IP request");
                ChMessage_recv(req_clients[client],&msg);
                req_handle_initnode(&msg,req_clients[client]);
                ChMessage_free(&msg);
}


#if CMK_IBVERBS_FAST_START
void req_one_client_partinit(int client){
   ChMessage partStartMsg;
         int clientNode;
                
         if(errorcheck_one_client_connect(client)){
           if (!skt_select1(req_clients[client],arg_timeout*1000))
                   client_connect_problem(client,"Timeout on partial init request");
                         
           ChMessage_recv(req_clients[client],&partStartMsg);
           clientNode =            ChMessageInt(*(ChMessageInt_t*)partStartMsg.data);
           assert(strncmp(partStartMsg.header.type,"partinit",8) == 0);
           ChMessage_free(&partStartMsg);
         } 
        
};
#endif


#ifdef HSTART
int nodeCount = 0;
/* To keep a global node numbering */
void add_singlenodeinfo_to_mynodeinfo(ChMessage * msg, SOCKET ctrlfd)
{
        /*add to myNodesInfo */                         
        ChSingleNodeinfo *nodeInfo = (ChSingleNodeinfo *)msg->data;
        
    /* need to change nodeNo */
        myNodesInfo[nodeCount].nodeNo = ChMessageInt_new(nodetab_rank0_table[ChMessageInt(nodeInfo->nodeNo)-mynodes_start]);
        myNodesInfo[nodeCount++].info = nodeInfo->info;

        /* Required for CCS */
        int nt=nodetab_rank0_table[ChMessageInt(nodeInfo->nodeNo)-mynodes_start];/*Nodetable index for this node*/
        int pe;
        for (pe=0;pe<nodetab_cpus(nt);pe++)
    {
            nodetab_table[nt+pe]->ctrlfd=ctrlfd;
        }
}
#endif

#ifndef HSTART
/* Original Function, need to check if modifications required*/ 
void req_set_client_connect(int start,int end) {
        fd_set sockset;
        ChMessage msg;
        int client,i;
        int done,maxdesc;
        int *finished;
        int curclient,curclientend,curclientstart;
        
        curclient=curclientend=curclientstart=start;

        finished=malloc((end-start)*sizeof(int));
        for(i=0;i<(end-start);i++)
                finished[i]=0;

#if CMK_USE_IBVERBS && !CMK_IBVERBS_FAST_START
        for (i=start;i<end;i++) {
                errorcheck_one_client_connect(curclientend++);
        }        
        if (req_nClients > 1) {
                /*  a barrier to make sure infiniband device gets initialized */
                for (i=start;i<end;i++) 
                        ChMessage_recv(req_clients[i],&msg);
                for (i=start;i<end;i++)
                        req_reply(req_clients[i], "barrier", "", 1);
        }
#endif

        done=0;
        while(!done) {
                /* check server socket for messages */
#if ! CMK_USE_IBVERBS || CMK_IBVERBS_FAST_START
                while(curclientstart==curclientend||skt_select1(server_fd,1)!=0) {
                        errorcheck_one_client_connect(curclientend++);
                }
#endif
                /* check appropriate clients for messages */
                for(client=curclientstart;client<curclientend;client++)
                        if(req_clients[client]>0) {
                                if(skt_select1(req_clients[client],1)!=0) {
                                        ChMessage_recv(req_clients[client],&msg);
                                        req_handle_initnode(&msg,req_clients[client]);
                                        finished[client-start]=1;
                                }
                        }


                /* test if done */
                done=1;
                for(i=curclientstart-start;i<(end-start);i++)
                        if(finished[i]==0) {
                                curclientstart=start+i;
                                done=0;
                                break;
                        }

        }
        ChMessage_free(&msg);

        free(finished);
}
#else
/*int charmrun_phase =0; meaningful for main charmun to decide what to receive*/
void req_set_client_connect(int start,int end) {
                        fd_set sockset;
                        ChMessage msg;
                        int client,i;
                        int done,maxdesc;
                        int *finished;
                        int curclient,curclientend,curclientstart;
                        
                        curclient=curclientend=curclientstart=start;

                        finished=malloc((end-start)*sizeof(int));
                        for(i=0;i<(end-start);i++)
                                finished[i]=0;

                        if(arg_child_charmrun && start==0 ) myNodesInfo = malloc(sizeof(ChSingleNodeinfo)*nodetab_rank0_size); 

#if CMK_USE_IBVERBS && !CMK_IBVERBS_FAST_START
        for (i=start;i<end;i++) {
                errorcheck_one_client_connect(curclientend++);
        }        
        if (req_nClients > 1) {
                /*  a barrier to make sure infiniband device gets initialized */
                for (i=start;i<end;i++) 
                        ChMessage_recv(req_clients[i],&msg);
                for (i=start;i<end;i++)
                        req_reply(req_clients[i], "barrier", "", 1);
        }
#endif

                        done=0;
                        while(!done) {
                                /* check server socket for messages */
#if ! CMK_USE_IBVERBS || CMK_IBVERBS_FAST_START
                while(curclientstart==curclientend||skt_select1(server_fd,1)!=0) {
                        errorcheck_one_client_connect(curclientend++);
                }
#endif
                                /* check appropriate clients for messages */
                                for(client=curclientstart;client<curclientend;client++)
                                        if(req_clients[client]>0) {
                                                if(skt_select1(req_clients[client],1)!=0) {
                                                        ChMessage_recv(req_clients[client],&msg);
                                                        if(!arg_hierarchical_start)
                                                                req_handle_initnode(&msg,req_clients[client]);
                                                        else{
                                                                        if(!arg_child_charmrun)
                                                                                {
                                                                                        if(charmrun_phase ==1) 
                                                                                                receive_nodeset_from_child(&msg, req_clients[client]);
                                                                                        else
                                                                                                set_sockets_list(&msg, req_clients[client]);    
                                                                                        //here we need to decide based upon the phase
                                                                                }
                                                                        else /* hier-start with 2nd leval*/
                                                                                add_singlenodeinfo_to_mynodeinfo(&msg,req_clients[client] );    
                                                        }
                                                        finished[client-start]=1;
                                                        }
                                                }
                                        
                                /* test if done */
                                done=1;
                                for(i=curclientstart-start;i<(end-start);i++)
                                        if(finished[i]==0) {
                                                curclientstart=start+i;
                                                done=0;
                                                break;
                                        }

                        }
                        ChMessage_free(&msg);

                        free(finished);
}
#endif
                                                
                                                
/* allow one client to connect */
void req_one_client_connect(int client)
{
        if(errorcheck_one_client_connect(client))
        { /*This client has just connected-- fetch his name and IP*/
                read_initnode_one_client(client);
        }
}

#if CMK_USE_IBVERBS
/* Each node has sent the qpn data for all the qpns it has created
   This data needs to be sent to all the other nodes
         This needs to be done for all nodes
**/
void exchange_qpdata_clients(){
        int proc,i;
        for( i=0;i<nodetab_rank0_size;i++){
                int nt=nodetab_rank0_table[i];/*Nodetable index for this node*/ 
                nodetab_table[nt]->qpData = malloc(sizeof(ChInfiAddr)*nodetab_rank0_size);
        }
        for(proc =0;proc< nodetab_rank0_size;proc++){
                int count=0;
                for(i=0;i<nodetab_rank0_size;i++){
                        if(i == proc){
                        }else{
                                int nt=nodetab_rank0_table[i];/*Nodetable index for this node*/ 
                                nodetab_table[nt]->qpData[proc] =  nodeinfo_arr[proc].qpList[count];
        //                      printf("Charmrun> nt %d proc %d lid 0x%x qpn 0x%x psn 0x%x\n",nt,proc,ChMessageInt(nodetab_table[nt]->qpData[proc].lid),ChMessageInt(nodetab_table[nt]->qpData[proc].qpn),ChMessageInt(nodetab_table[nt]->qpData[proc].psn));
                                count++;
                        }
                }
                                free(nodeinfo_arr[proc].qpList);
                        }
                };

                void    send_clients_nodeinfo_qpdata(){
                        int node;
                        int msgSize = sizeof(ChMessageInt_t)+sizeof(ChNodeinfo)*nodetab_rank0_size+sizeof(ChInfiAddr)*nodetab_rank0_size;
                        for(node=0;node<nodetab_rank0_size;node++){
                                int nt=nodetab_rank0_table[node];/*Nodetable index for this node*/
                //              printf("Charmrun> Node %d proc %d sending initnodetab \n",node,nt);
                                ChMessageHeader hdr;
                                ChMessageInt_t nNodes=ChMessageInt_new(nodetab_rank0_size);
                                ChMessageHeader_new("initnodetab",msgSize,&hdr);
                                skt_sendN(nodetab_table[nt]->ctrlfd,(const char *)&hdr,sizeof(hdr));
                                skt_sendN(nodetab_table[nt]->ctrlfd,(const char *)&nNodes,sizeof(nNodes));
                                skt_sendN(nodetab_table[nt]->ctrlfd,(const char *)nodeinfo_arr,sizeof(ChNodeinfo)*nodetab_rank0_size);
                                skt_sendN(nodetab_table[nt]->ctrlfd,(const char *)&nodetab_table[nt]->qpData[0],sizeof(ChInfiAddr)*nodetab_rank0_size);                         
                        }
                }
#endif

struct timeval tim;
#define  getthetime(x) gettimeofday(&tim,NULL); x = tim.tv_sec + (tim.tv_usec/1000000.0);
#define getthetime1(x) gettimeofday(&tim,NULL); x = tim.tv_sec ;
/*Wait for all the clients to connect to our server port*/
void req_client_connect(void)
{
        int client;
#ifdef HSTART
        if(!arg_hierarchical_start)
#endif
        nodeinfo_allocate();
        req_nClients=nodetab_rank0_size;
        req_clients=(SOCKET *)malloc(req_nClients*sizeof(SOCKET));
        for(client=0;client<req_nClients;client++)
                req_clients[client]=-1;
                        
        skt_set_abort(client_connect_problem);
                        
#if CMK_IBVERBS_FAST_START
        for (client=0;client<req_nClients;client++){
                req_one_client_partinit(client);
        }
        for (client=0;client<req_nClients;client++){
                read_initnode_one_client(client);
        }
#else

        req_set_client_connect(0,req_nClients);

#endif
                        
        if (portOk == 0) exit(1);
        if (arg_verbose) printf("Charmrun> All clients connected.\n");
#if CMK_USE_IBVERBS
        exchange_qpdata_clients();
        send_clients_nodeinfo_qpdata();
#else
#ifdef HSTART
        if(arg_hierarchical_start) {
          /* first we need to send data to parent charmrun and then send the nodeinfo to the clients*/
          send_myNodeInfo_to_parent();
          /*then receive from root */
          forward_nodetab_to_children();
        }

        else 
#endif
        for (client=0;client<req_nClients;client++)     {                       
                req_handle_initnodetab(NULL,req_clients[client]);
        }
                        
#endif
        if (arg_verbose) printf("Charmrun> IP tables sent.\n");
}

/*Wait for all the clients to connect to our server port, then collect and send nodetable to all */
#ifdef HSTART
void req_charmrun_connect(void)
{
        //      double t1, t2, t3, t4;
        int client;
        nodeinfo_allocate();
        req_nClients=branchfactor;
        req_clients=(SOCKET *)malloc(req_nClients*sizeof(SOCKET));
        charmrun_fds=(SOCKET *)malloc(req_nClients*sizeof(SOCKET));
        for(client=0;client<req_nClients;client++)
                req_clients[client]=-1;
                        
        skt_set_abort(client_connect_problem);
                        
#if CMK_IBVERBS_FAST_START
        for (client=0;client<req_nClients;client++){
                req_one_client_partinit(client);
        }
        for (client=0;client<req_nClients;client++){
                read_initnode_one_client(client);
        }
#else
//if(!arg_child_charmrun) getthetime(t1);

        req_set_client_connect(0,req_nClients);
//if(!arg_child_charmrun)       getthetime(t2);         /* also need to process received nodesets JIT */
#endif
        
        if (portOk == 0) exit(1);
        if (arg_verbose) printf("Charmrun> All clients connected.\n");
#if CMK_USE_IBVERBS
        exchange_qpdata_clients();
        send_clients_nodeinfo_qpdata();
#else
        for (client=0;client<req_nClients;client++)     {
                        // add flag to check what leval charmrun it is and what phase
                req_handle_initnodedistribution(NULL, charmrun_fds[client], client);
        }
//getthetime(t3);

        /* Now receive the nodetab from child charmruns*/
        charmrun_phase = 1;
                        
        skt_set_abort(client_connect_problem);

        req_set_client_connect(0,req_nClients);

        /* Already processed, so send*/
        for (client=0;client<req_nClients;client++)     {                       
                req_handle_initnodetab(NULL,req_clients[client]);
        }
//if(!arg_child_charmrun) getthetime(t4);
#endif
        if (arg_verbose) printf("Charmrun> IP tables sent.\n");
//if(!arg_child_charmrun) printf("Time for charmruns connect= %f , sending nodes to fire= %f, node clients connected= %f n ", t2-t1, t3-t2, t4-t3);
}

#endif

#ifndef CMK_BPROC

void start_one_node_rsh(int rank0no);
void finish_one_node(int rank0no);
void finish_set_nodes(int start, int stop);



void req_client_start_and_connect(void)
{
        int client, c;
        int batch = arg_batch_spawn;        /* fire several at a time */
        int clientgroup,clientstart;
        int counter;

#ifdef HSTART
        if(!arg_hierarchical_start)
#endif
        nodeinfo_allocate();
        req_nClients=nodetab_rank0_size;
        req_clients=(SOCKET *)malloc(req_nClients*sizeof(SOCKET));
        
        skt_set_abort(client_connect_problem);

        client=0;
        while(client<req_nClients) { /* initiate a batch */
                clientstart=client;

                for(counter=0;counter<batch;counter++) { /* initiate batch number of nodes */
                        clientgroup=start_set_node_rsh(client);
                        client+=clientgroup;
                        if(client>=req_nClients) {
                                client=req_nClients;
                                break;
                        }
                }
#if CMK_USE_RSH
                  /* ssh x11 forwarding will make sure ssh exit */
                if (!arg_ssh_display) 
#endif
                finish_set_nodes(clientstart,client);

#if CMK_IBVERBS_FAST_START
                for (c=clientstart;c<client;c++) { 
                        req_one_client_partinit(c);
                }
#else
                req_set_client_connect(clientstart,client);
#endif                                          
        }


#if CMK_IBVERBS_FAST_START
        for (client=0;client<req_nClients;client++){
                read_initnode_one_client(client);
        }
#endif
        if (portOk == 0) exit(1);
        if (arg_verbose) printf("Charmrun> All clients connected.\n");

#if CMK_USE_IBVERBS
        exchange_qpdata_clients();      
        send_clients_nodeinfo_qpdata();
#else
#ifdef HSTART
        if(arg_hierarchical_start) {
                                /* first we need to send data to parent charmrun and then send the nodeinfo to the clients*/
                        send_myNodeInfo_to_parent();
                        /*then receive from root */
                        forward_nodetab_to_children();
                        }

                        else
#endif
                        for (client=0;client<req_nClients;client++)     {                       
                                req_handle_initnodetab(NULL,req_clients[client]);
                        }
                        

#endif    
        if (arg_verbose) printf("Charmrun> IP tables sent.\n");
        free(rsh_pids); /* done with rsh_pids */
}

#endif

/*Start the server socket the clients will connect to.*/
void req_start_server(void)
{
  skt_ip_t ip=skt_innode_my_ip();
  if (arg_local)
          /* local execution, use localhost always */
        strcpy(server_addr, "127.0.0.1");
  else if (arg_charmrunip != NULL)
          /* user specify the IP at +useip */
        strcpy(server_addr, arg_charmrunip);
  else if ( (arg_charmrunip = getenv ("CHARMRUN_IP")) != NULL)
          /* user specify the env  */
        strcpy(server_addr, arg_charmrunip);
  else if (skt_ip_match(ip,_skt_invalid_ip)) {
          printf("Charmrun> Warning-- cannot find IP address for your hostname.  Using loopback.\n");
          strcpy(server_addr, "127.0.0.1");
  }
  else if (arg_usehostname || skt_ip_match(ip,skt_lookup_ip("127.0.0.1")))
          /*Use symbolic host name as charmrun address*/
        gethostname(server_addr,sizeof(server_addr));
  else 
        skt_print_ip(server_addr,ip);

  server_port = 0;
  server_fd=skt_server(&server_port);

  if (arg_verbose) {
        printf("Charmrun> Charmrun = %s, port = %d\n", server_addr, server_port);
  }
                  
#if CMK_CCS_AVAILABLE
#ifdef HSTART
  if(!arg_hierarchical_start || (arg_hierarchical_start && !arg_child_charmrun))
#endif
  if(arg_server == 1) CcsServer_new(NULL,&arg_server_port,arg_server_auth);
#endif
}

#ifdef HSTART
int unique_node_start;
                /* Function copied from machine.c file */
                void parse_netstart(void)
                {
                  char *ns;
                  int nread;
                  int port;
                  ns = getenv("NETSTART");
                  if (ns!=0) 
                  {/*Read values set by Charmrun*/
                                char parent_charmrun_name[1024];
                                nread = sscanf(ns, "%d%s%d%d%d",
                                                 &unique_node_start,
                                                 parent_charmrun_name, &parent_charmrun_port,
                                                 &parent_charmrun_pid, &port);
                        parent_charmrun_IP=skt_lookup_ip(parent_charmrun_name);
                        mynodes_start = nodetab_unique_table[unique_node_start]; /*Works only when init_hierarchical called in child charmrun*/


                                if (nread!=5) {
                                                fprintf(stderr,"Error parsing NETSTART '%s'\n",ns);
                                                exit(1);
                                }
                  } 
#if CMK_USE_IBVERBS | CMK_USE_IBUD
                        char *cmi_num_nodes = getenv("CmiNumNodes");
                        if(cmi_num_nodes != NULL){
                                sscanf(cmi_num_nodes,"%d",&_Cmi_numnodes);
                        }
#endif  
                }

                int nodetab_rank0_size_total;
                /* Receive nodes for which I am responsible*/
                void my_nodetab_store(ChMessage *msg)
                {
                        ChMessageInt_t * nodelistmsg = (ChMessageInt_t *)msg->data;     
                        nodetab_rank0_size = ChMessageInt(nodelistmsg[0]);
                        nodetab_rank0_size_total =  ChMessageInt(nodelistmsg[1]);
                        int k;
                        for(k =0; k<nodetab_rank0_size ; k++)
                        {
                                nodetab_rank0_table[k] = ChMessageInt(nodelistmsg[k+2]);
                        }
                }


/* In hierarchical startup, this function is used by child charmrun to obtains the list of nodes for which it is responsible */
                void nodelist_obtain(void)
                {
                  ChMessage nodelistmsg; /* info about all nodes*/
                  /*Contact charmrun for machine info.*/

#if CMK_USE_IBVERBS
                        {
                /*              int qpListSize = (_Cmi_numnodes-1)*sizeof(ChInfiAddr);
                                me.info.qpList = malloc(qpListSize);
                                copyInfiAddr(me.info.qpList);
                                MACHSTATE1(3,"me.info.qpList created and copied size %d bytes",qpListSize);
                                ctrl_sendone_nolock("initnode",(const char *)&me,sizeof(me),(const char *)me.info.qpList,qpListSize);
                                free(me.info.qpList);
                */      }
#else
                        ChMessageHeader hdr;
                        ChMessageInt_t node_start=ChMessageInt_new(unique_node_start);
                        ChMessageHeader_new("initnodetab",sizeof(ChMessageInt_t),&hdr);
                        skt_sendN(parent_charmrun_fd,(const char *)&hdr,sizeof(hdr));
                        skt_sendN(parent_charmrun_fd,(const char *)&node_start,sizeof(node_start));

                 #endif //CMK_USE_IBVERBS

                  
                        /*We get the other node addresses from a message sent
                          back via the charmrun control port.*/
                        if (!skt_select1(parent_charmrun_fd,1200*1000)){
                        exit(0);
                        }
                        ChMessage_recv(parent_charmrun_fd,&nodelistmsg);
                  
                  my_nodetab_store(&nodelistmsg);
                  ChMessage_free(&nodelistmsg);
                }


                void init_mynodes(void)
                {
                parse_netstart();
                if (!skt_ip_match(parent_charmrun_IP,_skt_invalid_ip)) {
                        dataskt=skt_server(&dataport);
                        parent_charmrun_fd = skt_connect(parent_charmrun_IP, parent_charmrun_port, 1800);
                          } else {
                        parent_charmrun_fd=-1;
                  }

                  nodelist_obtain();
                }
#endif

/****************************************************************************
 *
 *  The Main Program
 *
 ****************************************************************************/
void start_nodes_daemon(void);
void start_nodes_rsh(void);
void start_nodes_mpiexec();
#ifdef HSTART
void start_next_level_charmruns(void);
#endif
#if CMK_BPROC
void nodetab_init_for_scyld(void);
void start_nodes_scyld(void);
#endif
void start_nodes_local(char **envp);
void kill_nodes(void);
void open_gdb_info(void);
void read_global_segments_size(void);

static void fast_idleFn(void) {sleep(0);}
void finish_nodes(void);

int main(int argc, char **argv, char **envp)
{
  srand(time(0)); 
  skt_init();
  skt_set_idle(fast_idleFn);
  /* CrnSrand((int) time(0)); */
  /* notify charm developers that charm is in use */

#ifdef HSTART
  if(!arg_child_charmrun)
#endif
  ping_developers();    
  /* Compute the values of all constants */
  arg_init(argc, argv);
  if(arg_verbose) fprintf(stderr, "Charmrun> charmrun started...\n");
  start_timer = GetClock();
#if CMK_BPROC
  /* check scyld configuration */
  if (arg_nodelist)
    nodetab_init();
  else
    nodetab_init_for_scyld();
#else
  /* Initialize the node-table by reading nodesfile */
  nodetab_init();
#endif

/* Start the server port */
  req_start_server();
  
  /* Initialize the IO module */
  input_init();
  
#ifdef HSTART
  /* Hierarchical startup*/
  if(arg_child_charmrun)
  {
        init_mynodes(); /* contacts root charmrun and gets list  of nodes to start*/
   }
#endif
 /* start the node processes */
  if (0!=getenv("CONV_DAEMON"))
        start_nodes_daemon();
  else
#if CMK_BPROC
    start_nodes_scyld();
#else
#if CMK_USE_IBVERBS
  printf("Charmrun> IBVERBS version of charmrun\n");
#endif

#ifdef HSTART
/* Hierarchical-startup*/
        if(arg_hierarchical_start) {                    
                        if (!arg_local) {
                                if(!arg_child_charmrun){        
                                  start_next_level_charmruns();}
                                else {
                                        if (!arg_batch_spawn)
                                          start_nodes_rsh();
                                        else 
                                          req_client_start_and_connect();
                                        }
                                }
                        else
                          start_nodes_local(envp);
                }

/* Normanl startup*/    
        else 

#endif
  {
    if (!arg_local) {
      if (!arg_batch_spawn) {
        if (arg_mpiexec)
          start_nodes_mpiexec();
        else
          start_nodes_rsh();
      }
      else
        req_client_start_and_connect();
    }
    else
      start_nodes_local(envp);
  }
#endif

  if (arg_charmdebug) {
#if (defined(_WIN32) && !defined(__CYGWIN__)) || CMK_BPROC
        /* Gdb stream (and charmdebug) currently valid only with rsh subsystem */
        fprintf(stderr, "Charmdebug is supported currently only with the rsh subsystem\n");
        abort(); 
#else
        /* Open an additional connection to node 0 with a gdb to grab info */
        printf("opening connection with node 0 for info gdb\n");
        read_global_segments_size();
        open_gdb_info();
        gdb_stream = fdopen(dup(2), "a");
        dup2(1, 2);
#endif
                  }


if(arg_verbose) fprintf(stderr, "Charmrun> node programs all started\n");

/* Wait for all clients to connect */
#ifdef HSTART
 /* Hierarchical startup*/
 if(arg_hierarchical_start) {
#if !CMK_RSH_KILL
                  if (!arg_batch_spawn || (!arg_child_charmrun)) finish_nodes();
#endif

                  if(!arg_child_charmrun)                       
                                  req_charmrun_connect();
                  else if (!arg_batch_spawn)
                                req_client_connect();
 }
 /* Normal startup*/
 else 
#endif
 {
#if !CMK_RSH_KILL
  if (!arg_batch_spawn) finish_nodes();
#endif
  if (!arg_batch_spawn) req_client_connect();
 }
#if CMK_RSH_KILL
  kill_nodes();
#endif
  if(arg_verbose) fprintf(stderr, "Charmrun> node programs all connected\n");
    /* report time */
  fprintf(stderr, "Charmrun> started all node programs in %.3f seconds.\n", GetClock()-start_timer);

  /* enter request-service mode */
#ifdef HSTART
if(arg_hierarchical_start)
                while (1) req_poll_hierarchical();
else
#endif
  while (1) req_poll();

}

/*This little snippet creates a NETSTART 
environment variable entry for the given node #.
It uses the idiotic "return reference to static buffer"
string return idiom.
*/
char *create_netstart(int node)
{
  static char dest[1024];
  int port=0;
  if (arg_mpiexec)
    sprintf(dest,"$CmiMyNode %s %d %d %d",server_addr,server_port,getpid()&0x7FFF, port);
  else
    sprintf(dest,"%d %s %d %d %d",node,server_addr,server_port,getpid()&0x7FFF, port);
  return dest;
}

/* The remainder of charmrun is only concerned with starting all
the node-programs, also known as charmrun clients.  We have to
start nodetab_rank0_size processes on the remote machines.
*/

/*Ask the converse daemon running on each machine to start the node-programs.*/
void start_nodes_daemon(void)
{
  taskStruct task;
  char argBuffer[5000];/*Buffer to hold assembled program arguments*/
  int i,nodeNumber;

  /*Set the parts of the task structure that will be the same for all nodes*/
  /*Figure out the command line arguments (same for all PEs)*/
  argBuffer[0]=0;
  for (i=0;arg_argv[i];i++) 
  {
    if (arg_verbose)
      printf("Charmrun> packing arg: %s\n", arg_argv[i]);
    strcat(argBuffer," ");
    strcat(argBuffer,arg_argv[i]);
  }
  
  task.magic=ChMessageInt_new(DAEMON_MAGIC);

/*Start up the user program, by sending a message
  to PE 0 on each node.*/
  for (nodeNumber=0;nodeNumber<nodetab_rank0_size;nodeNumber++)
  {
    char nodeArgBuffer[5000];/*Buffer to hold assembled program arguments*/
    char *argBuf;
    char* arg_nodeprog_r, *arg_currdir_r;
    char statusCode='N';/*Default error code-- network problem*/
    int fd;
    int pe0=nodetab_rank0_table[nodeNumber];
    
    arg_currdir_r = pathfix(arg_currdir_a, nodetab_pathfixes(nodeNumber));
        strcpy(task.cwd,arg_currdir_r);

    arg_nodeprog_r = pathextfix(arg_nodeprog_a, nodetab_pathfixes(nodeNumber), nodetab_ext(nodeNumber));
    strcpy(task.pgm,arg_nodeprog_r);

if (arg_verbose)
  printf("Charmrun> Starting node program %d on '%s' as %s.\n",nodeNumber,nodetab_name(pe0), arg_nodeprog_r);

sprintf(task.env,"NETSTART=%s",create_netstart(nodeNumber));

if (nodetab_nice(nodeNumber) != -100) {
  if(arg_verbose) fprintf(stderr, "Charmrun> +nice %d\n", nodetab_nice(nodeNumber));
                  sprintf(nodeArgBuffer, "%s +nice %d", argBuffer, nodetab_nice(nodeNumber));
                  argBuf = nodeArgBuffer;
                }
                else 
                  argBuf = argBuffer;
                task.argLength=ChMessageInt_new(strlen(argBuf));

                /*Send request out to remote node*/
                fd = skt_connect(nodetab_ip(pe0),
                         DAEMON_IP_PORT,30);
                if (fd!=INVALID_SOCKET)
                {/*Contact!  Ask the daemon to start the program*/
                  skt_sendN(fd, (const char *)&task, sizeof(task));
                  skt_sendN(fd, (const char *)argBuf, strlen(argBuf));
                  skt_recvN(fd, &statusCode,sizeof(char));
                }
                if (statusCode!='G')
                {/*Something went wrong--*/
                  fprintf(stderr,"Error '%c' starting remote node program on %s--\n%s\n",
                          statusCode,nodetab_name(pe0),daemon_status2msg(statusCode));
                  exit(1);
                } else if (arg_verbose)
                  printf("Charmrun> Node program %d started.\n",nodeNumber);
                  }
                }

#if defined(_WIN32) && !defined(__CYGWIN__)
/*Sadly, interprocess communication on Win32 is quite
  different, so we can't use Rsh on win32 yet.  
  Fall back to the daemon.*/
void start_nodes_rsh() {start_nodes_daemon();}
void finish_nodes(void) {}
void start_one_node_rsh(int rank0no) {}
void finish_one_node(int rank0no) {}
void start_nodes_mpiexec() {}

int start_set_node_rsh(int client) { return 0; }
void finish_set_nodes(int start, int stop) {}

void envCat(char *dest,LPTSTR oldEnv)
{
  char *src=oldEnv;
  dest+=strlen(dest);//Advance to end of dest
  dest++;//Advance past terminating NULL character
  while ((*src)!='\0') {
    int adv=strlen(src)+1;//Length of newly-copied string plus NULL
    strcpy(dest,src);//Copy another environment string
    dest+=adv;//Advance past newly-copied string and NULL
    src+=adv;//Ditto for src
  }
  *dest='\0';//Paste on final terminating NULL character
  FreeEnvironmentStrings(oldEnv);
}


/* simple version of charmrun that avoids the rshd or charmd,   */
/* it spawn the node program just on local machine using exec. */
void start_nodes_local(char ** env)
{
  int ret, i;
  PROCESS_INFORMATION pi;     /* process Information for the process spawned */
  char **p;

  char environment[10000];/*Doubly-null terminated environment strings*/
  char cmdLine[10000];/*Program command line, including executable name*/
/*Command line too long.*/
/*
  if (strlen(pparam_argv[1])+strlen(args) > 10000) 
        return 0; 
*/
  strcpy(cmdLine,pparam_argv[1]);
  p = pparam_argv+2;
  while ((*p)!='\0') {
    strcat(cmdLine," ");
    strcat(cmdLine,*p);
    p++;
  }

                  for (i=0; i<arg_requested_pes; i++)
                  {
                        STARTUPINFO si={0};         /* startup info for the process spawned */

                        sprintf(environment, "NETSTART=%s",  create_netstart(i));
                        /*Paste all system environment strings */
                        envCat(environment,GetEnvironmentStrings());
                  
                        /* Initialise the security attributes for the process 
                         to be spawned */
                        si.cb = sizeof(si);   
                        if (arg_verbose)
                          printf("Charmrun> start %d node program on localhost.\n", i);

                        ret = CreateProcess(NULL,       /* application name */
                                        cmdLine,        /* command line */
                                        NULL,/*&sa,*/           /* process SA */
                                        NULL,/*&sa,*/           /* thread SA */
                                        FALSE,  /* inherit flag */
#if 1
                                        CREATE_NEW_PROCESS_GROUP|DETACHED_PROCESS, 
#else
                                        CREATE_NEW_PROCESS_GROUP|CREATE_NEW_CONSOLE,
#endif
                                        /* creation flags */
                                        environment,            /* environment block */
                                        ".",                    /* working directory */
                                        &si,                    /* startup info */
                                        &pi);
                 
                        if (ret==0)
                        {
                          /*Something went wrong!  Look up the Windows error code*/
                /*
                          int error=GetLastError();
                          char statusCode=daemon_err2status(error);
                          fprintf(logfile,"******************* ERROR *****************\n"
                                  "Error in creating process!\n"
                                  "Error code = %ld-- %s\n\n\n", error,
                                  daemon_status2msg(statusCode));
                          fflush(logfile);
                */
                          int error=GetLastError();
                          printf("startProcess failed to start process \"%s\" with status: %d\n", pparam_argv[1], error);
                          exit(1) ;
                        } 
                  }
                }

#elif CMK_BPROC

                int bproc_nodeisup(int node)
                {
                        int status = 0;
#if CMK_BPROC_VERSION < 4
                        if (bproc_nodestatus(node) == bproc_node_up) status = 1;
                        if (arg_verbose)
                          printf("Charmrun> node %d status: %s\n", node, status?"up":"down");
#else
                        char nodestatus[128];
                        if (node == -1) {               /* master node is always up */
                          strcpy(nodestatus, "up");
                          status = 1;
                        }
                        if (bproc_nodestatus(node, nodestatus, 128)) {
                          if (strcmp(nodestatus, "up")==0) status = 1;
                        }
                        if (arg_verbose)
                                printf("Charmrun> node %d status: %s\n", node, nodestatus);
#endif
                  return status;
                }

                void nodetab_init_for_scyld()
                {
                  int maxNodes, i, node, npes, rank;
                  nodetab_host group;
                  int tablesize;

                  tablesize = arg_requested_pes;
                  maxNodes = bproc_numnodes() + 1;
                  if (arg_endpe < maxNodes) maxNodes=arg_endpe+1;
                  if (maxNodes > tablesize) tablesize = maxNodes;
                  nodetab_table=(nodetab_host**)malloc(tablesize*sizeof(nodetab_host*));
                  nodetab_rank0_table=(int*)malloc(tablesize*sizeof(int));
                  nodetab_max=tablesize;

                  nodetab_reset(&group);

                  if (arg_ppn==0) arg_ppn=1;
                /*
#if CMK_SHARED_VARS_UNAVAILABLE
                  if (arg_ppn > 1) {
                        fprintf(stderr,"Warning> Invalid ppn %d in nodelist ignored.\n", arg_ppn);
                        arg_ppn=1;
                  }
#endif
                */
                  group.cpus = 1;
                  group.rank = 0;

                  /* check which slave node is available from frompe to endpe */
                  npes = 0;
                  for (i=-1; i<maxNodes && npes < arg_requested_pes; i++) {
                        char hostname[256];
                        if (!bproc_nodeisup(i)) continue;
                        if (i!= -1 && i<arg_startpe) continue;
                        if (i==-1 && arg_skipmaster) continue;    /* skip master node -1 */
                        sprintf(hostname, "%d", i);
#if ! CMK_SHARED_VARS_UNAVAILABLE
                        if (npes + arg_ppn > arg_requested_pes) group.cpus = arg_requested_pes-npes;
                        else group.cpus = arg_ppn;
#endif
                        for (rank = 0; rank<arg_ppn; rank++) {
#if ! CMK_SHARED_VARS_UNAVAILABLE
                          group.rank = rank;
#endif
                          nodetab_makehost(hostname, &group);
                          if (++npes == arg_requested_pes) break;
                        }   
                  }
                  if (nodetab_rank0_size == 0) {
                        fprintf(stderr, "Charmrun> no slave node available!\n");
                        exit (1);
                  }
                  if (arg_verbose)
                        printf("Charmrun> There are %d slave nodes available.\n", nodetab_rank0_size-(arg_skipmaster?0:1));

                  /* expand node table to arg_requested_pes */
                  if (arg_requested_pes > npes) {
                        int orig_size = npes;
                        int node;
                        int startnode = 0;
                        if (arg_singlemaster && nodetab_rank0_size > 1 && !arg_skipmaster) 
                                startnode = arg_ppn;      /* skip -1 */
                        node = startnode; 
                        while (npes < arg_requested_pes) {
#if ! CMK_SHARED_VARS_UNAVAILABLE
                          if (npes+arg_ppn > arg_requested_pes) group.cpus = arg_requested_pes-npes;
                          else group.cpus = arg_ppn;
#endif
                          for (rank = 0; rank<arg_ppn; rank++) {
#if ! CMK_SHARED_VARS_UNAVAILABLE
                                group.rank = rank;
#endif
                                nodetab_makehost(nodetab_name(node), &group);
                                if (++node == orig_size) node = startnode;
                                if (++npes == arg_requested_pes) break;
                          } 
                        }
                  }
                }

                void start_nodes_scyld(void)
                {
                  char *envp[2];
                  int i;

                  envp[0] = (char *)malloc(256);
                  envp[1] = 0;
                  for (i=0;i<nodetab_rank0_size;i++)
                  {
                        int status = 0;
                        int pid;
                        int pe=nodetab_rank0_table[i];
                        int nodeno = atoi(nodetab_name(pe));

                        if (arg_verbose)
                          printf("Charmrun> start node program on slave node: %d.\n", nodeno);
                        sprintf(envp[0], "NETSTART=%s",  create_netstart(i));
                        pid = 0;
                        pid = fork();
                        if (pid < 0) exit(1);
                        if (pid == 0)
                        {
                          int fd, fd1 = dup(1);
                          if (!(arg_debug || arg_debug_no_pause)) {   /* debug mode */
                                if (fd = open("/dev/null", O_RDWR)) {
                                  dup2(fd, 0); dup2(fd, 1); dup2(fd, 2);
                                }
                          }
                          if (nodeno == -1) {
                                status = execve(pparam_argv[1], pparam_argv+1, envp);
                                dup2(fd1, 1);
                                printf("execve failed to start process \"%s\" with status: %d\n", pparam_argv[1], status);
                          }
                          else {
                                status = bproc_execmove(nodeno, pparam_argv[1], pparam_argv+1, envp);
                                dup2(fd1, 1);
                                printf("bproc_execmove failed to start remote process \"%s\" with status: %d\n", pparam_argv[1], status);
                          }
                          kill(getppid(), 9);
                          exit(1);
                        }
                  }
                  free(envp[0]);
                }
                void finish_nodes(void) {}

#else
/*Unix systems can use Rsh normally*/
/********** RSH-ONLY CODE *****************************************/
/*                                                                          */
/* Rsh_etc                                                                  */
/*                                                                          */
/* this starts all the node programs.  It executes fully in the background. */
/*                                                                          */
/****************************************************************************/
#include <sys/wait.h>

extern char **environ;
void removeEnv(const char *doomedEnv)
{ /*Remove a value from the environment list*/
          char **oe, **ie;
          oe=ie=environ;
          while (*ie != NULL) {
                if (0!=strncmp(*ie,doomedEnv,strlen(doomedEnv)))
                  *oe++ = *ie;
                ie++;
          }
          *oe=NULL;/*NULL-terminate list*/
}

int rsh_fork(int nodeno,const char *startScript)
{
  char **rshargv;
  int pid;
  int num=0;
  char *s, *e;

  /* figure out size and dynamic allocate */
  s=nodetab_shell(nodeno); e=skipstuff(s);
  while (*s) {
        num++;
        s = skipblanks(e); e = skipstuff(s);
  }
  rshargv = (char **)malloc(sizeof(char *)*(num+6));

  num = 0;
  s=nodetab_shell(nodeno); e=skipstuff(s);
  while (*s) {
        rshargv[num++]=substr(s, e);
        s = skipblanks(e); e = skipstuff(s);
  }

  rshargv[num++]=nodetab_name(nodeno);
  rshargv[num++]="-l";
  rshargv[num++]=nodetab_login(nodeno);
  rshargv[num++]="/bin/sh -f";
  rshargv[num++]=0;
  if (arg_verbose) printf("Charmrun> Starting %s %s -l %s %s\n",nodetab_shell(nodeno), nodetab_name(nodeno),nodetab_login(nodeno), rshargv[num-2]);
                  
  pid = fork();
  if (pid < 0) 
        { perror("ERROR> starting rsh"); exit(1); }
  if (pid == 0)
  {/*Child process*/
          int i;
          int fdScript=open(startScript,O_RDONLY);
    unlink(startScript); 
          dup2(fdScript,0);/*Open script as standard input*/
          //removeEnv("DISPLAY="); /*No DISPLAY disables ssh's slow X11 forwarding*/
          for(i=3; i<1024; i++) close(i);
          execvp(rshargv[0], rshargv);
          fprintf(stderr,"Charmrun> Couldn't find rsh program '%s'!\n",rshargv[0]);
          exit(1);
  }
  free(rshargv);
  if (arg_verbose)
        fprintf(stderr,"Charmrun> remote shell (%s:%d) started\n",
                nodetab_name(nodeno),nodeno);
  return pid;
}

void fprint_arg(FILE *f,char **argv)
{
  while (*argv) { 
        fprintf(f," %s",*argv); 
        argv++; 
  }
}
void rsh_Find(FILE *f,const char *program,const char *dest)
{
    fprintf(f,"Find %s\n",program);
    fprintf(f,"%s=$loc\n",dest);
}
void rsh_script(FILE *f, int nodeno, int rank0no, char **argv, int restart)
{
  char *netstart;
  char *arg_nodeprog_r,*arg_currdir_r;
  char *dbg=nodetab_debugger(nodeno);
  char *host=nodetab_name(nodeno);

  if (arg_mpiexec)
        fprintf(f, "#!/bin/sh\n");

  fprintf(f, /*Echo: prints out status message*/
        "Echo() {\n"
        "  echo 'Charmrun remote shell(%s.%d)>' $*\n"
        "}\n",host,nodeno);
  fprintf(f, /*Exit: exits with return code*/
        "Exit() {\n"
        "  if [ $1 -ne 0 ]\n"
        "  then\n"
        "    Echo Exiting with error code $1\n"
        "  fi\n"
#if CMK_RSH_KILL /*End by killing ourselves*/
        "  sleep 5\n" /*Delay until any error messages are flushed*/
        "  kill -9 $$\n"
#else /*Exit normally*/
        "  exit $1\n"
#endif
        "}\n");
  fprintf(f, /*Find: locates a binary program in PATH, sets loc*/
        "Find() {\n"
        "  loc=''\n"
        "  for dir in `echo $PATH | sed -e 's/:/ /g'`\n"
        "  do\n"
        "    test -f \"$dir/$1\" && loc=\"$dir/$1\"\n"
        "  done\n"
        "  if [ \"x$loc\" = x ]\n"
        "  then\n"
        "    Echo $1 not found in your PATH \"($PATH)\"--\n"
        "    Echo set your path in your ~/.charmrunrc\n"
        "    Exit 1\n"
        "  fi\n"
        "}\n");
  
  if (arg_verbose) fprintf(f,"Echo 'remote responding...'\n");
  
  fprintf(f,"test -f \"$HOME/.charmrunrc\" && . \"$HOME/.charmrunrc\"\n");
/* let's leave DISPLAY untouched and rely on X11 forwarding,
   changing DISPLAY to charmrun does not always work if X11 forwarding presents 
*/
  if (arg_display && !arg_ssh_display)
    fprintf(f,"DISPLAY='%s';export DISPLAY\n",arg_display);

#ifdef HSTART
  if(arg_child_charmrun)
                  fprintf(f,"NETMAGIC=\"%d\";export NETMAGIC\n",parent_charmrun_pid&0x7FFF);
  else
#endif
                fprintf(f,"NETMAGIC=\"%d\";export NETMAGIC\n",getpid()&0x7FFF);

  if (arg_mpiexec) {
    fprintf(f,"CmiMyNode=$OMPI_COMM_WORLD_RANK\n");
    fprintf(f,"test -z \"$CmiMyNode\" && CmiMyNode=$MPIRUN_RANK\n");
    fprintf(f,"test -z \"$CmiMyNode\" && CmiMyNode=$PMI_RANK\n");
    fprintf(f,"test -z \"$CmiMyNode\" && CmiMyNode=$PMI_ID\n");
    fprintf(f,"test -z \"$CmiMyNode\" && (Echo Could not detect rank from environment ; Exit 1)\n");
    fprintf(f,"export CmiMyNode\n");
  }
#ifdef HSTART
  else  if(arg_hierarchical_start && arg_child_charmrun)
                         fprintf(f,"CmiMyNode='%d'; export CmiMyNode\n",mynodes_start+rank0no);
#endif
  else
    fprintf(f,"CmiMyNode='%d'; export CmiMyNode\n",rank0no);

#ifdef HSTART
  if(arg_hierarchical_start && arg_child_charmrun)
                          netstart = create_netstart(mynodes_start+rank0no);
  else
#endif
  netstart = create_netstart(rank0no);
  fprintf(f,"NETSTART=\"%s\";export NETSTART\n",netstart);

  fprintf(f,"CmiMyNodeSize='%d'; export CmiMyNodeSize\n",nodetab_getnodeinfo(rank0no)->cpus);

  if (restart || arg_mpiexec)    /* skip fork */
    fprintf(f,"CmiMyForks='%d'; export CmiMyForks\n",0);
  else
    fprintf(f,"CmiMyForks='%d'; export CmiMyForks\n",nodetab_getnodeinfo(rank0no)->forks);

  if (arg_mpiexec) {
    fprintf(f,"CmiNumNodes=$OMPI_COMM_WORLD_SIZE\n");
    fprintf(f,"test -z \"$CmiNumNodes\" && CmiNumNodes=$MPIRUN_NPROCS\n");
    fprintf(f,"test -z \"$CmiNumNodes\" && CmiNumNodes=$PMI_SIZE\n");
    fprintf(f,"test -z \"$CmiNumNodes\" && (Echo Could not detect node count from environment ; Exit 1)\n");
    fprintf(f,"export CmiNumNodes\n");
  }
#ifdef HSTART
  else  if(arg_hierarchical_start && arg_child_charmrun)
                 fprintf(f,"CmiNumNodes='%d'; export CmiNumNodes\n",nodetab_rank0_size_total);
#endif

  else
    fprintf(f,"CmiNumNodes='%d'; export CmiNumNodes\n",nodetab_rank0_size);

#if CONVERSE_VERSION_VMI
  /* VMI environment variable */
  fprintf (f, "VMI_PROCS='%d'; export VMI_PROCS\n", arg_requested_pes);
  fprintf (f, "VMI_KEY='charmrun%d'; export VMI_KEY\n", getpid ());
  fprintf (f, "VMI_SPECFILE='%s'; export VMI_SPECFILE\n", arg_vmispecfile);
#endif
#ifdef CMK_G95
  fprintf(f,"G95_UNBUFFERED_ALL=TRUE; export G95_UNBUFFERED_ALL\n");
#endif
#ifdef CMK_GFORTRAN
  fprintf(f,"GFORTRAN_UNBUFFERED_ALL=YES; export GFORTRAN_UNBUFFERED_ALL\n");
#endif
#if CMK_USE_MX
  fprintf(f,"MX_MONOTHREAD=1; export MX_MONOTHREAD\n");
  /*fprintf(f,"MX_RCACHE=1; export MX_RCACHE\n");*/
#endif
#if CMK_AIX && CMK_SMP
  fprintf(f,"MALLOCMULTIHEAP=1; export MALLOCMULTIHEAP\n");
#endif
  
  if (arg_verbose) {
    printf("Charmrun> Sending \"%s\" to client %d.\n", netstart, rank0no);
  }
  fprintf(f,"PATH=\"$PATH:/bin:/usr/bin:/usr/X/bin:/usr/X11/bin:/usr/local/bin:"
        "/usr/X11R6/bin:/usr/openwin/bin\"\n");
  
  /* find the node-program */
  arg_nodeprog_r = pathextfix(arg_nodeprog_a, nodetab_pathfixes(nodeno), nodetab_ext(nodeno));
  
  /* find the current directory, relative version */
  arg_currdir_r = pathfix(arg_currdir_a, nodetab_pathfixes(nodeno));
  
 if (arg_verbose) {
        printf("Charmrun> find the node program \"%s\" at \"%s\" for %d.\n", arg_nodeprog_r, arg_currdir_r, nodeno);
  }
  if (arg_debug || arg_debug_no_pause || arg_in_xterm) {
        rsh_Find(f,nodetab_xterm(nodeno),"F_XTERM");
        if(!arg_ssh_display && !arg_debug_no_xrdb)
          rsh_Find(f,"xrdb","F_XRDB");
        if(arg_verbose) fprintf(f,"Echo 'using xterm' $F_XTERM\n");
  }

  if (arg_debug || arg_debug_no_pause)
  {/*Look through PATH for debugger*/
        rsh_Find(f,dbg,"F_DBG");
        if (arg_verbose) fprintf(f,"Echo 'using debugger' $F_DBG\n");
  }

   if (!arg_ssh_display && !arg_debug_no_xrdb && 
       (arg_debug || arg_debug_no_pause || arg_in_xterm)) {
     /*    if (arg_debug || arg_debug_no_pause || arg_in_xterm) {*/
    fprintf(f,"$F_XRDB -query > /dev/null\n");
    fprintf(f,"if test $? != 0\nthen\n");
    fprintf(f,"  Echo 'Cannot contact X Server '$DISPLAY'.  You probably'\n");
    fprintf(f,"  Echo 'need to run xhost to authorize connections.'\n");
    fprintf(f,"  Echo '(See manual for xhost for security issues)'\n");
    fprintf(f,"  Echo 'Or try ++batch 1 ++ssh-display to rely on SSH X11 forwarding'\n");
    fprintf(f,"  Exit 1\n");
    fprintf(f,"fi\n");
  }
  
  fprintf(f,"if test ! -x \"%s\"\nthen\n",arg_nodeprog_r);
  fprintf(f,"  Echo 'Cannot locate this node-program: %s'\n",arg_nodeprog_r);
  fprintf(f,"  Exit 1\n");
  fprintf(f,"fi\n");
  
  fprintf(f,"cd \"%s\"\n",arg_currdir_r);
  fprintf(f,"if test $? = 1\nthen\n");
  fprintf(f,"  Echo 'Cannot propagate this current directory:'\n"); 
  fprintf(f,"  Echo '%s'\n",arg_currdir_r);
  fprintf(f,"  Exit 1\n");
  fprintf(f,"fi\n");
  
  if (strcmp(nodetab_setup(nodeno),"*")) {
    fprintf(f,"%s\n",nodetab_setup(nodeno));
    fprintf(f,"if test $? = 1\nthen\n");
    fprintf(f,"  Echo 'this initialization command failed:'\n");
    fprintf(f,"  Echo '\"%s\"'\n",nodetab_setup(nodeno));
    fprintf(f,"  Echo 'edit your nodes file to fix it.'\n");
    fprintf(f,"  Exit 1\n");
    fprintf(f,"fi\n");
  }

  fprintf(f,"rm -f /tmp/charmrun_err.$$\n");
  if(arg_verbose) fprintf(f,"Echo 'starting node-program...'\n");  
  /* This is the start of the the run-nodeprogram script */
  fprintf(f,"(");
  
  if (arg_debug || arg_debug_no_pause ) {
         if ( strcmp(dbg, "gdb") == 0 || strcmp(dbg, "idb") == 0 ) {
           fprintf(f,"cat > /tmp/charmrun_gdb.$$ << END_OF_SCRIPT\n");
           if ( strcmp(dbg, "idb") == 0 ) {
             fprintf(f,"set \\$cmdset=\"gdb\"\n");
           }
           fprintf(f,"shell /bin/rm -f /tmp/charmrun_gdb.$$\n");
           fprintf(f,"handle SIGPIPE nostop noprint\n");
           fprintf(f,"handle SIGWINCH nostop noprint\n");
           fprintf(f,"handle SIGWAITING nostop noprint\n");
           if(arg_debug_commands)
             fprintf(f,"%s\n", arg_debug_commands);
           fprintf(f,"set args");
           fprint_arg(f,argv);
           fprintf(f,"\n");
           if (arg_debug_no_pause) fprintf(f,"run\n");
           fprintf(f,"END_OF_SCRIPT\n");
           if (arg_runscript)
             fprintf(f,"\"%s\" ",arg_runscript);
           fprintf(f,"$F_XTERM");
           fprintf(f," -title 'Node %d (%s)' ",nodeno,nodetab_name(nodeno));
           if ( strcmp(dbg, "idb") == 0 )
             fprintf(f," -e $F_DBG %s -c /tmp/charmrun_gdb.$$ \n", arg_nodeprog_r);
           else 
             fprintf(f," -e $F_DBG %s -x /tmp/charmrun_gdb.$$ \n", arg_nodeprog_r);
         } else if ( strcmp(dbg, "dbx") == 0 ) {
           fprintf(f,"cat > /tmp/charmrun_dbx.$$ << END_OF_SCRIPT\n");
           fprintf(f,"sh /bin/rm -f /tmp/charmrun_dbx.$$\n");
           fprintf(f,"dbxenv suppress_startup_message 5.0\n");
           fprintf(f,"ignore SIGPOLL\n");
           fprintf(f,"ignore SIGPIPE\n");
           fprintf(f,"ignore SIGWINCH\n");
           fprintf(f,"ignore SIGWAITING\n");
           if(arg_debug_commands)
             fprintf(f,"%s\n", arg_debug_commands);
           fprintf(f,"END_OF_SCRIPT\n");
           if (arg_runscript)
             fprintf(f,"\"%s\" ",arg_runscript);
           fprintf(f,"$F_XTERM");
           fprintf(f," -title 'Node %d (%s)' ",nodeno,nodetab_name(nodeno));
           fprintf(f," -e $F_DBG %s ",arg_debug_no_pause?"-r":"");
           if(arg_debug) {
              fprintf(f,"-c \'runargs ");
              fprint_arg(f,argv);
              fprintf(f,"\' ");
           }
           fprintf(f, "-s/tmp/charmrun_dbx.$$ %s",arg_nodeprog_r);
           if(arg_debug_no_pause) 
              fprint_arg(f,argv);
           fprintf(f,"\n");
         } else { 
          fprintf(stderr, "Unknown debugger: %s.\n Exiting.\n", 
            nodetab_debugger(nodeno));
         }
  } else if (arg_in_xterm) {
    if(arg_verbose)
      fprintf(stderr, "Charmrun> node %d: xterm is %s\n", 
              nodeno, nodetab_xterm(nodeno));
    fprintf(f,"cat > /tmp/charmrun_inx.$$ << END_OF_SCRIPT\n");
    fprintf(f,"#!/bin/sh\n");
    fprintf(f,"/bin/rm -f /tmp/charmrun_inx.$$\n");
    fprintf(f,"%s", arg_nodeprog_r);
    fprint_arg(f,argv);
    fprintf(f,"\n");
    fprintf(f,"echo 'program exited with code '\\$?\n");
    fprintf(f,"read eoln\n");
    fprintf(f,"END_OF_SCRIPT\n");
    fprintf(f,"chmod 700 /tmp/charmrun_inx.$$\n");
    if (arg_runscript)
       fprintf(f,"\"%s\" ",arg_runscript);
    fprintf(f,"$F_XTERM -title 'Node %d (%s)' ",nodeno,nodetab_name(nodeno));
    fprintf(f," -sl 5000");
    fprintf(f," -e /tmp/charmrun_inx.$$\n");
  } else {
    if (arg_runscript)
       fprintf(f,"\"%s\" ",arg_runscript);
    if (arg_no_va_rand) {
      if(arg_verbose) fprintf(stderr, "Charmrun> setarch -R is used.\n");
      fprintf(f,"setarch `uname -m` -R ");
    }
    fprintf(f,"\"%s\" ",arg_nodeprog_r);
    fprint_arg(f,argv);
    if (nodetab_nice(nodeno) != -100) {
      if(arg_verbose) fprintf(stderr, "Charmrun> nice -n %d\n", nodetab_nice(nodeno));
      fprintf(f," +nice %d ",nodetab_nice(nodeno));
    }
    fprintf(f,"\nres=$?\n");
    /* If shared libraries fail to load, the program dies without
       calling charmrun back.  Since we *have* to close down stdin/out/err,
       we have to smuggle this failure information out via a file,
       /tmp/charmrun_err.<pid> */
    fprintf(f,
        "if [ $res -eq 127 ]\n"
        "then\n"
        "  ( \n" /* Re-run, spitting out errors from a subshell: */
        "    \"%s\" \n"
        "    ldd \"%s\"\n"
        "  ) > /tmp/charmrun_err.$$ 2>&1 \n"
        "fi\n",arg_nodeprog_r,arg_nodeprog_r);
  }
  
  /* End the node-program subshell. To minimize the number 
     of open ports on the front-end, we must close down rsh;
     to do this, we have to close stdin, stdout, stderr, and 
     run the subshell in the background. */
  fprintf(f,")");
  fprintf(f," < /dev/null 1> /dev/null 2> /dev/null");
  if (!arg_mpiexec)
      fprintf(f, " &");
  fprintf(f, "\n");
  
  if (arg_verbose) fprintf(f,"Echo 'rsh phase successful.'\n");
  fprintf(f, /* Check for startup errors: */
     "sleep 1\n"
     "if [ -r /tmp/charmrun_err.$$ ]\n"
     "then\n"
     "  cat /tmp/charmrun_err.$$ \n"
     "  rm -f /tmp/charmrun_err.$$ \n"
     "  Exit 1\n"
     "fi\n");
  fprintf(f,"Exit 0\n");
}


/* use the command "size" to get information about the position of the ".data"
   and ".bss" segments inside the program memory */
void read_global_segments_size() {
  char **rshargv;
  int childPid;

  /* find the node-program */
  arg_nodeprog_r = pathextfix(arg_nodeprog_a, nodetab_pathfixes(0), nodetab_ext(0));

  rshargv = (char **)malloc(sizeof(char *)*6);
  rshargv[0]=nodetab_shell(0);
  rshargv[1]=nodetab_name(0);
  rshargv[2]="-l";
  rshargv[3]=nodetab_login(0);
  rshargv[4] = (char *)malloc(sizeof(char)*9+strlen(arg_nodeprog_r));
  sprintf(rshargv[4],"size -A %s",arg_nodeprog_r);
  rshargv[5]=0;

  childPid = fork();
  if (childPid < 0) {
    perror("ERROR> getting the size of the global variables segments"); exit(1);
  } else if (childPid == 0) {
    /* child process */
    dup2(2, 1);
    /*printf("executing: \"%s\" \"%s\" \"%s\" \"%s\" \"%s\"\n",rshargv[0],rshargv[1],rshargv[2],rshargv[3],rshargv[4]);*/
    execvp(rshargv[0], rshargv);
    fprintf(stderr,"Charmrun> Couldn't find rsh program '%s'!\n",rshargv[0]);
    exit(1);
  } else {
    /* else we are in the parent */
    free(rshargv[4]);
    free(rshargv);
    waitpid(childPid, NULL, 0);
  }
}

/* open a rsh connection with processor 0 and open a gdb session for info */
void open_gdb_info() {
  char **rshargv;
  int fdin[2];
  int fdout[2];
  int fderr[2];
  int i;

  /* find the node-program */
  arg_nodeprog_r = pathextfix(arg_nodeprog_a, nodetab_pathfixes(0), nodetab_ext(0));

  rshargv = (char **)malloc(sizeof(char *)*6);
  rshargv[0]=nodetab_shell(0);
  rshargv[1]=nodetab_name(0);
  rshargv[2]="-l";
  rshargv[3]=nodetab_login(0);
  rshargv[4] = (char *)malloc(sizeof(char)*8+strlen(arg_nodeprog_r));
  sprintf(rshargv[4],"gdb -q %s",arg_nodeprog_r);
  rshargv[5]=0;

                  pipe(fdin);
                  pipe(fdout);
                  pipe(fderr);

                  gdb_info_pid = fork();
                  if (gdb_info_pid < 0) {
                        perror("ERROR> starting info gdb"); exit(1);
                  } else if (gdb_info_pid == 0) {
                        /* child process */
                        close(fdin[1]);
                        close(fdout[0]);
                        close(fderr[0]);
                        printf("executing: \"%s\" \"%s\" \"%s\" \"%s\" \"%s\"\n",rshargv[0],rshargv[1],rshargv[2],rshargv[3],rshargv[4]);
                        dup2(fdin[0],0);
                        dup2(fdout[1],1);
                        dup2(fderr[1],2);
                        for(i=3; i<1024; i++) close(i);
                        execvp(rshargv[0], rshargv);
                        fprintf(stderr,"Charmrun> Couldn't find rsh program '%s'!\n",rshargv[0]);
                        exit(1);
                  }
                  /* else we are in the parent */
                  free(rshargv[4]);
                  free(rshargv);
                  gdb_info_std[0] = fdin[1];
                  gdb_info_std[1] = fdout[0];
                  gdb_info_std[2] = fderr[0];
                  close(fdin[0]);
                  close(fdout[1]);
                  close(fderr[1]);
                }
#ifdef HSTART
                void start_next_level_charmruns()
                {
                        
                   static char buf[1024];
                   char * nodeprog_name = strrchr(arg_nodeprog_a, '/');
                   nodeprog_name[0] = 0;
                   sprintf(buf,"%s%s%s",arg_nodeprog_a,DIRSEP,"charmrun");
                   arg_nodeprog_a = strdup(buf);
                
                   int client;
                         int nextIndex =0;
                         client=0;
                         while(nextIndex<branchfactor){
                         /* need to index into unique_table*/
                         int rank0no = nodetab_unique_table[client];
                         int pe=nodetab_rank0_table[rank0no];
                         FILE *f;
                         char startScript[200];
                         sprintf(startScript,"/tmp/charmrun.%d.%d",getpid(),pe);
                         f=fopen(startScript,"w");
                         if (f==NULL) {
                           /* now try current directory */
                           sprintf(startScript,"charmrun.%d.%d",getpid(),pe);
                           f=fopen(startScript,"w");
                           if (f==NULL) {
                                 fprintf(stderr,"Charmrun> Can not write file %s!\n", startScript);
                                 exit(1);
                           }
                         }
                         rsh_script(f,pe,client,arg_argv,0);
                         fclose(f);
                        if (!rsh_pids)
                           rsh_pids=(int *)malloc(sizeof(int)*branchfactor);
                         rsh_pids[nextIndex++] = rsh_fork(pe,startScript);
                                 client += nodes_per_child;

                        }
                }
#endif
                                                                                
/* returns pid */
void start_one_node_rsh(int rank0no)
{
         int pe=nodetab_rank0_table[rank0no];
         FILE *f;
         char startScript[200];
         sprintf(startScript,"/tmp/charmrun.%d.%d",getpid(),pe);
         f=fopen(startScript,"w");
         if (f==NULL) {
           /* now try current directory */
           sprintf(startScript,"charmrun.%d.%d",getpid(),pe);
           f=fopen(startScript,"w");
           if (f==NULL) {
                 fprintf(stderr,"Charmrun> Can not write file %s!\n", startScript);
                 exit(1);
           }
         }
         rsh_script(f,pe,rank0no,arg_argv,0);
         fclose(f);
         if (!rsh_pids)
           rsh_pids=(int *)malloc(sizeof(int)*nodetab_rank0_size);
         rsh_pids[rank0no] = rsh_fork(pe,startScript);
}

int start_set_node_rsh(int client) {
        /* a search function could be inserted here instead of sequential lookup for more complex node lists (e.g. interleaving) */
        int clientgroup;
#if CMK_SMP || defined(_WIN32)
        clientgroup=client+1; /* smp already handles this functionality */
#else

#ifdef HSTART
        if(!arg_scalable_start && !arg_hierarchical_start)
                clientgroup=client+1; /* only launch 1 core per rsh call */
        else {
                clientgroup=client;
                do {
                        clientgroup++; /* add one more client to group if not greater than nodes and shares the same name as client */
                if(clientgroup>=nodetab_rank0_size)
                break;
            if(arg_scalable_start&&!arg_hierarchical_start)
                if(strcmp(nodetab_name(clientgroup),nodetab_name(client)))
                        break;
            /*Hierarchical-start*/
            if(strcmp(nodetab_name(nodetab_rank0_table[clientgroup]),nodetab_name(nodetab_rank0_table[client])))
                        break;
        }
        while(1);
        }

#else
        if(!arg_scalable_start)
                clientgroup=client+1; /* only launch 1 core per rsh call */
        else {
                clientgroup=client;
                do {
                        clientgroup++; /* add one more client to group if not greater than nodes and shares the same name as client */
        }while(clientgroup<nodetab_rank0_size&&(!strcmp(nodetab_name(clientgroup),nodetab_name(client))));
        }
#endif

#endif
                        nodetab_getnodeinfo(client)->forks=clientgroup-client-1; /* already have 1 process launching */
                        start_one_node_rsh(client);
                        return clientgroup-client; /* return number of entries in group */
                }

void start_nodes_rsh()
{
  int client,clientgroup;
  rsh_pids=(int *)malloc(sizeof(int)*nodetab_rank0_size);

  if (arg_verbose) printf("start_nodes_rsh\n");
  client=0;
  while(client<nodetab_rank0_size) {
        /* start a group of processes per node */
        clientgroup=start_set_node_rsh(client);
        client+=clientgroup;
  }
}


/* for mpiexec, for once calling mpiexec to start on all nodes  */
int rsh_fork_one(const char *startScript)
{
  char **rshargv;
  int pid;
  int num=0;
  char npes[128];
  char *s, *e;

  /* figure out size and dynamic allocate */
  s=nodetab_shell(0); e=skipstuff(s);
  while (*s) {
    num++;
    s = skipblanks(e); e = skipstuff(s);
  }
  rshargv = (char **)malloc(sizeof(char *)*(num+8));

  num = 0;
  s=nodetab_shell(0); e=skipstuff(s);
  while (*s) {
    rshargv[num++]=substr(s, e);
    s = skipblanks(e); e = skipstuff(s);
  }

  rshargv[num++]="-n";
  sprintf(npes, "%d", nodetab_rank0_size);
  rshargv[num++]=npes;
  rshargv[num++]=(char*)startScript;
  rshargv[num++]=0;
  if (arg_verbose) printf("Charmrun> Starting %s %s \n", nodetab_shell(0), startScript);
  
  pid = fork();
  if (pid < 0) 
        { perror("ERROR> starting mpiexec"); exit(1); }
  if (pid == 0)
  {/*Child process*/
      int i;
  /*  unlink(startScript); */
      //removeEnv("DISPLAY="); /*No DISPLAY disables ssh's slow X11 forwarding*/
      for(i=3; i<1024; i++) close(i);
      execvp(rshargv[0], rshargv);
      fprintf(stderr,"Charmrun> Couldn't find mpiexec program '%s'!\n",rshargv[0]);
      exit(1);
  }
  free(rshargv);
  if (arg_verbose)
    fprintf(stderr,"Charmrun> mpiexec started\n");
  return pid;
}

void start_nodes_mpiexec()
{
     int i;

     FILE *f;
     char startScript[200];
     sprintf(startScript,"./charmrun.%d",getpid());
     f=fopen(startScript,"w");
     chmod(startScript, S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IROTH);
     if (f==NULL) {
       /* now try current directory */
       sprintf(startScript,"./charmrun.%d",getpid());
       f=fopen(startScript,"w");
       if (f==NULL) {
         fprintf(stderr,"Charmrun> Can not write file %s!\n", startScript);
         exit(1);
       }
     }
     rsh_script(f,0,0,arg_argv,0);
     fclose(f);
     rsh_pids=(int *)malloc(sizeof(int)*nodetab_rank0_size);
     rsh_pids[0]=rsh_fork_one(startScript);
     for (i=0; i<nodetab_rank0_size; i++)
       rsh_pids[i] = 0;   /* skip finish_nodes */
}

void finish_set_nodes(int start, int stop) {
        int status,done,i;
        char *host;

        if (!rsh_pids) return; /*nothing to do*/

        done=0;
        while(!done) {
                done=1;
                for(i=start;i<stop;i++) { /* check all nodes */
                        if(rsh_pids[i]!=0) {
                                done=0; /* we are not finished yet */
                                status=0;
                                waitpid(rsh_pids[i],&status,0); /* check if the process is finished */
                                if(WIFEXITED(status)) {
                                        if (!WEXITSTATUS(status)) { /* good */
                                                rsh_pids[i]=0; /* process is finished */
                                        } else {
                                                host=nodetab_name(nodetab_rank0_table[i]);
                                                fprintf(stderr,"Charmrun> Error %d returned from rsh (%s:%d)\n",
                                                WEXITSTATUS(status),host,i);
                                                exit(1);
                                        } 
                                }
                        }
                }
        }
}

void finish_nodes()
{
#ifdef HSTART
        if(arg_hierarchical_start && !arg_child_charmrun)
                                        finish_set_nodes(0, branchfactor);
        else 
#endif
        finish_set_nodes(0,nodetab_rank0_size);
        free(rsh_pids);
}

void kill_nodes()
{
  int rank0no;
  if (!rsh_pids) return; /*nothing to do*/
  /*Now wait for all the rsh'es to finish*/
  for (rank0no=0;rank0no<nodetab_rank0_size;rank0no++)
  {
         const char *host=nodetab_name(nodetab_rank0_table[rank0no]);
         int status=0;
         if (arg_verbose) printf("Charmrun> waiting for rsh (%s:%d), pid %d\n",
                host,rank0no,rsh_pids[rank0no]);
         kill(rsh_pids[rank0no],9);
         waitpid(rsh_pids[rank0no],&status,0); /*<- no zombies*/
  }
  free(rsh_pids);
}

/* simple version of charmrun that avoids the rshd or charmd,   */
/* it spawn the node program just on local machine using exec. */
void start_nodes_local(char ** env)
{
  char **envp;
  int envc, rank0no, i;
  int extra = 0;

#if CMK_AIX && CMK_SMP
  extra = 1;
#endif

  /* copy environ and expanded to hold NETSTART and CmiNumNodes */ 
  for (envc=0; env[envc]; envc++);
  envp = (char **)malloc((envc+2+extra+1)*sizeof(void *));
  for (i=0; i<envc; i++) envp[i] = env[i];
  envp[envc] = (char *)malloc(256);
  envp[envc+1] = (char *)malloc(256);
#if CMK_AIX && CMK_SMP
  envp[envc+2] = (char *)malloc(256);
  sprintf(envp[envc+2], "MALLOCMULTIHEAP=1");
#endif
  envp[envc+2+extra] = 0;

  for (rank0no=0;rank0no<nodetab_rank0_size;rank0no++)
  {
        int status = 0;
        int pid;
        int pe=nodetab_rank0_table[rank0no];

        if (arg_verbose)
          printf("Charmrun> start %d node program on localhost.\n", pe);
        sprintf(envp[envc], "NETSTART=%s",  create_netstart(rank0no));
        sprintf(envp[envc+1],"CmiNumNodes=%d", nodetab_rank0_size);
        pid = 0;
        pid = fork();
        if (pid < 0) exit(1);
        if (pid == 0)
        {
          int fd, fd1 = dup(1);
          if (-1!=(fd = open("/dev/null", O_RDWR))) {
                dup2(fd, 0); dup2(fd, 1); dup2(fd, 2);
          }
          status = execve(pparam_argv[1], pparam_argv+1, envp);
          dup2(fd1, 1);
          printf("execve failed to start process \"%s\" with status: %d\n", pparam_argv[1], status);
          kill(getppid(), 9);
          exit(1);
        }
 }
 free(envp[envc]);
 free(envp[envc+1]);
#if CMK_AIX && CMK_SMP
 free(envp[envc+2]);
#endif
 free(envp);
}

#ifdef __FAULT__

int current_restart_phase = 1;

void refill_nodetab_entry(int crashed_node);
nodetab_host *replacement_host(int pe);

void restart_node(int crashed_node){
        int pe = nodetab_rank0_table[crashed_node];
        FILE *f;
        char startScript[200];
        int restart_rsh_pid;
        char **restart_argv;
        int status=0;
        char phase_str[10];
        int i;
        sprintf(startScript,"/tmp/charmrun.%d.%d",getpid(),pe);
        f=fopen(startScript,"w");
        
        i=0;
        while(arg_argv[i]!= NULL){
                i++;
        }
        restart_argv = (char **)malloc(sizeof(char *)*(i+4));
        i=0;
        while(arg_argv[i]!= NULL){
                restart_argv[i] = arg_argv[i];
                i++;
        }
        restart_argv[i] = "+restartaftercrash";
        sprintf(phase_str,"%d", ++current_restart_phase);
        restart_argv[i+1]=phase_str;
        restart_argv[i+2] = "+restartisomalloc";
        restart_argv[i+3]=NULL;

        rsh_script(f,pe,crashed_node,restart_argv,1);
        fclose(f);
        refill_nodetab_entry(crashed_node);
        restart_rsh_pid =rsh_fork(pe,startScript);
        status=0;
        if (arg_debug_no_pause || arg_debug) ;
        else {
        do{
                waitpid(restart_rsh_pid,&status,0);
        }while(!WIFEXITED(status));
        if (WEXITSTATUS(status)!=0){
         fprintf(stderr,"Charmrun> Error %d returned from new attempted rsh \n",
         WEXITSTATUS(status));
         exit(1);
        }     
        }
        printf("Charmrun finished launching new process in %fs\n", GetClock()-ftTimer);
}



void refill_nodetab_entry(int crashed_node){
        int pe =  nodetab_rank0_table[crashed_node];
        nodetab_host *h = nodetab_table[pe];
        *h = *(replacement_host(pe));
#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
fprintf(stderr,"Charmrun>>> New pe %d is on host %s \n",pe,nodetab_name(pe));
#endif
}

#if (defined(_FAULT_MLOG_) || defined(_FAULT_CAUSAL_))
nodetab_host *replacement_host(int pe){
    int x=loaded_max_pe+1;

    x = x%arg_read_pes;
    loaded_max_pe +=1;
/*  while(x == pe){
 *       x = rand()%nodetab_size;    
 *           }*/
    fprintf(stderr,"Charmrun>>> replacing pe %d with %d host %s with %s \n",pe,x,nodetab_name(pe),nodetab_name(x));
    return nodetab_table[x];
}
#else
nodetab_host *replacement_host(int pe){
        int x=pe;
        while(x == pe){
#ifdef HSTART         
         if(arg_hierarchical_start){ 
                x = nodetab_rank0_table[rand()%nodetab_rank0_size];     
                crashed_pe_id = pe;
                restarted_pe_id = x;
         }
         else
#endif
                x = rand()%nodetab_size;        
        }
        return nodetab_table[x];
}
#endif

void reconnect_crashed_client(int socket_index,int crashed_node){
        int i;
        unsigned int clientPort;
        skt_ip_t clientIP;
        ChSingleNodeinfo *in;
        if(0==skt_select1(server_fd,arg_timeout*1000)){
                        client_connect_problem(socket_index,"Timeout waiting forrestarted node-program to connect");
        }
        req_clients[socket_index] = skt_accept(server_fd,&clientIP,&clientPort);
        if(req_clients[socket_index] == SOCKET_ERROR){
                client_connect_problem(socket_index,"Failure in restarted node accept");
        }else{
                ChMessage msg;
                if(!skt_select1(req_clients[socket_index],arg_timeout*1000)){
                        client_connect_problem(socket_index,"Timeout on IP request for restarted processor");
                }

#ifdef HSTART
                if(arg_hierarchical_start){ 
                        req_forward_root(req_clients[socket_index]);
                if (_last_crash != 0) {
                  fprintf(stderr, "ERROR> Charmrun detected multiple crashes.\n");
                  exit(1);
                }

                        _last_crash = crashed_node;
                        _crash_socket_index = socket_index;
                        return;
                }
#endif
                ChMessage_recv(req_clients[socket_index],&msg);
                if(msg.len != sizeof(ChSingleNodeinfo)){
                  fprintf(stderr,"Charmrun: Bad initnode data length. Aborting\n");
                  fprintf(stderr,"Charmrun: possibly because: %s.\n", msg.data);
                }
fprintf(stdout,"socket_index %d crashed_node %d reconnected fd %d  \n",socket_index,crashed_node,req_clients[socket_index]);

                in = (ChSingleNodeinfo *)msg.data;
                nodeinfo_add(in,req_clients[socket_index]);
                for(i=0;i<req_nClients;i++){
                        if(i != socket_index){
                                req_handle_initnodetab(NULL,req_clients[i]);
                        }       
                }

                /* tell every one there is a crash */
                announce_crash(socket_index,crashed_node);
                if (_last_crash != 0) {
                  fprintf(stderr, "ERROR> Charmrun detected multiple crashes.\n");
                  exit(1);
                }
                _last_crash = crashed_node;
                _crash_socket_index = socket_index;
                /*holds the restarted process until I got ack back from
                  everyone in req_handle_crashack
                  now the restarted one can only continue until 
                  req_handle_crashack calls req_handle_initnodetab(socket_index)
                  req_handle_initnodetab(NULL,req_clients[socket_index]); */
                ChMessage_free(&msg);
        }       
}

void announce_crash(int socket_index,int crashed_node){
        int i;
        ChMessageHeader hdr;
        ChMessageInt_t crashNo=ChMessageInt_new(crashed_node);
        ChMessageHeader_new("crashnode",sizeof(ChMessageInt_t),&hdr);
        for(i=0;i<req_nClients;i++){
                if(i != socket_index){
                        skt_sendN(req_clients[i],(const char *)&hdr,sizeof(hdr));
                        skt_sendN(req_clients[i],(const char *)&crashNo,sizeof(ChMessageInt_t));
                }
        }
}

#endif

#endif /*CMK_USE_RSH*/

---