Merge branch 'master' of https://github.com/postgrespro/postgres_cluster

Author: knizhnik

contrib/mmts/arbiter.c

@@ -1051,6 +1051,7 @@ static void MtmReceiver(Datum arg)
 									MtmWakeUpBackend(ts);
 								}
 							} else { 
+								elog(WARNING, "Receive PRECOMMITTED response for aborted transaction"); // How it can happen? SHould we use assert here?
 								if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 									MtmWakeUpBackend(ts);
 								}

contrib/mmts/multimaster.c

@@ -942,6 +942,7 @@ void MtmPrecommitTransaction(char const* gid)
 			ts->csn = MtmAssignCSN();
 			MtmAdjustSubtransactions(ts);
 			MtmSend2PCMessage(ts, MSG_PRECOMMITTED);
+			SetPrepareTransactionState(ts->gid, "precommitted");
 		}
 	}
 	MtmUnlock();
@@ -1311,6 +1312,7 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 
 	if (MtmIsCoordinator(ts)) {
 		int i;
+		Assert(false); // All broadcasts are now done through logical decoding
 		for (i = 0; i < Mtm->nAllNodes; i++)
 		{
 			if (BIT_CHECK(ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask, i))
@@ -1560,7 +1562,10 @@ void MtmHandleApplyError(void)
 }
 
 /**
- * Check status of all prepared transactions with coordinator at disabled node
+ * Check status of all prepared transactions with coordinator at disabled node.
+ * Actually, if node is precommitted (state == UNKNOWN) at any of nodes, then is is prepared at all nodes and so can be committed. 
+ * But if coordinator of transaction is crashed, we made a decision about transaction commit only if transaction is precommitted at ALL live nodes.
+ * The reason is that we want to avoid extra polling to obtain maximum CSN from all nodes to assign it to committed transaction.
  * Called only from MtmDisableNode in critical section.
  */
 static void MtmPollStatusOfPreparedTransactions(int disabledNodeId)
@@ -1601,9 +1606,12 @@ static void MtmDisableNode(int nodeId)
 	Mtm->nodes[nodeId-1].lastStatusChangeTime = now;
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	if (nodeId != MtmNodeId) { 
-		Mtm->nLiveNodes -= 1;
+		Mtm->nLiveNodes -= 1;		
+	}
+	if (Mtm->nLiveNodes >= Mtm->nAllNodes/2+1) {
+		/* Make decision about prepared transaction status only in quorum */
+		MtmPollStatusOfPreparedTransactions(nodeId);
 	}
-	MtmPollStatusOfPreparedTransactions(nodeId);
 } 
 
 static void MtmEnableNode(int nodeId)
@@ -3228,9 +3236,10 @@ bool MtmFilterTransaction(char* record, int size)
 		   origin_node != 0 &&
 		   (Mtm->status == MTM_RECOVERY || origin_node == replication_node));	
 
-    switch(PGLOGICAL_XACT_EVENT(flags))
+    switch (PGLOGICAL_XACT_EVENT(flags))
     {
 	  case PGLOGICAL_PREPARE:
+	  case PGLOGICAL_PRECOMMIT_PREPARED:
 	  case PGLOGICAL_ABORT_PREPARED:
 		gid = pq_getmsgstring(&s);
 		break;

contrib/mmts/multimaster.h

@@ -44,7 +44,7 @@
 #define MTM_TXTRACE(tx, event)
 #else
 #define MTM_TXTRACE(tx, event) \
-		fprintf(stderr, "[MTM_TXTRACE], %s, %lld, %s, %d\n", tx->gid, (long long)MtmGetSystemTime(), event, getpid())
+		fprintf(stderr, "[MTM_TXTRACE], %s, %lld, %s, %d\n", tx->gid, (long long)MtmGetSystemTime(), event, MyProcPid)
 #endif
 
 #define MULTIMASTER_NAME                "multimaster"
@@ -160,6 +160,12 @@ typedef struct
 	pgid_t         gid;    /* Global transaction identifier */
 } MtmArbiterMessage;
 
+/*
+ * Abort logical message is send by replica when error is happen while applying prepared transaction.
+ * In this case we do not have prepared transaction and can not do abort-prepared.
+ * But we have to record the fact of abort to be able to replay it in case of crash of coordinator of this transaction.
+ * We are using logical abort message with code 'A' for it
+ */
 typedef struct MtmAbortLogicalMessage
 {
 	pgid_t         gid;

contrib/mmts/pglogical_apply.c

@@ -625,8 +625,12 @@ process_remote_commit(StringInfo in)
 	{
 	    case PGLOGICAL_PRECOMMIT_PREPARED:
 		{
+			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
+			MTM_LOG2("%d: PGLOGICAL_PRECOMMIT_PREPARED %s", MyProcPid, gid);
+			MtmBeginSession(origin_node);
 			MtmPrecommitTransaction(gid);
+			MtmEndSession(origin_node, true);
 			return;
 		}
 		case PGLOGICAL_COMMIT:
@@ -691,6 +695,7 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
+			/* MtmRollbackPreparedTransaction will set origin session itself */
 			MtmRollbackPreparedTransaction(origin_node, gid);
 			break;
 		}

contrib/mmts/pglogical_proto.c

@@ -192,7 +192,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	else
     	Assert(false);
 
-	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE || flags == PGLOGICAL_PRECOMMIT_PREPARED) { 
+	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) {
+		/* COMMIT and PREPARE are preceded by BEGIN, which set MtmIsFilteredTxn flag */
 		if (MtmIsFilteredTxn) { 
 			Assert(MtmTransactionRecords == 0);
 			return;

src/backend/access/transam/twophase.c

@@ -183,6 +183,9 @@ static TwoPhaseStateData *TwoPhaseState;
 static GlobalTransaction MyLockedGxact = NULL;
 
 static bool twophaseExitRegistered = false;
+static TransactionId cached_xid = InvalidTransactionId;
+static GlobalTransaction cached_gxact = NULL;
+
 
 static void RecordTransactionCommitPrepared(TransactionId xid,
 								int nchildren,
@@ -216,10 +219,10 @@ TwoPhaseShmemSize(void)
 	/* Need the fixed struct, the array of pointers, and the GTD structs */
 	size = offsetof(TwoPhaseStateData, prepXacts);
 	size = add_size(size, mul_size(max_prepared_xacts,
-								   sizeof(GlobalTransaction)));
+								   sizeof(GlobalTransaction)*2));
 	size = MAXALIGN(size);
 	size = add_size(size, mul_size(max_prepared_xacts,
-								   sizeof(GlobalTransactionData)*2));
+								   sizeof(GlobalTransactionData)));
 
 	return size;
 }
@@ -247,9 +250,9 @@ TwoPhaseShmemInit(void)
 		gxacts = (GlobalTransaction)
 			((char *) TwoPhaseState +
 			 MAXALIGN(offsetof(TwoPhaseStateData, prepXacts) +
-					  sizeof(GlobalTransaction) * max_prepared_xacts));
+					  sizeof(GlobalTransaction) * 2 * max_prepared_xacts));
 
-		TwoPhaseState->hashTable = (GlobalTransaction*)&gxacts[max_prepared_xacts];
+		TwoPhaseState->hashTable = &TwoPhaseState->prepXacts[max_prepared_xacts];
 
 		for (i = 0; i < max_prepared_xacts; i++)
 		{
@@ -438,6 +441,10 @@ MarkAsPreparing(TransactionId xid, const char *gid,
 	proc->lwWaitMode = 0;
 	proc->waitLock = NULL;
 	proc->waitProcLock = NULL;
+	
+	cached_xid = xid;
+	cached_gxact = gxact;
+		
 	for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
 		SHMQueueInit(&(proc->myProcLocks[i]));
 	/* subxid data must be filled later by GXactLoadSubxactData */
@@ -696,9 +703,7 @@ void SetPrepareTransactionState(char const* gid, char const* state)
 	strcpy(gxact->state_3pc, state);
 	EndPrepare(gxact);
 
-	/* Unlock GXact */
-	gxact->locking_backend = InvalidBackendId;
-	MyLockedGxact = NULL;
+	PostPrepare_Twophase();
 }
 
 /* Working status for pg_prepared_xact */
@@ -827,9 +832,6 @@ TwoPhaseGetGXact(TransactionId xid)
 	GlobalTransaction result = NULL;
 	int			i;
 
-	static TransactionId cached_xid = InvalidTransactionId;
-	static GlobalTransaction cached_gxact = NULL;
-
 	/*
 	 * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
 	 * repeatedly for the same XID.  We can save work with a simple cache.

src/backend/replication/logical/reorderbuffer.c

@@ -1346,7 +1346,7 @@ ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap)
  *
  * We currently can only decode a transaction's contents in when their commit
  * record is read because that's currently the only place where we know about
- * cache invalidations. Thus, once a toplevel commit is read, we iterate over
+ * cache invalidati ons. Thus, once a toplevel commit is read, we iterate over
  * the top and subtransactions (using a k-way merge) and replay the changes in
  * lsn order.
  */
@@ -1734,7 +1734,7 @@ ReorderBufferCommitBareXact(ReorderBuffer *rb, TransactionId xid,
 	txn->origin_lsn = origin_lsn;
 	txn->xact_action = rb->xact_action;
 	strcpy(txn->gid, rb->gid);
-	*txn->gid = '\0';
+	*txn->state_3pc = '\0';
 
 	rb->commit(rb, txn, commit_lsn);
 }

src/include/catalog/pg_proc.h

@@ -3057,7 +3057,7 @@ DATA(insert OID = 1371 (  pg_lock_status   PGNSP PGUID 12 1 1000 0 0 f f f f t t
 DESCR("view system lock information");
 DATA(insert OID = 2561 (  pg_blocking_pids PGNSP PGUID 12 1 0 0 0 f f f f t f v s 1 0 1007 "23" _null_ _null_ _null_ _null_ _null_ pg_blocking_pids _null_ _null_ _null_ ));
 DESCR("get array of PIDs of sessions blocking specified backend PID");
-DATA(insert OID = 1065 (  pg_prepared_xact PGNSP PGUID 12 1 1000 0 0 f f f f t t v s 0 0 2249 "" "{28,25,1184,26,26,25}" "{o,o,o,o,o,0}" "{transaction,gid,prepared,ownerid,dbid,state_3pc}" _null_ _null_ pg_prepared_xact _null_ _null_ _null_ ));
+DATA(insert OID = 1065 (  pg_prepared_xact PGNSP PGUID 12 1 1000 0 0 f f f f t t v s 0 0 2249 "" "{28,25,1184,26,26,25}" "{o,o,o,o,o,o}" "{transaction,gid,prepared,ownerid,dbid,state3pc}" _null_ _null_ pg_prepared_xact _null_ _null_ _null_ ));
 DESCR("view two-phase transactions");
 DATA(insert OID = 3445 (  pg_precommit_prepared PGNSP PGUID 12 1 0 0 0 f f f f t f v s 2 0 2278 "2275,2275" _null_ _null_ _null_ _null_ _null_ pg_precommit_prepared _null_ _null_ _null_ ));
 DESCR("alter state of prepared transaction (used for 3pc)");