Fix cancellation issues in the real-time executor (#1905)

src/backend/distributed/connection/remote_commands.c

@@ -182,7 +182,8 @@ NonblockingForgetResults(MultiConnection *connection)
 		}
 
 		result = PQgetResult(pgConn);
-		if (PQresultStatus(result) == PGRES_COPY_IN)
+		if (PQresultStatus(result) == PGRES_COPY_IN ||
+			PQresultStatus(result) == PGRES_COPY_OUT)
 		{
 			/* in copy, can't reliably recover without blocking */
 			return false;

src/backend/distributed/executor/multi_client_executor.c

@@ -924,15 +924,11 @@ MultiClientWait(WaitInfo *waitInfo)
 		{
 			/*
 			 * Signals that arrive can interrupt our poll(). In that case just
-			 * check for interrupts, and try again. Every other error is
+			 * return. Every other error is unexpected and treated as such.
-			 * unexpected and treated as such.
 			 */
 			if (errno == EAGAIN || errno == EINTR)
 			{
-				CHECK_FOR_INTERRUPTS();
+				return;
-
-				/* maximum wait starts at max again, but that's ok, it's just a stopgap */
-				continue;
 			}
 			else
 			{

src/test/regress/expected/multi_real_time_transaction.out

@@ -291,6 +291,64 @@ SELECT * FROM co_test_table;
 (1 row)
 
 ROLLBACK;
+-- Test cancelling behaviour. See https://github.com/citusdata/citus/pull/1905.
+-- Repeating it multiple times to increase the chance of failure before PR #1905.
+SET client_min_messages TO ERROR;
+alter system set deadlock_timeout TO '1ms';
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+SET client_min_messages TO DEFAULT;
+alter system set deadlock_timeout TO DEFAULT;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
 DROP SCHEMA multi_real_time_transaction CASCADE;
 NOTICE:  drop cascades to 4 other objects
 DETAIL:  drop cascades to table test_table

src/test/regress/expected/multi_real_time_transaction_0.out

@@ -299,6 +299,64 @@ SELECT * FROM co_test_table;
 (1 row)
 
 ROLLBACK;
+-- Test cancelling behaviour. See https://github.com/citusdata/citus/pull/1905.
+-- Repeating it multiple times to increase the chance of failure before PR #1905.
+SET client_min_messages TO ERROR;
+alter system set deadlock_timeout TO '1ms';
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ERROR:  canceling the transaction since it was involved in a distributed deadlock
+ROLLBACK;
+SET client_min_messages TO DEFAULT;
+alter system set deadlock_timeout TO DEFAULT;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
 DROP SCHEMA multi_real_time_transaction CASCADE;
 NOTICE:  drop cascades to 4 other objects
 DETAIL:  drop cascades to table test_table

src/test/regress/sql/multi_real_time_transaction.sql

@@ -176,4 +176,46 @@ DELETE FROM test_table where id = 1 or id = 3;
 SELECT * FROM co_test_table;
 ROLLBACK;
 
+-- Test cancelling behaviour. See https://github.com/citusdata/citus/pull/1905.
+-- Repeating it multiple times to increase the chance of failure before PR #1905.
+SET client_min_messages TO ERROR;
+alter system set deadlock_timeout TO '1ms';
+SELECT pg_reload_conf();
+
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+BEGIN;
+SELECT id, pg_advisory_lock(15) FROM test_table;
+ROLLBACK;
+
+SET client_min_messages TO DEFAULT;
+alter system set deadlock_timeout TO DEFAULT;
+SELECT pg_reload_conf();
+
+
 DROP SCHEMA multi_real_time_transaction CASCADE;