Some code changes after regression tests

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

@@ -160,6 +160,9 @@ BgwPoolMainLoop(BgwPool* poolDesc)
 			ProcessConfigFile(PGC_SIGHUP);
 		}
 
+		CHECK_FOR_INTERRUPTS();
+		AcceptInvalidationMessages();
+
 		// XXX: change to LWLock
 		LWLockAcquire(&poolDesc->lock, LW_EXCLUSIVE);
 
@@ -184,14 +187,17 @@ BgwPoolMainLoop(BgwPool* poolDesc)
 			LWLockRelease(&poolDesc->lock);
 
 			ConditionVariableSleep(&poolDesc->available_cv, PG_WAIT_EXTENSION);
+			ConditionVariableCancelSleep();
 			continue;
 		}
 
 		/* Wait for end of the node joining operation */
 		while (poolDesc->n_holders > 0 && !poolDesc->shutdown)
 		{
+			ConditionVariablePrepareToSleep(&Mtm->receiver_barrier_cv);
 			LWLockRelease(&poolDesc->lock);
 			ConditionVariableSleep(&Mtm->receiver_barrier_cv, PG_WAIT_EXTENSION);
+			ConditionVariableCancelSleep();
 			LWLockAcquire(&poolDesc->lock, LW_EXCLUSIVE);
 		}
 
@@ -259,7 +265,7 @@ static void BgwStartExtraWorker(BgwPool* pool)
 {
 	BackgroundWorker worker;
 	BackgroundWorkerHandle* handle;
-	MemoryContext oldcontext;
+	pid_t pid;
 
 	if (pool->nWorkers >= MtmMaxWorkers)
 		return;
@@ -268,21 +274,20 @@ static void BgwStartExtraWorker(BgwPool* pool)
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_restart_time = BGW_NEVER_RESTART;
+	worker.bgw_notify_pid = MyProcPid;
 	worker.bgw_main_arg = PointerGetDatum(pool);
 	sprintf(worker.bgw_library_name, "multimaster");
 	sprintf(worker.bgw_function_name, "BgwPoolDynamicWorkerMainLoop");
 	snprintf(worker.bgw_name, BGW_MAXLEN, "%s-dynworker-%d", pool->poolName, (int) pool->nWorkers + 1);
 
 	pool->lastDynamicWorkerStartTime = GetCurrentTimestamp();
 
-	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
-
 	if (RegisterDynamicBackgroundWorker(&worker, &handle))
 		pool->bgwhandles[pool->nWorkers++] = handle;
 	else
 		elog(WARNING, "Failed to start dynamic background worker");
 
-	MemoryContextSwitchTo(oldcontext);
+	WaitForBackgroundWorkerStartup(handle, &pid);
 }
 
 /*
@@ -365,6 +370,7 @@ BgwPoolExecute(BgwPool* poolDesc, void* work, int size, MtmReceiverContext *ctx)
 			ConditionVariablePrepareToSleep(&poolDesc->overflow_cv);
 			LWLockRelease(&poolDesc->lock);
 			ConditionVariableSleep(&poolDesc->overflow_cv, PG_WAIT_EXTENSION);
+			ConditionVariableCancelSleep();
 			LWLockAcquire(&poolDesc->lock, LW_EXCLUSIVE);
 		}
 	}

src/dmq.c

@@ -45,6 +45,7 @@
 #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)
@@ -992,7 +993,6 @@ dmq_receiver_at_exit(int status, Datum receiver)
 		dmq_receiver_stop_hook(sender_name);
 }
 
-
 Datum
 dmq_receiver_loop(PG_FUNCTION_ARGS)
 {
@@ -1152,6 +1152,7 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 
 		// XXX: is it enough?
 		CHECK_FOR_INTERRUPTS();
+		AcceptInvalidationMessages();
 
 		if (dmq_now() - last_message_at > recv_timeout)
 		{

src/multimaster.c

@@ -325,7 +325,7 @@ _PG_init(void)
 		"Maximal size of transaction after which transaction is written to the disk",
 		NULL,
 		&MtmTransSpillThreshold,
-		25 * 1024, /* 100Mb */
+		100 * 1024, /* 100Mb */
 		0,
 		MaxAllocSize/1024,
 		PGC_SIGHUP,

src/pglogical_apply.c

@@ -1107,10 +1107,10 @@ process_remote_insert(StringInfo s, Relation rel)
 	TupleDesc		tupDesc = RelationGetDescr(rel);
 	HeapTuple		tup;
 
+	PushActiveSnapshot(GetTransactionSnapshot());
 	estate = create_rel_estate(rel);
 	newslot = ExecInitExtraTupleSlot(estate, tupDesc);
 	oldslot = ExecInitExtraTupleSlot(estate, tupDesc);
-	PushActiveSnapshot(GetTransactionSnapshot());
 
 	ExecOpenIndices(estate->es_result_relation_info, false);
 	relinfo = estate->es_result_relation_info;

src/pglogical_receiver.c

@@ -677,10 +677,12 @@ pglogical_receiver_main(Datum main_arg)
 		{
 			int rc, hdr_len;
 
-			if (ProcDiePending && Mtm->pools[nodeId-1].nWorkers > 0)
-				PoolStateShutdown(&Mtm->pools[nodeId-1]);
-
-			CHECK_FOR_INTERRUPTS();
+			if (ProcDiePending)
+			{
+				if (Mtm->pools[nodeId-1].nWorkers > 0)
+					PoolStateShutdown(&Mtm->pools[nodeId-1]);
+				proc_exit(0);
+			}
 
 			/* Wait necessary amount of time */
 			rc = WaitLatchOrSocket(MyLatch,
@@ -707,16 +709,8 @@ pglogical_receiver_main(Datum main_arg)
 				proc_exit(1);
 			}
 
-			if (ProcDiePending)
-			{
-				dsm_handle handle = Mtm->pools[nodeId-1].dsmhandler;
-				dsm_segment *seg = dsm_find_mapping(handle);
-				dsm_detach(seg);
-
-				if (Mtm->pools[nodeId-1].nWorkers > 0)
-					PoolStateShutdown(&Mtm->pools[nodeId-1]);
-				return;
-			}
+			CHECK_FOR_INTERRUPTS();
+			AcceptInvalidationMessages();
 
 			if (count != MtmGetRecoveryCount())
 			{