Here's a patch against 7.1.3 that fixes a problem with sending larger

queries over non-blocking connections with libpq. "Larger" here
basically means that it doesn't fit into the output buffer.

The basic strategy is to fix pqFlush and pqPutBytes.

The problem with pqFlush as it stands now is that it returns EOF when an
error occurs or when not all data could be sent. The latter case is
clearly not an error for a non-blocking connection but the caller can't
distringuish it from an error very well.

The first part of the fix is therefore to fix pqFlush. This is done by
to renaming it to pqSendSome which only differs from pqFlush in its
return values to allow the caller to make the above distinction and a
new pqFlush which is implemented in terms of pqSendSome and behaves
exactly like the old pqFlush.

The second part of the fix modifies pqPutBytes to use pqSendSome instead
of pqFlush and to either send all the data or if not all data can be
sent on a non-blocking connection to at least put all data into the
output buffer, enlarging it if necessary. The callers of pqPutBytes
don't have to be changed because from their point of view pqPutBytes
behaves like before. It either succeeds in queueing all output data or
fails with an error.

I've also added a new API function PQsendSome which analogously to
PQflush just calls pqSendSome. Programs using non-blocking queries
should use this new function. The main difference is that this function
will have to be called repeatedly (calling select() properly in between)
until all data has been written.

AFAICT, the code in CVS HEAD hasn't changed with respect to non-blocking
queries and this fix should work there, too, but I haven't tested that
yet.

Bernhard Herzog

src/interfaces/libpq/fe-exec.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/interfaces/libpq/fe-exec.c,v 1.114 2002/03/04 23:59:14 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/interfaces/libpq/fe-exec.c,v 1.115 2002/03/05 05:20:12 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2429,3 +2429,11 @@ PQflush(PGconn *conn)
 
 	return (pqFlush(conn));
 }
+
+/* try to force data out, really only useful for non-blocking users.
+ * This implementation actually works for non-blocking connections */
+int
+PQsendSome(PGconn *conn)
+{
+	return pqSendSome(conn);
+}

src/interfaces/libpq/fe-misc.c

@@ -25,7 +25,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/interfaces/libpq/fe-misc.c,v 1.65 2001/12/03 00:28:24 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/interfaces/libpq/fe-misc.c,v 1.66 2002/03/05 05:20:12 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -110,55 +110,83 @@ pqPutc(char c, PGconn *conn)
 static int
 pqPutBytes(const char *s, size_t nbytes, PGconn *conn)
 {
-	size_t		avail = Max(conn->outBufSize - conn->outCount, 0);
-
 	/*
-	 * if we are non-blocking and the send queue is too full to buffer
-	 * this request then try to flush some and return an error
+	 * Strategy to handle blocking and non-blocking connections: Fill the
+	 * output buffer and flush it repeatedly until either all data has
+	 * been sent or is at least queued in the buffer.
+	 *
+	 * For non-blocking connections, grow the buffer if not all data fits
+	 * into it and the buffer can't be sent because the socket would
+	 * block.
 	 */
-	if (pqIsnonblocking(conn) && nbytes > avail && pqFlush(conn))
+
+	while (nbytes)
 	{
+		size_t		avail,
+					remaining;
+
+		/* fill the output buffer */
+		avail = Max(conn->outBufSize - conn->outCount, 0);
+		remaining = Min(avail, nbytes);
+		memcpy(conn->outBuffer + conn->outCount, s, remaining);
+		conn->outCount += remaining;
+		s += remaining;
+		nbytes -= remaining;
+
 		/*
-		 * even if the flush failed we may still have written some data,
-		 * recalculate the size of the send-queue relative to the amount
-		 * we have to send, we may be able to queue it afterall even
-		 * though it's not sent to the database it's ok, any routines that
-		 * check the data coming from the database better call pqFlush()
-		 * anyway.
+		 * if the data didn't fit completely into the buffer, try to flush
+		 * the buffer
 		 */
-		if (nbytes > Max(conn->outBufSize - conn->outCount, 0))
+		if (nbytes)
 		{
-			printfPQExpBuffer(&conn->errorMessage,
-							  libpq_gettext("could not flush enough data (space available: %d, space needed %d)\n"),
-						 (int) Max(conn->outBufSize - conn->outCount, 0),
-							  (int) nbytes);
+			int			send_result = pqSendSome(conn);
+
+			/* if there were errors, report them */
+			if (send_result < 0)
 				return EOF;
-		}
-		/* fixup avail for while loop */
-		avail = Max(conn->outBufSize - conn->outCount, 0);
-	}
 
 			/*
-	 * is the amount of data to be sent is larger than the size of the
-	 * output buffer then we must flush it to make more room.
-	 *
-	 * the code above will make sure the loop conditional is never true for
-	 * non-blocking connections
+			 * if not all data could be sent, increase the output buffer,
+			 * put the rest of s into it and return successfully. This
+			 * case will only happen in a non-blocking connection
 			 */
-	while (nbytes > avail)
+			if (send_result > 0)
 			{
-		memcpy(conn->outBuffer + conn->outCount, s, avail);
-		conn->outCount += avail;
-		s += avail;
-		nbytes -= avail;
-		if (pqFlush(conn))
+				/*
+				 * try to grow the buffer. FIXME: The new size could be
+				 * chosen more intelligently.
+				 */
+				size_t		buflen = conn->outCount + nbytes;
+
+				if (buflen > conn->outBufSize)
+				{
+					char	   *newbuf = realloc(conn->outBuffer, buflen);
+
+					if (!newbuf)
+					{
+						/* realloc failed. Probably out of memory */
+						printfPQExpBuffer(&conn->errorMessage,
+						   "cannot allocate memory for output buffer\n");
 						return EOF;
-		avail = conn->outBufSize;
 					}
-
+					conn->outBuffer = newbuf;
+					conn->outBufSize = buflen;
+				}
+				/* put the data into it */
 				memcpy(conn->outBuffer + conn->outCount, s, nbytes);
 				conn->outCount += nbytes;
 
+				/* report success. */
+				return 0;
+			}
+		}
+
+		/*
+		 * pqSendSome was able to send all data. Continue with the next
+		 * chunk of s.
+		 */
+	}							/* while */
+
 	return 0;
 }
 
@@ -603,11 +631,22 @@ definitelyFailed:
 	return -1;
 }
 
-/*
- * pqFlush: send any data waiting in the output buffer
+/* pqSendSome: send any data waiting in the output buffer and return 0
+ * if all data was sent, -1 if an error occurred or 1 if not all data
+ * could be written because the socket would have blocked.
+ *
+ * For a blocking connection all data will be sent unless an error
+ * occurrs. -1 will only be returned if the connection is non-blocking.
+ *
+ * Internally, the case of data remaining in the buffer after pqSendSome
+ * could be determined by looking at outCount, but this function also
+ * serves as the implementation of the new API function PQsendsome.
+ *
+ * FIXME: perhaps it would be more useful to return the number of bytes
+ * remaining?
  */
 int
-pqFlush(PGconn *conn)
+pqSendSome(PGconn *conn)
 {
 	char	   *ptr = conn->outBuffer;
 	int			len = conn->outCount;
@@ -685,14 +724,14 @@ pqFlush(PGconn *conn)
 					 * the socket open until pqReadData finds no more data
 					 * can be read.
 					 */
-					return EOF;
+					return -1;
 
 				default:
 					printfPQExpBuffer(&conn->errorMessage,
 					libpq_gettext("could not send data to server: %s\n"),
 									  SOCK_STRERROR(SOCK_ERRNO));
 					/* We don't assume it's a fatal error... */
-					return EOF;
+					return -1;
 			}
 		}
 		else
@@ -707,7 +746,7 @@ pqFlush(PGconn *conn)
 
 			/*
 			 * if the socket is in non-blocking mode we may need to abort
-			 * here
+			 * here and return 1 to indicate that data is still pending.
 			 */
 #ifdef USE_SSL
 			/* can't do anything for our SSL users yet */
@@ -719,14 +758,14 @@ pqFlush(PGconn *conn)
 					/* shift the contents of the buffer */
 					memmove(conn->outBuffer, ptr, len);
 					conn->outCount = len;
-					return EOF;
+					return 1;
 				}
 #ifdef USE_SSL
 			}
 #endif
 
 			if (pqWait(FALSE, TRUE, conn))
-				return EOF;
+				return -1;
 		}
 	}
 
@@ -738,6 +777,23 @@ pqFlush(PGconn *conn)
 	return 0;
 }
 
+
+/*
+ * pqFlush: send any data waiting in the output buffer
+ *
+ * Implemented in terms of pqSendSome to recreate the old behavior which
+ * returned 0 if all data was sent or EOF. EOF was sent regardless of
+ * whether an error occurred or not all data was sent on a non-blocking
+ * socket.
+ */
+int
+pqFlush(PGconn *conn)
+{
+	if (pqSendSome(conn))
+		return EOF;
+	return 0;
+}
+
 /*
  * pqWait: wait until we can read or write the connection socket
  *
@@ -756,7 +812,7 @@ pqWait(int forRead, int forWrite, PGconn *conn)
 	{
 		printfPQExpBuffer(&conn->errorMessage,
 						  libpq_gettext("connection not open\n"));
-		return EOF;
+		return -1;
 	}
 
 	if (forRead || forWrite)
@@ -863,6 +919,7 @@ const char *
 winsock_strerror(int eno)
 {
 	static char err_buf[512];
+
 #define WSSE_MAXLEN (sizeof(err_buf)-1-13)		/* 13 for " (0x00000000)" */
 	int			length;
 

src/interfaces/libpq/libpq-fe.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: libpq-fe.h,v 1.81 2002/03/04 23:59:14 momjian Exp $
+ * $Id: libpq-fe.h,v 1.82 2002/03/05 05:20:12 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -282,6 +282,11 @@ extern int	PQisnonblocking(const PGconn *conn);
 
 /* Force the write buffer to be written (or at least try) */
 extern int	PQflush(PGconn *conn);
+/*
+ * Force the write buffer to be written (or at least try)
+ * (better than PQflush)
+ */
+extern int	PQsendSome(PGconn *conn);
 
 /*
  * "Fast path" interface --- not really recommended for application

src/interfaces/libpq/libpq-int.h

@@ -12,7 +12,7 @@
  * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: libpq-int.h,v 1.44 2001/11/05 17:46:38 momjian Exp $
+ * $Id: libpq-int.h,v 1.45 2002/03/05 05:20:12 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -323,6 +323,7 @@ extern int	pqGetInt(int *result, size_t bytes, PGconn *conn);
 extern int	pqPutInt(int value, size_t bytes, PGconn *conn);
 extern int	pqReadData(PGconn *conn);
 extern int	pqFlush(PGconn *conn);
+extern int	pqSendSome(PGconn *conn);
 extern int	pqWait(int forRead, int forWrite, PGconn *conn);
 extern int	pqReadReady(PGconn *conn);
 extern int	pqWriteReady(PGconn *conn);