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

Commite0dadac

Browse files
committed
Fix timer initialziation
1 parent5ee48f4 commite0dadac

File tree

2 files changed

+21
-12
lines changed

2 files changed

+21
-12
lines changed

‎contrib/mmts/arbiter.c‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@ typedef struct
103103
staticint*sockets;
104104
staticintgateway;
105105
staticboolsend_heartbeat;
106+
staticTimeoutIdheartbeat_timer;
106107

107108
staticvoidMtmTransSender(Datumarg);
108109
staticvoidMtmTransReceiver(Datumarg);
@@ -526,7 +527,7 @@ static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, Mt
526527
buf->data[buf->used].csn=ts->csn;
527528
}else {
528529
buf->data[buf->used].code=MSG_HEARTBEAT;
529-
MTM_LOG3("Send HEARTBEATmessageto node %d from node %d\n",node+1,MtmNodeId);
530+
MTM_LOG3("Send HEARTBEAT to node %d from node %d at %ld\n",node+1,MtmNodeId,USEC_TO_MSEC(MtmGetSystemTime()));
530531
}
531532
buf->data[buf->used].node=MtmNodeId;
532533
buf->data[buf->used].disabledNodeMask=Mtm->disabledNodeMask;
@@ -540,8 +541,9 @@ static void MtmBroadcastMessage(MtmBuffer* txBuffer, MtmTransState* ts)
540541
intn=1;
541542
for (i=0;i<Mtm->nAllNodes;i++)
542543
{
543-
if (!BIT_CHECK(Mtm->disabledNodeMask,i)&& (ts==NULL||TransactionIdIsValid(ts->xids[i]))) {
544-
Assert(i+1!=MtmNodeId);
544+
if (i+1!=MtmNodeId&& !BIT_CHECK(Mtm->disabledNodeMask,i)
545+
&& (ts==NULL||TransactionIdIsValid(ts->xids[i])))
546+
{
545547
MtmAppendBuffer(txBuffer,ts ?ts->xids[i] :InvalidTransactionId,i,ts);
546548
n+=1;
547549
}
@@ -553,6 +555,7 @@ static void MtmSendHeartbeat()
553555
{
554556
send_heartbeat= true;
555557
PGSemaphoreUnlock(&Mtm->votingSemaphore);
558+
//enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
556559
}
557560

558561

@@ -561,15 +564,19 @@ static void MtmTransSender(Datum arg)
561564
sigset_tsset;
562565
intnNodes=MtmMaxNodes;
563566
inti;
567+
564568
MtmBuffer*txBuffer= (MtmBuffer*)palloc(sizeof(MtmBuffer)*nNodes);
565569

570+
InitializeTimeouts();
571+
566572
signal(SIGINT,SetStop);
567573
signal(SIGQUIT,SetStop);
568574
signal(SIGTERM,SetStop);
569575
sigfillset(&sset);
570576
sigprocmask(SIG_UNBLOCK,&sset,NULL);
571577

572-
RegisterTimeout(USER_TIMEOUT,MtmSendHeartbeat);
578+
heartbeat_timer=RegisterTimeout(USER_TIMEOUT,MtmSendHeartbeat);
579+
enable_timeout_after(heartbeat_timer,MtmHeartbeatSendTimeout);
573580

574581
MtmOpenConnections();
575582

@@ -584,6 +591,7 @@ static void MtmTransSender(Datum arg)
584591

585592
if (send_heartbeat) {
586593
send_heartbeat= false;
594+
enable_timeout_after(heartbeat_timer,MtmHeartbeatSendTimeout);
587595
MtmBroadcastMessage(txBuffer,NULL);
588596
}
589597
/*
@@ -725,7 +733,8 @@ static void MtmTransReceiver(Datum arg)
725733
Mtm->nodes[msg->node-1].oldestSnapshot=msg->oldestSnapshot;
726734
Mtm->nodes[msg->node-1].lastHeartbeat=MtmGetSystemTime();
727735

728-
if (msg->code==MSG_HEARTBEAT) {
736+
if (msg->code==MSG_HEARTBEAT) {
737+
MTM_LOG3("Receive HEARTBEAT from node %d at %ld",msg->node,USEC_TO_MSEC(MtmGetSystemTime()));
729738
continue;
730739
}
731740
ts= (MtmTransState*)hash_search(MtmXid2State,&msg->dxid,HASH_FIND,NULL);

‎contrib/mmts/multimaster.c‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -756,8 +756,8 @@ static void MtmWatchdog()
756756
if (Mtm->nodes[i].lastHeartbeat!=0
757757
&&now>Mtm->nodes[i].lastHeartbeat+MSEC_TO_USEC(MtmHeartbeatRecvTimeout))
758758
{
759-
elog(WARNING,"Disable node %d because last heartbeat was received %d msec ago",
760-
i+1, (int)USEC_TO_MSEC(now-Mtm->nodes[i].lastHeartbeat));
759+
elog(WARNING,"Disable node %d because last heartbeat was received %d msec ago (%ld)",
760+
i+1, (int)USEC_TO_MSEC(now-Mtm->nodes[i].lastHeartbeat),USEC_TO_MSEC(now));
761761
MtmOnNodeDisconnect(i+1);
762762
}
763763
}
@@ -802,7 +802,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
802802
/* wait votes from all nodes */
803803
while (!ts->votingCompleted) {
804804
MtmUnlock();
805-
MtmWatchdog();
805+
//MtmWatchdog();
806806
if (ts->status==TRANSACTION_STATUS_ABORTED) {
807807
elog(WARNING,"Transaction %d(%s) is aborted by watchdog",x->xid,x->gid);
808808
x->status=TRANSACTION_STATUS_ABORTED;
@@ -1694,7 +1694,7 @@ _PG_init(void)
16941694
"Timeout in milliseconds of receiving heartbeat messages",
16951695
"If no heartbeat message is received from node within this period, it assumed to be dead",
16961696
&MtmHeartbeatRecvTimeout,
1697-
2000,
1697+
100000,
16981698
1,
16991699
INT_MAX,
17001700
PGC_BACKEND,
@@ -1753,7 +1753,7 @@ _PG_init(void)
17531753
"Minamal amount of time (milliseconds) to wait 2PC confirmation from all nodes",
17541754
"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",
17551755
&Mtm2PCMinTimeout,
1756-
100000,/* 100 seconds */
1756+
10000,/* 100 seconds */
17571757
0,
17581758
INT_MAX,
17591759
PGC_BACKEND,
@@ -1814,8 +1814,8 @@ _PG_init(void)
18141814
DefineCustomIntVariable(
18151815
"multimaster.max_recovery_lag",
18161816
"Maximal lag of replication slot of failed node after which this slot is dropped to avoid transaction log overflow",
1817-
"Droppingslog makes it not possible to recover node using logical replication mechanism, it will be ncessary to completely copy content of some other nodes "
1818-
"usimg basebackup or similar tool. Zero value of parameter disabledroipping slot.",
1817+
"Droppingslot makes it not possible to recover node using logical replication mechanism, it will be ncessary to completely copy content of some other nodes "
1818+
"using basebackup or similar tool. Zero value of parameter disabledropping slot.",
18191819
&MtmMaxRecoveryLag,
18201820
100000000,
18211821
0,

0 commit comments

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