Fix locking in MtmRefreshClusterStatus

Author: knizhnik

contrib/mmts/arbiter.c

@@ -354,7 +354,7 @@ static void MtmSendHeartbeat()
 	msg.oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
 	msg.node = MtmNodeId;
 	msg.csn = now;
-	if (last_sent_heartbeat + MSEC_TO_USEC(MtmHeartbeatSendTimeout)*2 < now) { 
+	if (last_sent_heartbeat != 0 && last_sent_heartbeat + MSEC_TO_USEC(MtmHeartbeatSendTimeout)*2 < now) { 
 		MTM_LOG1("More than %ld microseconds since last heartbeat", now - last_sent_heartbeat);
 	}
 	last_sent_heartbeat = now;
@@ -1024,7 +1024,7 @@ static void MtmTransReceiver(Datum arg)
 							MTM_LOG1("Last hearbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
 						}
 					}
-					MTM_LOG1("epoll started %ld and finished %ld microseconds ago", now - startPolling, now - stopPolling);
+					MTM_LOG2("epoll started %ld and finished %ld microseconds ago", now - startPolling, now - stopPolling);
 				}
 				lastHeartbeatCheck = now;
 			}

contrib/mmts/multimaster.c

@@ -568,7 +568,7 @@ MtmAdjustOldestXid(TransactionId xid)
 	if (MtmUseDtm) 
 	{ 
 		if (prev != NULL) { 
-			MTM_LOG1("%d: MtmAdjustOldestXid: oldestXid=%d, prev->xid=%d, prev->status=%d, prev->snapshot=%ld, ts->xid=%d, ts->status=%d, ts->snapshot=%ld, oldestSnapshot=%ld", 
+			MTM_LOG2("%d: MtmAdjustOldestXid: oldestXid=%d, prev->xid=%d, prev->status=%d, prev->snapshot=%ld, ts->xid=%d, ts->status=%d, ts->snapshot=%ld, oldestSnapshot=%ld", 
 					 MyProcPid, xid, prev->xid, prev->status, prev->snapshot, (ts ? ts->xid : 0), (ts ? ts->status : -1), (ts ? ts->snapshot : -1), oldestSnapshot);
 			Mtm->transListHead = prev;
 			Mtm->oldestXid = xid = prev->xid;            
@@ -887,7 +887,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 	MtmLock(LW_EXCLUSIVE);
 	ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
 	Assert(ts != NULL);
-	if (x->gid[0]) MTM_LOG1("Preparing transaction %d (%s) at %ld", x->xid, x->gid, MtmGetCurrentTime());
+	//if (x->gid[0]) MTM_LOG1("Preparing transaction %d (%s) at %ld", x->xid, x->gid, MtmGetCurrentTime());
 	if (!MtmIsCoordinator(ts) || Mtm->status == MTM_RECOVERY) {
 		bool found;
 		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, &found);
@@ -941,7 +941,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		MTM_LOG3("%d: Result of vote: %d", MyProcPid, ts->status);
 		MtmUnlock();
 	}
-	if (x->gid[0]) MTM_LOG1("Prepared transaction %d (%s) csn=%ld at %ld: %d", x->xid, x->gid, ts->csn, MtmGetCurrentTime(), ts->status);
+	//if (x->gid[0]) MTM_LOG1("Prepared transaction %d (%s) csn=%ld at %ld: %d", x->xid, x->gid, ts->csn, MtmGetCurrentTime(), ts->status);
 	if (Mtm->inject2PCError == 3) { 
 		Mtm->inject2PCError = 0;
 		elog(ERROR, "ERROR INJECTION for transaction %d (%s)", x->xid, x->gid);
@@ -2597,7 +2597,9 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 		} else {
 			MtmUnlock();
 			MtmRefreshClusterStatus(true);
+			MtmLock(LW_SHARED);
 			if (BIT_CHECK(Mtm->disabledNodeMask, MtmReplicationNodeId-1)) {
+				MtmUnlock();
 				elog(ERROR, "Disabled node %d tries to reconnect without recovery", MtmReplicationNodeId); 
 			}
 		}

contrib/mmts/pglogical_apply.c

@@ -343,7 +343,7 @@ process_remote_begin(StringInfo s)
 
 	inside_tx = true;
 
-	MTM_LOG1("REMOTE begin node=%d xid=%d snapshot=%ld", gtid.node, gtid.xid, snapshot);
+	MTM_LOG2("REMOTE begin node=%d xid=%d snapshot=%ld", gtid.node, gtid.xid, snapshot);
 }
 
 static void
@@ -624,7 +624,7 @@ process_remote_commit(StringInfo in)
 			gid = pq_getmsgstring(in);
 			MTM_LOG2("PGLOGICAL_ABORT_PREPARED commit: gid=%s",  gid);
 			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
-				MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
+				MTM_LOG2("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
 				StartTransactionCommand();
 				MtmBeginSession();
 				MtmSetCurrentTransactionGID(gid);
@@ -967,7 +967,7 @@ void MtmExecutor(int id, void* work, size_t size)
     {    
         while (true) { 
             char action = pq_getmsgbyte(&s);
-            MTM_LOG1("%d: REMOTE process action %c", MyProcPid, action);
+            MTM_LOG2("%d: REMOTE process action %c", MyProcPid, action);
 #if 0
 			if (Mtm->status == MTM_RECOVERY) { 
 				MTM_LOG1("Replay action %c[%x]",   action, s.data[s.cursor]);