Location via proxy:   [ UP ]  
[Report a bug]   [Manage cookies]                
Skip to content

Commit af5645e

Browse files
knizhnikkelvich
knizhnik
authored and
kelvich
committed
Restore skip local transaction mechanism in pglogical_proto
1 parent c39b549 commit af5645e

File tree

4 files changed

+59
-55
lines changed

4 files changed

+59
-55
lines changed

‎arbiter.c

Lines changed: 11 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -898,14 +898,13 @@ static void MtmReceiver(Datum arg)
898898
msg->status = TRANSACTION_STATUS_ABORTED;
899899
} else {
900900
msg->status = tm->state->status;
901-
msg->csn = tm->state->csn;
901+
msg->csn = tm->state->csn;
902902
MTM_LOG1("Send response %d for transaction %s to node %d", msg->status, msg->gid, msg->node);
903903
}
904904
msg->disabledNodeMask = Mtm->disabledNodeMask;
905905
msg->connectivityMask = Mtm->connectivityMask;
906906
msg->oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
907907
msg->code = MSG_POLL_STATUS;
908-
msg->csn = ts->csn;
909908
MtmSendMessage(msg);
910909
continue;
911910
case MSG_POLL_STATUS:
@@ -918,11 +917,11 @@ static void MtmReceiver(Datum arg)
918917
BIT_SET(ts->votedMask, node-1);
919918
if (ts->status == TRANSACTION_STATUS_UNKNOWN) {
920919
if (msg->status == TRANSACTION_STATUS_IN_PROGRESS || msg->status == TRANSACTION_STATUS_ABORTED) {
921-
elog(LOG, "Abort transaction %s because it is in state %d at node %d",
922-
msg->gid, ts->status, node);
920+
elog(LOG, "Abort prepared transaction %s because it is in state %s at node %d",
921+
msg->gid, MtmNodeStatusMnem[msg->status], node);
923922
MtmFinishPreparedTransaction(ts, false);
924923
}
925-
else if (msg->status == TRANSACTION_STATUS_COMMITTED || msg->status == TRANSACTION_STATUS_UNKNOWN)
924+
else if (msg->status == TRANSACTION_STATUS_COMMITTED || msg->status == TRANSACTION_STATUS_UNKNOWN)
926925
{
927926
if (msg->csn > ts->csn) {
928927
ts->csn = msg->csn;
@@ -933,17 +932,17 @@ static void MtmReceiver(Datum arg)
933932
MtmFinishPreparedTransaction(ts, true);
934933
}
935934
} else {
936-
elog(LOG, "Receive response %d for transaction %s for node %d, votedMask=%llx, participantsMask=%llx",
937-
msg->status, msg->gid, node, (long long) ts->votedMask, (long long) (ts->participantsMask & ~Mtm->disabledNodeMask));
935+
elog(LOG, "Receive response %s for transaction %s for node %d, votedMask %llx, participantsMask %llx",
936+
MtmNodeStatusMnem[msg->status], msg->gid, node, (long long)ts->votedMask, (long long)(ts->participantsMask & ~Mtm->disabledNodeMask));
938937
continue;
939938
}
940939
} else if (ts->status == TRANSACTION_STATUS_ABORTED && msg->status == TRANSACTION_STATUS_COMMITTED) {
941940
elog(WARNING, "Transaction %s is aborted at node %d but committed at node %d", msg->gid, MtmNodeId, node);
942941
} else if (msg->status == TRANSACTION_STATUS_ABORTED && ts->status == TRANSACTION_STATUS_COMMITTED) {
943942
elog(WARNING, "Transaction %s is committed at node %d but aborted at node %d", msg->gid, MtmNodeId, node);
944943
} else {
945-
elog(LOG, "Receive response %d for transaction %s status %d for node %d, votedMask=%llx, participantsMask=%llx",
946-
msg->status, msg->gid, ts->status, node, (long long) ts->votedMask, (long long) (ts->participantsMask & ~Mtm->disabledNodeMask) );
944+
elog(LOG, "Receive response %s for transaction %s status %s for node %d, votedMask %llx, participantsMask %llx",
945+
MtmNodeStatusMnem[msg->status], msg->gid, MtmNodeStatusMnem[ts->status], node, (long long)ts->votedMask, (long long)(ts->participantsMask & ~Mtm->disabledNodeMask) );
947946
}
948947
}
949948
continue;
@@ -983,8 +982,8 @@ static void MtmReceiver(Datum arg)
983982
if ((~msg->disabledNodeMask & Mtm->disabledNodeMask) != 0) {
984983
/* Coordinator's disabled mask is wider than of this node: so reject such transaction to avoid
985984
commit on smaller subset of nodes */
986-
elog(WARNING, "Coordinator of distributed transaction see less nodes than node %d: %lx instead of %lx",
987-
node, (long) Mtm->disabledNodeMask, (long) msg->disabledNodeMask);
985+
elog(WARNING, "Coordinator of distributed transaction see less nodes than node %d: %llx instead of %llx",
986+
node, (long long) Mtm->disabledNodeMask, (long long) msg->disabledNodeMask);
988987
MtmAbortTransaction(ts);
989988
}
990989
if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
@@ -993,7 +992,7 @@ static void MtmReceiver(Datum arg)
993992
MtmWakeUpBackend(ts);
994993
} else {
995994
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
996-
MTM_LOG1("Transaction %s is prepared (status=%d participants=%lx disabled=%lx, voted=%lx)",
995+
MTM_LOG2("Transaction %s is prepared (status=%d participants=%lx disabled=%lx, voted=%lx)",
997996
ts->gid, ts->status, ts->participantsMask, Mtm->disabledNodeMask, ts->votedMask);
998997
ts->isPrepared = true;
999998
if (ts->isTwoPhase) {

‎multimaster.c

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1154,17 +1154,18 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
11541154
if (x->isPrepared) {
11551155
ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
11561156
Assert(ts != NULL);
1157+
Assert(strcmp(x->gid, ts->gid) == 0);
11571158
} else if (x->gid[0]) {
11581159
MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_FIND, NULL);
11591160
if (tm != NULL) {
11601161
ts = tm->state;
11611162
} else {
1162-
MTM_LOG3("%d: GID %s not found", MyProcPid, x->gid);
1163+
MTM_LOG1("%d: GID %s not found", MyProcPid, x->gid);
11631164
}
11641165
}
11651166
if (ts != NULL) {
11661167
if (*ts->gid)
1167-
MTM_LOG1("TRANSLOG: %s transaction git=%s xid=%d node=%d dxid=%d status %d",
1168+
MTM_LOG2("TRANSLOG: %s transaction gid=%s xid=%d node=%d dxid=%d status %d",
11681169
(commit ? "commit" : "rollback"), ts->gid, ts->xid, ts->gtid.node, ts->gtid.xid, ts->status);
11691170
if (commit) {
11701171
if (!(ts->status == TRANSACTION_STATUS_UNKNOWN
@@ -1177,6 +1178,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
11771178
ts->xid, ts->gid);
11781179
}
11791180
if (x->csn > ts->csn || Mtm->status == MTM_RECOVERY) {
1181+
Assert(x->csn != INVALID_CSN);
11801182
ts->csn = x->csn;
11811183
MtmSyncClock(ts->csn);
11821184
}
@@ -2844,7 +2846,7 @@ void MtmRollbackPreparedTransaction(int nodeId, char const* gid)
28442846
XidStatus status = MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED);
28452847
MTM_LOG1("Abort prepared transaction %s status %d", gid, status);
28462848
if (status == TRANSACTION_STATUS_UNKNOWN) {
2847-
MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
2849+
MTM_LOG2("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
28482850
MtmResetTransaction();
28492851
StartTransactionCommand();
28502852
MtmBeginSession(nodeId);
@@ -3193,9 +3195,7 @@ bool MtmFilterTransaction(char* record, int size)
31933195
duplicate = true;
31943196
}
31953197

3196-
//duplicate = Mtm->status == MTM_RECOVERY && origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLSN;
3197-
3198-
MTM_LOG1("%s transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx",
3198+
MTM_LOG2("%s transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx",
31993199
duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, flags, origin_node, origin_lsn, restart_lsn);
32003200
return duplicate;
32013201
}

‎pglogical_apply.c

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -641,7 +641,7 @@ process_remote_commit(StringInfo in)
641641
AbortCurrentTransaction();
642642
} else {
643643
/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
644-
MTM_LOG1("PGLOGICAL_PREPARE commit: gid=%s", gid);
644+
MTM_LOG2("PGLOGICAL_PREPARE commit: gid=%s", gid);
645645
BeginTransactionBlock();
646646
CommitTransactionCommand();
647647
StartTransactionCommand();
@@ -668,7 +668,7 @@ process_remote_commit(StringInfo in)
668668
Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
669669
csn = pq_getmsgint64(in);
670670
gid = pq_getmsgstring(in);
671-
MTM_LOG1("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%lx", csn, gid, end_lsn);
671+
MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%lx", csn, gid, end_lsn);
672672
MtmResetTransaction();
673673
StartTransactionCommand();
674674
MtmBeginSession(origin_node);

‎pglogical_proto.c

Lines changed: 40 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,9 @@
3939
#include "pglogical_relid_map.h"
4040

4141
static int MtmTransactionRecords;
42+
static bool MtmIsFilteredTxn;
4243
static TransactionId MtmCurrentXid;
43-
static bool DDLInProress = false;
44+
static bool DDLInProgress = false;
4445

4546
static void pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel);
4647

@@ -76,13 +77,13 @@ pglogical_write_rel(StringInfo out, PGLogicalOutputData *data, Relation rel)
7677
uint8 relnamelen;
7778
Oid relid;
7879

79-
if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN) {
80+
if (MtmIsFilteredTxn) {
8081
MTM_LOG2("%d: pglogical_write_message filtered", MyProcPid);
8182
return;
8283
}
8384

84-
if (DDLInProress) {
85-
MTM_LOG2("%d: pglogical_write_message filtered DDLInProress", MyProcPid);
85+
if (DDLInProgress) {
86+
MTM_LOG2("%d: pglogical_write_message filtered DDLInProgress", MyProcPid);
8687
return;
8788
}
8889

@@ -116,17 +117,22 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
116117
bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
117118
csn_t csn = MtmTransactionSnapshot(txn->xid);
118119

119-
MtmCurrentXid = txn->xid;
120-
121-
MTM_LOG3("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx",
122-
MyProcPid, txn->xid, MtmReplicationNodeId, csn, isRecovery, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
123-
124-
MTM_LOG3("%d: pglogical_write_begin XID=%d sent", MyProcPid, txn->xid);
125-
pq_sendbyte(out, 'B'); /* BEGIN */
126-
pq_sendint(out, MtmNodeId, 4);
127-
pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
128-
pq_sendint64(out, csn);
129-
MtmTransactionRecords = 0;
120+
if (!isRecovery && csn == INVALID_CSN) {
121+
MtmIsFilteredTxn = true;
122+
MTM_LOG3("%d: pglogical_write_begin XID=%d filtered", MyProcPid, txn->xid);
123+
} else {
124+
MtmCurrentXid = txn->xid;
125+
MtmIsFilteredTxn = false;
126+
MTM_LOG3("%d: pglogical_write_begin XID=%d node=%d CSN=%ld recovery=%d restart_decoding_lsn=%lx first_lsn=%lx end_lsn=%lx confirmed_flush=%lx",
127+
MyProcPid, txn->xid, MtmReplicationNodeId, csn, isRecovery, txn->restart_decoding_lsn, txn->first_lsn, txn->end_lsn, MyReplicationSlot->data.confirmed_flush);
128+
129+
MTM_LOG3("%d: pglogical_write_begin XID=%d sent", MyProcPid, txn->xid);
130+
pq_sendbyte(out, 'B'); /* BEGIN */
131+
pq_sendint(out, MtmNodeId, 4);
132+
pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
133+
pq_sendint64(out, csn);
134+
MtmTransactionRecords = 0;
135+
}
130136
}
131137

132138
static void
@@ -142,14 +148,14 @@ pglogical_write_message(StringInfo out,
142148
}
143149
break;
144150
case 'D':
145-
if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN) {
151+
if (MtmIsFilteredTxn) {
146152
MTM_LOG2("%d: pglogical_write_message filtered", MyProcPid);
147153
return;
148154
}
149-
DDLInProress = true;
155+
DDLInProgress = true;
150156
break;
151157
case 'E':
152-
DDLInProress = false;
158+
DDLInProgress = false;
153159
/*
154160
* we use End message only as indicator of DDL transaction finish,
155161
* so no need to send that to replicas.
@@ -187,11 +193,10 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
187193
Assert(false);
188194

189195
if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) {
190-
//Assert(txn->xid < 1000 || MtmTransactionRecords != 1);
191-
// if (MtmIsFilteredTxn) {
192-
// Assert(MtmTransactionRecords == 0);
193-
// return;
194-
// }
196+
if (MtmIsFilteredTxn) {
197+
Assert(MtmTransactionRecords == 0);
198+
return;
199+
}
195200
} else {
196201
csn_t csn = MtmTransactionSnapshot(txn->xid);
197202
bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
@@ -261,13 +266,13 @@ static void
261266
pglogical_write_insert(StringInfo out, PGLogicalOutputData *data,
262267
Relation rel, HeapTuple newtuple)
263268
{
264-
if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
269+
if (MtmIsFilteredTxn) {
265270
MTM_LOG2("%d: pglogical_write_insert filtered", MyProcPid);
266271
return;
267272
}
268273

269-
if (DDLInProress) {
270-
MTM_LOG2("%d: pglogical_write_insert filtered DDLInProress", MyProcPid);
274+
if (DDLInProgress) {
275+
MTM_LOG2("%d: pglogical_write_insert filtered DDLInProgress", MyProcPid);
271276
return;
272277
}
273278

@@ -284,13 +289,13 @@ static void
284289
pglogical_write_update(StringInfo out, PGLogicalOutputData *data,
285290
Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
286291
{
287-
if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
292+
if (MtmIsFilteredTxn) {
288293
MTM_LOG2("%d: pglogical_write_update filtered", MyProcPid);
289294
return;
290295
}
291296

292-
if (DDLInProress) {
293-
MTM_LOG2("%d: pglogical_write_update filtered DDLInProress", MyProcPid);
297+
if (DDLInProgress) {
298+
MTM_LOG2("%d: pglogical_write_update filtered DDLInProgress", MyProcPid);
294299
return;
295300
}
296301

@@ -317,13 +322,13 @@ static void
317322
pglogical_write_delete(StringInfo out, PGLogicalOutputData *data,
318323
Relation rel, HeapTuple oldtuple)
319324
{
320-
if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
325+
if (MtmIsFilteredTxn) {
321326
MTM_LOG2("%d: pglogical_write_delete filtered", MyProcPid);
322327
return;
323328
}
324329

325-
if (DDLInProress) {
326-
MTM_LOG2("%d: pglogical_write_delete filtered DDLInProress", MyProcPid);
330+
if (DDLInProgress) {
331+
MTM_LOG2("%d: pglogical_write_delete filtered DDLInProgress", MyProcPid);
327332
return;
328333
}
329334

@@ -354,13 +359,13 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
354359
int i;
355360
uint16 nliveatts = 0;
356361

357-
if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN){
362+
if (MtmIsFilteredTxn) {
358363
MTM_LOG2("%d: pglogical_write_tuple filtered", MyProcPid);
359364
return;
360365
}
361366

362-
if (DDLInProress) {
363-
MTM_LOG2("%d: pglogical_write_tuple filtered DDLInProress", MyProcPid);
367+
if (DDLInProgress) {
368+
MTM_LOG2("%d: pglogical_write_tuple filtered DDLInProgress", MyProcPid);
364369
return;
365370
}
366371

0 commit comments

Comments
 (0)