port referee back

kelvich · kelvich · commit 12b14911289f · 2019-03-13T15:42:28.000+03:00
diff --git a/Dockerfile b/Dockerfile
@@ -6,8 +6,8 @@ COPY ./ /pg/mmts/
 RUN export USE_PGXS=1 && \
     cd /pg/mmts && make clean && make install
 
-# RUN export USE_PGXS=1 && \
-#     cd /pg/src/contrib/referee && make clean && make install
+RUN export USE_PGXS=1 && \
+    cd /pg/src/contrib/referee && make clean && make install
 
 # pg_regress client assumes such dir exists on server
 RUN cp /pg/src/src/test/regress/*.so /pg/install/lib/postgresql/
diff --git a/src/dmq.c b/src/dmq.c
@@ -772,8 +772,8 @@ dmq_handle_message(StringInfo msg, shm_mq_handle **mq_handles, dsm_segment *seg)
 	}
 
 	mtm_log(DmqTraceIncoming,
-			"[DMQ] got message %s.%s, passing to %d", stream_name, body,
-			sub->procno);
+			"[DMQ] got message %s.%s (len=%d), passing to %d", stream_name, body,
+			body_len, sub->procno);
 
 	/* and send it */
 	res = shm_mq_send(mq_handles[sub->procno], body_len, body, false);
@@ -1546,8 +1546,8 @@ dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask)
 			*sender_id = i;
 
 			mtm_log(DmqTraceIncoming,
-					"[DMQ] dmq_pop_nb: got message %s from %s",
-					(char *) data, dmq_local.inhandles[i].name);
+					"[DMQ] dmq_pop_nb: got message %s (len=%zu) from %s",
+					(char *) data, len, dmq_local.inhandles[i].name);
 			return true;
 		}
 		else if (res == SHM_MQ_DETACHED)
diff --git a/src/include/resolver.h b/src/include/resolver.h
@@ -22,6 +22,7 @@ extern void ResolverInit(void);
 extern BackgroundWorkerHandle *ResolverStart(Oid db_id, Oid user_id);
 extern void ResolveTransactionsForNode(int node_id, int n_all_nodes);
 extern void ResolveAllTransactions(int n_all_nodes);
+extern void ResolveForRefereeWinner(int n_all_nodes);
 extern char *MtmTxStateMnem(MtmTxState state);
 
 #endif  /* RESOLVER_H */
diff --git a/src/pglogical_receiver.c b/src/pglogical_receiver.c
@@ -999,7 +999,7 @@ MtmStartReceiver(int nodeId, Oid db_id, Oid user_id, pid_t monitor_pid)
 	MemSet(&worker, 0, sizeof(BackgroundWorker));
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |	BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
-	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
+	worker.bgw_restart_time = 1;
 	worker.bgw_main_arg = Int32GetDatum(nodeId);
 	worker.bgw_notify_pid = monitor_pid;
 
diff --git a/src/resolver.c b/src/resolver.c
@@ -132,7 +132,7 @@ ResolverStart(Oid db_id, Oid user_id)
 	MemSet(&worker, 0, sizeof(BackgroundWorker));
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |	BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
-	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
+	worker.bgw_restart_time = 1;
 
 	memcpy(worker.bgw_extra, &db_id, sizeof(Oid));
 	memcpy(worker.bgw_extra + sizeof(Oid), &user_id, sizeof(Oid));
@@ -223,6 +223,46 @@ ResolveAllTransactions(int n_all_nodes)
 	ResolveTransactionsForNode(-1, n_all_nodes);
 }
 
+void
+ResolveForRefereeWinner(int n_all_nodes)
+{
+	HASH_SEQ_STATUS		hash_seq;
+	resolver_tx		   *tx;
+
+	mtm_log(LOG, "ResolveForRefereeWinner");
+
+	LWLockAcquire(resolver_state->lock, LW_EXCLUSIVE);
+	load_tasks(-1, n_all_nodes);
+
+	hash_seq_init(&hash_seq, gid2tx);
+	while ((tx = hash_seq_search(&hash_seq)) != NULL)
+	{
+		MtmTxState	state = tx->state[Mtm->my_node_id - 1];
+		bool		found;
+
+		if (state == MtmTxPrepared)
+		{
+			FinishPreparedTransaction(tx->gid, false, true);
+			mtm_log(ResolverTxFinish, "TXFINISH: %s aborted", tx->gid);
+			hash_search(gid2tx, tx->gid, HASH_REMOVE, &found);
+			Assert(found);
+		}
+		else if (state == MtmTxPreCommited)
+		{
+			FinishPreparedTransaction(tx->gid, true, true);
+			mtm_log(ResolverTxFinish, "TXFINISH: %s committed", tx->gid);
+			hash_search(gid2tx, tx->gid, HASH_REMOVE, &found);
+			Assert(found);
+		}
+		else
+		{
+			Assert(false);
+		}
+	}
+
+	LWLockRelease(resolver_state->lock);
+}
+
 char *
 MtmTxStateMnem(MtmTxState state)
 {
diff --git a/src/state.c b/src/state.c
@@ -261,7 +261,7 @@ MtmCheckState(void)
 			break;
 
 		case MTM_RECOVERY:
-			if (mtm_state->recovered)
+			if (mtm_state->recovered || mtm_state->referee_grant)
 			{
 				MtmSetClusterStatus(MTM_RECOVERED);
 
@@ -279,7 +279,8 @@ MtmCheckState(void)
 		 * in MTM_RECOVERED state.
 		 */
 		case MTM_RECOVERED:
-			if (nReceivers == nEnabled && nSenders == nEnabled && nEnabled == nConnected - 1)
+			if ((nReceivers == nEnabled && nSenders == nEnabled && nEnabled == nConnected - 1)
+				|| mtm_state->referee_grant)
 			{
 				MtmSetClusterStatus(MTM_ONLINE);
 
@@ -605,8 +606,6 @@ MtmRefreshClusterStatus()
 	// MtmCheckState();
 	// MtmUnlock();
 
-	return;
-
 	/*
 	 * Check for referee decision when only half of nodes are visible.
 	 * Do not hold lock here, but recheck later wheter mask changed.
@@ -642,8 +641,9 @@ MtmRefreshClusterStatus()
 					if (popcount(mtm_state->connected_mask) == 1)
 					{
 						// MtmPollStatusOfPreparedTransactions(true);
-						ResolveAllTransactions(popcount(mtm_state->configured_mask));
+						ResolveForRefereeWinner(popcount(mtm_state->configured_mask));
 					}
+					mtm_state->recovered = true;
 					MtmEnableNode(Mtm->my_node_id);
 					MtmCheckState();
 				}
@@ -659,7 +659,7 @@ MtmRefreshClusterStatus()
 	 * can get refereeGrant before start of walsender, so it start in recovered mode.
 	 */
 	if (MtmRefereeConnStr && *MtmRefereeConnStr && mtm_state->referee_winner_id &&
-		popcount(mtm_state->connected_mask) == popcount(mtm_state->configured_mask) &&
+		popcount(mtm_state->enabled_mask) == popcount(mtm_state->configured_mask) &&
 		MtmGetCurrentStatus() == MTM_ONLINE) /* restrict this actions only to major -> online transition */
 	{
 		if (MtmRefereeClearWinner())
@@ -670,6 +670,8 @@ MtmRefreshClusterStatus()
 		}
 	}
 
+	return;
+
 	// Mtm->clique = (((nodemask_t)1 << Mtm->nAllNodes) - 1);
 	// return;
 
@@ -1143,7 +1145,7 @@ MtmMonitorStart(Oid db_id, Oid user_id)
 	MemSet(&worker, 0, sizeof(BackgroundWorker));
 	worker.bgw_flags = BGWORKER_SHMEM_ACCESS |	BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
-	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
+	worker.bgw_restart_time = 1;
 	worker.bgw_main_arg = Int32GetDatum(0);
 
 	memcpy(worker.bgw_extra, &db_id, sizeof(Oid));
@@ -1609,6 +1611,8 @@ MtmMonitor(Datum arg)
 		// XXX: add tx start/stop to clear mcxt?
 		check_status_requests(mtm_cfg);
 
+		MtmRefreshClusterStatus();
+
 		rc = WaitLatch(MyLatch,
 					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
 					   1000, PG_WAIT_EXTENSION);
diff --git a/tests/docker-entrypoint.sh b/tests/docker-entrypoint.sh
@@ -25,7 +25,7 @@ if [ "$1" = 'postgres' ]; then
 			max_worker_processes = 50
 			max_replication_slots = 10
 			max_wal_senders = 10
-			# log_statement = all
+			log_statement = all
 
 			shared_preload_libraries = 'multimaster'
 			multimaster.volkswagen_mode = on
@@ -71,8 +71,8 @@ if [ "$1" = 'postgres' ]; then
 		EOSQL
 		echo
 
-		psql -U `whoami` $POSTGRES_DB -c 'CREATE EXTENSION multimaster;';
-		psql -U `whoami` $POSTGRES_DB -c "select mtm.init_node($NODE_ID, '{$CONNSTRS}');"
+		# psql -U `whoami` $POSTGRES_DB -c 'CREATE EXTENSION multimaster;';
+		# psql -U `whoami` $POSTGRES_DB -c "select mtm.init_node($NODE_ID, '{$CONNSTRS}');"
 
 		pg_ctl -D "$PGDATA" -m fast -w stop
 	fi
diff --git a/tests/lib/bank_client.py b/tests/lib/bank_client.py
@@ -83,7 +83,11 @@ def __init__(self, dsns, n_accounts=100000):
 
         self.total = 0
         self.aggregates = [{} for e in dsns]
-        keep_trying(40, 1, self.initdb, 'self.initdb')
+        keep_trying(40, 1, self.create_extension, 'self.create_extension')
+        keep_trying(40, 1, self.await_nodes, 'self.await_nodes')
+
+        self.initdb()
+
         self.nodes_state_fields = ["id", "disabled", "disconnected", "catchUp", "slotLag",
             "avgTransDelay", "lastStatusChange", "oldestSnapshot", "SenderPid",
             "SenderStartTime ", "ReceiverPid", "ReceiverStartTime", "connStr"]
@@ -116,8 +120,6 @@ def __init__(self, dsns, n_accounts=100000):
     def initdb(self):
         conn = psycopg2.connect(self.dsns[0])
         cur = conn.cursor()
-        # cur.execute('create extension if not exists multimaster')
-        conn.commit()
         cur.execute('drop table if exists bank_test')
         cur.execute('create table bank_test(uid int primary key, amount int)')
         cur.execute('create table insert_test(id text primary key)')
@@ -138,16 +140,39 @@ def execute(self, node_id, statements):
         cur.close()
         con.close()
 
+    def await_nodes(self):
+        print("await_nodes")
+
+        for dsn in self.dsns:
+            con = psycopg2.connect(dsn)
+            con.autocommit = True
+            cur = con.cursor()
+            cur.execute('select 1')
+            cur.close()
+            con.close()
+
+
     def create_extension(self):
 
+        print("create extension")
+
         for dsn in self.dsns:
             con = psycopg2.connect(dsn)
             con.autocommit = True
             cur = con.cursor()
-            cur.execute('create extension multimaster;')
+            cur.execute('create extension if not exists multimaster')
             cur.close()
             con.close()
 
+        conn = psycopg2.connect(self.dsns[0])
+        cur = conn.cursor()
+        cur.execute("select mtm.init_cluster($$%s$$, $${%s}$$);" %
+            ("dbname=regression user=pg host=node1",
+            '"dbname=regression user=pg host=node2"'))
+        conn.commit()
+        cur.close()
+        conn.close()
+
     def is_data_identic(self):
         hashes = set()
         hashes2 = set()
@@ -324,13 +349,13 @@ def total_tx(self, conn, cur, agg, conn_i):
             agg.isolation += 1
             print(datetime.datetime.utcnow(), 'Isolation error, total ', self.total, ' -> ', total[0], ', node ', conn_i+1)
             self.total = total[0]
-            print(self.oops)
-            yield from cur.execute('select * from pg_prepared_xacts order by prepared;')
-            pxacts = yield from cur.fetchall()
-            for pxact in pxacts:
-                for i, col in enumerate(["transaction", "gid", "prepared", "owner", "database", "state3pc"]):
-                    print(pxact[i], end="\t")
-                print("\n")
+            # print(self.oops)
+            # yield from cur.execute('select * from pg_prepared_xacts order by prepared;')
+            # pxacts = yield from cur.fetchall()
+            # for pxact in pxacts:
+            #     for i, col in enumerate(["transaction", "gid", "prepared", "owner", "database", "state3pc"]):
+            #         print(pxact[i], end="\t")
+            #     print("\n")
 
     def run(self):
         # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
diff --git a/tests/test_referee.py b/tests/test_referee.py
@@ -179,6 +179,7 @@ def test_saved_referee_decision(self):
         print('###########################')
         docker_api.containers.get('node2').start()
         self.awaitOnline("dbname=regression user=postgres host=127.0.0.1 port=15433")
+        self.awaitOnline("dbname=regression user=postgres host=127.0.0.1 port=15432")
 
         self.client.bgrun()
         time.sleep(3)
@@ -316,6 +317,7 @@ def test_consequent_shutdown(self):
         self.client.stop()
         docker_api.containers.get('node2').start()
         self.awaitOnline("dbname=regression user=postgres host=127.0.0.1 port=15433")
+        self.awaitOnline("dbname=regression user=postgres host=127.0.0.1 port=15432")
         self.client.bgrun()
         time.sleep(3)
 
