handcrafted fairness for MtmCommitBarrier LWLock

Author: kelvich

src/commit.c

@@ -163,6 +163,9 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	 *
 	 * See also comments at the end of MtmReplicationStartupHook().
 	 */
+	while (Mtm->stop_new_commits)
+		MtmSleep(USECS_PER_SEC);
+
 	LWLockAcquire(MtmCommitBarrier, LW_SHARED);
 
 	MtmLock(LW_SHARED);

src/include/multimaster.h

@@ -126,6 +126,7 @@ typedef struct
 typedef struct
 {
 	bool extension_created;
+	bool stop_new_commits;
 	bool recovered;
 	MtmNodeStatus status;              /* Status of this node */
 	char *statusReason;                /* A human-readable description of why the current status was set */

src/multimaster.c

@@ -565,6 +565,7 @@ static void MtmInitialize()
 	{
 		MemSet(Mtm, 0, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1));
 		Mtm->extension_created = false;
+		Mtm->stop_new_commits = false;
 		Mtm->recovered = false;
 		Mtm->status = MTM_DISABLED; //MTM_INITIALIZATION;
 		Mtm->recoverySlot = 0;

src/pglogical_proto.c

@@ -836,10 +836,12 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 			XLogRecPtr msg_xptr;
 			char *session_id = psprintf(INT64_FORMAT, hooks_data->session_id);
 
+			Mtm->stop_new_commits = true;
 			LWLockAcquire(MtmCommitBarrier, LW_EXCLUSIVE);
 			MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED, false);
 			msg_xptr = LogLogicalMessage("P", session_id, strlen(session_id) + 1, false);
 			LWLockRelease(MtmCommitBarrier);
+			Mtm->stop_new_commits = false;
 
 			XLogFlush(msg_xptr);
 		}

tests/docker-entrypoint.sh

@@ -59,6 +59,8 @@ if [ "$1" = 'postgres' ]; then
 			shared_preload_libraries = 'multimaster'
 			log_line_prefix = '%m [%p]: '
 			wal_writer_delay = 500ms
+			archive_mode = on
+			archive_command = '/bin/false'
 			# log_statement = all
 
 			multimaster.max_nodes = 3