From 1f6e5cde6f174024537f11ffced254b065b4644c Mon Sep 17 00:00:00 2001 From: Jacob Champion Date: Wed, 20 May 2026 01:39:04 -0700 Subject: [PATCH 3/3] WIP: libpq: Implement async DNS, with green tests Via the altsock. Appears to be working in the happy path...? libpq/004_load_balance_dns Still many TODOs. The async cleanup for authn and DNS needs to be refactored/merged as appropriate. --- src/common/ip.c | 270 +++++++++++++++++++++++++++++- src/include/common/ip.h | 5 + src/interfaces/libpq/fe-connect.c | 140 ++++++++++++++-- src/interfaces/libpq/libpq-fe.h | 1 + src/interfaces/libpq/libpq-int.h | 3 +- 5 files changed, 405 insertions(+), 14 deletions(-) diff --git a/src/common/ip.c b/src/common/ip.c index a5ac7293da3..fb5a85c77c1 100644 --- a/src/common/ip.c +++ b/src/common/ip.c @@ -23,6 +23,10 @@ #include "postgres_fe.h" #endif +#if defined(HAVE_GETADDRINFO_A) +#define ASYNC_LOOKUP_SUPPORTED +#endif + #include #include #include @@ -32,9 +36,13 @@ #include #include -#include "common/ip.h" - +#ifdef ASYNC_LOOKUP_SUPPORTED +#include +#include +#include +#endif +#include "common/ip.h" static int getaddrinfo_unix(const char *path, const struct addrinfo *hintsp, @@ -93,6 +101,264 @@ pg_getaddrinfo_all(const char *hostname, const char *servname, return rc; } +struct async_lookup_ctx +{ + atomic_int_fast8_t refcnt; /* reference count for the allocation */ + + struct gaicb cb; /* the getattrinfo() call description */ + char *hostname; + char *servname; + struct addrinfo *hintp; + + int self_pipe; /* write end of the pipe waking up select() */ +}; + +static void +release_async_lookup_ctx(struct async_lookup_ctx *ctx) +{ + if (atomic_fetch_sub(&ctx->refcnt, 1) != 1) + return; + + free(ctx->hostname); + free(ctx->servname); + free(ctx->hintp); + free(ctx); +} + +static void +lookup_finished(union sigval val) +{ + struct async_lookup_ctx *ctx = val.sival_ptr; + + close(ctx->self_pipe); /* wake up any select()ors */ + + release_async_lookup_ctx(ctx); +} + +/* + * pg_getaddrinfo_all_async - get address info for Unix, IPv4 and IPv6 sockets, + * possibly asynchronously + * + * If async DNS is supported for the given arguments on this platform, address + * lookup will be queued in the background. If successful, *async will be set + * to a non-NULL pointer, which must be passed to pg_getaddrinfo_all_finish() to + * retrieve the results. Otherwise, *async will be set to NULL and the function + * will behave identically to pg_getaddrinfo_all(). + * + * (TODO: intentionally fall back to sync if queuing fails?) + * + * As with pg_getaddrinfo_all(), hintp is mandatory. + * + * self_pipe must be the write side of an fd pair that will be used to wake up + * any select/poll clients once resolution completes. If the return code is zero + * and *async is set to non-NULL upon return, then ownership of this file + * descriptor has been passed to a background thread; it MUST NOT be written to, + * duplicated, or closed by the caller. (If the return code is nonzero, or if + * *async is NULL on return, ownership remains with the caller: the request has + * either completed or failed synchronously, and nothing needs to write to the + * pipe in that case.) + */ +int +pg_getaddrinfo_all_async(const char *hostname, const char *servname, + const struct addrinfo *hintp, struct addrinfo **result, + int self_pipe, void **async) +{ +#ifdef ASYNC_LOOKUP_SUPPORTED + int rc; + struct async_lookup_ctx *ctx; + struct gaicb *cb_list[1]; + struct sigevent sev = {0}; + pthread_attr_t attrs; + + *result = NULL; + *async = NULL; + + if (hintp->ai_family == AF_UNIX) + { + /* AF_UNIX lookup is nonblocking; defer to pg_getaddrinfo_all(). */ + return pg_getaddrinfo_all(hostname, servname, hintp, result); + } + + /* ctx holds all our asynchronous state. */ + ctx = calloc(1, sizeof(struct async_lookup_ctx)); + if (!ctx) + return EAI_MEMORY; + + /* (match pg_getaddrinfo_all behavior) */ + hostname = (!hostname || hostname[0] == '\0') ? NULL : hostname; + + /* Make copies of our inputs, since they may be stack-allocated. */ + if (hostname) + ctx->hostname = strdup(hostname); + if (servname) + ctx->servname = strdup(servname); + ctx->hintp = malloc(sizeof(*hintp)); + + if (!ctx->hostname || !ctx->servname || !ctx->hintp) + { + rc = EAI_MEMORY; + goto fail; + } + + memcpy(ctx->hintp, hintp, sizeof(*hintp)); + + /* + * ctx->cb corresponds to the getaddrinfo() arguments. ctx->cb.ar_result + * will be set upon completion and doesn't need to be initialized. + */ + ctx->cb.ar_name = ctx->hostname; + ctx->cb.ar_service = ctx->servname; + ctx->cb.ar_request = ctx->hintp; + + /* The notification thread will close this, if async queueing succeeds. */ + ctx->self_pipe = self_pipe; + + /* Set the refcount to 2 (pg_getaddrinfo_all_finish + lookup_finished). */ + atomic_init(&ctx->refcnt, 2); + + cb_list[0] = &ctx->cb; + + /* + * Set up lookup_finished as our notification callback. It will be called + * from the background, on an unspecified thread. + */ + sev.sigev_notify = SIGEV_THREAD; + sev.sigev_value.sival_ptr = ctx; + sev.sigev_notify_function = lookup_finished; + + /* + * XXX: possibly-pointless paranoia. getaddrinfo_a takes a list of + * optional thread attributes for the SIGEV_THREAD notification mode. The + * default list of attributes isn't documented (but it is *not* the same + * as the default pthread_attr_t!). In particular, we want to make sure + * that these tiny ephemeral threads are detached, not joinable. + * + * In practice, if you pass NULL, glibc does indeed assume that you want + * detachable threads as of 2026. But it feels dangerous to implicitly + * rely on that very important detail... + * + * TODO: possibly pull down the stack size? Anything else? + */ + if (pthread_attr_init(&attrs) + || pthread_attr_setdetachstate(&attrs, PTHREAD_CREATE_DETACHED)) + { + /* + * Documented to be impossible in the glibc implementation, but if the + * impossible happens anyway, errno should contain the failure mode. + */ + rc = EAI_SYSTEM; + goto fail; + } + + sev.sigev_notify_attributes = &attrs; + + /* Setup complete; queue our async request. */ + rc = getaddrinfo_a(GAI_NOWAIT, cb_list, lengthof(cb_list), &sev); + if (rc) + goto fail; /* TODO: it may be useful to fall back? */ + + /* + * Past this point, we race against the notification thread, which touches + * the contents of ctx. Nothing is allowed to free ctx without going + * through the refcount, and nothing is allowed to touch self_pipe except + * the notification thread. + */ + *async = ctx; + return 0; + +fail: + free(ctx->hostname); + free(ctx->servname); + free(ctx->hintp); + free(ctx); + + return rc; + +#else /* !ASYNC_LOOKUP_SUPPORTED */ + *async = NULL; + return pg_getaddrinfo_all(hostname, servname, hintp, result); +#endif +} + +/* + * Finish a started call to pg_getaddrinfo_all_async(). This is only relevant if + * the async pointer was set by that call, but it MUST be called to release the + * allocated resources, even if the database connection has failed for some + * other reason. That resource cleanup may complete at some unspecified point in + * the future; we don't wait for it here. + * + * async is the pointer that was provided by pg_getaddrinfo_all_async(). Its + * contents must not be touched after this call, because a concurrent thread may + * free it without warning. + * + * If the background lookup request has finished, the (optional) result array + * will be pointed to the fetched information. The return code matches the + * behavior of getaddrinfo: zero on success, or a non-zero EAI_* code on + * failure. + * + * If the request hasn't finished by this point, we will attempt to cancel it. + * (Even if the request cannot be canceled, the outstanding resources will be + * released once it completes.) *result will be set to NULL, and a relevant + * EAI_* code is returned. + * + * TODO: test the failure mode where many outstanding uncancellable requests + * pile up (high DNS latency?) + */ +int +pg_getaddrinfo_all_finish(void *async, struct addrinfo **result) +{ + struct async_lookup_ctx *ctx = async; + int rc; + + if (result) + *result = NULL; + + /* First figure out the current state of things. */ + rc = gai_cancel(&ctx->cb); + switch (rc) + { + case EAI_CANCELED: + + /* + * We have successfully removed the DNS request from the queue. + * The notification callback will still be invoked and we are + * actively racing against it now. + * + * TODO: is this true? *Will* the callback be invoked? + */ + Assert(false); /* TODO when does this happen? */ + break; + + case EAI_NOTCANCELED: + + /* + * The DNS request is already in flight. We are racing against + * both its manipulation of ctx->cb and the notification thread + * (once lookup completes), but we won't wait for its results. + */ + break; + + case EAI_ALLDONE: + + /* + * The request has finished, and we can take a look at the result. + * But we're *still* racing against the notification thread! We + * must not assume that it's already done with the ctx pointer. + */ + rc = gai_error(&ctx->cb); + if (!rc && result) + *result = ctx->cb.ar_result; + break; + + default: + /* Still want to clean up, but this should not be possible. */ + Assert(false); + } + + release_async_lookup_ctx(ctx); + + return rc; +} /* * pg_freeaddrinfo_all - free addrinfo structures for IPv4, IPv6, or Unix diff --git a/src/include/common/ip.h b/src/include/common/ip.h index 12444b79e28..77e9eb7a674 100644 --- a/src/include/common/ip.h +++ b/src/include/common/ip.h @@ -23,6 +23,11 @@ extern int pg_getaddrinfo_all(const char *hostname, const char *servname, const struct addrinfo *hintp, struct addrinfo **result); +extern int pg_getaddrinfo_all_async(const char *hostname, const char *servname, + const struct addrinfo *hintp, + struct addrinfo **result, pgsocket altsock, + void **async); +extern int pg_getaddrinfo_all_finish(void *async, struct addrinfo **result); extern void pg_freeaddrinfo_all(int hint_ai_family, struct addrinfo *ai); extern int pg_getnameinfo_all(const struct sockaddr_storage *addr, int salen, diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 39241002998..4913cd44ca5 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -466,7 +466,8 @@ static PGPing internal_ping(PGconn *conn); static void pqFreeCommandQueue(PGcmdQueueEntry *queue); static bool fillPGconn(PGconn *conn, PQconninfoOption *connOptions); static void freePGconn(PGconn *conn); -static int resolve_next_host(PGconn *conn); +static int resolve_next_host(PGconn *conn, bool *async); +static int finish_async_resolution(PGconn *conn); static void release_conn_addrinfo(PGconn *conn); static int store_conn_addrinfo(PGconn *conn, struct addrinfo *addrlist); static void sendTerminateConn(PGconn *conn); @@ -572,6 +573,11 @@ pqDropConnection(PGconn *conn, bool flushInput) conn->async_auth = NULL; /* cleanup_async_auth() should have done this, but make sure */ conn->altsock = PGINVALID_SOCKET; + + /* + * TODO gotta clean up async DNS too, but cleanup_async_auth needs to be + * generalized + */ #ifdef ENABLE_GSS { OM_uint32 min_s; @@ -2721,6 +2727,8 @@ setTCPUserTimeout(PGconn *conn) int pqConnectDBStart(PGconn *conn) { + int ret; + if (!conn) return 0; @@ -2772,8 +2780,11 @@ pqConnectDBStart(PGconn *conn) * asynchronous startup process. However, we must run it once here, * because callers expect a success return from this routine to mean that * we are in PGRES_POLLING_WRITING connection state. + * + * TODO: why writing? */ - if (PQconnectPoll(conn) == PGRES_POLLING_WRITING) + ret = PQconnectPoll(conn); + if (ret == PGRES_POLLING_WRITING || ret == PGRES_POLLING_READING) return 1; connect_errReturn: @@ -2913,7 +2924,8 @@ pqConnectDBComplete(PGconn *conn) * o If you call PQtrace, ensure that the stream object into which you trace * will not block. * o If you do not supply an IP address for the remote host (i.e. you - * supply a host name instead) then PQconnectStart will block on + * supply a host name instead) and asynchronous resolution is not + * implemented for your platform, then PQconnectStart will block on * getaddrinfo. You will be fine if using Unix sockets (i.e. by * supplying neither a host name nor a host address). * o If your backend wants to use Kerberos authentication then you must @@ -2964,6 +2976,11 @@ PQconnectPoll(PGconn *conn) break; } + /* Also a reading state, but handled separately */ + case CONNECTION_AWAITING_HOST: + /* TODO */ + break; + /* These are writing states, so we just proceed. */ case CONNECTION_STARTED: case CONNECTION_MADE: @@ -2999,21 +3016,46 @@ keep_going: /* We will come back to here until there is conn->try_next_addr = false; } - /* Time to advance to next connhost[] entry? */ + /* + * Time to advance to next connhost[] entry, or finish an asynchronous + * lookup? + */ if (conn->try_next_host) { int ret; + bool async = false; /* - * Look up info for the new host. On failure, log the problem in - * conn->errorMessage, then loop around to try the next host. (Note - * we don't clear try_next_host until we've succeeded.) + * Look up, or finish the lookup of, info for the new host. On + * failure, log the problem in conn->errorMessage, then loop around to + * try the next host. (Note we don't clear try_next_host until we've + * succeeded.) */ - ret = resolve_next_host(conn); + if (conn->status != CONNECTION_AWAITING_HOST) + ret = resolve_next_host(conn, &async); + else + ret = finish_async_resolution(conn); + if (ret > 0) /* non-fatal error; continue */ + { + conn->status = CONNECTION_NEEDED; /* TODO maybe just loop here + * instead */ goto keep_going; + } else if (ret < 0) /* internal error, alloc failure, etc. */ goto error_return; + else if (async) + { + /* + * Resolution is continuing in the background. Assert that we'll + * get back to this point safely. + */ + Assert(conn->altsock != PGINVALID_SOCKET); + Assert(!conn->try_next_addr && conn->try_next_host); + + conn->status = CONNECTION_AWAITING_HOST; + return PGRES_POLLING_READING; + } /* * If random load balancing is enabled we shuffle the addresses. @@ -4606,6 +4648,7 @@ keep_going: /* We will come back to here until there is goto keep_going; } + case CONNECTION_AWAITING_HOST: /* handled earlier, not here */ default: libpq_append_conn_error(conn, "invalid connection state %d, probably indicative of memory corruption", @@ -5104,7 +5147,7 @@ pqReleaseConnHosts(PGconn *conn) * error. */ static int -resolve_next_host(PGconn *conn) +resolve_next_host(PGconn *conn, bool *async) { pg_conn_host *ch; struct addrinfo hint; @@ -5113,6 +5156,8 @@ resolve_next_host(PGconn *conn) int ret; char portstr[MAXPGPATH]; + *async = false; + if (conn->whichhost + 1 < conn->nconnhost) conn->whichhost++; else @@ -5170,14 +5215,47 @@ resolve_next_host(PGconn *conn) switch (ch->type) { case CHT_HOST_NAME: - ret = pg_getaddrinfo_all(ch->host, portstr, &hint, - &addrlist); + int pipefds[2]; + + if (pipe(pipefds) < 0) + { + libpq_append_conn_error(conn, "could not create self-pipe: %m"); + return 1; + } + + ret = pg_getaddrinfo_all_async(ch->host, portstr, &hint, &addrlist, + pipefds[1], &conn->async_dns_ctx); + + if (conn->async_dns_ctx) + { + Assert(ret == 0); + + /* + * Request has been queued successfully. Set up the altsock to + * wake up clients of PQconnectPoll(). Ownership of pipefds[1] + * has passed into the background. + */ + conn->altsock = pipefds[0]; + *async = true; + return 0; + } + else + { + /* + * Synchronous return -- whether success or failure -- means + * we have no need for an altsock. + */ + close(pipefds[0]); + close(pipefds[1]); + } + if (ret || !addrlist) { libpq_append_conn_error(conn, "could not translate host name \"%s\" to address: %s", ch->host, gai_strerror(ret)); return 1; } + break; case CHT_HOST_ADDRESS: @@ -5230,6 +5308,46 @@ resolve_next_host(PGconn *conn) return 0; } +/* + * Finishes an asynchronous call to resolve_next_host(), storing the results in + * conn->addrs. The return value has the same meaning as resolve_next_host(). + */ +static int +finish_async_resolution(PGconn *conn) +{ + struct addrinfo *addrlist; + int ret; + + Assert(conn->altsock != PGINVALID_SOCKET); + Assert(conn->async_dns_ctx); + + /* + * Our side of the altsock can be released (the other side was closed as + * the wakeup signal). + */ + close(conn->altsock); + conn->altsock = PGINVALID_SOCKET; + + /* Retrieve the results. */ + ret = pg_getaddrinfo_all_finish(conn->async_dns_ctx, &addrlist); + conn->async_dns_ctx = NULL; /* we must not touch this now */ + + if (ret || !addrlist) + { + libpq_append_conn_error(conn, "could not translate host name \"%s\" to address: %s", + "TODO" /* FIXME */ , gai_strerror(ret)); + return 1; + } + + /* Store a copy of the addrlist, as in resolve_next_host(). */ + ret = store_conn_addrinfo(conn, addrlist); + pg_freeaddrinfo_all(AF_INET /* XXX */ , addrlist); + if (ret) + return -1; /* message already logged */ + + return 0; +} + /* * store_conn_addrinfo * - copy addrinfo to PGconn object diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index 8ecb9b4a4c7..2e8a90ce5ac 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -114,6 +114,7 @@ typedef enum * started. */ CONNECTION_AUTHENTICATING, /* Authentication is in progress with some * external system. */ + CONNECTION_AWAITING_HOST, /* Waiting for a getaddrinfo/DNS response. */ } ConnStatusType; typedef enum diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 3f921207a14..a3059b36ca3 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -527,7 +527,8 @@ struct pg_conn * know which auth response we're * sending */ - /* Callbacks for external async authentication */ + /* Support for external async requests (DNS, authn, etc.) */ + void *async_dns_ctx; PostgresPollingStatusType (*async_auth) (PGconn *conn); void (*cleanup_async_auth) (PGconn *conn); pgsocket altsock; /* alternative socket for client to poll */ -- 2.43.0