nocfbot.v0-0001-POC-Use-non-blocking-apis-for-frontend-tools.patch

From 0e37f6acca869ea471a840f9511adf1de856b5d2 Mon Sep 17 00:00:00 2001
From: Jelte Fennema-Nio <postgres@jeltef.nl>
Date: Sat, 13 Dec 2025 13:05:50 +0100
Subject: [PATCH v0] POC: Use non-blocking apis for frontend tools

POC by Claude Code, to see what would be needed to change all
fe_utils/cancel.c users to use non-blocking APIs. This code is an
extremely rough draft, only meant as a way to gauge feasibility of this
approach.
---
 src/bin/pgbench/pgbench.c     |  38 +-
 src/bin/psql/command.c        |   5 +-
 src/bin/psql/common.c         |  47 +-
 src/bin/psql/copy.c           |  54 +-
 src/bin/psql/large_obj.c      |  12 +-
 src/fe_utils/Makefile         |   2 +-
 src/fe_utils/cancel.c         | 914 ++++++++++++++++++++++++++++++----
 src/fe_utils/parallel_slot.c  |  45 +-
 src/fe_utils/query_utils.c    |   8 +-
 src/include/fe_utils/cancel.h |  34 +-
 10 files changed, 953 insertions(+), 206 deletions(-)

diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c
index 1dae918cc09..8ebefceeebf 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -867,7 +867,7 @@ get_table_relkind(PGconn *con, const char *table)
 	const char *sql =
 		"SELECT relkind FROM pg_catalog.pg_class WHERE oid=$1::pg_catalog.regclass";
 
-	res = PQexecParams(con, sql, 1, NULL, params, NULL, NULL, 0);
+	res = cancellable_exec_params(con, sql, 1, NULL, params, NULL, NULL, 0);
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
 		pg_log_error("query failed: %s", PQerrorMessage(con));
@@ -1490,13 +1490,13 @@ accumStats(StatsData *stats, bool skipped, double lat, double lag,
 	}
 }
 
-/* call PQexec() and exit() on failure */
+/* call cancellable_exec() and exit() on failure */
 static void
 executeStatement(PGconn *con, const char *sql)
 {
 	PGresult   *res;
 
-	res = PQexec(con, sql);
+	res = cancellable_exec(con, sql);
 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
 	{
 		pg_log_error("query failed: %s", PQerrorMessage(con));
@@ -1506,13 +1506,13 @@ executeStatement(PGconn *con, const char *sql)
 	PQclear(res);
 }
 
-/* call PQexec() and complain, but without exiting, on failure */
+/* call cancellable_exec() and complain, but without exiting, on failure */
 static void
 tryExecuteStatement(PGconn *con, const char *sql)
 {
 	PGresult   *res;
 
-	res = PQexec(con, sql);
+	res = cancellable_exec(con, sql);
 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
 	{
 		pg_log_error("%s", PQerrorMessage(con));
@@ -5059,7 +5059,7 @@ initPopulateTable(PGconn *con, const char *table, int64 base,
 	else if (n == -1)
 		pg_fatal("invalid format string");
 
-	res = PQexec(con, copy_statement);
+	res = cancellable_exec(con, copy_statement);
 
 	if (PQresultStatus(res) != PGRES_COPY_IN)
 		pg_fatal("unexpected copy in result: %s", PQerrorMessage(con));
@@ -5338,7 +5338,6 @@ runInitSteps(const char *initialize_steps)
 		pg_fatal("could not create connection for initialization");
 
 	setup_cancel_handler(NULL);
-	SetCancelConn(con);
 
 	for (step = initialize_steps; *step != '\0'; step++)
 	{
@@ -5399,7 +5398,6 @@ runInitSteps(const char *initialize_steps)
 	}
 
 	fprintf(stderr, "done in %.2f s (%s).\n", run_time, stats.data);
-	ResetCancelConn();
 	PQfinish(con);
 	termPQExpBuffer(&stats);
 }
@@ -5417,7 +5415,7 @@ GetTableInfo(PGconn *con, bool scale_given)
 	 * get the scaling factor that should be same as count(*) from
 	 * pgbench_branches if this is not a custom query
 	 */
-	res = PQexec(con, "select count(*) from pgbench_branches");
+	res = cancellable_exec(con, "select count(*) from pgbench_branches");
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
 		char	   *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
@@ -5456,17 +5454,17 @@ GetTableInfo(PGconn *con, bool scale_given)
 	 * We assume no partitioning on any failure, so as to avoid failing on an
 	 * old version without "pg_partitioned_table".
 	 */
-	res = PQexec(con,
-				 "select o.n, p.partstrat, pg_catalog.count(i.inhparent) "
-				 "from pg_catalog.pg_class as c "
-				 "join pg_catalog.pg_namespace as n on (n.oid = c.relnamespace) "
-				 "cross join lateral (select pg_catalog.array_position(pg_catalog.current_schemas(true), n.nspname)) as o(n) "
-				 "left join pg_catalog.pg_partitioned_table as p on (p.partrelid = c.oid) "
-				 "left join pg_catalog.pg_inherits as i on (c.oid = i.inhparent) "
-				 "where c.relname = 'pgbench_accounts' and o.n is not null "
-				 "group by 1, 2 "
-				 "order by 1 asc "
-				 "limit 1");
+	res = cancellable_exec(con,
+						   "select o.n, p.partstrat, pg_catalog.count(i.inhparent) "
+						   "from pg_catalog.pg_class as c "
+						   "join pg_catalog.pg_namespace as n on (n.oid = c.relnamespace) "
+						   "cross join lateral (select pg_catalog.array_position(pg_catalog.current_schemas(true), n.nspname)) as o(n) "
+						   "left join pg_catalog.pg_partitioned_table as p on (p.partrelid = c.oid) "
+						   "left join pg_catalog.pg_inherits as i on (c.oid = i.inhparent) "
+						   "where c.relname = 'pgbench_accounts' and o.n is not null "
+						   "group by 1, 2 "
+						   "order by 1 asc "
+						   "limit 1");
 
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index e6365d823ce..238d62dc86c 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -4299,7 +4299,6 @@ do_connect(enum trivalue reuse_previous_specification,
 				 */
 				PQfinish(o_conn);
 				pset.db = NULL;
-				ResetCancelConn();
 				UnsyncVariables();
 			}
 
@@ -6225,7 +6224,7 @@ lookup_object_oid(EditableObjectType obj_type, const char *desc,
 		destroyPQExpBuffer(query);
 		return false;
 	}
-	res = PQexec(pset.db, query->data);
+	res = cancellable_exec(pset.db, query->data);
 	if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) == 1)
 		*obj_oid = atooid(PQgetvalue(res, 0, 0));
 	else
@@ -6307,7 +6306,7 @@ get_create_object_cmd(EditableObjectType obj_type, Oid oid,
 		destroyPQExpBuffer(query);
 		return false;
 	}
-	res = PQexec(pset.db, query->data);
+	res = cancellable_exec(pset.db, query->data);
 	if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) == 1)
 	{
 		resetPQExpBuffer(buf);
diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c
index 2eadd391a9c..119ca99b418 100644
--- a/src/bin/psql/common.c
+++ b/src/bin/psql/common.c
@@ -382,7 +382,6 @@ CheckConnection(void)
 				PQfinish(pset.dead_conn);
 			pset.dead_conn = pset.db;
 			pset.db = NULL;
-			ResetCancelConn();
 			UnsyncVariables();
 		}
 		else
@@ -583,7 +582,7 @@ ClearOrSaveAllResults(void)
 {
 	PGresult   *result;
 
-	while ((result = PQgetResult(pset.db)) != NULL)
+	while ((result = cancellable_getresult(pset.db)) != NULL)
 		ClearOrSaveResult(result);
 }
 
@@ -681,11 +680,7 @@ PSQLexec(const char *query)
 			return NULL;
 	}
 
-	SetCancelConn(pset.db);
-
-	res = PQexec(pset.db, query);
-
-	ResetCancelConn();
+	res = cancellable_exec(pset.db, query);
 
 	if (!AcceptResult(res, true))
 	{
@@ -719,12 +714,8 @@ PSQLexecWatch(const char *query, const printQueryOpt *opt, FILE *printQueryFout,
 		return 0;
 	}
 
-	SetCancelConn(pset.db);
-
 	res = ExecQueryAndProcessResults(query, &elapsed_msec, NULL, true, min_rows, opt, printQueryFout);
 
-	ResetCancelConn();
-
 	/* Possible microtiming output */
 	if (timing)
 		PrintTiming(elapsed_msec);
@@ -943,8 +934,6 @@ HandleCopyResult(PGresult **resultp, FILE *copystream)
 	Assert(result_status == PGRES_COPY_OUT ||
 		   result_status == PGRES_COPY_IN);
 
-	SetCancelConn(pset.db);
-
 	if (result_status == PGRES_COPY_OUT)
 	{
 		success = handleCopyOut(pset.db,
@@ -973,7 +962,6 @@ HandleCopyResult(PGresult **resultp, FILE *copystream)
 							   PQbinaryTuples(*resultp),
 							   &copy_result);
 	}
-	ResetCancelConn();
 
 	/*
 	 * Replace the PGRES_COPY_OUT/IN result with COPY command's exit status,
@@ -1162,8 +1150,6 @@ SendQuery(const char *query)
 		fflush(pset.logfile);
 	}
 
-	SetCancelConn(pset.db);
-
 	transaction_status = PQtransactionStatus(pset.db);
 
 	if (transaction_status == PQTRANS_IDLE &&
@@ -1172,7 +1158,7 @@ SendQuery(const char *query)
 	{
 		PGresult   *result;
 
-		result = PQexec(pset.db, "BEGIN");
+		result = cancellable_exec(pset.db, "BEGIN");
 		if (PQresultStatus(result) != PGRES_COMMAND_OK)
 		{
 			pg_log_info("%s", PQerrorMessage(pset.db));
@@ -1190,7 +1176,7 @@ SendQuery(const char *query)
 	{
 		PGresult   *result;
 
-		result = PQexec(pset.db, "SAVEPOINT pg_psql_temporary_savepoint");
+		result = cancellable_exec(pset.db, "SAVEPOINT pg_psql_temporary_savepoint");
 		if (PQresultStatus(result) != PGRES_COMMAND_OK)
 		{
 			pg_log_info("%s", PQerrorMessage(pset.db));
@@ -1258,7 +1244,7 @@ SendQuery(const char *query)
 		{
 			PGresult   *svptres;
 
-			svptres = PQexec(pset.db, svptcmd);
+			svptres = cancellable_exec(pset.db, svptcmd);
 			if (PQresultStatus(svptres) != PGRES_COMMAND_OK)
 			{
 				pg_log_info("%s", PQerrorMessage(pset.db));
@@ -1293,9 +1279,6 @@ SendQuery(const char *query)
 
 sendquery_cleanup:
 
-	/* global cancellation reset */
-	ResetCancelConn();
-
 	/* reset \g's output-to-filename trigger */
 	if (pset.gfname)
 	{
@@ -1370,7 +1353,7 @@ DescribeQuery(const char *query, double *elapsed_msec)
 	 * anyway.  (So there's no great need to clear it when done, which is a
 	 * good thing because libpq provides no easy way to do that.)
 	 */
-	result = PQprepare(pset.db, "", query, 0, NULL);
+	result = cancellable_prepare(pset.db, "", query, 0, NULL);
 	if (PQresultStatus(result) != PGRES_COMMAND_OK)
 	{
 		pg_log_info("%s", PQerrorMessage(pset.db));
@@ -1380,7 +1363,7 @@ DescribeQuery(const char *query, double *elapsed_msec)
 	}
 	PQclear(result);
 
-	result = PQdescribePrepared(pset.db, "");
+	result = cancellable_describe_prepared(pset.db, "");
 	OK = AcceptResult(result, true) &&
 		(PQresultStatus(result) == PGRES_COMMAND_OK);
 	if (OK && result)
@@ -1428,7 +1411,7 @@ DescribeQuery(const char *query, double *elapsed_msec)
 			appendPQExpBufferStr(&buf, ") s(name, tp, tpm)");
 			PQclear(result);
 
-			result = PQexec(pset.db, buf.data);
+			result = cancellable_exec(pset.db, buf.data);
 			OK = AcceptResult(result, true);
 
 			if (timing)
@@ -1467,7 +1450,7 @@ discardAbortedPipelineResults(void)
 {
 	for (;;)
 	{
-		PGresult   *res = PQgetResult(pset.db);
+		PGresult   *res = cancellable_getresult(pset.db);
 		ExecStatusType result_status = PQresultStatus(res);
 
 		if (result_status == PGRES_PIPELINE_SYNC)
@@ -1491,7 +1474,7 @@ discardAbortedPipelineResults(void)
 			 * Fetch result to consume the end of the current query being
 			 * processed.
 			 */
-			fatal_res = PQgetResult(pset.db);
+			fatal_res = cancellable_getresult(pset.db);
 			Assert(fatal_res == NULL);
 			return res;
 		}
@@ -1759,8 +1742,8 @@ ExecQueryAndProcessResults(const char *query,
 		return 0;
 	}
 
-	/* first result */
-	result = PQgetResult(pset.db);
+	/* first result -- use cancellable wait for the potentially long wait */
+	result = cancellable_getresult(pset.db);
 	if (min_rows > 0 && PQntuples(result) < min_rows)
 	{
 		return_early = true;
@@ -1828,7 +1811,7 @@ ExecQueryAndProcessResults(const char *query,
 				result = discardAbortedPipelineResults();
 			}
 			else
-				result = PQgetResult(pset.db);
+				result = cancellable_getresult(pset.db);
 
 			/*
 			 * Get current timing measure in case an error occurs
@@ -1987,7 +1970,7 @@ ExecQueryAndProcessResults(const char *query,
 				ClearOrSaveResult(result);
 
 				/* get the next result, loop if it's PGRES_TUPLES_CHUNK */
-				result = PQgetResult(pset.db);
+				result = cancellable_getresult(pset.db);
 			} while (PQresultStatus(result) == PGRES_TUPLES_CHUNK);
 
 			/* We expect an empty PGRES_TUPLES_OK, else there's a problem */
@@ -2081,7 +2064,7 @@ ExecQueryAndProcessResults(const char *query,
 		 * to process.  We need to do that to check whether this is the last.
 		 */
 		if (PQpipelineStatus(pset.db) == PQ_PIPELINE_OFF)
-			next_result = PQgetResult(pset.db);
+			next_result = cancellable_getresult(pset.db);
 		else
 		{
 			/*
diff --git a/src/bin/psql/copy.c b/src/bin/psql/copy.c
index 6a8a9792e7d..fc28537acc1 100644
--- a/src/bin/psql/copy.c
+++ b/src/bin/psql/copy.c
@@ -18,6 +18,7 @@
 #include "common.h"
 #include "common/logging.h"
 #include "copy.h"
+#include "fe_utils/cancel.h"
 #include "libpq-fe.h"
 #include "pqexpbuffer.h"
 #include "prompt.h"
@@ -436,10 +437,24 @@ handleCopyOut(PGconn *conn, FILE *copystream, PGresult **res)
 	bool		OK = true;
 	char	   *buf;
 	int			ret;
+	bool		cancel_sent = false;
 
 	for (;;)
 	{
-		ret = PQgetCopyData(conn, &buf, 0);
+		/* Use async mode so we can watch for cancel interrupts */
+		ret = PQgetCopyData(conn, &buf, 1);
+
+		if (ret == 0)
+		{
+			/* No data available yet, wait for socket or cancel */
+			if (!cancellable_socket_wait(conn, &cancel_sent, false))
+			{
+				OK = false;
+				break;
+			}
+			PQconsumeInput(conn);
+			continue;
+		}
 
 		if (ret < 0)
 			break;				/* done or server/connection error */
@@ -480,7 +495,7 @@ handleCopyOut(PGconn *conn, FILE *copystream, PGresult **res)
 	 * but hasn't exited COPY_OUT state internally.  So we ignore the
 	 * possibility here.
 	 */
-	*res = PQgetResult(conn);
+	*res = cancellable_getresult(conn);
 	if (PQresultStatus(*res) != PGRES_COMMAND_OK)
 	{
 		pg_log_info("%s", PQerrorMessage(conn));
@@ -513,6 +528,10 @@ handleCopyIn(PGconn *conn, FILE *copystream, bool isbinary, PGresult **res)
 	bool		OK;
 	char		buf[COPYBUFSIZ];
 	bool		showprompt;
+	bool		cancel_sent = false;
+
+	/* Set non-blocking mode so PQputCopyData/End won't block internally */
+	PQsetnonblocking(conn, 1);
 
 	/*
 	 * Establish longjmp destination for exiting from wait-for-input. (This is
@@ -523,9 +542,10 @@ handleCopyIn(PGconn *conn, FILE *copystream, bool isbinary, PGresult **res)
 		/* got here with longjmp */
 
 		/* Terminate data transfer */
-		PQputCopyEnd(conn,
-					 (PQprotocolVersion(conn) < 3) ? NULL :
-					 _("canceled by user"));
+		cancellable_put_copy_end(conn,
+								 (PQprotocolVersion(conn) < 3) ? NULL :
+								 _("canceled by user"),
+								 &cancel_sent);
 
 		OK = false;
 		goto copyin_cleanup;
@@ -569,7 +589,7 @@ handleCopyIn(PGconn *conn, FILE *copystream, bool isbinary, PGresult **res)
 			if (buflen <= 0)
 				break;
 
-			if (PQputCopyData(conn, buf, buflen) <= 0)
+			if (cancellable_put_copy_data(conn, buf, buflen, &cancel_sent) <= 0)
 			{
 				OK = false;
 				break;
@@ -667,7 +687,7 @@ handleCopyIn(PGconn *conn, FILE *copystream, bool isbinary, PGresult **res)
 			 */
 			if (buflen >= COPYBUFSIZ - 5 || (copydone && buflen > 0))
 			{
-				if (PQputCopyData(conn, buf, buflen) <= 0)
+				if (cancellable_put_copy_data(conn, buf, buflen, &cancel_sent) <= 0)
 				{
 					OK = false;
 					break;
@@ -688,9 +708,10 @@ handleCopyIn(PGconn *conn, FILE *copystream, bool isbinary, PGresult **res)
 	 * keep the version checks just in case you're using a pre-v14 libpq.so at
 	 * runtime)
 	 */
-	if (PQputCopyEnd(conn,
-					 (OK || PQprotocolVersion(conn) < 3) ? NULL :
-					 _("aborted because of read failure")) <= 0)
+	if (cancellable_put_copy_end(conn,
+								 (OK || PQprotocolVersion(conn) < 3) ? NULL :
+								 _("aborted because of read failure"),
+								 &cancel_sent) <= 0)
 		OK = false;
 
 copyin_cleanup:
@@ -717,15 +738,20 @@ copyin_cleanup:
 	 * connection is lost.  But that's fine; it will get us out of COPY_IN
 	 * state, which is what we need.)
 	 */
-	while (*res = PQgetResult(conn), PQresultStatus(*res) == PGRES_COPY_IN)
+	while (*res = cancellable_getresult(conn), PQresultStatus(*res) == PGRES_COPY_IN)
 	{
 		OK = false;
 		PQclear(*res);
 		/* We can't send an error message if we're using protocol version 2 */
-		PQputCopyEnd(conn,
-					 (PQprotocolVersion(conn) < 3) ? NULL :
-					 _("trying to exit copy mode"));
+		cancellable_put_copy_end(conn,
+								 (PQprotocolVersion(conn) < 3) ? NULL :
+								 _("trying to exit copy mode"),
+								 &cancel_sent);
 	}
+
+	/* Restore blocking mode */
+	PQsetnonblocking(conn, 0);
+
 	if (PQresultStatus(*res) != PGRES_COMMAND_OK)
 	{
 		pg_log_info("%s", PQerrorMessage(conn));
diff --git a/src/bin/psql/large_obj.c b/src/bin/psql/large_obj.c
index 021f78e0f78..949f984ff58 100644
--- a/src/bin/psql/large_obj.c
+++ b/src/bin/psql/large_obj.c
@@ -147,9 +147,7 @@ do_lo_export(const char *loid_arg, const char *filename_arg)
 	if (!start_lo_xact("\\lo_export", &own_transaction))
 		return false;
 
-	SetCancelConn(NULL);
-	status = lo_export(pset.db, atooid(loid_arg), filename_arg);
-	ResetCancelConn();
+	status = cancellable_lo_export(pset.db, atooid(loid_arg), filename_arg);
 
 	/* of course this status is documented nowhere :( */
 	if (status != 1)
@@ -183,9 +181,7 @@ do_lo_import(const char *filename_arg, const char *comment_arg)
 	if (!start_lo_xact("\\lo_import", &own_transaction))
 		return false;
 
-	SetCancelConn(NULL);
-	loid = lo_import(pset.db, filename_arg);
-	ResetCancelConn();
+	loid = cancellable_lo_import(pset.db, filename_arg);
 
 	if (loid == InvalidOid)
 	{
@@ -245,9 +241,7 @@ do_lo_unlink(const char *loid_arg)
 	if (!start_lo_xact("\\lo_unlink", &own_transaction))
 		return false;
 
-	SetCancelConn(NULL);
-	status = lo_unlink(pset.db, loid);
-	ResetCancelConn();
+	status = cancellable_lo_unlink(pset.db, loid);
 
 	if (status == -1)
 	{
diff --git a/src/fe_utils/Makefile b/src/fe_utils/Makefile
index cbfbf93ac69..809ab21cc0c 100644
--- a/src/fe_utils/Makefile
+++ b/src/fe_utils/Makefile
@@ -17,7 +17,7 @@ subdir = src/fe_utils
 top_builddir = ../..
 include $(top_builddir)/src/Makefile.global
 
-override CPPFLAGS := -DFRONTEND -I$(libpq_srcdir) $(CPPFLAGS)
+override CPPFLAGS := -DFRONTEND -I$(libpq_srcdir) -I$(top_srcdir)/src/port $(CPPFLAGS)
 
 OBJS = \
 	archive.o \
diff --git a/src/fe_utils/cancel.c b/src/fe_utils/cancel.c
index e6b75439f56..514a8f54c82 100644
--- a/src/fe_utils/cancel.c
+++ b/src/fe_utils/cancel.c
@@ -2,9 +2,47 @@
  *
  * Query cancellation support for frontend code
  *
- * Assorted utility functions to control query cancellation with signal
- * handler for SIGINT.
+ * This module provides SIGINT/Ctrl-C handling for frontend tools that need
+ * to cancel queries running on the server.  It combines three completely
+ * independent mechanisms, any combination of which can be used by a caller:
  *
+ * 1. Server cancel request -- Often what applications need. When a query is
+ *    running, and the main thread is waiting for the result of that query in a
+ *    blocking manner, we want SIGINT/Ctrl-C to cancel that query. This can be
+ *    done by waiting for the query using cancellable_getresult() or
+ *    cancellable_exec() instead of PQgetResult() or PQexec(). These functions
+ *    wait on both the server connection and a cancel interrupt simultaneously.
+ *    When SIGINT/Ctrl-C is received the cancel request is sent to the server
+ *    from the main thread, which avoids race conditions that would occur if
+ *    the cancel were sent from a different thread.
+ *
+ * 2. CancelRequested flag -- A more involved but also much more flexible way
+ *    of cancelling. A volatile sig_atomic_t CancelRequested flag is set to
+ *    true whenever SIGINT is received. This means that the application code
+ *    can fully control what it does with this flag. The primary usecase for
+ *    this is when the application code is not blocked (indefinitely), but
+ *    needs to take an action when Ctrl-C is pressed, such as break out of a
+ *    long running loop.
+ *
+ * 3. Cancel callback -- The most complex way of handling a sigint. An optional
+ *    function pointer registered via setup_cancel_handler().  If set, it is
+ *    called directly from the signal handler, so it must be async-signal-safe.
+ *    Writing async-signal-safe code is not easy, so this is only recommended
+ *    as a last resort. psql uses this to longjmp back to the main loop when no
+ *    query is active.
+ *
+ * On Unix, the SIGINT signal handler cannot call PQcancelBlocking() directly
+ * because it is not async-signal-safe.  Instead, we use a pipe (the "self-pipe
+ * trick") to interrupt the main thread's select()/poll() call.  The signal
+ * handler writes a byte to the pipe, which makes the main thread's wait
+ * return.  The main thread then notices the cancel request and sends it to the
+ * server synchronously.
+ *
+ * On Windows, the console control handler runs in a separate OS-provided
+ * thread.  We use a Windows event object as the equivalent of the self-pipe:
+ * the console handler signals the event, and the main thread uses
+ * WaitForMultipleObjects to wait on both the socket and the cancel event
+ * simultaneously, so cancellation is noticed instantly.
  *
  * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -16,11 +54,18 @@
 
 #include "postgres_fe.h"
 
+#include <fcntl.h>
+#include <signal.h>
 #include <unistd.h>
 
-#include "common/connect.h"
+#ifndef WIN32
+#include <sys/select.h>
+#endif
+
+#include "common/logging.h"
 #include "fe_utils/cancel.h"
 #include "fe_utils/string_utils.h"
+#include "libpq/libpq-fs.h"
 
 
 /*
@@ -36,11 +81,6 @@
 		(void) rc_; \
 	} while (0)
 
-/*
- * Contains all the information needed to cancel a query issued from
- * a database connection to the backend.
- */
-static PGcancel *volatile cancelConn = NULL;
 
 /*
  * Predetermined localized error strings --- needed to avoid trying
@@ -58,169 +98,770 @@ static const char *cancel_not_sent_msg = NULL;
  */
 volatile sig_atomic_t CancelRequested = false;
 
-#ifdef WIN32
-static CRITICAL_SECTION cancelConnLock;
-#endif
-
 /*
  * Additional callback for cancellations.
  */
 static void (*cancel_callback) (void) = NULL;
 
+#ifndef WIN32
+/*
+ * On Unix, the SIGINT signal handler cannot call PQcancelBlocking() directly
+ * because it is not async-signal-safe.  Instead, we use a pipe to interrupt
+ * the main thread: the signal handler writes a byte to the pipe, and the main
+ * thread's select() call returns because the pipe's read end becomes readable.
+ * The main thread then sends the cancel request to the server.
+ */
+static int	cancel_pipe[2] = {-1, -1};
+#else
+/*
+ * On Windows, we use an event object to wake the main thread's
+ * WaitForMultipleObjects() call when Ctrl-C is pressed.
+ */
+static HANDLE cancel_event = NULL;
+#endif
+
 
 /*
- * SetCancelConn
- *
- * Set cancelConn to point to the current database connection.
+ * Send a cancel request to the given connection.
  */
 void
-SetCancelConn(PGconn *conn)
+send_cancel(PGconn *conn)
 {
-	PGcancel   *oldCancelConn;
+	PGcancelConn *cancelConn = PQcancelCreate(conn);
 
-#ifdef WIN32
-	EnterCriticalSection(&cancelConnLock);
-#endif
+	if (cancel_sent_msg)
+		write_stderr(cancel_sent_msg);
 
-	/* Free the old one if we have one */
-	oldCancelConn = cancelConn;
+	if (!PQcancelBlocking(cancelConn))
+	{
+		if (cancel_not_sent_msg)
+			write_stderr(cancel_not_sent_msg);
+		write_stderr(PQcancelErrorMessage(cancelConn));
+	}
 
-	/* be sure handle_sigint doesn't use pointer while freeing */
-	cancelConn = NULL;
+	PQcancelFinish(cancelConn);
+}
 
-	if (oldCancelConn != NULL)
-		PQfreeCancel(oldCancelConn);
 
-	cancelConn = PQgetCancel(conn);
+#ifndef WIN32
 
-#ifdef WIN32
-	LeaveCriticalSection(&cancelConnLock);
-#endif
+/*
+ * Drain any pending bytes from the cancel pipe.  This should be called
+ * before entering a cancellable wait, so that a stale signal from a
+ * previous query doesn't immediately trigger a cancel.
+ */
+static void
+drain_cancel_pipe(void)
+{
+	if (cancel_pipe[0] >= 0)
+	{
+		char		buf[16];
+		int			save_errno = errno;
+
+		while (read(cancel_pipe[0], buf, sizeof(buf)) > 0)
+			 /* loop */ ;
+
+		errno = save_errno;
+	}
 }
 
+#endif							/* !WIN32 */
+
+
 /*
- * ResetCancelConn
+ * cancellable_socket_wait
+ *
+ * Wait for the given connection's socket to become readable (or writable if
+ * for_write is true), while also watching for a cancel interrupt
+ * (SIGINT/Ctrl-C).
  *
- * Free the current cancel connection, if any, and set to NULL.
+ * If a cancel interrupt arrives during the wait, a cancel request is sent to
+ * the server.  The cancel is sent at most once per call (tracked via
+ * *cancel_sent, which the caller should initialize to false).  After sending
+ * the cancel, this function continues waiting for the socket, since the server
+ * is expected to respond with an error that the caller will process normally.
+ *
+ * Returns true if the socket is ready, false on error.
  */
-void
-ResetCancelConn(void)
+bool
+cancellable_socket_wait(PGconn *conn, bool *cancel_sent, bool for_write)
 {
-	PGcancel   *oldCancelConn;
+	int			sock = PQsocket(conn);
 
-#ifdef WIN32
-	EnterCriticalSection(&cancelConnLock);
-#endif
+	if (sock < 0)
+		return false;
+
+	for (;;)
+	{
+#ifndef WIN32
+		fd_set		sock_mask;
+		fd_set		cancel_mask;
+		fd_set	   *read_set;
+		fd_set	   *write_set;
+		int			maxFd;
+		int			rc;
+
+		FD_ZERO(&sock_mask);
+		FD_SET(sock, &sock_mask);
+		maxFd = sock;
+
+		FD_ZERO(&cancel_mask);
+		if (cancel_pipe[0] >= 0)
+		{
+			FD_SET(cancel_pipe[0], &cancel_mask);
+			if (cancel_pipe[0] > maxFd)
+				maxFd = cancel_pipe[0];
+		}
+
+		if (for_write)
+		{
+			read_set = &cancel_mask;
+			write_set = &sock_mask;
+		}
+		else
+		{
+			/*
+			 * Watch both the socket and the cancel pipe for readability.
+			 * Merge them into one fd_set.
+			 */
+			if (cancel_pipe[0] >= 0)
+				FD_SET(cancel_pipe[0], &sock_mask);
+			read_set = &sock_mask;
+			write_set = NULL;
+		}
 
-	oldCancelConn = cancelConn;
+		rc = select(maxFd + 1, read_set, write_set, NULL, NULL);
 
-	/* be sure handle_sigint doesn't use pointer while freeing */
-	cancelConn = NULL;
+		if (rc < 0)
+		{
+			if (errno == EINTR)
+				continue;
+			return false;
+		}
 
-	if (oldCancelConn != NULL)
-		PQfreeCancel(oldCancelConn);
+		/* Check cancel pipe (always in read_set for write, merged for read) */
+		if (cancel_pipe[0] >= 0 &&
+			((for_write && FD_ISSET(cancel_pipe[0], &cancel_mask)) ||
+			 (!for_write && FD_ISSET(cancel_pipe[0], &sock_mask))))
+		{
+			drain_cancel_pipe();
 
-#ifdef WIN32
-	LeaveCriticalSection(&cancelConnLock);
+			if (!*cancel_sent)
+			{
+				send_cancel(conn);
+				*cancel_sent = true;
+			}
+
+			/* Check if the socket is also ready */
+			if (for_write)
+			{
+				if (!FD_ISSET(sock, &sock_mask))
+					continue;
+			}
+			else
+			{
+				if (!FD_ISSET(sock, &sock_mask))
+					continue;
+			}
+		}
+
+		if (FD_ISSET(sock, &sock_mask))
+			return true;
+#else							/* WIN32 */
+		HANDLE		events[2];
+		WSAEVENT	sock_event;
+		DWORD		ret;
+		long		net_events;
+
+		net_events = for_write ? FD_WRITE : (FD_READ | FD_CLOSE);
+
+		sock_event = WSACreateEvent();
+		if (sock_event == WSA_INVALID_EVENT)
+			return false;
+
+		if (WSAEventSelect(sock, sock_event, net_events) != 0)
+		{
+			WSACloseEvent(sock_event);
+			return false;
+		}
+
+		events[0] = sock_event;
+		events[1] = cancel_event;
+
+		ret = WaitForMultipleObjects(cancel_event ? 2 : 1,
+									 events, FALSE, INFINITE);
+
+		WSAEventSelect(sock, sock_event, 0);
+		WSACloseEvent(sock_event);
+
+		if (ret == WAIT_OBJECT_0)
+		{
+			/* Socket is ready */
+			return true;
+		}
+		else if (ret == WAIT_OBJECT_0 + 1)
+		{
+			/* Cancel event signaled */
+			ResetEvent(cancel_event);
+
+			if (!*cancel_sent)
+			{
+				send_cancel(conn);
+				*cancel_sent = true;
+			}
+
+			/* Loop back to wait for the socket */
+			continue;
+		}
+		else
+		{
+			/* WAIT_FAILED or unexpected return */
+			return false;
+		}
+#endif							/* WIN32 */
+	}
+}
+
+
+/*
+ * cancellable_getresult
+ *
+ * Like PQgetResult, but cancellable via SIGINT.  When a cancel interrupt
+ * arrives while waiting for data from the server, a cancel request is sent
+ * to the server.  The server is then expected to respond with an error,
+ * which will be returned as the PGresult.
+ */
+PGresult *
+cancellable_getresult(PGconn *conn)
+{
+	bool		cancel_sent = false;
+
+#ifndef WIN32
+	drain_cancel_pipe();
+#else
+	if (cancel_event)
+		ResetEvent(cancel_event);
 #endif
+
+	while (PQisBusy(conn))
+	{
+		if (!cancellable_socket_wait(conn, &cancel_sent, false))
+			break;
+
+		if (!PQconsumeInput(conn))
+			break;
+	}
+
+	return PQgetResult(conn);
+}
+
+/*
+ * cancellable_exec
+ *
+ * Like PQexec, but cancellable via SIGINT.
+ */
+PGresult *
+cancellable_exec(PGconn *conn, const char *query)
+{
+	PGresult   *lastResult = NULL;
+	PGresult   *result;
+
+	if (!PQsendQuery(conn, query))
+		return PQgetResult(conn);
+
+	while ((result = cancellable_getresult(conn)) != NULL)
+	{
+		PQclear(lastResult);
+		lastResult = result;
+	}
+
+	return lastResult;
 }
 
 
 /*
- * Code to support query cancellation
+ * cancellable_exec_params
+ *
+ * Like PQexecParams, but cancellable via SIGINT.
+ */
+PGresult *
+cancellable_exec_params(PGconn *conn, const char *query,
+						int nParams, const Oid *paramTypes,
+						const char *const *paramValues,
+						const int *paramLengths, const int *paramFormats,
+						int resultFormat)
+{
+	PGresult   *lastResult = NULL;
+	PGresult   *result;
+
+	if (!PQsendQueryParams(conn, query, nParams, paramTypes,
+						   paramValues, paramLengths, paramFormats,
+						   resultFormat))
+		return PQgetResult(conn);
+
+	while ((result = cancellable_getresult(conn)) != NULL)
+	{
+		PQclear(lastResult);
+		lastResult = result;
+	}
+
+	return lastResult;
+}
+
+/*
+ * cancellable_prepare
  *
- * Note that sending the cancel directly from the signal handler is safe
- * because PQcancel() is written to make it so.  We use write() to report
- * to stderr because it's better to use simple facilities in a signal
- * handler.
+ * Like PQprepare, but cancellable via SIGINT.
+ */
+PGresult *
+cancellable_prepare(PGconn *conn, const char *stmtName,
+					const char *query, int nParams,
+					const Oid *paramTypes)
+{
+	PGresult   *lastResult = NULL;
+	PGresult   *result;
+
+	if (!PQsendPrepare(conn, stmtName, query, nParams, paramTypes))
+		return PQgetResult(conn);
+
+	while ((result = cancellable_getresult(conn)) != NULL)
+	{
+		PQclear(lastResult);
+		lastResult = result;
+	}
+
+	return lastResult;
+}
+
+/*
+ * cancellable_describe_prepared
  *
- * On Windows, the signal canceling happens on a separate thread, because
- * that's how SetConsoleCtrlHandler works.  The PQcancel function is safe
- * for this (unlike PQrequestCancel).  However, a CRITICAL_SECTION is required
- * to protect the PGcancel structure against being changed while the signal
- * thread is using it.
+ * Like PQdescribePrepared, but cancellable via SIGINT.
  */
+PGresult *
+cancellable_describe_prepared(PGconn *conn, const char *stmtName)
+{
+	PGresult   *lastResult = NULL;
+	PGresult   *result;
 
-#ifndef WIN32
+	if (!PQsendDescribePrepared(conn, stmtName))
+		return PQgetResult(conn);
+
+	while ((result = cancellable_getresult(conn)) != NULL)
+	{
+		PQclear(lastResult);
+		lastResult = result;
+	}
+
+	return lastResult;
+}
 
 /*
- * handle_sigint
+ * cancellable_flush
  *
- * Handle interrupt signals by canceling the current command, if cancelConn
- * is set.
+ * Flush pending output on the connection, waiting for the socket to become
+ * writable if needed.  Cancellable via SIGINT.
+ *
+ * Returns 0 on success, -1 on error or cancel.
  */
-static void
-handle_sigint(SIGNAL_ARGS)
+static int
+cancellable_flush(PGconn *conn, bool *cancel_sent)
 {
-	char		errbuf[256];
+	int			rc;
 
-	CancelRequested = true;
+	while ((rc = PQflush(conn)) > 0)
+	{
+		if (!cancellable_socket_wait(conn, cancel_sent, true))
+			return -1;
+	}
 
-	if (cancel_callback != NULL)
-		cancel_callback();
+	return rc;
+}
+
+/*
+ * cancellable_put_copy_data
+ *
+ * Like PQputCopyData, but cancellable via SIGINT.  The connection must be in
+ * non-blocking mode.
+ *
+ * Returns 1 on success, -1 on error or cancel.
+ */
+int
+cancellable_put_copy_data(PGconn *conn, const char *buffer, int nbytes,
+						  bool *cancel_sent)
+{
+	int			rc;
 
-	/* Send QueryCancel if we are processing a database query */
-	if (cancelConn != NULL)
+	for (;;)
 	{
-		if (PQcancel(cancelConn, errbuf, sizeof(errbuf)))
+		rc = PQputCopyData(conn, buffer, nbytes);
+
+		if (rc < 0)
+			return -1;
+
+		if (rc > 0)
 		{
-			write_stderr(cancel_sent_msg);
+			/* Data queued, flush it */
+			if (cancellable_flush(conn, cancel_sent) < 0)
+				return -1;
+			return 1;
 		}
-		else
+
+		/* rc == 0: would block, flush and retry */
+		if (cancellable_flush(conn, cancel_sent) < 0)
+			return -1;
+	}
+}
+
+/*
+ * cancellable_put_copy_end
+ *
+ * Like PQputCopyEnd, but cancellable via SIGINT.  The connection must be in
+ * non-blocking mode.
+ *
+ * Returns 1 on success, -1 on error or cancel.
+ */
+int
+cancellable_put_copy_end(PGconn *conn, const char *errormsg,
+						 bool *cancel_sent)
+{
+	int			rc;
+
+	for (;;)
+	{
+		rc = PQputCopyEnd(conn, errormsg);
+
+		if (rc < 0)
+			return -1;
+
+		if (rc > 0)
 		{
-			write_stderr(cancel_not_sent_msg);
-			write_stderr(errbuf);
+			/* End message queued, flush it */
+			if (cancellable_flush(conn, cancel_sent) < 0)
+				return -1;
+			return 1;
 		}
+
+		/* rc == 0: would block, flush and retry */
+		if (cancellable_flush(conn, cancel_sent) < 0)
+			return -1;
 	}
 }
 
 /*
- * setup_cancel_handler
+ * Helper to execute a lo_* SQL function via cancellable_exec_params and
+ * return a single integer column result.  Returns true on success, with
+ * the integer result in *result.  Returns false on failure.
+ */
+static bool
+lo_exec_int(PGconn *conn, const char *query, int nParams,
+			const char *const *paramValues, int *result)
+{
+	PGresult   *res;
+
+	res = cancellable_exec_params(conn, query, nParams, NULL,
+								  paramValues, NULL, NULL, 0);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK ||
+		PQntuples(res) != 1 || PQnfields(res) != 1)
+	{
+		PQclear(res);
+		return false;
+	}
+
+	*result = atoi(PQgetvalue(res, 0, 0));
+	PQclear(res);
+	return true;
+}
+
+#define LO_BUFSIZE 8192
+
+/*
+ * cancellable_lo_export
  *
- * Register query cancellation callback for SIGINT.
+ * Like lo_export, but each server round-trip is cancellable via SIGINT.
+ * Returns 1 on success, -1 on failure.
  */
-void
-setup_cancel_handler(void (*query_cancel_callback) (void))
+int
+cancellable_lo_export(PGconn *conn, Oid lobjId, const char *filename)
 {
-	cancel_callback = query_cancel_callback;
-	cancel_sent_msg = _("Cancel request sent\n");
-	cancel_not_sent_msg = _("Could not send cancel request: ");
+	char		oidbuf[32];
+	char		fdbuf[32];
+	char		lenbuf[32];
+	const char *params[2];
+	int			lobj;
+	int			fd;
+	int			result = 1;
+
+	/* lo_open */
+	sprintf(oidbuf, "%u", lobjId);
+	sprintf(lenbuf, "%d", INV_READ);
+	params[0] = oidbuf;
+	params[1] = lenbuf;
+	if (!lo_exec_int(conn, "SELECT lo_open($1::oid, $2::int4)",
+					 2, params, &lobj))
+		return -1;
+	if (lobj == -1)
+		return -1;
+
+	sprintf(fdbuf, "%d", lobj);
+
+	/* Open local file */
+	fd = open(filename, O_CREAT | O_WRONLY | O_TRUNC | PG_BINARY, 0666);
+	if (fd < 0)
+	{
+		pg_log_error("could not open file \"%s\": %m", filename);
+		params[0] = fdbuf;
+		cancellable_exec_params(conn, "SELECT lo_close($1::int4)",
+								1, NULL, params, NULL, NULL, 0);
+		return -1;
+	}
 
-	pqsignal(SIGINT, handle_sigint);
+	/* Read loop */
+	sprintf(lenbuf, "%d", LO_BUFSIZE);
+	params[0] = fdbuf;
+	params[1] = lenbuf;
+	for (;;)
+	{
+		PGresult   *res;
+		char	   *data;
+		int			nbytes;
+		int			written;
+
+		res = cancellable_exec_params(conn,
+									  "SELECT loread($1::int4, $2::int4)",
+									  2, NULL, params, NULL, NULL, 1);
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			PQclear(res);
+			result = -1;
+			break;
+		}
+
+		nbytes = PQgetlength(res, 0, 0);
+		if (nbytes == 0)
+		{
+			PQclear(res);
+			break;
+		}
+
+		data = PQgetvalue(res, 0, 0);
+		written = write(fd, data, nbytes);
+		PQclear(res);
+
+		if (written != nbytes)
+		{
+			pg_log_error("could not write to file \"%s\": %m", filename);
+			result = -1;
+			break;
+		}
+	}
+
+	close(fd);
+
+	/*
+	 * If the read loop failed, we're likely in an aborted transaction, so
+	 * skip lo_close to avoid overwriting the useful error message.
+	 */
+	if (result == 1)
+	{
+		int			close_result;
+
+		params[0] = fdbuf;
+		if (!lo_exec_int(conn, "SELECT lo_close($1::int4)",
+						 1, params, &close_result) ||
+			close_result != 0)
+			result = -1;
+	}
+
+	return result;
+}
+
+/*
+ * cancellable_lo_import
+ *
+ * Like lo_import, but each server round-trip is cancellable via SIGINT.
+ * Returns the OID of the new large object, or InvalidOid on failure.
+ */
+Oid
+cancellable_lo_import(PGconn *conn, const char *filename)
+{
+	char		oidbuf[32];
+	char		fdbuf[32];
+	char		modebuf[32];
+	const char *params[2];
+	int			paramLengths[2];
+	int			paramFormats[2];
+	int			fd;
+	Oid			lobjOid;
+	int			lobj;
+	int			tmp;
+	char		buf[LO_BUFSIZE];
+	int			nbytes;
+
+	/* Open local file */
+	fd = open(filename, O_RDONLY | PG_BINARY, 0666);
+	if (fd < 0)
+	{
+		pg_log_error("could not open file \"%s\": %m", filename);
+		return InvalidOid;
+	}
+
+	/* lo_creat */
+	sprintf(modebuf, "%d", INV_READ | INV_WRITE);
+	params[0] = modebuf;
+	if (!lo_exec_int(conn, "SELECT lo_creat($1::int4)",
+					 1, params, &tmp))
+	{
+		close(fd);
+		return InvalidOid;
+	}
+	lobjOid = (Oid) tmp;
+	if (lobjOid == InvalidOid)
+	{
+		close(fd);
+		return InvalidOid;
+	}
+
+	/* lo_open */
+	sprintf(oidbuf, "%u", lobjOid);
+	sprintf(modebuf, "%d", INV_WRITE);
+	params[0] = oidbuf;
+	params[1] = modebuf;
+	if (!lo_exec_int(conn, "SELECT lo_open($1::oid, $2::int4)",
+					 2, params, &lobj))
+	{
+		close(fd);
+		return InvalidOid;
+	}
+	if (lobj == -1)
+	{
+		close(fd);
+		return InvalidOid;
+	}
+
+	sprintf(fdbuf, "%d", lobj);
+
+	/* Write loop: read from file, write to large object */
+	while ((nbytes = read(fd, buf, LO_BUFSIZE)) > 0)
+	{
+		PGresult   *res;
+
+		paramFormats[0] = 0;	/* text for fd */
+		paramFormats[1] = 1;	/* binary for bytea data */
+		paramLengths[0] = 0;
+		paramLengths[1] = nbytes;
+		params[0] = fdbuf;
+		params[1] = buf;
+
+		res = cancellable_exec_params(conn,
+									  "SELECT lowrite($1::int4, $2::bytea)",
+									  2, NULL, params,
+									  paramLengths, paramFormats, 0);
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			PQclear(res);
+			close(fd);
+			return InvalidOid;
+		}
+
+		tmp = atoi(PQgetvalue(res, 0, 0));
+		PQclear(res);
+
+		if (tmp != nbytes)
+		{
+			close(fd);
+			return InvalidOid;
+		}
+	}
+
+	if (nbytes < 0)
+	{
+		pg_log_error("could not read from file \"%s\": %m", filename);
+		params[0] = fdbuf;
+		paramFormats[0] = 0;
+		cancellable_exec_params(conn, "SELECT lo_close($1::int4)",
+								1, NULL, params, NULL, NULL, 0);
+		close(fd);
+		return InvalidOid;
+	}
+
+	close(fd);
+
+	/* lo_close */
+	params[0] = fdbuf;
+	if (!lo_exec_int(conn, "SELECT lo_close($1::int4)",
+					 1, params, &tmp) ||
+		tmp != 0)
+		return InvalidOid;
+
+	return lobjOid;
+}
+
+/*
+ * cancellable_lo_unlink
+ *
+ * Like lo_unlink, but cancellable via SIGINT.
+ * Returns 1 on success, -1 on failure.
+ */
+int
+cancellable_lo_unlink(PGconn *conn, Oid lobjId)
+{
+	char		oidbuf[32];
+	const char *params[1];
+	int			result;
+
+	sprintf(oidbuf, "%u", lobjId);
+	params[0] = oidbuf;
+
+	if (!lo_exec_int(conn, "SELECT lo_unlink($1::oid)",
+					 1, params, &result))
+		return -1;
+
+	return result;
+}
+
+/*
+ * cancel_pipe_fd
+ *
+ * Return the read end of the cancel pipe, for use in select()/poll() by
+ * callers that manage their own wait loops (e.g. parallel_slot.c).
+ * Returns -1 on Windows or if the cancel handler hasn't been set up.
+ */
+int
+cancel_pipe_fd(void)
+{
+#ifndef WIN32
+	return cancel_pipe[0];
+#else
+	return -1;
+#endif
 }
 
-#else							/* WIN32 */
 
+#ifdef WIN32
+/*
+ * Console control handler for Windows.
+ *
+ * This runs in a separate thread created by the OS.  It sets
+ * CancelRequested and invokes the callback, but does not send a cancel
+ * request itself -- that is done by the main thread in the cancellable
+ * wait functions when it notices CancelRequested.
+ */
 static BOOL WINAPI
 consoleHandler(DWORD dwCtrlType)
 {
-	char		errbuf[256];
-
 	if (dwCtrlType == CTRL_C_EVENT ||
 		dwCtrlType == CTRL_BREAK_EVENT)
 	{
 		CancelRequested = true;
 
+		if (cancel_event)
+			SetEvent(cancel_event);
+
 		if (cancel_callback != NULL)
 			cancel_callback();
 
-		/* Send QueryCancel if we are processing a database query */
-		EnterCriticalSection(&cancelConnLock);
-		if (cancelConn != NULL)
-		{
-			if (PQcancel(cancelConn, errbuf, sizeof(errbuf)))
-			{
-				write_stderr(cancel_sent_msg);
-			}
-			else
-			{
-				write_stderr(cancel_not_sent_msg);
-				write_stderr(errbuf);
-			}
-		}
-
-		LeaveCriticalSection(&cancelConnLock);
-
 		return TRUE;
 	}
 	else
@@ -228,16 +869,77 @@ consoleHandler(DWORD dwCtrlType)
 		return FALSE;
 }
 
+#else							/* !WIN32 */
+
+/*
+ * Signal handler for SIGINT.  Sets CancelRequested and wakes up the main
+ * thread by writing to the pipe.
+ */
+static void
+handle_sigint(SIGNAL_ARGS)
+{
+	int			save_errno = errno;
+	char		c = 1;
+
+	CancelRequested = true;
+
+	if (cancel_callback != NULL)
+		cancel_callback();
+
+	/* Wake up the main thread's select() call */
+	if (cancel_pipe[1] >= 0)
+		(void) write(cancel_pipe[1], &c, 1);
+
+	errno = save_errno;
+}
+
+#endif							/* WIN32 */
+
+
+/*
+ * setup_cancel_handler
+ *
+ * Set up handler for SIGINT (Unix) or console events (Windows) to send a
+ * cancel request to the server.
+ *
+ * The optional callback is invoked directly from the signal handler context
+ * on every SIGINT (on Unix), so it must be async-signal-safe.
+ */
 void
-setup_cancel_handler(void (*callback) (void))
+setup_cancel_handler(void (*query_cancel_callback) (void))
 {
-	cancel_callback = callback;
+	cancel_callback = query_cancel_callback;
 	cancel_sent_msg = _("Cancel request sent\n");
 	cancel_not_sent_msg = _("Could not send cancel request: ");
 
-	InitializeCriticalSection(&cancelConnLock);
-
+#ifdef WIN32
+	cancel_event = CreateEvent(NULL, TRUE, FALSE, NULL);
+	if (cancel_event == NULL)
+	{
+		pg_log_error("could not create event for cancel: error code %lu",
+					 GetLastError());
+		exit(1);
+	}
 	SetConsoleCtrlHandler(consoleHandler, TRUE);
-}
+#else
 
-#endif							/* WIN32 */
+	/*
+	 * Create the pipe that the signal handler uses to wake the main thread.
+	 * See comment on cancel_pipe above.
+	 */
+	if (pipe(cancel_pipe) < 0)
+	{
+		pg_log_error("could not create pipe for cancel: %m");
+		exit(1);
+	}
+
+	/*
+	 * Make both ends non-blocking: the write end so that the signal handler
+	 * won't block, and the read end so that drain_cancel_pipe() won't block.
+	 */
+	fcntl(cancel_pipe[0], F_SETFL, O_NONBLOCK);
+	fcntl(cancel_pipe[1], F_SETFL, O_NONBLOCK);
+
+	pqsignal(SIGINT, handle_sigint);
+#endif
+}
diff --git a/src/fe_utils/parallel_slot.c b/src/fe_utils/parallel_slot.c
index fb9e6cc4ec1..dbf958c5656 100644
--- a/src/fe_utils/parallel_slot.c
+++ b/src/fe_utils/parallel_slot.c
@@ -60,13 +60,11 @@ consumeQueryResult(ParallelSlot *slot)
 	bool		ok = true;
 	PGresult   *result;
 
-	SetCancelConn(slot->connection);
-	while ((result = PQgetResult(slot->connection)) != NULL)
+	while ((result = cancellable_getresult(slot->connection)) != NULL)
 	{
 		if (!processQueryResult(slot, result))
 			ok = false;
 	}
-	ResetCancelConn();
 	return ok;
 }
 
@@ -81,6 +79,7 @@ select_loop(int maxFd, fd_set *workerset)
 {
 	int			i;
 	fd_set		saveSet = *workerset;
+	int			cancel_fd = cancel_pipe_fd();
 
 	if (CancelRequested)
 		return -1;
@@ -89,7 +88,8 @@ select_loop(int maxFd, fd_set *workerset)
 	{
 		/*
 		 * On Windows, we need to check once in a while for cancel requests;
-		 * on other platforms we rely on select() returning when interrupted.
+		 * on other platforms the cancel pipe makes select() return
+		 * immediately when SIGINT arrives.
 		 */
 		struct timeval *tvp;
 #ifdef WIN32
@@ -101,6 +101,16 @@ select_loop(int maxFd, fd_set *workerset)
 #endif
 
 		*workerset = saveSet;
+
+#ifndef WIN32
+		if (cancel_fd >= 0)
+		{
+			FD_SET(cancel_fd, workerset);
+			if (cancel_fd > maxFd)
+				maxFd = cancel_fd;
+		}
+#endif
+
 		i = select(maxFd + 1, workerset, NULL, NULL, tvp);
 
 #ifdef WIN32
@@ -115,10 +125,25 @@ select_loop(int maxFd, fd_set *workerset)
 
 		if (i < 0 && errno == EINTR)
 			continue;			/* ignore this */
-		if (i < 0 || CancelRequested)
-			return -1;			/* but not this */
+		if (i < 0)
+			return -1;
+
+#ifndef WIN32
+		if (cancel_fd >= 0 && FD_ISSET(cancel_fd, workerset))
+			return -1;			/* cancel requested */
+#endif
+
+		if (CancelRequested)
+			return -1;
 		if (i == 0)
 			continue;			/* timeout (Win32 only) */
+
+#ifndef WIN32
+		/* Remove the cancel pipe from the returned set */
+		if (cancel_fd >= 0)
+			FD_CLR(cancel_fd, workerset);
+#endif
+
 		break;
 	}
 
@@ -235,13 +260,15 @@ wait_on_slots(ParallelSlotArray *sa)
 	if (cancelconn == NULL)
 		return false;
 
-	SetCancelConn(cancelconn);
 	i = select_loop(maxFd, &slotset);
-	ResetCancelConn();
 
-	/* failure? */
+	/* failure or cancel? */
 	if (i < 0)
+	{
+		if (CancelRequested && cancelconn != NULL)
+			send_cancel(cancelconn);
 		return false;
+	}
 
 	for (i = 0; i < sa->numslots; i++)
 	{
diff --git a/src/fe_utils/query_utils.c b/src/fe_utils/query_utils.c
index c05fd9c21df..7e5b6676102 100644
--- a/src/fe_utils/query_utils.c
+++ b/src/fe_utils/query_utils.c
@@ -26,7 +26,7 @@ executeQuery(PGconn *conn, const char *query, bool echo)
 	if (echo)
 		printf("%s\n", query);
 
-	res = PQexec(conn, query);
+	res = cancellable_exec(conn, query);
 	if (!res ||
 		PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
@@ -51,7 +51,7 @@ executeCommand(PGconn *conn, const char *query, bool echo)
 	if (echo)
 		printf("%s\n", query);
 
-	res = PQexec(conn, query);
+	res = cancellable_exec(conn, query);
 	if (!res ||
 		PQresultStatus(res) != PGRES_COMMAND_OK)
 	{
@@ -79,9 +79,7 @@ executeMaintenanceCommand(PGconn *conn, const char *query, bool echo)
 	if (echo)
 		printf("%s\n", query);
 
-	SetCancelConn(conn);
-	res = PQexec(conn, query);
-	ResetCancelConn();
+	res = cancellable_exec(conn, query);
 
 	r = (res && PQresultStatus(res) == PGRES_COMMAND_OK);
 
diff --git a/src/include/fe_utils/cancel.h b/src/include/fe_utils/cancel.h
index e174fb83b92..1664c278a97 100644
--- a/src/include/fe_utils/cancel.h
+++ b/src/include/fe_utils/cancel.h
@@ -2,6 +2,7 @@
  *
  * Query cancellation support for frontend code
  *
+ * See cancel.c for an overview of the three cancellation mechanisms.
  *
  * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -20,13 +21,32 @@
 
 extern PGDLLIMPORT volatile sig_atomic_t CancelRequested;
 
-extern void SetCancelConn(PGconn *conn);
-extern void ResetCancelConn(void);
-
-/*
- * A callback can be optionally set up to be called at cancellation
- * time.
- */
 extern void setup_cancel_handler(void (*query_cancel_callback) (void));
 
+extern void send_cancel(PGconn *conn);
+extern PGresult *cancellable_exec(PGconn *conn, const char *query);
+extern PGresult *cancellable_exec_params(PGconn *conn, const char *query,
+										 int nParams, const Oid *paramTypes,
+										 const char *const *paramValues,
+										 const int *paramLengths,
+										 const int *paramFormats,
+										 int resultFormat);
+extern PGresult *cancellable_prepare(PGconn *conn, const char *stmtName,
+									 const char *query, int nParams,
+									 const Oid *paramTypes);
+extern PGresult *cancellable_describe_prepared(PGconn *conn,
+											   const char *stmtName);
+extern PGresult *cancellable_getresult(PGconn *conn);
+extern bool cancellable_socket_wait(PGconn *conn, bool *cancel_sent,
+									bool for_write);
+extern int	cancellable_put_copy_data(PGconn *conn, const char *buffer,
+									  int nbytes, bool *cancel_sent);
+extern int	cancellable_put_copy_end(PGconn *conn, const char *errormsg,
+									 bool *cancel_sent);
+extern int	cancellable_lo_export(PGconn *conn, Oid lobjId,
+								  const char *filename);
+extern Oid	cancellable_lo_import(PGconn *conn, const char *filename);
+extern int	cancellable_lo_unlink(PGconn *conn, Oid lobjId);
+extern int	cancel_pipe_fd(void);
+
 #endif							/* CANCEL_H */

base-commit: f95d73ed433207c4323802dc96e52f3e5553a86c
-- 
2.53.0