From ea5f93f0609b2a210cd8282e79dc3c2b934423a2 Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Thu, 2 Jul 2026 13:27:27 +0530 Subject: [PATCH 1/8] fix: segfault when too many confurrent request were sent on macos --- src/event.c | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/src/event.c b/src/event.c index df90b13..ae42475 100644 --- a/src/event.c +++ b/src/event.c @@ -195,16 +195,31 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in curl_multi_assign(wstate->curl_mhandle, sockfd, NULL); pfree(sock_info); } else { + int old_action = sock_info ? sock_info->action : CURL_POLL_NONE; + if (!sock_info) { sock_info = palloc(sizeof(SocketInfo)); sock_info->sockfd = sockfd; - sock_info->action = what; curl_multi_assign(wstate->curl_mhandle, sockfd, sock_info); } - if (what & CURL_POLL_IN) EV_SET(&ev[count++], sockfd, EVFILT_READ, EV_ADD, 0, 0, sock_info); + // curl can flip a socket's interest without ever sending CURL_POLL_REMOVE in + // between (e.g. CURL_POLL_OUT while sending the request, then CURL_POLL_IN + // while reading the response). Only touch the filters whose desired state + // actually changed, and keep `action` in sync so a later CURL_POLL_REMOVE + // deletes everything that's actually registered instead of leaking a kqueue + // entry whose sock_info we've already pfree'd (use-after-free). + if ((what & CURL_POLL_IN) && !(old_action & CURL_POLL_IN)) + EV_SET(&ev[count++], sockfd, EVFILT_READ, EV_ADD, 0, 0, sock_info); + else if (!(what & CURL_POLL_IN) && (old_action & CURL_POLL_IN)) + EV_SET(&ev[count++], sockfd, EVFILT_READ, EV_DELETE, 0, 0, sock_info); + + if ((what & CURL_POLL_OUT) && !(old_action & CURL_POLL_OUT)) + EV_SET(&ev[count++], sockfd, EVFILT_WRITE, EV_ADD, 0, 0, sock_info); + else if (!(what & CURL_POLL_OUT) && (old_action & CURL_POLL_OUT)) + EV_SET(&ev[count++], sockfd, EVFILT_WRITE, EV_DELETE, 0, 0, sock_info); - if (what & CURL_POLL_OUT) EV_SET(&ev[count++], sockfd, EVFILT_WRITE, EV_ADD, 0, 0, sock_info); + sock_info->action = what; } Assert(count <= 2); From 1b1b4c2023588397e1736132a0009a14e9da1488 Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Fri, 3 Jul 2026 20:24:34 +0530 Subject: [PATCH 2/8] fix: similar segfault on linux --- src/event.c | 33 ++++++++++++++++++++++----------- 1 file changed, 22 insertions(+), 11 deletions(-) diff --git a/src/event.c b/src/event.c index ae42475..d4a95a3 100644 --- a/src/event.c +++ b/src/event.c @@ -72,6 +72,10 @@ int multi_timer_cb(__attribute__((unused)) CURLM *multi, long timeout_ms, void * return 0; } +typedef struct { + curl_socket_t sockfd; +} SocketInfo; + int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, int what, void *userp, void *socketp) { WorkerState *wstate = (WorkerState *)userp; @@ -79,25 +83,30 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in "CURL_POLL_REMOVE"}; elog(DEBUG2, "multi_socket_cb: sockfd %d received %s", sockfd, whatstrs[what]); + SocketInfo *sock_info = (SocketInfo *)socketp; + int epoll_op; - if (!socketp) { + if (what == CURL_POLL_REMOVE) { + epoll_op = EPOLL_CTL_DEL; + curl_multi_assign(wstate->curl_mhandle, sockfd, NULL); + } else if (!sock_info) { + // socketp must point at storage that outlives this call, since curl hands it back to us + // verbatim on the next callback for this sockfd. The address of a local variable would be + // a dangling pointer as soon as we return. epoll_op = EPOLL_CTL_ADD; - bool socket_exists = true; - curl_multi_assign(wstate->curl_mhandle, sockfd, &socket_exists); - } else if (what == CURL_POLL_REMOVE) { - epoll_op = EPOLL_CTL_DEL; - bool socket_exists = false; - curl_multi_assign(wstate->curl_mhandle, sockfd, &socket_exists); + sock_info = palloc(sizeof(SocketInfo)); + sock_info->sockfd = sockfd; + curl_multi_assign(wstate->curl_mhandle, sockfd, sock_info); } else { epoll_op = EPOLL_CTL_MOD; } epoll_event ev = { .data.fd = sockfd, - .events = (what & CURL_POLL_IN) ? EPOLLIN - : (what & CURL_POLL_OUT) ? EPOLLOUT - : 0, // no event is assigned since here we get - // CURL_POLL_REMOVE and the sockfd will be removed + // CURL_POLL_INOUT sets both bits, so these must combine rather than pick just one - + // a ternary chain here would silently drop interest in one direction. + .events = (what & CURL_POLL_IN ? EPOLLIN : 0) | (what & CURL_POLL_OUT ? EPOLLOUT : 0), + // no event is assigned when `what` is CURL_POLL_REMOVE and the sockfd is being removed }; // epoll_ctl will copy ev, so there's no need to do palloc for the epoll_event @@ -109,6 +118,8 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in whatstrs[what], opstrs[epoll_op], sockfd, strerror(e))); } + if (epoll_op == EPOLL_CTL_DEL) pfree(sock_info); + return 0; } From 38d7f148df40ce4c36db66de067e36cc57f829e3 Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Mon, 6 Jul 2026 13:52:40 +0530 Subject: [PATCH 3/8] refactor: cleanup epoll impl of multi_socket_cb35 --- src/errors.h | 7 +++++++ src/event.c | 42 ++++++++++++++++++++++-------------------- 2 files changed, 29 insertions(+), 20 deletions(-) diff --git a/src/errors.h b/src/errors.h index 1831fb2..5b07530 100644 --- a/src/errors.h +++ b/src/errors.h @@ -19,6 +19,13 @@ ereport(ERROR, errmsg("Could not curl_multi_setopt(%s)", #opt)); \ } while (0) +#define EREPORT_CURL_MULTI_ASSIGN(multi_handle, sockfd, sockp) \ + do { \ + CURLMcode mresult = curl_multi_assign(multi_handle, sockfd, sockp); \ + if (mresult != CURLM_OK) \ + ereport(ERROR, errmsg("Could not curl_multi_assign(%s)", curl_multi_strerror(mresult))); \ + } while (0) + #define EREPORT_CURL_SLIST_APPEND(list, str) \ do { \ struct curl_slist *new_list = curl_slist_append(list, str); \ diff --git a/src/event.c b/src/event.c index d4a95a3..c0318a5 100644 --- a/src/event.c +++ b/src/event.c @@ -3,7 +3,9 @@ #include #include "pg_prelude.h" +#include "curl_prelude.h" +#include "errors.h" #include "event.h" #ifdef WAIT_USE_EPOLL @@ -72,9 +74,9 @@ int multi_timer_cb(__attribute__((unused)) CURLM *multi, long timeout_ms, void * return 0; } -typedef struct { - curl_socket_t sockfd; -} SocketInfo; +// A marker value only need to be assigned to a socket's socketp pointer via a call to +// curl_multi_assign. +static char socketp_marker; int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, int what, void *userp, void *socketp) { @@ -83,30 +85,32 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in "CURL_POLL_REMOVE"}; elog(DEBUG2, "multi_socket_cb: sockfd %d received %s", sockfd, whatstrs[what]); - SocketInfo *sock_info = (SocketInfo *)socketp; + // libcurl calls the multi_socket_cb with socketp set to null for a socketfd when it is first + // created. At that time we set the socketp to the marker value via a call to curl_multi_assign so + // that any subsequent calls will have that marker value set. This helps us distinguish between + // EPOLL_CTL_ADD and EPOLL_CTL_MOD scenarios. We could have set the socketp to a heap allocated + // value but since we don't need the actual value, we avoid those allocations and use the marker + // value. + bool seen_before = socketp != NULL; int epoll_op; - if (what == CURL_POLL_REMOVE) { + if (!seen_before) { + epoll_op = EPOLL_CTL_ADD; + EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, &socketp_marker); + } else if (what == CURL_POLL_REMOVE) { epoll_op = EPOLL_CTL_DEL; - curl_multi_assign(wstate->curl_mhandle, sockfd, NULL); - } else if (!sock_info) { - // socketp must point at storage that outlives this call, since curl hands it back to us - // verbatim on the next callback for this sockfd. The address of a local variable would be - // a dangling pointer as soon as we return. - epoll_op = EPOLL_CTL_ADD; - sock_info = palloc(sizeof(SocketInfo)); - sock_info->sockfd = sockfd; - curl_multi_assign(wstate->curl_mhandle, sockfd, sock_info); + EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, NULL); } else { epoll_op = EPOLL_CTL_MOD; } epoll_event ev = { .data.fd = sockfd, - // CURL_POLL_INOUT sets both bits, so these must combine rather than pick just one - - // a ternary chain here would silently drop interest in one direction. - .events = (what & CURL_POLL_IN ? EPOLLIN : 0) | (what & CURL_POLL_OUT ? EPOLLOUT : 0), - // no event is assigned when `what` is CURL_POLL_REMOVE and the sockfd is being removed + // We can get the `what` variable value equal to either CURL_POLL_IN, CURL_POLL_OUT, + // CURL_POLL_INOUT (equivalent to CURL_POLL_IN | CURL_POLL_OUT), or CURL_POLL_REMOVE. We want to + // set the events member correspondingly to EPOLLIN, EPOLLOUT, EPOLLIN | EPOLLOUT, or 0 (ignored + // when epoll_op is EPOLL_CTL_DEL). + .events = (what & CURL_POLL_IN ? EPOLLIN : 0) | (what & CURL_POLL_OUT ? EPOLLOUT : 0), }; // epoll_ctl will copy ev, so there's no need to do palloc for the epoll_event @@ -118,8 +122,6 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in whatstrs[what], opstrs[epoll_op], sockfd, strerror(e))); } - if (epoll_op == EPOLL_CTL_DEL) pfree(sock_info); - return 0; } From b290950557cae8ac748817e40509abd6ae5e45aa Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Tue, 7 Jul 2026 13:59:30 +0530 Subject: [PATCH 4/8] refactor: clean up fix on macOS --- src/event.c | 45 ++++++++++++--------------------------------- src/event.h | 12 ++++++++++++ 2 files changed, 24 insertions(+), 33 deletions(-) diff --git a/src/event.c b/src/event.c index c0318a5..b1e8010 100644 --- a/src/event.c +++ b/src/event.c @@ -198,42 +198,21 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in struct kevent ev[2]; int count = 0; - if (what == CURL_POLL_REMOVE) { - if (sock_info->action & CURL_POLL_IN) - EV_SET(&ev[count++], sockfd, EVFILT_READ, EV_DELETE, 0, 0, sock_info); + if (!sock_info) { + sock_info = palloc(sizeof(SocketInfo)); + sock_info->sockfd = sockfd; + sock_info->action = CURL_POLL_NONE; + EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, sock_info); + } else if (what == CURL_POLL_REMOVE) { + EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, NULL); + } - if (sock_info->action & CURL_POLL_OUT) - EV_SET(&ev[count++], sockfd, EVFILT_WRITE, EV_DELETE, 0, 0, sock_info); + UPDATE_FILTER(CURL_POLL_IN, EVFILT_READ); + UPDATE_FILTER(CURL_POLL_OUT, EVFILT_WRITE); - curl_multi_assign(wstate->curl_mhandle, sockfd, NULL); - pfree(sock_info); - } else { - int old_action = sock_info ? sock_info->action : CURL_POLL_NONE; + sock_info->action = what; - if (!sock_info) { - sock_info = palloc(sizeof(SocketInfo)); - sock_info->sockfd = sockfd; - curl_multi_assign(wstate->curl_mhandle, sockfd, sock_info); - } - - // curl can flip a socket's interest without ever sending CURL_POLL_REMOVE in - // between (e.g. CURL_POLL_OUT while sending the request, then CURL_POLL_IN - // while reading the response). Only touch the filters whose desired state - // actually changed, and keep `action` in sync so a later CURL_POLL_REMOVE - // deletes everything that's actually registered instead of leaking a kqueue - // entry whose sock_info we've already pfree'd (use-after-free). - if ((what & CURL_POLL_IN) && !(old_action & CURL_POLL_IN)) - EV_SET(&ev[count++], sockfd, EVFILT_READ, EV_ADD, 0, 0, sock_info); - else if (!(what & CURL_POLL_IN) && (old_action & CURL_POLL_IN)) - EV_SET(&ev[count++], sockfd, EVFILT_READ, EV_DELETE, 0, 0, sock_info); - - if ((what & CURL_POLL_OUT) && !(old_action & CURL_POLL_OUT)) - EV_SET(&ev[count++], sockfd, EVFILT_WRITE, EV_ADD, 0, 0, sock_info); - else if (!(what & CURL_POLL_OUT) && (old_action & CURL_POLL_OUT)) - EV_SET(&ev[count++], sockfd, EVFILT_WRITE, EV_DELETE, 0, 0, sock_info); - - sock_info->action = what; - } + if (what == CURL_POLL_REMOVE) pfree(sock_info); Assert(count <= 2); diff --git a/src/event.h b/src/event.h index d1a76ea..3e992ea 100644 --- a/src/event.h +++ b/src/event.h @@ -24,6 +24,18 @@ typedef struct epoll_event event; # include typedef struct kevent event; +// Diffs `what` (desired filter interest) against `sock_info->action` (currently +// registered interest) for a single poll bit / kqueue filter pair, and appends an +// EV_ADD or EV_DELETE to `ev`/`count` only when that filter's state actually changed. +// Relies on `what`, `sock_info`, `sockfd`, `ev`, and `count` from the caller's scope. +#define UPDATE_FILTER(poll_bit, filter) \ + do { \ + if ((what & (poll_bit)) && !(sock_info->action & (poll_bit))) \ + EV_SET(&ev[count++], sockfd, (filter), EV_ADD, 0, 0, sock_info); \ + else if (!(what & (poll_bit)) && (sock_info->action & (poll_bit))) \ + EV_SET(&ev[count++], sockfd, (filter), EV_DELETE, 0, 0, sock_info); \ + } while (0) + #endif int wait_event(int fd, event *events, size_t maxevents, int wait_milliseconds); From 729a2258b0ed8e8d62a3bc774a3383fd2905154e Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Tue, 7 Jul 2026 14:05:02 +0530 Subject: [PATCH 5/8] refactor: simplify implementation --- src/event.c | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/src/event.c b/src/event.c index b1e8010..fe8b9f7 100644 --- a/src/event.c +++ b/src/event.c @@ -91,10 +91,8 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in // EPOLL_CTL_ADD and EPOLL_CTL_MOD scenarios. We could have set the socketp to a heap allocated // value but since we don't need the actual value, we avoid those allocations and use the marker // value. - bool seen_before = socketp != NULL; - int epoll_op; - if (!seen_before) { + if (!socketp) { epoll_op = EPOLL_CTL_ADD; EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, &socketp_marker); } else if (what == CURL_POLL_REMOVE) { @@ -203,8 +201,6 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in sock_info->sockfd = sockfd; sock_info->action = CURL_POLL_NONE; EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, sock_info); - } else if (what == CURL_POLL_REMOVE) { - EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, NULL); } UPDATE_FILTER(CURL_POLL_IN, EVFILT_READ); @@ -212,7 +208,10 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in sock_info->action = what; - if (what == CURL_POLL_REMOVE) pfree(sock_info); + if (what == CURL_POLL_REMOVE) { + EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, NULL); + pfree(sock_info); + } Assert(count <= 2); From bc6aa4fa38fa9a61529b72d9008ed0381e599d93 Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Tue, 7 Jul 2026 14:21:40 +0530 Subject: [PATCH 6/8] refactor: remove unnecessary macro --- src/errors.h | 7 ------- src/event.c | 8 ++++---- 2 files changed, 4 insertions(+), 11 deletions(-) diff --git a/src/errors.h b/src/errors.h index 5b07530..1831fb2 100644 --- a/src/errors.h +++ b/src/errors.h @@ -19,13 +19,6 @@ ereport(ERROR, errmsg("Could not curl_multi_setopt(%s)", #opt)); \ } while (0) -#define EREPORT_CURL_MULTI_ASSIGN(multi_handle, sockfd, sockp) \ - do { \ - CURLMcode mresult = curl_multi_assign(multi_handle, sockfd, sockp); \ - if (mresult != CURLM_OK) \ - ereport(ERROR, errmsg("Could not curl_multi_assign(%s)", curl_multi_strerror(mresult))); \ - } while (0) - #define EREPORT_CURL_SLIST_APPEND(list, str) \ do { \ struct curl_slist *new_list = curl_slist_append(list, str); \ diff --git a/src/event.c b/src/event.c index fe8b9f7..c369f25 100644 --- a/src/event.c +++ b/src/event.c @@ -94,10 +94,10 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in int epoll_op; if (!socketp) { epoll_op = EPOLL_CTL_ADD; - EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, &socketp_marker); + EREPORT_MULTI(curl_multi_assign(wstate->curl_mhandle, sockfd, &socketp_marker)); } else if (what == CURL_POLL_REMOVE) { epoll_op = EPOLL_CTL_DEL; - EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, NULL); + EREPORT_MULTI(curl_multi_assign(wstate->curl_mhandle, sockfd, NULL)); } else { epoll_op = EPOLL_CTL_MOD; } @@ -200,7 +200,7 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in sock_info = palloc(sizeof(SocketInfo)); sock_info->sockfd = sockfd; sock_info->action = CURL_POLL_NONE; - EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, sock_info); + EREPORT_MULTI(curl_multi_assign(wstate->curl_mhandle, sockfd, sock_info)); } UPDATE_FILTER(CURL_POLL_IN, EVFILT_READ); @@ -209,7 +209,7 @@ int multi_socket_cb(__attribute__((unused)) CURL *easy, curl_socket_t sockfd, in sock_info->action = what; if (what == CURL_POLL_REMOVE) { - EREPORT_CURL_MULTI_ASSIGN(wstate->curl_mhandle, sockfd, NULL); + EREPORT_MULTI(curl_multi_assign(wstate->curl_mhandle, sockfd, NULL)); pfree(sock_info); } From b01fedc00bbdb6f47d7845b90be241f6c5ad782f Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Tue, 7 Jul 2026 14:27:59 +0530 Subject: [PATCH 7/8] refactor: update a comment --- src/event.h | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/event.h b/src/event.h index 3e992ea..bc6e5f0 100644 --- a/src/event.h +++ b/src/event.h @@ -24,10 +24,9 @@ typedef struct epoll_event event; # include typedef struct kevent event; -// Diffs `what` (desired filter interest) against `sock_info->action` (currently -// registered interest) for a single poll bit / kqueue filter pair, and appends an -// EV_ADD or EV_DELETE to `ev`/`count` only when that filter's state actually changed. -// Relies on `what`, `sock_info`, `sockfd`, `ev`, and `count` from the caller's scope. +// Sets the correct filters in ev array by comparing the existing and desired actions. +// Adds the filter if the desired action is set and the existing action is unset. +// Removes the filter if the desired action is unset and the existing action is set. #define UPDATE_FILTER(poll_bit, filter) \ do { \ if ((what & (poll_bit)) && !(sock_info->action & (poll_bit))) \ From 0bb21e77d12f30d397194a92575e4ad2add243ac Mon Sep 17 00:00:00 2001 From: Raminder Singh Date: Tue, 7 Jul 2026 15:04:30 +0530 Subject: [PATCH 8/8] chore: fix formatting --- src/event.c | 2 +- src/event.h | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/event.c b/src/event.c index c369f25..7bc0a04 100644 --- a/src/event.c +++ b/src/event.c @@ -2,8 +2,8 @@ #include #include -#include "pg_prelude.h" #include "curl_prelude.h" +#include "pg_prelude.h" #include "errors.h" #include "event.h" diff --git a/src/event.h b/src/event.h index bc6e5f0..c0a8dc0 100644 --- a/src/event.h +++ b/src/event.h @@ -27,7 +27,7 @@ typedef struct kevent event; // Sets the correct filters in ev array by comparing the existing and desired actions. // Adds the filter if the desired action is set and the existing action is unset. // Removes the filter if the desired action is unset and the existing action is set. -#define UPDATE_FILTER(poll_bit, filter) \ +# define UPDATE_FILTER(poll_bit, filter) \ do { \ if ((what & (poll_bit)) && !(sock_info->action & (poll_bit))) \ EV_SET(&ev[count++], sockfd, (filter), EV_ADD, 0, 0, sock_info); \