Index: /trunk/psModules/src/detrend/Makefile.am =================================================================== --- /trunk/psModules/src/detrend/Makefile.am (revision 7769) +++ /trunk/psModules/src/detrend/Makefile.am (revision 7770) @@ -10,5 +10,7 @@ pmNonLinear.c \ pmSubtractBias.c \ - pmDetrendDB.c + pmDetrendDB.c \ + psPipe.c \ + psIOBuffer.c # pmSubtractSky.c Index: /trunk/psModules/src/detrend/pmDetrendDB.c =================================================================== --- /trunk/psModules/src/detrend/pmDetrendDB.c (revision 7769) +++ /trunk/psModules/src/detrend/pmDetrendDB.c (revision 7770) @@ -85,23 +85,34 @@ pmDetrendSelectResults *pmDetrendSelect (pmDetrendSelectOptions *options) { - - char answer[128]; - char line[1024]; + bool status; + char *line = NULL; char *time = psTimeToISO (&options->time); char *type = pmDetrendTypeToString (options->type); - sprintf (line, "detselect -camera %s -time %s -type %s", - options->camera, time, type); + // generate the detselect command + psStringAppend (&line, "detselect -camera %s -time %s -type %s", options->camera, time, type); psFree (time); psFree (type); - // XXX put this in a fork/exec to catch the timeout - FILE *p = popen (line, "r"); - fgets (answer, 127, p); - pclose (p); + // use psPipe to exec the command, wait for response + psIOBuffer *buffer = psIOBufferAlloc (512); + psPipe *pipe = psPipeOpen (line); + status = psIOBufferReadEmpty (buffer, 100, pipe->stdout); + if (!status) { + psError (PS_ERR_IO, false, "detselect is not responding"); + psFree (buffer); + psFree (pipe); + psFree (line); + return NULL; + } + psPipeClose (pipe); + psFree (pipe); + psFree (line); - psList *list = psStringSplit (answer, " ", false); - psArray *array = psListToArray (list); + // XXX need to parse the response more robustly + // XXX for now, assume a single line + psArray *array = psStringSplitArray (buffer->data, " ", false); + psFree (buffer); if (!array) @@ -121,6 +132,4 @@ psFree (array); - psFree (list); - return results; } @@ -130,17 +139,29 @@ char *pmDetrendFile (char *detID, char *classID) { + bool status; + char *line = NULL; - char answer[128]; - char line[1024]; + // generate the detselect command + psStringAppend (&line, "detselect -select -detID %s -classID %s", detID, classID); - sprintf (line, "detselect -select -detID %s -classID %s", detID, classID); + // use psPipe to exec the command, wait for response + psIOBuffer *buffer = psIOBufferAlloc (512); + psPipe *pipe = psPipeOpen (line); + status = psIOBufferReadEmpty (buffer, 100, pipe->stdout); + if (!status) { + psError (PS_ERR_IO, false, "detselect is not responding"); + psFree (buffer); + psFree (pipe); + psFree (line); + return NULL; + } + psPipeClose (pipe); + psFree (pipe); + psFree (line); - // XXX put this in a fork/exec to catch the timeout - FILE *p = popen (line, "r"); - fgets (answer, 127, p); - pclose (p); - - psList *list = psStringSplit (answer, " ", false); - psArray *array = psListToArray (list); + // XXX need to parse the response more robustly + // XXX for now, assume a single line + psArray *array = psStringSplitArray (buffer->data, " ", false); + psFree (buffer); if (!array) @@ -154,6 +175,4 @@ psFree (array); - psFree (list); - return result; } Index: /trunk/psModules/src/detrend/pmDetrendDB.h =================================================================== --- /trunk/psModules/src/detrend/pmDetrendDB.h (revision 7769) +++ /trunk/psModules/src/detrend/pmDetrendDB.h (revision 7770) @@ -14,6 +14,6 @@ * @author EAM, IfA * -* @version $Revision: 1.1 $ $Name: not supported by cvs2svn $ -* @date $Date: 2006-06-17 01:50:43 $ +* @version $Revision: 1.2 $ $Name: not supported by cvs2svn $ +* @date $Date: 2006-07-01 00:00:11 $ * * Copyright 2004-2005 Institute for Astronomy, University of Hawaii @@ -66,3 +66,33 @@ char *pmDetrendFile (char *detID, char *classID); +// move these to pslib?? +typedef struct +{ + int stdin; + int stdout; + int stderr; +} +psPipe; + +typedef struct +{ + char *data; + int nAlloc; // current size of allocated buffer + int nReset; // size to set buffer after flush + int nBlock; // number of bytes to try to read at a time + int n; // current size of filled data +} +psIOBuffer; + +// psIOBuffer functions +psIOBuffer *psIOBufferAlloc (int nBuffer); +bool psIOBufferFlush (psIOBuffer *buffer); +int psIOBufferRead (psIOBuffer *buffer, int fd); +int psIOBufferReadEmpty (psIOBuffer *buffer, int maxRetries, int fd); + +// psPipe functions +psPipe *psPipeAlloc (); +psPipe *psPipeOpen (char *command); +bool psPipeClose (psPipe *pipe); + # endif Index: /trunk/psModules/src/detrend/psIOBuffer.c =================================================================== --- /trunk/psModules/src/detrend/psIOBuffer.c (revision 7770) +++ /trunk/psModules/src/detrend/psIOBuffer.c (revision 7770) @@ -0,0 +1,106 @@ +# include +# include +# include "pmFPA.h" +# include "pmDetrendDB.h" + +static void psIOBufferFree (psIOBuffer *buffer) +{ + if (buffer == NULL) + return; + + psFree (buffer->data); + return; +} + +psIOBuffer *psIOBufferAlloc (int nBuffer) +{ + + psIOBuffer *buffer = (psIOBuffer *)psAlloc(sizeof(psIOBuffer)); + psMemSetDeallocator(buffer, (psFreeFunc) psIOBufferFree); + + buffer->data = (char *) psAlloc (nBuffer); + + buffer->nAlloc = 0; + buffer->nReset = 0; + buffer->nBlock = 0; + buffer->n = 0; + return (buffer); +} + +bool psIOBufferFlush (psIOBuffer *buffer) +{ + + if (buffer == NULL) + return false; + buffer->n = 0; + buffer->nAlloc = buffer->nReset; + psRealloc (buffer->data, buffer->nAlloc); + bzero (buffer->data, buffer->nAlloc); + + return true; +} + +int psIOBufferRead (psIOBuffer *buffer, int fd) +{ + + int Nread, Nfree; + + if (fd == 0) { + /* pipe is closed */ + return (0); + } + + Nfree = buffer->nAlloc - buffer->n; + + // extend the data block if needed + if (Nfree < buffer->nBlock) { + buffer->nAlloc += 2*buffer->nBlock; + psRealloc (buffer->data, buffer->nAlloc); + Nfree = buffer->nAlloc - buffer->n; + bzero (buffer->data + buffer->n, Nfree); + } + + // attempt to read from the fd into the buffer + Nread = read (fd, &buffer->data[buffer->n], buffer->nBlock); + + if (Nread >= 0) { + buffer->n += Nread; + return (Nread); + } + + // check on exit status (try again if waiting for non-blocking fd) + if (Nread == -1) { + switch (errno) { + case EAGAIN: + case EIO: + /** no data available in pipe **/ + return (-1); + default: + /** error reading from pipe **/ + psError (PS_ERR_IO, true, "error on psIOBufferRead"); + return (-2); + } + } + return (Nread); +} + +/* read until buffer is empty (Nmax retries) */ +int psIOBufferReadEmpty (psIOBuffer *buffer, int Nmax, int fd) +{ + + int i, status; + + status = -1; + for (i = 0; (status != 0) && (i < Nmax); i++) { + status = psIOBufferRead (buffer, fd); + if (status == -2) + return false; + if (status == -1) + usleep (10000); + if (status > 0) + i = 0; + } + if (status == -1) + return false; + return true; +} Index: /trunk/psModules/src/detrend/psPipe.c =================================================================== --- /trunk/psModules/src/detrend/psPipe.c (revision 7770) +++ /trunk/psModules/src/detrend/psPipe.c (revision 7770) @@ -0,0 +1,144 @@ +# include +# include +# include +# include +# include "pmFPA.h" +# include "pmDetrendDB.h" + +void closePipes (int *stdin_fd, int *stdout_fd, int *stderr_fd) +{ + + if (stdin_fd[0] != 0) + close (stdin_fd[0]); + if (stdin_fd[1] != 0) + close (stdin_fd[0]); + 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]); +} + +static void psPipeFree (psPipe *pipe) +{ + return; +} + +psPipe *psPipeAlloc () +{ + psPipe *pipe = (psPipe *)psAlloc(sizeof(psPipe)); + psMemSetDeallocator(pipe, (psFreeFunc) psPipeFree); + + pipe->stdin = 0; + pipe->stdout = 0; + pipe->stderr = 0; + return (pipe); +} + +psPipe *psPipeOpen (char *command) +{ + + int stdin_fd[2], stdout_fd[2], stderr_fd[2], status; + pid_t pid; + + bzero (stdin_fd, 2*sizeof(int)); + bzero (stdout_fd, 2*sizeof(int)); + bzero (stderr_fd, 2*sizeof(int)); + + if (pipe (stdin_fd) < 0) { + psError (PS_ERR_UNKNOWN, true, "cannot create pipe file descriptor"); + closePipes (stdin_fd, stdout_fd, stderr_fd); + return NULL; + } + if (pipe (stdout_fd) < 0) { + psError (PS_ERR_UNKNOWN, true, "cannot create pipe file descriptor"); + closePipes (stdin_fd, stdout_fd, stderr_fd); + return NULL; + } + if (pipe (stderr_fd) < 0) { + psError (PS_ERR_UNKNOWN, true, "cannot create pipe file descriptor"); + closePipes (stdin_fd, stdout_fd, stderr_fd); + return NULL; + } + + psArray *cmd = psStringSplitArray (command, " ", false); + if (cmd->n <= 0) { + psError (PS_ERR_UNKNOWN, true, "empty command for pipe"); + psFree (cmd); + closePipes (stdin_fd, stdout_fd, stderr_fd); + return NULL; + } + + // create the command line array needed by execvp + char **argv = (char **)psAlloc((cmd->n+1)*sizeof(char *)); + for (int i = 0; i < cmd->n; i++) { + argv[i] = cmd->data[i]; + } + argv[cmd->n] = NULL; + + pid = fork (); + if (!pid) { /* must be child process */ + /* 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); + // fprintf (stderr, "error starting remote shell process\n"); + + // this statement exits the child, not the parent, process + exit (1); + } + psFree (cmd); + psFree (argv); + + if (pid == -1) { + psError (PS_ERR_UNKNOWN, true, "unable to create child process"); + closePipes (stdin_fd, stdout_fd, stderr_fd); + return NULL; + } + + /* 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); + + psPipe *pipe = psPipeAlloc(); + + pipe->stdin = stdin_fd[1]; + pipe->stdout = stdout_fd[0]; + pipe->stderr = stderr_fd[0]; + + return (pipe); +} + +bool psPipeClose (psPipe *pipe) +{ + + close (pipe->stdin); + close (pipe->stdout); + close (pipe->stderr); + return true; +} +