You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudberry.apache.org by ma...@apache.org on 2025/03/23 02:36:59 UTC
(cloudberry) 02/02: Reuse the epoll object to fix the dispatcher performance downgrade (#14800)
This is an automated email from the ASF dual-hosted git repository.
maxyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
commit a56d4f936fcee89b43ea1dffe3e87ce369682e05
Author: Hongxu Ma <in...@outlook.com>
AuthorDate: Thu Feb 9 16:12:25 2023 +0800
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)
```
---
contrib/interconnect/udp/ic_udpifc.c | 79 +++++++++++++++++++++++-------------
src/backend/storage/ipc/latch.c | 78 +++++++++++++++++++++++++++++++++++
src/include/storage/latch.h | 3 ++
3 files changed, 132 insertions(+), 28 deletions(-)
diff --git a/contrib/interconnect/udp/ic_udpifc.c b/contrib/interconnect/udp/ic_udpifc.c
index 44f5f83fe7..d2fcd8cb45 100644
--- a/contrib/interconnect/udp/ic_udpifc.c
+++ b/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)
{
diff --git a/src/backend/storage/ipc/latch.c b/src/backend/storage/ipc/latch.c
index 1c9104968c..e81041ae02 100644
--- a/src/backend/storage/ipc/latch.c
+++ b/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.
*
diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h
index 44f9368c64..c6cfafc8cd 100644
--- a/src/include/storage/latch.h
+++ b/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 */
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org