Merge branch 'master' of https://github.com/postgrespro/postgres_cluster

Author: knizhnik

contrib/mmts/arbiter.c

@@ -243,30 +243,34 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 	if (!MtmResolveHostByName(host, addrs, &n_addrs)) {
 		elog(ERROR, "Arbiter failed to resolve host '%s' by name", host);
 	}
-  Retry:
-	sd = socket(AF_INET, SOCK_STREAM, 0);
-	if (sd < 0) {
-	    elog(ERROR, "Arbiter failed to create socket: %d", errno);
-    }
+
+Retry:
+
     while (1) {
 		int rc = -1;
+
+		sd = socket(AF_INET, SOCK_STREAM, 0);
+		if (sd < 0) {
+			elog(ERROR, "Arbiter failed to create socket: %d", errno);
+		}
 		for (i = 0; i < n_addrs; ++i) {
 			memcpy(&sock_inet.sin_addr, &addrs[i], sizeof sock_inet.sin_addr);
 			do {
 				rc = connect(sd, (struct sockaddr*)&sock_inet, sizeof(sock_inet));
 			} while (rc < 0 && errno == EINTR);
-			
+
 			if (rc >= 0 || errno == EINPROGRESS) {
 				break;
 			}
 		}
 		if (rc < 0) {
 			if ((errno != ENOENT && errno != ECONNREFUSED && errno != EINPROGRESS) || max_attempts == 0) {
-				elog(WARNING, "Arbiter failed to connect to %s:%d: %d", host, port, errno);
+				elog(WARNING, "Arbiter failed to connect to %s:%d: error=%d", host, port, errno);
 				return -1;
 			} else { 
 				max_attempts -= 1;
-				MtmSleep(MtmConnectTimeout);
+				elog(WARNING, "Arbiter trying to connect to %s:%d: error=%d", host, port, errno);
+				MtmSleep(5*MtmConnectTimeout);
 			}
 			continue;
 		} else {
@@ -282,7 +286,7 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 			req.hdr.sxid = ShmemVariableCache->nextXid;
 			req.hdr.csn  = MtmGetCurrentTime();
 			req.hdr.disabledNodeMask = Mtm->disabledNodeMask;
-			strcpy(req.connStr, Mtm->nodes[MtmNodeId-1].connStr);
+			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);
 				close(sd);
@@ -321,7 +325,7 @@ static void MtmOpenConnections()
 
 	for (i = 0; i < nNodes; i++) {
 		if (i+1 != MtmNodeId) { 
-			sockets[i] = MtmConnectSocket(Mtm->nodes[i].hostName, MtmArbiterPort + i + 1, MtmConnectAttempts);
+			sockets[i] = MtmConnectSocket(Mtm->nodes[i].con.hostName, MtmArbiterPort + i + 1, MtmConnectAttempts);
 			if (sockets[i] < 0) { 
 				MtmOnNodeDisconnect(i+1);
 			} 
@@ -345,7 +349,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 		if (sockets[node] >= 0) { 
 			close(sockets[node]);
 		}
-		sockets[node] = MtmConnectSocket(Mtm->nodes[node].hostName, MtmArbiterPort + node + 1, MtmReconnectAttempts);
+		sockets[node] = MtmConnectSocket(Mtm->nodes[node].con.hostName, MtmArbiterPort + node + 1, MtmReconnectAttempts);
 		if (sockets[node] < 0) { 
 			MtmOnNodeDisconnect(node+1);
 			return false;
@@ -385,7 +389,7 @@ static void MtmAcceptOneConnection()
 			resp.dxid = HANDSHAKE_MAGIC;
 			resp.sxid = ShmemVariableCache->nextXid;
 			resp.csn  = MtmGetCurrentTime();
-			MtmUpdateNodeConnStr(req.hdr.node, req.connStr);
+			MtmUpdateNodeConnectionInfo(&Mtm->nodes[req.hdr.node-1].con, req.connStr);
 			if (!MtmWriteSocket(fd, &resp, sizeof resp)) { 
 				elog(WARNING, "Arbiter failed to write response for handshake message to node %d", resp.node);
 				close(fd);

contrib/mmts/multimaster.c

@@ -166,10 +166,12 @@ int   MtmConnectAttempts;
 int   MtmConnectTimeout;
 int   MtmKeepaliveTimeout;
 int   MtmReconnectAttempts;
+MtmConnectionInfo* MtmConnections;
 
 static char* MtmConnStrs;
 static int MtmQueueSize;
 static int MtmWorkers;
+static int MtmVacuumDelay;
 static int MtmMinRecoveryLag;
 static int MtmMaxRecoveryLag;
 
@@ -402,26 +404,90 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
  * We collest oldest CSNs from all nodes and choose minimum from them.
  * If no such XID can be located, then return previously observed oldest XID
  */
+#if 0
 static TransactionId 
 MtmAdjustOldestXid(TransactionId xid)
 {
     if (TransactionIdIsValid(xid)) { 
         MtmTransState *ts, *prev = NULL;
-        
+		csn_t oldestSnapshot = 0;
+		int i;
+       
 		MtmLock(LW_EXCLUSIVE);
-        ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
-        if (ts != NULL && ts->status == TRANSACTION_STATUS_COMMITTED) {  /* committed transactions have same CSNs at all nodes */
-			csn_t oldestSnapshot;
-			int i;
+		for (ts = Mtm->transListHead; ts != NULL; ts = ts->next) { 
+			if (TransactionIdPrecedes(ts->xid, xid)
+				&& ts->status == TRANSACTION_STATUS_COMMITTED
+				&& ts->csn > oldestSnapshot)
+			{
+				oldestSnapshot = ts->csn;
+			}
+		}		
+		Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
+		for (i = 0; i < MtmNodes; i++) { 
+			if (!BIT_CHECK(Mtm->disabledNodeMask, i)
+				&& Mtm->nodes[i].oldestSnapshot < oldestSnapshot) 
+			{ 
+				oldestSnapshot = Mtm->nodes[i].oldestSnapshot;
+			}
+		}
+		oldestSnapshot -= MtmVacuumDelay*USEC;
+		for (ts = Mtm->transListHead; 
+			 ts != NULL 
+				 && ts->csn < oldestSnapshot
+				 && TransactionIdPrecedes(ts->xid, xid)
+				 && (ts->status == TRANSACTION_STATUS_COMMITTED ||
+					 ts->status == TRANSACTION_STATUS_ABORTED);
+			 ts = ts->next) 
+		{
+			if (ts->status == TRANSACTION_STATUS_COMMITTED) { 
+				prev = ts;
+			}
+		}
+		if (prev != NULL) { 
+			for (ts = Mtm->transListHead; ts != prev; ts = ts->next) {
+				/* Remove information about too old transactions */
+				Assert(ts->status != TRANSACTION_STATUS_UNKNOWN);
+				hash_search(MtmXid2State, &ts->xid, HASH_REMOVE, NULL);
+			}
+			Mtm->transListHead = prev;
+			Mtm->oldestXid = xid = prev->xid;            
+        } else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) {
+            xid = Mtm->oldestXid;
+        }
+		MtmUnlock();
+    }
+    return xid;
+}
+#else
+static TransactionId 
+MtmAdjustOldestXid(TransactionId xid)
+{
+    if (TransactionIdIsValid(xid)) { 
+        MtmTransState *ts, *prev = NULL;
+        int i;
 
-			Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot = ts->csn;
+		MtmLock(LW_EXCLUSIVE);
+        ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+        if (ts != NULL && ts->status == TRANSACTION_STATUS_COMMITTED) { 
+			csn_t oldestSnapshot = ts->csn;
+			Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
 			for (i = 0; i < MtmNodes; i++) { 
-				if (Mtm->nodes[i].oldestSnapshot < oldestSnapshot) { 
+				if (!BIT_CHECK(Mtm->disabledNodeMask, i)
+					&& Mtm->nodes[i].oldestSnapshot < oldestSnapshot) 
+				{ 
 					oldestSnapshot = Mtm->nodes[i].oldestSnapshot;
 				}
 			}
-			for (ts = Mtm->transListHead; ts != NULL && ts->csn < oldestSnapshot; prev = ts, ts = ts->next) { 
-				Assert(ts->status == TRANSACTION_STATUS_COMMITTED || ts->status == TRANSACTION_STATUS_ABORTED || ts->status == TRANSACTION_STATUS_IN_PROGRESS);
+			oldestSnapshot -= MtmVacuumDelay*USEC;
+
+			for (ts = Mtm->transListHead; 
+				 ts != NULL 
+					 && ts->csn < oldestSnapshot
+					 && TransactionIdPrecedes(ts->xid, xid)
+					 && (ts->status == TRANSACTION_STATUS_COMMITTED ||
+						 ts->status == TRANSACTION_STATUS_ABORTED);
+				 prev = ts, ts = ts->next) 
+			{ 
 				if (prev != NULL) { 
 					/* Remove information about too old transactions */
 					hash_search(MtmXid2State, &prev->xid, HASH_REMOVE, NULL);
@@ -431,14 +497,14 @@ MtmAdjustOldestXid(TransactionId xid)
         if (prev != NULL) { 
             Mtm->transListHead = prev;
             Mtm->oldestXid = xid = prev->xid;            
-        } else { 
+        } else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) { 
             xid = Mtm->oldestXid;
         }
 		MtmUnlock();
     }
     return xid;
 }
-
+#endif
 /*
  * -------------------------------------------
  * Transaction list manipulation
@@ -989,7 +1055,7 @@ MtmBuildConnectivityMatrix(nodemask_t* matrix, bool nowait)
 	for (i = 0; i < n; i++) { 
 		if (i+1 != MtmNodeId) { 
 			void* data = PaxosGet(psprintf("node-mask-%d", i+1), NULL, NULL, nowait);
-			matrix[i] = *(nodemask_t*)data;
+			matrix[i] = data ? *(nodemask_t*)data : 0;
 		} else { 
 			matrix[i] = Mtm->connectivityMask;
 		}
@@ -1153,6 +1219,7 @@ static void MtmInitialize()
 		for (i = 0; i < MtmNodes; i++) {
 			Mtm->nodes[i].oldestSnapshot = 0;
 			Mtm->nodes[i].transDelay = 0;
+			Mtm->nodes[i].con = MtmConnections[i];
 		}
 		PGSemaphoreCreate(&Mtm->votingSemaphore);
 		PGSemaphoreReset(&Mtm->votingSemaphore);
@@ -1178,17 +1245,17 @@ MtmShmemStartup(void)
 	MtmInitialize();
 }
 
-void MtmUpdateNodeConnStr(int nodeId, char const* connStr)
+void MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr)
 {
 	char const* host;
 	char const* end;
 	int         hostLen;
 
 	if (strlen(connStr) >= MULTIMASTER_MAX_CONN_STR_SIZE) {
-		elog(ERROR, "Too long (%d) connection string '%s' for node %d, limit is %d", 
-			 (int)strlen(connStr), connStr, nodeId, MULTIMASTER_MAX_CONN_STR_SIZE-1);
+		elog(ERROR, "Too long (%d) connection string '%s': limit is %d", 
+			 (int)strlen(connStr), connStr, MULTIMASTER_MAX_CONN_STR_SIZE-1);
 	}
-	strcpy(Mtm->nodes[nodeId-1].connStr, connStr);
+	strcpy(conn->connStr, connStr);
 
 	host = strstr(connStr, "host=");
 	if (host == NULL) {
@@ -1198,30 +1265,46 @@ void MtmUpdateNodeConnStr(int nodeId, char const* connStr)
 	for (end = host; *end != ' ' && *end != '\0'; end++);
 	hostLen = end - host;
 	if (hostLen >= MULTIMASTER_MAX_HOST_NAME_SIZE) {
-		elog(ERROR, "Too long (%d) host name '%.*s' for node %d, limit is %d", 
-			 hostLen, hostLen, host, nodeId, MULTIMASTER_MAX_HOST_NAME_SIZE-1);
+		elog(ERROR, "Too long (%d) host name '%.*s': limit is %d", 
+			 hostLen, hostLen, host, MULTIMASTER_MAX_HOST_NAME_SIZE-1);
 	}
-	memcpy(Mtm->nodes[nodeId-1].hostName, host, hostLen);
-	Mtm->nodes[nodeId-1].hostName[hostLen] = '\0';
+	memcpy(conn->hostName, host, hostLen);
+	conn->hostName[hostLen] = '\0';
 }
 
 static void MtmSplitConnStrs(void)
 {
 	int i;
-	char* copy =  strdup(MtmConnStrs);
+	char* copy =  pstrdup(MtmConnStrs);
     char* connStr = copy;
     char* connStrEnd = connStr + strlen(connStr);
 
+	for (i = 0; connStr < connStrEnd; i++) { 
+		char* p = strchr(connStr, ',');
+        if (p == NULL) { 
+            p = connStrEnd;
+        }
+		connStr = p + 1;
+	}
+	if (i > MAX_NODES) { 
+		elog(ERROR, "Multimaster with more than %d nodes is not currently supported", MAX_NODES);
+	}	
+	if (i < 2) { 
+        elog(ERROR, "Multimaster should have at least two nodes");
+	}	
+	MtmNodes = i;
+	MtmConnections = (MtmConnectionInfo*)palloc(i*sizeof(MtmConnectionInfo));
+	connStr = copy;
+
 	for (i = 0; connStr < connStrEnd; i++) { 
         char* p = strchr(connStr, ',');
         if (p == NULL) { 
             p = connStrEnd;
         }
-		if (i == MAX_NODES) { 
-			elog(ERROR, "Multimaster with more than %d nodes is not currently supported", MAX_NODES);
-		}	
 		*p = '\0';
-		MtmUpdateNodeConnStr(i+1, connStr);
+
+		MtmUpdateNodeConnectionInfo(&MtmConnections[i], connStr);
+
 		if (i+1 == MtmNodeId) { 
 			char* dbName = strstr(connStr, "dbname=");
 			char* end;
@@ -1232,20 +1315,13 @@ static void MtmSplitConnStrs(void)
 			dbName += 7;
 			for (end = dbName; *end != ' ' && *end != '\0'; end++);
 			len = end - dbName;
-			MtmDatabaseName = (char*)malloc(len + 1);
+			MtmDatabaseName = (char*)palloc(len + 1);
 			memcpy(MtmDatabaseName, dbName, len);
 			MtmDatabaseName[len] = '\0';
 		}
 		connStr = p + 1;
     }
-	free(copy);
-	if (i < 2) { 
-        elog(ERROR, "Multimaster should have at least two nodes");
-	}	
-	MtmNodes = i;
-	if (MtmNodeId > MtmNodes) {
-		elog(ERROR, "Invalid node id %d for specified nubmer of nodes %d", MtmNodeId, MtmNodes);
-	}
+	pfree(copy);
 }		
 
 void
@@ -1309,6 +1385,21 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomIntVariable(
+		"multimaster.vacuum_delay",
+		"Minimal age of records which can be vacuumed (seconds)",
+		NULL,
+		&MtmVacuumDelay,
+		1,
+		1,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.queue_size",
 		"Multimaster queue size",

contrib/mmts/multimaster.h

@@ -81,10 +81,16 @@ typedef enum
 
 typedef struct
 {
+	char hostName[MULTIMASTER_MAX_HOST_NAME_SIZE];
+	char connStr[MULTIMASTER_MAX_CONN_STR_SIZE];
+} MtmConnectionInfo;
+
+
+typedef struct
+{
+	MtmConnectionInfo con;
 	time_t transDelay;
 	csn_t  oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
-	char   hostName[MULTIMASTER_MAX_HOST_NAME_SIZE];
-	char   connStr[MULTIMASTER_MAX_CONN_STR_SIZE];
 } MtmNodeInfo;
 
 typedef struct MtmTransState
@@ -152,6 +158,8 @@ extern int   MtmReconnectAttempts;
 extern int   MtmKeepaliveTimeout;
 extern HTAB* MtmXid2State;
 
+extern MtmConnectionInfo* MtmConnections;
+
 extern void  MtmArbiterInitialize(void);
 extern void  MtmStartReceivers(void);
 extern csn_t MtmTransactionSnapshot(TransactionId xid);
@@ -183,6 +191,6 @@ extern XidStatus MtmGetGlobalTransactionStatus(char const* gid);
 extern bool  MtmIsRecoveredNode(int nodeId);
 extern void  MtmRefreshClusterStatus(bool nowait);
 extern void  MtmSwitchClusterMode(MtmNodeStatus mode);
-extern void  MtmUpdateNodeConnStr(int nodeId, char const* connStr);
+extern void  MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr);
 
 #endif

contrib/mmts/pglogical_receiver.c

@@ -578,8 +578,8 @@ void MtmStartReceivers(void)
 
 	for (i = 0; i < MtmNodes; i++) {
         if (i+1 != MtmNodeId) {
-            ReceiverArgs* ctx = (ReceiverArgs*)malloc(sizeof(ReceiverArgs));
-            ctx->receiver_conn_string = psprintf("replication=database %s", Mtm->nodes[i].connStr);
+            ReceiverArgs* ctx = (ReceiverArgs*)palloc(sizeof(ReceiverArgs));
+            ctx->receiver_conn_string = psprintf("replication=database %s", MtmConnections[i].connStr);
             sprintf(ctx->receiver_slot, MULTIMASTER_SLOT_PATTERN, MtmNodeId);
             ctx->local_node = MtmNodeId;
             ctx->remote_node = i+1;