Skip to content

Commit bc92057

Browse files
knizhnikknizhnik
committed
2 parents 13eb740 + 6c492b4 commit bc92057

File tree

10 files changed

+76
-16
lines changed

10 files changed

+76
-16
lines changed

‎contrib/mmts/arbiter.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -867,6 +867,7 @@ static void MtmTransReceiver(Datum arg)
867867
if (MtmIsCoordinator(ts)) {
868868
switch (msg->code) {
869869
case MSG_READY:
870+
MTM_TXTRACE(ts, "MtmTransReceiver got MSG_READY");
870871
if (ts->nVotes >= Mtm->nLiveNodes) {
871872
MtmAbortTransaction(ts);
872873
MtmWakeUpBackend(ts);
@@ -889,6 +890,7 @@ static void MtmTransReceiver(Datum arg)
889890
} else if (MtmUseDtm) {
890891
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
891892
ts->nVotes = 1; /* I voted myself */
893+
MTM_TXTRACE(ts, "MtmTransReceiver send MSG_PREPARE");
892894
MtmSendNotificationMessage(ts, MSG_PREPARE);
893895
} else {
894896
Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
@@ -908,6 +910,7 @@ static void MtmTransReceiver(Datum arg)
908910
}
909911
break;
910912
case MSG_PREPARED:
913+
MTM_TXTRACE(ts, "MtmTransReceiver got MSG_PREPARED");
911914
if (ts->nVotes >= Mtm->nLiveNodes) {
912915
MtmAbortTransaction(ts);
913916
MtmWakeUpBackend(ts);

‎contrib/mmts/multimaster.c

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -717,7 +717,8 @@ MtmCreateTransState(MtmCurrentTrans* x)
717717
/* I am coordinator of transaction */
718718
ts->gtid.xid = x->xid;
719719
ts->gtid.node = MtmNodeId;
720-
ts->gid[0] = '\0';
720+
//ts->gid[0] = '\0';
721+
strcpy(ts->gid, x->gid);
721722
}
722723
return ts;
723724
}
@@ -733,6 +734,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
733734
{
734735
MtmTransState* ts;
735736
TransactionId* subxids;
737+
MTM_TXTRACE(x, "PrePrepareTransaction Start");
736738

737739
if (!x->isDistributed) {
738740
return;
@@ -791,7 +793,8 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
791793
MtmAddSubtransactions(ts, subxids, ts->nSubxids);
792794
MTM_LOG3("%d: MtmPrePrepareTransaction prepare commit of %d (gtid.xid=%d, gtid.node=%d, CSN=%ld)",
793795
MyProcPid, x->xid, ts->gtid.xid, ts->gtid.node, ts->csn);
794-
MtmUnlock();
796+
MtmUnlock();
797+
MTM_TXTRACE(x, "PrePrepareTransaction Finish");
795798
}
796799

797800
/*
@@ -819,6 +822,7 @@ static void
819822
MtmPostPrepareTransaction(MtmCurrentTrans* x)
820823
{
821824
MtmTransState* ts;
825+
MTM_TXTRACE(x, "PostPrepareTransaction Start");
822826

823827
if (!x->isDistributed) {
824828
return;
@@ -861,7 +865,9 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
861865
while (!ts->votingCompleted && Mtm->status == MTM_ONLINE && ts->status != TRANSACTION_STATUS_ABORTED && start + transTimeout >= MtmGetSystemTime())
862866
{
863867
MtmUnlock();
868+
MTM_TXTRACE(x, "PostPrepareTransaction WaitLatch Start");
864869
result = WaitLatch(&MyProc->procLatch, WL_LATCH_SET|WL_TIMEOUT, MtmHeartbeatRecvTimeout);
870+
MTM_TXTRACE(x, "PostPrepareTransaction WaitLatch Finish");
865871
if (result & WL_LATCH_SET) {
866872
ResetLatch(&MyProc->procLatch);
867873
}
@@ -884,6 +890,8 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
884890
Mtm->inject2PCError = 0;
885891
elog(ERROR, "ERROR INJECTION for transaction %d (%s)", x->xid, x->gid);
886892
}
893+
894+
MTM_TXTRACE(x, "PostPrepareTransaction Finish");
887895
}
888896

889897

@@ -1084,8 +1092,9 @@ csn_t MtmGetTransactionCSN(TransactionId xid)
10841092
}
10851093

10861094
void MtmWakeUpBackend(MtmTransState* ts)
1087-
{
1088-
if (!ts->votingCompleted) {
1095+
{
1096+
if (!ts->votingCompleted) {
1097+
MTM_TXTRACE(ts, "MtmWakeUpBackend");
10891098
MTM_LOG3("Wakeup backed procno=%d, pid=%d", ts->procno, ProcGlobal->allProcs[ts->procno].pid);
10901099
ts->votingCompleted = true;
10911100
SetLatch(&ProcGlobal->allProcs[ts->procno].procLatch);

‎contrib/mmts/multimaster.h

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,13 @@
3232
#define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
3333
#endif
3434

35+
#ifndef MTM_TRACE
36+
#define MTM_TXTRACE(tx, event)
37+
#else
38+
#define MTM_TXTRACE(tx, event) \
39+
fprintf(stderr, "[MTM_TXTRACE], %s, %lld, %s\n", tx->gid, (long long)MtmGetSystemTime(), event)
40+
#endif
41+
3542
#define MULTIMASTER_NAME "multimaster"
3643
#define MULTIMASTER_SCHEMA_NAME "mtm"
3744
#define MULTIMASTER_DDL_TABLE "ddl_log"

‎contrib/mmts/pglogical_proto.c

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -178,6 +178,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
178178
if (txn->xact_action != XLOG_XACT_COMMIT) {
179179
pq_sendstring(out, txn->gid);
180180
}
181+
182+
MTM_TXTRACE(txn, "pglogical_write_commit Finish");
181183
}
182184

183185
/*

‎contrib/mmts/tests/postgresql.conf.mm

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@
6161
# defaults to 'localhost'; use '*' for all
6262
# (change requires restart)
6363
port = 5432 # (change requires restart)
64-
max_connections = 200 # (change requires restart)
64+
max_connections = 30 # (change requires restart)
6565
# Note: Increasing max_connections costs ~400 bytes of shared memory per
6666
# connection slot, plus lock space (see max_locks_per_transaction).
6767
#superuser_reserved_connections = 3 # (change requires restart)
@@ -624,6 +624,27 @@
624624

625625
# Add settings for extensions here
626626

627-
multimaster.workers=8
628-
multimaster.queue_size=104857600 # 100mb
629-
multimaster.ignore_tables_without_pk=1
627+
#multimaster.workers=8
628+
#multimaster.queue_size=104857600 # 100mb
629+
#multimaster.ignore_tables_without_pk=1
630+
631+
listen_addresses='*'
632+
max_prepared_transactions = 100
633+
synchronous_commit = off
634+
wal_level = logical
635+
max_worker_processes = 15
636+
max_replication_slots = 10
637+
max_wal_senders = 10
638+
shared_preload_libraries = 'raftable,multimaster'
639+
default_transaction_isolation = 'repeatable read'
640+
log_checkpoints = on
641+
log_autovacuum_min_duration = 0
642+
643+
multimaster.workers = 4
644+
multimaster.use_raftable = true
645+
multimaster.queue_size=52857600
646+
multimaster.ignore_tables_without_pk = 1
647+
multimaster.heartbeat_recv_timeout = 1000
648+
multimaster.heartbeat_send_timeout = 250
649+
multimaster.twopc_min_timeout = 40000
650+

‎contrib/mmts/tests/reinit-mm.sh

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,20 +5,21 @@ pkill -9 postgres
55
pkill -9 arbiter
66

77
cd ~/code/postgres_cluster/contrib/mmts/
8-
make install
8+
make clean && make install
99
cd ~/code/postgres_cluster/contrib/raftable/
10-
make install
10+
make clean && make install
1111
cd ~/code/postgres_cluster/contrib/mmts/tests
1212

1313

1414
rm -fr node? *.log dtm
15-
mkdir dtm
1615
conn_str=""
1716
sep=""
1817
for ((i=1;i<=n_nodes;i++))
1918
do
20-
port=$((5431+i))
19+
port=$((5431 + i))
20+
raft_port=$((6665 + i))
2121
conn_str="$conn_str${sep}dbname=postgres host=localhost port=$port sslmode=disable"
22+
raft_conn_str="$raft_conn_str${sep}${i}:localhost:$raft_port"
2223
sep=","
2324
initdb node$i
2425
done
@@ -35,6 +36,9 @@ do
3536
sed "s/5432/$port/g" < postgresql.conf.mm > node$i/postgresql.conf
3637
echo "multimaster.conn_strings = '$conn_str'" >> node$i/postgresql.conf
3738
echo "multimaster.node_id = $i" >> node$i/postgresql.conf
39+
echo "raftable.id = $i" >> node$i/postgresql.conf
40+
echo "raftable.peers = '$raft_conn_str'" >> node$i/postgresql.conf
41+
3842
cp pg_hba.conf node$i
3943
pg_ctl -w -D node$i -l node$i.log start
4044
done

‎contrib/mmts/tests2/docker-entrypoint.sh

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ if [ "$1" = 'postgres' ]; then
8888
multimaster.conn_strings = '$CONNSTRS'
8989
multimaster.heartbeat_recv_timeout = 1000
9090
multimaster.heartbeat_send_timeout = 250
91+
multimaster.twopc_min_timeout = 40000
9192
EOF
9293

9394
tail -n 20 $PGDATA/postgresql.conf

‎contrib/mmts/tests2/provision.yml

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
pg_dst: "{{pg_prefix}}/install"
1010
pg_datadir: "{{pg_prefix}}/data_{{pg_port}}"
1111
pg_repo: https://github.com/postgrespro/postgres_cluster.git
12-
pg_version_tag: master
12+
pg_version_tag: txtrace
1313
pg_destroy_and_init: true
1414
makejobs: 4
1515

@@ -127,16 +127,17 @@
127127
- "max_worker_processes = 15"
128128
- "max_replication_slots = 10"
129129
- "max_wal_senders = 10"
130+
- "max_worker_processes = 30"
130131
- "log_checkpoints = on"
131132
- "log_autovacuum_min_duration = 0"
132133
- "shared_preload_libraries = 'raftable,multimaster'"
133134
- "default_transaction_isolation = 'repeatable read'"
134135
- "raftable.id = {{ node_id }}"
135136
- "raftable.peers = '{{ raft_connstr }}'"
136-
- "multimaster.workers = 4"
137+
- "multimaster.workers = 20"
137138
- "multimaster.arbiter_port = {{ 7777 }}"
138139
- "multimaster.use_raftable = true"
139-
- "multimaster.queue_size=52857600"
140+
- "multimaster.queue_size=528576000"
140141
- "multimaster.ignore_tables_without_pk = 1"
141142
- "multimaster.node_id = {{ node_id }}"
142143
- "multimaster.conn_strings = '{{ mm_connstr }}'"

‎contrib/raftable/worker.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -496,7 +496,7 @@ static void worker_main(Datum arg)
496496

497497
void worker_register(WorkerConfig *cfg)
498498
{
499-
BackgroundWorker worker;
499+
BackgroundWorker worker = {{0}};
500500
strcpy(worker.bgw_name, "raftable worker");
501501
worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
502502
worker.bgw_start_time = BgWorkerStart_ConsistentState;

‎src/backend/replication/logical/decode.c

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,9 @@
2424
*
2525
* -------------------------------------------------------------------------
2626
*/
27+
#include <sys/time.h>
28+
#include <time.h>
29+
2730
#include "postgres.h"
2831

2932
#include "access/heapam.h"
@@ -643,6 +646,15 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
643646
int i;
644647
TransactionId xid = parsed->twophase_xid;
645648

649+
// struct timeval tv;
650+
// int64 ts;
651+
652+
// gettimeofday(&tv, NULL);
653+
// ts = (int64)tv.tv_sec*USECS_PER_SEC + tv.tv_usec;
654+
655+
// fprintf(stderr, "[MTM_TXTRACE], %s, %lld, DecodePrepare Started\n",
656+
// parsed->twophase_gid, (long long) ts);
657+
646658
if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
647659
{
648660
origin_lsn = parsed->origin_lsn;

0 commit comments

Comments
 (0)