Fix the problem with DROP ROLE in the receiver pool of workers.

Author: danolivo

Makefile

@@ -34,7 +34,7 @@ submake-regress:
 	$(MAKE) -C $(top_builddir)/src/test/regress tablespace-setup
 
 ifndef MTM_ALL
-#PROVE_TESTS=t/0000_precommit.pl t/000_regress.pl t/000_truncate.pl t/000_deadlock.pl t/001_basic_recovery.pl
+PROVE_TESTS=t/0000_precommit.pl t/000_regress.pl t/000_truncate.pl t/000_deadlock.pl t/001_basic_recovery.pl
 endif
 check: temp-install submake-regress
 	$(prove_check)

src/bgwpool.c

@@ -198,7 +198,8 @@ BgwPoolMainLoop(BgwPool* poolDesc)
 			LWLockRelease(&poolDesc->lock);
 			ConditionVariableSleep(&Mtm->receiver_barrier_cv, PG_WAIT_EXTENSION);
 			ConditionVariableCancelSleep();
-			LWLockAcquire(&poolDesc->lock, LW_EXCLUSIVE);
+			continue;
+			//LWLockAcquire(&poolDesc->lock, LW_EXCLUSIVE);
 		}
 
 		size = *(int *) &queue[poolDesc->head];
@@ -360,7 +361,7 @@ BgwPoolExecute(BgwPool* poolDesc, void* work, int size, MtmReceiverContext *ctx)
 			if (poolDesc->tail == poolDesc->size)
 				poolDesc->tail = 0;
 
-			ConditionVariableBroadcast(&poolDesc->available_cv);
+			ConditionVariableSignal(&poolDesc->available_cv);
 			break;
 		}
 		else

src/ddl.c

@@ -1141,8 +1141,8 @@ MtmApplyDDLMessage(const char *messageBody, bool transactional)
 	/*
 	 * Set proper context for running receiver DDL.
 	 *
-	 * MtmProcessUtilityReciever() will work only when DDLApplyInProgress is
-	 * set ti true. Captured non-transactional DDL will be placed into
+	 * MtmProcessUtilityReceiver() will work only when DDLApplyInProgress is
+	 * set to true. Captured non-transactional DDL will be placed into
 	 * MtmCapturedDDL. In case of error both of this variables are reset by
 	 * MtmDDLResetApplyState().
 	 */
@@ -1223,7 +1223,6 @@ MtmApplyDDLMessage(const char *messageBody, bool transactional)
 
 			default:
 				Assert(false);
-
 		}
 
 		pfree(MtmCapturedDDL);

src/dmq.c

@@ -45,7 +45,6 @@
 #include "storage/ipc.h"
 #include "tcop/tcopprot.h"
 #include "utils/dynahash.h"
-#include "utils/inval.h"
 #include "utils/ps_status.h"
 
 #define DMQ_MQ_SIZE  ((Size) 65536)
@@ -1152,7 +1151,6 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 
 		// XXX: is it enough?
 		CHECK_FOR_INTERRUPTS();
-		AcceptInvalidationMessages();
 
 		if (dmq_now() - last_message_at > recv_timeout)
 		{

src/pglogical_apply.c

@@ -962,14 +962,17 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid, MtmRecei
 			strncpy(gid, pq_getmsgstring(in), sizeof gid);
 			MtmBeginSession(origin_node);
 
-			if (!IsTransactionState()) {
+			if (!IsTransactionState())
+			{
 				StartTransactionCommand();
 				SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED);
 				CommitTransactionCommand();
+
 				MemoryContextSwitchTo(MtmApplyContext);
-			} else {
-				SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED);
 			}
+			else
+				SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED);
+
 			mtm_log(MtmTxFinish, "TXFINISH: %s precommitted", gid);
 
 			if (receiver_ctx->parallel_allowed)
@@ -1030,7 +1033,6 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid, MtmRecei
 			current_gtid->my_xid = InvalidTransactionId;
 			MtmDeadlockDetectorRemoveXact(xid);
 
-
 			MtmEndSession(origin_node, true);
 
 			mtm_log(MtmApplyTrace,
@@ -1455,77 +1457,76 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
     s.len = size;
     s.maxlen = -1;
 	s.cursor = 0;
-	
+
 	if (MtmApplyContext == NULL)
-	{
 		MtmApplyContext = AllocSetContextCreate(TopMemoryContext,
 												"ApplyContext",
 												ALLOCSET_DEFAULT_SIZES);
-    }
+
 	top_context = MemoryContextSwitchTo(MtmApplyContext);
 	replorigin_session_origin = InvalidRepOriginId;
 
 	PG_TRY();
-
-	AcceptInvalidationMessages();
-	if (!receiver_mtm_cfg_valid)
 	{
-		if (receiver_mtm_cfg)
-			pfree(receiver_mtm_cfg);
+		bool inside_transaction = true;
 
-		receiver_mtm_cfg = MtmLoadConfig();
+		AcceptInvalidationMessages();
 
-		if (receiver_mtm_cfg->my_node_id == 0 ||
-			MtmNodeById(receiver_mtm_cfg, receiver_ctx->node_id) == NULL) //XXX
-			proc_exit(0);
+		/* Clear authorization settings */
+		StartTransactionCommand();
+		SetPGVariable("session_authorization", NIL, false);
+		ResetAllOptions();
+		CommitTransactionCommand();
 
-		receiver_mtm_cfg_valid = true;
-	}
+		if (!receiver_mtm_cfg_valid)
+		{
+			if (receiver_mtm_cfg)
+				pfree(receiver_mtm_cfg);
+			receiver_mtm_cfg = MtmLoadConfig();
+			if (receiver_mtm_cfg->my_node_id == 0 ||
+				MtmNodeById(receiver_mtm_cfg, receiver_ctx->node_id) == NULL) //XXX
+				proc_exit(0);
+
+			receiver_mtm_cfg_valid = true;
+		}
 
-	StartTransactionCommand();
-	SetPGVariable("session_authorization", NIL, false);
-	ResetAllOptions();
-	CommitTransactionCommand();
+		do {
+			char action = pq_getmsgbyte(&s);
 
-    {    
-		bool inside_transaction = true;
-        do { 
-            char action = pq_getmsgbyte(&s);
 			old_context = MemoryContextSwitchTo(MtmApplyContext);
-
 			mtm_log(MtmApplyTrace, "got action '%c'", action);
 
 			switch (action)
 			{
-                /* BEGIN */
-            case 'B':
-			    inside_transaction = process_remote_begin(&s, &current_gtid);
+				/* BEGIN */
+			case 'B':
+				inside_transaction = process_remote_begin(&s, &current_gtid);
 				break;
-                /* COMMIT */
-            case 'C':
-  			    close_rel(rel);
+				/* COMMIT */
+			case 'C':
+				close_rel(rel);
 				process_remote_commit(&s, &current_gtid, receiver_ctx);
 				inside_transaction = false;
-                break;
-                /* INSERT */
-            case 'I':
+				break;
+				/* INSERT */
+			case 'I':
 				Assert(rel);
-			    process_remote_insert(&s, rel);
-                break;
-                /* UPDATE */
-            case 'U':
+				process_remote_insert(&s, rel);
+				break;
+				/* UPDATE */
+			case 'U':
 				Assert(rel);
-                process_remote_update(&s, rel);
-                break;
-                /* DELETE */
-            case 'D':
+				process_remote_update(&s, rel);
+				break;
+				/* DELETE */
+			case 'D':
 				Assert(rel);
-                process_remote_delete(&s, rel);
-                break;
-            case 'R':
-  			    close_rel(rel);
-                rel = read_rel(&s, RowExclusiveLock);
-                break;
+				process_remote_delete(&s, rel);
+				break;
+			case 'R':
+				close_rel(rel);
+				rel = read_rel(&s, RowExclusiveLock);
+				break;
 			case 'F':
 			{
 				int node_id = pq_getmsgint(&s, 4);
@@ -1534,9 +1535,10 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 				spill_file = MtmOpenSpillFile(node_id, file_id);
 				break;
 			}
- 		    case '(':
+			case '(':
 			{
-			    size_t size = pq_getmsgint(&s, 4);    
+				size_t size = pq_getmsgint(&s, 4);
+
 				s.data = MemoryContextAlloc(TopMemoryContext, size);
 				save_cursor = s.cursor;
 				save_len = s.len;
@@ -1545,14 +1547,12 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 				MtmReadSpillFile(spill_file, s.data, size);
 				break;
 			}
-  		    case ')':
-			{
-  			    pfree(s.data);
+			case ')':
+				pfree(s.data);
 				s.data = work;
-  			    s.cursor = save_cursor;
+				s.cursor = save_cursor;
 				s.len = save_len;
 				break;
-			}
 			case 'N':
 			{
 				int64 next;
@@ -1565,17 +1565,15 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 				AdjustSequence(relid, next);
 				break;
 			}
-		    case '0':
-			    Assert(rel != NULL);
-			    heap_truncate_one_rel(rel);
+			case '0':
+				Assert(rel != NULL);
+				heap_truncate_one_rel(rel);
 				break;
 			case 'M':
-			{
-  			    close_rel(rel);
+				close_rel(rel);
 				rel = NULL;
 				inside_transaction = !process_remote_message(&s, receiver_ctx);
 				break;
-			}
 			case 'Z':
 			{
 				int			rc;
@@ -1647,10 +1645,6 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 	}
 	PG_END_TRY();
 
-	// Assert(s.cursor == s.len);
-	// only non-error scenario
-	// Assert(s.data == work);
-
 	if (s.data != work)
 		pfree(s.data);
 	MemoryContextSwitchTo(top_context);

src/pglogical_output.c

@@ -106,16 +106,16 @@ static bool startup_message_sent = false;
 
 void MtmOutputPluginWrite(LogicalDecodingContext *ctx, bool last_write, bool flush)
 {
-	if (flush) {
+	if (flush)
 		OutputPluginWrite(ctx, last_write);
-	}
 }
 
 void MtmOutputPluginPrepareWrite(LogicalDecodingContext *ctx, bool last_write, bool flush)
 {
-	if (!ctx->prepared_write) { 
+	if (!ctx->prepared_write)
 		OutputPluginPrepareWrite(ctx, last_write);
-	} else if (flush || ctx->out->len > OUTPUT_BUFFER_SIZE) {
+	else if (flush || ctx->out->len > OUTPUT_BUFFER_SIZE)
+	{
 		OutputPluginWrite(ctx, false);
 		OutputPluginPrepareWrite(ctx, last_write);
 	}