Changeset 20047 for trunk/Ohana/src/opihi/pcontrol/GetJobOutput.c

Timestamp:

Oct 10, 2008, 12:51:04 PM (18 years ago)

Author:

eugene

Message:

allow job stdout / stderr from pclient to arrive slowly

File:

• trunk/Ohana/src/opihi/pcontrol/GetJobOutput.c (modified) (1 diff)

trunk/Ohana/src/opihi/pcontrol/GetJobOutput.c

@@ -1 +1 @@
 # include "pcontrol.h"
-# define PCLIENT_TIMEOUT 500
+# define PCLIENT_TIMEOUT 100
+# define DEBUG 0
 
-/* we read Nbytes from the host, then watch for the prompt */ 
-int GetJobOutput (char *cmd, Host *host, IOBuffer *buffer, int Nbytes) {
+// we are trying to read a total of Nbytes from the host.  This function may be called
+// repeatedly until the buffer has the complete set of data.  We need to read output[0].size
+// bytes, then look for the PCLIENT_PROMPT in the output stream
+
+int GetJobOutput (char *command, Host *host, JobOutput *output) {
   
-  int i, status, Nstart;
   char *line;
+  int i, status;
   struct timespec request, remain;
 
-  ASSERT (cmd, "cmd missing");
+  ASSERT (command, "command missing");
   ASSERT (host, "host missing");
-  ASSERT (buffer, "buffer missing");
+  ASSERT (output, "output missing");
 
-  /* flush any earlier messages */
-  ReadtoIOBuffer (buffer, host[0].stdout_fd);
-  FlushIOBuffer (buffer);
-  Nstart = buffer[0].Nbuffer;
-
-  /* avoid blocking on waitpid, test every 100 usec, up to 50 msec */
+  /* avoid blocking on waitpid, test every 100 usec, up to 10 msec */
   request.tv_sec = 0;
   request.tv_nsec = 100000;
 
-  /* send cmd (stdout / stderr) */
-  status = write_fmt (host[0].stdin_fd, "%s\n", cmd);
+  if (!output[0].requested) {
+      /* send command (stdout / stderr) */
+      status = write_fmt (host[0].stdin_fd, "%s\n", command);
 
-  /* is pipe still open? */
-  if ((status == -1) && (errno == EPIPE)) return (PCLIENT_DOWN);
+      /* is pipe still open? */
+      if ((status == -1) && (errno == EPIPE)) return PCLIENT_DOWN;
+      output[0].requested = TRUE;
+  }
 
-  /* read at least Nbytes, then watch for PCLIENT_PROMPT */
+  if (output[0].completed) return PCLIENT_GOOD;
+
+  // attempt to read the output->size bytes from the host 
+  if (output[0].buffer.Nbuffer < output[0].size) {
+      status = -1;
+      for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (output[0].buffer.Nbuffer < output[0].size); i++) {
+          status = ReadtoIOBuffer (&output[0].buffer, host[0].stdout_fd);
+          if (status == -1) nanosleep (&request, &remain);
+      }
+      if (VerboseMode()) gprint (GP_ERR, "%s\n Read %d of %d bytes so far\n", output[0].buffer.buffer, output[0].buffer.Nbuffer, output[0].size);
+      if (status == 0) {
+          if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname);
+          return PCLIENT_DOWN;
+      }
+      if (output[0].buffer.Nbuffer < output[0].size) {
+          if (VerboseMode()) gprint (GP_ERR, "host %s still has data, keep trying\n", host[0].hostname);
+          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 (buffer, host[0].stdout_fd);
-    if ((buffer[0].Nbuffer - Nstart) >= Nbytes) {
-      line = memstr (buffer[0].buffer, PCLIENT_PROMPT, buffer[0].Nbuffer);
-    }
+    status = ReadtoIOBuffer (&output[0].buffer, host[0].stdout_fd);
+    line = memstr (output[0].buffer.buffer, PCLIENT_PROMPT, output[0].buffer.Nbuffer);
     if (status == -1) nanosleep (&request, &remain);
   }
-  if (status ==  0) return (PCLIENT_DOWN);
-  if (status == -1) return (PCLIENT_HUNG);
-
-  /* check on success of pclient command */
-  switch (status) {
-    case -1:
-      if (VerboseMode()) gprint (GP_ERR, "host %s is not responding\n", host[0].hostname);
-      return (FALSE);
-
-    case 0:
-      if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname);
-      return (FALSE);
-
-    default:
-      if (VerboseMode()) gprint (GP_ERR, "message received (GetJobOutput : %s)\n", cmd);  
-      /* drop extra bytes from pclient (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 (TRUE);
+  if (VerboseMode()) gprint (GP_ERR, "%s\n Read %d of %d bytes so far\n", output[0].buffer.buffer, output[0].buffer.Nbuffer, output[0].size);
+  if (status == 0) {
+    if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname);
+    return PCLIENT_DOWN;
+  }
+  if (line == NULL) {
+    if (VerboseMode()) gprint (GP_ERR, "host %s not yet at prompt, keep trying\n", host[0].hostname);
+    return PCLIENT_HUNG;
   }
 
-  gprint (GP_ERR, "programming error: should not reach here (GetJobOutput)\n");
-  pcontrol_exit (50);
+  output[0].completed = TRUE;
+  return PCLIENT_GOOD;
 }