Add WIN32 pipe implementation that uses sockets.

Claudio Natoli

configure

@@ -12227,7 +12227,8 @@ esac
 case $host_os in mingw*)
 LIBOBJS="$LIBOBJS dirmod.$ac_objext"
 LIBOBJS="$LIBOBJS copydir.$ac_objext"
-LIBOBJS="$LIBOBJS gettimeofday.$ac_objext" ;;
+LIBOBJS="$LIBOBJS gettimeofday.$ac_objext"
+LIBOBJS="$LIBOBJS pipe.$ac_objext" ;;
 esac
 
 if test "$with_readline" = yes; then

configure.in

@@ -1,5 +1,5 @@
 dnl Process this file with autoconf to produce a configure script.
-dnl $PostgreSQL: pgsql/configure.in,v 1.309 2003/12/23 18:40:52 tgl Exp $
+dnl $PostgreSQL: pgsql/configure.in,v 1.310 2004/01/09 04:58:09 momjian Exp $
 dnl
 dnl Developers, please strive to achieve this order:
 dnl
@@ -924,7 +924,8 @@ esac
 case $host_os in mingw*)
 AC_LIBOBJ(dirmod)
 AC_LIBOBJ(copydir)
-AC_LIBOBJ(gettimeofday) ;;
+AC_LIBOBJ(gettimeofday)
+AC_LIBOBJ(pipe) ;;
 esac
 
 if test "$with_readline" = yes; then

src/backend/postmaster/pgstat.c

@@ -13,7 +13,7 @@
  *
  *	Copyright (c) 2001-2003, PostgreSQL Global Development Group
  *
- *	$PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.51 2004/01/06 23:15:22 momjian Exp $
+ *	$PostgreSQL: pgsql/src/backend/postmaster/pgstat.c,v 1.52 2004/01/09 04:58:09 momjian Exp $
  * ----------
  */
 #include "postgres.h"
@@ -135,6 +135,19 @@ static void pgstat_recv_tabpurge(PgStat_MsgTabpurge *msg, int len);
 static void pgstat_recv_dropdb(PgStat_MsgDropdb *msg, int len);
 static void pgstat_recv_resetcounter(PgStat_MsgResetcounter *msg, int len);
 
+/*
+ *	WIN32 doesn't allow descriptors returned by pipe() to be used in select(),
+ *	so for that platform we use socket() instead of pipe().
+ */
+#ifndef WIN32
+#define pgpipe(a)			pipe(a)
+#define piperead(a,b,c)		read(a,b,c)
+#define pipewrite(a,b,c)	write(a,b,c)
+#else
+/* pgpipe() is in /src/port */
+#define piperead(a,b,c)		recv(a,b,c,0)
+#define pipewrite(a,b,c)	send(a,b,c,0)
+#endif
 
 /* ------------------------------------------------------------
  * Public functions called from postmaster follow
@@ -1380,7 +1393,7 @@ pgstat_main(PGSTAT_FORK_ARGS)
 	 * two buffer processes competing to read from the UDP socket --- not
 	 * good.
 	 */
-	if (pipe(pgStatPipe) < 0)
+	if (pgpipe(pgStatPipe) < 0)
 	{
 		ereport(LOG,
 				(errcode_for_socket_access(),
@@ -1595,7 +1608,7 @@ pgstat_mainChild(PGSTAT_FORK_ARGS)
 
 			while (nread < targetlen)
 			{
-				len = read(readPipe,
+				len = piperead(readPipe,
 						   ((char *) &msg) + nread,
 						   targetlen - nread);
 				if (len < 0)
@@ -1920,7 +1933,7 @@ pgstat_recvbuffer(void)
 			if (xfr > msg_have)
 				xfr = msg_have;
 			Assert(xfr > 0);
-			len = write(writePipe, msgbuffer + msg_send, xfr);
+			len = pipewrite(writePipe, msgbuffer + msg_send, xfr);
 			if (len < 0)
 			{
 				if (errno == EINTR || errno == EAGAIN)

src/port/pipe.c

@@ -0,0 +1,55 @@
+/*-------------------------------------------------------------------------
+ *
+ * pipe.c
+ *	  pipe()
+ *
+ * Copyright (c) 1996-2003, PostgreSQL Global Development Group
+ *
+ *	This is a replacement version of pipe for Win32 which allows
+ *	returned handles to be used in select(). Note that read/write calls
+ *	must be replaced with recv/send.
+ *
+ * IDENTIFICATION
+ *	  $PostgreSQL: pgsql/src/port/pipe.c,v 1.1 2004/01/09 04:58:09 momjian Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+int
+pgpipe(int handles[2])
+{
+	SOCKET		s;
+	struct sockaddr_in serv_addr;
+	int			len = sizeof(serv_addr);
+
+	handles[0] = handles[1] = INVALID_SOCKET;
+
+	if ((s = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
+		return -1;
+
+	memset((void *) &serv_addr, 0, sizeof(serv_addr));
+	serv_addr.sin_family = AF_INET;
+	serv_addr.sin_port = htons(0);
+	serv_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+	if (bind(s, (SOCKADDR *) & serv_addr, len) == SOCKET_ERROR ||
+		listen(s, 1) == SOCKET_ERROR ||
+		getsockname(s, (SOCKADDR *) & serv_addr, &len) == SOCKET_ERROR ||
+		(handles[1] = socket(PF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
+	{
+		closesocket(s);
+		return -1;
+	}
+
+	if (connect(handles[1], (SOCKADDR *) & serv_addr, len) == SOCKET_ERROR ||
+		(handles[0] = accept(s, (SOCKADDR *) & serv_addr, &len)) == INVALID_SOCKET)
+	{
+		closesocket(handles[1]);
+		handles[1] = INVALID_SOCKET;
+		closesocket(s);
+		return -1;
+	}
+	closesocket(s);
+	return 0;
+}