Commit670cb78

authored and

committed

Rewrite 2pc support

1 parent75b1bbc commit670cb78Copy full SHA for 670cb78

File tree

-11

lines changed

-11

lines changed

Lines changed: 3 additions & 4 deletions

Original file line number	Diff line number	Diff line change
`@@ -126,7 +126,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError);`
`126`	`126`	`staticvoidMtmVoteForTransaction(MtmTransState*ts);`
`127`	`127`
`128`	`128`	`staticHTAB*xid2state;`
`129`		`-staticHTAB*git2xid;`
	`129`	`+staticHTAB*gid2xid;`
`130`	`130`	`staticMtmCurrentTransdtmTx;`
`131`	`131`	`staticMtmState*dtm;`
`132`	`132`
`@@ -529,7 +529,6 @@ MtmBeginTransaction(MtmCurrentTrans* x)`
`529`	`529`	`elog(ERROR,"Multimaster node is not online: current status %s",MtmNodeStatusMnem[dtm->status]);`
`530`	`530`	`}`
`531`	`531`	`x->containsDML= false;`
`532`		`-x->isPrepared= false;`
`533`	`532`	`x->snapshot=MtmAssignCSN();`
`534`	`533`	`x->gtid.xid=InvalidTransactionId;`
`535`	`534`	`x->gid[0]='\0';`
`@@ -541,7 +540,8 @@ MtmBeginTransaction(MtmCurrentTrans* x)`
`541`	`540`	`}`
`542`	`541`
`543`	`542`	`/*`
`544`		`- * Prepare transaction for two-phase commit`
	`543`	`+ * Prepare transaction for two-phase commit.`
	`544`	`+ * This code is executed by PRE_PREPARE hook before PREPARE message is sent to replicas by logical replication`
`545`	`545`	`*/`
`546`	`546`	`MtmPrePrepareTransaction(MtmCurrentTrans*x)`
`547`	`547`	`{`
`@@ -575,7 +575,6 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)`
`575`	`575`	`ts->gtid=x->gtid;`
`576`	`576`	`ts->procno=MyProc->pgprocno;`
`577`	`577`	`ts->nVotes=0;`
`578`		`-ts->voteCompleted= false;`
`579`	`578`
`580`	`579`	`x->isPrepared= true;`
`581`	`580`	`x->csn=csn;`

Lines changed: 0 additions & 2 deletions

Original file line number	Diff line number	Diff line change
`@@ -84,11 +84,9 @@ typedef struct MtmTransState`
`84`	`84`	`finally should be nNodes-1 */`
`85`	`85`	`intprocno;/* pgprocno of transaction coordinator waiting for responses from replicas,`
`86`	`86`	`used to notify coordinator by arbiter */`
`87`		`-boolvoteCompleted;/* Responses necessary to make a decision are received by coordinator of transaction */`
`88`	`87`	`intnSubxids;/* Number of subtransanctions */`
`89`	`88`	`structMtmTransStatenextVoting;/ Next element in L1-list of voting transactions. */`
`90`	`89`	`structMtmTransStatenext;/ Next element in L1 list of all finished transaction present in xid2state hash */`
`91`		`-TransactionIdxids[1];/* transaction ID at replicas: varying size MtmNodes */`
`92`	`90`	`}MtmTransState;`
`93`	`91`
`94`	`92`	`typedefstruct`

Lines changed: 2 additions & 2 deletions

Original file line number	Diff line number	Diff line change
`@@ -499,7 +499,7 @@ process_remote_commit(StringInfo in)`
`499`	`499`	`CommitTransactionCommand();`
`500`	`500`	`StartTransactionCommand();`
`501`	`501`	`/* PREPARE itself */`
`502`		`-MtmSetCurrentTransactionGid(gid);`
	`502`	`+MtmSetCurrentTransactionGID(gid);`
`503`	`503`	`PrepareTransactionBlock(gid);`
`504`	`504`	`CommitTransactionCommand();`
`505`	`505`	`break;`
`@@ -519,7 +519,7 @@ process_remote_commit(StringInfo in)`
`519`	`519`	`{`
`520`	`520`	`gid=pq_getmsgstring(in);`
`521`	`521`	`StartTransactionCommand();`
`522`		`-MtmSetCurrentTransactionGid(gid);`
	`522`	`+MtmSetCurrentTransactionGID(gid);`
`523`	`523`	`FinishPreparedTransaction(gid, false);`
`524`	`524`	`CommitTransactionCommand();`
`525`	`525`	`break;`

Lines changed: 3 additions & 3 deletions

Original file line number	Diff line number	Diff line change
`@@ -122,12 +122,12 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,`
`122`	`122`
`123`	`123`
`124`	`124`	`staticcsn_t`
`125`		`-MtmGetCSN(TransactionIdxid)`
	`125`	`+MtmGetCSN(PGLogicalProtoMM*mm,TransactionIdxid)`
`126`	`126`	`{`
`127`	`127`	`MtmTransState*ts;`
`128`	`128`	`csn_tcsn;`
`129`	`129`	`MtmLock(LW_SHARED);`
`130`		`-ts= (MtmTransState*)hash_search(xid2state,&xid,HASH_FIND,NULL);`
	`130`	`+ts= (MtmTransState*)hash_search(mm->xid2state,&xid,HASH_FIND,NULL);`
`131`	`131`	`Assert(ts!=NULL);`
`132`	`132`	`csn=ts->csn;`
`133`	`133`	`MtmUnlock();`
`@@ -171,7 +171,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,`
`171`	`171`	`pq_sendint64(out,txn->commit_time);`
`172`	`172`
`173`	`173`	`if (txn->xact_action==XLOG_XACT_COMMIT_PREPARED) {`
`174`		`-pq_sendint64(out,MtmGetCSN(txn->xid));`
	`174`	`+pq_sendint64(out,MtmGetCSN(mm,txn->xid));`
`175`	`175`	`}`
`176`	`176`	`if (txn->xact_action==XLOG_XACT_PREPARE\|\|`
`177`	`177`	`txn->xact_action==XLOG_XACT_COMMIT_PREPARED\|\|`

Comments

(0)