Move xact static metadata (node id, generation, etc) from gid to state_3pc.

Prepares the ground for explicit 2PC: changing on the fly user provided gid
would be ugly and inefficient, so let's move everything out.

Author: arssher

src/commit.c

@@ -277,25 +277,17 @@ MtmBeginTransaction()
 }
 
 /*
- * Genenerate global transaction identifier for two-phase commit.
- * It should be unique for all nodes.
- * MTM-node_id-xid is part ensuring uniqueness; the rest is necessary payload.
- * gen_num identifies the number of generation xact belongs to.
- * configured is mask of configured nodes of this generation; it is required
- * by resolver.
- * (in theory we could get rid of it if we remembered generation history for
- *  some time, but we don't currently)
- *
- * Beware that GlobalTxGCInTableProposals parses gid from SQL.
- *
- * TODO: add version and kinda backwards compatibility.
+ * Generate global transaction identifier for two-phase commit.
+ * It should be unique for all nodes. This is basically the only requirement;
+ * we have some important metadata associated with xact, but we put
+ * everything into state_3pc for seamless integration with explicit 2PC.
+ * (we still keep gen_num here for logging)
  */
 void
-MtmGenerateGid(char *gid, int node_id, TransactionId xid, uint64 gen_num,
-			   nodemask_t configured)
+MtmGenerateGid(char *gid, int node_id, TransactionId xid, uint64 gen_num)
 {
-	sprintf(gid, "MTM-%d-" XID_FMT "-%" INT64_MODIFIER "X-%" INT64_MODIFIER "X",
-			node_id, xid, gen_num, configured);
+	sprintf(gid, "MTM-%d-" XID_FMT "-" UINT64_FORMAT,
+			node_id, xid, gen_num);
 	return;
 }
 
@@ -329,20 +321,6 @@ MtmGidParseXid(const char *gid)
 	return xid;
 }
 
-/* parse generation configured mask from gid */
-nodemask_t
-MtmGidParseConfigured(const char *gid)
-{
-	TransactionId xid;
-	uint64 gen_num;
-	nodemask_t configured = 0;
-
-	sscanf(gid, "MTM-%*d-" XID_FMT "-%" INT64_MODIFIER "X-%" INT64_MODIFIER "X",
-		   &xid, &gen_num, &configured);
-	Assert(configured != 0);
-	return configured;
-}
-
 /* ensure we get the right PREPARE ack */
 static bool
 PrepareGatherHook(MtmMessage *anymsg, Datum arg)
@@ -385,6 +363,7 @@ MtmTwoPhaseCommit(void)
 	nodemask_t	pc_success_cohort;
 	MtmGeneration xact_gen;
 	char dmq_stream_name[DMQ_STREAM_NAME_MAXLEN];
+	GTxState gtx_state;
 
 	if (MtmNo3PC)
 	{
@@ -449,28 +428,33 @@ MtmTwoPhaseCommit(void)
 		xact_gen = MtmGetCurrentGen(true);
 		xid = GetTopTransactionId();
 		MtmGenerateGid(mtm_commit_state.gid, mtm_cfg->my_node_id, xid,
-					   xact_gen.num, xact_gen.configured);
+					   xact_gen.num);
 		sprintf(dmq_stream_name, "xid" XID_FMT, xid);
 		dmq_stream_subscribe(dmq_stream_name);
 		mtm_log(MtmTxTrace, "%s subscribed for %s", mtm_commit_state.gid,
 				dmq_stream_name);
 
 		/* prepare transaction on our node */
-
 		mtm_commit_state.gtx = GlobalTxAcquire(mtm_commit_state.gid, true,
-											   false, NULL);
+											   false, NULL, 0);
 		/*
 		 * it is simpler to mark gtx originated here as orphaned from the
 		 * beginning rather than in error handler; resolver won't touch gtx
 		 * while it is locked on us anyway
 		 */
 		mtm_commit_state.gtx->orphaned = true;
+		mtm_commit_state.gtx->xinfo.coordinator = mtm_cfg->my_node_id;
+		mtm_commit_state.gtx->xinfo.xid = xid;
+		mtm_commit_state.gtx->xinfo.gen_num = xact_gen.num;
+		mtm_commit_state.gtx->xinfo.configured = xact_gen.configured;
 		Assert(mtm_commit_state.gtx->state.status == GTXInvalid);
 		/*
 		 * PREPARE doesn't happen here; ret 0 just means we were already in
 		 * aborted transaction block and we expect the callee to handle this.
 		 */
-		ret = PrepareTransactionBlock(mtm_commit_state.gid);
+		ret = PrepareTransactionBlockWithState3PC(
+			mtm_commit_state.gid,
+			serialize_xstate(&mtm_commit_state.gtx->xinfo, &mtm_commit_state.gtx->state));
 		if (!ret)
 		{
 			Assert(false);
@@ -559,18 +543,18 @@ MtmTwoPhaseCommit(void)
 		}
 
 		/* ok, we have all prepare responses, precommit */
-		SetPreparedTransactionState(mtm_commit_state.gid,
-			serialize_gtx_state(GTXPreCommitted,
-								InitialGTxTerm,
-								InitialGTxTerm),
+		gtx_state.status = GTXPreCommitted;
+		gtx_state.proposal = InitialGTxTerm;
+		gtx_state.accepted = InitialGTxTerm;
+		SetPreparedTransactionState(
+			mtm_commit_state.gid,
+			serialize_xstate(&mtm_commit_state.gtx->xinfo, &gtx_state),
 			false);
 		/*
 		 * since this moment direct aborting is not allowed; others can
 		 * receive our precommit and resolve xact to commit without us
 		 */
-		mtm_commit_state.gtx->state.status = GTXPreCommitted;
-		mtm_commit_state.gtx->state.proposal = InitialGTxTerm;
-		mtm_commit_state.gtx->state.accepted = InitialGTxTerm;
+		mtm_commit_state.gtx->state = gtx_state;
 		mtm_log(MtmTxFinish, "TXFINISH: %s precommitted", mtm_commit_state.gid);
 
 		/*

src/global_tx.c

@@ -67,78 +67,81 @@ term_cmp(GlobalTxTerm t1, GlobalTxTerm t2)
 		return 1;
 }
 
+#define XStateVersion 1
 
 char *
-serialize_gtx_state(GlobalTxStatus status, GlobalTxTerm term_prop, GlobalTxTerm term_acc)
+serialize_xstate(XactInfo *xinfo, GTxState *gtx_state)
 {
 	char	   *state;
 	char	   *status_abbr;
 
-	if (status == GTXInvalid)
+	if (gtx_state->status == GTXInvalid)
 		status_abbr = "in";
-	else if (status == GTXPreCommitted)
+	else if (gtx_state->status == GTXPreCommitted)
 		status_abbr = "pc";
-	else if (status == GTXPreAborted)
+	else if (gtx_state->status == GTXPreAborted)
 		status_abbr = "pa";
-	else if (status == GTXCommitted)
+	else if (gtx_state->status == GTXCommitted)
 		status_abbr = "cm";
 	else
 	{
-		Assert(status == GTXAborted);
+		Assert(gtx_state->status == GTXAborted);
 		status_abbr = "ab";
 	}
 
-	state = psprintf("%s-%d:%d-%d:%d",
+	state = psprintf("%d-%d-" XID_FMT "-%" INT64_MODIFIER "X-%" INT64_MODIFIER "X-%s-%d:%d-%d:%d",
+					 XStateVersion,
+					 xinfo->coordinator,
+					 xinfo->xid,
+					 xinfo->gen_num,
+					 xinfo->configured,
 					 status_abbr,
-					 term_prop.ballot, term_prop.node_id,
-					 term_acc.ballot, term_acc.node_id);
+					 gtx_state->proposal.ballot, gtx_state->proposal.node_id,
+					 gtx_state->accepted.ballot, gtx_state->accepted.node_id);
 	return state;
 }
 
-void
-parse_gtx_state(const char *state, GlobalTxStatus *status,
-				GlobalTxTerm *term_prop, GlobalTxTerm *term_acc)
+/* returns 0 on success */
+int
+deserialize_xstate(const char *state, XactInfo *xinfo, GTxState *gtx_state,
+				   int elevel)
 {
 	int			n_parsed = 0;
+	char		status_abbr[3]; /* must be big enough for '\0' */
 
 	Assert(state);
 
-	/*
-	 * Might be immediately after PrepareTransaction. It would be better to
-	 * also pass state_3pc to it, but...
-	 */
-	if (state[0] == '\0')
+	n_parsed = sscanf(state, "%*d-%d-" XID_FMT "-%" INT64_MODIFIER "X-%" INT64_MODIFIER "X-%2s-%d:%d-%d:%d",
+					  &xinfo->coordinator,
+					  &xinfo->xid,
+					  &xinfo->gen_num,
+					  &xinfo->configured,
+					  status_abbr,
+					  &gtx_state->proposal.ballot,
+					  &gtx_state->proposal.node_id,
+					  &gtx_state->accepted.ballot,
+					  &gtx_state->accepted.node_id);
+	if (n_parsed != 9)
 	{
-		*status = GTXInvalid;
-		*term_prop = InitialGTxTerm;
-		*term_acc = InvalidGTxTerm;
+		mtm_log(elevel, "GlobalTxLoadAll: failed to deparse state_3pc %s, ignoring it (res=%d)",
+				state, n_parsed);
+		return n_parsed;
 	}
+
+	if (strncmp(status_abbr, "in", 2) == 0)
+		gtx_state->status = GTXInvalid;
+	else if (strncmp(status_abbr, "pc", 2) == 0)
+		gtx_state->status = GTXPreCommitted;
+	else if (strncmp(status_abbr, "pa", 2) == 0)
+		gtx_state->status = GTXPreAborted;
+	else if (strncmp(status_abbr, "cm", 3) == 0)
+		gtx_state->status = GTXCommitted;
 	else
 	{
-		if (strncmp(state, "in-", 3) == 0)
-			*status = GTXInvalid;
-		else if (strncmp(state, "pc-", 3) == 0)
-			*status = GTXPreCommitted;
-		else if (strncmp(state, "pa-", 3) == 0)
-			*status = GTXPreAborted;
-		else if (strncmp(state, "cm-", 3) == 0)
-			*status = GTXCommitted;
-		else
-		{
-			Assert((strncmp(state, "ab-", 3) == 0));
-			*status = GTXAborted;
-		}
-
-		n_parsed = sscanf(state + 3, "%d:%d-%d:%d",
-						&term_prop->ballot, &term_prop->node_id,
-						&term_acc->ballot, &term_acc->node_id);
-
-		if (n_parsed != 4)
-		{
-			Assert(false);
-			mtm_log(PANIC, "wrong state_3pc format: %s", state);
-		}
+		Assert((strncmp(status_abbr, "ab", 2) == 0));
+		gtx_state->status = GTXAborted;
 	}
+	return 0;
 }
 
 void
@@ -219,10 +222,11 @@ GlobalTxEnsureBeforeShmemExitHook(void)
  * If nowait_own_live is true, gtx is already locked, I am the coordinator and
  * gtx is not orphaned, don't wait for release -- backend is still working on
  * xact, which may be very long. *busy (if provided) is set to true in this
- * case.
+ * case. coordinator must be passed for this to work.
  */
 GlobalTx *
-GlobalTxAcquire(const char *gid, bool create, bool nowait_own_live, bool *busy)
+GlobalTxAcquire(const char *gid, bool create, bool nowait_own_live, bool *busy,
+				int coordinator)
 {
 	GlobalTx   *gtx = NULL;
 	bool		found;
@@ -274,9 +278,7 @@ GlobalTxAcquire(const char *gid, bool create, bool nowait_own_live, bool *busy)
 
 		if (nowait_own_live)
 		{
-			int			tx_node_id = MtmGidParseNodeId(gid);
-
-			if (tx_node_id == Mtm->my_node_id && !gtx->orphaned)
+			if (coordinator == Mtm->my_node_id && !gtx->orphaned)
 			{
 				if (busy)
 					*busy = true;
@@ -374,8 +376,15 @@ GlobalTxLoadAll()
 		Assert(!found);
 
 		gtx->acquired_by = InvalidBackendId;
-		parse_gtx_state(pxacts[i].state_3pc, &gtx->state.status,
-						&gtx->state.proposal, &gtx->state.accepted);
+		/*
+		 * Allow instance to start even if we have problems parsing xstate...
+		 */
+		if (deserialize_xstate(pxacts[i].state_3pc, &gtx->xinfo, &gtx->state,
+							   WARNING) != 0)
+		{
+			hash_search(gtx_shared->gid2gtx, pxacts[i].gid, HASH_REMOVE, &found);
+			continue;
+		}
 		gtx->prepared = true;
 		gtx->orphaned = true;
 		gtx->resolver_stage = GTRS_AwaitStatus;
@@ -420,8 +429,7 @@ GlobalTxMarkOrphaned(int node_id)
 	hash_seq_init(&hash_seq, gtx_shared->gid2gtx);
 	while ((gtx = hash_seq_search(&hash_seq)) != NULL)
 	{
-		int			tx_node_id = MtmGidParseNodeId(gtx->gid);
-		if (tx_node_id == node_id)
+		if (gtx->xinfo.coordinator == node_id)
 		{
 			gtx->orphaned = true;
 			mtm_log(MtmTxTrace, "%s is orphaned", gtx->gid);

src/include/commit.h

@@ -17,12 +17,11 @@
 
 #include "messaging.h"
 
-extern void MtmGenerateGid(char *gid, int node_id, TransactionId xid, uint64 gen_num,
-						   nodemask_t configured);
+extern void MtmGenerateGid(char *gid, int node_id, TransactionId xid,
+						   uint64 gen_num);
 extern uint64 MtmGidParseGenNum(const char *gid);
 extern int	MtmGidParseNodeId(const char *gid);
 extern TransactionId MtmGidParseXid(const char *gid);
-extern nodemask_t MtmGidParseConfigured(const char *gid);
 
 extern bool MtmTwoPhaseCommit(void);
 extern void MtmBeginTransaction(void);

src/include/global_tx.h

@@ -45,14 +45,38 @@ typedef enum
 
 typedef struct
 {
-	GlobalTxTerm	proposal; /* nextBal in terms of The Part-Time Paliament */
-	GlobalTxTerm	accepted; /* prevBal in terms of The Part-Time Paliament */
-	GlobalTxStatus	status; /* prevDec in terms of The Part-Time Paliament */
+	GlobalTxTerm	proposal; /* nextBal in terms of The Part-Time Parliament */
+	GlobalTxTerm	accepted; /* prevBal in terms of The Part-Time Parliament */
+	GlobalTxStatus	status; /*
+							 * prevDec in terms of The Part-Time Parliament
+							 * (or special never voted | commit | abort)
+							 */
 } GTxState;
 
+/*
+ * Constant xact metadata which we encode into state_3pc. We could (and
+ * previously did) carry that directly in gid, but this intervenes with
+ * explicit 2PC usage: applier must know generation of the xact, and
+ * scribbling over user-provided gid is ugly and/or inefficient.
+ */
+typedef struct
+{
+	int coordinator; /* node id who initiated the transaction */
+	TransactionId xid; /* xid at coordinator */
+	uint64 gen_num; /* the number of generation xact belongs to */
+	nodemask_t configured; /* mask of configured nodes of this generation;
+							* the idea was to use this by resolver, but it
+							* wasn't finished. We shouldn't have any problems
+							* with this anyway if all xacts created before
+							* first node add-rm are resolved before the
+							* second one is started
+							*/
+} XactInfo;
+
 typedef struct GlobalTx
 {
 	char		gid[GIDSIZE];
+	XactInfo	xinfo;
 	XLogRecPtr	coordinator_end_lsn;
 	BackendId	acquired_by;
 	/* paxos voting state for this xact */
@@ -86,21 +110,18 @@ void MtmGlobalTxInit(void);
 void MtmGlobalTxShmemStartup(void);
 void GlobalTxEnsureBeforeShmemExitHook(void);
 GlobalTx *GlobalTxAcquire(const char *gid, bool create, bool nowait_own_live,
-						  bool *busy);
+						  bool *busy, int coordinator);
 void GlobalTxRelease(GlobalTx *gtx);
 void GlobalTxAtExit(int code, Datum arg);
 void GlobalTxLoadAll(void);
-char *serialize_gtx_state(GlobalTxStatus status, GlobalTxTerm term_prop,
-						  GlobalTxTerm term_acc);
+char *serialize_xstate(XactInfo *xinfo, GTxState *gtx_state);
 int term_cmp(GlobalTxTerm t1, GlobalTxTerm t2);
-void parse_gtx_state(const char *state, GlobalTxStatus *status,
-				GlobalTxTerm *term_prop, GlobalTxTerm *term_acc);
+int deserialize_xstate(const char *state, XactInfo *xinfo, GTxState *gtx_state,
+					   int elevel);
 GlobalTxTerm GlobalTxGetMaxProposal(void);
 void GlobalTxSaveInTable(const char *gid, XLogRecPtr coordinator_end_lsn,
 						 GlobalTxStatus status,
 						 GlobalTxTerm term_prop, GlobalTxTerm term_acc);
-void GlobalTxDeleteFromTable(const char *gid);
-void GlobalTxGCInTableProposals(void);
 void GlobalTxMarkOrphaned(int node_id);
 
 char *GlobalTxToString(GlobalTx *gtx);

src/include/messaging.h

@@ -97,6 +97,8 @@ typedef struct
 	MtmTxRequestValue	type;
 	GlobalTxTerm		term;
 	const char		   *gid;
+	int					coordinator;
+	uint64				gen_num;
 	XLogRecPtr			coordinator_end_lsn; /* matters for 1a */
 } MtmTxRequest;