Skip to content

Commit a8b2483

Browse files
kelvichkelvich
committed
merge with pg_tstmd repo
2 parents e6885f3 + b507fd2 commit a8b2483

30 files changed

+1349
-151
lines changed

‎.gitignore

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,3 +38,4 @@ lib*.pc
3838
/Debug/
3939
/Release/
4040
/tmp_install/
41+

‎contrib/pg_tsdtm/Makefile

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,3 +14,10 @@ top_builddir = ../..
1414
include $(top_builddir)/src/Makefile.global
1515
include $(top_srcdir)/contrib/contrib-global.mk
1616
endif
17+
18+
# ifndef PG_CONFIG
19+
# PG_CONFIG = pg_config
20+
# endif
21+
22+
# PGXS := $(shell $(PG_CONFIG) --pgxs)
23+
# include $(PGXS)

‎contrib/pg_tsdtm/dtm_recovery/dtm_recovery.cpp

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
#include <iostream>
2+
#include <string>
3+
#include <vector>
4+
#include <set>
5+
6+
#include <pqxx/connection>
7+
#include <pqxx/transaction>
8+
#include <pqxx/nontransaction>
9+
10+
using namespace std;
11+
using namespace pqxx;
12+
13+
int main (int argc, char* argv[])
14+
{
15+
if (argc == 1){
16+
printf("Use -h to show usage options\n");
17+
return 1;
18+
}
19+
vector<string> connections;
20+
set<string> prepared_xacts;
21+
set<string> committed_xacts;
22+
bool verbose = false;
23+
for (int i = 1; i < argc; i++) {
24+
if (argv[i][0] == '-') {
25+
switch (argv[i][1]) {
26+
case 'C':
27+
case 'c':
28+
connections.push_back(string(argv[++i]));
29+
continue;
30+
case 'v':
31+
verbose = true;
32+
continue;
33+
}
34+
}
35+
printf("Perform recovery of pg_tsdtm cluster.\n"
36+
"Usage: dtm_recovery {options}\n"
37+
"Options:\n"
38+
"\t-c STR\tdatabase connection string\n"
39+
"\t-v\tverbose mode: print extra information while processing\n");
40+
return 1;
41+
}
42+
if (verbose) {
43+
cout << "Collecting information about prepared transactions...\n";
44+
}
45+
for (vector<string>::iterator ic = connections.begin(); ic != connections.end(); ++ic)
46+
{
47+
if (verbose) {
48+
cout << "Connecting to " << *ic << "...\n";
49+
}
50+
connection con(*ic);
51+
work txn(con);
52+
result r = txn.exec("select gid from pg_prepared_xacts");
53+
for (result::const_iterator it = r.begin(); it != r.end(); ++it)
54+
{
55+
string gid = it.at("gid").as(string());
56+
prepared_xacts.insert(gid);
57+
}
58+
txn.commit();
59+
}
60+
if (verbose) {
61+
cout << "Prepared transactions: ";
62+
for (set<string>::iterator it = prepared_xacts.begin(); it != prepared_xacts.end(); ++it)
63+
{
64+
cout << *it << ", ";
65+
}
66+
cout << "\nChecking which of them are committed...\n";
67+
}
68+
for (vector<string>::iterator ic = connections.begin(); ic != connections.end(); ++ic)
69+
{
70+
if (verbose) {
71+
cout << "Connecting to " << *ic << "...\n";
72+
}
73+
connection con(*ic);
74+
work txn(con);
75+
con.prepare("commit-check", "select * from pg_committed_xacts where gid=$1");
76+
for (set<string>::iterator it = prepared_xacts.begin(); it != prepared_xacts.end(); ++it)
77+
{
78+
string gid = *it;
79+
result r = txn.prepared("commit-check")(gid).exec();
80+
if (!r.empty()) {
81+
committed_xacts.insert(gid);
82+
}
83+
}
84+
txn.commit();
85+
}
86+
if (verbose) {
87+
cout << "Committed transactions: ";
88+
for (set<string>::iterator it = committed_xacts.begin(); it != committed_xacts.end(); ++it)
89+
{
90+
cout << *it << ", ";
91+
}
92+
cout << "\nCommitting them at all nodes...\n";
93+
}
94+
for (vector<string>::iterator ic = connections.begin(); ic != connections.end(); ++ic)
95+
{
96+
if (verbose) {
97+
cout << "Connecting to " << *ic << "...\n";
98+
}
99+
connection con(*ic);
100+
work txn(con);
101+
con.prepare("commit-check", "select * from pg_committed_xacts where gid=$1");
102+
con.prepare("commit-prepared", "commit prepared $1");
103+
con.prepare("rollback-prepared", "rollback prepared $1");
104+
result r = txn.exec("select gid from pg_prepared_xacts");
105+
for (result::const_iterator it = r.begin(); it != r.end(); ++it)
106+
{
107+
string gid = it.at("gid").as(string());
108+
result rc = txn.prepared("commit-check")(gid).exec();
109+
if (rc.empty()) {
110+
if (committed_xacts.find(gid) != committed_xacts.end()) {
111+
if (verbose) {
112+
cout << "Commit transaction " << gid << "\n";
113+
}
114+
txn.prepared("commit-prepared")(gid);
115+
} else {
116+
if (verbose) {
117+
cout << "Rollback transaction " << gid << "\n";
118+
}
119+
txn.prepared("rollback-prepared")(gid);
120+
}
121+
}
122+
}
123+
txn.commit();
124+
}
125+
if (verbose) {
126+
cout << "Recovery completed\n";
127+
}
128+
return 0;
129+
}

‎contrib/pg_tsdtm/dtm_recovery/makefile

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
CXX=g++
2+
CXXFLAGS=-g -Wall -O0 -pthread
3+
4+
all: dtm_recovery
5+
6+
dtm_recovery: dtm_recovery.cpp
7+
$(CXX) $(CXXFLAGS) -o dtm_recovery dtm_recovery.cpp -lpqxx
8+
9+
clean:
10+
rm -f dtm_recovery

‎contrib/pg_tsdtm/pg_dtm--1.0.sql

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,3 +20,7 @@ LANGUAGE C;
2020
CREATE FUNCTION dtm_end_prepare(gtid cstring, csn bigint) RETURNS void
2121
AS 'MODULE_PATHNAME','dtm_end_prepare'
2222
LANGUAGE C;
23+
24+
CREATE FUNCTION dtm_get_csn(xid integer) RETURNS bigint
25+
AS 'MODULE_PATHNAME','dtm_get_csn'
26+
LANGUAGE C;

0 commit comments

Comments
 (0)