Reuse the epoll object to fix the dispatcher performance downgrade (#14800)

In commit e30909f, I used the "WaitEventSet" to refactor the related dispatch logic.
But in the following pgbench test, we found the performance downgrade 50% due to this commit. Did a perf on it, the bottleneck is native_queued_spin_lock_slowpath, means there is a lot of lock contention in it.

After investigation, we found the culprit is a large number of concurrent creation and destruction of epoll objects. Fix it in the commit: reuse the epoll object.And here are the pgbench results for reference:

```
./pgbench tpcb -j 200 -c 200  -n -P 1 -h mdw -p 5432 -U gpadmin -T 60 -S
// main
tps = 48662.387634 (including connections establishing)
tps = 48697.870330 (excluding connections establishing)

// revert e30909f
...
tps = 151641.552024 (including connections establishing)
tps = 151757.591140 (excluding connections establishing)

// main with this commit (fixed)
...
tps = 148574.817871 (including connections establishing)
tps = 148666.308622 (excluding connections establishing)

// 6x
...
tps = 114706.905303 (including connections establishing)
tps = 114795.896798 (excluding connections establishing)
```

Author: interma

contrib/interconnect/udp/ic_udpifc.c

@@ -385,6 +385,8 @@ struct SendControlInfo
  * and congestion control.
  */
 static SendControlInfo snd_control_info;
+/* WaitEventSet for the icudp */
+static WaitEventSet *ICWaitSet = NULL;
 
 /*
  * ICGlobalControlInfo
@@ -779,6 +781,8 @@ static inline void logPkt(char *prefix, icpkthdr *pkt);
 static void aggregateStatistics(ChunkTransportStateEntry *pChunkEntry);
 
 static inline bool pollAcks(ChunkTransportState *transportStates, int fd, int timeout);
+static TupleChunkListItem receiveChunksUDPIFCLoop(ChunkTransportState *pTransportStates, ChunkTransportStateEntry *pEntry,
+						int16 *srcRoute, MotionConn *conn, WaitEventSet *waitset, int nevent);
 
 /* #define TRANSFER_PROTOCOL_STATS */
 
@@ -3820,15 +3824,10 @@ static TupleChunkListItem
 receiveChunksUDPIFC(ChunkTransportState *pTransportStates, ChunkTransportStateEntry *pEntry,
 					int16 motNodeID, int16 *srcRoute, MotionConn *mConn)
 {
-	int			retries = 0;
 	bool		directed = false;
 	int 		nFds = 0;
 	int 		*waitFds = NULL;
 	int 		nevent = 0;
-	MotionConn 	*rxconn = NULL;
-	MotionConnUDP 	*udpRXconn = NULL;
-	WaitEvent	*rEvents = NULL;
-	WaitEventSet		*waitset = NULL;
 	TupleChunkListItem	tcItem = NULL;
 	MotionConnUDP *conn = NULL;
 
@@ -3865,17 +3864,53 @@ receiveChunksUDPIFC(ChunkTransportState *pTransportStates, ChunkTransportStateEn
 
 	}
 
-	/* init WaitEventSet */
-	waitset = CreateWaitEventSet(CurrentMemoryContext, nevent);
-	rEvents = palloc(nevent * sizeof(WaitEvent)); /* returned events */
-	AddWaitEventToSet(waitset, WL_LATCH_SET, PGINVALID_SOCKET, &ic_control_info.latch, NULL);
-	AddWaitEventToSet(waitset, WL_POSTMASTER_DEATH, PGINVALID_SOCKET, NULL, NULL);
+	/* reset WaitEventSet */
+	ResetWaitEventSet(&ICWaitSet, TopMemoryContext, nevent);
+
+	/*
+	 * Use PG_TRY() - PG_CATCH() to make sure destroy the waiteventset (close the epoll fd)
+	 * The main receive logic is in receiveChunksUDPIFCLoop()
+	 */
+	PG_TRY();
+	{
+		AddWaitEventToSet(ICWaitSet, WL_LATCH_SET, PGINVALID_SOCKET, &ic_control_info.latch, NULL);
+		AddWaitEventToSet(ICWaitSet, WL_POSTMASTER_DEATH, PGINVALID_SOCKET, NULL, NULL);
+		for (int i = 0; i < nFds; i++)
+		{
+			AddWaitEventToSet(ICWaitSet, WL_SOCKET_READABLE, waitFds[i], NULL, NULL);
+		}
 
-	for (int i = 0; i < nFds; i++)
+		tcItem = receiveChunksUDPIFCLoop(pTransportStates, pEntry, srcRoute, mConn, ICWaitSet, nevent);
+	}
+	PG_CATCH();
 	{
-		AddWaitEventToSet(waitset, WL_SOCKET_READABLE, waitFds[i], NULL, NULL);
+		if (waitFds != NULL)
+			pfree(waitFds);
+		PG_RE_THROW();
 	}
+	PG_END_TRY();
+
+	if (waitFds != NULL)
+		pfree(waitFds);
+
+	return tcItem;
+}
+
+static TupleChunkListItem
+receiveChunksUDPIFCLoop(ChunkTransportState *pTransportStates, ChunkTransportStateEntry *pEntry,
+						int16 *srcRoute, MotionConn *conn, WaitEventSet *waitset, int nevent)
+{
+	TupleChunkListItem	tcItem = NULL;
+	MotionConn 	*rxconn = NULL;
+	MotionConnUDP   *udpRXconn = NULL;
+	int			retries = 0;
+	bool		directed = false;
+	WaitEvent	*rEvents = NULL;
 
+	if (conn != NULL)
+		directed = true;
+
+	rEvents = palloc(nevent * sizeof(WaitEvent)); /* returned events */
 	/* we didn't have any data, so we've got to read it from the network. */
 	for (;;)
 	{
@@ -3905,13 +3940,7 @@ receiveChunksUDPIFC(ChunkTransportState *pTransportStates, ChunkTransportStateEn
 
 			if (!directed)
 				*srcRoute = udpRXconn->route;
-
-			FreeWaitEventSet(waitset);
-			if (rEvents != NULL)
-				pfree(rEvents);
-			if (waitFds != NULL)
-				pfree(waitFds);
-
+			pfree(rEvents);
 			return tcItem;
 		}
 
@@ -3963,7 +3992,7 @@ receiveChunksUDPIFC(ChunkTransportState *pTransportStates, ChunkTransportStateEn
 		}
 
 		/*
-		 * 1. NIC on master (and thus the QD connection) may become bad, check
+		 * 1. NIC on coordinator (and thus the QD connection) may become bad, check
 		 * it. 2. Postmaster may become invalid, check it
 		 */
 		if ((retries & 0x3f) == 0)
@@ -3978,19 +4007,13 @@ receiveChunksUDPIFC(ChunkTransportState *pTransportStates, ChunkTransportStateEn
 		}
 
 		pthread_mutex_lock(&ic_control_info.lock);
-
-	}							/* for (;;) */
-
-	FreeWaitEventSet(waitset);
-	if (rEvents != NULL)
-		pfree(rEvents);
-	if (waitFds != NULL)
-		pfree(waitFds);
+	} /* for (;;) */
 
 	/* We either got data, or get cancelled. We never make it out to here. */
 	return NULL;				/* make GCC behave */
 }
 
+
 TupleChunkListItem
 RecvTupleChunkUDPIFC(MotionConn *conn, ChunkTransportState *transportStates)
 {

src/backend/storage/ipc/latch.c

@@ -785,6 +785,84 @@ CreateWaitEventSet(MemoryContext context, int nevents)
 	return set;
 }
 
+/*
+ * If *pset == NULL create a new WaitEventSet and set it to *pset
+ * Else reset the existed WaitEventSet:
+ *  - if enable epoll, it frees the old memory and reuse the epoll object.
+ *  - else it just Free+Create a new WaitEventSet.
+ *
+ * Note: Why we want to reuse the epoll object? In previous code, it create
+ * and free waiteventset several times for one query. And it leads performance
+ * downgrade in pgbench test. After investigation, we found the culprit is
+ * a large number of concurrent creation and destruction of epoll objects. So
+ * we introduce the function to reuse epoll object.
+ */
+void
+ResetWaitEventSet(WaitEventSet **pset, MemoryContext context, int nevents)
+{
+	if (*pset == NULL)
+	{
+		*pset = CreateWaitEventSet(context, nevents);
+		return;
+	}
+
+#if defined(WAIT_USE_EPOLL)
+	WaitEventSet *set = *pset;
+	/* delete existed events */
+	for (int i = 0; i < set->nevents; i++)
+	{
+		WaitEvent  *event = &(set->events[i]);
+		int			rc;
+		rc = epoll_ctl(set->epoll_fd, EPOLL_CTL_DEL, event->fd, NULL);
+		if (rc < 0)
+		{
+			ereport(DEBUG1,
+				(errcode_for_socket_access(),
+				/* translator: %s is a syscall name, such as "poll()" */
+				 errmsg("%s cleanup epoll events failed: %m",
+					 "epoll_ctl()")));
+			/*
+			 * in some scenarios, the event's fd has been cleanup,
+			 * so need to destory the whold epoll object
+			 */
+			goto CREATE_NEW_EV;
+		}
+	}
+
+	/* the same alloc logic as CreateWaitEventSet() */
+	char	   *data;
+	Size		sz = 0;
+	sz += MAXALIGN(sizeof(WaitEventSet));
+	sz += MAXALIGN(sizeof(WaitEvent) * nevents);
+	sz += MAXALIGN(sizeof(struct epoll_event) * nevents);
+	data = (char *) MemoryContextAllocZero(context, sz);
+
+	/* reuse the epoll object and free the old memory */
+	int	old_epoll_fd = set->epoll_fd;
+	pfree(set);
+
+	set = (WaitEventSet *) data;
+	data += MAXALIGN(sizeof(WaitEventSet));
+	set->events = (WaitEvent *) data;
+	data += MAXALIGN(sizeof(WaitEvent) * nevents);
+	set->epoll_ret_events = (struct epoll_event *) data;
+
+	set->latch = NULL;
+	set->nevents_space = nevents;
+	set->exit_on_postmaster_death = false;
+
+	/* reuse the epoll object (is empty now) */
+	set->epoll_fd = old_epoll_fd;
+
+	*pset = set;
+	return;
+#endif
+CREATE_NEW_EV:
+	FreeWaitEventSet(*pset);
+	*pset = CreateWaitEventSet(context, nevents);
+
+}
+
 /*
  * Free a previously created WaitEventSet.
  *

src/include/storage/latch.h

@@ -180,5 +180,8 @@ extern int	WaitLatchOrSocket(Latch *latch, int wakeEvents,
 							  pgsocket sock, long timeout, uint32 wait_event_info);
 extern void InitializeLatchWaitSet(void);
 extern int	GetNumRegisteredWaitEvents(WaitEventSet *set);
+/* specifial function for gpdb */
+extern void ResetWaitEventSet(WaitEventSet **pset, MemoryContext context, int nevents);
+
 
 #endif							/* LATCH_H */