Index: trunk/Ohana/src/opihi/pantasks/CheckJobs.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/CheckJobs.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/CheckJobs.c	(revision 4450)
@@ -0,0 +1,74 @@
+# include "scheduler.h"
+
+int CheckJobs () {
+
+  Job *job;
+  Task *task;
+  Macro *macro;
+  int i, found, status;
+  int Ntest;
+
+  /** test all jobs: ready to test?  finished? **/
+  while ((job = NextJob ()) != NULL) {
+
+    /* check for timeout */
+    if (GetTaskTimer(job[0].start) >= job[0].task[0].timeout_period) {
+      fprintf (stderr, "timeout on %s\n", job[0].task[0].name);
+      /* run task[0].timeout macro, if it exists */
+      if (job[0].task[0].timeout != NULL) {
+	exec_loop (job[0].task[0].timeout);
+      }
+      DeleteJob (job);
+      continue;
+    }
+
+    /* check poll period (ready to run again?) */
+    if (GetTaskTimer(job[0].last) < job[0].task[0].poll_period) continue;
+
+    /* check current status */
+    status = CheckJob (job);
+    switch (status) {
+      case JOB_BUSY:
+	fprintf (stderr, "job %s (%d) busy\n", job[0].task[0].name, job[0].JobID);
+	break;
+
+      case JOB_CRASH:
+	fprintf (stderr, "job %s (%d) crash\n", job[0].task[0].name, job[0].JobID);
+	/* run task[0].crash macro, if it exists */
+	/* set the stdout and stderr variables with job.stdout, job.stderr */
+	/* XXX this will break on 0 values in output streams */
+	set_str_variable ("stdout", job[0].stdout.buffer);
+	set_str_variable ("stderr", job[0].stderr.buffer);
+	if (job[0].task[0].crash != NULL) {
+	  exec_loop (job[0].task[0].crash);
+	}
+	DeleteJob (job);
+	continue;
+	break;
+
+      case JOB_EXIT:
+	fprintf (stderr, "job %s (%d) exit\n", job[0].task[0].name, job[0].JobID);
+	/* run corresponding task[0].exit macro, if it exists */
+	macro = job[0].task[0].def;
+	for (i = 0; i < job[0].task[0].Nexit; i++) {
+	  if (job[0].exit_status == atoi(job[0].task[0].exit[i][0].name)) {
+	    macro = job[0].task[0].exit[i];
+	    break;
+	  }
+	}
+	if (macro != NULL) exec_loop (macro);
+	DeleteJob (job);
+	continue;
+	break;
+
+      default:
+	fprintf (stderr, "unknown exit status\n");
+	/** do something more useful here ?? **/
+	break;
+    }
+
+    /* reset polling clock */
+    SetTaskTimer (&job[0].last);
+  }
+  return (TRUE);
+}
Index: trunk/Ohana/src/opihi/pantasks/CheckSystem.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/CheckSystem.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/CheckSystem.c	(revision 4450)
@@ -0,0 +1,8 @@
+# include "scheduler.h"
+
+int CheckSystem () {
+
+  CheckTasks ();
+  CheckJobs ();
+
+}
Index: trunk/Ohana/src/opihi/pantasks/CheckTasks.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/CheckTasks.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/CheckTasks.c	(revision 4450)
@@ -0,0 +1,39 @@
+# include "scheduler.h"
+
+int CheckTasks () {
+
+  Job *job;
+  Task *task;
+  Macro *macro;
+  int i, found, status;
+  int Ntest;
+
+  /** test all tasks: ready to test? ready to run? **/
+  while ((task = NextTask ()) != NULL) {
+
+    /* ready to test? : check exec period */
+    if (GetTaskTimer(task[0].last) < task[0].exec_period) continue;
+
+    SetCurrentTask (task[0].name);
+    fprintf (stderr, "trying task %s\n", task[0].name);
+
+    /* ready to run? : run task.exec macro */
+    if (task[0].exec != NULL) {
+      status = exec_loop (task[0].exec);
+      if (!status) continue;
+    }
+
+    /* is task valid?  check state of task.(argc, argv) */
+    /*** ADD CODE HERE ***/
+
+    /* construct job from task */
+    job = CreateJob (task);
+
+    /* execute job - XXX add status test */
+    SubmitJob (job);
+
+    /* reset timer on task (don't do this if Create/Submit fails)*/
+    gettimeofday (&task[0].last, (void *) NULL);
+  }
+  return (TRUE);
+}
Index: trunk/Ohana/src/opihi/pantasks/ControllerOps.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/ControllerOps.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/ControllerOps.c	(revision 4450)
@@ -0,0 +1,322 @@
+# include "scheduler.h"
+# define CONTROLLER_TIMEOUT 20
+# define CONNECT_TIMEOUT 300
+
+int status = FALSE;
+int stdin_cntl, stdout_cntl, stderr_cntl;
+
+/* we read Nbytes from the host, then watch for the prompt */ 
+int GetJobOutput (char *cmd, int pid, IOBuffer *buffer, int Nbytes) {
+  
+  int i, status, Nstart;
+  char *line;
+
+  /* flush any earlier messages */
+  ReadtoIOBuffer (buffer, stdout_cntl);
+  FlushIOBuffer (buffer);
+  Nstart = buffer[0].Nbuffer;
+
+  /* send command to get appropriate channel */
+  ALLOCATE (line, char, MAX (1, strlen(cmd) + 15));
+  sprintf (line, "%s %d\n", cmd, pid);
+  status = write (stdin_cntl, line, strlen(line));
+  free (line);
+
+  /* is pipe still open? */
+  if ((status == -1) && (errno == EPIPE)) return (CONTROLLER_DOWN);
+
+  /* read at least Nbytes, then watch for CONTROLLER_PROMPT */
+  line = NULL;
+  status = -1;
+  for (i = 0; (i < CONTROLLER_TIMEOUT) && (status == -1) && (line == NULL); i++) {
+    status = ReadtoIOBuffer (buffer, stdout_cntl);
+    if ((buffer[0].Nbuffer - Nstart) >= Nbytes) {
+      line = memstr (buffer[0].buffer, CONTROLLER_PROMPT, buffer[0].Nbuffer);
+    }
+    if (status == -1) usleep (10000);
+  }
+  if (status ==  0) return (CONTROLLER_DOWN);
+  if (status == -1) return (CONTROLLER_HUNG);
+
+  fprintf (stderr, "message received (GetJobOutput : %s)\n", cmd);  
+  /* drop extra bytes from pcontrol (not pclient:job) */
+  buffer[0].Nbuffer = Nstart + Nbytes;
+  if (buffer[0].Nalloc > buffer[0].Nbuffer) {
+    bzero (buffer[0].buffer + buffer[0].Nbuffer, buffer[0].Nalloc - buffer[0].Nbuffer);
+  }
+  return (CONTROLLER_GOOD);
+}
+
+int CheckControllerJob (Job *job) {
+
+  CheckControllerJobStatus (job);
+
+  if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) {
+    GetJobOutput ("stdout", job[0].pid, &job[0].stdout, job[0].stdout_size);
+    GetJobOutput ("stderr", job[0].pid, &job[0].stderr, job[0].stderr_size);
+  }  
+}
+
+int CheckControllerJobStatus (Job *job) {
+
+  int outstate;
+  char cmd[128], status_string[64];
+  char *p;
+  IOBuffer buffer;
+  
+
+  sprintf (cmd, "check job %d", job[0].pid);
+  InitIOBuffer (&buffer, 0x100);
+  status = ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer);
+
+  /* check on success of controller command */
+  switch (status) {
+    case CONTROLLER_DOWN:
+      fprintf (stderr, "controller is down\n");
+      FreeIOBuffer (&buffer);
+      return (FALSE);
+
+    case CONTROLLER_HUNG:
+      fprintf (stderr, "controller is not responding\n");
+      FreeIOBuffer (&buffer);
+      return (FALSE);
+
+    case CONTROLLER_GOOD:
+      fprintf (stderr, "message received (CheckControllerJobStatus)\n");
+      break;
+
+    default:
+      fprintf (stderr, "unknown status for controller command: programming error\n");  
+      exit (1);
+  }
+
+  /** was this a valid job? **/
+  p = memstr (buffer.buffer, "job not found", buffer.Nbuffer);
+  if (p != NULL) {
+    fprintf (stderr, "unknown job %d\n", job[0].pid);
+    return (FALSE);
+  }
+
+  /** parse status message **/
+  p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
+  sscanf (p, "%*s %s", status_string);
+  p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer);
+  sscanf (p, "%*s %d", &job[0].exit_status);
+  p = memstr (buffer.buffer, "STDOUT", buffer.Nbuffer);
+  sscanf (p, "%*s %d", &job[0].stdout_size);
+  p = memstr (buffer.buffer, "STDERR", buffer.Nbuffer);
+  sscanf (p, "%*s %d", &job[0].stderr_size);
+  FreeIOBuffer (&buffer);
+
+  /* possible exit status values */
+  outstate = -1;
+  if (!strcmp(status_string, "BUSY"))    outstate = JOB_BUSY;
+  if (!strcmp(status_string, "DONE"))    outstate = JOB_BUSY;
+  if (!strcmp(status_string, "PENDING")) outstate = JOB_PENDING;
+  if (!strcmp(status_string, "EXIT"))    outstate = JOB_EXIT;
+  if (!strcmp(status_string, "CRASH"))   outstate = JOB_CRASH;
+  if (outstate == -1) {
+    fprintf (stderr, "programming error?\n");
+    exit (1);
+  }
+  job[0].state = outstate;
+  return (TRUE);
+}
+
+/* submitting a job to the controller automatically starts controller */
+int SubmitControllerJob (Job *job) {
+
+  int i, Nchar;
+  char *cmd, *p, string[64];
+  IOBuffer buffer;
+
+  if (job[0].task[0].host == NULL) return (FALSE); 
+
+  StartController ();   // XXX check for success 
+
+  /** construct the line to be sent to the controller **/
+
+  /* determine the total line length */
+  Nchar = 0;
+  for (i = 0; i < job[0].argc; i++) {
+    Nchar += strlen (job[0].argv[i]) + 1;
+  }
+  if (job[0].task[0].host) {
+    Nchar += strlen (job[0].task[0].host) + 1;
+  }
+  Nchar += 10;
+  ALLOCATE (cmd, char, Nchar);
+  bzero (cmd, Nchar);
+
+  /* construct the controller command portion */
+  if (!strcmp (job[0].task[0].host, "NONE")) {
+    sprintf (cmd, "job", cmd);
+  } else {
+    if (job[0].task[0].host_required) {
+      sprintf (cmd, "job +host %s", job[0].task[0].host);
+    } else {
+      sprintf (cmd, "job -host %s", job[0].task[0].host);
+    }
+  }
+
+  /* add the command arguments */
+  for (i = 0; i < job[0].task[0].argc; i++) {
+    strcat (cmd, " ");
+    strcat (cmd, job[0].task[0].argv[i]);
+  }
+
+  InitIOBuffer (&buffer, 0x100);
+  ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer);
+  free (cmd);
+
+  /* extract the job PID from the controller response */
+  p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
+  if (p == NULL) {
+    fprintf (stderr, "missing PID in pcontrol message : programming error\n");
+    exit (1);
+  }
+  sscanf (p, "%*s %s", string);
+  job[0].pid = atoi (string);
+  FreeIOBuffer (&buffer);
+  return (TRUE);
+}
+
+int StartController () {
+
+  char *p;
+  char **argv, cmd[128];
+  int i, pid;
+  int stdin_fd[2], stdout_fd[2], stderr_fd[2];
+  IOBuffer buffer;
+
+  if (status) return (TRUE);
+
+  if (VarConfig ("CONTROLLER", "%s", cmd) == NULL) strcpy (cmd, "pcontrol");
+
+  bzero (stdin_fd,  2*sizeof(int));
+  bzero (stdout_fd, 2*sizeof(int));
+  bzero (stderr_fd, 2*sizeof(int));
+
+  if (pipe (stdin_fd)  < 0) goto pipe_error;
+  if (pipe (stdout_fd) < 0) goto pipe_error;
+  if (pipe (stderr_fd) < 0) goto pipe_error;
+
+  ALLOCATE (argv, char *, 2);
+  argv[0] = cmd;
+  argv[1] = 0;
+
+  pid = fork ();
+  if (!pid) { /* must be child process */
+    fprintf (stderr, "starting controller connection\n");
+
+    /* close the other ends of the pipes */
+    close (stdin_fd[1]);
+    close (stdout_fd[0]);
+    close (stderr_fd[0]);
+
+    /* tie our ends of the pipes to stdin, stdout, stderr */
+    dup2 (stdin_fd[0],  STDIN_FILENO);
+    dup2 (stdout_fd[1], STDOUT_FILENO);
+    dup2 (stderr_fd[1], STDERR_FILENO);
+
+    /* set all three unblocking */
+    setvbuf (stdin,  (char *) NULL, _IONBF, BUFSIZ);
+    setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ);
+    setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ);
+
+    status = execvp (argv[0], argv); 
+    exit (1);
+  }
+  free (argv);
+
+  /* close the other ends of the pipes */
+  close (stdin_fd[0]);  stdin_fd[0]  = 0;
+  close (stdout_fd[1]); stdout_fd[1] = 0;
+  close (stderr_fd[1]); stderr_fd[1] = 0;
+
+  /* make the pipes non-blocking */
+  fcntl (stdin_fd[1],  F_SETFL, O_NONBLOCK);
+  fcntl (stdout_fd[0], F_SETFL, O_NONBLOCK);
+  fcntl (stderr_fd[0], F_SETFL, O_NONBLOCK);
+
+  /* perform handshake with controller to verify alive & running */
+  /** this handshake is similar to ControllerCommand, but has important differences **/
+  InitIOBuffer (&buffer, 0x100);
+
+  /* send handshake command */
+  sprintf (cmd, "echo CONNECTED\n");
+  status = write (stdin_fd[1], cmd, strlen(cmd));
+  if ((status == -1) && (errno == EPIPE)) goto pipe_error;
+
+  /* try to get evidence connection is alive - wait upto a few seconds */
+  p = NULL;
+  status = -1;
+  for (i = 0; (i < CONNECT_TIMEOUT) && (status == -1) && (p == NULL); i++) {
+    status = ReadtoIOBuffer (&buffer, stdout_fd[0]);
+    p = memstr (buffer.buffer, "CONNECTED", buffer.Nbuffer);
+    usleep (10000);
+  }
+  if (status == 0) goto pipe_error;
+  if (status == -1) goto io_error;
+  FreeIOBuffer (&buffer);
+
+  /* set local static vars to pipe connections */
+  stdin_cntl  = stdin_fd[1];
+  stdout_cntl = stdout_fd[0];
+  stderr_cntl = stderr_fd[0];
+
+  fprintf (stderr, "Connected\n");
+  return (TRUE);
+
+pipe_error:
+  perror ("pipe error:");
+  goto close_pipes;
+
+io_error:
+  fprintf (stderr, "timeout while connecting\n");
+  goto close_pipes;
+
+close_pipes:
+  if (stdin_fd[0]  != 0) close (stdin_fd[0]);
+  if (stdin_fd[1]  != 0) close (stdin_fd[1]);
+  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);
+}
+
+int ControllerCommand (char *cmd, char *response, IOBuffer *buffer) {
+
+  int i;
+  int status;
+  char *line;
+
+  ReadtoIOBuffer (buffer, stdout_cntl);
+  FlushIOBuffer (buffer);
+
+  /* send command to client (adding on \n) */
+  /* fprintf (stderr, "send: %s (%d)\n", command, buffer[0].Nbuffer); */
+  ALLOCATE (line, char, MAX (1, strlen(cmd)));
+  sprintf (line, "%s\n", cmd);
+  status = write (stdin_cntl, line, strlen(line));
+  free (line);
+
+  /* is pipe still open? */
+  if ((status == -1) && (errno == EPIPE)) return (CONTROLLER_DOWN);
+  
+  /* watch for response - wait up to 1 second */
+  line = NULL;
+  status = -1;
+  for (i = 0; (i < CONTROLLER_TIMEOUT) && (status == -1) && (line == NULL); i++) {
+    status = ReadtoIOBuffer (buffer, stdout_cntl);
+    line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer);
+    if (status == -1) usleep (10000);
+  }
+  if (status ==  0) return (CONTROLLER_DOWN);
+  if (status == -1) return (CONTROLLER_HUNG);
+  /* fprintf (stderr, "buffer.buffer: %s\n", buffer[0].buffer); */
+  return (CONTROLLER_GOOD);
+}
+
+/* memstr returns a view, not an allocated string : don't free */
Index: trunk/Ohana/src/opihi/pantasks/IOBufferOps.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/IOBufferOps.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/IOBufferOps.c	(revision 4450)
@@ -0,0 +1,68 @@
+# include "pclient.h"
+
+int InitIOBuffer (IOBuffer *buffer, int Nalloc) {
+
+  buffer[0].Nalloc = Nalloc;
+  buffer[0].Nreset = Nalloc;
+  buffer[0].Nblock = Nalloc / 2;
+  buffer[0].Nbuffer = 0;
+
+  ALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc);
+  bzero (buffer[0].buffer, buffer[0].Nalloc);
+
+  return (TRUE);
+}
+
+int FlushIOBuffer (IOBuffer *buffer) {
+
+  buffer[0].Nbuffer = 0;
+  buffer[0].Nalloc = buffer[0].Nreset;
+  REALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc);
+
+  return (TRUE);
+}
+
+int ReadtoIOBuffer (IOBuffer *buffer, int fd) {
+
+  int Nread, Nfree;
+
+  if (fd == 0) {
+    fprintf (stderr, "pipe is closed (ReadIOBuffer)\n");
+    return (0);
+    /* pipe is closed */
+  }
+
+  Nfree = buffer[0].Nalloc - buffer[0].Nbuffer;
+  if (Nfree < buffer[0].Nblock) {
+    buffer[0].Nalloc += 2*buffer[0].Nblock;
+    REALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc);
+    Nfree = buffer[0].Nalloc - buffer[0].Nbuffer;
+  }
+
+  Nread = read (fd, &buffer[0].buffer[buffer[0].Nbuffer], buffer[0].Nblock);
+
+  if (Nread >= 0) {
+    buffer[0].Nbuffer += Nread;
+    return (Nread);
+  }
+
+  if (Nread == -1) {
+    switch (errno) {
+    case EAGAIN:
+    case EIO:
+      /** no data available in pipe **/
+      return (-1);
+    default:
+      /** error reading from pipe **/
+      perror ("ReadtoIOBuffer read error");
+      return (-2);
+    }
+  }
+}
+
+void FreeIOBuffer (IOBuffer *buffer) {
+
+  if (buffer[0].buffer != (char *) NULL) {
+    free (buffer[0].buffer);
+  }
+}
Index: trunk/Ohana/src/opihi/pantasks/JobOps.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/JobOps.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/JobOps.c	(revision 4450)
@@ -1,3 +1,2 @@
-# include "opihi.h"
 # include "scheduler.h"
 
@@ -9,6 +8,6 @@
 
 static Job *jobs;
-static int   Njobs;
-static int   NJOBS;
+static int  Njobs;
+static int  NJOBS;
 
 static char *JobName = dot;
@@ -65,5 +64,5 @@
 
   for (i = 0; i < Njobs; i++) {
-    fprintf (stderr, "%d: %-15s %5d %20s (%x)\n", Njobs, jobs[i].task.name, jobs[i].JobID, jobs[i].task.argv[0], jobs[i].task.argv);
+    fprintf (stderr, "%d: %-15s %5d %20s (%x)\n", Njobs, jobs[i].task[0].name, jobs[i].JobID, jobs[i].argv[0], jobs[i].argv);
   }
   return;
@@ -75,5 +74,5 @@
   int i;
 
-  /* try for an exact match first */
+  /* return job with matching JobID */
   for (i = 0; i < Njobs; i++) {
     if (jobs[i].JobID == JobID) {
@@ -90,13 +89,31 @@
 
   jobs[Njobs].JobID = NextJobID ();
-  jobs[Njobs].PID = 0;
-  jobs[Njobs].task = task[0];
-
-  /* we need our own copy of task[0].argv */
-  ALLOCATE (jobs[Njobs].task.argv, char *, MAX (jobs[Njobs].task.argc, 1));
-  for (i = 0; i < jobs[Njobs].task.argc; i++) {
-    jobs[Njobs].task.argv[i] = strcreate (task[0].argv[i]);
-  }
-  jobs[Njobs].task.host = strcreate (task[0].host);
+  jobs[Njobs].pid = 0;
+  jobs[Njobs].mode = JOB_LOCAL;
+  if (task[0].host != NULL) {
+    jobs[Njobs].mode = JOB_CONTROLLER;
+  }
+
+  /* we need our own copy of task[0].argv 
+   *  argc is the number of valid args, like the usual command line.
+   *  we allocate one extra element, with value 0 to be passed to execvp
+   */
+  jobs[Njobs].argc = task[0].argc;
+  ALLOCATE (jobs[Njobs].argv, char *, MAX (task[0].argc + 1, 1));
+  for (i = 0; i < task[0].argc; i++) {
+    jobs[Njobs].argv[i] = strcreate (task[0].argv[i]);
+  }
+  jobs[Njobs].argv[i] = 0;
+
+  /* other data from the task is needed by the job 
+     we carry a pointer back to the task.  this means we 
+     cannot modify the task once a job is created, or the changes will 
+     be applied to the existing jobs */
+
+  jobs[Njobs].task = task;
+  
+  /* if we decide we need to be able to dynamically set task qualities
+     (like host, timeouts, etc), the we will need to have matched 
+     entries to these quantites in the job structure */
 
   Njobs ++;
@@ -108,4 +125,21 @@
 }
 
+void FreeJob (Job *job) {
+  
+  int i;
+
+  if (job == NULL) return;
+
+  if ((job[0].JobID >= 0) || (job[0].JobID < MAX_N_JOBS)) {
+    JobIDList[job[0].JobID] = FALSE;
+  }
+
+  for (i = 0; i < job[0].argc; i++) {
+    free (job[0].argv[i]);
+  }
+  free (job[0].argv);
+  return;
+}
+
 void SetCurrentJob (char *name) {
   JobName = name;
@@ -114,22 +148,4 @@
 char *GetCurrentJob () {
   return (JobName);
-}
-
-void FreeJob (Job *job) {
-  
-  int i;
-
-  if (job == NULL) return;
-
-  if ((job[0].JobID >= 0) || (job[0].JobID < MAX_N_JOBS)) {
-    JobIDList[job[0].JobID] = FALSE;
-  }
-
-  for (i = 0; i < job[0].task.argc; i++) {
-    free (job[0].task.argv[i]);
-  }
-  free (job[0].task.argv);
-  free (job[0].task.host);
-  return;
 }
 
@@ -161,29 +177,11 @@
 }
 
-/* this needs to: 
-   1) distinguish local from controller jobs
-   2) fork the local jobs in the background
-   3) send the controller jobs to the controller */
-
 int SubmitJob (Job *job) {
 
-  int i, Nchar;
-  char *string;
-
-  Nchar = 0;
-  for (i = 0; i < job[0].task.argc; i++) {
-    Nchar += strlen (job[0].task.argv[i]) + 1;
-  }
-  ALLOCATE (string, char, Nchar);
-  bzero (string, Nchar);
-
-  strcat (string, job[0].task.argv[0]);
-  for (i = 1; i < job[0].task.argc; i++) {
-    strcat (string, " ");
-    strcat (string, job[0].task.argv[i]);
-  }
-  
-  fprintf (stderr, "executing job %d ...%s...\n", job[0].JobID, string);
-  free (string);
+  if (job[0].mode == JOB_LOCAL) {
+    SubmitLocalJob (job);
+  } else {
+    SubmitControllerJob (job);
+  }
 
   /* reset clock for start and poll-test */
@@ -196,26 +194,11 @@
 int CheckJob (Job *job) {
 
-  float f;
-  f = drand48 ();
-
-  if ((0.00 < f) && (f < 0.25)) {
-    job[0].state = JOB_BUSY;
-    job[0].exit_status = -1;
-    return (JOB_BUSY);
-  }
-  if ((0.25 < f) && (f < 0.50)) {
-    job[0].state = JOB_BUSY;
-    job[0].exit_status = -1;
-    return (JOB_CRASH);
-  }
-  if ((0.50 < f) && (f < 0.75)) {
-    job[0].state = JOB_BUSY;
-    job[0].exit_status = 0;
-    return (JOB_EXIT);
-  }
-  if ((0.75 < f) && (f < 1.00)) {
-    job[0].state = JOB_BUSY;
-    job[0].exit_status = 1;
-    return (JOB_EXIT);
-  }
-}
+  /* add checks for timeouts */
+
+  if (job[0].mode == JOB_LOCAL) {
+    CheckLocalJob (job);
+  } else {
+    CheckControllerJob (job);
+  }
+  return (job[0].state);
+}
Index: trunk/Ohana/src/opihi/pantasks/LocalJob.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/LocalJob.c	(revision 3907)
+++ 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;
+}
Index: trunk/Ohana/src/opihi/pantasks/Makefile
===================================================================
--- trunk/Ohana/src/opihi/pantasks/Makefile	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/Makefile	(revision 4450)
@@ -25,15 +25,25 @@
 # sched user commands and support functions ########################
 
-sched = \
+funcs = \
+$(SDIR)/CheckJobs.$(ARCH).o \
+$(SDIR)/CheckSystem.$(ARCH).o \
+$(SDIR)/CheckTasks.$(ARCH).o \
+$(SDIR)/ControllerOps.$(ARCH).o \
+$(SDIR)/LocalJob.$(ARCH).o \
+$(SDIR)/JobOps.$(ARCH).o \
+$(SDIR)/TaskOps.$(ARCH).o \
+$(SDIR)/IOBufferOps.$(ARCH).o \
+$(SDIR)/memstr.$(ARCH).o \
+$(SDIR)/init.$(ARCH).o
+
+cmds = \
+$(SDIR)/controller.$(ARCH).o \
 $(SDIR)/run.$(ARCH).o \
+$(SDIR)/scheduler.$(ARCH).o \
 $(SDIR)/task.$(ARCH).o \
 $(SDIR)/task_command.$(ARCH).o \
 $(SDIR)/task_host.$(ARCH).o \
 $(SDIR)/task_macros.$(ARCH).o \
-$(SDIR)/task_periods.$(ARCH).o \
-$(SDIR)/JobOps.$(ARCH).o \
-$(SDIR)/TaskOps.$(ARCH).o \
-$(SDIR)/init.$(ARCH).o \
-$(SDIR)/scheduler.$(ARCH).o
+$(SDIR)/task_periods.$(ARCH).o
 
 libs = \
@@ -49,5 +59,5 @@
 	@echo done
 
-$(BIN)/scheduler.$(ARCH) : $(sched) $(libs)
+$(BIN)/scheduler.$(ARCH) : $(funcs) $(cmds) $(libs)
 
 install: $(DESTBIN)/scheduler
Index: trunk/Ohana/src/opihi/pantasks/TaskOps.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/TaskOps.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/TaskOps.c	(revision 4450)
@@ -108,4 +108,5 @@
 
   tasks[Ntasks].host = NULL;
+  tasks[Ntasks].host_required = FALSE;
 
   tasks[Ntasks].argc = 0;
@@ -124,4 +125,7 @@
   tasks[Ntasks].poll_period = 1.0;
   tasks[Ntasks].timeout_period = 1.0;
+
+  /* init task timer (is reset by 'run') */  
+  gettimeofday (&tasks[Ntasks].last, (void *) NULL);
 
   Ntasks ++;
@@ -155,2 +159,12 @@
   gettimeofday (timer, (void *) NULL);
 }
+
+/* start the clock for all tasks */
+void InitTaskTimers () {
+
+  Task *task;
+
+  while ((task = NextTask ()) != NULL) {
+    gettimeofday (&task[0].last, (void *) NULL);
+ }
+}
Index: trunk/Ohana/src/opihi/pantasks/controller.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/controller.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/controller.c	(revision 4450)
@@ -0,0 +1,48 @@
+# include "scheduler.h"
+
+int controller (int argc, char **argv) {
+
+  int status;
+  char command[1024];
+  IOBuffer buffer;
+
+  if (argc != 3) {
+    fprintf (stderr, "USAGE: controller host (hostname)\n");
+    return (FALSE);
+  }
+
+  if (strcmp (argv[1], "host")) {
+    fprintf (stderr, "invalid controller command\n");
+    return (FALSE);
+  }
+
+  /* start controller connection (if needed) */
+  StartController ();
+
+  sprintf (command, "host %s", argv[2]);
+  InitIOBuffer (&buffer, 0x100);
+  status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer);
+  FreeIOBuffer (&buffer);
+
+  /* check on success of controller command */
+  switch (status) {
+    case CONTROLLER_DOWN:
+      fprintf (stderr, "controller is down\n");
+      return (FALSE);
+
+    case CONTROLLER_HUNG:
+      fprintf (stderr, "controller is not responding\n");
+      return (FALSE);
+
+    case CONTROLLER_GOOD:
+      fprintf (stderr, "controller command sent\n");  
+      return (TRUE);
+
+    default:
+      fprintf (stderr, "unknown status for controller command: programming error\n");  
+      exit (1);
+  }
+
+  fprintf (stderr, "programming error: should not reach here\n");  
+  exit (1);
+}
Index: trunk/Ohana/src/opihi/pantasks/init.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/init.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/init.c	(revision 4450)
@@ -2,4 +2,5 @@
 # include "scheduler.h"
 
+int controller      PROTO((int, char **));
 int task	    PROTO((int, char **));
 int task_host	    PROTO((int, char **));
@@ -8,13 +9,16 @@
 int task_periods    PROTO((int, char **));
 int run             PROTO((int, char **));
+int stop            PROTO((int, char **));
 
 static Command cmds[] = {  
-  {"task",      task,         "define a schedulable task"},
-  {"host",      task_host,    "define host machine for a task"},
-  {"task.exit", task_macros,  "define exit macros for a task"},
-  {"task.exec", task_macros,  "define pre-exec macro for a task"},
-  {"command",   task_command, "define executed command for a task"},
-  {"periods",   task_periods, "define time scales for a task"},
-  {"run",       run,          "run the scheduler"},
+  {"controller", controller,   "controller commands"},
+  {"task",       task,         "define a schedulable task"},
+  {"host",       task_host,    "define host machine for a task"},
+  {"task.exit",  task_macros,  "define exit macros for a task"},
+  {"task.exec",  task_macros,  "define pre-exec macro for a task"},
+  {"command",    task_command, "define executed command for a task"},
+  {"periods",    task_periods, "define time scales for a task"},
+  {"run",        run,          "run the scheduler"},
+  {"stop",       stop,         "stop the scheduler"},
 }; 
 
Index: trunk/Ohana/src/opihi/pantasks/memstr.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/memstr.c	(revision 4450)
+++ trunk/Ohana/src/opihi/pantasks/memstr.c	(revision 4450)
@@ -0,0 +1,13 @@
+# include "pcontrol.h"
+
+/* returns pointer to start of m2 in m1, or NULL if failure */ 
+char *memstr (char *m1, char *m2, int n) {
+
+  int i, N;
+
+  N = strlen (m2);
+  for (i = 0; (i < n - N + 1) && memcmp (m1, m2, N); i++, m1++);
+  if (memcmp (m1, m2, N)) return (NULL);
+  return (m1);
+
+}
Index: trunk/Ohana/src/opihi/pantasks/run.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/run.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/run.c	(revision 4450)
@@ -1,12 +1,5 @@
-# include "basic.h"
 # include "scheduler.h"
 
 int run (int argc, char **argv) {
-
-  Job *job;
-  Task *task;
-  Macro *macro;
-  int i, found, status;
-  int Ntest;
 
   if (argc != 1) {
@@ -15,107 +8,21 @@
   }
 
-  /* start the clock for all tasks */
-  while ((task = NextTask ()) != NULL) {
-    gettimeofday (&task[0].last, (void *) NULL);
+  InitTaskTimers ();
+  rl_event_hook = CheckSystem;
+  rl_set_keyboard_input_timeout (1000000); 
+
+  return (TRUE);
+}
+
+int stop (int argc, char **argv) {
+
+  if (argc != 1) {
+    fprintf (stderr, "USAGE: stop\n");
+    return (FALSE);
   }
 
-  Ntest = 0;
+  rl_event_hook = NULL;
+  rl_set_keyboard_input_timeout (1000000); 
 
-  /* loop forever, checking for completed jobs and ready tasks */
-  while (1) {
-    if (Ntest > 5) { 
-      ListJobs ();
-      Ntest = 0;
-    }
-    usleep (10000);
-    Ntest ++;
-
-    /** test all tasks: ready to test? ready to run? **/
-    while ((task = NextTask ()) != NULL) {
-
-      /* ready to test? : check exec period */
-      if (GetTaskTimer(task[0].last) < task[0].exec_period) continue;
-
-      SetCurrentTask (task[0].name);
-      fprintf (stderr, "trying task %s\n", task[0].name);
-
-      /* ready to run? : run task.exec macro */
-      if (task[0].exec != NULL) {
-	status = exec_loop (task[0].exec);
-	if (!status) continue;
-      }
-
-      /* is task valid?  check state of task.(argc, argv) */
-      /*** ADD CODE HERE ***/
-
-      /* construct job from task */
-      job = CreateJob (task);
-
-      /* execute job - XXX add status test */
-      SubmitJob (job);
-
-      /* reset timer on task (don't do this if Create/Submit fails)*/
-      gettimeofday (&task[0].last, (void *) NULL);
-    }
-
-    /** test all jobs: ready to test?  finished? **/
-    while ((job = NextJob ()) != NULL) {
-
-      /* check for timeout */
-      if (GetTaskTimer(job[0].start) >= job[0].task.timeout_period) {
-	fprintf (stderr, "timeout on %s\n", job[0].task.name);
-	/* run task.timeout macro, if it exists */
-	if (job[0].task.timeout != NULL) {
-	  exec_loop (job[0].task.timeout);
-	}
-	DeleteJob (job);
-	continue;
-      }
-
-      /* check poll period (ready to run again?) */
-      if (GetTaskTimer(job[0].last) < job[0].task.poll_period) continue;
-
-      /* check current status */
-      status = CheckJob (job);
-      switch (status) {
-	case JOB_BUSY:
-	  fprintf (stderr, "job %s (%d) busy\n", job[0].task.name, job[0].JobID);
-	  break;
-
-	case JOB_CRASH:
-	  fprintf (stderr, "job %s (%d) crash\n", job[0].task.name, job[0].JobID);
-	  /* run task.crash macro, if it exists */
-	  if (job[0].task.crash != NULL) {
-	    exec_loop (job[0].task.crash);
-	  }
-	  DeleteJob (job);
-	  continue;
-	  break;
-
-	case JOB_EXIT:
-	  fprintf (stderr, "job %s (%d) exit\n", job[0].task.name, job[0].JobID);
-	  /* run corresponding task.exit macro, if it exists */
-	  macro = job[0].task.def;
-	  for (i = 0; i < job[0].task.Nexit; i++) {
-	    if (job[0].exit_status == atoi(job[0].task.exit[i][0].name)) {
-	      macro = job[0].task.exit[i];
-	      break;
-	    }
-	  }
-	  if (macro != NULL) exec_loop (macro);
-	  DeleteJob (job);
-	  continue;
-	  break;
-
-	default:
-	  fprintf (stderr, "unknown exit status\n");
-	  /** do something more useful here ?? **/
-	  break;
-      }
-
-      /* reset polling clock */
-      SetTaskTimer (&job[0].last);
-    }
-  }
   return (TRUE);
 }
Index: trunk/Ohana/src/opihi/pantasks/scheduler.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/scheduler.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/scheduler.c	(revision 4450)
@@ -49,4 +49,6 @@
   rl_readline_name = opihi_name;
   rl_attempted_completion_function = command_completer;
+  rl_event_hook = NULL;
+  rl_set_keyboard_input_timeout (1000000); 
 
   set_str_variable ("HISTORY", opihi_history);
Index: trunk/Ohana/src/opihi/pantasks/task_command.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/task_command.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/task_command.c	(revision 4450)
@@ -41,10 +41,2 @@
   return (TRUE);
 }
-
-
-/** 
-    careful with this: the command is supposed to be realized 
-    for the Job, not the Task (at execution)
-    the code is right, but who calls it when needs to be clarified.
- **/
-
Index: trunk/Ohana/src/opihi/pantasks/task_host.c
===================================================================
--- trunk/Ohana/src/opihi/pantasks/task_host.c	(revision 3907)
+++ trunk/Ohana/src/opihi/pantasks/task_host.c	(revision 4450)
@@ -4,10 +4,20 @@
 int task_host (int argc, char **argv) {
 
+  int N, RequiredHost;
   Task *task;
   char *taskname;
 
+  RequiredHost = FALSE;
+  if (N = get_argument (argc, argv, "-required")) {
+    remove_argument (N, &argc, argv);
+    RequiredHost = TRUE;
+  }
+
   if (argc != 2) {
-    fprintf (stderr, "USAGE: host <name>\n");
-    fprintf (stderr, "  (define host machine for this task (or 'none'))\n");
+    fprintf (stderr, "USAGE: host <name> [-required]\n");
+    fprintf (stderr, "  define host machine for this task\n");
+    fprintf (stderr, "  -required flags indicates controller must use this host\n");
+    fprintf (stderr, "  value of 'local' for host indicates process not using controller\n");
+    fprintf (stderr, "  value of 'none' for host indicates controller may assign at will\n");
     return (FALSE);
   }
@@ -23,9 +33,10 @@
     return (FALSE);
   }
+  task[0].host_required = RequiredHost;
 
   if (task[0].host != NULL) free (task[0].host);
   task[0].host = NULL;
 
-  if (!strcasecmp (argv[1], "NONE")) return (TRUE);
+  if (!strcasecmp (argv[1], "LOCAL")) return (TRUE);
 
   task[0].host = strcreate (argv[1]);