Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitea75de1

Browse files
committed
Prevent recusive broadcast by setting special applicastion name
1 parent810ca68 commitea75de1

File tree

3 files changed

+14
-19
lines changed

3 files changed

+14
-19
lines changed

‎contrib/mmts/multimaster.c‎

Lines changed: 11 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -800,7 +800,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
800800
* Send notification only if ABORT happens during transaction processing at replicas,
801801
* do not send notification if ABORT is receiver from master
802802
*/
803-
MTM_TRACE("%d: send ABORT notification to coordinator %d\n",MyProcPid,x->gtid.node);
803+
MTM_INFO("%d: send ABORT notificationabort transaction %dto coordinator %d\n",MyProcPid,x->gtid.xid,x->gtid.node);
804804
if (ts==NULL) {
805805
Assert(TransactionIdIsValid(x->xid));
806806
ts=hash_search(MtmXid2State,&x->xid,HASH_ENTER,NULL);
@@ -1604,6 +1604,11 @@ MtmSlotMode MtmReceiverSlotMode(int nodeId)
16041604
returnMtm->recoverySlot ?SLOT_CREATE_NEW :SLOT_OPEN_ALWAYS;
16051605
}
16061606

1607+
staticboolMtmIsBroadcast()
1608+
{
1609+
returnapplication_name!=NULL&&strcmp(application_name,MULTIMASTER_BROADCAST_SERVICE)==0;
1610+
}
1611+
16071612
voidMtmRecoverNode(intnodeId)
16081613
{
16091614
if (nodeId <=0||nodeId>Mtm->nNodes)
@@ -1613,7 +1618,7 @@ void MtmRecoverNode(int nodeId)
16131618
if (!BIT_CHECK(Mtm->disabledNodeMask,nodeId-1)) {
16141619
elog(ERROR,"Node %d was not disabled",nodeId);
16151620
}
1616-
if (!IsTransactionBlock())
1621+
if (!MtmIsBroadcast())
16171622
{
16181623
MtmBroadcastUtilityStmt(psprintf("select pg_create_logical_replication_slot('"MULTIMASTER_SLOT_PATTERN"', '"MULTIMASTER_NAME"')",nodeId), true);
16191624
}
@@ -1630,7 +1635,7 @@ void MtmDropNode(int nodeId, bool dropSlot)
16301635
}
16311636
BIT_SET(Mtm->disabledNodeMask,nodeId-1);
16321637
Mtm->nNodes-=1;
1633-
if (!IsTransactionBlock())
1638+
if (!MtmIsBroadcast())
16341639
{
16351640
MtmBroadcastUtilityStmt(psprintf("select mtm.drop_node(%d,%s)",nodeId,dropSlot ?"true" :"false"), true);
16361641
}
@@ -1650,7 +1655,6 @@ MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
16501655
staticbool
16511656
MtmReplicationTxnFilterHook(structPGLogicalTxnFilterArgs*args)
16521657
{
1653-
elog(WARNING,"MtmReplicationTxnFilterHook: args->origin_id=%d, MtmReplicationNodeId=%d",args->origin_id,MtmReplicationNodeId);
16541658
returnargs->origin_id==InvalidRepOriginId||MtmIsRecoveredNode(MtmReplicationNodeId);
16551659
}
16561660

@@ -1797,7 +1801,6 @@ static bool MtmRunUtilityStmt(PGconn* conn, char const* sql, char **errmsg)
17971801
{
17981802
PGresult*result=PQexec(conn,sql);
17991803
intstatus=PQresultStatus(result);
1800-
char*errstr;
18011804

18021805
boolret=status==PGRES_COMMAND_OK||status==PGRES_TUPLES_OK;
18031806

@@ -1817,25 +1820,18 @@ static bool MtmRunUtilityStmt(PGconn* conn, char const* sql, char **errmsg)
18171820

18181821
staticvoidMtmBroadcastUtilityStmt(charconst*sql,boolignoreError)
18191822
{
1820-
char*conn_str=pstrdup(MtmConnStrs);
1821-
char*conn_str_end=conn_str+strlen(conn_str);
18221823
inti=0;
18231824
nodemask_tdisabledNodeMask=Mtm->disabledNodeMask;
18241825
intfailedNode=-1;
18251826
charconst*errorMsg=NULL;
18261827
PGconn**conns=palloc0(sizeof(PGconn*)*MtmNodes);
18271828
char*utility_errmsg;
18281829

1829-
while (conn_str<conn_str_end)
1830+
for (i=0;i<MtmNodes;i++)
18301831
{
1831-
char*p=strchr(conn_str,',');
1832-
if (p==NULL) {
1833-
p=conn_str_end;
1834-
}
1835-
*p='\0';
18361832
if (!BIT_CHECK(disabledNodeMask,i))
18371833
{
1838-
conns[i]=PQconnectdb(conn_str);
1834+
conns[i]=PQconnectdb(psprintf("%s application_name=%s",Mtm->nodes[i].con.connStr,MULTIMASTER_BROADCAST_SERVICE));
18391835
if (PQstatus(conns[i])!=CONNECTION_OK)
18401836
{
18411837
if (ignoreError)
@@ -1847,12 +1843,10 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError)
18471843
do {
18481844
PQfinish(conns[i]);
18491845
}while (--i >=0);
1850-
elog(ERROR,"Failed to establish connection '%s' to node %d",conn_str,failedNode);
1846+
elog(ERROR,"Failed to establish connection '%s' to node %d",Mtm->nodes[i].con.connStr,failedNode);
18511847
}
18521848
}
18531849
}
1854-
conn_str=p+1;
1855-
i+=1;
18561850
}
18571851
Assert(i==MtmNodes);
18581852

‎contrib/mmts/multimaster.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
#defineMULTIMASTER_MAX_SLOT_NAME_SIZE 16
2727
#defineMULTIMASTER_MAX_CONN_STR_SIZE 128
2828
#defineMULTIMASTER_MAX_HOST_NAME_SIZE 64
29+
#defineMULTIMASTER_BROADCAST_SERVICE "mtm_broadcast"
2930

3031
#defineUSEC 1000000
3132

‎contrib/mmts/pglogical_apply.c‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -936,10 +936,10 @@ void MtmExecutor(int id, void* work, size_t size)
936936
{
937937
EmitErrorReport();
938938
FlushErrorState();
939-
MTM_TRACE("%d: REMOTE begin abort transaction %d\n",MyProcPid,MtmGetCurrentTransactionId());
939+
MTM_INFO("%d: REMOTE begin abort transaction %d\n",MyProcPid,MtmGetCurrentTransactionId());
940940
MtmEndSession(false);
941941
AbortCurrentTransaction();
942-
MTM_TRACE("%d: REMOTE end abort transaction %d\n",MyProcPid,MtmGetCurrentTransactionId());
942+
MTM_INFO("%d: REMOTE end abort transaction %d\n",MyProcPid,MtmGetCurrentTransactionId());
943943
}
944944
PG_END_TRY();
945945

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp