Do not use epoll

Author: knizhnik

arbiter.c

@@ -59,7 +59,7 @@
 
 #ifndef USE_EPOLL
 #ifdef __linux__
-#define USE_EPOLL 1
+#define USE_EPOLL 0
 #else
 #define USE_EPOLL 0
 #endif
@@ -105,7 +105,7 @@ typedef struct
 static int*        sockets;
 static int         gateway;
 static bool        send_heartbeat;
-static timestamp_t last_sent_hearbeat;
+static timestamp_t last_sent_heartbeat;
 static TimeoutId   heartbeat_timer;
 static int         busy_socket;
 
@@ -266,17 +266,20 @@ static bool MtmWriteSocket(int sd, void const* buf, int size)
 
 static int MtmReadSocket(int sd, void* buf, int buf_size)
 {
-	int rc = MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout);
-	if (rc == 1) { 
-		int rc = recv(sd, buf, buf_size, 0);
-		if (rc <= 0) { 
-			Assert(errno != EINTR); /* should not happen in non-blocking call */
-			return -1;
+	int rc = recv(sd, buf, buf_size, 0);
+	if (rc < 0 && errno == EAGAIN) { 
+		rc = MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout);
+		if (rc == 1) { 
+			rc = recv(sd, buf, buf_size, 0);
+			if (rc < 0) { 
+				Assert(errno != EINTR); /* should not happen in non-blocking call */
+				return -1;
+			}
+		} else { 
+			return 0;
 		}
-		return rc;
-	} else { 
-		return 0;
 	}
+	return rc;
 }
 
 
@@ -343,7 +346,7 @@ static void MtmCheckResponse(MtmArbiterMessage* resp)
 
 static void MtmScheduleHeartbeat()
 {
-//	Assert(!last_sent_hearbeat || last_sent_hearbeat + MSEC_TO_USEC(MtmHeartbeatRecvTimeout) >= MtmGetSystemTime());
+//	Assert(!last_sent_heartbeat || last_sent_heartbeat + MSEC_TO_USEC(MtmHeartbeatRecvTimeout) >= MtmGetSystemTime());
 	enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
 	send_heartbeat = true;
 	PGSemaphoreUnlock(&Mtm->votingSemaphore);
@@ -353,11 +356,16 @@ static void MtmSendHeartbeat()
 {
 	int i;
 	MtmArbiterMessage msg;
+	timestamp_t now = MtmGetSystemTime();
 	msg.code = MSG_HEARTBEAT;
 	msg.disabledNodeMask = Mtm->disabledNodeMask;
 	msg.oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
 	msg.node = MtmNodeId;
-	last_sent_hearbeat = MtmGetSystemTime();
+	msg.csn = now;
+	if (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;
 
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
@@ -366,6 +374,8 @@ static void MtmSendHeartbeat()
 		{ 
 			if (!MtmSendToNode(i, &msg, sizeof(msg))) { 
 				elog(LOG, "Arbiter failed to send heartbeat to node %d", i+1);
+			} else {
+				MTM_LOG1("Send heartbeat to node %d with timestamp %ld", i+1, now);    
 			}
 		}
 	}
@@ -558,7 +568,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 static int MtmReadFromNode(int node, void* buf, int buf_size)
 {
 	int rc = MtmReadSocket(sockets[node], buf, buf_size);
-	if (rc <= 0) { 
+	if (rc < 0) { 
 		elog(WARNING, "Arbiter failed to read from node=%d, rc=%d, errno=%d", node+1, rc, errno);
 		MtmDisconnect(node);
 	}
@@ -812,6 +822,8 @@ static void MtmTransReceiver(Datum arg)
 	}
 
 	while (!stop) {
+		timestamp_t startPolling = MtmGetSystemTime();
+		timestamp_t stopPolling;
 #if USE_EPOLL
         n = epoll_wait(epollfd, events, nNodes, MtmHeartbeatRecvTimeout);
 		if (n < 0) { 
@@ -820,13 +832,17 @@ static void MtmTransReceiver(Datum arg)
 			}
 			elog(ERROR, "Arbiter failed to poll sockets: %d", errno);
 		}
+		stopPolling = MtmGetSystemTime();
+
 		for (j = 0; j < n; j++) {
 			i = events[j].data.u32;
 			if (events[j].events & EPOLLERR) {
 				elog(WARNING, "Arbiter lost connection with node %d", i+1);
 				MtmDisconnect(i);
 			} 
-			else if (events[j].events & EPOLLIN)  
+		}
+		for (j = 0; j < n; j++) {
+			if (events[j].events & EPOLLIN)  
 #else
         fd_set events;
 		do { 
@@ -842,6 +858,8 @@ static void MtmTransReceiver(Datum arg)
 		if (n < 0) {
 			elog(ERROR, "Arbiter failed to select sockets: %d", errno);
 		}
+		stopPolling = MtmGetSystemTime();
+
 		for (i = 0; i < nNodes; i++) { 
 			if (sockets[i] >= 0 && FD_ISSET(sockets[i], &events)) 
 #endif
@@ -871,7 +889,8 @@ static void MtmTransReceiver(Datum arg)
 					Mtm->nodes[msg->node-1].lastHeartbeat = MtmGetSystemTime();
 
 					if (msg->code == MSG_HEARTBEAT) {
-						MTM_LOG3("Receive HEARTBEAT from node %d at %ld", msg->node, USEC_TO_MSEC(MtmGetSystemTime())); 
+						MTM_LOG1("Receive HEARTBEAT from node %d with timestamp %ld delay %ld", 
+								 msg->node, msg->csn, USEC_TO_MSEC(MtmGetSystemTime() - msg->csn)); 
 						continue;
 					}
 					if (BIT_CHECK(msg->disabledNodeMask, msg->node-1)) {
@@ -985,7 +1004,14 @@ static void MtmTransReceiver(Datum arg)
 		}
 		now = MtmGetSystemTime();
 		if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
-			MtmWatchdog();
+			if (!MtmWatchdog(stopPolling)) { 
+				for (i = 0; i < nNodes; i++) { 
+					if (Mtm->nodes[i].lastHeartbeat != 0 && sockets[i] >= 0) {
+						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);
+			}
 			lastHeartbeatCheck = now;
 		}
 		if (n == 0 && Mtm->disabledNodeMask != 0) { 

multimaster.c

@@ -248,7 +248,13 @@ void MtmLock(LWLockMode mode)
 #ifdef USE_SPINLOCK
 	SpinLockAcquire(&Mtm->spinlock);
 #else
+	timestamp_t start, stop;
+	start = MtmGetSystemTime();
 	LWLockAcquire((LWLockId)&Mtm->locks[MTM_STATE_LOCK_ID], mode);
+	stop = MtmGetSystemTime();
+	if (stop > start + MSEC_TO_USEC(MtmHeartbeatSendTimeout)) { 
+		MTM_LOG1("%d: obtaining %s lock takes %ld microseconds", MyProcPid, (mode == LW_EXCLUSIVE ? "exclusive" : "shared"), stop - start);
+	}	
 #endif
 	Mtm->lastLockHolder = MyProcPid;
 }
@@ -819,10 +825,10 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 /*
  * Check heartbeats
  */
-void MtmWatchdog(void)
+bool MtmWatchdog(timestamp_t now)
 {
 	int i, n = Mtm->nAllNodes;
-	timestamp_t now = MtmGetSystemTime();
+	bool allAlive = true;
 	for (i = 0; i < n; i++) { 
 		if (i+1 != MtmNodeId && !BIT_CHECK(Mtm->disabledNodeMask, i)) {
 			if (Mtm->nodes[i].lastHeartbeat != 0
@@ -831,9 +837,11 @@ void MtmWatchdog(void)
 				elog(WARNING, "Heartbeat is not received from node %d during %d msec", 
 					 i+1, (int)USEC_TO_MSEC(now - Mtm->nodes[i].lastHeartbeat));
 				MtmOnNodeDisconnect(i+1);				
+				allAlive = false;
 			}
 		}
 	}
+	return allAlive;
 }
 
 
@@ -926,7 +934,7 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 		MtmLock(LW_EXCLUSIVE);
 		tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
 		Assert(tm != NULL && tm->state != NULL);
-		MTM_LOG1("%ld: Abort prepared transaction %d with gid='%s'", MtmGetSystemTime(), x->xid, x->gid);
+		MTM_LOG1("Abort prepared transaction %d with gid='%s'", x->xid, x->gid);
 		MtmAbortTransaction(tm->state);
 		MtmUnlock();
 		x->status = TRANSACTION_STATUS_ABORTED;
@@ -1184,6 +1192,7 @@ static void MtmDisableNode(int nodeId)
 static void MtmEnableNode(int nodeId)
 { 
 	BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
+	BIT_CLEAR(Mtm->reconnectMask, nodeId-1);
 	Mtm->nodes[nodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	Mtm->nLiveNodes += 1;			
@@ -1569,6 +1578,7 @@ void MtmOnNodeConnect(int nodeId)
 {
 	MtmLock(LW_EXCLUSIVE);	
 	BIT_CLEAR(Mtm->connectivityMask, nodeId-1);
+	BIT_CLEAR(Mtm->reconnectMask, nodeId-1);
 	MtmUnlock();
 
 	MTM_LOG1("Reconnect node %d", nodeId);
@@ -1881,7 +1891,7 @@ _PG_init(void)
 	DefineCustomIntVariable(
 		"multimaster.heartbeat_send_timeout", 
 		"Timeout in milliseconds of sending heartbeat messages",
-		"Period of broadcasting heartbeat messages by abiter to all nodes",
+		"Period of broadcasting heartbeat messages by arbiter to all nodes",
 		&MtmHeartbeatSendTimeout,
 		1000,
 		1,
@@ -2286,6 +2296,7 @@ MtmSlotMode MtmReceiverSlotMode(int nodeId)
 				/* Choose for recovery first available slot */
 				MTM_LOG1("Start recovery from node %d", nodeId);
 				Mtm->recoverySlot = nodeId;
+				FinishAllPreparedTransactions(false);
 				return SLOT_OPEN_EXISTED;
 			}
 		}

multimaster.h

@@ -277,7 +277,7 @@ extern void  MtmMakeTableLocal(char* schema, char* name);
 extern void  MtmHandleApplyError(void);
 extern void  MtmUpdateLsnMapping(int nodeId, XLogRecPtr endLsn);
 extern XLogRecPtr MtmGetFlushPosition(int nodeId);
-extern void MtmWatchdog(void);
+extern bool MtmWatchdog(timestamp_t now);
 extern void MtmCheckHeartbeat(void);
 
 

pglogical_apply.c

@@ -538,11 +538,11 @@ process_remote_commit(StringInfo in)
 			Assert(IsTransactionState() && TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
 			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_IN_PROGRESS) == TRANSACTION_STATUS_ABORTED) { 
-				MTM_LOG1("%ld: avoid prepare of previously aborted global transaction %s", MtmGetSystemTime(), gid);	
+				MTM_LOG1("Avoid prepare of previously aborted global transaction %s", gid);	
 				AbortCurrentTransaction();
 			} else { 				
 				/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
-				MTM_LOG2("%ld: PGLOGICAL_PREPARE commit: gid=%s", MtmGetSystemTime(), gid);
+				MTM_LOG2("PGLOGICAL_PREPARE commit: gid=%s", gid);
 				BeginTransactionBlock();
 				CommitTransactionCommand();
 				StartTransactionCommand();
@@ -554,7 +554,7 @@ process_remote_commit(StringInfo in)
 				CommitTransactionCommand();
 
 				if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_UNKNOWN) == TRANSACTION_STATUS_ABORTED) { 
-					MTM_LOG1("%ld: perform delayed rollback of prepared global transaction %s", MtmGetSystemTime(), gid);	
+					MTM_LOG1("Perform delayed rollback of prepared global transaction %s", gid);	
 					StartTransactionCommand();
 					MtmSetCurrentTransactionGID(gid);
 					FinishPreparedTransaction(gid, false);
@@ -568,7 +568,7 @@ process_remote_commit(StringInfo in)
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			csn = pq_getmsgint64(in); 
 			gid = pq_getmsgstring(in);
-			MTM_LOG2("%ld: PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s", MtmGetSystemTime(), csn, gid);
+			MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s", csn, gid);
 			StartTransactionCommand();
 			MtmBeginSession();
 			MtmSetCurrentTransactionCSN(csn);
@@ -581,9 +581,9 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			MTM_LOG2("%ld: PGLOGICAL_ABORT_PREPARED commit: gid=%s",  MtmGetSystemTime(), gid);
+			MTM_LOG2("PGLOGICAL_ABORT_PREPARED commit: gid=%s",  gid);
 			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
-				MTM_LOG1("%ld: PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", MtmGetSystemTime(), gid);
+				MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
 				StartTransactionCommand();
 				MtmSetCurrentTransactionGID(gid);
 				FinishPreparedTransaction(gid, false);

pglogical_proto.c

@@ -161,7 +161,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	}
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
-	MTM_LOG2("%ld: PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx", MtmGetSystemTime(), flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());
+	MTM_LOG2("PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx", flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());
 
     /* send the flags field */
     pq_sendbyte(out, flags);

pglogical_receiver.c

@@ -40,7 +40,7 @@
 #include "spill.h"
 
 #define ERRCODE_DUPLICATE_OBJECT_STR  "42710"
-#define RECEIVER_SUSPEND_TIMEOUT (10*USECS_PER_SEC)
+#define RECEIVER_SUSPEND_TIMEOUT (1*USECS_PER_SEC)
 
 /* Signal handling */
 static volatile sig_atomic_t got_sigterm = false;
@@ -357,9 +357,10 @@ pglogical_receiver_main(Datum main_arg)
 			if (rc & WL_POSTMASTER_DEATH)
 				proc_exit(1);
 
-			if (Mtm->status == MTM_OFFLINE || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId)) {
+			if (Mtm->status == MTM_OFFLINE || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId)) 
+			{
 				ereport(LOG, (errmsg("%s: suspending WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
-				goto OnError;
+				break;
 			}
 
 
@@ -578,6 +579,9 @@ pglogical_receiver_main(Datum main_arg)
 				goto OnError;
 			}
 		}
+		PQfinish(conn);
+		continue;
+
 	  OnError:
 		PQfinish(conn);
 		MtmSleep(RECEIVER_SUSPEND_TIMEOUT);		

raftable.c

@@ -19,13 +19,19 @@ void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, bool now
 void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 {
 	if (MtmUseRaftable) {
+		timestamp_t start, stop;
+		start = MtmGetSystemTime();
 		if (nowait) {
 			raftable_set(key, value, size, 0);
 		} else { 
 			while (!raftable_set(key, value, size, MtmHeartbeatSendTimeout)) { 
 				MtmCheckHeartbeat();
 			}
 		}
+		stop = MtmGetSystemTime();
+		if (stop > start + MSEC_TO_USEC(MtmHeartbeatSendTimeout)) { 
+			MTM_LOG1("Raftable set nowait=%d takes %ld microseconds", nowait, stop - start);
+		}
 	}		
 }