MODULE_big = multimaster

OBJS = multimaster.o ../arbiter/lib/libarbiter.a ../arbiter/sockhub/libsockhub.a bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o

overrideCPPFLAGS += -I../arbiter/api -I../arbiter/sockhub

EXTENSION = multimaster

DATA = multimaster--1.0.sql

.PHONY: all

all: multimaster.o multimaster.so

../arbiter/sockhub/libsockhub.a:

make -C ../arbiter/sockhub

../arbiter/lib/libarbiter.a:

make -C ../arbiter

PG_CPPFLAGS = -I$(libpq_srcdir) -DUSE_PGLOGICAL_OUTPUT

SHLIB_LINK =$(libpq)

PG_CONFIG = pg_config

PGXS :=$(shell$(PG_CONFIG) --pgxs)

else

subdir = contrib/multimaster

top_builddir = ../..

include$(top_builddir)/src/Makefile.global

include$(top_srcdir)/contrib/contrib-global.mk

This repo implements distributed transaction manager using Snapshot Sharing mechanism. General concepts and alternative approaches described in postgres wikihttps://wiki.postgresql.org/wiki/DTM.

Backend-DTM protocol description can be found in[dtmd/README](dtmd/README).

* Patch postgres using xtm.patch. After that build and install postgres in usual way.

cd~/code/postgres

patch -p1<~/code/pg_dtm/xtm.patch

* Install pg_dtm extension.

export PATH=/path/to/pgsql/bin/:$PATH

cd~/code/pg_dtm

make&& make install

* Run dtmd.

cd~/code/pg_dtm/dtmd

make

mkdir /tmp/clog

./bin/dtmd&

* To run something meaningful you need at leat two postgres instances. Also pg_dtm requires presense in```shared_preload_libraries```.

initdb -D ./install/data1

initdb -D ./install/data2

echo"port = 5433">> ./install/data2/postgresql.conf

echo"shared_preload_libraries = 'pg_dtm'">> ./install/data1/postgresql.conf

echo"shared_preload_libraries = 'pg_dtm'">> ./install/data2/postgresql.conf

pg_ctl -D ./install/data1 -l ./install/data1/log start

pg_ctl -D ./install/data2 -l ./install/data2/log start

For a cluster-wide deploy we use ansible, more details in tests/deploy_layouts. (Ansible instructions will be later)

Now cluster is running and you can use global tx between two nodes. Let's connect to postgres instances at different ports:

create extension pg_dtm;-- node1

createtableaccounts(user_idint, amountint);-- node1

insert into accounts (select2*generate_series(1,100)-1,0);-- node1, odd user_id's

create extension pg_dtm;-- node2

createtableaccounts(user_idint, amountint);-- node2

insert into accounts (select2*generate_series(1,100),0);-- node2, even user_id's

select dtm_begin_transaction();-- node1, returns global xid, e.g. 42

select dtm_join_transaction(42);-- node2, join global tx

begin;-- node1

begin;-- node2

update accountsset amount=amount-100where user_id=1;-- node1, transfer money from user#1

update accountsset amount=amount+100where user_id=2;-- node2, to user#2

commit;-- node1, blocks until second commit happend

commit;-- node2

To ensure consistency we use simple bank test: perform a lot of simultaneous transfers between accounts on different servers, while constantly checking total amount of money on all accounts. This test can be found in tests/perf.

> go run ./tests/perf/*

-C value

Connection string (repeatfor multiple connections)

-a int

The number of bank accounts (default 100000)

-b string

Backend to use. Possible optinos: transfers, fdw, pgshard, readers. (default"transfers")

-gUse DTM to keep global consistency

-iInit database

-lUse'repeatable read' isolation level instead of'read committed'

-n int

The number updates each writer (readerincase of Reades backend) performs (default 10000)

-pUse parallel execs

-r int

The number of readers (default 1)

-s int

StartID. Script will update rows starting from this value

-vShow progress and other stufffor mortals

-w int

The number of writers (default 8)

go run ./tests/perf/*.go \

-C"dbname=postgres port=5432" \

-C"dbname=postgres port=5433" \

-g -i

go run ./tests/perf/*.go \

-C"dbname=postgres port=5432" \

-C"dbname=postgres port=5433" \

-g

We also provide a patch, that enables support of global transactions with postres_fdw. After patching and installing postres_fdw it is possible to run sametest via fdw usig key```-b fdw```.

pg_shard.use_dtm_transactions = 1

{

BgwPoolConstructorconstructor;

intid;

}BgwPoolExecutorCtx;

size_tn_snapshots;

size_tn_active;

staticvoidBgwPoolMainLoop(Datumarg)

{

BgwPoolExecutorCtx*ctx= (BgwPoolExecutorCtx*)arg;

intid=ctx->id;

BgwPool*pool=ctx->constructor();

intsize;

void*work;

BackgroundWorkerUnblockSignals();

BackgroundWorkerInitializeConnection(pool->dbname,NULL);

elog(WARNING,"Start background worker %d",id);

while(true) {

PGSemaphoreLock(&pool->available);

SpinLockAcquire(&pool->lock);

size=*(int*)&pool->queue[pool->head];

Assert(size<pool->size);

work=palloc(size);

pool->active-=1;

if (pool->head+size+4>pool->size) {

memcpy(work,pool->queue,size);

pool->head=INTALIGN(size);

}else {

memcpy(work,&pool->queue[pool->head+4],size);

pool->head+=4+INTALIGN(size);

}

if (pool->size==pool->head) {

pool->head=0;

}

if (pool->producerBlocked) {

pool->producerBlocked= false;

PGSemaphoreUnlock(&pool->overflow);

}

SpinLockRelease(&pool->lock);

pool->executor(id,work,size);

pfree(work);

}

}

voidBgwPoolInit(BgwPool*pool,BgwPoolExecutorexecutor,charconst*dbname,size_tqueueSize)

{

pool->queue= (char*)ShmemAlloc(queueSize);

pool->executor=executor;

PGSemaphoreCreate(&pool->available);

PGSemaphoreCreate(&pool->overflow);

PGSemaphoreReset(&pool->available);

PGSemaphoreReset(&pool->overflow);

SpinLockInit(&pool->lock);

pool->producerBlocked= false;

pool->head=0;

pool->tail=0;

pool->size=queueSize;

pool->active=0;

strcpy(pool->dbname,dbname);

}

voidBgwPoolStart(intnWorkers,BgwPoolConstructorconstructor)

{

inti;

BackgroundWorkerworker;

MemSet(&worker,0,sizeof(BackgroundWorker));

worker.bgw_flags=BGWORKER_SHMEM_ACCESS |BGWORKER_BACKEND_DATABASE_CONNECTION;

worker.bgw_start_time=BgWorkerStart_ConsistentState;

worker.bgw_main=BgwPoolMainLoop;

worker.bgw_restart_time=10;/* Wait 10 seconds for restart before crash */

for (i=0;i<nWorkers;i++) {

BgwPoolExecutorCtx*ctx= (BgwPoolExecutorCtx*)malloc(sizeof(BgwPoolExecutorCtx));

snprintf(worker.bgw_name,BGW_MAXLEN,"bgw_pool_worker_%d",i+1);

ctx->id=i;

ctx->constructor=constructor;

worker.bgw_main_arg= (Datum)ctx;

RegisterBackgroundWorker(&worker);

}

}

voidBgwPoolExecute(BgwPool*pool,void*work,size_tsize)

{

Assert(size+4 <=pool->size);

SpinLockAcquire(&pool->lock);

while (true) {

if ((pool->head <=pool->tail&&pool->size-pool->tail<size+4&&pool->head<size)

|| (pool->head>pool->tail&&pool->head-pool->tail<size+4))

{

pool->producerBlocked= true;

SpinLockRelease(&pool->lock);

PGSemaphoreLock(&pool->overflow);

SpinLockAcquire(&pool->lock);

}else {

pool->active+=1;

n_snapshots+=1;

n_active+=pool->active;

*(int*)&pool->queue[pool->tail]=size;

if (pool->size-pool->tail >=size+4) {

memcpy(&pool->queue[pool->tail+4],work,size);

pool->tail+=4+INTALIGN(size);

}else {

memcpy(pool->queue,work,size);

pool->tail=INTALIGN(size);

}

if (pool->tail==pool->size) {

pool->tail=0;

}

PGSemaphoreUnlock(&pool->available);

break;

}

}

SpinLockRelease(&pool->lock);

}

typedefvoid(*BgwPoolExecutor)(intid,void*work,size_tsize);

#defineMAX_DBNAME_LEN 30

{

BgwPoolExecutorexecutor;

volatileslock_tlock;

PGSemaphoreDataavailable;

PGSemaphoreDataoverflow;

size_thead;

size_ttail;

size_tsize;

size_tactive;

boolproducerBlocked;

chardbname[MAX_DBNAME_LEN];

char*queue;

}BgwPool;

typedefBgwPool*(*BgwPoolConstructor)(void);

externvoidBgwPoolStart(intnWorkers,BgwPoolConstructorconstructor);

externvoidBgwPoolInit(BgwPool*pool,BgwPoolExecutorexecutor,charconst*dbname,size_tqueueSize);

externvoidBgwPoolExecute(BgwPool*pool,void*work,size_tsize);

#defineINIT_BUF_SIZE 1024

voidByteBufferAlloc(ByteBuffer*buf)

{

buf->size=INIT_BUF_SIZE;

buf->data=palloc(buf->size);

buf->used=0;

}

voidByteBufferAppend(ByteBuffer*buf,void*data,intlen)

{

if (buf->used+len>buf->size) {

buf->size=buf->used+len>buf->size*2 ?buf->used+len :buf->size*2;

buf->data= (char*)repalloc(buf->data,buf->size);

}

memcpy(&buf->data[buf->used],data,len);

buf->used+=len;

}

voidByteBufferAppendInt32(ByteBuffer*buf,intdata)

{

ByteBufferAppend(buf,&data,sizeofdata);

}

voidByteBufferFree(ByteBuffer*buf)

{

pfree(buf->data);

}

voidByteBufferReset(ByteBuffer*buf)

{

buf->used=0;

}

{

char*data;

intsize;

intused;

}ByteBuffer;

externvoidByteBufferAlloc(ByteBuffer*buf);

externvoidByteBufferAppend(ByteBuffer*buf,void*data,intlen);

externvoidByteBufferAppendInt32(ByteBuffer*buf,intdata);

externvoidByteBufferFree(ByteBuffer*buf);

externvoidByteBufferReset(ByteBuffer*buf);