Unlock multimaster lock in atexit handler and FINALLy blocks

Author: knizhnik

arbiter.c

@@ -406,7 +406,7 @@ static int MtmConnectSocket(int node, int port, time_t timeout)
 	char portstr[MAXPGPATH];
 	MtmHandshakeMessage req;
 	MtmArbiterMessage   resp;
-	int sd;
+	int sd = -1;
 	int ret;
 	timestamp_t start = MtmGetSystemTime();
 	char const* host = Mtm->nodes[node].con.hostName;
@@ -424,7 +424,7 @@ static int MtmConnectSocket(int node, int port, time_t timeout)
 	ret = pg_getaddrinfo_all(host, portstr, &hint, &addrs);
 	if (ret != 0) 
 	{
-		MTM_ELOG(LOG, "Arbiter failed to resolve host '%s' by name: %s", host, gai_strerror(ret));
+		MTM_ELOG(LOG, "Arbiter failed to resolve host '%s' by name: (%d) %s", host, ret, gai_strerror(ret));
 		return -1;
 	}
 	BIT_SET(busy_mask, node);
@@ -435,15 +435,12 @@ static int MtmConnectSocket(int node, int port, time_t timeout)
 		sd = socket(AF_INET, SOCK_STREAM, 0);
 		if (sd < 0) {
 			MTM_ELOG(LOG, "Arbiter failed to create socket: %d", errno);
-			busy_mask = save_mask;
-			return -1;
+			goto Error;
 		}
 		rc = fcntl(sd, F_SETFL, O_NONBLOCK);
 		if (rc < 0) {
 			MTM_ELOG(LOG, "Arbiter failed to switch socket to non-blocking mode: %d", errno);
-			close(sd);
-			busy_mask = save_mask;
-			return -1;
+			goto Error;
 		}
 		for (addr = addrs; addr != NULL; addr = addr->ai_next)
 		{
@@ -461,18 +458,14 @@ static int MtmConnectSocket(int node, int port, time_t timeout)
 		beforeWait = MtmGetSystemTime();
 		if (errno != EINPROGRESS || start + MSEC_TO_USEC(timeout) < beforeWait ) {
 			MTM_ELOG(WARNING, "Arbiter failed to connect to %s:%d: error=%d", host, port, errno);
-			close(sd);
-			busy_mask = save_mask;
-			return -1;
+			goto Error;
 		} else {
 			rc = MtmWaitSocket(sd, true, MtmHeartbeatSendTimeout);
 			if (rc == 1) {
 				socklen_t optlen = sizeof(int); 
 				if (getsockopt(sd, SOL_SOCKET, SO_ERROR, (void*)&rc, &optlen) < 0) { 
 					MTM_ELOG(WARNING, "Arbiter failed to getsockopt for %s:%d: error=%d", host, port, errno);
-					close(sd);
-					busy_mask = save_mask;
-					return -1;
+					goto Error;
 				}
 				if (rc == 0) { 
 					break;
@@ -513,16 +506,28 @@ static int MtmConnectSocket(int node, int port, time_t timeout)
 		close(sd);
 		goto Retry;
 	}
-	
+	if (addrs)
+		pg_freeaddrinfo_all(hint.ai_family, addrs);
+
 	MtmLock(LW_EXCLUSIVE);
 	MtmCheckResponse(&resp);
 	MtmUnlock();
 
 	MtmOnNodeConnect(node+1);
 
 	busy_mask = save_mask;
-
+	
 	return sd;
+
+  Error:
+	busy_mask = save_mask;
+	if (sd >= 0) { 
+		close(sd);
+	}
+	if (addrs) {
+		pg_freeaddrinfo_all(hint.ai_family, addrs);
+	}
+	return -1;
 }
 
 

multimaster.c

@@ -233,7 +233,7 @@ int   MtmMin2PCTimeout;
 int   MtmMax2PCRatio;
 bool  MtmUseDtm;
 bool  MtmPreserveCommitOrder;
-bool  MtmVolksWagenMode;
+bool  MtmVolksWagenMode; /* Pretend to be normal postgres. This means skip some NOTICE's and use local sequences */
 
 TransactionId  MtmUtilityProcessedInXid;
 
@@ -263,6 +263,23 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 ProcessUtilityContext context, ParamListInfo params,
 							 DestReceiver *dest, char *completionTag);
 
+static bool MtmAtExitHookRegistered = false;
+
+/*
+ * Release multimaster main lock if been hold.
+ * This function is called when backend is terminated because of critical error or when error is catched 
+ * by FINALLY block 
+ */
+void MtmReleaseLock(void)
+{
+	if (MtmLockCount != 0) { 
+		Assert(Mtm->lastLockHolder == MyProcPid);
+		MtmLockCount = 0;
+		Mtm->lastLockHolder = 0;
+		LWLockRelease((LWLockId)&Mtm->locks[MTM_STATE_LOCK_ID]);
+	}
+}
+		
 /*
  * -------------------------------------------
  * Synchronize access to MTM structures.
@@ -276,8 +293,13 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 void MtmLock(LWLockMode mode)
 {
 	timestamp_t start, stop;
+	if (!MtmAtExitHookRegistered) { 
+		atexit(MtmReleaseLock);
+		MtmAtExitHookRegistered = true;
+	}
 	if (mode == LW_EXCLUSIVE || MtmLockCount != 0) { 
 		if (MtmLockCount++ != 0) { 
+			Assert(Mtm->lastLockHolder == MyProcPid);
 			return;
 		}
 	}
@@ -293,6 +315,7 @@ void MtmLock(LWLockMode mode)
 void MtmUnlock(void)
 {
 	if (MtmLockCount != 0 && --MtmLockCount != 0) { 
+		Assert(Mtm->lastLockHolder == MyProcPid);
 		return;
 	}
 	LWLockRelease((LWLockId)&Mtm->locks[MTM_STATE_LOCK_ID]);
@@ -2421,6 +2444,7 @@ static void MtmInitialize()
 	Mtm = (MtmState*)ShmemInitStruct(MULTIMASTER_NAME, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1), &found);
 	if (!found)
 	{
+		MemSet(Mtm, 0, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1));
 		Mtm->status = MTM_INITIALIZATION;
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);

multimaster.h

@@ -419,5 +419,6 @@ extern void MtmPrecommitTransaction(char const* gid);
 extern char* MtmGucSerialize(void);
 extern bool MtmTransIsActive(void);
 extern MtmTransState* MtmGetActiveTransaction(MtmL2List* list);
+extern void MtmReleaseLock(void);
 
 #endif

pglogical_apply.c

@@ -1041,6 +1041,7 @@ void MtmExecutor(void* work, size_t size)
 	int spill_file = -1;
 	int save_cursor = 0;
 	int save_len = 0;
+	MemoryContext topContext;
 
     s.data = work;
     s.len = size;
@@ -1054,7 +1055,7 @@ void MtmExecutor(void* work, size_t size)
 												ALLOCSET_DEFAULT_INITSIZE,
 												ALLOCSET_DEFAULT_MAXSIZE);
     }
-	MemoryContextSwitchTo(MtmApplyContext);
+	topContext = MemoryContextSwitchTo(MtmApplyContext);
 
 	replorigin_session_origin = InvalidRepOriginId;
     PG_TRY();
@@ -1144,7 +1145,9 @@ void MtmExecutor(void* work, size_t size)
     }
     PG_CATCH();
     {
-		MemoryContext oldcontext = MemoryContextSwitchTo(MtmApplyContext);
+		MemoryContext oldcontext;
+		MtmReleaseLock();
+		oldcontext = MemoryContextSwitchTo(MtmApplyContext);
 		MtmHandleApplyError();
 		MemoryContextSwitchTo(oldcontext);
 		EmitErrorReport();
@@ -1162,5 +1165,6 @@ void MtmExecutor(void* work, size_t size)
 	}
 #endif
     MemoryContextResetAndDeleteChildren(MtmApplyContext);
+	MemoryContextSwitchTo(topContext);
 }