dmq integration: make setup procedures more compatible with previous mm versions

Author: kelvich

Cluster.pm

@@ -120,13 +120,12 @@ sub configure
 			listen_addresses = '$host'
 			unix_socket_directories = '$unix_sock_dir'
 			port = $pgport
-			max_prepared_transactions = 10
+			max_prepared_transactions = 30
 			max_connections = 10
 			max_worker_processes = 100
 			wal_level = logical
 			max_wal_senders = 6
 			wal_sender_timeout = 0
-			default_transaction_isolation = 'repeatable read'
 			max_replication_slots = 6
 			shared_preload_libraries = 'multimaster'
 			shared_buffers = 16MB
@@ -160,6 +159,7 @@ sub start
 	foreach my $node (@$nodes)
 	{
 		$node->start();
+		$node->safe_psql('postgres', "create extension multimaster;");
 		note( "Starting node with connstr 'dbname=postgres port=@{[ $node->port() ]} host=@{[ $node->host() ]}'");
 	}
 }
@@ -230,7 +230,7 @@ sub stop
 		$node->stop($mode);
 	}
 
-	$self->dumplogs();
+	# $self->dumplogs();
 
 	return $ok;
 }

bgwpool.h

@@ -8,7 +8,7 @@
 
 typedef void(*BgwPoolExecutor)(void* work, size_t size);
 
-typedef ulong64 timestamp_t;
+typedef long timestamp_t;
 
 
 #define MAX_DBNAME_LEN 30

commit.c

@@ -15,6 +15,7 @@
 #include "utils/guc.h"
 #include "miscadmin.h"
 #include "commands/dbcommands.h"
+#include "tcop/tcopprot.h"
 
 #include "logger.h"
 
@@ -63,18 +64,19 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 	x->containsDML = false; // will be set by executor hook
 	x->isTransactionBlock = IsTransactionBlock();
 
-	// /* Application name can be changed using PGAPPNAME environment variable */
-	// if (x->isDistributed && Mtm->status != MTM_ONLINE
-	// 	&& strcmp(application_name, MULTIMASTER_ADMIN) != 0
-	// 	&& strcmp(application_name, MULTIMASTER_BROADCAST_SERVICE) != 0)
-	// {
-	// 	/* Reject all user's transactions at offline cluster.
-	// 	 * Allow execution of transaction by bg-workers to makeit possible to perform recovery.
-	// 	 */
-	// 	MTM_ELOG(ERROR,
-	// 			"Multimaster node is not online: current status %s",
-	// 			MtmNodeStatusMnem[Mtm->status]);
-	// }
+	/* Application name can be changed using PGAPPNAME environment variable */
+	if (x->isDistributed && Mtm->status != MTM_ONLINE
+		&& strcmp(application_name, MULTIMASTER_ADMIN) != 0
+		&& strcmp(application_name, MULTIMASTER_BROADCAST_SERVICE) != 0
+		&& debug_query_string && pg_strcasecmp(debug_query_string, "create extension multimaster;") != 0)
+	{
+		/* Reject all user's transactions at offline cluster.
+		 * Allow execution of transaction by bg-workers to makeit possible to perform recovery.
+		 */
+		MTM_ELOG(ERROR,
+				"Multimaster node is not online: current status %s",
+				MtmNodeStatusMnem[Mtm->status]);
+	}
 }
 
 static void
@@ -104,7 +106,7 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	char	stream[DMQ_NAME_MAXLEN];
 	pgid_t  gid;
 
-	if (!x->isDistributed || !x->containsDML)
+	if (!x->isDistributed || !x->containsDML || !Mtm->extension_created)
 		return false;
 
 	if (!DmqSubscribed)

multimaster.c

@@ -435,20 +435,26 @@ timestamp_t MtmGetCurrentTime(void)
 	return MtmGetSystemTime() + Mtm->timeShift;
 }
 
-void MtmSleep(timestamp_t interval)
+void MtmSleep(timestamp_t usec)
 {
-	timestamp_t waketm = MtmGetCurrentTime() + interval;
+	timestamp_t waketm = MtmGetCurrentTime() + usec;
+
 	for (;;)
 	{
+		int rc;
 		timestamp_t sleepfor = waketm - MtmGetCurrentTime();
 
-		pg_usleep(sleepfor);
-		if (MtmGetCurrentTime() < waketm)
-		{
-			/* Assert(errno == EINTR); */
-			continue;
-		}
-		break;
+		CHECK_FOR_INTERRUPTS();
+
+		rc = WaitLatch(MyLatch,
+						WL_TIMEOUT | WL_POSTMASTER_DEATH,
+						sleepfor/1000.0, WAIT_EVENT_BGWORKER_STARTUP);
+
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+		if (MtmGetCurrentTime() > waketm)
+			break;
 	}
 }
 
@@ -1177,10 +1183,10 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 		MtmLock(LW_EXCLUSIVE);
 		if (MtmMin2PCTimeout != 0 && now > deadline) {
 			if (ts->isPrepared) {
-				MTM_ELOG(LOG, "Distributed transaction %s (%llu) is not committed in %lld msec", ts->gid, (long64)ts->xid, USEC_TO_MSEC(now - start));
+				// MTM_ELOG(LOG, "Distributed transaction %s (%llu) is not committed in %lld msec", ts->gid, (long64)ts->xid, USEC_TO_MSEC(now - start));
 			} else {
-				MTM_ELOG(WARNING, "Commit of distributed transaction %s (%llu) is canceled because of %lld msec timeout expiration",
-					 ts->gid, (long64)ts->xid, USEC_TO_MSEC(timeout));
+				// MTM_ELOG(WARNING, "Commit of distributed transaction %s (%llu) is canceled because of %lld msec timeout expiration",
+					//  ts->gid, (long64)ts->xid, USEC_TO_MSEC(timeout));
 				MtmAbortTransaction(ts);
 				break;
 			}
@@ -2398,6 +2404,7 @@ static void MtmInitialize()
 	if (!found)
 	{
 		MemSet(Mtm, 0, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1));
+		Mtm->extension_created = false;
 		Mtm->status = MTM_DISABLED; //MTM_INITIALIZATION;
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);
@@ -2481,19 +2488,6 @@ static void MtmInitialize()
 
 	MtmCheckControlFile();
 
-	/* Start dmq senders */
-	for (i = 0; i < MtmNodes; i++)
-	{
-		int destination_id;
-
-		if (i + 1 == MtmNodeId)
-			continue;
-		destination_id = dmq_destination_add(MtmConnections[i].connStr,
-											 psprintf("node%d", MtmNodeId),
-											 psprintf("node%d", i + 1),
-											 MtmHeartbeatSendTimeout);
-		Mtm->nodes[i].destination_id = destination_id;
-	}
 }
 
 static void
@@ -3394,7 +3388,7 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
  * During recovery we need to open only one replication slot from which node should receive all transactions.
  * Slots at other nodes should be removed.
  */
-MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown)
+MtmReplicationMode MtmGetReplicationMode(int nodeId)
 {
 	MtmLock(LW_EXCLUSIVE);
 
@@ -3410,8 +3404,6 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 			BIT_CHECK(EFFECTIVE_CONNECTIVITY_MASK, MtmNodeId - 1))
 	{
 		MtmUnlock();
-		if (*shutdown)
-			return REPLMODE_EXIT;
 		MtmSleep(STATUS_POLL_DELAY);
 		MtmLock(LW_EXCLUSIVE);
 	}
@@ -3435,8 +3427,6 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 		while (BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId - 1))
 		{
 			MtmUnlock();
-			if (*shutdown)
-				return REPLMODE_EXIT;
 			MtmSleep(STATUS_POLL_DELAY);
 			MtmLock(LW_EXCLUSIVE);
 		}
@@ -4267,7 +4257,8 @@ typedef struct
 	int		  nodeId;
 } MtmGetClusterInfoCtx;
 
-static void erase_option_from_connstr(const char *option, char *connstr)
+void
+erase_option_from_connstr(const char *option, char *connstr)
 {
 	char *needle = psprintf("%s=", option);
 	while (1) {
@@ -5772,3 +5763,22 @@ MtmToggleDML(void)
 {
 	MtmTx.containsDML = true;
 }
+
+
+void
+MtmWaitForExtensionCreation(void)
+{
+	for (;;)
+	{
+		RangeVar   *rv;
+		Oid			rel_oid;
+		int			rc;
+
+		StartTransactionCommand();
+		rv = makeRangeVar(MULTIMASTER_SCHEMA_NAME, "local_tables", -1);
+		rel_oid = RangeVarGetRelid(rv, NoLock, true);
+		CommitTransactionCommand();
+
+		MtmSleep(USECS_PER_SEC);
+	}
+}

multimaster.h

@@ -316,6 +316,7 @@ typedef struct {
 
 typedef struct
 {
+	bool extension_created;
 	MtmNodeStatus status;              /* Status of this node */
 	/* A human-readable description of why the current status was set */
 	char *statusReason;
@@ -451,6 +452,9 @@ extern void MtmGenerateGid(char *gid, TransactionId xid);
 extern int  MtmGidParseNodeId(const char* gid);
 extern TransactionId MtmGidParseXid(const char* gid);
 
+extern void MtmWaitForExtensionCreation(void);
+extern void erase_option_from_connstr(const char *option, char *connstr);
+
 extern void ResolverMain(void);
 extern void ResolverInit(void);
 extern void ResolveTransactionsForNode(int node_id);
@@ -468,7 +472,7 @@ extern csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, no
 extern csn_t MtmAssignCSN(void);
 extern csn_t MtmSyncClock(csn_t csn);
 extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot, nodemask_t participantsMask);
-extern MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown);
+extern MtmReplicationMode MtmGetReplicationMode(int nodeId);
 extern void  MtmExecutor(void* work, size_t size);
 extern void  MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd);
 extern void  MtmSendMessage(MtmArbiterMessage* msg);

pglogical_receiver.c

@@ -40,6 +40,7 @@
 #include "replication/origin.h"
 #include "utils/portal.h"
 #include "tcop/pquery.h"
+#include "tcop/tcopprot.h"
 
 #ifdef WITH_RSOCKET
 #include "libpq-int.h"
@@ -55,7 +56,6 @@
 #define RECEIVER_SUSPEND_TIMEOUT (1*USECS_PER_SEC)
 
 /* Signal handling */
-static volatile sig_atomic_t got_sigterm = false;
 static volatile sig_atomic_t got_sighup = false;
 
 /* GUC variables */
@@ -75,16 +75,6 @@ static int64 fe_recvint64(char *buf);
 
 void pglogical_receiver_main(Datum main_arg);
 
-static void
-receiver_raw_sigterm(SIGNAL_ARGS)
-{
-	int save_errno = errno;
-	got_sigterm = true;
-	if (MyProc)
-		SetLatch(&MyProc->procLatch);
-	errno = save_errno;
-}
-
 static void
 receiver_raw_sighup(SIGNAL_ARGS)
 {
@@ -258,7 +248,8 @@ pglogical_receiver_main(Datum main_arg)
 
 	/* Register functions for SIGTERM/SIGHUP management */
 	pqsignal(SIGHUP, receiver_raw_sighup);
-	pqsignal(SIGTERM, receiver_raw_sigterm);
+	// pqsignal(SIGTERM, receiver_raw_sigterm);
+	pqsignal(SIGTERM, die);
 
 	MtmCreateSpillDirectory(nodeId);
 
@@ -277,6 +268,8 @@ pglogical_receiver_main(Datum main_arg)
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";
 
+	MtmWaitForExtensionCreation();
+
 	BgwPoolStart(&Mtm->nodes[nodeId-1].pool, worker_proc);
 
 	/*
@@ -305,7 +298,7 @@ pglogical_receiver_main(Datum main_arg)
 	 * In case of errors we will try to reestablish connection.
 	 * Also reconnet is forced when node is switch to recovery mode
 	 */
-	while (!got_sigterm)
+	for(;;)
 	{
 		int	 count;
 		ConnStatusType status;
@@ -317,7 +310,7 @@ pglogical_receiver_main(Datum main_arg)
 		 * Druing recovery we need to open only one replication slot from which node should receive all transactions.
 		 * Slots at other nodes should be removed
 		 */
-		mode = MtmGetReplicationMode(nodeId, &got_sigterm);
+		mode = MtmGetReplicationMode(nodeId);
 		MTM_LOG1("[STATE] Node %i: wal_receiver starts in %s mode", nodeId, MtmReplicationModeName[mode]);
 
 		// if (mode == REPLMODE_RECOVERY)
@@ -419,7 +412,7 @@ pglogical_receiver_main(Datum main_arg)
 
 		MtmStateProcessNeighborEvent(nodeId, MTM_NEIGHBOR_WAL_RECEIVER_START);
 
-		while (!got_sigterm)
+		for(;;)
 		{
 			int rc, hdr_len;
 			/* Wait necessary amount of time */
@@ -437,13 +430,6 @@ pglogical_receiver_main(Datum main_arg)
 				ereport(LOG, (MTM_ERRMSG("%s: processed SIGHUP", worker_proc)));
 			}
 
-			if (got_sigterm)
-			{
-				/* Simply exit */
-				ereport(LOG, (MTM_ERRMSG("%s: processed SIGTERM", worker_proc)));
-				proc_exit(0);
-			}
-
 			/* Emergency bailout if postmaster has died */
 			if (rc & WL_POSTMASTER_DEATH)
 			{
@@ -606,12 +592,12 @@ pglogical_receiver_main(Datum main_arg)
 								} else {
 									if (MtmPreserveCommitOrder && buf.used == msg_len) {
 										/* Perform commit-prepared and rollback-prepared requested directly in receiver */
-										timestamp_t stop, start = MtmGetSystemTime();
+										// timestamp_t stop, start = MtmGetSystemTime();
 										MtmExecutor(buf.data, buf.used);
-										stop = MtmGetSystemTime();
-										if (stop - start > USECS_PER_SEC) {
-											elog(WARNING, "Commit of prepared transaction takes %lld usec, flags=%x", stop - start, stmt[1]);
-										}
+										// stop = MtmGetSystemTime();
+										// if (stop - start > USECS_PER_SEC) {
+										// 	elog(WARNING, "Commit of prepared transaction takes %lld usec, flags=%x", stop - start, stmt[1]);
+										// }
 									} else {
 										/* all other commits should be applied in place */
 										// Assert(stmt[1] == PGLOGICAL_PREPARE || stmt[1] == PGLOGICAL_COMMIT || stmt[1] == PGLOGICAL_PRECOMMIT_PREPARED);

resolver.c

@@ -428,8 +428,13 @@ ResolverMain(void)
 	/* init this worker */
 	pqsignal(SIGTERM, die);
 	BackgroundWorkerUnblockSignals();
+
+	MtmBackgroundWorker = true;
+
 	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL, 0);
 
+	MtmWaitForExtensionCreation();
+
 	/* init map with current unresolved transactions */
 	ctl.keysize = GIDSIZE;
 	ctl.entrysize = sizeof(resolver_tx);