Skip to content

Commit 45945a8

Browse files
knizhnikknizhnik
committed
Change flush position reporting method
1 parent 08994d4 commit 45945a8

File tree

5 files changed

+67
-8
lines changed

5 files changed

+67
-8
lines changed

‎contrib/mmts/multimaster.c

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,8 @@ static bool MtmIsRecoverySession;
147147

148148
static MtmCurrentTrans MtmTx;
149149

150+
static dlist_head MtmLsnMapping = DLIST_STATIC_INIT(MtmLsnMapping);
151+
150152
static TransactionManager MtmTM = {
151153
PgTransactionIdGetStatus,
152154
PgTransactionIdSetTreeStatus,
@@ -1033,6 +1035,7 @@ void MtmHandleApplyError(void)
10331035
kill(PostmasterPid, SIGQUIT);
10341036
break;
10351037
}
1038+
FreeErrorData(edata);
10361039
}
10371040

10381041

@@ -1507,6 +1510,7 @@ static void MtmInitialize()
15071510
Mtm->nodes[i].transDelay = 0;
15081511
Mtm->nodes[i].lastStatusChangeTime = time(NULL);
15091512
Mtm->nodes[i].con = MtmConnections[i];
1513+
Mtm->nodes[i].flushPos = 0;
15101514
}
15111515
PGSemaphoreCreate(&Mtm->votingSemaphore);
15121516
PGSemaphoreReset(&Mtm->votingSemaphore);
@@ -2084,6 +2088,45 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
20842088
on_shmem_exit(MtmOnProcExit, 0);
20852089
}
20862090

2091+
XLogRecPtr MtmGetFlushPosition(int nodeId)
2092+
{
2093+
return Mtm->nodes[nodeId-1].flushPos;
2094+
}
2095+
2096+
void MtmUpdateLsnMapping(int node_id, XLogRecPtr end_lsn)
2097+
{
2098+
dlist_mutable_iter iter;
2099+
MtmFlushPosition* flushpos;
2100+
XLogRecPtr local_flush = GetFlushRecPtr();
2101+
MemoryContext old_context = MemoryContextSwitchTo(TopMemoryContext);
2102+
2103+
/* Track commit lsn */
2104+
flushpos = (MtmFlushPosition *) palloc(sizeof(MtmFlushPosition));
2105+
flushpos->node_id = node_id;
2106+
flushpos->local_end = XactLastCommitEnd;
2107+
flushpos->remote_end = end_lsn;
2108+
dlist_push_tail(&MtmLsnMapping, &flushpos->node);
2109+
2110+
MtmLock(LW_EXCLUSIVE);
2111+
dlist_foreach_modify(iter, &MtmLsnMapping)
2112+
{
2113+
flushpos = dlist_container(MtmFlushPosition, node, iter.cur);
2114+
if (flushpos->local_end <= local_flush)
2115+
{
2116+
if (Mtm->nodes[node_id-1].flushPos < local_flush) {
2117+
Mtm->nodes[node_id-1].flushPos = local_flush;
2118+
}
2119+
dlist_delete(iter.cur);
2120+
pfree(flushpos);
2121+
} else {
2122+
break;
2123+
}
2124+
}
2125+
MtmUnlock();
2126+
MemoryContextSwitchTo(old_context);
2127+
}
2128+
2129+
20872130
static void
20882131
MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
20892132
{

‎contrib/mmts/multimaster.h

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@ typedef struct
124124
MtmConnectionInfo con;
125125
time_t transDelay;
126126
time_t lastStatusChangeTime;
127+
XLogRecPtr flushPos;
127128
csn_t oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
128129
} MtmNodeInfo;
129130

@@ -179,6 +180,15 @@ typedef struct
179180
MtmNodeInfo nodes[1]; /* [MtmNodes]: per-node data */
180181
} MtmState;
181182

183+
typedef struct MtmFlushPosition
184+
{
185+
dlist_node node;
186+
int node_id;
187+
XLogRecPtr local_end;
188+
XLogRecPtr remote_end;
189+
} MtmFlushPosition;
190+
191+
182192
#define MtmIsCoordinator(ts) (ts->gtid.node == MtmNodeId)
183193

184194
extern char const* const MtmNodeStatusMnem[];
@@ -240,5 +250,7 @@ extern bool MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN);
240250
extern void MtmRecoveryCompleted(void);
241251
extern void MtmMakeTableLocal(char* schema, char* name);
242252
extern void MtmHandleApplyError(void);
253+
extern void MtmUpdateLsnMapping(int nodeId, XLogRecPtr endLsn);
254+
extern XLogRecPtr MtmGetFlushPosition(int nodeId);
243255

244256
#endif

‎contrib/mmts/pglogical_apply.c

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -509,14 +509,14 @@ process_remote_commit(StringInfo in)
509509
uint8 flags;
510510
csn_t csn;
511511
const char *gid = NULL;
512-
512+
XLogRecPtr end_lsn;
513513
/* read flags */
514514
flags = pq_getmsgbyte(in);
515515
MtmReplicationNode = pq_getmsgbyte(in);
516516

517517
/* read fields */
518518
replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
519-
pq_getmsgint64(in); /* end_lsn */
519+
end_lsn = pq_getmsgint64(in); /* end_lsn */
520520
replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
521521

522522
Assert(replorigin_session_origin == InvalidRepOriginId);
@@ -581,6 +581,7 @@ process_remote_commit(StringInfo in)
581581
Assert(false);
582582
}
583583
MtmEndSession(true);
584+
MtmUpdateLsnMapping(MtmReplicationNodeId, end_lsn);
584585
if (flags & PGLOGICAL_CAUGHT_UP) {
585586
MtmRecoveryCompleted();
586587
}
@@ -951,7 +952,9 @@ void MtmExecutor(int id, void* work, size_t size)
951952
}
952953
PG_CATCH();
953954
{
955+
MemoryContext oldcontext = MemoryContextSwitchTo(ApplyContext);
954956
MtmHandleApplyError();
957+
MemoryContextSwitchTo(oldcontext);
955958
EmitErrorReport();
956959
FlushErrorState();
957960
MTM_LOG2("%d: REMOTE begin abort transaction %d", MyProcPid, MtmGetCurrentTransactionId());

‎contrib/mmts/pglogical_receiver.c

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -89,17 +89,18 @@ receiver_raw_sighup(SIGNAL_ARGS)
8989
* Send a Standby Status Update message to server.
9090
*/
9191
static bool
92-
sendFeedback(PGconn *conn, int64 now, RepOriginId originId)
92+
sendFeedback(PGconn *conn, int64 now, int node_id)
9393
{
9494
char replybuf[1 + 8 + 8 + 8 + 8 + 1];
9595
int len = 0;
96-
XLogRecPtr output_applied_lsn = replorigin_get_progress(originId, true);
96+
XLogRecPtr output_applied_lsn = output_written_lsn;
97+
XLogRecPtr output_flushed_lsn = MtmGetFlushPosition(node_id);
9798

9899
replybuf[len] = 'r';
99100
len += 1;
100101
fe_sendint64(output_written_lsn, &replybuf[len]); /* write */
101102
len += 8;
102-
fe_sendint64(output_applied_lsn, &replybuf[len]); /* flush */
103+
fe_sendint64(output_flushed_lsn, &replybuf[len]); /* flush */
103104
len += 8;
104105
fe_sendint64(output_applied_lsn, &replybuf[len]); /* apply */
105106
len += 8;
@@ -421,7 +422,7 @@ pglogical_receiver_main(Datum main_arg)
421422
int64 now = feGetCurrentTimestamp();
422423

423424
/* Leave is feedback is not sent properly */
424-
if (!sendFeedback(conn, now, originId))
425+
if (!sendFeedback(conn, now, args->remote_node))
425426
proc_exit(1);
426427
}
427428
continue;

‎contrib/mmts/tests/dtmbench.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -150,8 +150,8 @@ void* writer(void* arg)
150150
for (int i = 0; i < cfg.nIterations; i++)
151151
{
152152
//work
153-
transaction<repeatable_read> txn(*conns[random() % conns.size()]);
154-
//transaction<read_committed> txn(*conns[random() % conns.size()]);
153+
//transaction<repeatable_read> txn(*conns[random() % conns.size()]);
154+
transaction<read_committed> txn(*conns[random() % conns.size()]);
155155
int srcAcc = random() % cfg.nAccounts;
156156
int dstAcc = random() % cfg.nAccounts;
157157
if (cfg.scatter) {

0 commit comments

Comments
 (0)