Function wait_on_slots

src/fe_utils/parallel_slot.c:197–281

* Wait for any slot's connection to have query results, consume the results, * and update the slot's status as appropriate. Returns true on success, * false on cancellation, on error, or if no slots are connected. */

 * false on cancellation, on error, or if no slots are connected.
 */
static bool
wait_on_slots(ParallelSlotArray *sa)
{
	int			i;
	fd_set		slotset;
	int			maxFd = 0;
	PGconn	   *cancelconn = NULL;

	/* We must reconstruct the fd_set for each call to select_loop */
	FD_ZERO(&slotset);

	for (i = 0; i < sa->numslots; i++)
	{
		int			sock;

		/* We shouldn't get here if we still have slots without connections */
		Assert(sa->slots[i].connection != NULL);

		sock = PQsocket(sa->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;

		/* Keep track of the first valid connection we see. */
		if (cancelconn == NULL)
			cancelconn = sa->slots[i].connection;

		FD_SET(sock, &slotset);
		if (sock > maxFd)
			maxFd = sock;
	}

	/*
	 * If we get this far with no valid connections, processing cannot
	 * continue.
	 */
	if (cancelconn == NULL)
		return false;

	SetCancelConn(sa->slots->connection);
	i = select_loop(maxFd, &slotset);
	ResetCancelConn();

	/* failure? */
	if (i < 0)
		return false;

	for (i = 0; i < sa->numslots; i++)
	{
		int			sock;

		sock = PQsocket(sa->slots[i].connection);

		if (sock >= 0 && FD_ISSET(sock, &slotset))