Handle invalid libpq sockets in more places

Also, make error messages consistent.

From: Michael Paquier <michael.paquier@gmail.com>

src/backend/replication/libpqwalreceiver/libpqwalreceiver.c

@@ -331,7 +331,7 @@ libpq_select(int timeout_ms)
 	if (PQsocket(streamConn) < 0)
 		ereport(ERROR,
 				(errcode_for_socket_access(),
-				 errmsg("socket not open")));
+				 errmsg("invalid socket: %s", PQerrorMessage(streamConn))));
 
 	/* We use poll(2) if available, otherwise select(2) */
 	{

src/bin/pg_basebackup/pg_recvlogical.c

@@ -360,6 +360,14 @@ StreamLogicalLog(void)
 			struct timeval timeout;
 			struct timeval *timeoutptr = NULL;
 
+			if (PQsocket(conn) < 0)
+			{
+				fprintf(stderr,
+						_("%s: invalid socket: %s"),
+						progname, PQerrorMessage(conn));
+				goto error;
+			}
+
 			FD_ZERO(&input_mask);
 			FD_SET(PQsocket(conn), &input_mask);
 

src/bin/pg_basebackup/receivelog.c

@@ -956,7 +956,8 @@ CopyStreamPoll(PGconn *conn, long timeout_ms)
 
 	if (PQsocket(conn) < 0)
 	{
-		fprintf(stderr, _("%s: socket not open"), progname);
+		fprintf(stderr, _("%s: invalid socket: %s"), progname,
+				PQerrorMessage(conn));
 		return -1;
 	}
 

src/bin/pgbench/pgbench.c

@@ -3797,7 +3797,7 @@ threadRun(void *arg)
 			sock = PQsocket(st->con);
 			if (sock < 0)
 			{
-				fprintf(stderr, "bad socket: %s", PQerrorMessage(st->con));
+				fprintf(stderr, "invalid socket: %s", PQerrorMessage(st->con));
 				goto done;
 			}
 
@@ -3867,7 +3867,8 @@ threadRun(void *arg)
 
 				if (sock < 0)
 				{
-					fprintf(stderr, "bad socket: %s", PQerrorMessage(st->con));
+					fprintf(stderr, "invalid socket: %s",
+							PQerrorMessage(st->con));
 					goto done;
 				}
 				if (FD_ISSET(sock, &input_mask) ||

src/bin/scripts/vacuumdb.c

@@ -70,7 +70,7 @@ 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 init_slot(ParallelSlot *slot, PGconn *conn, const char *progname);
 
 static void help(const char *progname);
 
@@ -421,14 +421,14 @@ vacuum_one_database(const char *dbname, vacuumingOptions *vacopts,
 	 * array contains the connection.
 	 */
 	slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * concurrentCons);
-	init_slot(slots, conn);
+	init_slot(slots, conn, progname);
 	if (parallel)
 	{
 		for (i = 1; i < concurrentCons; i++)
 		{
 			conn = connectDatabase(dbname, host, port, username, prompt_password,
 								   progname, false, true);
-			init_slot(slots + i, conn);
+			init_slot(slots + i, conn, progname);
 		}
 	}
 
@@ -917,11 +917,18 @@ select_loop(int maxFd, fd_set *workerset, bool *aborting)
 }
 
 static void
-init_slot(ParallelSlot *slot, PGconn *conn)
+init_slot(ParallelSlot *slot, PGconn *conn, const char *progname)
 {
 	slot->connection = conn;
 	slot->isFree = true;
 	slot->sock = PQsocket(conn);
+
+	if (slot->sock < 0)
+	{
+		fprintf(stderr, _("%s: invalid socket: %s"), progname,
+				PQerrorMessage(conn));
+		exit(1);
+	}
 }
 
 static void

src/interfaces/libpq/fe-misc.c

@@ -1058,7 +1058,7 @@ pqSocketCheck(PGconn *conn, int forRead, int forWrite, time_t end_time)
 	if (conn->sock == PGINVALID_SOCKET)
 	{
 		printfPQExpBuffer(&conn->errorMessage,
-						  libpq_gettext("socket not open\n"));
+						  libpq_gettext("invalid socket\n"));
 		return -1;
 	}
 

src/test/isolation/isolationtester.c

@@ -705,6 +705,12 @@ try_complete_step(Step *step, int flags)
 	PGresult   *res;
 	bool		canceled = false;
 
+	if (sock < 0)
+	{
+		fprintf(stderr, "invalid socket: %s", PQerrorMessage(conn));
+		exit_nicely();
+	}
+
 	gettimeofday(&start_time, NULL);
 	FD_ZERO(&read_set);