0001-Export-vacuumdb-s-parallel-infrastructure-v3.patch

From db00f7c7546724ef28cdaa8aafedec8491be6040 Mon Sep 17 00:00:00 2001
From: Julien Rouhaud <julien.rouhaud@free.fr>
Date: Fri, 28 Jun 2019 13:11:29 +0200
Subject: [PATCH 1/2] Export vacuumdb's parallel infrastructure

---
 src/bin/scripts/Makefile           |   4 +-
 src/bin/scripts/common.c           |  78 +++++++
 src/bin/scripts/common.h           |   7 +
 src/bin/scripts/scripts_parallel.c | 279 +++++++++++++++++++++++
 src/bin/scripts/scripts_parallel.h |  32 +++
 src/bin/scripts/vacuumdb.c         | 349 ++---------------------------
 6 files changed, 417 insertions(+), 332 deletions(-)
 create mode 100644 src/bin/scripts/scripts_parallel.c
 create mode 100644 src/bin/scripts/scripts_parallel.h

diff --git a/src/bin/scripts/Makefile b/src/bin/scripts/Makefile
index 9f352b5e2b..3cd793b134 100644
--- a/src/bin/scripts/Makefile
+++ b/src/bin/scripts/Makefile
@@ -28,7 +28,7 @@ createuser: createuser.o common.o $(WIN32RES) | submake-libpq submake-libpgport
 dropdb: dropdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 dropuser: dropuser.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 clusterdb: clusterdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
-vacuumdb: vacuumdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
+vacuumdb: vacuumdb.o common.o scripts_parallel.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 reindexdb: reindexdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 pg_isready: pg_isready.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 
@@ -50,7 +50,7 @@ uninstall:
 
 clean distclean maintainer-clean:
 	rm -f $(addsuffix $(X), $(PROGRAMS)) $(addsuffix .o, $(PROGRAMS))
-	rm -f common.o $(WIN32RES)
+	rm -f common.o scripts_parallel.o $(WIN32RES)
 	rm -rf tmp_check
 
 check:
diff --git a/src/bin/scripts/common.c b/src/bin/scripts/common.c
index 296029d809..7250b57bc4 100644
--- a/src/bin/scripts/common.c
+++ b/src/bin/scripts/common.c
@@ -22,6 +22,8 @@
 #include "fe_utils/connect.h"
 #include "fe_utils/string_utils.h"
 
+#define ERRCODE_UNDEFINED_TABLE  "42P01"
+
 
 static PGcancel *volatile cancelConn = NULL;
 bool		CancelRequested = false;
@@ -178,6 +180,28 @@ connectMaintenanceDatabase(const char *maintenance_db,
 	return conn;
 }
 
+/*
+ * Disconnect the given connection, canceling any statement if one is active.
+ */
+void
+disconnectDatabase(PGconn *conn)
+{
+	char		errbuf[256];
+
+	if (PQtransactionStatus(conn) == PQTRANS_ACTIVE)
+	{
+		PGcancel   *cancel;
+
+		if ((cancel = PQgetCancel(conn)))
+		{
+			(void) PQcancel(cancel, errbuf, sizeof(errbuf));
+			PQfreeCancel(cancel);
+		}
+	}
+
+	PQfinish(conn);
+}
+
 /*
  * Run a query, return the results, exit program on failure.
  */
@@ -255,6 +279,60 @@ executeMaintenanceCommand(PGconn *conn, const char *query, bool echo)
 	return r;
 }
 
+/*
+ * getQuerySucess
+ *
+ * Pump the conn till it's dry of results; return false if any are errors.
+ * Note that this will block if the conn is busy.
+ */
+bool
+getQuerySuccess(PGconn *conn, const char *progname)
+{
+	bool		ok = true;
+	PGresult   *result;
+
+	SetCancelConn(conn);
+	while ((result = PQgetResult(conn)) != NULL)
+	{
+		if (!processQueryResult(conn, result, progname))
+			ok = false;
+	}
+	ResetCancelConn();
+	return ok;
+}
+
+/*
+ * processQueryResult
+ *
+ * Process (and delete) a query result.  Returns true if there's no error,
+ * false otherwise -- but errors about trying to vacuum a missing relation
+ * are reported and subsequently ignored.
+ */
+bool
+processQueryResult(PGconn *conn, PGresult *result, const char *progname)
+{
+	/*
+	 * If it's an error, report it.  Errors about a missing table are harmless
+	 * so we continue processing; but die for other errors.
+	 */
+	if (PQresultStatus(result) != PGRES_COMMAND_OK)
+	{
+		char	   *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
+
+		pg_log_error("processing of database \"%s\" failed: %s",
+					 PQdb(conn), PQerrorMessage(conn));
+
+		if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
+		{
+			PQclear(result);
+			return false;
+		}
+	}
+
+	PQclear(result);
+	return true;
+}
+
 
 /*
  * Split TABLE[(COLUMNS)] into TABLE and [(COLUMNS)] portions.  When you
diff --git a/src/bin/scripts/common.h b/src/bin/scripts/common.h
index 35d1a3e0d5..9f10e1da96 100644
--- a/src/bin/scripts/common.h
+++ b/src/bin/scripts/common.h
@@ -39,6 +39,8 @@ extern PGconn *connectMaintenanceDatabase(const char *maintenance_db,
 										  const char *pguser, enum trivalue prompt_password,
 										  const char *progname, bool echo);
 
+extern void disconnectDatabase(PGconn *conn);
+
 extern PGresult *executeQuery(PGconn *conn, const char *query,
 							  const char *progname, bool echo);
 
@@ -48,6 +50,11 @@ extern void executeCommand(PGconn *conn, const char *query,
 extern bool executeMaintenanceCommand(PGconn *conn, const char *query,
 									  bool echo);
 
+extern bool getQuerySuccess(PGconn *conn, const char *progname);
+
+extern bool processQueryResult(PGconn *conn, PGresult *result,
+							   const char *progname);
+
 extern void splitTableColumnsSpec(const char *spec, int encoding,
 								  char **table, const char **columns);
 
diff --git a/src/bin/scripts/scripts_parallel.c b/src/bin/scripts/scripts_parallel.c
new file mode 100644
index 0000000000..d75f1e1960
--- /dev/null
+++ b/src/bin/scripts/scripts_parallel.c
@@ -0,0 +1,279 @@
+/*-------------------------------------------------------------------------
+ *
+ *	scripts_parallel.c
+ *		Parallel support for bin/scripts/
+ *
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/bin/scripts/scripts_parallel.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres_fe.h"
+
+#ifdef HAVE_SYS_SELECT_H
+#include <sys/select.h>
+#endif
+
+#include "common.h"
+#include "common/logging.h"
+#include "scripts_parallel.h"
+
+static ParallelSlot *get_idle_slot(ParallelSlot slots[], int numslots,
+								   const char *progname);
+static void init_slot(ParallelSlot *slot, PGconn *conn);
+static int	select_loop(int maxFd, fd_set *workerset, bool *aborting);
+
+
+/*
+ * Get the connection slot to use.  If there are multiples slots, here we wait
+ * for one connection to become available if none already is, returning NULL if
+ * an error occured.  Otherwise, we simply use the only slot we have, which we know to
+ * be free.
+ */
+ParallelSlot *
+ConsumeIdleSlot(ParallelSlot *slots, int numslots,
+				const char *progname)
+{
+	ParallelSlot *free_slot;
+
+	if (numslots > 1)
+	{
+		/*
+		 * Get a free slot, waiting until one becomes free if none currently
+		 * is.
+		 */
+		free_slot = get_idle_slot(slots, numslots, progname);
+		if (!free_slot)
+			return NULL;
+
+		free_slot->isFree = false;
+	}
+	else
+		free_slot = slots;
+
+	return free_slot;
+}
+
+/*
+ * get_idle_slot
+ *		Return a connection slot that is ready to execute a command.
+ *
+ * We return the first slot we find that is marked isFree, if one is;
+ * otherwise, we loop on select() until one socket becomes available.  When
+ * this happens, we read the whole set and mark as free all sockets that become
+ * available.
+ *
+ * If an error occurs, NULL is returned.
+ */
+static ParallelSlot *
+get_idle_slot(ParallelSlot slots[], int numslots,
+			  const char *progname)
+{
+	int			i;
+	int			firstFree = -1;
+
+	/* Any connection already known free? */
+	for (i = 0; i < numslots; i++)
+	{
+		if (slots[i].isFree)
+			return slots + i;
+	}
+
+	/*
+	 * No free slot found, so wait until one of the connections has finished
+	 * its task and return the available slot.
+	 */
+	while (firstFree < 0)
+	{
+		fd_set		slotset;
+		int			maxFd = 0;
+		bool		aborting;
+
+		/* We must reconstruct the fd_set for each call to select_loop */
+		FD_ZERO(&slotset);
+
+		for (i = 0; i < numslots; i++)
+		{
+			int			sock = PQsocket(slots[i].connection);
+
+			/*
+			 * We don't really expect any connections to lose their sockets
+			 * after startup, but just in case, cope by ignoring them.
+			 */
+			if (sock < 0)
+				continue;
+
+			FD_SET(sock, &slotset);
+			if (sock > maxFd)
+				maxFd = sock;
+		}
+
+		SetCancelConn(slots->connection);
+		i = select_loop(maxFd, &slotset, &aborting);
+		ResetCancelConn();
+
+		if (aborting)
+		{
+			/*
+			 * We set the cancel-receiving connection to the one in the zeroth
+			 * slot above, so fetch the error from there.
+			 */
+			getQuerySuccess(slots->connection, progname);
+			return NULL;
+		}
+		Assert(i != 0);
+
+		for (i = 0; i < numslots; i++)
+		{
+			int			sock = PQsocket(slots[i].connection);
+
+			if (sock >= 0 && FD_ISSET(sock, &slotset))
+			{
+				/* select() says input is available, so consume it */
+				PQconsumeInput(slots[i].connection);
+			}
+
+			/* Collect result(s) as long as any are available */
+			while (!PQisBusy(slots[i].connection))
+			{
+				PGresult   *result = PQgetResult(slots[i].connection);
+
+				if (result != NULL)
+				{
+					/* Check and discard the command result */
+					if (!processQueryResult(slots[i].connection, result,
+											progname))
+						return NULL;
+				}
+				else
+				{
+					/* This connection has become idle */
+					slots[i].isFree = true;
+					if (firstFree < 0)
+						firstFree = i;
+					break;
+				}
+			}
+		}
+	}
+
+	return slots + firstFree;
+}
+
+static void
+init_slot(ParallelSlot *slot, PGconn *conn)
+{
+	slot->connection = conn;
+	/* Initially assume connection is idle */
+	slot->isFree = true;
+}
+
+ParallelSlot *
+SetupParallelSlots(const char *dbname, const char *host, const char *port,
+				   const char *username, bool prompt_password, const char *progname,
+				   bool echo, PGconn *conn, int numslots)
+{
+	ParallelSlot *slots;
+	int			i;
+
+	slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * numslots);
+	init_slot(slots, conn);
+	if (numslots > 1)
+	{
+		for (i = 1; i < numslots; i++)
+		{
+			conn = connectDatabase(dbname, host, port, username, prompt_password,
+								   progname, echo, false, true);
+			init_slot(slots + i, conn);
+		}
+	}
+
+	return slots;
+}
+
+/*
+ * Loop on select() until a descriptor from the given set becomes readable.
+ *
+ * If we get a cancel request while we're waiting, we forego all further
+ * processing and set the *aborting flag to true.  The return value must be
+ * ignored in this case.  Otherwise, *aborting is set to false.
+ */
+static int
+select_loop(int maxFd, fd_set *workerset, bool *aborting)
+{
+	int			i;
+	fd_set		saveSet = *workerset;
+
+	if (CancelRequested)
+	{
+		*aborting = true;
+		return -1;
+	}
+	else
+		*aborting = false;
+
+	for (;;)
+	{
+		/*
+		 * On Windows, we need to check once in a while for cancel requests;
+		 * on other platforms we rely on select() returning when interrupted.
+		 */
+		struct timeval *tvp;
+#ifdef WIN32
+		struct timeval tv = {0, 1000000};
+
+		tvp = &tv;
+#else
+		tvp = NULL;
+#endif
+
+		*workerset = saveSet;
+		i = select(maxFd + 1, workerset, NULL, NULL, tvp);
+
+#ifdef WIN32
+		if (i == SOCKET_ERROR)
+		{
+			i = -1;
+
+			if (WSAGetLastError() == WSAEINTR)
+				errno = EINTR;
+		}
+#endif
+
+		if (i < 0 && errno == EINTR)
+			continue;			/* ignore this */
+		if (i < 0 || CancelRequested)
+			*aborting = true;	/* but not this */
+		if (i == 0)
+			continue;			/* timeout (Win32 only) */
+		break;
+	}
+
+	return i;
+}
+
+/* Wait for all connections to finish, returning true if no error occured.. */
+bool
+WaitForSlotsCompletion(ParallelSlot *slots, int numslots, const char *progname)
+{
+	int			i;
+
+	/*
+	 * If the callers didn't use multiple connections, it already waited for
+	 * query completion
+	 */
+	if (numslots <= 1)
+		return true;
+
+	for (i = 0; i < numslots; i++)
+	{
+		if (!getQuerySuccess((slots + i)->connection, progname))
+			return false;
+	}
+
+	return true;
+}
diff --git a/src/bin/scripts/scripts_parallel.h b/src/bin/scripts/scripts_parallel.h
new file mode 100644
index 0000000000..34111fd310
--- /dev/null
+++ b/src/bin/scripts/scripts_parallel.h
@@ -0,0 +1,32 @@
+/*
+ *	scripts_parallel.h
+ *		Parallel support for bin/scripts/
+ *
+ *	Copyright (c) 2003-2019, PostgreSQL Global Development Group
+ *
+ *	src/bin/scripts/scripts_parallel.h
+ */
+#ifndef SCRIPTS_PARALLEL_H
+#define SCRIPTS_PARALLEL_H
+
+/* Parallel processing stuff */
+typedef struct ParallelSlot
+{
+	PGconn	   *connection;		/* One connection */
+	bool		isFree;			/* Is it known to be idle? */
+} ParallelSlot;
+
+extern ParallelSlot *ConsumeIdleSlot(ParallelSlot *slots, int numslots,
+									 const char *progname);
+
+extern ParallelSlot *SetupParallelSlots(const char *dbname, const char *host,
+										const char *port,
+										const char *username, bool prompt_password,
+										const char *progname, bool echo,
+										PGconn *conn, int numslots);
+
+extern bool WaitForSlotsCompletion(ParallelSlot *slots, int numslots,
+								   const char *progname);
+
+
+#endif							/* SCRIPTS_PARALLEL_H */
diff --git a/src/bin/scripts/vacuumdb.c b/src/bin/scripts/vacuumdb.c
index 3bcd14b4dc..78e315eb44 100644
--- a/src/bin/scripts/vacuumdb.c
+++ b/src/bin/scripts/vacuumdb.c
@@ -12,10 +12,6 @@
 
 #include "postgres_fe.h"
 
-#ifdef HAVE_SYS_SELECT_H
-#include <sys/select.h>
-#endif
-
 #include "catalog/pg_class_d.h"
 
 #include "common.h"
@@ -23,17 +19,9 @@
 #include "fe_utils/connect.h"
 #include "fe_utils/simple_list.h"
 #include "fe_utils/string_utils.h"
+#include "scripts_parallel.h"
 
 
-#define ERRCODE_UNDEFINED_TABLE  "42P01"
-
-/* Parallel vacuuming stuff */
-typedef struct ParallelSlot
-{
-	PGconn	   *connection;		/* One connection */
-	bool		isFree;			/* Is it known to be idle? */
-} ParallelSlot;
-
 /* vacuum options controlled by user flags */
 typedef struct vacuumingOptions
 {
@@ -71,20 +59,6 @@ static void prepare_vacuum_command(PQExpBuffer sql, int serverVersion,
 static void run_vacuum_command(PGconn *conn, const char *sql, bool echo,
 							   const char *table, const char *progname, bool async);
 
-static ParallelSlot *GetIdleSlot(ParallelSlot slots[], int numslots,
-								 const char *progname);
-
-static bool ProcessQueryResult(PGconn *conn, PGresult *result,
-							   const char *progname);
-
-static bool GetQueryResult(PGconn *conn, const char *progname);
-
-static void DisconnectDatabase(ParallelSlot *slot);
-
-static int	select_loop(int maxFd, fd_set *workerset, bool *aborting);
-
-static void init_slot(ParallelSlot *slot, PGconn *conn);
-
 static void help(const char *progname);
 
 /* For analyze-in-stages mode */
@@ -625,17 +599,9 @@ vacuum_one_database(const char *dbname, vacuumingOptions *vacopts,
 	 */
 	if (concurrentCons <= 0)
 		concurrentCons = 1;
-	slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * concurrentCons);
-	init_slot(slots, conn);
-	if (parallel)
-	{
-		for (i = 1; i < concurrentCons; i++)
-		{
-			conn = connectDatabase(dbname, host, port, username, prompt_password,
-								   progname, echo, false, true);
-			init_slot(slots + i, conn);
-		}
-	}
+
+	slots = SetupParallelSlots(dbname, host, port, username, prompt_password,
+							   progname, echo, conn, concurrentCons);
 
 	/*
 	 * Prepare all the connections to run the appropriate analyze stage, if
@@ -666,29 +632,12 @@ vacuum_one_database(const char *dbname, vacuumingOptions *vacopts,
 			goto finish;
 		}
 
-		/*
-		 * Get the connection slot to use.  If in parallel mode, here we wait
-		 * for one connection to become available if none already is.  In
-		 * non-parallel mode we simply use the only slot we have, which we
-		 * know to be free.
-		 */
-		if (parallel)
+		free_slot = ConsumeIdleSlot(slots, concurrentCons, progname);
+		if (!free_slot)
 		{
-			/*
-			 * Get a free slot, waiting until one becomes free if none
-			 * currently is.
-			 */
-			free_slot = GetIdleSlot(slots, concurrentCons, progname);
-			if (!free_slot)
-			{
-				failed = true;
-				goto finish;
-			}
-
-			free_slot->isFree = false;
+			failed = true;
+			goto finish;
 		}
-		else
-			free_slot = slots;
 
 		prepare_vacuum_command(&sql, PQserverVersion(free_slot->connection),
 							   vacopts, tabname);
@@ -704,24 +653,20 @@ vacuum_one_database(const char *dbname, vacuumingOptions *vacopts,
 		cell = cell->next;
 	} while (cell != NULL);
 
-	if (parallel)
-	{
-		int			j;
-
-		/* wait for all connections to finish */
-		for (j = 0; j < concurrentCons; j++)
-		{
-			if (!GetQueryResult((slots + j)->connection, progname))
-			{
-				failed = true;
-				goto finish;
-			}
-		}
-	}
+	if (!WaitForSlotsCompletion(slots, concurrentCons, progname))
+		failed = true;
 
 finish:
 	for (i = 0; i < concurrentCons; i++)
-		DisconnectDatabase(slots + i);
+	{
+		PGconn	   *conn = slots[i].connection;
+
+		if (conn == NULL)
+			continue;
+
+		disconnectDatabase(conn);
+		conn = NULL;
+	}
 	pfree(slots);
 
 	termPQExpBuffer(&sql);
@@ -953,262 +898,6 @@ run_vacuum_command(PGconn *conn, const char *sql, bool echo,
 	}
 }
 
-/*
- * GetIdleSlot
- *		Return a connection slot that is ready to execute a command.
- *
- * We return the first slot we find that is marked isFree, if one is;
- * otherwise, we loop on select() until one socket becomes available.  When
- * this happens, we read the whole set and mark as free all sockets that become
- * available.
- *
- * If an error occurs, NULL is returned.
- */
-static ParallelSlot *
-GetIdleSlot(ParallelSlot slots[], int numslots,
-			const char *progname)
-{
-	int			i;
-	int			firstFree = -1;
-
-	/* Any connection already known free? */
-	for (i = 0; i < numslots; i++)
-	{
-		if (slots[i].isFree)
-			return slots + i;
-	}
-
-	/*
-	 * No free slot found, so wait until one of the connections has finished
-	 * its task and return the available slot.
-	 */
-	while (firstFree < 0)
-	{
-		fd_set		slotset;
-		int			maxFd = 0;
-		bool		aborting;
-
-		/* We must reconstruct the fd_set for each call to select_loop */
-		FD_ZERO(&slotset);
-
-		for (i = 0; i < numslots; i++)
-		{
-			int			sock = PQsocket(slots[i].connection);
-
-			/*
-			 * We don't really expect any connections to lose their sockets
-			 * after startup, but just in case, cope by ignoring them.
-			 */
-			if (sock < 0)
-				continue;
-
-			FD_SET(sock, &slotset);
-			if (sock > maxFd)
-				maxFd = sock;
-		}
-
-		SetCancelConn(slots->connection);
-		i = select_loop(maxFd, &slotset, &aborting);
-		ResetCancelConn();
-
-		if (aborting)
-		{
-			/*
-			 * We set the cancel-receiving connection to the one in the zeroth
-			 * slot above, so fetch the error from there.
-			 */
-			GetQueryResult(slots->connection, progname);
-			return NULL;
-		}
-		Assert(i != 0);
-
-		for (i = 0; i < numslots; i++)
-		{
-			int			sock = PQsocket(slots[i].connection);
-
-			if (sock >= 0 && FD_ISSET(sock, &slotset))
-			{
-				/* select() says input is available, so consume it */
-				PQconsumeInput(slots[i].connection);
-			}
-
-			/* Collect result(s) as long as any are available */
-			while (!PQisBusy(slots[i].connection))
-			{
-				PGresult   *result = PQgetResult(slots[i].connection);
-
-				if (result != NULL)
-				{
-					/* Check and discard the command result */
-					if (!ProcessQueryResult(slots[i].connection, result,
-											progname))
-						return NULL;
-				}
-				else
-				{
-					/* This connection has become idle */
-					slots[i].isFree = true;
-					if (firstFree < 0)
-						firstFree = i;
-					break;
-				}
-			}
-		}
-	}
-
-	return slots + firstFree;
-}
-
-/*
- * ProcessQueryResult
- *
- * Process (and delete) a query result.  Returns true if there's no error,
- * false otherwise -- but errors about trying to vacuum a missing relation
- * are reported and subsequently ignored.
- */
-static bool
-ProcessQueryResult(PGconn *conn, PGresult *result, const char *progname)
-{
-	/*
-	 * If it's an error, report it.  Errors about a missing table are harmless
-	 * so we continue processing; but die for other errors.
-	 */
-	if (PQresultStatus(result) != PGRES_COMMAND_OK)
-	{
-		char	   *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
-
-		pg_log_error("vacuuming of database \"%s\" failed: %s",
-					 PQdb(conn), PQerrorMessage(conn));
-
-		if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
-		{
-			PQclear(result);
-			return false;
-		}
-	}
-
-	PQclear(result);
-	return true;
-}
-
-/*
- * GetQueryResult
- *
- * Pump the conn till it's dry of results; return false if any are errors.
- * Note that this will block if the conn is busy.
- */
-static bool
-GetQueryResult(PGconn *conn, const char *progname)
-{
-	bool		ok = true;
-	PGresult   *result;
-
-	SetCancelConn(conn);
-	while ((result = PQgetResult(conn)) != NULL)
-	{
-		if (!ProcessQueryResult(conn, result, progname))
-			ok = false;
-	}
-	ResetCancelConn();
-	return ok;
-}
-
-/*
- * DisconnectDatabase
- *		Disconnect the connection associated with the given slot
- */
-static void
-DisconnectDatabase(ParallelSlot *slot)
-{
-	char		errbuf[256];
-
-	if (!slot->connection)
-		return;
-
-	if (PQtransactionStatus(slot->connection) == PQTRANS_ACTIVE)
-	{
-		PGcancel   *cancel;
-
-		if ((cancel = PQgetCancel(slot->connection)))
-		{
-			(void) PQcancel(cancel, errbuf, sizeof(errbuf));
-			PQfreeCancel(cancel);
-		}
-	}
-
-	PQfinish(slot->connection);
-	slot->connection = NULL;
-}
-
-/*
- * Loop on select() until a descriptor from the given set becomes readable.
- *
- * If we get a cancel request while we're waiting, we forego all further
- * processing and set the *aborting flag to true.  The return value must be
- * ignored in this case.  Otherwise, *aborting is set to false.
- */
-static int
-select_loop(int maxFd, fd_set *workerset, bool *aborting)
-{
-	int			i;
-	fd_set		saveSet = *workerset;
-
-	if (CancelRequested)
-	{
-		*aborting = true;
-		return -1;
-	}
-	else
-		*aborting = false;
-
-	for (;;)
-	{
-		/*
-		 * On Windows, we need to check once in a while for cancel requests;
-		 * on other platforms we rely on select() returning when interrupted.
-		 */
-		struct timeval *tvp;
-#ifdef WIN32
-		struct timeval tv = {0, 1000000};
-
-		tvp = &tv;
-#else
-		tvp = NULL;
-#endif
-
-		*workerset = saveSet;
-		i = select(maxFd + 1, workerset, NULL, NULL, tvp);
-
-#ifdef WIN32
-		if (i == SOCKET_ERROR)
-		{
-			i = -1;
-
-			if (WSAGetLastError() == WSAEINTR)
-				errno = EINTR;
-		}
-#endif
-
-		if (i < 0 && errno == EINTR)
-			continue;			/* ignore this */
-		if (i < 0 || CancelRequested)
-			*aborting = true;	/* but not this */
-		if (i == 0)
-			continue;			/* timeout (Win32 only) */
-		break;
-	}
-
-	return i;
-}
-
-static void
-init_slot(ParallelSlot *slot, PGconn *conn)
-{
-	slot->connection = conn;
-	/* Initially assume connection is idle */
-	slot->isFree = true;
-}
-
 static void
 help(const char *progname)
 {
-- 
2.20.1