Index: /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/Makefile
===================================================================
--- /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/Makefile	(revision 33364)
+++ /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/Makefile	(revision 33365)
@@ -42,5 +42,6 @@
 $(SRC)/AssignSkyToHost.$(ARCH).o        \
 $(SRC)/CopyToHostLocation.$(ARCH).o     \
-$(SRC)/CopyFromHostLocation.$(ARCH).o
+$(SRC)/CopyFromHostLocation.$(ARCH).o   \
+$(SRC)/PclientCommands.$(ARCH).o
 
 OLD = \
Index: /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/include/dvodist.h
===================================================================
--- /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/include/dvodist.h	(revision 33364)
+++ /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/include/dvodist.h	(revision 33365)
@@ -4,4 +4,6 @@
 # include <md5.h>
 # include <unistd.h> // needed?
+
+# define PCLIENT_PROMPT "pclient:"
 
 // used by the md5 stuff
@@ -37,2 +39,7 @@
 
 int           get_md5_with_copy (char *input, char *output, md5_byte_t *digest);
+
+int CheckBusyJob (HostInfo *host, IOBuffer *stdout_buf, IOBuffer *stderr_buf);
+int GetJobOutput (char *command, HostInfo *host, IOBuffer *output, int size);
+int PclientResponse (HostInfo *host, char *response, IOBuffer *buffer);
+int PclientCommand (HostInfo *host, char *command, IOBuffer *buffer);
Index: /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/CopyToHostLocation.c
===================================================================
--- /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/CopyToHostLocation.c	(revision 33364)
+++ /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/CopyToHostLocation.c	(revision 33365)
@@ -4,4 +4,8 @@
 # define DELETE_ORIGINAL 0
 // make this a user option
+
+int get_md5_from_remote (HostInfo *host, char *filename, md5_byte_t *digest);
+int get_md5_from_pclient (HostInfo *host, char *filename, md5_byte_t *digest);
+int hexchar_to_int (char input);
 
 // XXX collapse this and CopyFromHostLocation (only difference is name of src/tgt and bits set on success)
@@ -95,6 +99,6 @@
 	
       // need to re-open and re-read tgtname to check md5sum
-      // XXX this would be safer and faster if we used the remote shell to perform the md5sum locally.  
-      if (!get_md5_with_copy (tgtname, NULL, tgtDigest)) {
+      char *absname = abspath (tgtname, 1024);
+      if (!get_md5_from_pclient (host, absname, tgtDigest)) {
 	fprintf (stderr, "error reading %s or getting md5\n", tgtname);
 	success = FALSE;
@@ -135,9 +139,9 @@
 	sprintf (tgtname, "%s/%s.%s", host->pathname, skylist->regions[i]->name, extname[j]);
 	sprintf (srcname, "%s/%s.%s", catdir, skylist->regions[i]->name, extname[j]);
-
-	// do not rename or unlink original if it does not exist..
-	status = stat (srcname, &filestat);
-	if (status && (errno == ENOENT)) continue;  
-	
+ 
+ 	// do not rename or unlink original if it does not exist..
+ 	status = stat (srcname, &filestat);
+ 	if (status && (errno == ENOENT)) continue;  
+
 	if (!DEBUG && DELETE_ORIGINAL) {
 	  unlink (srcname);
@@ -190,2 +194,111 @@
  * read the SPLIT headers to see if MEASURE, MISSING, SECFILT fields exist?
  */
+
+// XXX this would be better controlled if we used a remote pclient.  then we could
+// check things like the exit status and 
+int get_md5_from_remote (HostInfo *host, char *filename, md5_byte_t *digest) {
+
+  IOBuffer buffer;
+
+  char line[1024];
+  int Nline = snprintf (line, 1024, "md5sum %s\n", filename);
+  assert (Nline < 1024);
+
+  write (host->stdio[0], line, Nline);
+
+  InitIOBuffer (&buffer, 100);
+  int status = EmptyIOBuffer (&buffer, 10000, host->stdio[1]);
+  fprintf (stderr, "md5sum status: %d\n", status);
+      
+  fprintf (stderr, "buffer: %s\n", buffer.buffer);
+
+  char result[1024], fileout[1024];
+  int Nscan = sscanf (buffer.buffer, "%s %s", result, fileout);
+  assert (Nscan == 2);
+  assert (strlen(result) == NDIGEST);
+  
+  memcpy (digest, result, NDIGEST);
+  return TRUE;
+}
+
+// pclient version of remote client
+int get_md5_from_pclient (HostInfo *host, char *filename, md5_byte_t *digest) {
+
+  IOBuffer buffer, stdout_buf, stderr_buf;
+
+  char line[1024];
+  int Nline = snprintf (line, 1024, "job md5sum %s\n", filename);
+  assert (Nline < 1024);
+
+  // fprintf (stderr, "command: %s\n", line);
+
+  // start the md5sum command
+  InitIOBuffer (&buffer, 100);
+  PclientCommand (host, line, &buffer);
+  PclientResponse (host, PCLIENT_PROMPT, &buffer);
+  // XXX for the moment, abort if anything goes wrong
+
+  // wait for result
+  while (!CheckBusyJob (host, &stdout_buf, &stderr_buf)) {
+    usleep (10000);
+  }
+  if (stderr_buf.Nbuffer) {
+    fprintf (stderr, "error in md5sum: %s\n", stderr_buf.buffer);
+    exit (1);
+  }
+
+  // fprintf (stderr, "result from md5sum: %s\n", stdout_buf.buffer);
+
+  char result[1024], fileout[1024];
+  int Nscan = sscanf (stdout_buf.buffer, "%s %s", result, fileout);
+  assert (Nscan == 2);
+  assert (strlen(result) == 2*NDIGEST);
+  
+  // result is the set of 0-f nibbles, convert to the digest
+  int i;
+  for (i = 0; i < NDIGEST; i++) {
+    int value1 = hexchar_to_int (result[2*i+0]);
+    int value2 = hexchar_to_int (result[2*i+1]);
+    digest[i] = (value1 << 4) + value2;
+    // fprintf (stderr, "%02x %c %c\n", digest[i], result[2*i], result[2*i+1]);
+  }
+  
+  // need to reset pclient for next command...
+  PclientCommand (host, "reset", &buffer);
+  PclientResponse (host, PCLIENT_PROMPT, &buffer);
+
+  return TRUE;
+}
+
+int hexchar_to_int (char input) {
+
+  switch (input) {
+    case '0':
+    case '1':
+    case '2':
+    case '3':
+    case '4':
+    case '5':
+    case '6':
+    case '7':
+    case '8':
+    case '9':
+      return input - '0';
+    case 'a':
+    case 'b':
+    case 'c':
+    case 'd':
+    case 'e':
+    case 'f':
+      return input - 'a' + 10;
+    case 'A':
+    case 'B':
+    case 'C':
+    case 'D':
+    case 'E':
+    case 'F':
+      return input - 'A' + 10;
+    default:
+      abort();
+  }
+}
Index: /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/HostTableLoad.c
===================================================================
--- /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/HostTableLoad.c	(revision 33364)
+++ /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/HostTableLoad.c	(revision 33365)
@@ -5,10 +5,9 @@
 
   int i;
-  int stdio[3];
   char command[64];
   char shell[64];
 
   strcpy (command, "ssh");
-  strcpy (shell, "csh");
+  strcpy (shell, "pclient");
 
   for (i = 0; i < table->Nhosts; i++) {
@@ -18,5 +17,5 @@
 
     int errorInfo;
-    int pid = rconnect (command, table->hosts[i].hostname, shell, stdio, &errorInfo, TRUE);
+    int pid = rconnect (command, table->hosts[i].hostname, shell, table->hosts[i].stdio, &errorInfo, TRUE);
     if (!pid) {     
       /** failure to start: extend retry period **/
@@ -24,11 +23,6 @@
       exit (1);
     }
-
-    // XXX check anything else?
-
-    // close the connection
-    close(stdio[0]);
-    close(stdio[1]);
-    close(stdio[2]);
+    table->hosts[i].pid = pid;
+    // keep the connection open to use for md5sum check
 
     int status = check_dir_access (table->hosts[i].pathname, DEBUG);
Index: /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/PclientCommands.c
===================================================================
--- /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/PclientCommands.c	(revision 33365)
+++ /branches/eam_branches/ipp-20111122/Ohana/src/photdbc/src/PclientCommands.c	(revision 33365)
@@ -0,0 +1,179 @@
+# include "dvodist.h"
+# define PCLIENT_TIMEOUT 100
+# define DEBUG 0
+
+// send a command and check for errors; ignore output
+int PclientCommand (HostInfo *host, char *command, IOBuffer *buffer) {
+
+  int status;
+
+  // flush the stdout and stderr buffers here
+  ReadtoIOBuffer (buffer, host->stdio[1]);
+  FlushIOBuffer (buffer);
+  ReadtoIOBuffer (buffer, host->stdio[2]);
+  FlushIOBuffer (buffer);
+
+  /* send command to client (adding on \n) */
+  status = write_fmt (host->stdio[0], "%s\n", command);
+
+  /* is pipe still open? */
+  if ((status == -1) && (errno == EPIPE)) {
+    fprintf (stderr, "pclient read gives pipe error for %s\n", command);
+    abort();
+  }
+
+  FlushIOBuffer (buffer);
+
+  return TRUE;
+}
+  
+// check for response; message must end with specified string.
+// accumulate the response in the buffer
+int PclientResponse (HostInfo *host, char *response, IOBuffer *buffer) {
+
+  int i;
+  int status;
+  char *line;
+  struct timespec request, remain;
+
+  /* avoid blocking very long on read, test every 100 usec, up to 0.1 sec */
+  request.tv_sec = 0;
+  request.tv_nsec = 100000;
+
+  /* watch for response - wait up to 1 second */
+  line = NULL;
+  status = -1;
+
+  // how long does each cycle really take?
+  for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (line == NULL); i++) {
+    status = ReadtoIOBuffer (buffer, host->stdio[1]);
+    line = memstr (buffer->buffer, response, buffer->Nbuffer);
+    if (status == -1) nanosleep (&request, &remain);
+  }
+  if (status ==  0) {
+    fprintf (stderr, "pclient read returns 0 for %s\n", response);
+    abort();
+  }
+  assert (line != NULL); // return (PCLIENT_HUNG);
+  assert (status != -1); // return (PCLIENT_HUNG);
+
+  // fprintf (stderr, "response: %s\n", buffer->buffer);
+
+  return TRUE;
+}
+
+int GetJobOutput (char *command, HostInfo *host, IOBuffer *output, int size) {
+  
+  char *line;
+  int i, status;
+  struct timespec request, remain;
+
+  InitIOBuffer (output, 100);
+
+  /* avoid blocking on waitpid, test every 100 usec, up to 10 msec */
+  request.tv_sec = 0;
+  request.tv_nsec = 100000;
+
+  status = write_fmt (host->stdio[0], "%s\n", command);
+
+  if ((status == -1) && (errno == EPIPE)) {
+    fprintf (stderr, "host crashed\n");
+    exit (2);
+  }
+
+  // attempt to read the output->size bytes from the host 
+  for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (output->Nbuffer < size); i++) {
+    status = ReadtoIOBuffer (output, host->stdio[1]);
+    if (status == -1) nanosleep (&request, &remain);
+  }
+  if (DEBUG) fprintf (stderr, "Read %d of %d bytes so far\n", output->Nbuffer, size);
+  assert (status != 0);		   // return PCLIENT_DOWN;
+  assert (output->Nbuffer >= size); // return PCLIENT_HUNG;
+
+  // keep trying to read until we get the prompt
+  line = NULL;
+  status = -1;
+  for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (line == NULL); i++) {
+    status = ReadtoIOBuffer (output, host->stdio[1]);
+    line = memstr (output->buffer, PCLIENT_PROMPT, output->Nbuffer);
+    if (status == -1) nanosleep (&request, &remain);
+  }
+  if (DEBUG) fprintf (stderr, "Read %d of %d bytes so far\n", output->Nbuffer, size);
+  assert (status != 0);	      // return PCLIENT_DOWN;
+  assert (line);		      // return PCLIENT_HUNG;
+  return TRUE;
+}
+
+int CheckBusyJob (HostInfo *host, IOBuffer *stdout_buf, IOBuffer *stderr_buf) {
+
+  int status;
+  char *p;
+  char string[64];
+  IOBuffer buffer;
+
+  InitIOBuffer (&buffer, 100);
+  status = PclientCommand (host, "status", &buffer);
+  status = PclientResponse (host, PCLIENT_PROMPT, &buffer);
+
+  /** need to parse message **/
+  p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer);
+  if (p == NULL) {
+    if (DEBUG) fprintf (stderr, "missing STATUS in response; try again?\n");
+    return FALSE;
+  }
+
+  sscanf (p, "%*s %s", string);
+  assert (strcmp(string, "NONE")); // "no current job\n");
+  
+  /** no status change, return to BUSY stack **/
+  if (!strcmp(string, "BUSY")) {
+    if (DEBUG) fprintf (stderr, "not yet done...\n");
+    return FALSE;
+  }
+  
+  /* exit status better be either EXIT or CRASH */
+  if (!strcmp(string, "CRASH")) {
+    fprintf (stderr, "crash on host...\n");
+    exit (1);
+  }
+
+  assert (!strcmp(string, "EXIT"));
+
+  int exit_status, stdout_buf_size, stderr_buf_size;
+
+  /* parse the exit status and sizes of output buffers */
+  p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer);
+  sscanf (p, "%*s %d", &exit_status);
+  p = memstr (buffer.buffer, "STDOUT", buffer.Nbuffer);
+  sscanf (p, "%*s %d", &stdout_buf_size);
+  p = memstr (buffer.buffer, "STDERR", buffer.Nbuffer);
+  sscanf (p, "%*s %d", &stderr_buf_size);
+
+  // XXX runaway job if output too large?
+  if (stdout_buf_size > 0x1000000) abort();
+  if (stderr_buf_size > 0x1000000) abort();
+
+  if (stdout_buf_size) {
+    GetJobOutput ("stdout", host, stdout_buf, stdout_buf_size);
+  } else {
+    ReadtoIOBuffer (&buffer, host->stdio[1]);
+    FlushIOBuffer (&buffer);
+  }
+  if (stderr_buf_size) {
+    GetJobOutput ("stderr", host, stderr_buf, stderr_buf_size);
+  } else {
+    ReadtoIOBuffer (&buffer, host->stdio[2]);
+    FlushIOBuffer (&buffer);
+  } 
+
+  return TRUE;
+}
+
+/* memstr returns a view, not an allocated string : don't free */
+/* ReadtoIOBuffer returns : 
+    0 - pipe closed
+   -1 - no more data in pipe, data not ready
+   -2 - serious error reading from pipe
+   >0 - data read from pipe
+*/
+