What I've done:

1. Rewritten libpq to allow asynchronous clients.

2. Implemented client side of cancel protocol in library,
   and patched psql.c to send a cancel request upon SIGINT.  The
   backend doesn't notice it yet :-(

3. Implemented 'Z' protocol message addition and renaming of
   copy in/out start messages.  These are implemented conditionally,
   ie, the client protocol version is checked; so the code should
   still work with 1.0 clients.

4. Revised protocol and libpq sgml documents (don't have an SGML
   compiler, though, so there may be some markup glitches here).


What remains to be done:

1. Implement addition of atttypmod field to RowDescriptor messages.
   The client-side code is there but ifdef'd out.  I have no idea
   what to change on the backend side.  The field should be sent
   only if protocol >= 2.0, of course.

2. Implement backend response to cancel requests received as OOB
   messages.  (This prolly need not be conditional on protocol
   version; just do it if you get SIGURG.)

3. Update libpq.3.  (I'm hoping this can be generated mechanically
   from libpq.sgml... if not, will do it by hand.)  Is there any
   other doco to fix?

4. Update non-libpq interfaces as necessary.  I patched libpgtcl
   so that it would compile, but haven't tested it.  Dunno what
   needs to be done with the other interfaces.

Have at it!

Tom Lane

1 files changed, 390 insertions, 71 deletions

diff --git a/src/interfaces/libpq/fe-misc.c b/src/interfaces/libpq/fe-misc.c
index 8a16950cdc..d7fc71dd13 100644
--- a/src/interfaces/libpq/fe-misc.c
+++ b/src/interfaces/libpq/fe-misc.c
@@ -5,177 +5,496 @@
  *
  *	 DESCRIPTION
  *		 miscellaneous useful functions
- *	 these routines are analogous to the ones in libpq/pqcomm.c
+ *
+ * The communication routines here are analogous to the ones in
+ * backend/libpq/pqcomm.c and backend/libpq/pqcomprim.c, but operate
+ * in the considerably different environment of the frontend libpq.
+ * In particular, we work with a bare nonblock-mode socket, rather than
+ * a stdio stream, so that we can avoid unwanted blocking of the application.
+ *
+ * XXX: MOVE DEBUG PRINTOUT TO HIGHER LEVEL.  As is, block and restart
+ * will cause repeat printouts.
+ *
+ * We must speak the same transmitted data representations as the backend
+ * routines.  Note that this module supports *only* network byte order
+ * for transmitted ints, whereas the backend modules (as of this writing)
+ * still handle either network or little-endian byte order.
  *
  * Copyright (c) 1994, Regents of the University of California
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/interfaces/libpq/fe-misc.c,v 1.10 1998/02/26 04:45:09 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/interfaces/libpq/fe-misc.c,v 1.11 1998/05/06 23:51:14 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
 
 #include <stdlib.h>
 #include <stdio.h>
+#include <string.h>
+#include <time.h>
+#if !defined(NO_UNISTD_H)
+#include <unistd.h>
+#endif
+#include <sys/types.h>			/* for fd_set stuff */
+#ifdef HAVE_SYS_SELECT_H
+#include <sys/select.h>
+#endif
 
 #include "postgres.h"
-
 #include "libpq-fe.h"
 
 /* --------------------------------------------------------------------- */
 /* pqGetc:
-   get a character from stream f
+   get a character from the connection
 
-   if debug is set, also echo the character fetched
+   All these routines return 0 on success, EOF on error.
+   Note that for the Get routines, EOF only means there is not enough
+   data in the buffer, not that there is necessarily a hard error.
 */
 int
-pqGetc(FILE *fin, FILE *debug)
+pqGetc(char *result, PGconn *conn)
 {
-	int			c;
+	if (conn->inCursor >= conn->inEnd)
+		return EOF;
 
-	c = getc(fin);
+	*result = conn->inBuffer[conn->inCursor++];
 
-	if (debug && c != EOF)
-		fprintf(debug, "From backend> %c\n", c);
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "From backend> %c\n", *result);
 
-	return c;
+	return 0;
 }
 
+
 /* --------------------------------------------------------------------- */
-/* pqPutnchar:
-   send a string of exactly len length into stream f
+/* pqPutBytes: local routine to write N bytes to the connection,
+   with buffering
+ */
+static int
+pqPutBytes(const char *s, int nbytes, PGconn *conn)
+{
+	int avail = conn->outBufSize - conn->outCount;
+
+	while (nbytes > avail)
+	{
+		memcpy(conn->outBuffer + conn->outCount, s, avail);
+		conn->outCount += avail;
+		s += avail;
+		nbytes -= avail;
+		if (pqFlush(conn))
+			return EOF;
+		avail = conn->outBufSize;
+	}
 
-   returns 1 if there was an error, 0 otherwise.
+	memcpy(conn->outBuffer + conn->outCount, s, nbytes);
+	conn->outCount += nbytes;
+
+	return 0;
+}
+
+/* --------------------------------------------------------------------- */
+/* pqGets:
+   get a null-terminated string from the connection,
+   and store it in a buffer of size maxlen bytes.
+   If the incoming string is >= maxlen bytes, all of it is read,
+   but the excess characters are silently discarded.
 */
 int
-pqPutnchar(const char *s, int len, FILE *f, FILE *debug)
+pqGets(char *s, int maxlen, PGconn *conn)
+{
+	/* Copy conn data to locals for faster search loop */
+	char	*inBuffer = conn->inBuffer;
+	int		inCursor = conn->inCursor;
+	int		inEnd = conn->inEnd;
+	int		slen;
+
+	while (inCursor < inEnd && inBuffer[inCursor])
+		inCursor++;
+
+	if (inCursor >= inEnd)
+		return EOF;
+
+	slen = inCursor - conn->inCursor;
+	if (slen < maxlen)
+		strcpy(s, inBuffer + conn->inCursor);
+	else
+	{
+		strncpy(s, inBuffer + conn->inCursor, maxlen-1);
+		s[maxlen-1] = '\0';
+	}
+
+	conn->inCursor = ++inCursor;
+
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "From backend> \"%s\"\n", s);
+
+	return 0;
+}
+
+/* --------------------------------------------------------------------- */
+int
+pqPuts(const char *s, PGconn *conn)
 {
-	if (debug)
-		fprintf(debug, "To backend> %s\n", s);
+	if (pqPutBytes(s, strlen(s)+1, conn))
+		return EOF;
+
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "To backend> %s\n", s);
 
-	return (pqPutNBytes(s, len, f) == EOF ? 1 : 0);
+	return 0;
 }
 
 /* --------------------------------------------------------------------- */
 /* pqGetnchar:
    get a string of exactly len bytes in buffer s (which must be 1 byte
-   longer) from stream f and terminate it with a '\0'.
+   longer) and terminate it with a '\0'.
 */
 int
-pqGetnchar(char *s, int len, FILE *f, FILE *debug)
+pqGetnchar(char *s, int len, PGconn *conn)
 {
-	int			status;
+	if (len < 0 || len > conn->inEnd - conn->inCursor)
+		return EOF;
+
+	memcpy(s, conn->inBuffer + conn->inCursor, len);
+	s[len] = '\0';
 
-	status = pqGetNBytes(s, len, f);
+	conn->inCursor += len;
 
-	if (debug)
-		fprintf(debug, "From backend (%d)> %s\n", len, s);
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "From backend (%d)> %s\n", len, s);
 
-	return (status == EOF ? 1 : 0);
+	return 0;
 }
 
 /* --------------------------------------------------------------------- */
-/* pqGets:
-   get a string of up to length len from stream f
+/* pqPutnchar:
+   send a string of exactly len bytes
+   The buffer should have a terminating null, but it's not sent.
 */
 int
-pqGets(char *s, int len, FILE *f, FILE *debug)
+pqPutnchar(const char *s, int len, PGconn *conn)
 {
-	int			status;
-
-	status = pqGetString(s, len, f);
+	if (pqPutBytes(s, len, conn))
+		return EOF;
 
-	if (debug)
-		fprintf(debug, "From backend> \"%s\"\n", s);
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "To backend> %s\n", s);
 
-	return (status == EOF ? 1 : 0);
+	return 0;
 }
 
 /* --------------------------------------------------------------------- */
-/* pgPutInt
-   send an integer of 2 or 4 bytes to the file stream, compensate
-   for host endianness.
-   returns 0 if successful, 1 otherwise
+/* pgGetInt
+   read a 2 or 4 byte integer and convert from network byte order
+   to local byte order
 */
 int
-pqPutInt(const int integer, int bytes, FILE *f, FILE *debug)
+pqGetInt(int *result, int bytes, PGconn *conn)
 {
-	int			retval = 0;
+	uint16 tmp2;
+	uint32 tmp4;
 
 	switch (bytes)
 	{
 		case 2:
-			retval = pqPutShort(integer, f);
+			if (conn->inCursor + 2 > conn->inEnd)
+				return EOF;
+			memcpy(&tmp2, conn->inBuffer + conn->inCursor, 2);
+			conn->inCursor += 2;
+			*result = (int) ntohs(tmp2);
 			break;
 		case 4:
-			retval = pqPutLong(integer, f);
+			if (conn->inCursor + 4 > conn->inEnd)
+				return EOF;
+			memcpy(&tmp4, conn->inBuffer + conn->inCursor, 4);
+			conn->inCursor += 4;
+			*result = (int) ntohl(tmp4);
 			break;
 		default:
 			fprintf(stderr, "** int size %d not supported\n", bytes);
-			retval = 1;
+			return EOF;
 	}
 
-	if (debug)
-		fprintf(debug, "To backend (%d#)> %d\n", bytes, integer);
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "From backend (#%d)> %d\n", bytes, *result);
 
-	return retval;
+	return 0;
 }
 
 /* --------------------------------------------------------------------- */
-/* pgGetInt
-   read a 2 or 4 byte integer from the stream and swab it around
-   to compensate for different endianness
-   returns 0 if successful
+/* pgPutInt
+   send an integer of 2 or 4 bytes, converting from host byte order
+   to network byte order.
 */
 int
-pqGetInt(int *result, int bytes, FILE *f, FILE *debug)
+pqPutInt(int value, int bytes, PGconn *conn)
 {
-	int			retval = 0;
+	uint16 tmp2;
+	uint32 tmp4;
 
 	switch (bytes)
 	{
 		case 2:
-			retval = pqGetShort(result, f);
+			tmp2 = htons((uint16) value);
+			if (pqPutBytes((const char*) &tmp2, 2, conn))
+				return EOF;
 			break;
 		case 4:
-			retval = pqGetLong(result, f);
+			tmp4 = htonl((uint32) value);
+			if (pqPutBytes((const char*) &tmp4, 4, conn))
+				return EOF;
 			break;
 		default:
 			fprintf(stderr, "** int size %d not supported\n", bytes);
-			retval = 1;
+			return EOF;
 	}
 
-	if (debug)
-		fprintf(debug, "From backend (#%d)> %d\n", bytes, *result);
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "To backend (%d#)> %d\n", bytes, value);
 
-	return retval;
+	return 0;
 }
 
 /* --------------------------------------------------------------------- */
+/* pqReadReady: is select() saying the file is ready to read?
+ */
+static int
+pqReadReady(PGconn *conn)
+{
+	fd_set			input_mask;
+	struct timeval	timeout;
+
+	if (conn->sock < 0)
+		return 0;
+
+	FD_ZERO(&input_mask);
+	FD_SET(conn->sock, &input_mask);
+	timeout.tv_sec = 0;
+	timeout.tv_usec = 0;
+	if (select(conn->sock+1, &input_mask, (fd_set *) NULL, (fd_set *) NULL,
+			   &timeout) < 0)
+	{
+		sprintf(conn->errorMessage,
+				"pqReadReady() -- select() failed: errno=%d\n%s\n",
+				errno, strerror(errno));
+		return 0;
+	}
+	return FD_ISSET(conn->sock, &input_mask);
+}
+
+/* --------------------------------------------------------------------- */
+/* pqReadData: read more data, if any is available
+ * Possible return values:
+ *   1: successfully loaded at least one more byte
+ *   0: no data is presently available, but no error detected
+ *  -1: error detected (including EOF = connection closure);
+ *      conn->errorMessage set
+ * NOTE: callers must not assume that pointers or indexes into conn->inBuffer
+ * remain valid across this call!
+ */
 int
-pqPuts(const char *s, FILE *f, FILE *debug)
+pqReadData(PGconn *conn)
 {
-	if (pqPutString(s, f) == EOF)
-		return 1;
+	int nread;
+
+	if (conn->sock < 0)
+	{
+		strcpy(conn->errorMessage, "pqReadData() -- connection not open\n");
+		return -1;
+	}
 
-	fflush(f);
+	/* Left-justify any data in the buffer to make room */
+	if (conn->inStart < conn->inEnd)
+	{
+		memmove(conn->inBuffer, conn->inBuffer + conn->inStart,
+				conn->inEnd - conn->inStart);
+		conn->inEnd -= conn->inStart;
+		conn->inCursor -= conn->inStart;
+		conn->inStart = 0;
+	}
+	else
+	{
+		conn->inStart = conn->inCursor = conn->inEnd = 0;
+	}
+	/* If the buffer is fairly full, enlarge it.
+	 * We need to be able to enlarge the buffer in case a single message
+	 * exceeds the initial buffer size.  We enlarge before filling the
+	 * buffer entirely so as to avoid asking the kernel for a partial packet.
+	 * The magic constant here should be at least one TCP packet.
+	 */
+	if (conn->inBufSize - conn->inEnd < 2000)
+	{
+		int newSize = conn->inBufSize * 2;
+		char * newBuf = (char *) realloc(conn->inBuffer, newSize);
+		if (newBuf)
+		{
+			conn->inBuffer = newBuf;
+			conn->inBufSize = newSize;
+		}
+	}
 
-	if (debug)
-		fprintf(debug, "To backend> %s\n", s);
+	/* OK, try to read some data */
+tryAgain:
+	nread = recv(conn->sock, conn->inBuffer + conn->inEnd, 
+				 conn->inBufSize - conn->inEnd, 0);
+	if (nread < 0)
+	{
+		if (errno == EINTR)
+			goto tryAgain;
+		sprintf(conn->errorMessage,
+				"pqReadData() --  read() failed: errno=%d\n%s\n",
+				errno, strerror(errno));
+		return -1;
+	}
+	if (nread > 0)
+	{
+		conn->inEnd += nread;
+		return 1;
+	}
 
-	return 0;
+	/* A return value of 0 could mean just that no data is now available,
+	 * or it could mean EOF --- that is, the server has closed the connection.
+	 * Since we have the socket in nonblock mode, the only way to tell the
+	 * difference is to see if select() is saying that the file is ready.
+	 * Grumble.  Fortunately, we don't expect this path to be taken much,
+	 * since in normal practice we should not be trying to read data unless
+	 * the file selected for reading already.
+	 */
+	if (! pqReadReady(conn))
+		return 0;				/* definitely no data available */
+
+	/* Still not sure that it's EOF,
+	 * because some data could have just arrived.
+	 */
+tryAgain2:
+	nread = recv(conn->sock, conn->inBuffer + conn->inEnd, 
+				 conn->inBufSize - conn->inEnd, 0);
+	if (nread < 0)
+	{
+		if (errno == EINTR)
+			goto tryAgain2;
+		sprintf(conn->errorMessage,
+				"pqReadData() --  read() failed: errno=%d\n%s\n",
+				errno, strerror(errno));
+		return -1;
+	}
+	if (nread > 0)
+	{
+		conn->inEnd += nread;
+		return 1;
+	}
+
+	/* OK, we are getting a zero read even though select() says ready.
+	 * This means the connection has been closed.  Cope.
+	 */
+	sprintf(conn->errorMessage,
+			"pqReadData() -- backend closed the channel unexpectedly.\n"
+			"\tThis probably means the backend terminated abnormally"
+			" before or while processing the request.\n");
+	conn->status = CONNECTION_BAD; /* No more connection to
+									* backend */
+	close(conn->sock);
+	conn->sock = -1;
+	
+	return -1;
 }
 
 /* --------------------------------------------------------------------- */
-void
-pqFlush(FILE *f, FILE *debug)
+/* pqFlush: send any data waiting in the output buffer
+ */
+int
+pqFlush(PGconn *conn)
 {
-	if (f)
-		fflush(f);
+	char * ptr = conn->outBuffer;
+	int len = conn->outCount;
+
+	if (conn->sock < 0)
+	{
+		strcpy(conn->errorMessage, "pqFlush() -- connection not open\n");
+		return EOF;
+	}
 
-	if (debug)
-		fflush(debug);
+	while (len > 0)
+	{
+		int sent = send(conn->sock, ptr, len, 0);
+		if (sent < 0)
+		{
+			/* Anything except EAGAIN or EWOULDBLOCK is trouble */
+			switch (errno)
+			{
+#ifdef EAGAIN
+				case EAGAIN:
+					break;
+#endif
+#if defined(EWOULDBLOCK) && (!defined(EAGAIN) || (EWOULDBLOCK != EAGAIN))
+				case EWOULDBLOCK:
+					break;
+#endif
+				default:
+					sprintf(conn->errorMessage,
+							"pqFlush() --  couldn't send data: errno=%d\n%s\n",
+							errno, strerror(errno));
+					return EOF;
+			}
+		}
+		else
+		{
+			ptr += sent;
+			len -= sent;
+		}
+		if (len > 0)
+		{
+			/* We didn't send it all, wait till we can send more */
+			if (pqWait(FALSE, TRUE, conn))
+				return EOF;
+		}
+	}
+
+	conn->outCount = 0;
+
+	if (conn->Pfdebug)
+		fflush(conn->Pfdebug);
+
+	return 0;
 }
 
 /* --------------------------------------------------------------------- */
+/* pqWait: wait until we can read or write the connection socket
+ */
+int
+pqWait(int forRead, int forWrite, PGconn *conn)
+{
+	fd_set	input_mask;
+	fd_set	output_mask;
+
+	if (conn->sock < 0)
+	{
+		strcpy(conn->errorMessage, "pqWait() -- connection not open\n");
+		return EOF;
+	}
+
+	/* loop in case select returns EINTR */
+	for (;;) {
+		FD_ZERO(&input_mask);
+		FD_ZERO(&output_mask);
+		if (forRead)
+			FD_SET(conn->sock, &input_mask);
+		if (forWrite)
+			FD_SET(conn->sock, &output_mask);
+		if (select(conn->sock+1, &input_mask, &output_mask, (fd_set *) NULL,
+				   (struct timeval *) NULL) < 0)
+		{
+			if (errno == EINTR)
+				continue;
+			sprintf(conn->errorMessage,
+					"pqWait() -- select() failed: errno=%d\n%s\n",
+					errno, strerror(errno));
+			return EOF;
+		}
+		/* On nonerror return, assume we're done */
+		break;
+	}
+
+	return 0;
+}