Skip to content

Commit 1c7fe33

Browse files
kvapkvap
committed
Implement traditional postgres snapshots in DTM.
1 parent cc7cb1a commit 1c7fe33

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

47 files changed

+7936
-0
lines changed

‎contrib/pg_xtm/Makefile

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
MODULE_big = pg_dtm
2+
OBJS = pg_dtm.o libdtm.o
3+
4+
EXTENSION = pg_dtm
5+
DATA = pg_dtm--1.0.sql
6+
7+
ifdef USE_PGXS
8+
PG_CONFIG = pg_config
9+
PGXS := $(shell $(PG_CONFIG) --pgxs)
10+
include $(PGXS)
11+
else
12+
subdir = contrib/pg_dtm
13+
top_builddir = ../..
14+
include $(top_builddir)/src/Makefile.global
15+
include $(top_srcdir)/contrib/contrib-global.mk
16+
endif

‎contrib/pg_xtm/README

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
===
2+
dtm
3+
===
4+
5+
Distributed transaction management tools for PostgreSQL.
6+
7+
--------------------
8+
Communication scheme
9+
--------------------
10+
11+
.- Backend -.
12+
/ \
13+
/ \
14+
DTM ---- Backend ---- Coordinator
15+
\ /
16+
\ /
17+
`- Backend -´
18+
19+
20+
-----------------------
21+
Coordinator-Backend API
22+
-----------------------
23+
24+
This API includes a set of postgres procedures that
25+
the coordinator can call with "select" statement.
26+
27+
FIXME: document the API
28+
29+
----------
30+
libdtm api
31+
----------
32+
33+
typedef unsigned long long xid_t;
34+
35+
// Connects to the specified DTM.
36+
DTMConn DtmConnect(char *host, int port);
37+
38+
// Disconnects from the DTM. Do not use the 'dtm' pointer after this call, or
39+
// bad things will happen.
40+
void DtmDisconnect(DTMConn dtm);
41+
42+
// Asks DTM for a fresh snapshot. Returns a snapshot on success, or NULL
43+
// otherwise. Please free the snapshot memory yourself after use.
44+
Snapshot DtmGlobalGetSnapshot(DTMConn dtm);
45+
46+
// Starts a transaction. Returns the 'gxid' on success, or INVALID_GXID otherwise.
47+
xid_t DtmGlobalBegin(DTMConn dtm);
48+
49+
// Marks a given transaction as 'committed'. Returns 'true' on success,
50+
// 'false' otherwise.
51+
bool DtmGlobalCommit(DTMConn dtm, xid_t gxid);
52+
53+
// Marks a given transaction as 'aborted'.
54+
void DtmGlobalRollback(DTMConn dtm, xid_t gxid);
55+
56+
// Gets the status of the transaction identified by 'gxid'. Returns the status
57+
// on success, or -1 otherwise.
58+
int DtmGlobalGetTransStatus(DTMConn dtm, xid_t gxid);
59+
60+
--------------------
61+
Backend-DTM Protocol
62+
--------------------
63+
64+
DTM <--> Backend:
65+
66+
<- 'b'<hex16 self> - "begin"
67+
-> '+'<hex16 gxid> - "transaction started"
68+
-> '-' - "something went wrong"
69+
70+
<- 'c'<hex16 gxid> - "commit"
71+
-> '+' - "commit saved"
72+
-> '-' - "something went wrong"
73+
74+
<- 'a'<hex16 gxid> - "abort"
75+
-> '+' - "abort saved"
76+
-> '-' - "something went wrong"
77+
78+
<- 'h' - "snapshot"
79+
-> '+'<snapshot> - "here is a fresh snapshot for you"
80+
-> '-' - "something went wrong"
81+
82+
<- 's'<hex16 gxid> - "status"
83+
-> '+''c|a|?' - "here is the transaction status"
84+
(c)ommitted, (a)borted or (?)unknown
85+
-> '-' - "something went wrong"
86+
87+
<snapshot> = <hex16 xmin><hex16 xmax><hex16 n><hex16 active[n]>
88+
89+
Backend disconnection is considered as an abort of all incomplete transactions
90+
started by that backend.

‎contrib/pg_xtm/dtmd/Makefile

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
CC=gcc
2+
CFLAGS=-g -Wall -Iinclude -D_LARGEFILE64_SOURCE
3+
LIBUV_PREFIX=$(HOME)/libuv-build
4+
LIBUV_CFLAGS=-I"$(LIBUV_PREFIX)/include" -L"$(LIBUV_PREFIX)/lib"
5+
LIBUV_LDFLAGS=-luv -pthread
6+
7+
SYSTEM=$(shell uname -s)
8+
ifeq ($(SYSTEM),Darwin)
9+
CFLAGS += -D_DARWIN_C_SOURCE
10+
endif
11+
12+
.PHONY: all clean check bindir objdir
13+
14+
all: bin/dtmd
15+
@echo Done.
16+
@echo Feel free to run the tests with \'make check\'.
17+
18+
bin/dtmd: obj/intset.o obj/eventwrap.o obj/main.o obj/parser.o obj/clog.o obj/clogfile.o obj/util.o | bindir objdir
19+
$(CC) -o bin/dtmd $(CFLAGS) $(LIBUV_CFLAGS) \
20+
obj/intset.o obj/eventwrap.o obj/main.o obj/parser.o \
21+
obj/clog.o obj/clogfile.o obj/util.o \
22+
$(LIBUV_LDFLAGS)
23+
24+
obj/eventwrap.o: src/eventwrap.c | objdir
25+
$(CC) -c -o obj/eventwrap.o $(CFLAGS) $(LIBUV_CFLAGS) src/eventwrap.c
26+
27+
check: bin/intset-test bin/util-test bin/clog-test bin/parser-test
28+
./check.sh intset util parser clog
29+
30+
obj/%.o: src/%.c | objdir
31+
$(CC) $(CFLAGS) -c -o $@ $<
32+
33+
bin/intset-test: obj/intset-test.o obj/intset.o | bindir
34+
$(CC) -o bin/intset-test $(CFLAGS) obj/intset-test.o obj/intset.o
35+
36+
bin/util-test: obj/util-test.o obj/util.o | bindir
37+
$(CC) -o bin/util-test $(CFLAGS) obj/util-test.o obj/util.o
38+
39+
bin/clog-test: obj/clog-test.o obj/clog.o obj/clogfile.o obj/util.o | bindir
40+
$(CC) -o bin/clog-test $(CFLAGS) obj/clog-test.o obj/clog.o obj/clogfile.o obj/util.o
41+
42+
bin/parser-test: obj/parser-test.o obj/parser.o | bindir
43+
$(CC) -o bin/parser-test $(CFLAGS) obj/parser-test.o obj/parser.o
44+
45+
bindir:
46+
mkdir -p bin
47+
48+
objdir:
49+
mkdir -p obj
50+
51+
clean:
52+
rm -rfv bin obj test.log

‎contrib/pg_xtm/dtmd/check.sh

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
#!/bin/sh
2+
rm -rf test.log
3+
failed=0
4+
passed=0
5+
for testname in $@; do
6+
if bin/${testname}-test >> test.log 2>&1; then
7+
echo "${testname} ok"
8+
passed=$((passed + 1))
9+
else
10+
echo "${testname} FAILED"
11+
failed=$((failed + 1))
12+
fi
13+
done
14+
15+
echo "tests passed: $passed"
16+
echo "tests failed: $failed"
17+
if [ $failed -eq 0 ]; then
18+
rm -rf test.log
19+
else
20+
echo "see test.log for details"
21+
exit 1
22+
fi

‎contrib/pg_xtm/dtmd/include/clog.h

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
/*
2+
* This module provides a high-level API to access clog files.
3+
*/
4+
5+
#ifndef CLOG_H
6+
#define CLOG_H
7+
8+
#include <stdbool.h>
9+
#include "int.h"
10+
11+
#define INVALID_GXID 0
12+
#define MIN_GXID 42
13+
#define MAX_GXID 0xdeadbeefcafebabe
14+
15+
#define COMMIT_UNKNOWN 0
16+
#define COMMIT_YES 1
17+
#define COMMIT_NO 2
18+
19+
typedef struct clog_data_t *clog_t;
20+
21+
// Open the clog at the specified path. Try not to open the same datadir twice
22+
// or in two different processes. Return a clog object on success, NULL
23+
// otherwise.
24+
clog_t clog_open(char *datadir);
25+
26+
// Get the status of the specified global commit.
27+
int clog_read(clog_t clog, xid_t gxid);
28+
29+
// Set the status of the specified global commit. Return 'true' on success,
30+
// 'false' otherwise.
31+
bool clog_write(clog_t clog, xid_t gxid, int status);
32+
33+
// Allocate a fresh unused gxid. Return INVALID_GXID on error.
34+
xid_t clog_advance(clog_t clog);
35+
36+
// Get the first unknown commit id (used as a snapshot). Return INVALID_GXID on
37+
// error.
38+
xid_t clog_horizon(clog_t clog);
39+
40+
// Forget about the commits before the given one ('until'), and free the
41+
// occupied space if possible. Return 'true' on success, 'false' otherwise.
42+
bool clog_forget(clog_t clog, xid_t until);
43+
44+
// Close the specified clog. Do not use the clog object after closing. Return
45+
// 'true' on success, 'false' otherwise.
46+
bool clog_close(clog_t clog);
47+
48+
#endif

‎contrib/pg_xtm/dtmd/include/clogfile.h

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
/*
2+
* This module provides a low-level API to access clog files.
3+
*/
4+
5+
#include <stdbool.h>
6+
#include "int.h"
7+
8+
#ifndef CLOGFILE_H
9+
#define CLOGFILE_H
10+
11+
#define BITS_PER_COMMIT 2
12+
#define COMMIT_MASK ((1 << BITS_PER_COMMIT) - 1)
13+
#define COMMITS_PER_BYTE 4
14+
#define COMMITS_PER_FILE 1024 // 0x100000000
15+
#define BYTES_PER_FILE ((COMMITS_PER_FILE) / (COMMITS_PER_BYTE))
16+
#define GXID_TO_FILEID(GXID) ((GXID) / (COMMITS_PER_FILE))
17+
#define GXID_TO_OFFSET(GXID) (((GXID) % (COMMITS_PER_FILE)) / (COMMITS_PER_BYTE))
18+
#define GXID_TO_SUBOFFSET(GXID) (((GXID) % (COMMITS_PER_FILE)) % (COMMITS_PER_BYTE))
19+
20+
typedef struct clogfile_t {
21+
char *path;
22+
xid_t min;
23+
xid_t max;
24+
void *data; // ptr for mmap
25+
} clogfile_t;
26+
27+
// Open a clog file with the gived id. Create before opening if 'create' is
28+
// true. Return 'true' on success, 'false' otherwise.
29+
bool clogfile_open_by_id(clogfile_t *clogfile, char *datadir, int fileid, bool create);
30+
31+
// Close and remove the given clog file. Return 'true' on success, 'false'
32+
// otherwise.
33+
bool clogfile_remove(clogfile_t *clogfile);
34+
35+
// Close the specified clogfile. Return 'true' on success, 'false' otherwise.
36+
bool clogfile_close(clogfile_t *clogfile);
37+
38+
// Get the status of the specified global commit from the clog file.
39+
int clogfile_get_status(clogfile_t *clogfile, xid_t gxid);
40+
41+
// Set the status of the specified global commit in the clog file. Return
42+
// 'true' on success, 'false' otherwise.
43+
bool clogfile_set_status(clogfile_t *clogfile, xid_t gxid, int status);
44+
45+
#endif

‎contrib/pg_xtm/dtmd/include/eventwrap.h

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
/*
2+
* This module is used as a layer of abstraction between the main logic and the
3+
* event library. This should, theoretically, allow us to switch to another
4+
* library with less effort.
5+
*/
6+
7+
#ifndef EVENTWRAP_H
8+
#define EVENTWRAP_H
9+
10+
int eventwrap(
11+
const char *host,
12+
int port,
13+
char *(*ondata)(void *client, size_t len, char *data),
14+
void (*onconnect)(void **client),
15+
void (*ondisconnect)(void *client)
16+
);
17+
18+
#endif

‎contrib/pg_xtm/dtmd/include/int.h

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
#ifndef INT_H
2+
#define INT_H
3+
4+
typedef unsigned long long xid_t;
5+
6+
#endif

‎contrib/pg_xtm/dtmd/include/intset.h

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
#ifndef INTSET_H
2+
#define INTSET_H
3+
4+
#include "int.h"
5+
6+
// TODO: use tree structures instead of this uglyness
7+
8+
typedef struct intset_t {
9+
xid_t *data;
10+
int shift;
11+
int size;
12+
int capacity;
13+
} intset_t;
14+
15+
// Constructor and destructor
16+
intset_t *intset_create(int capacity);
17+
void intset_destroy(intset_t *this);
18+
19+
// Appends the value at the end. The value should be greater than all the
20+
// previously added values.
21+
void intset_add(intset_t *this, xid_t value);
22+
23+
// Returns the value of the i-th element.
24+
xid_t intset_get(intset_t *this, int i);
25+
26+
// Removes the value, shifting everything else to the beginning.
27+
void intset_remove(intset_t *this, xid_t value);
28+
29+
// Returns the size of the intset.
30+
int intset_size(intset_t *this);
31+
32+
#endif

‎contrib/pg_xtm/dtmd/include/parser.h

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
#ifndef PARSER_H
2+
#define PARSER_H
3+
4+
#include <stdbool.h>
5+
6+
#include "int.h"
7+
8+
#define CMD_BEGIN 'b'
9+
#define CMD_COMMIT 'c'
10+
#define CMD_ABORT 'a'
11+
#define CMD_SNAPSHOT 'h'
12+
#define CMD_STATUS 's'
13+
14+
typedef struct cmd_t {
15+
char cmd;
16+
xid_t arg;
17+
} cmd_t;
18+
19+
// Do not rely on the inner structure, it may change tomorrow.
20+
typedef struct parser_data_t *parser_t;
21+
22+
// Allocate and initialize a parser.
23+
parser_t parser_create();
24+
25+
// Destroy the parser. The 'p' handle becomes invalid, so do not refer to it
26+
// after destroying the parser.
27+
void parser_destroy(parser_t p);
28+
29+
// Initialize the parser.
30+
void parser_init(parser_t p);
31+
32+
// Check if parser has failed.
33+
bool parser_failed(parser_t p);
34+
35+
// Get the error message for the parser.
36+
char *parser_errormsg(parser_t p);
37+
38+
// Feeds a character to the parser, and returns a parsed command if it is
39+
// complete. Returns NULL if command is not complete. The caller should check
40+
// for errors, please use parser_failed() method for that. Also the caller
41+
// should free the cmd after use.
42+
cmd_t *parser_feed(parser_t p, char c);
43+
44+
#endif

0 commit comments

Comments
 (0)