checkpointserverproto.c

Path MPICH-V for MPICH the MPI Implementation

      }
      else if(ret == 0)
      {
        printi("CS_proto", "Data channel connection closed for checkpoint %d:%d(%d) : %d data bytes read %d", group, rank, seq, dataread);
        close(sdata);
        sdata = -1;
        continue;
      }
      if(datawriteCheckpointFile(f, protosize, dataread, buff, ret) != ret)
      {
        printw("Writting to checkpoint file %d:%d(%d) failed", group, rank, seq);
        discardCheckpointFile(f, group, rank, seq);
        printq("Aborting due to internat write failure");
      }
      dataread += ret;
    }
  }

  if(read(sproto, buff, 1) != 1)
  {
    printe("Failure reading confirm message from %d:%d(%d)", group, rank, seq);
    discardCheckpointFile(f, group, rank, seq);
    printq("Aborting due to read failure (probably client failure)");
  }
  if(*buff == 'C')
  {
    if(read(sproto, buff, sizeof(int)) != sizeof(int))
    {
      printe("Failure reading confirm message from %d:%d(%d)", group, rank, seq);
      discardCheckpointFile(f, group, rank, seq);
      printq("Aborting due to read failure (probably client failure)");
    }
    datasize = ntohl(*((int *)buff));
    while(dataread < datasize)
    {
      if(sdata == -1)
      {
        printw("Not enough data were available in data channel to finish reception of checkpoint %d:%d(%d) (read %d of %d)", group, rank, seq, dataread, datasize);
        discardCheckpointFile(f, group, rank, seq);
        printq("Aborting due to incomplete data (probably client or network failure)");
      }
      if((ret = read(sdata, buff, BUFFSIZE)) < 0)
      {
        printe("Read failed on data channel during reception of checkpoint %d:%d(%d)", group, rank, seq);
        discardCheckpointFile(f, group, rank, seq);
        printq("Aborting due to read failure (probably client failure)");
      }
      else if(ret == 0)
      {
        printi("CS_proto", "Data channel connection closed for checkpoint %d:%d(%d) : %d data bytes read %d", group, rank, seq, dataread);
        close(sdata);
        sdata = -1;
        continue;
      }
      if(datawriteCheckpointFile(f, protosize, dataread, buff, ret) != ret)
      {
        printw("Writting to checkpoint file %d:%d(%d) failed", group, rank, seq);
        discardCheckpointFile(f, group, rank, seq);
        printq("Aborting due to internat write failure");
      }
      dataread += ret;
      if(dataread > datasize)
      {
        printw("To much data were read from client %d:%d(%d) (read %d waited %d)", group, rank, seq, dataread, datasize);
        discardCheckpointFile(f, group, rank, seq);
        printq("Aborting due to inconsistancy in data size");
      }
    }
    if(datasizewriteCheckpointFile(f, protosize, datasize) < 0)
    {
      printw("Could not write datasize=%d to image file %d:%d(%d)", datasize, group, rank, seq);
      discardCheckpointFile(f, group, rank, seq);
      printq("Aborting due to internal write failure");
    }
    confirmCheckpointFile(f, group, rank, seq);
    printi("CS", "Sending bak confirmation ACK to client %d:%d(%d)", group, rank, seq);
    if(write(sproto, &c, 1) != 1)
    {
      qerror("Failure sending confirmation ACK to client %d:%d(%d) (probably due to client failure), checkpoint saved", group, rank, seq);
    }
  }
  else if(*buff == 'T')
  {
    printw("Client %d:%d(%d) is requesting a Third party confirm, not implemented", group, rank, seq);
    discardCheckpointFile(f, group, rank, seq);
    printq("Aborting due to call to non implemented failure");
  }
  else
  {
    printw("Client %d:%d(%d) sent an invalid confirm message \"%d\"", group, rank, seq, *buff);
    discardCheckpointFile(f, group, rank, seq);
    printq("Aborting due to invalid confirm message");
  }
  printi("CS", "Checkpoint %d finished for client %d:%d (protosize=%d datasize=%d)", seq, group, rank, protosize, datasize);
  return 0;
}

static int del_checkpoint(int sproto, int sdata)
{
  int group, rank, seq;
  int minseq, maxseq;

  switch(recv_del_ids(sproto, &group, &rank, &seq))
  {
    case DEL_UPTO : 
      printi("CS_del", "removing checkpoint files from repository up to %d:%d#%d %s", group, rank, seq);
      if(delCheckpointFileUpto(group, rank, seq) < 0)
      {
        printi("CS_del", "Failure notification to client : files up to %d:d#%d deletion failed", group, rank, seq);
        *buff = 'F';
      }
      else
      {
        printi("CS_del", "Success notification to client : files up to %d:%d#%d deleted", group, rank, seq);
        *buff = 'A';
      }
      *((int *) (buff + sizeof(char))) = htonl(minseq);
      *((int *) (buff + sizeof(char) + sizeof(int))) = htonl(maxseq);
      if(write(sproto, buff, sizeof(char) + 2 * sizeof(int)) != (sizeof(char) + 2 * sizeof(int))) qerror("Sending del notification to client %d:%d#%d", group, rank, seq);
      return 0;
    case DEL_EXACT : 
      if(delCheckpointFileExact(group, rank, seq) < 0) 
      {
        printi("CS_del", "Failure notification to client : file %d:%d#%d still exists", group, rank, seq);
        *buff = 'F';
      }
      else
      {
        printi("CS_del", "Success notification to client : files up %d:%d#%d deleted", group, rank, seq);
        *buff = 'A';
      }
      *((int *) (buff + sizeof(char))) = htonl(minseq);
      *((int *) (buff + sizeof(char) + sizeof(int))) = htonl(maxseq);
      if(write(sproto, buff, sizeof(char) + 2 * sizeof(int)) != (sizeof(char) + 2 * sizeof(int))) qerror("Sending del notification to client %d:%d#%d", group, rank, seq);
      return 0;     
    default : 
      return -1;
  }
}

/*
int send_memory_channel_ack(char *memserv,int wid,int rankid)
{
  int s;
  struct sockaddr_in addr;
  socklen_t saddr;
  struct hostent *he;
  int msg[2];

  if((s = socket(AF_INET,SOCK_STREAM,0)) == -1) {printe("socket");return -1;}
  if((he = gethostbyname(memserv)) == NULL) {herror("resolving channel memory server name");return -1;}

  addr.sin_family = AF_INET;
  memcpy(&addr.sin_addr.s_addr,he->h_addr,he->h_length);
  addr.sin_port = htons(CHANNEL_MEM_PORT);

  if(connect(s,(struct sockaddr *) &addr,sizeof(addr)) == -1) {printe("contacting channel memory server");return -1;}

  wid = htonl(wid);
  rankid = htonl(rankid);
  msg[0]=wid;
  msg[1]=rankid;
  if(write(s,msg,sizeof(int) * 2) != (sizeof(int) * 2)) {printe("Sending wid and rankid info to memory server");return -1;}
  if(read(s,msg,sizeof(int)) == -1) {printe("Receiving ACK from memory server");return -1;}
  if(! msg[0]) {printi("CS-proto","Memory server sent failure notification");return -1;}
  close(s);
  return 0;
}
*/

static int recv_type_message(int s)
{
  int c = 0;

  switch(read(s, &c, 1))
  {
    case 0 :
      printi("CS_proto", "Request is empty, aborting");
      return ERROR_SYNTAX;
    case -1 :
      printe("Request failed during type determination");
      return ERROR_CO;
  }
  switch(c)
  {
    case 'G' :
      printi("CS_proto", "Request is a GET");
      return GET;
    case 'P' :
      printi("CS_proto", "Request is a PUT");
      return PUT;
    case 'D' :
      printi("CS_proto", "Request is a DELETE");
      return DEL;
    default :
      printi("CS_proto", "Type of request %c is invalid", c);
      return ERROR_SYNTAX;
  }
}

static int recv_get_ids(int s, int *wid, int *rankid, int *seq)
{
  if(read(s, buff, sizeof(int) * 3) != (sizeof(int) * 3))
  {
    printe("error reading group, rank");
    return -1;
  }
  *wid = ntohl(*((int *) buff));
  *rankid = ntohl(*((int *) (buff + sizeof(int))));
  *seq = ntohl(*((int *) (buff + 2 * sizeof(int))));
  printi("CS_get", "Client wants %d:%d#%d checkpoint image", *wid, *rankid, *seq);
  return 2;
}

static int recv_put_ids(int s,int *wid,int *rankid, int *seq, int *protosize)
{
  if(read(s, buff, sizeof(int) * 4) != (sizeof(int) * 4))
  {
    printe("error reading group, rank, seq, protosize");
    return -1;
  }
  *wid = ntohl(*((int *) buff));
  *rankid = ntohl(*((int *)(buff + sizeof(int))));
  *seq = ntohl(*((int *)(buff + 2 * sizeof(int))));
  *protosize = ntohl(*((int *)(buff + 3 * sizeof(int))));
  printi("CS", "Client %d:%d is sending checkpoint #%d, protocol implementation data is size %d", *wid, *rankid, *seq, *protosize);
  return 0;
}

static char recv_del_ids(int s, int *group, int *rank, int *seq)
{
  if(read(s, buff, sizeof(char) + 3 * sizeof(int)) != (sizeof(char) + 3 * sizeof(int)))
  {
    printe("error reading group, rank");
    return -1;
  }
  *group = ntohl(*((int *) (buff + sizeof(char))));
  *rank = ntohl(*((int *) (buff + sizeof(char) + sizeof(int))));
  *seq = ntohl(*((int *) (buff + sizeof(char) + 2 * sizeof(int))));
  if((*buff != DEL_EXACT) && (*buff != DEL_UPTO)) 
  {
    printw("Invalid protocol literal %c", *buff);
    return -1;
  }
  printi("CS_del", "Client wants to delete %s %d:%d#%d checkpoint image", "U", *group, *rank, *seq);
  return *buff;

}

static int qsend_protofailure(int s, int group, int rank)
{
  char c = 'F';
  if(write(s, &c, 1) != 1) qerror("Failed sending failure notification to client %d:%d", group, rank);
  close(s);
  printq("Failure notification sent to client %d:%d : aborting", group, rank);
}