Index: /trunk/Ohana/src/opihi/doc/pcontrol.txt
===================================================================
--- /trunk/Ohana/src/opihi/doc/pcontrol.txt	(revision 8423)
+++ /trunk/Ohana/src/opihi/doc/pcontrol.txt	(revision 8424)
@@ -1,2 +1,11 @@
+
+2006.08.18
+
+ Outstanding issues related to pcontrol:
+
+ * disposition of HUNG jobs?
+ * probably should not save the history for pcontrol or pclient
+   (these will be many lines long very quickly...)
+ * need to add options to run/stop for hosts and jobs independently
 
 2006.08.11
Index: /trunk/Ohana/src/opihi/include/pcontrol.h
===================================================================
--- /trunk/Ohana/src/opihi/include/pcontrol.h	(revision 8423)
+++ /trunk/Ohana/src/opihi/include/pcontrol.h	(revision 8424)
@@ -1,4 +1,5 @@
 # include "data.h"
 # include "basic.h"
+# define THREADED
 
 typedef struct timeval Ptime;
@@ -98,5 +99,7 @@
   int    Nobject;
   int    NOBJECT;
-  // pthread_mutex_t mutex;
+# ifdef THREADED    
+  pthread_mutex_t mutex;
+# endif
 } Stack;
 
@@ -108,4 +111,9 @@
 # define DTIME(A,B) ((A.tv_sec - B.tv_sec) + 1e-6*(A.tv_usec - B.tv_usec))
 # define ZTIME(A) ((A.tv_sec == 0) && (A.tv_usec == 0))
+
+// # define ASSERT(TEST,STRING) { if (!(TEST)) { gprint (GP_ERR, "programming error: %s\n", STRING); abort (); }}
+// # define ABORT(STRING) { gprint (GP_ERR, "programming error: %s\n", STRING); abort (); }
+# define ASSERT(TEST,STRING) { if (!(TEST)) { gprint (GP_ERR, "programming error: %s\n", STRING); raise (SIGINT); exit (2); }}
+# define ABORT(STRING) { gprint (GP_ERR, "programming error: %s\n", STRING); raise (SIGINT); exit (2); }
 
 void InitPcontrol ();
@@ -117,42 +125,57 @@
 void  *PullStackByName (Stack *stack, char *name);
 void  *PullStackByID (Stack *stack, int id);
-void  *FindStackByID (Stack *stack, int id);
-void  *FindStackByName (Stack *stack, char *name);
+int    RemoveStackEntry (Stack *stack, int where);
+void  *RemoveStackByID (Stack *stack, int id);
 void   LockStack (Stack *stack);
 void   UnlockStack (Stack *stack);
 
-int CheckSystem ();
-int CheckBusyJobs (float delay);
-int CheckDoneJobs (float delay);
-int CheckKillJobs (float delay);
-int CheckDoneHosts (float delay);
-int CheckLiveHosts (float delay);
-int CheckDownHosts (float delay);
-int CheckIdleHosts (float delay);
+// void  *FindStackByID (Stack *stack, int id);
+// void  *FindStackByName (Stack *stack, char *name);
+
+/*** CheckSystem.c ***/
+int   CheckSystem ();
+void *CheckSystem_Threaded (void *data);
+int   CheckBusyJobs (float delay);
+int   CheckDoneJobs (float delay);
+int   CheckKillJobs (float delay);
+int   CheckDoneHosts (float delay);
+int   CheckDownHosts (float delay);
+int   CheckIdleHosts (float delay);
+int   CheckLiveHosts (float delay);
+int   SetRunSystem (int state);
+
+/*** own files ***/
+int CheckHost (Host *host);
+int StartHost (Host *host);
 int CheckIdleHost (Host *host);
+int CheckDoneHost (Host *host);
+int CheckBusyJob (Job *job, Host *host);
+int CheckDoneJob (Job *job, Host *host);
+int KillJob (Job *job, Host *host);
+int StartJob (Job *job, Host *host);
+int ResetJob (Job *job);
+int GetJobOutput (char *command, Host *host, IOBuffer *buffer, int Nbytes);
+int PclientCommand (Host *host, char *command, char *response, IOBuffer *buffer);
+int rconnect (char *command, char *hostname, char *shell, int *stdio);
+
+/*** misc files ***/
+int    VerboseMode ();  // in verbose.c
+void   gotsignal (int signum); // in pcontrol.c
+
+/*** IDops.c ***/
 void InitIDs ();
 IDtype NextJobID ();
 IDtype NextHostID ();
 void PrintID (gpDest dest, IDtype ID);
-int CheckBusyJob (Job *job);
-int StartHost (Host *host);
-int CheckDoneHost (Host *host);
-int CheckDoneJob (Job *job);
-int GetJobOutput (char *command, Host *host, IOBuffer *buffer, int Nbytes);
-int ResetJob (Job *job);
-int PclientCommand (Host *host, char *command, char *response, IOBuffer *buffer);
-int rconnect (char *command, char *hostname, char *shell, int *stdio);
-int CheckHost (Host *host);
-
-int PrintJobStack (int Nstack);
-int PrintHostStack (int Nstack);
-
-int    VerboseMode ();
-
-void   gotsignal (int signum);
+
+/*** CheckPoint.c ***/
+int SetCheckPoint ();
+int ClearCheckPoint ();
+int TestCheckPoint ();
 
 /*** HostOps.c ***/
 void   InitHostStacks ();
 Stack *GetHostStack (int StackID);
+char  *GetHostStackName (int StackID);
 Stack *GetHostStackByName (char *name);
 int    PutHost (Host *host, int StackID, int where);
@@ -161,20 +184,18 @@
 Host  *PullHostFromStackByID (int StackID, IDtype ID);
 Host  *PullHostFromStackByName (int StackID, char *name);
-Host  *FindHostByID (IDtype HostID, int *StackID);
-Host  *FindHostByName (char *name, int *StackID);
-Host  *FindHostInStackByID (int StackID, IDtype ID);
-Host  *FindHostInStackByName (int StackID, char *name);
 IDtype AddHost (char *hostname);
 void   DelHost (Host *host);
 
+/*** StopHosts.c ***/
 void   DownHost (Host *host);
 void   OffHost (Host *host);
 int    DownHosts ();
+int    StopHost (Host *host);
 int    HarvestHost (int pid);
-int    StopHost (Host *host);
 
 /*** JobOps.c ***/
 void   InitJobStacks ();
 Stack *GetJobStack (int StackID);
+char  *GetJobStackName (int StackID);
 Stack *GetJobStackByName (char *name);
 int    PutJob (Job *job, int StackID, int where);
@@ -182,6 +203,4 @@
 Job   *PullJobByID (IDtype JobID, int *StackID);
 Job   *PullJobFromStackByID (int StackID, int ID);
-Job   *FindJobByID (IDtype JobID, int *StackID);
-Job   *FindJobInStackByID (int StackID, int ID);
 IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv);
 void   DelJob (Job *job);
@@ -189,4 +208,8 @@
 void   LinkJobAndHost (Job *job, Host *host);
 
-int    KillJob (Job *job);
-int    StartJob (Job *job);
+// Job   *FindJobByID (IDtype JobID, int *StackID);
+// Job   *FindJobInStackByID (int StackID, int ID);
+// Host  *FindHostByID (IDtype HostID, int *StackID);
+// Host  *FindHostByName (char *name, int *StackID);
+// Host  *FindHostInStackByID (int StackID, IDtype ID);
+// Host  *FindHostInStackByName (int StackID, char *name);
Index: /trunk/Ohana/src/opihi/include/shell.h
===================================================================
--- /trunk/Ohana/src/opihi/include/shell.h	(revision 8423)
+++ /trunk/Ohana/src/opihi/include/shell.h	(revision 8424)
@@ -11,4 +11,8 @@
 # define MACRO_STRING(s) #s
 # define MACRO_NAME(s) MACRO_STRING(s)
+
+/* enums used by gprint functions */
+typedef enum {GP_FILE, GP_BUFF} gpMode;
+typedef enum {GP_LOG, GP_ERR} gpDest;
 
 typedef int CommandF ();
@@ -33,4 +37,13 @@
 } List;
 
+/* structure used to represent the gprint i/o stream */
+typedef struct {
+  FILE *file;
+  IOBuffer *buffer;
+  gpMode mode;
+  gpDest dest;
+  pthread_t thread;
+} gpStream;
+
 /*** globals used to track the shell language concepts  ***/
 List         *lists;			/* variable to store the list of all lists */
@@ -47,4 +60,5 @@
 void          program_init            	PROTO((int *argc, char **argv));
 void          startup               	PROTO((int *argc, char **argv));
+int           opihi                     PROTO((int argc, char **argv));
 int           multicommand          	PROTO((char *line));
 void          multicommand_InitServer   PROTO((void));
@@ -118,34 +132,20 @@
 int 	      macro_list_f 		PROTO((int, char **));  /* "macro_list" is a readline func */
 int 	      macro_read   		PROTO((int, char **));
-int 	      macro_write  		PROTO((int, char **));
+int 	      macro_write   		PROTO((int, char **));
 
-char *memstr (char *m1, char *m2, int n);
-int write_fmt (int fd, char *format, ...);
-char *opihi_version ();
-char *strip_version (char *input);
+char 	     *memstr        		PROTO((char *m1, char *m2, int n));
+int  	      write_fmt     		PROTO((int fd, char *format, ...));
+char 	     *opihi_version 		PROTO(());
+char 	     *strip_version 		PROTO((char *input));
 
-/*** gprint defines ***/
-
-/* enums used by gprint functions */
-typedef enum {GP_FILE, GP_BUFF} gpMode;
-typedef enum {GP_LOG, GP_ERR} gpDest;
-
-/* structure used to represent the gprint i/o stream */
-typedef struct {
-  FILE *file;
-  IOBuffer *buffer;
-  gpMode mode;
-  gpDest dest;
-  pthread_t thread;
-} gpStream;
-
-void gprintInit ();
-gpStream *gprintGetStream (gpDest dest);
-void gprintSetBuffer (gpDest dest);
-IOBuffer *gprintGetBuffer (gpDest dest);
-void gprintSetFile (gpDest dest, char *filename);
-FILE *gprintGetFile (gpDest dest);
-int gprint (gpDest dest, char *format, ...);
-int gwrite (char *buffer, int size, int N, gpDest dest);
+/* gprint functions */
+void          gprintInit      		PROTO(());
+gpStream     *gprintGetStream 		PROTO((gpDest dest));
+void          gprintSetBuffer 		PROTO((gpDest dest));
+IOBuffer     *gprintGetBuffer 		PROTO((gpDest dest));
+void          gprintSetFile   		PROTO((gpDest dest, char *filename));
+FILE         *gprintGetFile   		PROTO((gpDest dest));
+int           gprint          		PROTO((gpDest dest, char *format, ...));
+int           gwrite          		PROTO((char *buffer, int size, int N, gpDest dest));
 
 # endif
Index: /trunk/Ohana/src/opihi/pcontrol/CheckBusyJob.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckBusyJob.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckBusyJob.c	(revision 8424)
@@ -1,5 +1,5 @@
 # include "pcontrol.h"
 
-int CheckBusyJob (Job *job) {
+int CheckBusyJob (Job *job, Host *host) {
 
   int      status;
@@ -8,5 +8,4 @@
   char     string[64];
   IOBuffer buffer;
-  Host    *host;
 
   /* we are checking a job which is currently busy.  it has been pulled from the
@@ -14,6 +13,6 @@
      XXX need to check on state of HOST on return */
 
-  /** must have a valid host : if not? **/
-  host = (Host *) job[0].host;
+  ASSERT (host == (Host *) job[0].host, "invalid host");
+  ASSERT (job  == (Job *) host[0].job, "invalid job");
 
   InitIOBuffer (&buffer, 0x100);
@@ -25,5 +24,7 @@
     case PCLIENT_DOWN:
       HarvestHost (host[0].pid);
-      UnlinkJobAndHost (job);
+      // unlink host & job
+      job[0].host = NULL;
+      host[0].job = NULL;
       PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
       PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
@@ -32,4 +33,9 @@
 
     case PCLIENT_HUNG:
+      HarvestHost (host[0].pid);
+      // unlink host & job
+      job[0].host = NULL;
+      host[0].job = NULL;
+      PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
       PutJobSetState (job, PCONTROL_JOB_BUSY, STACK_BOTTOM, PCONTROL_JOB_HUNG);
       FreeIOBuffer (&buffer);
@@ -37,25 +43,21 @@
 
     case PCLIENT_GOOD:
-      if (VerboseMode()) gprint (GP_ERR, "message received (CheckBusyJob)\n");
+      if (VerboseMode()) gprint (GP_ERR, "message received (CheckBusyJob)");
       break;
 
     default:
-      gprint (GP_ERR, "programming error: unknown status for pclient command\n");  
-      exit (1);
+      ABORT ("unknown status for pclient command");  
   }
 
   /** host is up, need to parse message **/
   p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
-  if (p == NULL) {
-    gprint (GP_ERR, "programming error: missing STATUS in pclient message\n");
-    exit (1);
-  }
+  ASSERT (p != NULL, "missing STATUS in pclient message");
+
   sscanf (p, "%*s %s", string);
-  if (!strcmp(string, "NONE")) {
-    gprint (GP_ERR, "programming error: no current job\n");
-    exit (1);
-  }
+  ASSERT (strcmp(string, "NONE"), "no current job\n");
+
   /** no status change, return to BUSY stack **/
   if (!strcmp(string, "BUSY")) {
+    PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
     PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM);
     FreeIOBuffer (&buffer);
@@ -67,8 +69,5 @@
   if (!strcmp(string, "EXIT")) outstate = PCONTROL_JOB_EXIT;
   if (!strcmp(string, "CRASH")) outstate = PCONTROL_JOB_CRASH;
-  if (outstate == PCONTROL_JOB_BUSY) {
-    gprint (GP_ERR, "programming error : should not reach here (CheckJob)\n");
-    exit (1);
-  }
+  ASSERT (outstate != PCONTROL_JOB_BUSY, "should not reach here (CheckJob)");
 
   /* parse the exit status and sizes of output buffers */
@@ -80,9 +79,10 @@
   sscanf (p, "%*s %d", &job[0].stderr_size);
 
-  /** job has exited : move to DONE stack (host still BUSY) **/
+  // job has exited : move to DONE stack 
+  // the host is still BUSY until job output is gathered (int CheckDoneJob)
+  // don't unlink job and host yet
+  PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
   PutJobSetState (job, PCONTROL_JOB_DONE, STACK_BOTTOM, outstate);
   FreeIOBuffer (&buffer);
   return (TRUE);
 }
-
-/** need to add timeout check here **/
Index: /trunk/Ohana/src/opihi/pcontrol/CheckDoneHost.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckDoneHost.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckDoneHost.c	(revision 8424)
@@ -20,7 +20,8 @@
       FreeIOBuffer (&buffer);
       return (FALSE);
-      /** do we need to close the connection? **/
+      // XXX do we need to close the connection?
 
     case PCLIENT_HUNG:
+      // XXX should this be DONE or DOWN?/
       PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM);
       if (VerboseMode()) gprint (GP_ERR, "host %s is not responding\n", host[0].hostname);
@@ -33,19 +34,15 @@
 
     default:
-      if (VerboseMode()) gprint (GP_ERR, "unknown status for pclient command: programming error\n");  
-      exit (1);
+      ABORT ("unknown status for pclient command");  
   }
 
   /** successful command, examine result **/
   p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
-  if (p == NULL) {
-    gprint (GP_ERR, "programming error: missing STATUS in pclient message (CheckDoneHost)\n");
-    exit (1);
-  }
+  ASSERT (p != NULL, "missing STATUS in pclient message (CheckDoneHost)");
+
   sscanf (p, "%*s %d", &status);
   switch (status) {
     case -1:
-      gprint (GP_ERR, "programming error: reset syntax error\n");
-      exit (1);
+      ABORT ("reset syntax error");
       
     case 0:
@@ -63,9 +60,7 @@
 
     default:
-      gprint (GP_ERR, "programming error: should not reach here (CheckDoneHost)\n");
-      exit (1);
+      ABORT ("should not reach here (CheckDoneHost)");
   }
-  gprint (GP_ERR, "programming error: should not reach here either (CheckDoneHost)\n");
-  exit (1);
+  ABORT ("should not reach here (CheckDoneHost)");
 }
 
Index: /trunk/Ohana/src/opihi/pcontrol/CheckDoneJob.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckDoneJob.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckDoneJob.c	(revision 8424)
@@ -1,15 +1,19 @@
 # include "pcontrol.h"
 
-int CheckDoneJob (Job *job) {
+int CheckDoneJob (Job *job, Host *host) {
   
-  Host *host;
+  int success;
 
-  if (!GetJobOutput ("stdout", (Host *) job[0].host, &job[0].stdout, job[0].stdout_size)) {
-    /* strip off first and last lines */
-    PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM);
-    return (FALSE);
-  }
+  ASSERT (host == (Host *) job[0].host, "invalid host");
+  ASSERT (job == (Job *) host[0].job, "invalid job");
 
-  if (!GetJobOutput ("stderr", (Host *) job[0].host, &job[0].stderr, job[0].stderr_size)) {
+  success = TRUE;
+  success &= GetJobOutput ("stdout", host, &job[0].stdout, job[0].stdout_size);
+  success &= GetJobOutput ("stderr", host, &job[0].stderr, job[0].stderr_size);
+
+  if (!success) {
+    // XXX some kind of error?
+    // XXX try again later?
+    PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
     PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM);
     return (FALSE);
@@ -17,5 +21,7 @@
 
   /* job's state is either EXIT or CRASH (verify?) */
-  host = UnlinkJobAndHost (job);
+  // unlink host & job
+  job[0].host = NULL;
+  host[0].job = NULL;
   PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM);
   PutJob (job, job[0].state, STACK_BOTTOM);
Index: /trunk/Ohana/src/opihi/pcontrol/CheckHost.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckHost.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckHost.c	(revision 8424)
@@ -5,5 +5,14 @@
   int status;
   IOBuffer buffer;
-  Job *job;
+
+  if (host[0].job != NULL) return (TRUE);
+
+  /* if this host has been marked to be turned off, do that and return */
+  if (host[0].markoff) {
+    host[0].markoff = FALSE;
+    StopHost (host);
+    OffHost (host);
+    return (TRUE);
+  }
 
   InitIOBuffer (&buffer, 0x100);
@@ -13,10 +22,4 @@
     case 0:
       if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname);
-      /* if host has a job, job is dead, return to Pending */
-      job = (Job *) host[0].job;
-      if (job != NULL) {
-	UnlinkJobAndHost (job);
-	PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
-      }
       HarvestHost (host[0].pid);
       PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
@@ -36,5 +39,7 @@
       return (TRUE);
   }
-  gprint (GP_ERR, "programming error: should not reach here (Check Host)\n"); 
-  return (FALSE);
+  ABORT ("should not reach here (Check Host)"); 
 }
+
+// if the host has a job, we skip it (down or crash state will be caught elsewhere)
+// in fact, just touch the IDLE hosts, not the BUSY hosts?
Index: /trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c	(revision 8424)
@@ -8,4 +8,12 @@
   Job *job;
 
+  /* if this host has been marked to be turned off, do that and return */
+  if (host[0].markoff) {
+    host[0].markoff = FALSE;
+    StopHost (host);
+    OffHost (host);
+    return (TRUE);
+  }
+    
   /* search the JOB_PENDING stack for an appropriate job */
   stack = GetJobStack (PCONTROL_JOB_PENDING);
@@ -16,8 +24,5 @@
     job = (Job *) stack[0].object[i];
     if (job[0].mode != PCONTROL_JOB_NEEDHOST) continue;
-    if (job[0].hostname == NULL) {
-      gprint (GP_ERR, "programming error: NEEDHOST hostname missing\n");
-      exit (2);
-    }
+    ASSERT (job[0].hostname != NULL, "NEEDHOST hostname missing");
     if (strcasecmp (job[0].hostname, host[0].hostname)) continue;
 
@@ -29,5 +34,5 @@
     RemoveStackEntry (stack, i);
     UnlockStack (stack);
-    StartJob (job);
+    StartJob (job, host);
     return (TRUE);
   }
@@ -37,8 +42,5 @@
     job = (Job *) stack[0].object[i];
     if (job[0].mode != PCONTROL_JOB_WANTHOST) continue;
-    if (job[0].hostname == NULL) {
-      gprint (GP_ERR, "programming error: WANTHOST hostname missing\n");
-      exit (2);
-    }
+    ASSERT (job[0].hostname != NULL, "WANTHOST hostname missing");
     if (strcasecmp (job[0].hostname, host[0].hostname)) continue;
 
@@ -50,5 +52,5 @@
     RemoveStackEntry (stack, i);
     UnlockStack (stack);
-    StartJob (job);
+    StartJob (job, host);
     return (TRUE);
   }
@@ -66,6 +68,6 @@
     RemoveStackEntry (stack, i);
     UnlockStack (stack);
-    StartJob (job);
-   return (TRUE);
+    StartJob (job, host);
+    return (TRUE);
   }
 
@@ -76,5 +78,5 @@
     job = (Job *) stack[0].object[i];
     if (job[0].mode != PCONTROL_JOB_WANTHOST) continue;
-    // test the job age and skip if too young
+    // XXX test the job age and skip if too young
 
     /* we have found an appropriate job; link it to the host and send to StartJob */
@@ -85,5 +87,5 @@
     RemoveStackEntry (stack, i);
     UnlockStack (stack);
-    StartJob (job);
+    StartJob (job, host);
    return (TRUE);
   }
Index: /trunk/Ohana/src/opihi/pcontrol/CheckPoint.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckPoint.c	(revision 8424)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckPoint.c	(revision 8424)
@@ -0,0 +1,66 @@
+# include "pcontrol.h"
+
+// one thread (user) interacts with the user and blocks for long periods on input.
+// one thread (client) spins continuously and monitors the hosts and jobs
+// in some cases, the user thread needs to set a check point on the client thread 
+// to ensure the all HOSTs and JOBs are on one of the stacks (nothing 'in flight').
+// these are not symmetric: the client thread should not call Set/Clear, the user 
+// thread should not call Test
+
+# ifdef THREADED
+static pthread_mutex_t client = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t user = PTHREAD_MUTEX_INITIALIZER;
+# endif
+
+// The user thread calls this to stop the client thread from shuffling the Host/Job stacks
+int SetCheckPoint () {
+
+# ifdef THREADED
+  int status;
+
+  // set my lock
+  pthread_mutex_lock (&user);
+
+  // wait until client thread sets its lock
+  while (1) {
+    status = pthread_mutex_trylock (&client);
+    if (status == EBUSY) {
+      // client has reached the check-point
+      return (TRUE);
+    }
+    pthread_mutex_unlock (&client);
+    usleep (10000);
+  }
+  // put in a timeout?  (client thread not spinning...)
+  return (FALSE);
+# else
+  return (TRUE);
+# endif
+}
+
+// The user thread calls this to allow the client thread to continue
+int ClearCheckPoint () {
+  // clear my lock
+# ifdef THREADED
+  pthread_mutex_unlock (&user);
+# endif
+  return (TRUE);
+}
+
+// The client thread calls in the thread loop somewhere the stacks are stable
+// (ie, no jobs or hosts currently in flight)
+int TestCheckPoint () {
+
+# ifdef THREADED
+  // set my lock
+  pthread_mutex_lock (&client);
+  
+  // try the user-thread lock
+  pthread_mutex_lock (&user);
+  pthread_mutex_unlock (&user);
+
+  // clear my lock
+  pthread_mutex_unlock (&client);
+# endif
+  return (TRUE);
+}
Index: /trunk/Ohana/src/opihi/pcontrol/CheckSystem.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/CheckSystem.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/CheckSystem.c	(revision 8424)
@@ -1,5 +1,14 @@
 # include "pcontrol.h"
+# define DEBUG 0
 
 static struct timeval lastlive = {0, 0};
+static int RunSystem = FALSE;
+
+int SetRunSystem (int state) {
+  int oldstate;
+  oldstate = RunSystem;
+  RunSystem = state;
+  return oldstate;
+}
 
 int CheckSystem () {
@@ -29,5 +38,5 @@
   } 
 
-  if (0) { 
+  if (DEBUG) { 
     Stack *stack;
     int Nidle, Ndown, Nbusy;
@@ -44,10 +53,65 @@
 }
 
+void *CheckSystem_Threaded (void *data) {
+
+  struct timeval now;
+  float dtime;
+
+  gprintInit ();
+
+  while (1) {
+    // stop here if the user-thread requests (no objects in flight) 
+    TestCheckPoint ();
+
+    // don't run the system checks if RunSystem is FALSE
+    if (!RunSystem) {
+      usleep (50000);
+      continue;
+    }
+
+    // we want to give each block a maximum allowed time
+    CheckIdleHosts(0.020); /* submit a new job */
+
+    CheckBusyJobs(0.020);  /* get job status */
+    CheckDoneJobs(0.020);  /* harvest job stdout/stderr */
+    CheckKillJobs(0.020);  /* harvest job stdout/stderr */
+
+    CheckDoneHosts(0.020); /* reset the host */
+    CheckDownHosts(0.100); /* launch the host */
+
+    /* always allow at least one test */
+    /* most tests require about 2ms per host.  
+       CheckDoneJobs must depend on the size of the output buffer */
+
+    gettimeofday (&now, (void *) NULL);
+    dtime = DTIME (now, lastlive);
+    if (dtime > 1.0) {
+      CheckLiveHosts(0.040);
+      lastlive = now;
+    } 
+
+    if (DEBUG) { 
+      Stack *stack;
+      int Nidle, Ndown, Nbusy;
+      stack = GetHostStack (PCONTROL_HOST_IDLE);
+      Nidle = stack[0].Nobject;
+      stack = GetHostStack (PCONTROL_HOST_DOWN);
+      Ndown = stack[0].Nobject;
+      stack = GetHostStack (PCONTROL_HOST_BUSY);
+      Nbusy = stack[0].Nobject;
+      gprint (GP_ERR, "busy, idle, down: %2d %2d %2d\n", Nbusy, Nidle, Ndown);
+    }
+  }
+  return (NULL);
+}
+
 int CheckBusyJobs (float MaxDelay) {
 
   struct timeval start, stop;
   int i, Nobject;
-  Stack *stack;
+  Stack *hoststack;
+  Stack *jobstack;
   Job   *job;
+  Host  *host;
   float dtime;
 
@@ -65,6 +129,5 @@
   dtime = 0.0;
   for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) {
-    /* pull both job and host from their stacks */
-    /* XXX is the subject to the Dangerous Embrace? */
+    // pull both job and host from their stacks
     LockStack (hoststack);
     job = PullStackByLocation (jobstack, STACK_TOP);
@@ -73,12 +136,13 @@
       break;
     }
-    host = RemoveStackByID (hoststack, job[0].host[0].HostID);
+    host = (Host *) job[0].host;
+    RemoveStackByID (hoststack, host[0].HostID);
     UnlockStack (hoststack);
 
-    CheckBusyJob (job);
-    gettimeofday (&stop, (void *) NULL);
-    dtime = DTIME (stop, start);
-  }
-  if (0 && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject);
+    CheckBusyJob (job, host);
+    gettimeofday (&stop, (void *) NULL);
+    dtime = DTIME (stop, start);
+  }
+  if (DEBUG && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject);
   return (TRUE);
 }
@@ -88,23 +152,34 @@
   struct timeval start, stop;
   int i, Nobject;
-  Stack *stack;
+  Stack *hoststack;
+  Stack *jobstack;
   Job   *job;
-  float dtime;
-
-  /* Loop through objects on the stack, no more than once. see note above */
-  stack = GetJobStack (PCONTROL_JOB_DONE);
-  Nobject = stack[0].Nobject;
-
-  /* always allow at least one test */
-  gettimeofday (&start, (void *) NULL);
-  dtime = 0.0;
-  for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) {
-    job = PullStackByLocation (stack, STACK_TOP);
-    if (job == NULL) break;
-    CheckDoneJob (job);
-    gettimeofday (&stop, (void *) NULL);
-    dtime = DTIME (stop, start);
-  }
-  if (0 && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject);
+  Host  *host;
+  float dtime;
+
+  /* Loop through objects on the stack, no more than once. see note above */
+  hoststack = GetHostStack (PCONTROL_HOST_BUSY);
+  jobstack  = GetJobStack (PCONTROL_JOB_DONE);
+  Nobject   = jobstack[0].Nobject;
+
+  /* always allow at least one test */
+  gettimeofday (&start, (void *) NULL);
+  dtime = 0.0;
+  for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) {
+    LockStack (hoststack);
+    job = PullStackByLocation (jobstack, STACK_TOP);
+    if (job == NULL) {
+      UnlockStack (hoststack);
+      break;
+    }
+    host = (Host *) job[0].host;
+    RemoveStackByID (hoststack, host[0].HostID);
+    UnlockStack (hoststack);
+
+    CheckDoneJob (job, host);
+    gettimeofday (&stop, (void *) NULL);
+    dtime = DTIME (stop, start);
+  }
+  if (DEBUG && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject);
   return (TRUE);
 }
@@ -114,23 +189,34 @@
   struct timeval start, stop;
   int i, Nobject;
-  Stack *stack;
+  Stack *hoststack;
+  Stack *jobstack;
   Job   *job;
-  float dtime;
-
-  /* Loop through objects on the stack, no more than once. see note above */
-  stack = GetJobStack (PCONTROL_JOB_KILL);
-  Nobject = stack[0].Nobject;
-
-  /* always allow at least one test */
-  gettimeofday (&start, (void *) NULL);
-  dtime = 0.0;
-  for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) {
-    job = PullStackByLocation (stack, STACK_TOP);
-    if (job == NULL) break;
-    KillJob (job);
-    gettimeofday (&stop, (void *) NULL);
-    dtime = DTIME (stop, start);
-  }
-  if (0 && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject);
+  Host  *host;
+  float dtime;
+
+  /* Loop through objects on the stack, no more than once. see note above */
+  hoststack = GetHostStack (PCONTROL_HOST_BUSY);
+  jobstack = GetJobStack (PCONTROL_JOB_KILL);
+  Nobject = jobstack[0].Nobject;
+
+  /* always allow at least one test */
+  gettimeofday (&start, (void *) NULL);
+  dtime = 0.0;
+  for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) {
+    LockStack (hoststack);
+    job = PullStackByLocation (jobstack, STACK_TOP);
+    if (job == NULL) {
+      UnlockStack (hoststack);
+      break;
+    }
+    host = (Host *) job[0].host;
+    RemoveStackByID (hoststack, host[0].HostID);
+    UnlockStack (hoststack);
+
+    KillJob (job, host);
+    gettimeofday (&stop, (void *) NULL);
+    dtime = DTIME (stop, start);
+  }
+  if (DEBUG && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject);
   return (TRUE);
 }
@@ -158,5 +244,5 @@
     dtime = DTIME (stop, start);
   }
-  if (0) gprint (GP_ERR, "checked %d hosts\n", i);
+  if (DEBUG) gprint (GP_ERR, "checked %d hosts\n", i);
   return (TRUE);
 }
@@ -180,4 +266,9 @@
     host = PullStackByLocation (stack, STACK_TOP);
     if (host == NULL) break;
+    if (host[0].markoff) {
+      host[0].markoff = FALSE;
+      OffHost (host);
+      return (TRUE);
+    }
     dtime = DTIME (host[0].nexttry, start);
     if (dtime > 0) {
@@ -189,5 +280,5 @@
     dtime = DTIME (stop, start);
   }
-  if (0) gprint (GP_ERR, "checked %d hosts\n", i);
+  if (DEBUG) gprint (GP_ERR, "checked %d hosts\n", i);
   return (TRUE);
 }
@@ -219,9 +310,9 @@
     dtime = DTIME (stop, start);
   }
-  if (0) gprint (GP_ERR, "checked %d hosts\n", i);
-  return (TRUE);
-}
-
-/* this is just a heartbeat check */
+  if (DEBUG) gprint (GP_ERR, "checked %d hosts\n", i);
+  return (TRUE);
+}
+
+/* this is just a heartbeat check (only IDLE hosts) */
 int CheckLiveHosts (float MaxDelay) {
 
@@ -246,19 +337,5 @@
     dtime = DTIME (stop, start);
   }
-  if (0) gprint (GP_ERR, "checked %d idle hosts\n", i);
-
-  /* Loop through objects on the stack, no more than once. see note above */
-  stack = GetHostStack (PCONTROL_HOST_BUSY);
-  Nobject = stack[0].Nobject;
-
-  dtime = 0.0;
-  for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) {
-    host = PullStackByLocation (stack, STACK_TOP);
-    if (host == NULL) break;
-    CheckHost (host);
-    gettimeofday (&stop, (void *) NULL);
-    dtime = DTIME (stop, start);
-  }
-  if (0) gprint (GP_ERR, "checked %d busy hosts\n", i);
+  if (DEBUG) gprint (GP_ERR, "checked %d idle hosts\n", i);
   return (TRUE);
 }
Index: /trunk/Ohana/src/opihi/pcontrol/HostOps.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/HostOps.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/HostOps.c	(revision 8424)
@@ -15,19 +15,23 @@
 }
 
+char *GetHostStackName (int StackID) {
+  switch (StackID) {
+    case PCONTROL_HOST_IDLE: return ("IDLE");
+    case PCONTROL_HOST_DOWN: return ("DOWN");
+    case PCONTROL_HOST_DONE: return ("DONE");
+    case PCONTROL_HOST_BUSY: return ("BUSY");
+    case PCONTROL_HOST_OFF:  return ("OFF");
+  }
+  gprint (GP_ERR, "error: unknown host stack : programming error\n");
+  exit (1);
+}
+
 Stack *GetHostStack (int StackID) {
   switch (StackID) {
-    case PCONTROL_HOST_IDLE:
-      return (HostPool_Idle);
-    case PCONTROL_HOST_DOWN:
-      return (HostPool_Down);
-    case PCONTROL_HOST_DONE:
-      return (HostPool_Done);
-    case PCONTROL_HOST_BUSY:
-      return (HostPool_Busy);
-    case PCONTROL_HOST_OFF:
-      return (HostPool_Off);
-    default:
-      gprint (GP_ERR, "error: unknown host stack : programming error\n");
-      exit (1);
+    case PCONTROL_HOST_IDLE: return (HostPool_Idle);
+    case PCONTROL_HOST_DOWN: return (HostPool_Down);
+    case PCONTROL_HOST_DONE: return (HostPool_Done);
+    case PCONTROL_HOST_BUSY: return (HostPool_Busy);
+    case PCONTROL_HOST_OFF:  return (HostPool_Off);
   }
   gprint (GP_ERR, "error: unknown host stack : programming error\n");
Index: /trunk/Ohana/src/opihi/pcontrol/JobOps.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/JobOps.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/JobOps.c	(revision 8424)
@@ -17,21 +17,25 @@
 }
 
+char *GetJobStackName (int StackID) {
+  switch (StackID) {
+    case PCONTROL_JOB_PENDING: return ("PENDING");
+    case PCONTROL_JOB_BUSY:    return ("BUSY");
+    case PCONTROL_JOB_DONE:    return ("DONE");
+    case PCONTROL_JOB_KILL:    return ("KILL");
+    case PCONTROL_JOB_EXIT:    return ("EXIT");
+    case PCONTROL_JOB_CRASH:   return ("CRASH");
+  }
+  gprint (GP_ERR, "error: unknown host stack : programming error\n");
+  exit (1);
+}
+
 Stack *GetJobStack (int StackID) {
   switch (StackID) {
-    case PCONTROL_JOB_PENDING:
-      return (JobPool_Pending);
-    case PCONTROL_JOB_BUSY:
-      return (JobPool_Busy);
-    case PCONTROL_JOB_DONE:
-      return (JobPool_Done);
-    case PCONTROL_JOB_KILL:
-      return (JobPool_Kill);
-    case PCONTROL_JOB_EXIT:
-      return (JobPool_Exit);
-    case PCONTROL_JOB_CRASH:
-      return (JobPool_Crash);
-    default:
-      gprint (GP_ERR, "error: unknown job stack : programming error\n");
-      exit (1);
+    case PCONTROL_JOB_PENDING: return (JobPool_Pending);
+    case PCONTROL_JOB_BUSY:    return (JobPool_Busy);
+    case PCONTROL_JOB_DONE:    return (JobPool_Done);
+    case PCONTROL_JOB_KILL:    return (JobPool_Kill);
+    case PCONTROL_JOB_EXIT:    return (JobPool_Exit);
+    case PCONTROL_JOB_CRASH:   return (JobPool_Crash);
   }
   gprint (GP_ERR, "error: unknown job stack : programming error\n");
@@ -191,5 +195,5 @@
   JobID = job[0].JobID;
   PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
-  gprint (GP_ERR, "added new job\n");
+  if (VerboseMode()) gprint (GP_ERR, "added new job\n");
   return (JobID);
 }
@@ -210,26 +214,2 @@
   FREE (job);
 }
-
-/* unlink job and host, pull host from its stack */
-Host *UnlinkJobAndHost (Job *job) {
-
-  Host *host;
-
-  host = (Host *) job[0].host;
-  if (host == NULL) {
-    gprint (GP_ERR, "programming error: job has no host\n");
-    exit (2);
-  }
-
-  /* unlink host & job */
-  job[0].host = NULL;
-  host[0].job = NULL;
-  
-  /* remove host from correct stack */
-  XXXX does this step asuume the host is in this stack??
-  if (PullHostFromStackByID (host[0].stack, host[0].HostID) == NULL) {
-    gprint (GP_ERR, "programming error: host is not found in current stack\n");
-    exit (2);
-  }
-  return (host);
-}
Index: /trunk/Ohana/src/opihi/pcontrol/KillJob.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/KillJob.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/KillJob.c	(revision 8424)
@@ -1,13 +1,12 @@
 # include "pcontrol.h"
 
-int KillJob (Job *job) {
+int KillJob (Job *job, Host *host) {
   
-  Host *host;
   IOBuffer buffer;
   int status;
   char *p;
 
-  /** must have a valid host : if not? **/
-  host = (Host *) job[0].host;
+  ASSERT (host == (Host *) job[0].host, "invalid host");
+  ASSERT (job  == (Job *) host[0].job, "invalid job");
 
   InitIOBuffer (&buffer, 0x100);
@@ -19,11 +18,18 @@
     case PCLIENT_DOWN:
       HarvestHost (host[0].pid);
-      UnlinkJobAndHost (job);
-      PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM);
+      // unlink host & job
+      job[0].host = NULL;
+      host[0].job = NULL;
       PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
+      PutJob (job, PCONTROL_JOB_CRASH, STACK_BOTTOM);
       FreeIOBuffer (&buffer);
       return (FALSE);
 
     case PCLIENT_HUNG:
+      HarvestHost (host[0].pid);
+      // unlink host & job
+      job[0].host = NULL;
+      host[0].job = NULL;
+      PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM);
       PutJobSetState (job, PCONTROL_JOB_BUSY, STACK_BOTTOM, PCONTROL_JOB_HUNG);
       FreeIOBuffer (&buffer);
@@ -31,19 +37,16 @@
 
     case PCLIENT_GOOD:
-      gprint (GP_ERR, "message received (KillJob)\n");  
+      if (VerboseMode()) gprint (GP_ERR, "message received (KillJob)\n");  
       break;
 
     default:
-      gprint (GP_ERR, "unknown status for pclient command: programming error\n");  
-      exit (1);
+      ABORT ("unknown status for pclient command");  
   }
 
   /** host is up, need to parse message **/
   p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
-  if (p == NULL) {
-    gprint (GP_ERR, "missing STATUS in pclient message : programming error\n");
-    exit (1);
-  }
-  gprint (GP_ERR, "client message: %s\n", buffer.buffer);
+  ASSERT (p != NULL, "missing STATUS in pclient message");
+  if (VerboseMode()) gprint (GP_ERR, "client message: %s\n", buffer.buffer);
+
   sscanf (p, "%*s %d", &status);
   FreeIOBuffer (&buffer);
@@ -52,22 +55,21 @@
   switch (status) {
     case -1:
-      gprint (GP_ERR, "programming error (syntax error to pclient)\n");
-      return (FALSE);
-      break;
+      ABORT ("syntax error to pclient");
     case 0:
       gprint (GP_ERR, "failure to kill child process\n");
-      PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM);
+      PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
+      PutJob (job, PCONTROL_JOB_KILL, STACK_BOTTOM);
       return (FALSE);
     case 1:
-      PutJobSetState (job, PCONTROL_JOB_DONE, STACK_BOTTOM, PCONTROL_JOB_CRASH);
+      gprint (GP_ERR, "killing job %s on %s\n", job[0].argv[0], host[0].hostname);
+      // unlink host & job
+      job[0].host = NULL;
+      host[0].job = NULL;
+      PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM);
+      PutJob (job, PCONTROL_JOB_CRASH, STACK_BOTTOM);
       return (TRUE);
     case 2:
-      gprint (GP_ERR, "programming error (client has no job)\n");
-      return (FALSE);
+      ABORT ("client has no job");
   }
-
-  gprint (GP_ERR, "programming error : should not reach here (CheckJob)\n");
-  exit (1);
+  ABORT ("should not reach here (KillJob)");
 }
-
-/** XXX need to do something appropriate with host? ***/
Index: /trunk/Ohana/src/opihi/pcontrol/Makefile
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/Makefile	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/Makefile	(revision 8424)
@@ -19,5 +19,5 @@
 # link flags 
 LIBS    = 	-L$(LIB) -L$(LLIB) -L$(XLIB)
-LIBS1   =       -lsocket -lnsl -lreadline $(TLIB) -lkapa -lFITS -lohana -lX11 -lm
+LIBS1   =       -lsocket -lnsl -lreadline $(TLIB) -lkapa -lFITS -lohana -lX11 -lpthread -lm
 LIBS2   =       -lbasiccmd -lshell -ldata 
 LFLAGS  =       $(LIBS) $(LIBS2) $(LIBS1) 
@@ -27,5 +27,4 @@
 funcs = \
 $(SDIR)/init.$(ARCH).o \
-$(SDIR)/pclient.$(ARCH).o \
 $(SDIR)/pcontrol.$(ARCH).o \
 $(SDIR)/rconnect.$(ARCH).o \
@@ -35,4 +34,5 @@
 $(SDIR)/CheckHost.$(ARCH).o \
 $(SDIR)/CheckIdleHost.$(ARCH).o \
+$(SDIR)/CheckPoint.$(ARCH).o \
 $(SDIR)/CheckSystem.$(ARCH).o \
 $(SDIR)/GetJobOutput.$(ARCH).o \
@@ -41,4 +41,5 @@
 $(SDIR)/JobOps.$(ARCH).o \
 $(SDIR)/StackOps.$(ARCH).o \
+$(SDIR)/PclientCommand.$(ARCH).o \
 $(SDIR)/ResetJob.$(ARCH).o \
 $(SDIR)/StartHost.$(ARCH).o \
Index: /trunk/Ohana/src/opihi/pcontrol/PclientCommand.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/PclientCommand.c	(revision 8424)
+++ /trunk/Ohana/src/opihi/pcontrol/PclientCommand.c	(revision 8424)
@@ -0,0 +1,38 @@
+# include "pcontrol.h"
+# define PCLIENT_TIMEOUT 500
+
+int PclientCommand (Host *host, char *command, char *response, IOBuffer *buffer) {
+
+  int i;
+  int status;
+  char *line;
+  struct timespec request, remain;
+
+  /* avoid blocking on read, test every 100 usec, up to 50 msec */
+  request.tv_sec = 0;
+  request.tv_nsec = 100000;
+
+  ReadtoIOBuffer (buffer, host[0].stdout);
+  FlushIOBuffer (buffer);
+
+  /* send command to client (adding on \n) */
+  status = write_fmt (host[0].stdin, "%s\n", command);
+
+  /* is pipe still open? */
+  if ((status == -1) && (errno == EPIPE)) return (PCLIENT_DOWN);
+  
+  /* watch for response - wait up to 1 second */
+  line = NULL;
+  status = -1;
+  for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (line == NULL); i++) {
+    status = ReadtoIOBuffer (buffer, host[0].stdout);
+    line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer);
+    if (status == -1) nanosleep (&request, &remain);
+  }
+  if (status ==  0) return (PCLIENT_DOWN);
+  if (status == -1) return (PCLIENT_HUNG);
+  /* gprint (GP_ERR, "buffer.buffer: %s\n", buffer[0].buffer); */
+  return (PCLIENT_GOOD);
+}
+
+/* memstr returns a view, not an allocated string : don't free */
Index: /trunk/Ohana/src/opihi/pcontrol/StackOps.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/StackOps.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/StackOps.c	(revision 8424)
@@ -25,7 +25,7 @@
   ALLOCATE (stack[0].id,     int,    stack[0].NOBJECT);
 
-  // we need to use a mutex of type 
-  // stack[0].mutex = PTHREAD_MUTEX_INITIALIZER;
-
+# ifdef THREADED
+  pthread_mutex_init (&stack[0].mutex, NULL);
+# endif
   return (stack);
 }
@@ -72,5 +72,4 @@
 void *PullStackByLocation (Stack *stack, int where) {
 
-  int i;
   void *object;
   
@@ -96,5 +95,5 @@
 void *PullStackByName (Stack *stack, char *name) {
 
-  int i, j;
+  int i;
   void *object;
 
@@ -106,5 +105,5 @@
     /* here is the element of interest */
     object = stack[0].object[i];
-    RemoveStackEntry (i);
+    RemoveStackEntry (stack, i);
     UnlockStack (stack); 
     return (object);
@@ -117,5 +116,5 @@
 void *PullStackByID (Stack *stack, int id) {
 
-  int i, j;
+  int i;
   void *object;
   
@@ -137,4 +136,6 @@
 /* should only be called if you know where is a valid entry */
 int RemoveStackEntry (Stack *stack, int where) {
+
+  int i;
 
   if (where < 0) abort();
@@ -152,8 +153,8 @@
 }
 
-/* should only be called if manually lock the stack */
-int RemoveStackByID (Stack *stack, int id) {
-
-  int i, j;
+/* should only be called if you manually lock the stack */
+void *RemoveStackByID (Stack *stack, int id) {
+
+  int i;
   void *object;
   
@@ -205,10 +206,14 @@
 
 void LockStack (Stack *stack) {
+# ifdef THREADED
+  pthread_mutex_lock (&stack[0].mutex);
+# endif
   return;
 }
 
 void UnlockStack (Stack *stack) {
+# ifdef THREADED
+  pthread_mutex_unlock (&stack[0].mutex);
+# endif
   return;
 }
-
-// Safe with PTHREAD_MUTEX_INITIALIZER lock
Index: /trunk/Ohana/src/opihi/pcontrol/StartHost.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/StartHost.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/StartHost.c	(revision 8424)
@@ -13,4 +13,6 @@
   if (VarConfig ("COMMAND", "%s", command) == NULL) strcpy (command, "ssh");
   if (VarConfig ("SHELL", "%s", shell)     == NULL) strcpy (shell, "pclient");
+
+  gprint (GP_ERR, "starting host within thread %d\n", pthread_self());
 
   pid = rconnect (command, host[0].hostname, shell, stdio);
Index: /trunk/Ohana/src/opihi/pcontrol/StartJob.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/StartJob.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/StartJob.c	(revision 8424)
@@ -1,9 +1,8 @@
 # include "pcontrol.h"
 
-int StartJob (Job *job) {
+int StartJob (Job *job, Host *host) {
 
   int  i, Nline, status;
   char *line, *p;
-  Host *host;
   IOBuffer buffer;
 
@@ -11,9 +10,6 @@
 
   /* job must have assigned host */
-  host = (Host *) job[0].host;
-  if (host == NULL) {
-    gprint (GP_ERR, "programming error: no assigned host\n");
-    exit (1);
-  }
+  ASSERT (host == (Host *) job[0].host, "invalid host");
+  ASSERT (job  == (Job *) host[0].job, "invalid job");
 
   /* construct command line : job arg0 arg1 ... argN\n */
@@ -48,14 +44,11 @@
 
     default:
-      if (VerboseMode()) gprint (GP_ERR, "unknown status for pclient command: programming error\n");  
-      exit (1);
+      ABORT ("unknown status for pclient command");  
   }
 
   /* check on result of pclient command */
   p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
-  if (p == NULL) {
-    gprint (GP_ERR, "programming error: missing STATUS in pclient message\n");
-    exit (1);
-  }
+  ASSERT (p != NULL, "missing STATUS in pclient message");
+
   sscanf (p, "%*s %d", &status);
   switch (status) {
@@ -65,10 +58,8 @@
 
     case -2:
-      gprint (GP_ERR, "programming error: syntax error in pclient command\n");
-      exit (1);
+      ABORT ("syntax error in pclient command");
 
     case -3:
-      gprint (GP_ERR, "programming error: existing child on pclient\n");
-      exit (1);
+      ABORT ("existing child on pclient");
 
     default:
@@ -81,9 +72,8 @@
   }
   /* we should never reach here */
-  gprint (GP_ERR, "programming error: should not reach here (StartJob)\n");
-  exit (1);
+  ABORT ("should not reach here (StartJob)");
 
 failure:
-  /* unlink host & job */
+  // unlink host & job
   job[0].host = NULL;
   host[0].job = NULL;
@@ -94,4 +84,2 @@
   return (FALSE);
 }
-
-/** note : host and job popped off stacks : can't use UnlinkJobAndHost **/
Index: /trunk/Ohana/src/opihi/pcontrol/StopHosts.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/StopHosts.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/StopHosts.c	(revision 8424)
@@ -19,5 +19,4 @@
 int DownHosts () {
 
-  int i, Nobject;
   Stack *stack;
   Host  *host;
@@ -68,10 +67,19 @@
 int HarvestHost (int pid) {
   
-  int       result;
-  int       waitstatus;
+  int i, result, waitstatus;
 
-  /* I probably should loop a few time with max timeout larger than 10ms... */
-  usleep (10000);
-  result = waitpid (pid, &waitstatus, WNOHANG);
+  gprint (GP_ERR, "harvesting within thread %d\n", pthread_self());
+  gprint (GP_ERR, "child process %d is down, wait for exit status\n", pid);
+  
+  // Loop a few times waiting for child to exit
+  for (i = 0; i < 50; i++) {
+    result = waitpid (pid, &waitstatus, WNOHANG);
+    if ((result == -1) && (errno == ECHILD)) {
+      usleep (10000);
+      continue;
+    } else {
+      break;
+    }
+  }
   switch (result) {
     case -1:  /* error with waitpid */
@@ -81,13 +89,9 @@
 	  gprint (GP_ERR, "did process already exit?  programming error?\n");
 	  break;
+	case EINTR:
 	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);
+	  perror ("unexpected error");
+	  ABORT ("(HarvestHost)");
       }
       break;
Index: /trunk/Ohana/src/opihi/pcontrol/check.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/check.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/check.c	(revision 8424)
@@ -1,6 +1,3 @@
 # include "pcontrol.h"
-
-char jobstate[7][32]  = {"PENDING", "BUSY", "HUNG", "DONE", "KILL", "EXIT", "CRASH"};
-char hoststate[5][32] = {"IDLE",    "BUSY", "DOWN", "DONE", "OFF"};
 
 int check (int argc, char **argv) {
@@ -18,14 +15,18 @@
   if (!strcasecmp (argv[1], "JOB")) {
     JobID = atoi (argv[2]);
+
+    SetCheckPoint ();  // ensure the JOB is on one of the stacks
     job = PullJobByID (JobID, &StackID);
     if (job == NULL) {
       gprint (GP_LOG, "job not found\n");
+      ClearCheckPoint ();
       return (FALSE);
     }
-    gprint (GP_LOG, "STATUS %s\n", jobstate[StackID]);
+    gprint (GP_LOG, "STATUS %s\n", GetJobStackName(StackID));
     gprint (GP_LOG, "EXITST %d\n", job[0].exit_status);
     gprint (GP_LOG, "STDOUT %d\n", job[0].stdout_size);
     gprint (GP_LOG, "STDERR %d\n", job[0].stderr_size);
     PutJob (job, StackID, STACK_BOTTOM);
+    ClearCheckPoint ();
     return (TRUE);
   }
@@ -33,11 +34,15 @@
   if (!strcasecmp (argv[1], "HOST")) {
     HostID = atoi (argv[2]);
+
+    SetCheckPoint ();  // ensure the HOST is on one of the stacks
     host = PullHostByID (HostID, &StackID);
     if (host == NULL) {
       gprint (GP_LOG, "host not found\n");
+      ClearCheckPoint ();
       return (FALSE);
     }
-    gprint (GP_LOG, "host %s\n", hoststate[StackID]);
+    gprint (GP_LOG, "host %s\n", GetHostStackName(StackID));
     PutHost (host, StackID, STACK_BOTTOM);
+    ClearCheckPoint ();
     return (TRUE);
   }
@@ -46,4 +51,2 @@
   return (FALSE);
 }
-
-XXX how do I handle objects which are in flight??
Index: /trunk/Ohana/src/opihi/pcontrol/host.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/host.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/host.c	(revision 8424)
@@ -1,7 +1,7 @@
 # include "pcontrol.h"
 
+// we use CheckPoints in this function to prevent objects in flight from being missing.
 int host (int argc, char **argv) {
 
-  int N, Ns;
   int StackID;
   IDtype HostID;
@@ -21,8 +21,10 @@
       return (FALSE);
     }
+    host[0].markoff = FALSE;
     DownHost (host);
     return (TRUE);
   }
   if (!strcasecmp (argv[1], "RETRY")) {
+    // no need to use a check point [thief: CheckDownHost (DOWN->IDLE)]
     host = PullHostFromStackByName (PCONTROL_HOST_DOWN, argv[2]);
     if (!host) {
@@ -39,28 +41,37 @@
   }
   if (!strcasecmp (argv[1], "CHECK")) {
+    SetCheckPoint ();  // ensure the host is on one of the stacks
     host = PullHostByName (argv[2], &StackID);
-    switch (StackID) {
-      case PCONTROL_HOST_IDLE:
-	gprint (GP_LOG, "host %s is IDLE\n", argv[2]);
-      case PCONTROL_HOST_BUSY:
-	gprint (GP_LOG, "host %s is BUSY\n", argv[2]);
-      case PCONTROL_HOST_DONE:
-	gprint (GP_LOG, "host %s is DONE\n", argv[2]);
-      case PCONTROL_HOST_DOWN:
-	gprint (GP_LOG, "host %s is DOWN\n", argv[2]);
-      case PCONTROL_HOST_OFF:
-	gprint (GP_LOG, "host %s is OFF\n", argv[2]);
-      default:
-	gprint (GP_LOG, "host %s not found\n", argv[2]);
-	return (FALSE);
+    if (host == NULL) {
+      gprint (GP_LOG, "host %s not found\n", argv[2]);
+      ClearCheckPoint ();
+      return (FALSE);
     }
     PutHost (host, StackID, STACK_BOTTOM);
-    return (FALSE);
+    ClearCheckPoint ();
+
+    gprint (GP_LOG, "host %s is %s\n", argv[2], GetHostStackName (StackID));
+    return (TRUE);
   }
   if (!strcasecmp (argv[1], "OFF")) {
+    SetCheckPoint (); // ensure we can find the specified host
+    host = PullHostByName (argv[2], &StackID);
+    if (host == NULL) {
+      gprint (GP_LOG, "host %s not found\n", argv[2]);
+      ClearCheckPoint ();
+      return (FALSE);
+    }
+    host[0].markoff = TRUE;
+    PutHost (host, StackID, STACK_BOTTOM);
+    ClearCheckPoint ();
+    return (TRUE);
+  }
+
+# if 0
     host = PullHostFromStackByName (PCONTROL_HOST_IDLE, argv[2]);
     if (host) {
       StopHost (host);
       OffHost (host);
+      ClearCheckPoint ();
       return (TRUE);
     }
@@ -68,17 +79,31 @@
     if (host) {
       OffHost (host);
+      ClearCheckPoint ();
       return (TRUE);
     }
-    /* XXX the 'markoff' flag is not being used */
+    /* XXX the 'markoff' flag is not being checked */
     host = PullHostFromStackByName (PCONTROL_HOST_BUSY, argv[2]);
     if (host) {
       host[0].markoff  = TRUE;
       PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM);
+      ClearCheckPoint ();
       return (TRUE);
     }
-    gprint (GP_LOG, "host %s is not BUSY, IDLE, or DOWN\n", argv[2]);
+    /* XXX the 'markoff' flag is not being checked */
+    host = PullHostFromStackByName (PCONTROL_HOST_DONE, argv[2]);
+    if (host) {
+      host[0].markoff  = TRUE;
+      PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM);
+      ClearCheckPoint ();
+      return (TRUE);
+    }
+    gprint (GP_LOG, "host %s is not found (already off?)\n", argv[2]);
+    ClearCheckPoint ();
     return (FALSE);
   }
+# endif
+
   if (!strcasecmp (argv[1], "DELETE")) {
+    // a check point is not required: no possible thief
     host = PullHostFromStackByName (PCONTROL_HOST_OFF, argv[2]);
     if (!host) {
Index: /trunk/Ohana/src/opihi/pcontrol/init.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/init.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/init.c	(revision 8424)
@@ -8,5 +8,4 @@
 int jobstack    PROTO((int, char **));
 int kill_pc     PROTO((int, char **));
-int pulse       PROTO((int, char **));
 int status      PROTO((int, char **));
 int run         PROTO((int, char **));
@@ -16,4 +15,7 @@
 int verbose     PROTO((int, char **));
 int version         PROTO((int, char **));
+
+// pulse is only available in the un-threaded version
+int pulse       PROTO((int, char **));
 
 static Command cmds[] = {  
@@ -25,5 +27,4 @@
   {"verbose",   verbose,   "set the verbose mode for job"},
   {"version",   version,   "show version information"},
-  {"pulse",     pulse,     "set system pulse"},
   {"job",       job,       "add job"},
   {"jobstack",  jobstack,  "list jobs for a single stack"},
@@ -33,4 +34,7 @@
   {"stderr",    stderr_pc, "get stderr buffer for job"},
   {"stdout",    stdout_pc, "get stdout buffer for job"},
+# ifndef THREADED
+  {"pulse",     pulse,     "set system pulse"},
+# endif
 }; 
 
@@ -42,7 +46,3 @@
     AddCommand (&cmds[i]);
   }
-
-  /* XXX temporary : put this elsewhere? */
-  InitJobStacks ();
-  InitHostStacks ();
 }
Index: unk/Ohana/src/opihi/pcontrol/pclient.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/pclient.c	(revision 8423)
+++ 	(revision )
@@ -1,38 +1,0 @@
-# include "pcontrol.h"
-# define PCLIENT_TIMEOUT 500
-
-int PclientCommand (Host *host, char *command, char *response, IOBuffer *buffer) {
-
-  int i;
-  int status;
-  char *line;
-  struct timespec request, remain;
-
-  /* avoid blocking on read, test every 100 usec, up to 50 msec */
-  request.tv_sec = 0;
-  request.tv_nsec = 100000;
-
-  ReadtoIOBuffer (buffer, host[0].stdout);
-  FlushIOBuffer (buffer);
-
-  /* send command to client (adding on \n) */
-  status = write_fmt (host[0].stdin, "%s\n", command);
-
-  /* is pipe still open? */
-  if ((status == -1) && (errno == EPIPE)) return (PCLIENT_DOWN);
-  
-  /* watch for response - wait up to 1 second */
-  line = NULL;
-  status = -1;
-  for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (line == NULL); i++) {
-    status = ReadtoIOBuffer (buffer, host[0].stdout);
-    line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer);
-    if (status == -1) nanosleep (&request, &remain);
-  }
-  if (status ==  0) return (PCLIENT_DOWN);
-  if (status == -1) return (PCLIENT_HUNG);
-  /* gprint (GP_ERR, "buffer.buffer: %s\n", buffer[0].buffer); */
-  return (PCLIENT_GOOD);
-}
-
-/* memstr returns a view, not an allocated string : don't free */
Index: /trunk/Ohana/src/opihi/pcontrol/pcontrol.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/pcontrol.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/pcontrol.c	(revision 8424)
@@ -10,4 +10,8 @@
 void program_init (int *argc, char **argv) {
   
+# ifdef THREADED  
+  pthread_t clientsThread;
+# endif
+
   auto_break = TRUE;
 
@@ -18,8 +22,17 @@
   gprintInit ();
 
+  InitJobStacks ();
+  InitHostStacks ();
+
   rl_readline_name = opihi_name;
   rl_attempted_completion_function = command_completer;
+# ifdef THREADED
+  pthread_create (&clientsThread, NULL, &CheckSystem_Threaded, NULL);
+  rl_event_hook = NULL;
+  rl_set_keyboard_input_timeout (1000); 
+# else
   rl_event_hook = CheckSystem;
   rl_set_keyboard_input_timeout (1000); 
+# endif  
 
   set_str_variable ("HISTORY", opihi_history);
Index: /trunk/Ohana/src/opihi/pcontrol/run.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/run.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/run.c	(revision 8424)
@@ -8,18 +8,11 @@
   }
 
+# ifdef THREADED
+  SetRunSystem (TRUE);
+# else
   rl_event_hook = CheckSystem;
+# endif
 
   return (TRUE);
 }
 
-int run_threaded (int argc, char **argv) {
-
-  if (argc != 1) {
-    gprint (GP_ERR, "USAGE: run\n");
-    return (FALSE);
-  }
-
-  // some action
-
-  return (TRUE);
-}
Index: /trunk/Ohana/src/opihi/pcontrol/status.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/status.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/status.c	(revision 8424)
@@ -1,6 +1,6 @@
 # include "pcontrol.h"
 
-char jobname[7][32]  = {"PENDING", "BUSY", "HUNG", "DONE", "KILL", "EXIT", "CRASH"};
-char hostname[5][32] = {"IDLE",    "BUSY", "DOWN", "DONE", "OFF"};
+int PrintJobStack (int Nstack);
+int PrintHostStack (int Nstack);
 
 int status (int argc, char **argv) {
@@ -31,5 +31,5 @@
   LockStack (stack);
   Nobject = stack[0].Nobject;
-  gprint (GP_LOG, "job stack %s:  %d objects\n", jobname[Nstack], Nobject);
+  gprint (GP_LOG, "job stack %s:  %d objects\n", GetJobStackName(Nstack), Nobject);
 
   for (i = 0; i < Nobject; i++) {
@@ -57,5 +57,5 @@
   LockStack (stack);
   Nobject = stack[0].Nobject;
-  gprint (GP_LOG, "host stack %s:  %d objects\n", hostname[Nstack], Nobject);
+  gprint (GP_LOG, "host stack %s:  %d objects\n", GetHostStackName(Nstack), Nobject);
 
   for (i = 0; i < Nobject; i++) {
Index: /trunk/Ohana/src/opihi/pcontrol/stop.c
===================================================================
--- /trunk/Ohana/src/opihi/pcontrol/stop.c	(revision 8423)
+++ /trunk/Ohana/src/opihi/pcontrol/stop.c	(revision 8424)
@@ -8,5 +8,9 @@
   }
 
+# ifdef THREADED
+  SetRunSystem (FALSE);
+# else
   rl_event_hook = NULL;
+# endif
 
   return (TRUE);