Send deadlock graph ysing logical messages

Author: knizhnik

arbiter.c

@@ -341,6 +341,7 @@ static void MtmSendHeartbeat()
 	timestamp_t now = MtmGetSystemTime();
 	msg.code = MSG_HEARTBEAT;
 	msg.disabledNodeMask = Mtm->disabledNodeMask;
+	msg.connectivityMask = Mtm->connectivityMask;
 	msg.oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
 	msg.node = MtmNodeId;
 	msg.csn = now;
@@ -464,6 +465,7 @@ static int MtmConnectSocket(int node, int port, int timeout)
 	req.hdr.sxid = ShmemVariableCache->nextXid;
 	req.hdr.csn  = MtmGetCurrentTime();
 	req.hdr.disabledNodeMask = Mtm->disabledNodeMask;
+	req.hdr.connectivityMask = Mtm->connectivityMask;
 	strcpy(req.connStr, Mtm->nodes[MtmNodeId-1].con.connStr);
 	if (!MtmWriteSocket(sd, &req, sizeof req)) { 
 		elog(WARNING, "Arbiter failed to send handshake message to %s:%d: %d", host, port, errno);
@@ -600,6 +602,7 @@ static void MtmAcceptOneConnection()
 
 			resp.code = MSG_STATUS;
 			resp.disabledNodeMask = Mtm->disabledNodeMask;
+			resp.connectivityMask = Mtm->connectivityMask;
 			resp.dxid = HANDSHAKE_MAGIC;
 			resp.sxid = ShmemVariableCache->nextXid;
 			resp.csn  = MtmGetCurrentTime();
@@ -882,6 +885,8 @@ static void MtmReceiver(Datum arg)
 
 					Assert(node > 0 && node <= nNodes && node != MtmNodeId);
 					Mtm->nodes[node-1].oldestSnapshot = msg->oldestSnapshot;
+					Mtm->nodes[node-1].disabledNodeMask = msg->disabledNodeMask;
+					Mtm->nodes[node-1].connectivityMask = msg->connectivityMask;
 					Mtm->nodes[node-1].lastHeartbeat = MtmGetSystemTime();
 
 					switch (msg->code) {

bgwpool.c

@@ -130,7 +130,8 @@ static void BgwStartExtraWorker(BgwPool* pool)
 {
 	if (pool->nWorkers < MtmMaxWorkers) { 
 		timestamp_t now = MtmGetSystemTime();
-		if (pool->lastDynamicWorkerStartTime + MULTIMASTER_BGW_RESTART_TIMEOUT*USECS_PER_SEC < now) { 
+		/*if (pool->lastDynamicWorkerStartTime + MULTIMASTER_BGW_RESTART_TIMEOUT*USECS_PER_SEC < now)*/
+		{ 
 			BackgroundWorker worker;
 			BackgroundWorkerHandle* handle;
 			MemSet(&worker, 0, sizeof(BackgroundWorker));

multimaster--1.0.sql

@@ -70,6 +70,10 @@ CREATE FUNCTION mtm.inject_2pc_error(stage integer) RETURNS void
 AS 'MODULE_PATHNAME','mtm_inject_2pc_error'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.check_deadlock(xid integer) RETURNS boolean
+AS 'MODULE_PATHNAME','mtm_check_deadlock'
+LANGUAGE C;
+
 -- CREATE TABLE IF NOT EXISTS public.ddl_log (issued timestamp with time zone not null, query text);
 
 -- CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema text, rel_name text, primary key(rel_schema, rel_name));

multimaster.c

@@ -38,6 +38,7 @@
 #include "access/twophase.h"
 #include "utils/guc.h"
 #include "utils/hsearch.h"
+#include "utils/timeout.h"
 #include "utils/tqual.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
@@ -93,7 +94,7 @@ typedef enum
 #define MTM_MAP_SIZE   MTM_HASH_SIZE
 #define MIN_WAIT_TIMEOUT 1000
 #define MAX_WAIT_TIMEOUT 100000
-#define MAX_WAIT_LOOPS   100
+#define MAX_WAIT_LOOPS   100 // 1000000 
 #define STATUS_POLL_DELAY USECS_PER_SEC
 
 void _PG_init(void);
@@ -116,6 +117,7 @@ PG_FUNCTION_INFO_V1(mtm_get_cluster_info);
 PG_FUNCTION_INFO_V1(mtm_make_table_local);
 PG_FUNCTION_INFO_V1(mtm_dump_lock_graph);
 PG_FUNCTION_INFO_V1(mtm_inject_2pc_error);
+PG_FUNCTION_INFO_V1(mtm_check_deadlock);
 
 static Snapshot MtmGetSnapshot(Snapshot snapshot);
 static void MtmInitialize(void);
@@ -274,15 +276,15 @@ void MtmUnlock(void)
 	Mtm->lastLockHolder = 0;
 }
 
-void MtmLockNode(int nodeId)
+void MtmLockNode(int nodeId, LWLockMode mode)
 {
-	Assert(nodeId > 0 && nodeId <= Mtm->nAllNodes);
-	LWLockAcquire((LWLockId)&Mtm->locks[nodeId], LW_EXCLUSIVE);
+	Assert(nodeId > 0 && nodeId <= MtmMaxNodes*2);
+	LWLockAcquire((LWLockId)&Mtm->locks[nodeId], mode);
 }
 
 void MtmUnlockNode(int nodeId)
 {
-	Assert(nodeId > 0 && nodeId <= Mtm->nAllNodes);
+	Assert(nodeId > 0 && nodeId <= MtmMaxNodes*2);
 	LWLockRelease((LWLockId)&Mtm->locks[nodeId]);	
 }
 
@@ -437,6 +439,7 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
     static timestamp_t totalSleepTime;
     static timestamp_t maxSleepTime;
 #endif
+	timestamp_t start = MtmGetSystemTime();
     timestamp_t delay = MIN_WAIT_TIMEOUT;
 	int i;
     Assert(xid != InvalidTransactionId);
@@ -460,7 +463,10 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
             if (ts->csn > MtmTx.snapshot) { 
                 MTM_LOG4("%d: tuple with xid=%d(csn=%ld) is invisibile in snapshot %ld",
 						 MyProcPid, xid, ts->csn, MtmTx.snapshot);
-                MtmUnlock();
+				if (MtmGetSystemTime() - start > USECS_PER_SEC) { 
+					elog(WARNING, "Backend %d waits for transaction %x status %ld usecs", MyProcPid,  xid, MtmGetSystemTime() - start);
+				}
+				MtmUnlock();
                 return true;
             }
             if (ts->status == TRANSACTION_STATUS_UNKNOWN)
@@ -499,6 +505,9 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
                 MTM_LOG4("%d: tuple with xid=%d(csn= %ld) is %s in snapshot %ld",
 						 MyProcPid, xid, ts->csn, invisible ? "rollbacked" : "committed", MtmTx.snapshot);
                 MtmUnlock();
+				if (MtmGetSystemTime() - start > USECS_PER_SEC) { 
+					elog(WARNING, "Backend %d waits for %s transaction %x %ld usecs", MyProcPid, invisible ? "rollbacked" : "committed", xid, MtmGetSystemTime() - start);
+				}
                 return invisible;
             }
         }
@@ -510,7 +519,7 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
         }
     }
 	MtmUnlock();
-	elog(ERROR, "Failed to get status of XID %d", xid);
+	elog(ERROR, "Failed to get status of XID %d in %ld usec", xid, MtmGetSystemTime() - start);
 	return true;
 }    
 
@@ -1091,6 +1100,7 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 	msg.sxid = ts->xid;
 	msg.csn  = ts->csn;
 	msg.disabledNodeMask = Mtm->disabledNodeMask;
+	msg.connectivityMask = Mtm->connectivityMask;
 	msg.oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
 	memcpy(msg.gid, ts->gid, MULTIMASTER_MAX_GID_SIZE);
 
@@ -1118,6 +1128,7 @@ void MtmBroadcastPollMessage(MtmTransState* ts)
 	MtmArbiterMessage msg;
 	msg.code = MSG_POLL_REQUEST;
 	msg.disabledNodeMask = Mtm->disabledNodeMask;
+	msg.connectivityMask = Mtm->connectivityMask;
 	msg.oldestSnapshot = Mtm->nodes[MtmNodeId-1].oldestSnapshot;
 	memcpy(msg.gid, ts->gid, MULTIMASTER_MAX_GID_SIZE);
 
@@ -1703,8 +1714,6 @@ void MtmCheckQuorum(void)
 
 void MtmOnNodeDisconnect(int nodeId)
 { 
-	MtmTransState *ts;
-
 	if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
 	{
 		/* Node is already disabled */
@@ -1733,11 +1742,13 @@ void MtmOnNodeDisconnect(int nodeId)
 	}
 
 	MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout));
-
+#if 0
 	if (!MtmUseRaftable) 
 	{
 		MtmLock(LW_EXCLUSIVE);
 		if (!BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) { 
+			MtmTransState *ts;
+
 			MtmDisableNode(nodeId);
 			MtmCheckQuorum();
 			/* Interrupt voting for active transaction and abort them */
@@ -1751,7 +1762,10 @@ void MtmOnNodeDisconnect(int nodeId)
 			}
 		}
 		MtmUnlock();
-	} else {
+	} 
+	else 
+#endif
+	{ 
 		MtmRefreshClusterStatus(false, 0);
     }
 }
@@ -1964,6 +1978,11 @@ static void MtmInitialize()
 		Mtm->freeQueue = NULL;
 		for (i = 0; i < MtmNodes; i++) {
 			Mtm->nodes[i].oldestSnapshot = 0;
+			Mtm->nodes[i].disabledNodeMask = 0;
+			Mtm->nodes[i].connectivityMask = 0;
+			Mtm->nodes[i].lockGraphUsed = 0;
+			Mtm->nodes[i].lockGraphAllocated = 0;
+			Mtm->nodes[i].lockGraphData = NULL;
 			Mtm->nodes[i].transDelay = 0;
 			Mtm->nodes[i].lastStatusChangeTime = MtmGetSystemTime();
 			Mtm->nodes[i].con = MtmConnections[i];
@@ -2603,7 +2622,7 @@ _PG_init(void)
 	 * resources in mtm_shmem_startup().
 	 */
 	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmQueueSize);
-	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes);
+	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2);
 
     BgwPoolStart(MtmWorkers, MtmPoolConstructor);
 
@@ -3260,7 +3279,7 @@ Datum mtm_dump_lock_graph(PG_FUNCTION_ARGS)
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
 		size_t size;
-		char *data = RaftableGet(psprintf("lock-graph-%d", i+1), &size, NULL, false);
+		char* data = RaftableGet(psprintf("lock-graph-%d", i+1), &size, NULL, false);
 		if (data) { 
 			GlobalTransactionId *gtid = (GlobalTransactionId *)data;
 			GlobalTransactionId *last = (GlobalTransactionId *)(data + size);
@@ -3672,12 +3691,28 @@ static bool MtmProcessDDLCommand(char const* queryString)
 	}
 
 	MTM_LOG1("Sending utility: %s", queryWithContext);
-	LogLogicalMessage("MTM:GUC", queryWithContext, strlen(queryWithContext), true);
+	LogLogicalMessage("G", queryWithContext, strlen(queryWithContext)+1, true);
 
 	MtmTx.containsDML = true;
 	return false;
 }
 
+void MtmUpdateLockGraph(int nodeId, void const* messageBody, int messageSize)
+{
+	int allocated;
+	MtmLockNode(nodeId + MtmMaxNodes, LW_EXCLUSIVE);
+	allocated = Mtm->nodes[nodeId-1].lockGraphAllocated;
+	if (messageSize > allocated) { 
+		allocated = Max(Max(MULTIMASTER_LOCK_BUF_INIT_SIZE, allocated*2), messageSize);
+		Mtm->nodes[nodeId-1].lockGraphData = ShmemAlloc(allocated);
+		Mtm->nodes[nodeId-1].lockGraphAllocated = allocated;
+	}
+	memcpy(Mtm->nodes[nodeId-1].lockGraphData, messageBody, messageSize);
+	Mtm->nodes[nodeId-1].lockGraphUsed = messageSize;
+	MtmUnlockNode(nodeId + MtmMaxNodes);
+	MTM_LOG1("Update deadlock graph for node %d size %d", nodeId, messageSize);
+}
+
 static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 ProcessUtilityContext context, ParamListInfo params,
 							 DestReceiver *dest, char *completionTag)
@@ -3996,20 +4031,19 @@ MtmSerializeLock(PROCLOCK* proclock, void* arg)
 }
 
 static bool 
-MtmDetectGlobalDeadLock(PGPROC* proc)
+MtmDetectGlobalDeadLockFortXid(TransactionId xid)
 {
-    ByteBuffer buf;
-    PGXACT* pgxact = &ProcGlobal->allPgXact[proc->pgprocno];
 	bool hasDeadlock = false;
-
-    if (TransactionIdIsValid(pgxact->xid)) { 
+    if (TransactionIdIsValid(xid)) { 
+		ByteBuffer buf;
 		MtmGraph graph;
 		GlobalTransactionId gtid; 
 		int i;
 
         ByteBufferAlloc(&buf);
         EnumerateLocks(MtmSerializeLock, &buf);
 		RaftableSet(psprintf("lock-graph-%d", MtmNodeId), buf.data, buf.used, false);
+		MtmSleep(MSEC_TO_USEC(DeadlockTimeout));
 		MtmGraphInit(&graph);
 		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.used/sizeof(GlobalTransactionId));
         ByteBufferFree(&buf);
@@ -4024,9 +4058,9 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 				}
 			}
 		}
-		MtmGetGtid(pgxact->xid, &gtid);
+		MtmGetGtid(xid, &gtid);
 		hasDeadlock = MtmGraphFindLoop(&graph, &gtid);
-		elog(WARNING, "Distributed deadlock check for %u:%u = %d", gtid.node, gtid.xid, hasDeadlock);
+		elog(WARNING, "Distributed deadlock check by backend %d for %u:%u = %d", MyProcPid, gtid.node, gtid.xid, hasDeadlock);
 		if (!hasDeadlock) { 
 			/* There is no deadlock loop in graph, but deadlock can be caused by lack of apply workers: if all of them are busy, then some transactions
 			 * can not be appied just because there are no vacant workers and it cause additional dependency between transactions which is not 
@@ -4037,8 +4071,27 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 				hasDeadlock = true;
 				elog(WARNING, "Apply workers were blocked more than %d msec", 
 					 (int)USEC_TO_MSEC(MtmGetSystemTime() - lastPeekTime));
+			} else { 
+				MTM_LOG1("Enable deadlock timeout in backend %d for transaction %d", MyProcPid, xid);
+				enable_timeout_after(DEADLOCK_TIMEOUT, DeadlockTimeout);
 			}
 		}
 	}
     return hasDeadlock;
 }
+
+static bool 
+MtmDetectGlobalDeadLock(PGPROC* proc)
+{
+    PGXACT* pgxact = &ProcGlobal->allPgXact[proc->pgprocno];
+
+	MTM_LOG1("Detect global deadlock for %d by backend %d", pgxact->xid, MyProcPid);
+
+    return MtmDetectGlobalDeadLockFortXid(pgxact->xid);
+}
+
+Datum mtm_check_deadlock(PG_FUNCTION_ARGS)
+{
+	TransactionId xid = PG_GETARG_INT32(0);
+    PG_RETURN_BOOL(MtmDetectGlobalDeadLockFortXid(xid));
+}

multimaster.h

@@ -53,6 +53,7 @@
 #define MULTIMASTER_MAX_HOST_NAME_SIZE  64
 #define MULTIMASTER_MAX_LOCAL_TABLES    256
 #define MULTIMASTER_MAX_CTL_STR_SIZE    256
+#define MULTIMASTER_LOCK_BUF_INIT_SIZE  4096
 #define MULTIMASTER_BROADCAST_SERVICE   "mtm_broadcast"
 #define MULTIMASTER_ADMIN               "mtm_admin"
 
@@ -138,8 +139,9 @@ typedef struct
 	TransactionId  sxid;   /* Transaction ID at sender node */  
     XidStatus      status; /* Transaction status */	
 	csn_t          csn;    /* Local CSN in case of sending data from replica to master, global CSN master->replica */
-	nodemask_t     disabledNodeMask; /* Bitmask of disabled nodes at the sender of message */
 	csn_t          oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
+	nodemask_t     disabledNodeMask; /* Bitmask of disabled nodes at the sender of message */
+	nodemask_t     connectivityMask; /* Connectivity bittmask at the sender of message */
 	pgid_t         gid;    /* Global transaction identifier */
 } MtmArbiterMessage;
 
@@ -179,13 +181,18 @@ typedef struct
 	timestamp_t receiverStartTime;
 	timestamp_t senderStartTime;
 	timestamp_t lastHeartbeat;
+	nodemask_t  disabledNodeMask;      /* Bitmask of disabled nodes received from this node */
+	nodemask_t  connectivityMask;      /* Connectivity mask at this node */
 	int         senderPid;
 	int         receiverPid;
 	XLogRecPtr  flushPos;
-	csn_t       oldestSnapshot; /* Oldest snapshot used by active transactions at this node */	
+	csn_t       oldestSnapshot;        /* Oldest snapshot used by active transactions at this node */	
 	XLogRecPtr  restartLsn;
 	RepOriginId originId;
 	int         timeline;
+	void*       lockGraphData;
+	int         lockGraphAllocated;
+	int         lockGraphUsed;
 } MtmNodeInfo;
 
 typedef struct MtmTransState
@@ -223,7 +230,7 @@ typedef struct
 	MtmNodeStatus status;              /* Status of this node */
 	int recoverySlot;                  /* NodeId of recovery slot or 0 if none */
 	volatile slock_t spinlock;         /* spinlock used to protect access to hash table */
-	PGSemaphoreData sendSemaphore;   /* semaphore used to notify mtm-sender about new responses to coordinator */
+	PGSemaphoreData sendSemaphore;     /* semaphore used to notify mtm-sender about new responses to coordinator */
 	LWLockPadded *locks;               /* multimaster lock tranche */
 	TransactionId oldestXid;           /* XID of oldest transaction visible by any active transaction (local or global) */
 	nodemask_t disabledNodeMask;       /* bitmask of disabled nodes */
@@ -310,7 +317,7 @@ extern void  MtmAdjustSubtransactions(MtmTransState* ts);
 extern void  MtmBroadcastPollMessage(MtmTransState* ts);
 extern void  MtmLock(LWLockMode mode);
 extern void  MtmUnlock(void);
-extern void  MtmLockNode(int nodeId);
+extern void  MtmLockNode(int nodeId, LWLockMode mode);
 extern void  MtmUnlockNode(int nodeId);
 extern void  MtmDropNode(int nodeId, bool dropSlot);
 extern void  MtmRecoverNode(int nodeId);
@@ -340,6 +347,7 @@ extern XLogRecPtr MtmGetFlushPosition(int nodeId);
 extern bool MtmWatchdog(timestamp_t now);
 extern void MtmCheckHeartbeat(void);
 extern void MtmResetTransaction(void);
+extern void MtmUpdateLockGraph(int nodeId, void const* messageBody, int messageSize);
 extern PGconn *PQconnectdb_safe(const char *conninfo);