Add interconnect teardown hook

and add ability to get UDP IC stats from anywhere outside udpifc module.
Both features are to be used in monitoring extensions.

Author: Smyatkin-Maxim

gpcontrib/gp_interconnect_stats/gp_interconnect_stats.c

@@ -8,7 +8,7 @@
 #include "postgres.h"
 
 #include "cdb/cdbvars.h"
-#include "cdb/ml_ipc.h"
+#include "cdb/ic_udpifc.h"
 #include "fmgr.h"
 
 PG_MODULE_MAGIC;

src/backend/cdb/motion/ic_common.c

@@ -25,6 +25,7 @@
 #include "cdb/ml_ipc.h"
 #include "cdb/cdbvars.h"
 #include "cdb/cdbdisp.h"
+#include "cdb/ic_udpifc.h"
 
 #include <unistd.h>
 #include <arpa/inet.h>
@@ -58,6 +59,8 @@ int			UDP_listenerFd;
 static interconnect_handle_t *open_interconnect_handles;
 static bool interconnect_resowner_callback_registered;
 
+ic_teardown_hook_type ic_teardown_hook = NULL;
+
 /*=========================================================================
  * FUNCTIONS PROTOTYPES
  */
@@ -576,6 +579,9 @@ TeardownInterconnect(ChunkTransportState *transportStates, bool hasErrors)
 {
 	interconnect_handle_t *h = find_interconnect_handle(transportStates);
 
+	if (ic_teardown_hook)
+		ic_teardown_hook(transportStates, hasErrors);
+
 	if (Gp_interconnect_type == INTERCONNECT_TYPE_UDPIFC)
 	{
 		TeardownUDPIFCInterconnect(transportStates, hasErrors);

src/backend/cdb/motion/ic_udpifc.c

@@ -53,6 +53,7 @@
 #include "cdb/cdbdisp.h"
 #include "cdb/cdbdispatchresult.h"
 #include "cdb/cdbicudpfaultinjection.h"
+#include "cdb/ic_udpifc.h"
 
 #include <fcntl.h>
 #include <limits.h>
@@ -586,52 +587,6 @@ typedef struct AckSendParam
 	socklen_t	peer_len;
 } AckSendParam;
 
-/*
- * ICStatistics
- *
- * A structure keeping various statistics about interconnect internal.
- *
- * Note that the statistics for ic are not accurate for multiple cursor case on QD.
- *
- * totalRecvQueueSize        - receive queue size sum when main thread is trying to get a packet.
- * recvQueueSizeCountingTime - counting times when computing totalRecvQueueSize.
- * totalCapacity             - the capacity sum when packets are tried to be sent.
- * capacityCountingTime      - counting times used to compute totalCapacity.
- * totalBuffers              - total buffers available when sending packets.
- * bufferCountingTime        - counting times when compute totalBuffers.
- * activeConnectionsNum      - the number of active connections.
- * retransmits               - the number of packet retransmits.
- * mismatchNum               - the number of mismatched packets received.
- * crcErrors                 - the number of crc errors.
- * sndPktNum                 - the number of packets sent by sender.
- * recvPktNum                - the number of packets received by receiver.
- * disorderedPktNum          - disordered packet number.
- * duplicatedPktNum          - duplicate packet number.
- * recvAckNum                - the number of Acks received.
- * statusQueryMsgNum         - the number of status query messages sent.
- *
- */
-typedef struct ICStatistics
-{
-	uint64		totalRecvQueueSize;
-	uint64		recvQueueSizeCountingTime;
-	uint64		totalCapacity;
-	uint64		capacityCountingTime;
-	uint64		totalBuffers;
-	uint64		bufferCountingTime;
-	uint32		activeConnectionsNum;
-	int32		retransmits;
-	int32		startupCachedPktNum;
-	int32		mismatchNum;
-	int32		crcErrors;
-	int32		sndPktNum;
-	int32		recvPktNum;
-	int32		disorderedPktNum;
-	int32		duplicatedPktNum;
-	int32		recvAckNum;
-	int32		statusQueryMsgNum;
-} ICStatistics;
-
 /* Statistics for UDP interconnect. */
 static ICStatistics ic_statistics;
 
@@ -7125,6 +7080,15 @@ getActiveMotionConns(void)
 	return ic_statistics.activeConnectionsNum;
 }
 
+ICStatistics
+UDPIFCGetICStats(void)
+{
+	pthread_mutex_lock(&ic_control_info.lock);
+	ICStatistics stats = ic_statistics;
+	pthread_mutex_unlock(&ic_control_info.lock);
+	return stats;
+}
+
 Datum
 GpInterconnectGetStatsUDPIFC(PG_FUNCTION_ARGS)
 {

src/include/cdb/ic_udpifc.h

@@ -0,0 +1,80 @@
+/*-------------------------------------------------------------------------
+ * ic_udpifc.h
+ *	  Motion Layer IPC Layer.
+ *
+ * Portions Copyright (c) 2005-2008, Greenplum inc
+ * Portions Copyright (c) 2012-2025 Pivotal Software, Inc.
+ * Portions Copyright (c) 2025- Present open-gpdb
+ *
+ *
+ * IDENTIFICATION
+ *	    src/include/cdb/ic_udpifc.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef IC_UDPIFC_H
+#define IC_UDPIFC_H
+
+#include "postgres.h"
+#include "fmgr.h"
+
+
+#include <unistd.h>
+
+/*
+ * ICStatistics
+ *
+ * A structure keeping various statistics about interconnect internal.
+ *
+ * Note that the statistics for ic are not accurate for multiple cursor case on QD.
+ *
+ * totalRecvQueueSize        - receive queue size sum when main thread is trying to get a packet.
+ * recvQueueSizeCountingTime - counting times when computing totalRecvQueueSize.
+ * totalCapacity             - the capacity sum when packets are tried to be sent.
+ * capacityCountingTime      - counting times used to compute totalCapacity.
+ * totalBuffers              - total buffers available when sending packets.
+ * bufferCountingTime        - counting times when compute totalBuffers.
+ * activeConnectionsNum      - the number of active connections.
+ * retransmits               - the number of packet retransmits.
+ * mismatchNum               - the number of mismatched packets received.
+ * crcErrors                 - the number of crc errors.
+ * sndPktNum                 - the number of packets sent by sender.
+ * recvPktNum                - the number of packets received by receiver.
+ * disorderedPktNum          - disordered packet number.
+ * duplicatedPktNum          - duplicate packet number.
+ * recvAckNum                - the number of Acks received.
+ * statusQueryMsgNum         - the number of status query messages sent.
+ *
+ */
+typedef struct ICStatistics
+{
+	uint64		totalRecvQueueSize;
+	uint64		recvQueueSizeCountingTime;
+	uint64		totalCapacity;
+	uint64		capacityCountingTime;
+	uint64		totalBuffers;
+	uint64		bufferCountingTime;
+	uint32		activeConnectionsNum;
+	int32		retransmits;
+	int32		startupCachedPktNum;
+	int32		mismatchNum;
+	int32		crcErrors;
+	int32		sndPktNum;
+	int32		recvPktNum;
+	int32		disorderedPktNum;
+	int32		duplicatedPktNum;
+	int32		recvAckNum;
+	int32		statusQueryMsgNum;
+} ICStatistics;
+
+extern void InterconnectShmemInitUDPIFC(void);
+extern Size InterconnectShmemSizeUDPIFC(void);
+extern void WaitInterconnectQuitUDPIFC(void);
+
+/* Get interconnect statistics local to this slice */
+extern ICStatistics UDPIFCGetICStats(void);
+
+/* Get global cummulative IC stats */
+extern Datum GpInterconnectGetStatsUDPIFC(PG_FUNCTION_ARGS);
+
+#endif

src/include/cdb/ml_ipc.h

@@ -307,16 +307,12 @@ extern ChunkTransportStateEntry *removeChunkTransportState(ChunkTransportState *
 extern TupleChunkListItem RecvTupleChunk(MotionConn *conn, ChunkTransportState *transportStates);
 
 extern Size InterconnectShmemSize(void);
-extern Size InterconnectShmemSizeUDPIFC(void);
 extern void InterconnectShmemInit(void);
-extern void InterconnectShmemInitUDPIFC(void);
-extern Datum GpInterconnectGetStatsUDPIFC(PG_FUNCTION_ARGS);
 extern void InitMotionTCP(int *listenerSocketFd, uint16 *listenerPort);
 extern void InitMotionUDPIFC(int *listenerSocketFd, uint16 *listenerPort);
 extern void markUDPConnInactiveIFC(MotionConn *conn);
 extern void CleanupMotionTCP(void);
 extern void CleanupMotionUDPIFC(void);
-extern void WaitInterconnectQuitUDPIFC(void);
 extern void SetupTCPInterconnect(EState *estate);
 extern void SetupUDPIFCInterconnect(EState *estate);
 extern void TeardownTCPInterconnect(ChunkTransportState *transportStates,
@@ -328,4 +324,8 @@ extern uint32 getActiveMotionConns(void);
 
 extern char *format_sockaddr(struct sockaddr_storage *sa, char *buf, size_t len);
 
+#define IC_TEARDOWN_HOOK
+typedef void (*ic_teardown_hook_type)(ChunkTransportState *, bool);
+extern PGDLLIMPORT ic_teardown_hook_type ic_teardown_hook;
+
 #endif   /* ML_IPC_H */