v2run_internals.c

Path MPICH-V for MPICH the MPI Implementation

	    {
	      newCS   = (CS *)malloc(sizeof(CS));
	      sscanf(line, "%s %s %s %d %s",
		     type,
		     newCS->ipAddress,
		     newCS->fastIpAddress,
		     &newCS->port,
		     newCS->tmp);
	      newCS->autoLaunch = js->autoLaunch;
	      strcpy(newCS->debugString, "");
	      if (totalCS == 0)
		{
		  js->csList = newCS;
		  lastCS = js->csList;
		}
	      else
		{
		  lastCS->next = newCS;
		  lastCS = lastCS->next;
		}
	      lastCS->next = NULL;
	      /*addAuxiliary(newCS->ipAddress, TYPE_CS);*/
	      totalCS ++;
	    }
	  else if (line[1] == 'C')
	    {
	      newSC   = (SC *)malloc(sizeof(SC));
	      sscanf(line, "%s %s %s %d",
		     type,
		     newSC->ipAddress,
		     newSC->fastIpAddress,
		     &newSC->port);
	      newSC->autoLaunch = js->autoLaunch;
	      strcpy(newSC->debugString, "");
	      if (totalSC == 0)
		{
		  js->scList = newSC;
		  lastSC = js->scList;
		}
	      else
		{
		  lastSC->next = newSC;
		  lastSC = lastSC->next;
		}
	      lastSC->next = NULL;
	      /*addAuxiliary(newSC->ipAddress, TYPE_SC);*/
	      totalSC ++;
	    }
	  else if (line[1] == 'N')
	    {
	      newCN = (CN *)malloc(sizeof(CN));
	      sscanf(line, "%s %d %s %s %s %d %s %d %s %d %s %d",
		     type,
		     &newCN->rank,
		     newCN->hostName,
		     newCN->ipAddress,
		     newCN->fastIpAddress,
		     &newCN->communicationPort,
		     newCN->eventLogger,
		     &newCN->eventLoggerPort,
		     newCN->checkpointServer,
		     &newCN->checkpointServerPort,
		     newCN->checkpointScheduler,
		     &newCN->checkpointSchedulerPort);
	      newCN->autoLaunch = js->autoLaunch;
	      strcpy(newCN->debugString, "");
	      if (totalCN == 0)
		{
		  js->nodeList = newCN;
		  lastCN = js->nodeList;
		}
	      else
		{
		  lastCN->next = newCN;
		  lastCN = lastCN->next;
		}
	      lastCN->next = NULL;
	      totalCN ++;
	    }
	}
      fclose(fd);
      js->nprocs = totalCN;
      i=0;
      ccn = js->nodeList;
      /* Are we using a high performance network? If so, we need to translate the addresses into the address world of the network */
      while (ccn != NULL)
	{
	  inet_aton(ccn->fastIpAddress, &(js->nodeListArray[i].sin_addr));
	  js->nodeListArray[i].sin_port = htons(ccn->communicationPort);
	  i++;
	  ccn = ccn->next;
	}
    }
  else
    {
      printf("Cannot open program file %s\n", js->v2pgFile);
      exit(1);
    }
}

void parseDebugFile(JS * js) {
  FILE * fd;
  char line[4096];
  char type[2];
  int  rank;
  char host[128];
  char ip1[15];
  char ip2[15];
  char ip3[15];
  char ip4[15];
  int  p0, p1, p2, p3;

  int i, j, jj, n, np, group, iPort;

  CN * currentCN;
  EL * currentEL;
  CS * currentCS;
  SC * currentSC;
  CN * ccn;
  char noauto[6];
  char debug[25];
  boolean autoLaunch;
  char c[6];

  currentCN = (CN *)malloc(sizeof(CN));
  currentEL = (EL *)malloc(sizeof(EL));
  currentCS = (CS *)malloc(sizeof(CS));
  currentSC = (SC *)malloc(sizeof(SC));
  ccn    = (CN *)malloc(sizeof(CN));

  if ((fd = fopen(js->debugFile, "r")) != NULL ) {
    while (fgets(line, 4096, fd) ) {
      j = 3;
      while (line[j] != ' ') {
	c[j-3] = line[j];
	j++;
      }
      i = atoi(c);

      jj = j+1;
      j = jj;
      while (line[j] != ' ') {
	noauto[j-jj] = line[j];
	j++;
      }

      jj = j+2;
      j = jj;
      while (line[j] != '"') {
        debug[j-jj] = line[j];
	j++;
      }

      autoLaunch = !(!(strcmp(noauto, "noauto")));
      if (line[1] == 'L') {
	currentEL = js->elList;
	for (j=0; j<i; j++)
	  currentEL = currentEL->next;
	currentEL->autoLaunch = autoLaunch;
        strcpy(currentEL->debugString, debug);
      } else if (line[1] == 'S') {
	currentCS = js->csList;
	for (j=0; j<i; j++)
          currentCS = currentCS->next;
	currentCS->autoLaunch = autoLaunch;
	strcpy(currentCS->debugString, debug);
      } else if (line[1] == 'C') {
	currentSC = js->scList;
	for (j=0; j<i; j++)
          currentSC = currentSC->next;
	currentSC->autoLaunch = autoLaunch;
	strcpy(currentSC->debugString, debug);
      } else if (line[1] == 'N') {
	currentCN = js->nodeList;
	for (j=0; j<i; j++)
          currentCN = currentCN->next;
	currentCN->autoLaunch = autoLaunch;
	strcpy(currentCN->debugString, debug);
      }
    }
  }
}

void createFifoFile(JS * js) {
  /*sprintf(js->fifoFile, "%s/%d:dispatcher.pipe", js->pwd, js->jobId);*/
  sprintf(js->fifoFile, "%s/1:dispatcher.pipe", js->pwd);
  mknod(js->fifoFile, S_IFIFO | S_IREAD | S_IWRITE, 0);
}

void moveExecutionOneNode(JS * js, int rank, char * hostName, char newIP[15], char newFastIP[15], int newPort) {
  CN * currentNode;
  currentNode = (CN *)malloc(sizeof(CN));

  if (rank > js->nprocs) return;

  currentNode = js->nodeList;
  while (currentNode != NULL) {
    if (currentNode->rank == rank) break;
    currentNode = currentNode->next;
  }

  /* We update the socket connections */

  /* We kill the node! */
  killComputingNode(js, currentNode);

  /* We update the structure of the computing node */
  strcpy(currentNode->ipAddress, newIP);
  strcpy(currentNode->fastIpAddress, newFastIP);
  currentNode->communicationPort = newPort;

  /* We update the 'nodes list array' */
  inet_aton(currentNode->fastIpAddress, &(js->nodeListArray[rank].sin_addr));
  js->nodeListArray[rank].sin_port = htons(currentNode->communicationPort);

}

void killComputingNode(JS * js, CN * node) {
  char command[256];
  char log[256];
  sprintf(command, "%s %s kill -9 %d ", js->rshCmd, node->ipAddress, node->pid);
  system(command);
  sprintf(log, "  killing node of rank %d on host %s", node->rank, node->hostName);
  v2logMessage(log);
}

void killAllAuxiliaries(JS * js) {
  EL * elc;
  CS * csc;
  SC * scc;
  char log[256];

  elc = js->elList;
  while (elc != NULL) {
    sprintf(log, "Killing EL on %s", elc->ipAddress);
    v2logMessage(log);
    killFromIPpID(js, elc->ipAddress, elc->pid);
    elc = elc->next;
  }
  csc = js->csList;
  while (csc != NULL) {
    sprintf(log, "Killing CS on %s", csc->ipAddress);
    v2logMessage(log);
    killFromIPpID(js, csc->ipAddress, csc->pid);
    csc = csc->next;
  }
  scc = js->scList;
  while (scc != NULL) {
    sprintf(log, "Killing SC on %s", scc->ipAddress);
    v2logMessage(log);
    killFromIPpID(js, scc->ipAddress, scc->pid);
    scc = scc->next;
  }
}

void killFromIPpID(JS * js, char * ip, pid_t pid) {
  char command[256];
  char log[256];
  sprintf(command, "%s %s kill -9 %d 2> /dev/null" , js->rshCmd, ip, pid);
  system(command);
  sprintf(log, "  killed (IP: %s PID: %d)", ip, pid);
}

void cleanAuxiliaryFiles(JS *js) {
  char filename[PATH_LENGTH];
  char *homePath;
  int i;
  homePath = strdup(getenv("HOME"));
  /* We remove first the files used for socket connection between nodes */
  /*  and the files used for checkpointing */
  for (i=0; i<js->nprocs; i++) {
    sprintf(filename, "%s/%d:%d.v2d.sock", homePath, js->jobId, i);
    unlink(filename);
    sprintf(filename, "%s/%d:%d.ckpt.pipe.tmp", homePath, js->jobId, i);
    unlink(filename);
    sprintf(filename, "%s/%d:%d.restart.pipe", homePath, js->jobId, i);
    unlink(filename);
    sprintf(filename, "%s/%d:%d.signal.pipe", homePath, js->jobId, i);
    unlink(filename);
  }
  /* Lastly, we remove the dispatcher pipe */

  sprintf(filename, "%s/%d:dispatcher.pipe", js->pwd, 1);
  unlink(filename);


  free(homePath);

}

void v2logMessage(char *s) {
  FILE * fd;
  char * logFile;
  time_t t;
  char *stime;
  boolean logging = true;

  if (!logging) return;
  stime = (char *)calloc(24, sizeof(char));

  t = time(NULL);
  strncpy(stime, ctime(&t), 24);
  logFile = (char *)calloc(512, sizeof(char));
  strcpy(logFile, "./mpich_v.log");
  if ((fd = fopen(logFile, "a")) != NULL) {
    fprintf(fd, "[ %s ] %s\n", stime, s);
    fclose(fd);
  }
  free(stime);
  free(logFile);
}