Accommodate for AF_INET6 when doing a motion layer IPC teardown

Previously on commit 70306db18e2, we removed pg_getaddrinfo_all for
signal handlers. However, in doing so, the capability of supporting both
AF_INET6 and AF_INET was lost; this responsibility must now be handled
by us. The commit mentioned above fixed the issue for AF_INET (IPv4),
but not for AF_INET6 (IPv6).

This commit addresses the situation for both AF_INET and AF_INET6.

Reviewed-by: Soumyadeep Chakraborty <soumyadeep2007@gmail.com>

Author: Chibin

contrib/interconnect/udp/ic_udpifc.c

@@ -640,8 +640,8 @@ typedef struct ICStatistics
 /* Statistics for UDP interconnect. */
 static ICStatistics ic_statistics;
 
-static struct addrinfo udp_dummy_packet_addrinfo;
-static struct sockaddr udp_dummy_packet_sockaddr;
+/* Cached sockaddr of the listening udp socket */
+static struct sockaddr_storage udp_dummy_packet_sockaddr;
 
 /* UDP listen fd */
 int			UDP_listenerFd;
@@ -670,10 +670,15 @@ static void setRxThreadError(int eno);
 static void resetRxThreadError(void);
 static void SendDummyPacket(void);
 
+static void ConvertToIPv4MappedAddr(struct sockaddr_storage *sockaddr, socklen_t *o_len);
+#if defined(__darwin__)
+#define	s6_addr32 __u6_addr.__u6_addr32
+static void ConvertIPv6WildcardToLoopback(struct sockaddr_storage* dest);
+#endif
 static void getSockAddr(struct sockaddr_storage *peer, socklen_t *peer_len, const char *listenerAddr, int listenerPort);
 static uint32 setUDPSocketBufferSize(int ic_socket, int buffer_type);
 static void setupUDPListeningSocket(int *listenerSocketFd, int32 *listenerPort,
-							int *txFamily, struct addrinfo *listenerAddrinfo, struct sockaddr *listenerSockaddr);
+							int *txFamily, struct sockaddr_storage *listenerSockaddr);
 static ChunkTransportStateEntry *startOutgoingUDPConnections(ChunkTransportState *transportStates,
 							ExecSlice *sendSlice,
 							int *pOutgoingCount);
@@ -1166,7 +1171,7 @@ resetRxThreadError()
  * 		Setup udp listening socket.
  */
 static void
-setupUDPListeningSocket(int *listenerSocketFd, int32 *listenerPort, int *txFamily, struct addrinfo *listenerAddrinfo, struct sockaddr *listenerSockaddr)
+setupUDPListeningSocket(int *listenerSocketFd, int32 *listenerPort, int *txFamily, struct sockaddr_storage *listenerSockaddr)
 {
 	struct addrinfo 		*addrs = NULL;
 	struct addrinfo 		*addr;
@@ -1287,16 +1292,6 @@ setupUDPListeningSocket(int *listenerSocketFd, int32 *listenerPort, int *txFamil
 	if (!addr || ic_socket == PGINVALID_SOCKET)
 		goto startup_failed;
 
-	/*
-	 * cache the successful addrinfo and sockaddr of the listening socket, so
-	 * we can use this information to connect to the listening socket.
-	 */
-	if (listenerAddrinfo != NULL && listenerSockaddr != NULL )
-	{
-		memcpy(listenerAddrinfo, addr, sizeof(udp_dummy_packet_addrinfo));
-		memcpy(listenerSockaddr, addr->ai_addr, sizeof(udp_dummy_packet_sockaddr));
-	}
-
 	/* Memorize the socket fd, kernel assigned port and address family */
 	*listenerSocketFd = ic_socket;
 	if (listenerAddr.ss_family == AF_INET6)
@@ -1310,6 +1305,13 @@ setupUDPListeningSocket(int *listenerSocketFd, int32 *listenerPort, int *txFamil
 		*txFamily = AF_INET;
 	}
 
+	/*
+	 * cache the successful sockaddr of the listening socket, so
+	 * we can use this information to connect to the listening socket.
+	 */
+	if (listenerSockaddr != NULL)
+		memcpy(listenerSockaddr, &listenerAddr, sizeof(struct sockaddr_storage));
+
 	/* Set up socket non-blocking mode */
 	if (!pg_set_noblock(ic_socket))
 	{
@@ -1442,9 +1444,8 @@ InitMotionUDPIFC(int *listenerSocketFd, int32 *listenerPort)
 	/*
 	 * setup listening socket and sending socket for Interconnect.
 	 */
-	setupUDPListeningSocket(listenerSocketFd, listenerPort, &txFamily,
-			&udp_dummy_packet_addrinfo, &udp_dummy_packet_sockaddr);
-	setupUDPListeningSocket(&ICSenderSocket, &ICSenderPort, &ICSenderFamily, NULL, NULL);
+	setupUDPListeningSocket(listenerSocketFd, listenerPort, &txFamily, &udp_dummy_packet_sockaddr);
+	setupUDPListeningSocket(&ICSenderSocket, &ICSenderPort, &ICSenderFamily, NULL);
 
 	/* Initialize receive control data. */
 	resetMainThreadWaiting(&rx_control_info.mainWaitingState);
@@ -1552,7 +1553,6 @@ static inline void CleanupMotionUDPIFC(void)
 	ICSenderPort = 0;
 	ICSenderFamily = 0;
 
-	memset(&udp_dummy_packet_addrinfo, 0, sizeof(udp_dummy_packet_addrinfo));
 	memset(&udp_dummy_packet_sockaddr, 0, sizeof(udp_dummy_packet_sockaddr));
 
 #ifdef USE_ASSERT_CHECKING
@@ -2848,30 +2848,8 @@ setupOutgoingUDPConnection(ChunkTransportState *transportStates, ChunkTransportS
 			 */
 			if (pEntry->txfd_family == AF_INET6)
 			{
-				struct sockaddr_storage temp;
-				const struct sockaddr_in *in = (const struct sockaddr_in *) &conn->peer;
-				struct sockaddr_in6 *in6_new = (struct sockaddr_in6 *) &temp;
-
-				memset(&temp, 0, sizeof(temp));
-
 				elog(DEBUG1, "We are inet6, remote is inet.  Converting to v4 mapped address.");
-
-				/* Construct a V4-to-6 mapped address.  */
-				temp.ss_family = AF_INET6;
-				in6_new->sin6_family = AF_INET6;
-				in6_new->sin6_port = in->sin_port;
-				in6_new->sin6_flowinfo = 0;
-
-				memset(&in6_new->sin6_addr, '\0', sizeof(in6_new->sin6_addr));
-				/* in6_new->sin6_addr.s6_addr16[5] = 0xffff; */
-				((uint16 *) &in6_new->sin6_addr)[5] = 0xffff;
-				/* in6_new->sin6_addr.s6_addr32[3] = in->sin_addr.s_addr; */
-				memcpy(((char *) &in6_new->sin6_addr) + 12, &(in->sin_addr), 4);
-				in6_new->sin6_scope_id = 0;
-
-				/* copy it back */
-				memcpy(&conn->peer, &temp, sizeof(struct sockaddr_in6));
-				conn->peer_len = sizeof(struct sockaddr_in6);
+				ConvertToIPv4MappedAddr(&conn->peer, &conn->peer_len);
 			}
 			else
 			{
@@ -7237,28 +7215,94 @@ WaitInterconnectQuitUDPIFC(void)
 	ic_control_info.threadCreated = false;
 }
 
+/*
+ * If the socket was created AF_INET6, but the address we want to
+ * send to is IPv4 (AF_INET), we need to change the address
+ * format. On Linux, this is not necessary: glibc automatically
+ * handles this. But on MAC OSX and Solaris, we need to convert
+ * the IPv4 address to IPv4-mapped IPv6 address in AF_INET6 format.
+ *
+ * The comment above relies on getaddrinfo() via function getSockAddr to get
+ * the correct V4-mapped address. We need to be careful here as we need to
+ * ensure that the platform we are using is POSIX 1003-2001 compliant.
+ * Just to be on the safeside, we'll be keeping this function for
+ * now to be used for all platforms and not rely on POSIX.
+ *
+ * Since this can be called in a signal handler, we avoid the use of
+ * async-signal unsafe functions such as memset/memcpy
+ */
+static void
+ConvertToIPv4MappedAddr(struct sockaddr_storage *sockaddr, socklen_t *o_len)
+{
+	const struct sockaddr_in *in = (const struct sockaddr_in *) sockaddr;
+	struct sockaddr_storage temp = {0};
+	struct sockaddr_in6 *in6_new = (struct sockaddr_in6 *) &temp;
+
+	/* Construct a IPv4-to-IPv6 mapped address.  */
+	temp.ss_family = AF_INET6;
+	in6_new->sin6_family = AF_INET6;
+	in6_new->sin6_port = in->sin_port;
+	in6_new->sin6_flowinfo = 0;
+
+	((uint16 *) &in6_new->sin6_addr)[5] = 0xffff;
+
+	in6_new->sin6_addr.s6_addr32[3] = in->sin_addr.s_addr;
+	in6_new->sin6_scope_id = 0;
+
+	/* copy it back */
+	*sockaddr = temp;
+	*o_len = sizeof(struct sockaddr_in6);
+}
+
+#if defined(__darwin__)
+/* macos does not accept :: as the destination, we will need to covert this to the IPv6 loopback */
+static void
+ConvertIPv6WildcardToLoopback(struct sockaddr_storage* dest)
+{
+	char address[INET6_ADDRSTRLEN];
+	/* we want to terminate our own process, so this should be local */
+	const struct sockaddr_in6 *in6 = (const struct sockaddr_in6 *) &udp_dummy_packet_sockaddr;
+	inet_ntop(AF_INET6, &in6->sin6_addr, address, sizeof(address));
+	if (strcmp("::", address) == 0)
+		((struct sockaddr_in6 *)dest)->sin6_addr = in6addr_loopback;
+}
+#endif
+
 /*
  * Send a dummy packet to interconnect thread to exit poll() immediately
  */
 static void
 SendDummyPacket(void)
 {
 	int					ret;
-	in_port_t			udp_listener_port;
 	char				*dummy_pkt = "stop it";
 	int					counter;
-	struct sockaddr_in	*addr_in = NULL;
-	struct sockaddr_in	dest_addr;
-	/*
-	 * Get address info from interconnect udp listener port
-	 */
-	udp_listener_port = (Gp_listener_port >> 16) & 0x0ffff;
+	struct sockaddr_storage dest;
+	socklen_t	dest_len;
+
+	Assert(udp_dummy_packet_sockaddr.ss_family == AF_INET || udp_dummy_packet_sockaddr.ss_family == AF_INET6);
+	Assert(ICSenderFamily == AF_INET || ICSenderFamily == AF_INET6);
 
-	addr_in = (struct sockaddr_in *) &udp_dummy_packet_sockaddr;
-	memset(&dest_addr, 0, sizeof(dest_addr));
-	dest_addr.sin_family = addr_in->sin_family;
-	dest_addr.sin_port = htons(udp_listener_port);
-	dest_addr.sin_addr.s_addr = addr_in->sin_addr.s_addr;
+	dest = udp_dummy_packet_sockaddr;
+	dest_len = (ICSenderFamily == AF_INET) ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6);
+
+	if (ICSenderFamily == AF_INET6)
+	{
+#if defined(__darwin__)
+		if (udp_dummy_packet_sockaddr.ss_family == AF_INET6)
+			ConvertIPv6WildcardToLoopback(&dest);
+#endif
+		if (udp_dummy_packet_sockaddr.ss_family == AF_INET)
+			ConvertToIPv4MappedAddr(&dest, &dest_len);
+	}
+
+	if (ICSenderFamily == AF_INET && udp_dummy_packet_sockaddr.ss_family == AF_INET6)
+	{
+		/* the size of AF_INET6 is bigger than the side of IPv4, so
+		 * converting from IPv6 to IPv4 may potentially not work. */
+		ereport(LOG, errmsg("sending dummy packet failed: cannot send from AF_INET to receiving on AF_INET6"));
+		return;
+	}
 
 	/*
 	 * Send a dummy package to the interconnect listener, try 10 times.
@@ -7269,24 +7313,22 @@ SendDummyPacket(void)
 	while (counter < 10)
 	{
 		counter++;
-		ret = sendto(ICSenderSocket, dummy_pkt, strlen(dummy_pkt), 0, (struct sockaddr *) &dest_addr, sizeof(dest_addr));
+		ret = sendto(ICSenderSocket, dummy_pkt, strlen(dummy_pkt), 0, (struct sockaddr *) &dest, dest_len);
 		if (ret < 0)
 		{
 			if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
 				continue;
 			else
 			{
-				elog(LOG, "send dummy packet failed, sendto failed: %m");
+				ereport(LOG, errmsg("send dummy packet failed, sendto failed: %m"));
 				return;
 			}
 		}
 		break;
 	}
 
 	if (counter >= 10)
-	{
-		elog(LOG, "send dummy packet failed, sendto failed with 10 times: %m");
-	}
+		ereport(LOG, errmsg("send dummy packet failed, sendto failed with 10 times: %m"));
 }
 
 void logChunkParseDetails(MotionConn *conn, uint32 ic_instance_id)

contrib/interconnect/udp/test/Makefile

@@ -0,0 +1,12 @@
+subdir=contrib/interconnect/udp
+top_builddir=../../../../..
+include $(top_builddir)/src/Makefile.global
+
+TARGETS=cdbsenddummypacket
+
+include $(top_builddir)/src/backend/mock.mk
+
+cdbsenddummypacket.t: EXCL_OBJS += contrib/interconnect/udp/ic_udpifc.o
+cdbsenddummypacket.t: \
+	$(MOCK_DIR)/backend/access/hash/hash_mock.o \
+	$(MOCK_DIR)/backend/utils/fmgr/fmgr_mock.o