port 3pc stuff

Author: kelvich

src/backend/access/transam/twophase.c

@@ -171,6 +171,7 @@ typedef struct GlobalTransactionData
 	bool		ondisk;			/* TRUE if prepare state file is on disk */
 	bool		inredo;			/* TRUE if entry was added via xlog_redo */
 	char		gid[GIDSIZE];	/* The GID assigned to the prepared xact */
+	char		state_3pc[MAX_3PC_STATE_SIZE]; /* 3PC transaction state  */
 }			GlobalTransactionData;
 
 /*
@@ -422,6 +423,7 @@ MarkAsPreparing(TransactionId xid, const char *gid,
 	TwoPhaseState->freeGXacts = gxact->next;
 
 	MarkAsPreparingGuts(gxact, xid, gid, prepared_at, owner, databaseid);
+	*gxact->state_3pc = '\0';
 
 	gxact->ondisk = false;
 
@@ -735,7 +737,7 @@ pg_prepared_xact(PG_FUNCTION_ARGS)
 
 		/* build tupdesc for result tuples */
 		/* this had better match pg_prepared_xacts view in system_views.sql */
-		tupdesc = CreateTemplateTupleDesc(5, false);
+		tupdesc = CreateTemplateTupleDesc(6, false);
 		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "transaction",
 						   XIDOID, -1, 0);
 		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "gid",
@@ -746,6 +748,8 @@ pg_prepared_xact(PG_FUNCTION_ARGS)
 						   OIDOID, -1, 0);
 		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "dbid",
 						   OIDOID, -1, 0);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "state3pc",
+						   TEXTOID, -1, 0);
 
 		funcctx->tuple_desc = BlessTupleDesc(tupdesc);
 
@@ -770,8 +774,8 @@ pg_prepared_xact(PG_FUNCTION_ARGS)
 		GlobalTransaction gxact = &status->array[status->currIdx++];
 		PGPROC	   *proc = &ProcGlobal->allProcs[gxact->pgprocno];
 		PGXACT	   *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
-		Datum		values[5];
-		bool		nulls[5];
+		Datum		values[6];
+		bool		nulls[6];
 		HeapTuple	tuple;
 		Datum		result;
 
@@ -789,6 +793,7 @@ pg_prepared_xact(PG_FUNCTION_ARGS)
 		values[2] = TimestampTzGetDatum(gxact->prepared_at);
 		values[3] = ObjectIdGetDatum(gxact->owner);
 		values[4] = ObjectIdGetDatum(proc->databaseId);
+		values[5] = CStringGetTextDatum(gxact->state_3pc);
 
 		tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
 		result = HeapTupleGetDatum(tuple);
@@ -916,6 +921,7 @@ typedef struct TwoPhaseFileHeader
 	uint16		gidlen;			/* length of the GID - GID follows the header */
 	XLogRecPtr	origin_lsn;		/* lsn of this record at origin node */
 	TimestampTz origin_timestamp; /* time of prepare at origin node */
+	char		state_3pc[MAX_3PC_STATE_SIZE];	/* 3PC state */
 } TwoPhaseFileHeader;
 
 /*
@@ -1026,7 +1032,8 @@ StartPrepare(GlobalTransaction gxact)
 	hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels);
 	hdr.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
 														  &hdr.initfileinval);
-	hdr.gidlen = strlen(gxact->gid) + 1;	/* Include '\0' */
+	hdr.gidlen = strlen(gxact->gid) + 1;		/* Include '\0' */
+	*hdr.state_3pc = '\0';
 
 	save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
 	save_state_data(gxact->gid, hdr.gidlen);
@@ -1330,6 +1337,7 @@ ParsePrepareRecord(uint8 info, char *xlrec, xl_xact_parsed_prepare *parsed)
 	parsed->ncommitrels = hdr->ncommitrels;
 	parsed->nabortrels = hdr->nabortrels;
 	parsed->nmsgs = hdr->ninvalmsgs;
+	strncpy(parsed->state_3pc, hdr->state_3pc, MAX_3PC_STATE_SIZE);
 
 	strncpy(parsed->twophase_gid, bufptr, hdr->gidlen);
 	bufptr += MAXALIGN(hdr->gidlen);
@@ -1657,11 +1665,27 @@ RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
 	pg_crc32c	statefile_crc;
 	int			fd;
 
+	/* Crutch to fix crc and len check on 2pc file reading */
+	Assert( ((TwoPhaseFileHeader *) content)->total_len - sizeof(pg_crc32c) <= len);
+	len = ((TwoPhaseFileHeader *) content)->total_len - sizeof(pg_crc32c);
+
 	/* Recompute CRC */
 	INIT_CRC32C(statefile_crc);
 	COMP_CRC32C(statefile_crc, content, len);
 	FIN_CRC32C(statefile_crc);
 
+	/*
+	 * 3PC hacky support. Xid for xlog record is set during xlog insert
+	 * via GetCurrentTransactionIdIfAny() call. However this tx isn't already
+	 * active so allow it to be zero in xlog, but override here during recovery
+	 * so the file name will be valid xid.
+	 */
+	if (!TransactionIdIsValid(xid))
+	{
+		Assert( *(((TwoPhaseFileHeader *) content)->state_3pc) != '\0');
+		xid = ((TwoPhaseFileHeader *) content)->xid;
+	}
+
 	TwoPhaseFilePath(path, xid);
 
 	fd = OpenTransientFile(path,
@@ -2388,6 +2412,34 @@ RecordTransactionAbortPrepared(TransactionId xid,
 	SyncRepWaitForLSN(recptr, false);
 }
 
+
+int
+GetPreparedTransactions(PreparedTransaction* pxacts)
+{
+	int			num;
+	int			i;
+
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+
+	num = TwoPhaseState->numPrepXacts;
+	if (num == 0)
+	{
+		LWLockRelease(TwoPhaseStateLock);
+		*pxacts = NULL;
+		return 0;
+	}
+
+	*pxacts = (PreparedTransaction)palloc(sizeof(PreparedTransactionData) * num);
+	for (i = 0; i < num; i++) {
+		(*pxacts)[i].owner = TwoPhaseState->prepXacts[i]->owner;
+		strcpy((*pxacts)[i].gid, TwoPhaseState->prepXacts[i]->gid);
+		strcpy((*pxacts)[i].state_3pc, TwoPhaseState->prepXacts[i]->state_3pc);
+	}
+	LWLockRelease(TwoPhaseStateLock);
+
+	return num;
+}
+
 /*
  * PrepareRedoAdd
  *
@@ -2404,6 +2456,7 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
 	char	   *bufptr;
 	const char *gid;
 	GlobalTransaction gxact;
+	int				  i;
 
 	Assert(LWLockHeldByMeInMode(TwoPhaseStateLock, LW_EXCLUSIVE));
 	Assert(RecoveryInProgress());
@@ -2422,6 +2475,24 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
 	 * that it got added in the redo phase
 	 */
 
+	/*
+	 * MTM-SPECIFIC.
+	 * In our case 3PC is done via logging prepare record twice with different state_3pc.
+	 * So try to find state with same xid and change the state. Other option is to use
+	 * twice bigger numPrepXacts, but then we anyway need iterate whole array upon
+	 * deletion to wipe all mentions of this tx.
+	 */
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		gxact = TwoPhaseState->prepXacts[i];
+		if (gxact->xid == hdr->xid)
+		{
+			Assert(gxact->inredo);
+			strncpy(gxact->state_3pc, hdr->state_3pc, MAX_3PC_STATE_SIZE);
+			return;
+		}
+	}
+
 	/* Get a free gxact from the freelist */
 	if (TwoPhaseState->freeGXacts == NULL)
 		ereport(ERROR,
@@ -2442,6 +2513,7 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
 	gxact->ondisk = XLogRecPtrIsInvalid(start_lsn);
 	gxact->inredo = true;		/* yes, added in redo */
 	strcpy(gxact->gid, gid);
+	strncpy(gxact->state_3pc, hdr->state_3pc, MAX_3PC_STATE_SIZE);
 
 	/* And insert it into the active array */
 	Assert(TwoPhaseState->numPrepXacts < max_prepared_xacts);
@@ -2607,3 +2679,84 @@ pg_prepared_xact_status(PG_FUNCTION_ARGS)
 	XLogReaderFree(xlogreader);
 	PG_RETURN_TEXT_P(cstring_to_text(xact_status));
 }
+
+bool GetPreparedTransactionState(char const* gid, char* state)
+{
+	int i;
+	GlobalTransaction gxact;
+	bool result = false;
+
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		gxact = TwoPhaseState->prepXacts[i];
+		if (strcmp(gxact->gid, gid) == 0)
+		{
+			strcpy(state, gxact->state_3pc);
+			result = true;
+			break;
+		}
+	}
+	LWLockRelease(TwoPhaseStateLock);
+	return result;
+}
+
+
+/*
+ * SetPrepareTransactionState
+ * Alter 3PC state of prepared transaction
+ */
+void SetPreparedTransactionState(char const* gid, char const* state)
+{
+	GlobalTransaction gxact;
+	PGXACT	   *pgxact;
+	TwoPhaseFileHeader *hdr;
+	char* buf;
+	bool replorigin;
+
+	if (strlen(state) >= MAX_3PC_STATE_SIZE)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("transaction state \"%s\" is too long",
+						state)));
+
+	replorigin = (replorigin_session_origin != InvalidRepOriginId &&
+				  replorigin_session_origin != DoNotReplicateId);
+
+	gxact = LockGXact(gid, GetUserId(), false);
+	pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
+	strcpy(gxact->state_3pc, state);
+
+	if (gxact->ondisk)
+		buf = ReadTwoPhaseFile(pgxact->xid, true);
+	else
+		XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
+	hdr = (TwoPhaseFileHeader *)buf;
+	strcpy(hdr->state_3pc, state);
+
+	START_CRIT_SECTION();
+	MyPgXact->delayChkpt = true;
+
+	hdr->origin_lsn = replorigin_session_origin_lsn;
+	hdr->origin_timestamp = replorigin_session_origin_timestamp;
+
+	XLogBeginInsert();
+	XLogRegisterData(buf, hdr->total_len - sizeof(pg_crc32c));
+	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+	gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
+
+	if (replorigin)
+		/* Move LSNs forward for this replication origin */
+		replorigin_session_advance(replorigin_session_origin_lsn,
+								   gxact->prepare_end_lsn);
+
+	XLogFlush(gxact->prepare_end_lsn);
+
+	gxact->prepare_start_lsn = ProcLastRecPtr;
+	MyPgXact->delayChkpt = false;
+
+	END_CRIT_SECTION();
+
+	PostPrepare_Twophase();
+}

src/backend/catalog/system_views.sql

@@ -283,7 +283,7 @@ CREATE VIEW pg_available_extension_versions AS
 
 CREATE VIEW pg_prepared_xacts AS
     SELECT P.transaction, P.gid, P.prepared,
-           U.rolname AS owner, D.datname AS database
+           U.rolname AS owner, D.datname AS database, P.state3pc AS state3pc
     FROM pg_prepared_xact() AS P
          LEFT JOIN pg_authid U ON P.ownerid = U.oid
          LEFT JOIN pg_database D ON P.dbid = D.oid;

src/backend/replication/logical/decode.c

@@ -657,7 +657,7 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 {
 	XLogRecPtr	origin_lsn = parsed->origin_lsn;
 	TimestampTz	commit_time = parsed->origin_timestamp;
-	XLogRecPtr	origin_id = XLogRecGetOrigin(buf->record);
+	RepOriginId	origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
 	TransactionId xid = parsed->twophase_xid;
 
@@ -666,7 +666,7 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	 * transaction's contents, since the various caches need to always be
 	 * consistent.
 	 */
-	if (parsed->nmsgs > 0)
+	if (parsed->nmsgs > 0 && *parsed->state_3pc == '\0')
 	{
 		ReorderBufferAddInvalidations(ctx->reorder, xid, buf->origptr,
 									  parsed->nmsgs, parsed->msgs);
@@ -686,6 +686,23 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		return;
 	}
 
+	/* Shorthack for PRECOMMIT */
+	if (*parsed->state_3pc != '\0')
+	{ 
+		ReorderBufferTXN txn;
+		txn.xid = xid;
+		txn.first_lsn = buf->origptr;
+		txn.final_lsn = buf->origptr;
+		txn.end_lsn = buf->endptr;
+		txn.commit_time = commit_time;
+		txn.origin_id = origin_id;
+		txn.origin_lsn = origin_lsn;
+		strcpy(txn.gid, parsed->twophase_gid);
+		strcpy(txn.state_3pc, parsed->state_3pc);
+		ctx->reorder->prepare(ctx->reorder, &txn, buf->origptr);
+		return;
+	}
+
 	/* tell the reorderbuffer about the surviving subtransactions */
 	for (i = 0; i < parsed->nsubxacts; i++)
 	{
@@ -695,7 +712,8 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 
 	/* replay actions of all transaction + subtransactions in order */
 	ReorderBufferPrepare(ctx->reorder, xid, buf->origptr, buf->endptr,
-						 commit_time, origin_id, origin_lsn, parsed->twophase_gid);
+						commit_time, origin_id, origin_lsn,
+						parsed->twophase_gid, parsed->state_3pc);
 }
 
 /*

src/backend/replication/logical/reorderbuffer.c

@@ -1768,7 +1768,7 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
 					XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
 					TimestampTz commit_time,
 					RepOriginId origin_id, XLogRecPtr origin_lsn,
-					char *gid)
+					char *gid, char *state_3pc)
 {
 	ReorderBufferTXN *txn;
 
@@ -1781,6 +1781,7 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
 
 	txn->txn_flags |= TXN_PREPARE;
 	strcpy(txn->gid, gid);
+	strcpy(txn->state_3pc, state_3pc);
 
 	ReorderBufferCommitInternal(txn, rb, xid, commit_lsn, end_lsn,
 								commit_time, origin_id, origin_lsn);

src/include/access/twophase.h

@@ -19,6 +19,14 @@
 #include "datatype/timestamp.h"
 #include "storage/lock.h"
 
+typedef struct PreparedTransactionData
+{
+	Oid			owner;
+	char		gid[GIDSIZE];	/* The GID assigned to the prepared xact */
+	char        state_3pc[MAX_3PC_STATE_SIZE]; /* 3PC transaction state  */	
+} PreparedTransactionData, *PreparedTransaction;
+
+
 /*
  * GlobalTransactionData is defined in twophase.c; other places have no
  * business knowing the internal definition.
@@ -57,6 +65,12 @@ extern void CheckPointTwoPhase(XLogRecPtr redo_horizon);
 extern void FinishPreparedTransaction(const char *gid, bool isCommit,
 									  bool missing_ok);
 
+extern int GetPreparedTransactions(PreparedTransaction* pxacts);
+
+extern void SetPreparedTransactionState(char const* gid, char const* state);
+
+extern bool GetPreparedTransactionState(char const* gid, char* state);
+
 extern void PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
 			   XLogRecPtr end_lsn, RepOriginId origin_id);
 extern void PrepareRedoRemove(TransactionId xid, bool giveWarning);

src/include/access/xact.h

@@ -29,6 +29,11 @@
  */
 #define GIDSIZE 200
 
+/*
+ * Maximal size of 3PC transaction state
+ */
+#define MAX_3PC_STATE_SIZE 16
+
 /*
  * Xact isolation levels
  */
@@ -342,6 +347,7 @@ typedef struct xl_xact_parsed_prepare
 
 	TransactionId twophase_xid;
 	char 		twophase_gid[GIDSIZE];
+	char		state_3pc[MAX_3PC_STATE_SIZE];
 
 	XLogRecPtr	origin_lsn;
 	TimestampTz origin_timestamp;