Merge branch 'txtrace'

Author: kelvich

contrib/mmts/arbiter.c

@@ -867,6 +867,7 @@ static void MtmTransReceiver(Datum arg)
 					if (MtmIsCoordinator(ts)) {
 						switch (msg->code) { 
 						  case MSG_READY:
+							MTM_TXTRACE(ts, "MtmTransReceiver got MSG_READY");
 							if (ts->nVotes >= Mtm->nLiveNodes) {
 								MtmAbortTransaction(ts);
 								MtmWakeUpBackend(ts);
@@ -889,6 +890,7 @@ static void MtmTransReceiver(Datum arg)
 									} else if (MtmUseDtm) { 
 										Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
 										ts->nVotes = 1; /* I voted myself */
+										MTM_TXTRACE(ts, "MtmTransReceiver send MSG_PREPARE");
 										MtmSendNotificationMessage(ts, MSG_PREPARE);									  
 									} else { 
 										Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
@@ -908,6 +910,7 @@ static void MtmTransReceiver(Datum arg)
 							}
 							break;
 						  case MSG_PREPARED:
+							MTM_TXTRACE(ts, "MtmTransReceiver got MSG_PREPARED");
 							if (ts->nVotes >= Mtm->nLiveNodes) {					
 								MtmAbortTransaction(ts);
 								MtmWakeUpBackend(ts);

contrib/mmts/multimaster.c

@@ -708,7 +708,8 @@ MtmCreateTransState(MtmCurrentTrans* x)
 		/* I am coordinator of transaction */
 		ts->gtid.xid = x->xid;
 		ts->gtid.node = MtmNodeId;
-		ts->gid[0] = '\0';
+		//ts->gid[0] = '\0';
+		strcpy(ts->gid, x->gid);
 	}
 	return ts;
 }
@@ -724,6 +725,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 { 
 	MtmTransState* ts;
 	TransactionId* subxids;
+	MTM_TXTRACE(x, "PrePrepareTransaction Start");
 
 	if (!x->isDistributed) {
 		return;
@@ -782,7 +784,8 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	MtmAddSubtransactions(ts, subxids, ts->nSubxids);
 	MTM_LOG3("%d: MtmPrePrepareTransaction prepare commit of %d (gtid.xid=%d, gtid.node=%d, CSN=%ld)", 
 			 MyProcPid, x->xid, ts->gtid.xid, ts->gtid.node, ts->csn);
-	MtmUnlock(); 
+	MtmUnlock();
+	MTM_TXTRACE(x, "PrePrepareTransaction Finish");
 }
 
 /*
@@ -810,6 +813,7 @@ static void
 MtmPostPrepareTransaction(MtmCurrentTrans* x)
 { 
 	MtmTransState* ts;
+	MTM_TXTRACE(x, "PostPrepareTransaction Start");
 
 	if (!x->isDistributed) {
 		return;
@@ -849,7 +853,9 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		while (!ts->votingCompleted && Mtm->status == MTM_ONLINE && ts->status != TRANSACTION_STATUS_ABORTED && start + transTimeout >= MtmGetSystemTime()) 
 		{
 			MtmUnlock();
+			MTM_TXTRACE(x, "PostPrepareTransaction WaitLatch Start");
 			result = WaitLatch(&MyProc->procLatch, WL_LATCH_SET|WL_TIMEOUT, MtmHeartbeatRecvTimeout);
+			MTM_TXTRACE(x, "PostPrepareTransaction WaitLatch Finish");
 			if (result & WL_LATCH_SET) { 
 				ResetLatch(&MyProc->procLatch);			
 			} 
@@ -872,6 +878,8 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		Mtm->inject2PCError = 0;
 		elog(ERROR, "ERROR INJECTION for transaction %d (%s)", x->xid, x->gid);
 	}
+
+	MTM_TXTRACE(x, "PostPrepareTransaction Finish");
 }
 
 
@@ -1068,8 +1076,9 @@ csn_t MtmGetTransactionCSN(TransactionId xid)
 }
 
 void MtmWakeUpBackend(MtmTransState* ts)
-{																		
-	if (!ts->votingCompleted) { 
+{
+	if (!ts->votingCompleted) {
+		MTM_TXTRACE(ts, "MtmWakeUpBackend");
 		MTM_LOG3("Wakeup backed procno=%d, pid=%d", ts->procno, ProcGlobal->allProcs[ts->procno].pid);
 		ts->votingCompleted = true;
 		SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch); 

contrib/mmts/multimaster.h

@@ -32,6 +32,13 @@
 #define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__) 
 #endif
 
+#ifndef MTM_TRACE
+#define MTM_TXTRACE(tx, event)
+#else
+#define MTM_TXTRACE(tx, event) \
+		fprintf(stderr, "[MTM_TXTRACE], %s, %lld, %s\n", tx->gid, (long long)MtmGetSystemTime(), event)
+#endif
+
 #define MULTIMASTER_NAME                "multimaster"
 #define MULTIMASTER_SCHEMA_NAME         "mtm"
 #define MULTIMASTER_DDL_TABLE           "ddl_log"

contrib/mmts/pglogical_proto.c

@@ -178,6 +178,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     if (txn->xact_action != XLOG_XACT_COMMIT) { 
     	pq_sendstring(out, txn->gid);
 	}
+
+	MTM_TXTRACE(txn, "pglogical_write_commit Finish");
 }
 
 /*

contrib/mmts/tests/postgresql.conf.mm

@@ -61,7 +61,7 @@
 					# defaults to 'localhost'; use '*' for all
 					# (change requires restart)
 port = 5432				# (change requires restart)
-max_connections = 200			# (change requires restart)
+max_connections = 30			# (change requires restart)
 # Note:  Increasing max_connections costs ~400 bytes of shared memory per
 # connection slot, plus lock space (see max_locks_per_transaction).
 #superuser_reserved_connections = 3	# (change requires restart)
@@ -624,6 +624,27 @@
 
 # Add settings for extensions here
 
-multimaster.workers=8
-multimaster.queue_size=104857600 # 100mb
-multimaster.ignore_tables_without_pk=1
+#multimaster.workers=8
+#multimaster.queue_size=104857600 # 100mb
+#multimaster.ignore_tables_without_pk=1
+
+listen_addresses='*'
+max_prepared_transactions = 100
+synchronous_commit = off
+wal_level = logical
+max_worker_processes = 15
+max_replication_slots = 10
+max_wal_senders = 10
+shared_preload_libraries = 'raftable,multimaster'
+default_transaction_isolation = 'repeatable read'
+log_checkpoints = on
+log_autovacuum_min_duration = 0
+
+multimaster.workers = 4
+multimaster.use_raftable = true
+multimaster.queue_size=52857600
+multimaster.ignore_tables_without_pk = 1
+multimaster.heartbeat_recv_timeout = 1000
+multimaster.heartbeat_send_timeout = 250
+multimaster.twopc_min_timeout = 40000
+

contrib/mmts/tests/reinit-mm.sh

@@ -5,20 +5,21 @@ pkill -9 postgres
 pkill -9 arbiter
 
 cd ~/code/postgres_cluster/contrib/mmts/
-make install
+make clean && make install
 cd ~/code/postgres_cluster/contrib/raftable/
-make install
+make clean && make install
 cd ~/code/postgres_cluster/contrib/mmts/tests
 
 
 rm -fr node? *.log dtm
-mkdir dtm
 conn_str=""
 sep=""
 for ((i=1;i<=n_nodes;i++))
 do    
-    port=$((5431+i))
+    port=$((5431 + i))
+	raft_port=$((6665 + i))
     conn_str="$conn_str${sep}dbname=postgres host=localhost port=$port sslmode=disable"
+	raft_conn_str="$raft_conn_str${sep}${i}:localhost:$raft_port"
     sep=","
     initdb node$i
 done
@@ -35,6 +36,9 @@ do
     sed "s/5432/$port/g" < postgresql.conf.mm > node$i/postgresql.conf
     echo "multimaster.conn_strings = '$conn_str'" >> node$i/postgresql.conf
     echo "multimaster.node_id = $i" >> node$i/postgresql.conf
+	echo "raftable.id = $i" >> node$i/postgresql.conf
+	echo "raftable.peers = '$raft_conn_str'" >> node$i/postgresql.conf
+
     cp pg_hba.conf node$i
     pg_ctl -w -D node$i -l node$i.log start
 done

contrib/mmts/tests2/docker-entrypoint.sh

@@ -88,6 +88,7 @@ if [ "$1" = 'postgres' ]; then
 			multimaster.conn_strings = '$CONNSTRS'
 			multimaster.heartbeat_recv_timeout = 1000
 			multimaster.heartbeat_send_timeout = 250
+			multimaster.twopc_min_timeout = 40000
 		EOF
 
 		tail -n 20 $PGDATA/postgresql.conf

contrib/mmts/tests2/provision.yml

@@ -9,7 +9,7 @@
     pg_dst: "{{pg_prefix}}/install"
     pg_datadir: "{{pg_prefix}}/data_{{pg_port}}"
     pg_repo: https://github.com/postgrespro/postgres_cluster.git
-    pg_version_tag: master
+    pg_version_tag: txtrace
     pg_destroy_and_init: true
     makejobs: 4
 
@@ -127,16 +127,17 @@
     - "max_worker_processes = 15"
     - "max_replication_slots = 10"
     - "max_wal_senders = 10"
+    - "max_worker_processes = 30"
     - "log_checkpoints = on"
     - "log_autovacuum_min_duration = 0"
     - "shared_preload_libraries = 'raftable,multimaster'"
     - "default_transaction_isolation = 'repeatable read'"
     - "raftable.id = {{ node_id }}"
     - "raftable.peers = '{{ raft_connstr }}'"
-    - "multimaster.workers = 4"
+    - "multimaster.workers = 20"
     - "multimaster.arbiter_port = {{ 7777 }}"
     - "multimaster.use_raftable = true"
-    - "multimaster.queue_size=52857600"
+    - "multimaster.queue_size=528576000"
     - "multimaster.ignore_tables_without_pk = 1"
     - "multimaster.node_id = {{ node_id }}"
     - "multimaster.conn_strings = '{{ mm_connstr }}'"

src/backend/replication/logical/decode.c

@@ -24,6 +24,9 @@
  *
  * -------------------------------------------------------------------------
  */
+#include <sys/time.h>
+#include <time.h>
+
 #include "postgres.h"
 
 #include "access/heapam.h"
@@ -643,6 +646,15 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	int			i;
 	TransactionId xid = parsed->twophase_xid;
 
+	struct timeval tv;
+	int64 ts;
+
+	gettimeofday(&tv, NULL);
+	ts = (int64)tv.tv_sec*USECS_PER_SEC + tv.tv_usec;
+
+	fprintf(stderr, "[MTM_TXTRACE], %s, %lld, DecodePrepare Started\n",
+									parsed->twophase_gid, (long long) ts);
+
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
 		origin_lsn = parsed->origin_lsn;