Add hooks to support extensible interconnect

The interconnect module is traditionally single-thread. It's hard
to refactor/replace the current implementation for thread-safe.

This commit adds hooks to support extensible interconnect. The hooks
contain 3 functions in struct ExtInterconnectFuncs:

struct ExtInterconnectFuncs {
	void (*init)();
	int (*port)();
	void (*exit)();
};

init: create socket to receive interconnect packets.
port: return the port number created in init.
exit: disconnect all interconnections and cleanup related resources.

Author: gfphoenix78

src/backend/cdb/cdbutil.c

@@ -79,6 +79,7 @@
 
 MemoryContext CdbComponentsContext = NULL;
 static CdbComponentDatabases *cdb_component_dbs = NULL;
+const struct ExtInterconnectFuncs *ext_interconnect = NULL;
 
 #ifdef USE_INTERNAL_FTS
 
@@ -1149,6 +1150,9 @@ cdb_setup(void)
 		ensureInterconnectAddress();
 		/* Initialize the Motion Layer IPC subsystem. */
 		CurrentMotionIPCLayer->InitMotionLayerIPC();
+
+		if (ext_interconnect && ext_interconnect->init)
+			ext_interconnect->init();
 	}
 
 	/*
@@ -1213,6 +1217,9 @@ cdb_cleanup(int code pg_attribute_unused(), Datum arg
 	{
 		/* shutdown our listener socket */
 		CurrentMotionIPCLayer->CleanUpMotionLayerIPC();
+
+		if (ext_interconnect && ext_interconnect->exit)
+			ext_interconnect->exit();
 	}
 }
 
@@ -3542,6 +3549,9 @@ cdb_setup(void)
 		ensureInterconnectAddress();
 		/* Initialize the Motion Layer IPC subsystem. */
 		CurrentMotionIPCLayer->InitMotionLayerIPC();
+
+		if (ext_interconnect && ext_interconnect->init)
+			ext_interconnect->init();
 	}
 
 	/*
@@ -3595,6 +3605,9 @@ cdb_cleanup(int code pg_attribute_unused(), Datum arg
 	{
 		/* shutdown our listener socket */
 		CurrentMotionIPCLayer->CleanUpMotionLayerIPC();
+
+		if (ext_interconnect && ext_interconnect->exit)
+			ext_interconnect->exit();
 	}
 }
 

src/backend/cdb/dispatcher/cdbconn.c

@@ -31,7 +31,7 @@
 #include "cdb/cdbgang.h"
 
 
-static uint32 cdbconn_get_motion_listener_port(PGconn *conn);
+static int64 cdbconn_get_motion_listener_port(PGconn *conn);
 static void cdbconn_disconnect(SegmentDatabaseDescriptor *segdbDesc);
 
 static void MPPnoticeReceiver(void *arg, const PGresult *res);
@@ -83,6 +83,7 @@ cdbconn_createSegmentDescriptor(struct CdbComponentDatabaseInfo *cdbinfo, int id
 	/* Connection info, set in function cdbconn_doConnect */
 	segdbDesc->conn = NULL;
 	segdbDesc->motionListener = 0;
+	segdbDesc->motionExtListener = 0;
 	segdbDesc->backendPid = 0;
 
 	/* whoami */
@@ -294,16 +295,19 @@ cdbconn_doConnectComplete(SegmentDatabaseDescriptor *segdbDesc)
 	 * giving us the TCP port number where it listens for connections from the
 	 * gang below.
 	 */
-	segdbDesc->motionListener = cdbconn_get_motion_listener_port(segdbDesc->conn);
+	int64 ports = cdbconn_get_motion_listener_port(segdbDesc->conn);
+	segdbDesc->motionListener = (uint32) (ports & 0xFFFFFFFF);
+	segdbDesc->motionExtListener = ports >> 32;
 	segdbDesc->backendPid = PQbackendPID(segdbDesc->conn);
 
 	if (segdbDesc->motionListener != 0 &&
 		gp_log_gang >= GPVARS_VERBOSITY_DEBUG)
 	{
-		elog(LOG, "Connected to %s motionListenerPorts=%u/%u with options %s",
+		elog(LOG, "Connected to %s motionListenerPorts=%u/%u/%u with options %s",
 			 segdbDesc->whoami,
 			 (segdbDesc->motionListener & 0x0ffff),
 			 ((segdbDesc->motionListener >> 16) & 0x0ffff),
+			 segdbDesc->motionExtListener,
 			 PQoptions(segdbDesc->conn));
 	}
 }
@@ -483,25 +487,32 @@ cdbconn_signalQE(SegmentDatabaseDescriptor *segdbDesc,
 
 
 /* GPDB function to retrieve QE-backend details (motion listener) */
-static uint32
+static int64
 cdbconn_get_motion_listener_port(PGconn *conn)
 {
 	const char *val;
 	char	   *endptr;
-	uint32		result;
+	int64		result;
+	int64		ext_port;
 
 	val = PQparameterStatus(conn, "qe_listener_port");
 	if (!val)
 		return 0;
 
 	errno = 0;
-	result = strtoul(val, &endptr, 10);
+	result = strtol(val, &endptr, 10);
+	if (endptr == val || *endptr != ':' || errno == ERANGE)
+		return 0;
+
+	val = endptr + 1;
+	ext_port = strtol(val, &endptr, 10);
 	if (endptr == val || *endptr != '\0' || errno == ERANGE)
 		return 0;
 
-	return result;
-}
+	if (result < 0 || ext_port < 0) return 0;
 
+	return (result | (ext_port << 32));
+}
 
 /*-------------------------------------------------------------------------
  * QE Notice receiver support

src/backend/cdb/dispatcher/cdbgang.c

@@ -627,6 +627,7 @@ makeCdbProcess(SegmentDatabaseDescriptor *segdbDesc)
 	process->listenerAddr = pstrdup(qeinfo->config->hostip);
 
 	process->listenerPort = (segdbDesc->motionListener & 0x0ffff);
+	process->listenerExtPort = segdbDesc->motionExtListener;
 	process->pid = segdbDesc->backendPid;
 	process->contentid = segdbDesc->segindex;
 	process->dbid = qeinfo->config->dbid;
@@ -1101,6 +1102,7 @@ gp_backend_info(PG_FUNCTION_ARGS)
 		qddesc->segindex = -1;
 		qddesc->conn = NULL;
 		qddesc->motionListener = 0;
+		qddesc->motionExtListener = 0;
 		qddesc->backendPid = MyProcPid;
 		qddesc->whoami = NULL;
 		qddesc->isWriter = false;

src/backend/nodes/copyfuncs.c

@@ -5747,6 +5747,7 @@ _copyCdbProcess(const CdbProcess *from)
 
 	COPY_STRING_FIELD(listenerAddr);
 	COPY_SCALAR_FIELD(listenerPort);
+	COPY_SCALAR_FIELD(listenerExtPort);
 	COPY_SCALAR_FIELD(pid);
 	COPY_SCALAR_FIELD(contentid);
 	COPY_SCALAR_FIELD(dbid);

src/backend/nodes/outfuncs_common.c

@@ -1460,6 +1460,7 @@ _outCdbProcess(StringInfo str, const CdbProcess *node)
 	WRITE_NODE_TYPE("CDBPROCESS");
 	WRITE_STRING_FIELD(listenerAddr);
 	WRITE_INT_FIELD(listenerPort);
+	WRITE_INT_FIELD(listenerExtPort);
 	WRITE_INT_FIELD(pid);
 	WRITE_INT_FIELD(contentid);
 	WRITE_INT_FIELD(dbid);

src/backend/nodes/readfuncs_common.c

@@ -536,6 +536,7 @@ _readCdbProcess(void)
 
 	READ_STRING_FIELD(listenerAddr);
 	READ_INT_FIELD(listenerPort);
+	READ_INT_FIELD(listenerExtPort);
 	READ_INT_FIELD(pid);
 	READ_INT_FIELD(contentid);
 	READ_INT_FIELD(dbid);

src/backend/tcop/dest.c

@@ -329,9 +329,13 @@ void
 sendQEDetails(void)
 {
 	StringInfoData msgbuf;
-	char		port_str[11];
+	char		port_str[32];
+	int ext_port = 0;
 
-	snprintf(port_str, sizeof(port_str), "%u", CurrentMotionIPCLayer->GetListenPort());
+	if (ext_interconnect && ext_interconnect->port)
+		ext_port = ext_interconnect->port();
+
+	snprintf(port_str, sizeof(port_str), "%u:%d", CurrentMotionIPCLayer->GetListenPort(), ext_port);
 
 	pq_beginmessage(&msgbuf, 'S');
 	pq_sendstring(&msgbuf, "qe_listener_port");

src/include/cdb/cdbconn.h

@@ -37,24 +37,25 @@ typedef struct SegmentDatabaseDescriptor
 	 */
 	int32         			segindex;
 
-    /*
+	/*
 	 * A non-NULL value points to the PGconn block of a successfully
 	 * established connection to the segment database.
 	 */
 	PGconn				   *conn;		
 
-    /*
-     * Connection info saved at most recent PQconnectdb.
-     *
-     * NB: Use malloc/free, not palloc/pfree, for the items below.
-     */
-    uint32		            motionListener; /* interconnect listener port */
-    int32					backendPid;
-    char                   *whoami;         /* QE identifier for msgs */
-	bool					isWriter;
-	int						identifier;		/* unique identifier in the cdbcomponent segment pool */
-	double					establishConnTime; /* the time of establish connection to the segment,
-												* -1 means this connection is cached */
+	/*
+	* Connection info saved at most recent PQconnectdb.
+	*
+	* NB: Use malloc/free, not palloc/pfree, for the items below.
+	*/
+	uint32		motionListener; /* interconnect listener port */
+	int32		backendPid;
+	char		*whoami;         /* QE identifier for msgs */
+	bool		isWriter;
+	uint16		motionExtListener; /* extensible interconnect listener port */
+	int			identifier;		/* unique identifier in the cdbcomponent segment pool */
+	double		establishConnTime; /* the time of establish connection to the segment,
+									* -1 means this connection is cached */
 } SegmentDatabaseDescriptor;
 
 SegmentDatabaseDescriptor *

src/include/cdb/cdbgang.h

@@ -126,6 +126,7 @@ typedef struct CdbProcess
 	char *listenerAddr; /* Interconnect listener IPv4 address, a C-string */
 
 	int listenerPort; /* Interconnect listener port */
+	int listenerExtPort; /* extensible interconnect listener port */
 	int pid; /* Backend PID of the process. */
 
 	int contentid;

src/include/cdb/cdbutil.h

@@ -51,6 +51,13 @@ typedef enum SegmentType
 	SEGMENTTYPE_ANY
 }SegmentType;
 
+struct ExtInterconnectFuncs {
+	void (*init)();
+	int (*port)();
+	void (*exit)();
+};
+extern const struct ExtInterconnectFuncs *ext_interconnect;
+
 /*
  * performs all necessary setup required for initializing Apache Cloudberry components.
  *