Avoid writes in critical section

Author: knizhnik

contrib/mmts/arbiter.c

@@ -73,9 +73,9 @@
 
 #include "multimaster.h"
 
-#define MAX_ROUTES      16
-#define BUFFER_SIZE     1024
-#define HANDSHAKE_MAGIC 0xCAFEDEED
+#define MAX_ROUTES       16
+#define INIT_BUFFER_SIZE 1024
+#define HANDSHAKE_MAGIC  0xCAFEDEED
 
 typedef struct
 {
@@ -97,7 +97,8 @@ typedef struct
 typedef struct 
 {
 	int used;
-	MtmArbiterMessage data[BUFFER_SIZE];
+	int size;
+	MtmArbiterMessage* data;
 } MtmBuffer;
 
 static int*      sockets;
@@ -448,11 +449,14 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 	}
 
 	/* Some node considered that I am dead, so switch to recovery mode */
+	MtmLock(LW_EXCLUSIVE);
 	if (BIT_CHECK(resp.disabledNodeMask, MtmNodeId-1)) { 
 		elog(WARNING, "Node %d thinks that I was dead", resp.node);
 		BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
 		MtmSwitchClusterMode(MTM_RECOVERY);
 	}
+	MtmUnlock();
+
 	return sd;
 }
 
@@ -489,7 +493,9 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 	while (true) {
 		if (sockets[node] >= 0 && BIT_CHECK(Mtm->reconnectMask, node)) {
 			elog(WARNING, "Arbiter is forced to reconnect to node %d", node+1); 
+			MtmLock(LW_EXCLUSIVE);		
 			BIT_CLEAR(Mtm->reconnectMask, node);
+			MtmUnlock();
 			close(sockets[node]);
 			sockets[node] = -1;
 		}
@@ -504,7 +510,9 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 				MtmOnNodeDisconnect(node+1);
 				return false;
 			}
+			MtmLock(LW_EXCLUSIVE);		
 			BIT_CLEAR(Mtm->reconnectMask, node);
+			MtmUnlock();
 			MTM_LOG3("Arbiter restablished connection with node %d", node+1);
 		} else { 
 			return true;
@@ -550,7 +558,6 @@ static void MtmAcceptOneConnection()
 				close(fd);
 			} else { 
 				MTM_LOG1("Arbiter established connection with node %d", req.hdr.node); 
-				BIT_CLEAR(Mtm->connectivityMask, req.hdr.node-1);
 				MtmRegisterSocket(fd, req.hdr.node-1);
 				sockets[req.hdr.node-1] = fd;
 				MtmOnNodeConnect(req.hdr.node);
@@ -596,12 +603,9 @@ static void MtmAcceptIncomingConnections()
 static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, MtmTransState* ts)
 {
 	MtmBuffer* buf = &txBuffer[node];
-	if (buf->used == BUFFER_SIZE) { 
-		if (!MtmSendToNode(node, buf->data, buf->used*sizeof(MtmArbiterMessage))) {			
-			buf->used = 0;
-			return;
-		}
-		buf->used = 0;
+	if (buf->used == buf->size) {
+		buf->size = buf->size ? buf->size*2 : INIT_BUFFER_SIZE;
+		buf->data = repalloc(buf->data, buf->size * sizeof(MtmArbiterMessage));
 	}
 	buf->data[buf->used].dxid = xid;
 
@@ -638,7 +642,7 @@ static void MtmTransSender(Datum arg)
 	int nNodes = MtmMaxNodes;
 	int i;
 
-	MtmBuffer* txBuffer = (MtmBuffer*)palloc(sizeof(MtmBuffer)*nNodes);
+	MtmBuffer* txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
 
 	InitializeTimeouts();
 
@@ -653,10 +657,6 @@ static void MtmTransSender(Datum arg)
 
 	MtmOpenConnections();
 
-	for (i = 0; i < nNodes; i++) { 
-		txBuffer[i].used = 0;
-	}
-
 	while (!stop) {
 		MtmTransState* ts;		
 		PGSemaphoreLock(&Mtm->votingSemaphore);
@@ -721,7 +721,7 @@ static void MtmTransReceiver(Datum arg)
 	int nNodes = MtmMaxNodes;
 	int nResponses;
 	int i, j, n, rc;
-	MtmBuffer* rxBuffer = (MtmBuffer*)palloc(sizeof(MtmBuffer)*nNodes);
+	MtmBuffer* rxBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
 	timestamp_t lastHeartbeatCheck = MtmGetSystemTime();
 	timestamp_t now;
 
@@ -742,7 +742,8 @@ static void MtmTransReceiver(Datum arg)
 	MtmAcceptIncomingConnections();
 
 	for (i = 0; i < nNodes; i++) { 
-		rxBuffer[i].used = 0;
+		rxBuffer[i].size = INIT_BUFFER_SIZE;
+		rxBuffer[i].data = palloc(INIT_BUFFER_SIZE*sizeof(MtmArbiterMessage));
 	}
 
 	while (!stop) {
@@ -786,7 +787,7 @@ static void MtmTransReceiver(Datum arg)
 					continue;
 				}  
 
-				rc = MtmReadFromNode(i, (char*)rxBuffer[i].data + rxBuffer[i].used, BUFFER_SIZE-rxBuffer[i].used);
+				rc = MtmReadFromNode(i, (char*)rxBuffer[i].data + rxBuffer[i].used, rxBuffer[i].size-rxBuffer[i].used);
 				if (rc <= 0) { 
 					continue;
 				}

contrib/mmts/multimaster.c

@@ -1433,8 +1433,11 @@ void MtmOnNodeDisconnect(int nodeId)
 		/* Avoid false detection of node failure and prevent node status blinking */
 		return;
 	}
+	MtmLock(LW_EXCLUSIVE);
 	BIT_SET(Mtm->connectivityMask, nodeId-1);
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
+	MtmUnlock();
+
 	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false);
 
 	MtmSleep(MSEC_TO_USEC(MtmRaftPollDelay));
@@ -1461,7 +1464,10 @@ void MtmOnNodeDisconnect(int nodeId)
 
 void MtmOnNodeConnect(int nodeId)
 {
+	MtmLock(LW_EXCLUSIVE);	
 	BIT_CLEAR(Mtm->connectivityMask, nodeId-1);
+	MtmUnlock();
+
 	MTM_LOG1("Reconnect node %d", nodeId);
 	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false);
 }
@@ -2195,6 +2201,8 @@ void MtmRecoverNode(int nodeId)
 
 void MtmDropNode(int nodeId, bool dropSlot)
 {
+	MtmLock(LW_EXCLUSIVE);
+
 	if (!BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
 	{
 		if (nodeId <= 0 || nodeId > Mtm->nLiveNodes) 
@@ -2212,6 +2220,8 @@ void MtmDropNode(int nodeId, bool dropSlot)
 			ReplicationSlotDrop(psprintf(MULTIMASTER_SLOT_PATTERN, nodeId));
 		}		
 	}
+
+	MtmUnlock();
 }
 static void
 MtmOnProcExit(int code, Datum arg)
@@ -2402,7 +2412,9 @@ mtm_add_node(PG_FUNCTION_ARGS)
 	} 
 	else 
 	{ 
-		int nodeId = Mtm->nAllNodes;
+		int nodeId;
+		MtmLock(LW_EXCLUSIVE);	
+		nodeId = Mtm->nAllNodes;
 		elog(NOTICE, "Add node %d: '%s'", nodeId+1, connStr);
 		MtmUpdateNodeConnectionInfo(&Mtm->nodes[nodeId].con, connStr);
 		Mtm->nodes[nodeId].transDelay = 0;
@@ -2412,6 +2424,8 @@ mtm_add_node(PG_FUNCTION_ARGS)
 
 		BIT_SET(Mtm->disabledNodeMask, nodeId);
 		Mtm->nAllNodes += 1;
+		MtmUnlock();
+
 		MtmStartReceiver(nodeId+1, true);
 	}
     PG_RETURN_VOID();