Index: trunk/Ohana/src/opihi/pantasks/LocalJob.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/LocalJob.c	(revision 3392)
+++ trunk/Ohana/src/opihi/pantasks/LocalJob.c	(revision 4450)
@@ -1,17 +1,150 @@
+# include "scheduler.h"
 
-/* local jobs are forked in the background */
+/* this could be written a just a one-way pipe */
+int SubmitLocalJob (Job *job) {
 
-SubmitLocalJob (Job *job) {
+  int status, pid;
+  int stdout_fd[2], stderr_fd[2];
 
-  /* 
-     construct the command line
-     fork the command, get back the PID
-     increment local job counter 
-  */
+  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 */
+    fprintf (stderr, "starting controller connection\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);
 }
 
-CheckLocalJob (Job *job) {
+/* update current state, drain stdout/stderr buffers */
+int CheckLocalJob (Job *job) {
 
-  /* 
-     
+  int Nread;
+
+  // XXX do something useful with exit status?
+  CheckLocalJobStatus (job);
+
+  /* read stdout buffer */
+  Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd);
+  switch (Nread) {
+    case -2:  /* error in read (programming error?  system level error?) */
+      fprintf (stderr, "serious IO error\n");
+      exit (2);
+    case -1:  /* no data in pipe */
+      break;
+    case 0:   /* pipe is closed */
+      /** change child state? **/
+      break;
+    default:  /* data in pipe */
+      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?) */
+      fprintf (stderr, "serious IO error\n");
+      exit (2);
+    case -1:  /* no data in pipe */
+      break;
+    case 0:   /* pipe is closed */
+      /** change child state? **/
+      break;
+    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:
+	  fprintf (stderr, "unknown PID, not a child proc\n");
+	  fprintf (stderr, "did process already exit?  programming error?\n");
+	  job[0].state = JOB_NONE;
+	  job[0].exit_status = 0;
+	  return (FALSE);
+	case EINVAL:
+	  fprintf (stderr, "error EINVAL (waitpid): programming error\n");
+	  exit (1);
+	case EINTR:
+	  fprintf (stderr, "error EINTR (waitpid): programming error\n");
+	  exit (1);
+	default:
+	  fprintf (stderr, "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) {
+	fprintf (stderr, "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)) {
+	fprintf (stderr, "waitpid returns 'stopped': programming error\n");
+	exit (1);
+      }
+  }
+  return;
+}