# include "pantasks.h"

/* local jobs are forked in the background 
   we might need to limit the maximum number of local jobs.
   should we have a queue/stack of pending local jobs, much
   like controller? */

/* update current state, drain stdout/stderr buffers */
int CheckLocalJob (Job *job) {

  int Nread;

  // XXX do something useful with exit status?
  CheckLocalJobStatus (job);

  if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) {
    EmptyIOBuffer (&job[0].stdout, 10, job[0].stdout_fd);
    EmptyIOBuffer (&job[0].stderr, 10, job[0].stderr_fd);
    close (job[0].stdout_fd);
    close (job[0].stderr_fd);
  } else {
    /* read stdout buffer */
    Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd);
    switch (Nread) {
      case -2:  /* error in read (programming error?  system level error?) */
	gprint (GP_ERR, "serious IO error\n");
	exit (2);
      case -1:  /* no data in pipe */
      case 0:   /* pipe is closed, change child state? **/
      default:  /* data in pipe */
	// fprintf (stderr, "read %d bytes (Nblock: %d, Nbuffer: %d)\n", Nread, job[0].stdout.Nblock, job[0].stdout.Nbuffer);
	break;
    }
  
    /* read stderr buffer */
    Nread = ReadtoIOBuffer (&job[0].stderr, job[0].stderr_fd);
    switch (Nread) {
      case -2:  /* error in read (programming error?  system level error?) */
	gprint (GP_ERR, "serious IO error\n");
	exit (2);
      case -1:  /* no data in pipe */
      case 0:   /* pipe is closed, change child state? **/
      default:  /* data in pipe */
	break;
    }
  }
  return (TRUE);
}

int CheckLocalJobStatus (Job *job) {

  int result, waitstatus;

  /* check local job status */
  result = waitpid (job[0].pid, &waitstatus, WNOHANG);
  switch (result) {
    case -1:  /* error with waitpid */
      switch (errno) {
	case ECHILD:
	  gprint (GP_ERR, "unknown PID, not a child proc\n");
	  gprint (GP_ERR, "did process already exit?  programming error?\n");
	  job[0].state = JOB_NONE;
	  job[0].exit_status = 0;
	  return (FALSE);
	case EINVAL:
	  gprint (GP_ERR, "error EINVAL (waitpid): programming error\n");
	  exit (1);
	case EINTR:
	  gprint (GP_ERR, "error EINTR (waitpid): programming error\n");
	  exit (1);
	default:
	  gprint (GP_ERR, "unknown error for waitpid (%d): programming error\n", errno);
	  exit (1);
      }
      break;
      
    case 0:  /* process not exited */
      job[0].state = JOB_BUSY;
      job[0].exit_status = 0;
      return (TRUE);

    default:
      if (result != job[0].pid) {
	gprint (GP_ERR, "waitpid error: mis-matched PID (%d vs %d).  programming error\n", result, job[0].pid);
	exit (1);
      }
      if (WIFEXITED(waitstatus)) {
	job[0].state = JOB_EXIT;
	job[0].exit_status = WEXITSTATUS(waitstatus);
      }
      if (WIFSIGNALED(waitstatus)) {
	job[0].state = JOB_CRASH;
	job[0].exit_status = WTERMSIG(waitstatus);
      }
      if (WIFSTOPPED(waitstatus)) {
	gprint (GP_ERR, "waitpid returns 'stopped': programming error\n");
	exit (1);
      }
      job[0].dtime = GetTaskTimer (job[0].start);
      break;
  }
  return (FALSE);
}

/* this could be written a just a one-way pipe */
int SubmitLocalJob (Job *job) {

  int status, pid;
  int stdout_fd[2], stderr_fd[2];

  bzero (stdout_fd, 2*sizeof(int));
  bzero (stderr_fd, 2*sizeof(int));

  if (pipe (stdout_fd) < 0) goto pipe_error;
  if (pipe (stderr_fd) < 0) goto pipe_error;

  pid = fork ();
  if (!pid) { /* must be child process */
    if (VerboseMode()) gprint (GP_ERR, "starting local job\n");

    /* close the other ends of the pipes */
    close (stdout_fd[0]);
    close (stderr_fd[0]);

    /* tie our ends of the pipes to stdin, stdout, stderr */
    dup2 (stdout_fd[1], STDOUT_FILENO);
    dup2 (stderr_fd[1], STDERR_FILENO);

    /* set all three unblocking */
    setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ);
    setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ);

    status = execvp (job[0].argv[0], job[0].argv); 
    exit (1);
  }

  /* close the other ends of the pipes */
  close (stdout_fd[1]); stdout_fd[1] = 0;
  close (stderr_fd[1]); stderr_fd[1] = 0;

  /* make the pipes non-blocking */
  fcntl (stdout_fd[0], F_SETFL, O_NONBLOCK);
  fcntl (stderr_fd[0], F_SETFL, O_NONBLOCK);

  job[0].stdout_fd = stdout_fd[0];
  job[0].stderr_fd = stderr_fd[0];
  job[0].pid = pid;

  return (TRUE);

pipe_error:
  perror ("pipe error:");
  if (stdout_fd[0] != 0) close (stdout_fd[0]);
  if (stdout_fd[1] != 0) close (stdout_fd[1]);
  if (stderr_fd[0] != 0) close (stderr_fd[0]);
  if (stderr_fd[1] != 0) close (stderr_fd[1]);
  return (FALSE);
}

/* should this function close the fd's? */
int KillLocalJob (Job *job) {

  int i, result, waitstatus;

  if (job[0].state != JOB_BUSY) return (TRUE);

  /* send SIGTERM signal to job */
  kill (job[0].pid, SIGTERM);
  result = 0;
  for (i = 0; (i < 10) && (result == 0); i++) {
    usleep (10000);  // wait for job to exit
    result = waitpid (job[0].pid, &waitstatus, WNOHANG);
  }
  if (result) return (TRUE);

  /* send SIGKILL signal to job */
  kill (job[0].pid, SIGKILL);
  result = 0;
  for (i = 0; (i < 10) && (result == 0); i++) {
    usleep (10000);  // wait for job to exit
    result = waitpid (job[0].pid, &waitstatus, WNOHANG);
  }
  if (result) return (TRUE);

  /* total failure, don't reset */
  return (FALSE);
}

