Commitcbdf8bb

committed

Multimaster first sign of life

1 parent23aa459 commitcbdf8bbCopy full SHA for cbdf8bb

File tree

+191

-75

lines changed

+191

-75

lines changed

Lines changed: 6 additions & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -108,8 +108,10 @@ decoder_raw_begin_txn(LogicalDecodingContext ctx, ReorderBufferTXN txn)`
`108`	`108`	`data->isLocal= true;`
`109`	`109`	`}else {`
`110`	`110`	`OutputPluginPrepareWrite(ctx, true);`
	`111`	`+elog(WARNING,"Send transation to %u to replica",txn->xid);`
`111`	`112`	`appendStringInfo(ctx->out,"BEGIN %u;",txn->xid);`
`112`	`113`	`OutputPluginWrite(ctx, true);`
	`114`	`+data->isLocal= false;`
`113`	`115`	`}`
`114`	`116`	`}`
`115`	`117`
`@@ -120,6 +122,7 @@ decoder_raw_commit_txn(LogicalDecodingContext ctx, ReorderBufferTXN txn,`
`120`	`122`	`{`
`121`	`123`	`DecoderRawData*data=ctx->output_plugin_private;`
`122`	`124`	`if (!data->isLocal) {`
	`125`	`+elog(WARNING,"Send commit of %u to replica",txn->xid);`
`123`	`126`	`OutputPluginPrepareWrite(ctx, true);`
`124`	`127`	`appendStringInfoString(ctx->out,"COMMIT;");`
`125`	`128`	`OutputPluginWrite(ctx, true);`
`@@ -476,7 +479,9 @@ decoder_raw_change(LogicalDecodingContext ctx, ReorderBufferTXN txn,`
`476`	`479`	`if (data->isLocal) {`
`477`	`480`	`return;`
`478`	`481`	`}`
`479`		`-/* Avoid leaking memory by using and resetting our own context */`
	`482`	`+elog(WARNING,"Send action %d in transaction %u to replica",change->action,txn->xid);`
	`483`	`+`
	`484`	`+/* Avoid leaking memory by using and resetting our own context */`
`480`	`485`	`old=MemoryContextSwitchTo(data->context);`
`481`	`486`
`482`	`487`	`/*`

Lines changed: 2 additions & 3 deletions

Original file line number	Diff line number	Diff line change
`@@ -270,7 +270,7 @@ static xid_t get_global_xmin() {`
`270`	`270`	`staticvoidonbegin(client_tclient,intargc,xid_t*argv) {`
`271`	`271`	`Transaction*t;`
`272`	`272`	`CHECK(`
`273`		`-argc==1,`
	`273`	`+argc==2,`
`274`	`274`	`client,`
`275`	`275`	`"BEGIN: wrong number of arguments"`
`276`	`276`	`);`
`@@ -292,7 +292,7 @@ static void onbegin(client_t client, int argc, xid_t *argv) {`
`292`	`292`
`293`	`293`	`prev_gxid=t->xid=next_gxid++;`
`294`	`294`	`t->snapshots_count=0;`
`295`		`-t->size=1;`
	`295`	`+t->size=argv[1];`
`296`	`296`
`297`	`297`	`t->collision=transaction_hash[t->xid %MAX_TRANSACTIONS];`
`298`	`298`	`transaction_hash[t->xid %MAX_TRANSACTIONS]=t;`
`@@ -448,7 +448,6 @@ static void onsnapshot(client_t client, int argc, xid_t *argv) {`
`448`	`448`	`if (CLIENT_XID(client)==INVALID_XID) {`
`449`	`449`	`CLIENT_SNAPSENT(client)=0;`
`450`	`450`	`CLIENT_XID(client)=t->xid;`
`451`		`-t->size+=1;`
`452`	`451`	`}`
`453`	`452`
`454`	`453`	`CHECK(`

Lines changed: 2 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+create extension multimaster;`
	`2`	`+createtablefoo(kintegerprimary key, vvarchar);`

Lines changed: 2 additions & 2 deletions

Original file line number	Diff line number	Diff line change
`@@ -296,7 +296,7 @@ void DtmInitSnapshot(Snapshot snapshot)`
`296`	`296`	`#endif`
`297`	`297`	`}`
`298`	`298`
`299`		`-TransactionIdDtmGlobalStartTransaction(Snapshotsnapshot,TransactionId*gxmin)`
	`299`	`+TransactionIdDtmGlobalStartTransaction(Snapshotsnapshot,TransactionId*gxmin,intnParticipants)`
`300`	`300`	`{`
`301`	`301`	`inti;`
`302`	`302`	`xid_txid;`
`@@ -307,7 +307,7 @@ TransactionId DtmGlobalStartTransaction(Snapshot snapshot, TransactionId *gxmin)`
`307`	`307`	`assert(snapshot!=NULL);`
`308`	`308`
`309`	`309`	`// command`
`310`		`-if (!dtm_send_command(dtm,CMD_BEGIN,0)) gotofailure;`
	`310`	`+if (!dtm_send_command(dtm,CMD_BEGIN,1,nParticipants)) gotofailure;`
`311`	`311`
`312`	`312`	`// results`
`313`	`313`	`reslen=dtm_recv_results(dtm,RESULTS_SIZE,results);`

Lines changed: 1 addition & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,7 @@ void DtmInitSnapshot(Snapshot snapshot);`
`22`	`22`	`* smallest xmin among all snapshots known to arbiter. Returns INVALID_XID`
`23`	`23`	`* otherwise.`
`24`	`24`	`*/`
`25`		`-TransactionIdDtmGlobalStartTransaction(Snapshotsnapshot,TransactionId*gxmin);`
	`25`	`+TransactionIdDtmGlobalStartTransaction(Snapshotsnapshot,TransactionId*gxmin,intnParticipants);`
`26`	`26`
`27`	`27`	`/**`
`28`	`28`	`* Asks the arbiter for a fresh snapshot. Fills the 'snapshot' and 'gxmin' on`

Comments

(0)