Merge branch 'master' into raftable-dynamic

Author: kvap

multimaster.c

@@ -1708,6 +1708,26 @@ static void MtmLoadLocalTables(void)
 	}
 }
 
+static void MtmRaftableInitialize()
+{
+	int i;
+
+	for (i = 0; i < MtmNodes; i++)
+	{
+		char const* raftport = strstr(MtmConnections[i].connStr, "raftport=");
+		int port;
+		if (raftport != NULL) {
+			if (sscanf(raftport+9, "%d", &port) != 1) { 
+				elog(ERROR, "Invalid raftable port: %s", raftport+9);
+			}
+		} else {
+			port = MtmRaftablePort + i;
+		}
+		raftable_peer(i, MtmConnections[i].hostName, port);
+	}
+	raftable_start(MtmNodeId - 1);
+}
+
 
 static void MtmInitialize()
 {
@@ -1863,33 +1883,6 @@ static void MtmSplitConnStrs(void)
 	pfree(copy);
 }		
 
-static void MtmRaftableInitialize()
-{
-	int i;
-	WorkerConfig wcfg;
-
-	for (i = 0; i < RAFTABLE_PEERS_MAX; i++) 
-	{
-		wcfg.peers[i].up = false;
-	}
-
-	for (i = 0; i < MtmNodes; i++)
-	{
-		char const* raftport = strstr(MtmConnections[i].connStr, "raftport=");
-		if (raftport != NULL) {
-			if (sscanf(raftport+9, "%d", &wcfg.peers[i].port) != 1) { 
-				elog(ERROR, "Invalid raftable port: %s", raftport+9);
-			}
-		} else {
-			wcfg.peers[i].port = MtmRaftablePort + i;
-		}
-		wcfg.peers[i].up = true;
-		strncpy(wcfg.peers[i].host, MtmConnections[i].hostName, sizeof(wcfg.peers[i].host));
-	}
-	wcfg.id = MtmNodeId-1;
-	worker_register(&wcfg);
-}
-
 void
 _PG_init(void)
 {
@@ -2249,7 +2242,7 @@ _PG_init(void)
 
     BgwPoolStart(MtmWorkers, MtmPoolConstructor);
 
-	//MtmRaftableInitialize();
+	MtmRaftableInitialize();
 	MtmArbiterInitialize();
 
 	/*

raftable.c

@@ -9,23 +9,36 @@
  */
 void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, bool nowait)
 {
+	void *value;
+	size_t vallen;
 	if (!MtmUseRaftable) { 
 		return NULL;
 	}
-	return raftable_get(key, size);
+	value = raftable_get(key, &vallen, MtmHeartbeatSendTimeout);
+	if (size != NULL) {
+		*size = vallen;
+	}
+	return value;
 }
 
 
 void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 {
 	if (MtmUseRaftable) {
+		int tries = 10;
 		timestamp_t start, stop;
 		start = MtmGetSystemTime();
 		if (nowait) {
 			raftable_set(key, value, size, 0);
 		} else { 
-			while (!raftable_set(key, value, size, MtmHeartbeatSendTimeout)) { 
+			while (!raftable_set(key, value, size, MtmHeartbeatSendTimeout))
+			{
 				MtmCheckHeartbeat();
+				if (tries-- <= 0)
+				{
+					MTM_LOG1("RaftableSet nowait=%d, all attempts failed", nowait);
+					break;
+				}
 			}
 		}
 		stop = MtmGetSystemTime();