foreachmy$i (1..$nodenum)

{

my$host ="127.0.0.1";

my ($pgport,$raftport) = allocate_ports($host, 2);

my ($pgport,$arbiter_port) = allocate_ports($host, 2);

my$node = new PostgresNode("node$i",$host,$pgport);

$node->{id} =$i;

$node->{raftport} =$raftport;

$node->{arbiter_port} =$arbiter_port;

push(@$nodes,$node);

}

{

my ($self) =@_;

my$nodes =$self->{nodes};

my$nnodes =scalar @{$nodes };

my$connstr =join(',',map {"${\$_->connstr('postgres') }" }@$nodes);

my$raftpeers =join(',',map {join(':',$_->{id},$_->host,$_->{raftport}) }@$nodes);

my$connstr =join(',',map {"${\$_->connstr('postgres') } arbiter_port=${\$_->{arbiter_port} }" }@$nodes);

foreachmy$node (@$nodes)

{

my$id =$node->{id};

my$host =$node->host;

my$pgport =$node->port;

my$raftport =$node->{raftport};

my$arbiter_port =$node->{arbiter_port};

$node->append_conf("postgresql.conf",qq(

));

$node->append_conf("pg_hba.conf",qq(

MODULE_big = multimaster

OBJS = multimaster.oraftable.oarbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o

OBJS = multimaster.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o

RAFTABLE_PATH = ../raftable

overrideCPPFLAGS += -I$(RAFTABLE_PATH) -I$(RAFTABLE_PATH)/raft/include

EXTRA_INSTALL = contrib/raftable contrib/mmts

EXTENSION = multimaster

DATA = multimaster--1.0.sql

typedefstruct {

TransactionIdxid;/* local transaction ID */

intMtmNodeId;

intMtmReplicationNodeId;

intMtmArbiterPort;

intMtmRaftablePort;

intMtmConnectTimeout;

intMtmReconnectTimeout;

intMtmNodeDisableDelay;

intMtmHeartbeatRecvTimeout;

intMtmMin2PCTimeout;

intMtmMax2PCRatio;

boolMtmUseRaftable;

boolMtmUseDtm;

boolMtmPreserveCommitOrder;

boolMtmVolksWagenMode;

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

if (i+1!=MtmNodeId) {

void*data=RaftableGet(psprintf("node-mask-%d",i+1),NULL,NULL,nowait);

void*data=&Mtm->nodes[i].connectivityMask;

if (data==NULL) {

return false;

}

MTM_LOG1("Disconnect node %d connectivity mask %llx",nodeId, (long long)Mtm->connectivityMask);

MtmUnlock();

if (!RaftableSet(psprintf("node-mask-%d",MtmNodeId),&Mtm->connectivityMask,sizeofMtm->connectivityMask, false))

{

elog(WARNING,"Disable node which is in minority according to RAFT");

MtmLock(LW_EXCLUSIVE);

if (Mtm->status==MTM_ONLINE) {

MtmSwitchClusterMode(MTM_IN_MINORITY);

}

MtmUnlock();

return;

}

MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout));

MtmRefreshClusterStatus(false);

}

BIT_CLEAR(Mtm->connectivityMask,nodeId-1);

BIT_CLEAR(Mtm->reconnectMask,nodeId-1);

MtmUnlock();

MTM_LOG1("Reconnect node %d, connectivityMask=%llx",nodeId, (long long)Mtm->connectivityMask);

RaftableSet(psprintf("node-mask-%d",MtmNodeId),&Mtm->connectivityMask,sizeofMtm->connectivityMask, false);

}

heap_close(rel,RowExclusiveLock);

}

}

staticvoidMtmRaftableInitialize()

{

inti;

for (i=0;i<MtmNodes;i++)

{

intport=MtmConnections[i].raftablePort;

if (port==0) {

port=MtmRaftablePort+i;

}

raftable_peer(i,MtmConnections[i].hostName,port);

}

raftable_start(MtmNodeId-1);

}

staticvoidMtmCheckControlFile(void)

{

memcpy(conn->hostName,host,hostLen);

conn->hostName[hostLen]='\0';

port=strstr(connStr,"raftport=");

port=strstr(connStr,"arbiter_port=");

if (port!=NULL) {

if (sscanf(port+9,"%d",&conn->raftablePort)!=1) {

elog(ERROR,"Invalid raftable port: %s",port+9);

}

}else {

conn->raftablePort=0;

}

port=strstr(connStr,"arbiterport=");

if (port!=NULL) {

if (sscanf(port+12,"%d",&conn->arbiterPort)!=1) {

elog(ERROR,"Invalid arbiter port: %s",port+12);

if (sscanf(port+13,"%d",&conn->arbiterPort)!=1) {

elog(ERROR,"Invalid arbiter port: %s",port+13);

}

}else {

conn->arbiterPort=MULTIMASTER_DEFAULT_ARBITER_PORT;

);

DefineCustomBoolVariable(

"multimaster.use_raftable",

"Use raftable plugin for internode communication",

NULL,

&MtmUseRaftable,

true,

PGC_BACKEND,

0,

NULL,

NULL,

);

DefineCustomBoolVariable(

"multimaster.ignore_tables_without_pk",

"Do not replicate tables withpout primary key",

);

DefineCustomIntVariable(

"multimaster.raftable_port",

"Base value for assigning raftable ports",

NULL,

&MtmRaftablePort,

6543,

0,

INT_MAX,

PGC_BACKEND,

0,

NULL,

NULL,

);

DefineCustomStringVariable(

"multimaster.conn_strings",

"Multimaster node connection strings separated by commas, i.e. 'replication=database dbname=postgres host=localhost port=5001,replication=database dbname=postgres host=localhost port=5002'",

BgwPoolStart(MtmWorkers,MtmPoolConstructor);

if (MtmUseRaftable) {

MtmRaftableInitialize();

}

MtmArbiterInitialize();

{

PGconn*conn;

char*safe_connstr=pstrdup(conninfo);

erase_option_from_connstr("raftport",safe_connstr);

erase_option_from_connstr("arbiterport",safe_connstr);

erase_option_from_connstr("arbiter_port",safe_connstr);

conn=PQconnectdb(safe_connstr);

inti;

for (i=0;i<Mtm->nAllNodes;i++)

{

size_tsize;

char*data=RaftableGet(psprintf("lock-graph-%d",i+1),&size,NULL, false);

if (data) {

GlobalTransactionId*gtid= (GlobalTransactionId*)data;

GlobalTransactionId*last= (GlobalTransactionId*)(data+size);

size_tlockGraphSize;

char*lockGraphData;

MtmLockNode(i+MtmMaxNodes,LW_SHARED);

lockGraphSize=Mtm->nodes[i].lockGraphUsed;

lockGraphData=palloc(lockGraphSize);

memcpy(lockGraphData,Mtm->nodes[i].lockGraphData,lockGraphSize);

MtmUnlockNode(i+MtmMaxNodes);

if (lockGraphData) {

GlobalTransactionId*gtid= (GlobalTransactionId*)lockGraphData;

GlobalTransactionId*last= (GlobalTransactionId*) (lockGraphData+lockGraphSize);

appendStringInfo(s,"node-%d lock graph: ",i+1);

while (gtid!=last) {

GlobalTransactionId*src=gtid++;

ByteBufferAlloc(&buf);

EnumerateLocks(MtmSerializeLock,&buf);

RaftableSet(psprintf("lock-graph-%d",MtmNodeId),buf.data,buf.used, false);

Assert(replorigin_session_origin==InvalidRepOriginId);

XLogFlush(LogLogicalMessage("L",buf.data,buf.used, false));

MtmSleep(MSEC_TO_USEC(DeadlockTimeout));

MtmGraphInit(&graph);

MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data,buf.used/sizeof(GlobalTransactionId));

ByteBufferFree(&buf);

for (i=0;i<Mtm->nAllNodes;i++) {

if (i+1!=MtmNodeId&& !BIT_CHECK(Mtm->disabledNodeMask,i)) {

size_tsize;

void*data=RaftableGet(psprintf("lock-graph-%d",i+1),&size,NULL, false);

if (data==NULL) {

size_tlockGraphSize;

void*lockGraphData;

MtmLockNode(i+MtmMaxNodes,LW_SHARED);

lockGraphSize=Mtm->nodes[i].lockGraphUsed;

lockGraphData=palloc(lockGraphSize);

memcpy(lockGraphData,Mtm->nodes[i].lockGraphData,lockGraphSize);

MtmUnlockNode(i+MtmMaxNodes);

if (lockGraphData==NULL) {

return true;/* If using Raftable is disabled */

}else {

MtmGraphAdd(&graph, (GlobalTransactionId*)data,size/sizeof(GlobalTransactionId));

MtmGraphAdd(&graph, (GlobalTransactionId*)lockGraphData,lockGraphSize/sizeof(GlobalTransactionId));

}

}

}