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

src/backend/commands/async.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.31 1998/04/27 04:05:08 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.32 1998/05/06 23:49:52 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -20,8 +20,7 @@
  *	  end of commit),
  *	  2.a  If the process is the same as the backend process that issued
  *		   notification (we are notifying something that we are listening),
- *		   signal the corresponding frontend over the comm channel using the
- *		   out-of-band channel.
+ *		   signal the corresponding frontend over the comm channel.
  *	  2.b  For all other listening processes, we send kill(2) to wake up
  *		   the listening backend.
  * 3. Upon receiving a kill(2) signal from another backend process notifying
@@ -30,7 +29,7 @@
  *	  3.a  We are sleeping, wake up and signal our frontend.
  *	  3.b  We are in middle of another transaction, wait until the end of
  *		   of the current transaction and signal our frontend.
- * 4. Each frontend receives this notification and prcesses accordingly.
+ * 4. Each frontend receives this notification and processes accordingly.
  *
  * -- jw, 12/28/93
  *
@@ -547,12 +546,6 @@ Async_UnlistenOnExit(int code,	/* from exitpg */
  * Results:
  *		XXX
  *
- * Side effects:
- *
- *		We make use of the out-of-band channel to transmit the
- *		notification to the front end.	The actual data transfer takes
- *		place at the front end's request.
- *
  * --------------------------------------------------------------
  */
 GlobalMemory notifyContext = NULL;

src/backend/tcop/dest.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/tcop/dest.c,v 1.17 1998/02/26 04:36:24 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/tcop/dest.c,v 1.18 1998/05/06 23:49:59 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -15,7 +15,8 @@
  *	 INTERFACE ROUTINES
  *		BeginCommand - prepare destination for tuples of the given type
  *		EndCommand - tell destination that no more tuples will arrive
- *		NullCommand - tell dest that the last of a query sequence was processed
+ *		NullCommand - tell dest that an empty query string was recognized
+ *		ReadyForQuery - tell dest that we are ready for a new query
  *
  *	 NOTES
  *		These routines do the appropriate work before and after
@@ -115,16 +116,10 @@ EndCommand(char *commandTag, CommandDest dest)
 			sprintf(buf, "%s%s", commandTag, CommandInfo);
 			CommandInfo[0] = 0;
 			pq_putstr(buf);
-			pq_flush();
 			break;
 
 		case Local:
 		case Debug:
-			break;
-		case CopyEnd:
-			pq_putnchar("Z", 1);
-			pq_flush();
-			break;
 		case None:
 		default:
 			break;
@@ -139,28 +134,37 @@ EndCommand(char *commandTag, CommandDest dest)
  *
  * COPY rel FROM stdin
  *
+ * NOTE: the message code letters are changed at protocol version 2.0
+ * to eliminate possible confusion with data tuple messages.
  */
 void
 SendCopyBegin(void)
 {
-	pq_putnchar("B", 1);
-/*	  pq_putint(0, 4); */
-	pq_flush();
+	if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2)
+		pq_putnchar("H", 1);	/* new way */
+	else
+		pq_putnchar("B", 1);	/* old way */
 }
 
 void
 ReceiveCopyBegin(void)
 {
-	pq_putnchar("D", 1);
-/*	  pq_putint(0, 4); */
+	if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2)
+		pq_putnchar("G", 1);	/* new way */
+	else
+		pq_putnchar("D", 1);	/* old way */
+	/* We *must* flush here to ensure FE knows it can send. */
 	pq_flush();
 }
 
 /* ----------------
- *		NullCommand - tell dest that the last of a query sequence was processed
+ *		NullCommand - tell dest that an empty query string was recognized
  *
- *		Necessary to implement the hacky FE/BE interface to handle
- *		multiple-return queries.
+ *		In FE/BE protocol version 1.0, this hack is necessary to support
+ *		libpq's crufty way of determining whether a multiple-command
+ *		query string is done.  In protocol 2.0 it's probably not really
+ *		necessary to distinguish empty queries anymore, but we still do it
+ *		for backwards compatibility with 1.0.
  * ----------------
  */
 void
@@ -168,46 +172,46 @@ NullCommand(CommandDest dest)
 {
 	switch (dest)
 	{
-			case RemoteInternal:
-			case Remote:
+		case RemoteInternal:
+		case Remote:
 			{
-#if 0
-
-				/*
-				 * Do any asynchronous notification.  If front end wants
-				 * to poll, it can send null queries to call this
-				 * function.
-				 */
-				PQNotifyList *nPtr;
-				MemoryContext orig;
-
-				if (notifyContext == NULL)
-				{
-					notifyContext = CreateGlobalMemory("notify");
-				}
-				orig = MemoryContextSwitchTo((MemoryContext) notifyContext);
-
-				for (nPtr = PQnotifies();
-					 nPtr != NULL;
-					 nPtr = (PQNotifyList *) SLGetSucc(&nPtr->Node))
-				{
-					pq_putnchar("A", 1);
-					pq_putint(0, sizeof(int4));
-					pq_putstr(nPtr->relname);
-					pq_putint(nPtr->be_pid, sizeof(nPtr->be_pid));
-					PQremoveNotify(nPtr);
-				}
-				pq_flush();
-				PQcleanNotify();/* garbage collect */
-				MemoryContextSwitchTo(orig);
-#endif
 				/* ----------------
-				 *		tell the fe that the last of the queries has finished
+				 *		tell the fe that we saw an empty query string
 				 * ----------------
 				 */
-/*		pq_putnchar("I", 1);  */
 				pq_putstr("I");
-				/* pq_putint(0, 4); */
+			}
+			break;
+
+		case Local:
+		case Debug:
+		case None:
+		default:
+			break;
+	}
+}
+
+/* ----------------
+ *		ReadyForQuery - tell dest that we are ready for a new query
+ *
+ *		The ReadyForQuery message is sent in protocol versions 2.0 and up
+ *		so that the FE can tell when we are done processing a query string.
+ *
+ *		Note that by flushing the stdio buffer here, we can avoid doing it
+ *		most other places and thus reduce the number of separate packets sent.
+ * ----------------
+ */
+void
+ReadyForQuery(CommandDest dest)
+{
+	switch (dest)
+	{
+		case RemoteInternal:
+		case Remote:
+			{
+				if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2)
+					pq_putnchar("Z", 1);
+				/* Flush output at end of cycle in any case. */
 				pq_flush();
 			}
 			break;
@@ -264,7 +268,6 @@ BeginCommand(char *pname,
 			 *		send fe info on tuples we're about to send
 			 * ----------------
 			 */
-			pq_flush();
 			pq_putnchar("P", 1);/* new portal.. */
 			pq_putstr(pname);	/* portal name */
 

src/backend/tcop/fastpath.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/tcop/fastpath.c,v 1.16 1998/04/26 04:07:22 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/tcop/fastpath.c,v 1.17 1998/05/06 23:50:10 momjian Exp $
  *
  * NOTES
  *	  This cruft is the server side of PQfn.
@@ -113,7 +113,6 @@ SendFunctionResult(Oid fid,		/* function id */
 	}
 
 	pq_putnchar("0", 1);
-	pq_flush();
 }
 
 /*

src/backend/tcop/postgres.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/tcop/postgres.c,v 1.67 1998/02/26 04:36:31 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/tcop/postgres.c,v 1.68 1998/05/06 23:50:19 momjian Exp $
  *
  * NOTES
  *	  this is the "main" module of the postgres backend and
@@ -1302,7 +1302,7 @@ PostgresMain(int argc, char *argv[])
 	if (IsUnderPostmaster == false)
 	{
 		puts("\nPOSTGRES backend interactive interface");
-		puts("$Revision: 1.67 $ $Date: 1998/02/26 04:36:31 $");
+		puts("$Revision: 1.68 $ $Date: 1998/05/06 23:50:19 $");
 	}
 
 	/* ----------------
@@ -1316,6 +1316,12 @@ PostgresMain(int argc, char *argv[])
 
 	for (;;)
 	{
+		/* ----------------
+		 *	 (0) tell the frontend we're ready for a new query.
+		 * ----------------
+		 */
+		ReadyForQuery(Remote);
+
 		/* ----------------
 		 *	 (1) read a command.
 		 * ----------------
@@ -1391,8 +1397,8 @@ PostgresMain(int argc, char *argv[])
 				 * ----------------
 				 */
 			case 'X':
-				IsEmptyQuery = true;
 				pq_close();
+				exitpg(0);
 				break;
 
 			default:

src/bin/psql/psql.c

@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/bin/psql/Attic/psql.c,v 1.139 1998/05/04 02:02:01 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/bin/psql/Attic/psql.c,v 1.140 1998/05/06 23:50:23 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -283,6 +283,38 @@ PSQLexec(PsqlSettings *pset, char *query)
 	return NULL;
 }
 
+/*
+ * Code to support command cancellation.
+ * If interactive, we enable a SIGINT signal catcher that sends
+ * a cancel request to the backend.
+ * Note that sending the cancel directly from the signal handler
+ * is safe only because the cancel is sent as an OOB message.
+ * If it were inline data, then we'd risk inserting it into the
+ * middle of a normal data message by doing this.
+ * (It's probably not too cool to write on stderr, for that matter...
+ *  but for debugging purposes we'll risk that.)
+ */
+
+static PGconn * cancelConn = NULL; /* connection to try cancel on */
+
+static void
+handle_sigint (SIGNAL_ARGS)
+{
+	if (cancelConn == NULL)
+		exit(1);				/* accept signal if no connection */
+	/* Try to send cancel request */
+	if (PQrequestCancel(cancelConn))
+	{
+		fprintf(stderr, "\nCANCEL request sent\n");
+	}
+	else
+	{
+		fprintf(stderr, "\nCannot send cancel request:\n%s\n",
+				PQerrorMessage(cancelConn));
+	}
+}
+
+
 /*
  * listAllDbs
  *
@@ -1099,8 +1131,7 @@ SendQuery(bool *success_p, PsqlSettings *pset, const char *query,
 			exit(2);			/* we are out'ta here */
 		}
 		/* check for asynchronous returns */
-		notify = PQnotifies(pset->db);
-		if (notify)
+		while ((notify = PQnotifies(pset->db)) != NULL)
 		{
 			fprintf(stderr,
 				 "ASYNC NOTIFY of '%s' from backend pid '%d' received\n",
@@ -1416,6 +1447,7 @@ do_connect(const char *new_dbname,
 		}
 		else
 		{
+			cancelConn = pset->db; /* redirect sigint's loving attentions */
 			PQfinish(olddb);
 			free(pset->prompt);
 			pset->prompt = malloc(strlen(PQdb(pset->db)) + 10);
@@ -2462,11 +2494,18 @@ main(int argc, char **argv)
 	settings.opt.fieldSep = strdup(DEFAULT_FIELD_SEP);
 	settings.opt.pager = 1;
 	if (!isatty(0) || !isatty(1))
+	{
+		/* Noninteractive defaults */
 		settings.notty = 1;
-#ifdef USE_READLINE
+	}
 	else
+	{
+		/* Interactive defaults */
+		pqsignal(SIGINT, handle_sigint);	/* control-C => cancel */
+#ifdef USE_READLINE
 		settings.useReadline = 1;
 #endif
+	}
 #ifdef PSQL_ALWAYS_GET_PASSWORDS
 	settings.getPassword = 1;
 #else
@@ -2580,6 +2619,9 @@ main(int argc, char **argv)
 		PQfinish(settings.db);
 		exit(1);
 	}
+
+	cancelConn = settings.db;	/* enable SIGINT to send cancel */
+
 	if (listDatabases)
 	{
 		exit(listAllDbs(&settings));

src/include/libpq/pqcomm.h

@@ -6,7 +6,7 @@
  *
  * Copyright (c) 1994, Regents of the University of California
  *
- * $Id: pqcomm.h,v 1.24 1998/03/02 05:42:15 scrappy Exp $
+ * $Id: pqcomm.h,v 1.25 1998/05/06 23:50:32 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -66,7 +66,7 @@ typedef union SockAddr
 /* The earliest and latest frontend/backend protocol version supported. */
 
 #define PG_PROTOCOL_EARLIEST	PG_PROTOCOL(0,0)
-#define PG_PROTOCOL_LATEST	PG_PROTOCOL(1,0)
+#define PG_PROTOCOL_LATEST	PG_PROTOCOL(2,0)
 
 /*
  * All packets sent to the postmaster start with the length.  This is omitted

src/include/tcop/dest.h

@@ -26,7 +26,7 @@
  *
  * Copyright (c) 1994, Regents of the University of California
  *
- * $Id: dest.h,v 1.13 1998/02/26 04:43:39 momjian Exp $
+ * $Id: dest.h,v 1.14 1998/05/06 23:50:49 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -46,10 +46,6 @@ typedef enum
 	Debug,						/* results go to debugging output */
 	Local,						/* results go in local portal buffer */
 	Remote,						/* results sent to frontend process */
-	CopyBegin,					/* results sent to frontend process but
-								 * are strings */
-	CopyEnd,					/* results sent to frontend process but
-								 * are strings */
 	RemoteInternal,				/* results sent to frontend process in
 								 * internal (binary) form */
 	SPI							/* results sent to SPI manager */
@@ -70,6 +66,7 @@ extern void EndCommand(char *commandTag, CommandDest dest);
 extern void SendCopyBegin(void);
 extern void ReceiveCopyBegin(void);
 extern void NullCommand(CommandDest dest);
+extern void ReadyForQuery(CommandDest dest);
 extern void
 BeginCommand(char *pname, int operation, TupleDesc attinfo,
 			 bool isIntoRel, bool isIntoPortal, char *tag,

src/interfaces/libpgtcl/pgtclId.c

@@ -12,7 +12,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/interfaces/libpgtcl/Attic/pgtclId.c,v 1.8 1998/03/15 08:03:00 scrappy Exp $
+ *	  $Header: /cvsroot/pgsql/src/interfaces/libpgtcl/Attic/pgtclId.c,v 1.9 1998/05/06 23:51:00 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -48,7 +48,7 @@ int PgInputProc(DRIVER_INPUT_PROTO)
 {
     Pg_ConnectionId	*connid;
     PGconn		*conn;
-    int			c;
+    char		c;
     int			avail;
 
     connid = (Pg_ConnectionId *)cData;
@@ -64,13 +64,24 @@ int PgInputProc(DRIVER_INPUT_PROTO)
 	return PgEndCopy(connid, errorCodePtr);
     }
 
+    /* Try to load any newly arrived data */
+    errno = 0;
+
+    if (pqReadData(conn) < 0) {
+      *errorCodePtr = errno ? errno : EIO;
+      return -1;
+    }
+
+    /* Move data from libpq's buffer to tcl's */
+
+    conn->inCursor = conn->inStart;
+
     avail = bufSize;
     while (avail > 0 &&
-	    (c = pqGetc(conn->Pfin, conn->Pfdebug)) != EOF) {
-	/* fprintf(stderr, "%d: got char %c\n", bufSize-avail, c); */
+	   pqGetc(&c, conn) == 0) {
 	*buf++ = c;
 	--avail;
-	if (c == '\n' && bufSize-avail > 3) {
+	if (c == '\n' && bufSize-avail >= 3) {
 	    if ((bufSize-avail == 3 || buf[-4] == '\n') &&
 		    buf[-3] == '\\' && buf[-2] == '.') {
 		avail += 3;
@@ -79,6 +90,8 @@ int PgInputProc(DRIVER_INPUT_PROTO)
 	    }
 	}
     }
+    /* Accept the data permanently */
+    conn->inStart = conn->inCursor;
     /* fprintf(stderr, "returning %d chars\n", bufSize - avail); */
     return bufSize - avail;
 }
@@ -100,16 +113,15 @@ int PgOutputProc(DRIVER_OUTPUT_PROTO)
 	return -1;
     }
 
-    /*
-    fprintf(stderr, "PgOutputProc called: bufSize=%d: atend:%d <", bufSize,
-	strncmp(buf, "\\.\n", 3));
-    fwrite(buf, 1, bufSize, stderr);
-    fputs(">\n", stderr);
-    */
-    fwrite(buf, 1, bufSize, conn->Pfout);
-    if (bufSize > 2 && strncmp(&buf[bufSize-3], "\\.\n", 3) == 0) {
-	/* fprintf(stderr,"checking closure\n"); */
-	fflush(conn->Pfout);
+    errno = 0;
+
+    if (pqPutnchar(buf, bufSize, conn)) {
+      *errorCodePtr = errno ? errno : EIO;
+      return -1;
+    }
+
+    if (bufSize >= 3 && strncmp(&buf[bufSize-3], "\\.\n", 3) == 0) {
+	(void) pqFlush(conn);
 	if (PgEndCopy(connid, errorCodePtr) == -1)
 	    return -1;
     }
@@ -156,10 +168,10 @@ PgSetConnectionId(Tcl_Interp *interp, PGconn *conn)
     for (i = 0; i < RES_START; i++) connid->results[i] = NULL;
     Tcl_InitHashTable(&connid->notify_hash, TCL_STRING_KEYS);
 
-    sprintf(connid->id, "pgsql%d", fileno(conn->Pfout));
+    sprintf(connid->id, "pgsql%d", PQsocket(conn));
 
 #if TCL_MAJOR_VERSION == 7 && TCL_MINOR_VERSION == 5
-    conn_chan = Tcl_CreateChannel(&Pg_ConnType, connid->id, conn->Pfin, conn->Pfout, (ClientData)connid);
+    conn_chan = Tcl_CreateChannel(&Pg_ConnType, connid->id, NULL, NULL, (ClientData)connid);
 #else
     conn_chan = Tcl_CreateChannel(&Pg_ConnType, connid->id, (ClientData)connid,
 	TCL_READABLE | TCL_WRITABLE);

src/interfaces/libpq/Makefile.in

@@ -7,7 +7,7 @@
 #
 #
 # IDENTIFICATION
-#    $Header: /cvsroot/pgsql/src/interfaces/libpq/Attic/Makefile.in,v 1.16 1998/04/27 14:55:46 scrappy Exp $
+#    $Header: /cvsroot/pgsql/src/interfaces/libpq/Attic/Makefile.in,v 1.17 1998/05/06 23:51:06 momjian Exp $
 #
 #-------------------------------------------------------------------------
 
@@ -25,14 +25,13 @@ ifdef KRBVERS
 CFLAGS+= $(KRBFLAGS)
 endif
 
-OBJS= fe-auth.o fe-connect.o fe-exec.o fe-misc.o fe-lobj.o \
-      dllist.o pqsignal.o pqcomprim.o
+OBJS= fe-auth.o fe-connect.o fe-exec.o fe-misc.o fe-print.o fe-lobj.o \
+      dllist.o pqsignal.o
 
 # Shared library stuff
 shlib := 
 install-shlib-dep :=
 ifeq ($(PORTNAME), linux)
-  LINUX_ELF=@LINUX_ELF@
   ifdef LINUX_ELF
     install-shlib-dep := install-shlib
     shlib := libpq.so.$(SO_MAJOR_VERSION).$(SO_MINOR_VERSION)
@@ -84,9 +83,6 @@ fe-lobj.o: $(SRCDIR)/backend/fmgr.h
 dllist.c:	$(SRCDIR)/backend/lib/dllist.c
 	-ln -s $(SRCDIR)/backend/lib/dllist.c .
 
-pqcomprim.c:	    $(SRCDIR)/backend/libpq/pqcomprim.c
-	-ln -s $(SRCDIR)/backend/libpq/pqcomprim.c .
-
 # The following rules cause dependencies in the backend directory to 
 # get made if they don't exist, but don't cause them to get remade if they
 # are out of date.
@@ -183,7 +179,7 @@ depend dep:
 
 .PHONY: clean
 clean:
-	rm -f libpq.a $(shlib) $(OBJS) c.h dllist.c pqcomprim.c libpq.so
+	rm -f libpq.a $(shlib) $(OBJS) c.h dllist.c libpq.so
 
 ifeq (depend,$(wildcard depend))
 include depend