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

Author: kelvich

contrib/mmts/multimaster.c

@@ -530,7 +530,7 @@ MtmAdjustOldestXid(TransactionId xid)
 		if (prev != NULL) { 
 			Mtm->transListHead = prev;
 			Mtm->oldestXid = xid = prev->xid;            
-			MTM_LOG2("%d: MtmAdjustOldestXid: oldestXid=%d, olderstSnapshot=%ld", MyProcPid, xid, oldestSnapshot);
+			MTM_LOG2("%d: MtmAdjustOldestXid: oldestXid=%d, oldestSnapshot=%ld", MyProcPid, xid, oldestSnapshot);
 		} else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) {  
 			xid = Mtm->oldestXid;
 		}

contrib/mmts/tests2/lib/bank_client.py

@@ -133,6 +133,7 @@ def check_total(self):
         def tx(conn, cur):
             cur.execute('select sum(amount) from bank_test')
             res = cur.fetchone()
+            conn.commit()
             if res[0] != 0:
                 print("Isolation error, total = %d" % (res[0],))
                 raise BaseException

contrib/raftable/raft/include/raft.h

@@ -60,6 +60,11 @@ raft_t raft_init(raft_config_t *config);
  */
 raft_bool_t raft_peer_up(raft_t r, int id, char *host, int port, raft_bool_t self);
 
+/*
+ * Returns the number of entried applied by the leader.
+ */
+int raft_progress(raft_t r);
+
 /*
  * Remove a previously added peer named 'id'.
  */

contrib/raftable/raft/src/raft.c

@@ -1182,3 +1182,7 @@ bool raft_is_leader(raft_t r) {
 int raft_get_leader(raft_t r) {
 	return r->leader;
 }
+
+int raft_progress(raft_t r) {
+	return r->log.applied;
+}

contrib/raftable/raftable--1.0.sql

@@ -1,6 +1,12 @@
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION raftable" to load this file. \quit
 
+-- sync
+CREATE FUNCTION raftable_sync(timeout_ms int)
+RETURNS void
+AS 'MODULE_PATHNAME','raftable_sql_sync'
+LANGUAGE C;
+
 -- get
 CREATE FUNCTION raftable(key varchar(64))
 RETURNS text

contrib/raftable/raftable.c

@@ -38,6 +38,7 @@ PG_MODULE_MAGIC;
 
 PG_FUNCTION_INFO_V1(raftable_sql_get);
 PG_FUNCTION_INFO_V1(raftable_sql_set);
+PG_FUNCTION_INFO_V1(raftable_sql_sync);
 PG_FUNCTION_INFO_V1(raftable_sql_list);
 
 static struct {
@@ -327,6 +328,47 @@ static bool try_sending_update(RaftableUpdate *ru, size_t size, timeout_t *timeo
 	return true;
 }
 
+bool raftable_sync(int timeout_ms)
+{
+	RaftableUpdate ru;
+	size_t size = sizeof(ru);
+	timeout_t timeout;
+
+	Assert(wcfg.id >= 0);
+
+	ru.expector = wcfg.id;
+	ru.fieldnum = 0;
+
+	if (timeout_ms < 0)
+	{
+		while (true)
+		{
+			timeout_start(&timeout, 100);
+
+			if (try_sending_update(&ru, size, &timeout))
+				return true;
+			else
+				disconnect_leader();
+		}
+	}
+	else
+	{
+		timeout_start(&timeout, timeout_ms);
+
+		TIMEOUT_LOOP_START(&timeout);
+		{
+			if (try_sending_update(&ru, size, &timeout))
+				return true;
+			else
+				disconnect_leader();
+		}
+		TIMEOUT_LOOP_END(&timeout);
+	}
+
+	elog(WARNING, "failed to sync after %d ms", timeout_elapsed_ms(&timeout));
+	return false;
+}
+
 bool raftable_set(const char *key, const char *value, size_t vallen, int timeout_ms)
 {
 	RaftableField *f;
@@ -408,6 +450,16 @@ raftable_sql_set(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+Datum
+raftable_sql_sync(PG_FUNCTION_ARGS)
+{
+	int timeout_ms = PG_GETARG_INT32(0);
+
+	raftable_sync(timeout_ms);
+
+	PG_RETURN_VOID();
+}
+
 void raftable_every(void (*func)(const char *, const char *, size_t, void *), void *arg)
 {
 	void *scan;

contrib/raftable/raftable.h

@@ -1,6 +1,9 @@
 #ifndef __RAFTABLE_H__
 #define __RAFTABLE_H__
 
+/* Syncs with the raftable leader. Gives up after 'timeout_ms' milliseconds. */
+bool raftable_sync(int timeout_ms);
+
 /* Gets value by key. Returns the value or NULL if not found. */
 char *raftable_get(const char *key, size_t *vallen);
 

contrib/raftable/t/000_basic.pl

@@ -77,7 +77,7 @@ sub start_nodes
 $able->psql('postgres', "select raftable('world', '$tests{world}', $timeout_ms);");
 
 $baker->start;
-sleep(5);
+$baker->psql('postgres', "select raftable_sync($timeout_ms);");
 while (my ($key, $value) = each(%tests))
 {
 	my ($rc, $stdout, $stderr) = $baker->psql('postgres', "select raftable('$key');");

contrib/raftable/worker.c

@@ -322,14 +322,30 @@ static void attend(Client *c)
 
 		Assert(c->expect.id == NOBODY); /* client shouldn't send multiple updates at once */
 
-		u.len = c->msglen;
-		u.data = c->msg;
 		c->expect.id = ru->expector;
-		index = raft_emit(raft, u);
-		if (index >= 0)
-			c->expect.index = index;
-		else
-			c->good = false;
+		if (ru->fieldnum > 0) {
+			// an actual update
+			u.len = c->msglen;
+			u.data = c->msg;
+			index = raft_emit(raft, u);
+			if (index >= 0)
+				c->expect.index = index;
+			else
+				c->good = false;
+		} else {
+			// a sync command
+			c->expect.index = raft_progress(raft);
+			if (raft_applied(raft, c->expect.id, c->expect.index))
+			{
+				int ok = 1;
+				if (send(c->sock, &ok, sizeof(ok), 0) != sizeof(ok))
+				{
+					fprintf(stderr, "failed to notify client\n");
+					c->good = false;
+				}
+				c->expect.id = NOBODY;
+			}
+		}
 		pfree(c->msg);
 		c->msg = NULL;
 	}