Add ATX fixes for multimaster

Author: knizhnik

contrib/mmts/multimaster.c

@@ -166,6 +166,8 @@ static void*  MtmCreateSavepointContext(void);
 static void	  MtmRestoreSavepointContext(void* ctx);
 static void	  MtmReleaseSavepointContext(void* ctx);
 static void   MtmSetRemoteFunction(char const* list, void* extra);
+static void*  MtmSuspendTransaction(void);
+static void   MtmResumeTransaction(void* ctx);
 
 // static void MtmCheckClusterLock(void);
 static void MtmCheckSlots(void);
@@ -218,7 +220,9 @@ static TransactionManager MtmTM =
 	MtmInitializeSequence,
 	MtmCreateSavepointContext,
 	MtmRestoreSavepointContext,
-	MtmReleaseSavepointContext
+	MtmReleaseSavepointContext,
+	MtmSuspendTransaction,
+	MtmResumeTransaction
 };
 
 char const* const MtmNodeStatusMnem[] =
@@ -557,6 +561,22 @@ static void	 MtmReleaseSavepointContext(void* ctx)
 {
 }
 
+static void* MtmSuspendTransaction(void)
+{
+	MtmCurrentTrans* ctx = malloc(sizeof(MtmCurrentTrans));
+	*ctx = MtmTx;
+	MtmResetTransaction();
+	MtmBeginTransaction(&MtmTx);
+	return ctx;
+}
+
+static void MtmResumeTransaction(void* ctx)
+{
+	MtmTx = *(MtmCurrentTrans*)ctx;
+	MtmInsideTransaction = true;
+	free(ctx);
+}
+
 
 /*
  * -------------------------------------------
@@ -964,6 +984,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 		x->isSuspended = false;
 		x->isTwoPhase = false;
 		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
@@ -4715,6 +4736,7 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 
 	if (!x->isReplicated && x->isDistributed && x->containsDML) {
 		MtmGenerateGid(x->gid);
+
 		if (!x->isTransactionBlock) {
 			BeginTransactionBlock(false);
 			x->isTransactionBlock = true;

src/backend/access/transam/xact.c

@@ -1860,6 +1860,7 @@ typedef struct {
 	void *PgStatState;
 	void *TriggerState;
 	void *SPIState;
+	void *XTMState;
 	void *SnapshotState;
 	void *PredicateState;
 	void *StorageState;
@@ -1945,26 +1946,29 @@ StartTransaction(void)
 	AtStart_Memory();
 	AtStart_ResourceOwner();
 
-	/*
-	 * Assign a new LocalTransactionId, and combine it with the backendId to
-	 * form a virtual transaction id.
-	 */
-	vxid.backendId = MyBackendId;
-	vxid.localTransactionId = GetNextLocalTransactionId();
+	if (getNestLevelATX() == 0)
+	{
+		/*
+		 * Assign a new LocalTransactionId, and combine it with the backendId to
+		 * form a virtual transaction id.
+		 */
+		vxid.backendId = MyBackendId;
+		vxid.localTransactionId = GetNextLocalTransactionId();
 
-	/*
-	 * Lock the virtual transaction id before we announce it in the proc array
-	 */
-	VirtualXactLockTableInsert(vxid);
+		/*
+		 * Lock the virtual transaction id before we announce it in the proc array
+		 */
+		VirtualXactLockTableInsert(vxid);
 
-	/*
-	 * Advertise it in the proc array.  We assume assignment of
-	 * LocalTransactionID is atomic, and the backendId should be set already.
-	 */
-	Assert(MyProc->backendId == vxid.backendId);
-	MyProc->lxid = vxid.localTransactionId;
+		/*
+		 * Advertise it in the proc array.  We assume assignment of
+		 * LocalTransactionID is atomic, and the backendId should be set already.
+		 */
+		Assert(MyProc->backendId == vxid.backendId);
+		MyProc->lxid = vxid.localTransactionId;
 
-	TRACE_POSTGRESQL_TRANSACTION_START(vxid.localTransactionId);
+		TRACE_POSTGRESQL_TRANSACTION_START(vxid.localTransactionId);
+	}
 
 	/*
 	 * set transaction_timestamp() (a/k/a now()).  We want this to be the same
@@ -1982,7 +1986,7 @@ StartTransaction(void)
 	 * note: prevXactReadOnly is not used at the outermost level
 	 */
 	s->nestingLevel = 1;
-	s->gucNestLevel = 1;
+	s->gucNestLevel = getNestLevelATX() != 0 ? NewGUCNestLevel() : 1;
 	s->childXids = NULL;
 	s->nChildXids = 0;
 	s->maxChildXids = 0;
@@ -2505,7 +2509,7 @@ PrepareTransaction(void)
 	PostPrepare_Twophase();
 
 	/* PREPARE acts the same as COMMIT as far as GUC is concerned */
-	AtEOXact_GUC(true, 1);
+	AtEOXact_GUC(true, CurrentTransactionState->gucNestLevel);
 	AtEOXact_SPI(true);
 	AtEOXact_on_commit_actions(true);
 	AtEOXact_Namespace(true, false);
@@ -2959,7 +2963,6 @@ CommitTransactionCommand(void)
 		case TBLOCK_PREPARE:
 			PrepareTransaction();
 			s->blockState = TBLOCK_DEFAULT;
-			ResumeTransaction();
 			break;
 
 			/*
@@ -3225,7 +3228,6 @@ AbortCurrentTransaction(void)
 			AbortTransaction();
 			CleanupTransaction();
 			s->blockState = TBLOCK_DEFAULT;
-			ResumeTransaction();
 			break;
 
 			/*
@@ -3545,6 +3547,9 @@ void SuspendTransaction(void)
 	//Assert(XactTopTransactionId != InvalidTransactionId);
 	Assert(nParallelCurrentXids == 0);
 
+	if (suspendedXactNum >= MAX_SUSPENDED_XACTS)
+		elog(ERROR, "Too many recursive autonomouse transactions");
+
 	SuspendPgXact(MyPgXact);
 	{
 		TransactionState tts;
@@ -3616,6 +3621,7 @@ void SuspendTransaction(void)
 		MOVELEFT(sus->vxid.localTransactionId, MyProc->lxid, GetNextLocalTransactionId());
 		MOVELEFT(sus->on_commit_actions, pg_on_commit_actions, NULL);
 
+		sus->XTMState = TM->SuspendTransaction();
 		sus->PgStatState = PgStatSuspend();
 		sus->TriggerState = TriggerSuspend();
 		sus->SPIState = SuspendSPI();
@@ -3638,6 +3644,7 @@ bool ResumeTransaction(void)
 		ResumeSPI(sus->SPIState);
 		TriggerResume(sus->TriggerState);
 		PgStatResume(sus->PgStatState);
+		TM->ResumeTransaction(sus->XTMState);
 
 		TopTransactionStateData = sus->TopTransactionStateData;
 		CurrentTransactionState = sus->CurrentTransactionState;
@@ -3671,6 +3678,8 @@ bool ResumeTransaction(void)
 		CurTransactionContext = sus->CurTransactionContext;
 		TransactionAbortContext = sus->TransactionAbortContext;
 
+		Assert(TopTransactionContext && CurTransactionContext && TransactionAbortContext);
+
 		CurrentResourceOwner = sus->CurrentResourceOwner;
 		CurTransactionResourceOwner = sus->CurTransactionResourceOwner;
 		TopTransactionResourceOwner = sus->TopTransactionResourceOwner;

src/backend/access/transam/xtm.c

@@ -72,6 +72,15 @@ void PgReleaseSavepointContext(void* ctx)
 }
 
 
+void* PgSuspendTransaction(void)
+{
+	return NULL;
+}
+
+void PgResumeTransaction(void* ctx)
+{
+}
+
 TransactionManager PgTM = {
 	PgTransactionIdGetStatus,
 	PgTransactionIdSetTreeStatus,
@@ -89,7 +98,9 @@ TransactionManager PgTM = {
 	PgInitializeSequence,
 	PgCreateSavepointContext,
 	PgRestoreSavepointContext,
-	PgReleaseSavepointContext
+	PgReleaseSavepointContext,
+	PgSuspendTransaction,
+	PgResumeTransaction
 };
 
 TransactionManager *TM = &PgTM;

src/backend/executor/spi.c

@@ -2801,7 +2801,7 @@ SPI_register_trigger_data(TriggerData *tdata)
 	return SPI_OK_TD_REGISTER;
 }
 
-#define MOVELEFT(A, B, C) do { (A) = (B); } while (0)
+#define MOVELEFT(A, B, C) do { (A) = (B); (B) = (C); } while (0)
 void *
 SuspendSPI(void)
 {

src/include/access/xtm.h

@@ -110,6 +110,16 @@ typedef struct
 	 */
 	void        (*ReleaseSavepointContext)(void* ctx);
 
+	/*
+	 * Suspend transaction (ATX support)
+	 */
+	void*       (*SuspendTransaction)(void);
+
+	/*
+	 * Resume transaction state saved by SuspendTransaction (ATX support)
+	 */
+	void        (*ResumeTransaction)(void* state);
+
 }	TransactionManager;
 
 /* Get pointer to transaction manager: actually returns content of TM variable */
@@ -147,6 +157,7 @@ extern void PgInitializeSequence(int64* start, int64* step);
 extern void* PgCreateSavepointContext(void);
 extern void PgRestoreSavepointContext(void*);
 extern void PgReleaseSavepointContext(void*);
-
+extern void* PgSuspendTransaction(void);
+extern void PgResumeTransaction(void* ctx);
 
 #endif

src/pl/plpgsql/src/pl_exec.c

@@ -1181,7 +1181,8 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block)
 	volatile int rc = -1;
 	int			i;
 	int			n;
-
+	int         spi_rc;
+	
 	/*
 	 * First initialize all variables declared in this block
 	 */
@@ -1275,6 +1276,9 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block)
 
 		SuspendTransaction();
 
+		if ((spi_rc = SPI_connect()) != SPI_OK_CONNECT)
+			elog(ERROR, "SPI_connect failed: %s", SPI_result_code_string(spi_rc));
+
 		PG_TRY();
 		{
 			EState *old_shared_estate;
@@ -1304,11 +1308,15 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block)
 											   resTypByVal, resTypLen);
 			}
 
+			if ((spi_rc = SPI_finish()) != SPI_OK_FINISH)
+				elog(ERROR, "SPI_finish failed: %s", SPI_result_code_string(spi_rc));
+
 			plpgsql_destroy_econtext(estate);
 
 			old_shared_estate = shared_simple_eval_estate;
 			shared_simple_eval_estate = NULL;
 			error_inside_commit = true;
+
 			CommitTransactionCommand();
 			shared_simple_eval_estate = old_shared_estate;
 
@@ -1332,6 +1340,9 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block)
 
 			if (!error_inside_commit)
 			{
+				if ((spi_rc = SPI_finish()) != SPI_OK_FINISH)
+					elog(ERROR, "SPI_finish failed: %s", SPI_result_code_string(spi_rc));
+
 				plpgsql_destroy_econtext(estate);
 				old_shared_estate = shared_simple_eval_estate;
 				shared_simple_eval_estate = NULL;
@@ -1396,9 +1407,9 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block)
 
 			/* If no match found, re-throw the error */
 			if (e == NULL)
-			{
 				ReThrowError(edata);
-			}
+			else
+				FreeErrorData(edata);
 		}
 		PG_END_TRY();
 	}

src/pl/plpython/plpy_atxobject.c

@@ -105,8 +105,13 @@ PLy_atx_dealloc(PyObject *atx)
 static PyObject *
 PLy_atx_enter(PyObject *self, PyObject *unused)
 {
+	int spi_rc;
+
 	SuspendTransaction();
 
+	if ((spi_rc = SPI_connect()) != SPI_OK_CONNECT)
+		elog(ERROR, "SPI_connect failed: %s", SPI_result_code_string(spi_rc));
+
 	Py_INCREF(self);
 	return self;
 }
@@ -128,10 +133,14 @@ PLy_atx_exit(PyObject *self, PyObject *args)
 	PyObject   *type;
 	PyObject   *value;
 	PyObject   *traceback;
+	int         spi_rc;
 
 	if (!PyArg_ParseTuple(args, "OOO", &type, &value, &traceback))
 		return NULL;
 
+	if ((spi_rc = SPI_finish()) != SPI_OK_FINISH)
+		elog(ERROR, "SPI_finish failed: %s", SPI_result_code_string(spi_rc));
+
 	if (type != Py_None)
 	{
 		AbortCurrentTransaction();