Index: trunk/Ohana/src/libohana/Makefile =================================================================== --- trunk/Ohana/src/libohana/Makefile (revision 5237) +++ trunk/Ohana/src/libohana/Makefile (revision 5273) @@ -35,4 +35,6 @@ $(SRC)/config.$(ARCH).o \ $(SRC)/Fread.$(ARCH).o \ +$(SRC)/IOBufferOps.$(ARCH).o \ +$(SRC)/CommOps.$(ARCH).o \ $(SRC)/version.$(ARCH).o Index: trunk/Ohana/src/libohana/include/ohana.h =================================================================== --- trunk/Ohana/src/libohana/include/ohana.h (revision 5237) +++ trunk/Ohana/src/libohana/include/ohana.h (revision 5273) @@ -87,4 +87,13 @@ # endif /* FOPEN */ +/* socket / pipe communication buffer */ +typedef struct { + char *buffer; + int Nalloc; + int Nreset; + int Nblock; + int Nbuffer; +} IOBuffer; + extern double hypot(); @@ -143,4 +152,18 @@ int str_to_radec PROTO((double *ra, double *dec, char *str1, char *str2)); +/* IO Buffer functions */ +int InitIOBuffer (IOBuffer *buffer, int Nalloc); +int FlushIOBuffer (IOBuffer *buffer); +int ReadtoIOBuffer (IOBuffer *buffer, int fd); +int EmptyIOBuffer (IOBuffer *buffer, int Nmax, int fd); +void FreeIOBuffer (IOBuffer *buffer); + +/* communication functions */ +int ExpectMessage (int device, double timeout, IOBuffer *message); +int ExpectCommand (int device, int length, double timeout, IOBuffer *buffer); +int SendMessage (int device, char *format, ...); +int SendCommand (int device, int length, char *format, ...); +int SendCommandV (int device, int length, char *format, va_list argp); + /* # define F_SETFL 4 Index: trunk/Ohana/src/libohana/src/CommOps.c =================================================================== --- trunk/Ohana/src/libohana/src/CommOps.c (revision 5273) +++ trunk/Ohana/src/libohana/src/CommOps.c (revision 5273) @@ -0,0 +1,114 @@ +# include + +int ExpectMessage (int device, double timeout, IOBuffer *message) { + + int status, length; + IOBuffer command; + + status = ExpectCommand (device, 16, timeout, &command); + if (status != 0) { + FreeIOBuffer (&command); + return (status); + } + + /* buffer contains an EOL NULL, we can just sscan it */ + sscanf (command.buffer, "%*s %d", &length); + FreeIOBuffer (&command); + + status = ExpectCommand (device, length, timeout, message); + return (status); +} + +int ExpectCommand (int device, int length, double timeout, IOBuffer *buffer) { + + /* read from device until we have length bytes, or timeout */ + + int Nread; + double dtime; + struct timespec request, remain; + struct timeval start, stop; + + gettimeofday (&start, NULL); + + /* avoid blocking on waitpid, test every 1000 usec, up to timeout msec */ + request.tv_sec = 0; + request.tv_nsec = 1000000; + + InitIOBuffer (buffer, length + 1); + + while (buffer[0].Nbuffer < length) { + Nread = read (device, &buffer[0].buffer[buffer[0].Nbuffer], length - buffer[0].Nbuffer); + + if (Nread > 0) { + buffer[0].Nbuffer += Nread; + continue; + } + + if (Nread == -1) { + switch (errno) { + case EAGAIN: + case EIO: + /** no data available in pipe, wait a bit, check for timeout **/ + nanosleep (&request, &remain); + break; + default: + /** error reading from pipe **/ + perror ("ReadtoIOBuffer read error"); + return (-2); + } + } + + if (Nread == 0) return (-3); + + gettimeofday (&stop, NULL); + dtime = DTIME (stop, start); + if (dtime > timeout) return (-1); + } + return (0); +} + +/* send a message of arbitrary size, sending the size first */ +int SendMessage (int device, char *format, ...) { + + int Nbyte, status; + char tmp; + va_list argp; + + va_start (argp, format); + Nbyte = vsnprintf (&tmp, 0, format, argp); + va_end (argp); + + if (!Nbyte) return (FALSE); + + va_start (argp, format); + SendCommand (device, 16, "NBYTES: %6d", Nbyte); + status = SendCommandV (device, Nbyte, format, argp); + va_end (argp); + return (status); +} + +int SendCommand (int device, int length, char *format, ...) { + + int status; + va_list argp; + + va_start (argp, format); + status = SendCommandV (device, length, format, argp); + va_end (argp); + return (status); +} + +int SendCommandV (int device, int length, char *format, va_list argp) { + + char *string; + + /* I allocated and zero 1 extra byte */ + ALLOCATE (string, char, length + 1); + memset (string, 0, length + 1); + vsnprintf (string, length + 1, format, argp); + + /* fprintf (stderr, "msg: %s\n", string); */ + write (device, string, length); + free (string); + return (TRUE); +} Index: trunk/Ohana/src/libohana/src/IOBufferOps.c =================================================================== --- trunk/Ohana/src/libohana/src/IOBufferOps.c (revision 5273) +++ trunk/Ohana/src/libohana/src/IOBufferOps.c (revision 5273) @@ -0,0 +1,85 @@ +# include + +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); + bzero (buffer[0].buffer, buffer[0].Nalloc); + + return (TRUE); +} + +int ReadtoIOBuffer (IOBuffer *buffer, int fd) { + + int Nread, Nfree; + + if (fd == 0) { + /* pipe is closed */ + return (0); + } + + 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; + bzero (buffer[0].buffer + buffer[0].Nbuffer, Nfree); + } + + 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); + } + } + return (Nread); +} + +/* read until buffer is empty (Nmax retries) */ +int EmptyIOBuffer (IOBuffer *buffer, int Nmax, int fd) { + + int i, status; + + status = -1; + for (i = 0; (status != 0) && (i < Nmax); i++) { + status = ReadtoIOBuffer (buffer, fd); + if (status == -1) usleep (10000); + if (status > 0) i = 0; + } + if (status == -1) return (FALSE); + return (TRUE); +} + +void FreeIOBuffer (IOBuffer *buffer) { + + if (buffer[0].buffer != (char *) NULL) { + free (buffer[0].buffer); + } +} Index: trunk/Ohana/src/libohana/src/ohana_allocate.c =================================================================== --- trunk/Ohana/src/libohana/src/ohana_allocate.c (revision 5237) +++ trunk/Ohana/src/libohana/src/ohana_allocate.c (revision 5273) @@ -235,4 +235,5 @@ S[3] = 'X'; Ns[0] = Ns[1] = Ns[2] = Ns[3] = 0; + N = 0; for (i = 0; i < Nmemlist; i++) {