Multimaster first sign of life

Author: knizhnik

contrib/multimaster/decoder_raw.c

@@ -108,8 +108,10 @@ decoder_raw_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
         data->isLocal = true;
     } else { 
         OutputPluginPrepareWrite(ctx, true);
+        elog(WARNING, "Send transation to %u to replica", txn->xid);
         appendStringInfo(ctx->out, "BEGIN %u;", txn->xid);
         OutputPluginWrite(ctx, true);
+        data->isLocal = false;
     }
 }
 
@@ -120,6 +122,7 @@ decoder_raw_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 {
 	DecoderRawData *data = ctx->output_plugin_private;
     if (!data->isLocal) { 
+        elog(WARNING, "Send commit of %u to replica", txn->xid);
         OutputPluginPrepareWrite(ctx, true);
         appendStringInfoString(ctx->out, "COMMIT;");
         OutputPluginWrite(ctx, true);
@@ -476,7 +479,9 @@ decoder_raw_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
     if (data->isLocal) { 
         return;
     }
-	/* Avoid leaking memory by using and resetting our own context */
+    elog(WARNING, "Send action %d in transaction  %u to replica", change->action, txn->xid);
+
+ 	/* Avoid leaking memory by using and resetting our own context */
 	old = MemoryContextSwitchTo(data->context);
 
 	/*

contrib/multimaster/dtmd/src/main.c

@@ -270,7 +270,7 @@ static xid_t get_global_xmin() {
 static void onbegin(client_t client, int argc, xid_t *argv) {
 	Transaction *t;
 	CHECK(
-		argc == 1,
+		argc == 2,
 		client,
 		"BEGIN: wrong number of arguments"
 	);
@@ -292,7 +292,7 @@ static void onbegin(client_t client, int argc, xid_t *argv) {
 
 	prev_gxid = t->xid = next_gxid++;
 	t->snapshots_count = 0;
-	t->size = 1;
+	t->size = argv[1];
 
     t->collision = transaction_hash[t->xid % MAX_TRANSACTIONS];
     transaction_hash[t->xid % MAX_TRANSACTIONS] = t;
@@ -448,7 +448,6 @@ static void onsnapshot(client_t client, int argc, xid_t *argv) {
 	if (CLIENT_XID(client) == INVALID_XID) {
 		CLIENT_SNAPSENT(client) = 0;
 		CLIENT_XID(client) = t->xid;
-		t->size += 1;
 	}
 
 	CHECK(

contrib/multimaster/init.sql

@@ -0,0 +1,2 @@
+create extension multimaster;
+create table foo(k integer primary key, v varchar);

contrib/multimaster/libdtm.c

@@ -296,7 +296,7 @@ void DtmInitSnapshot(Snapshot snapshot)
 	#endif
 }
 
-TransactionId DtmGlobalStartTransaction(Snapshot snapshot, TransactionId *gxmin)
+TransactionId DtmGlobalStartTransaction(Snapshot snapshot, TransactionId *gxmin, int nParticipants)
 {
 	int i;
 	xid_t xid;
@@ -307,7 +307,7 @@ TransactionId DtmGlobalStartTransaction(Snapshot snapshot, TransactionId *gxmin)
 	assert(snapshot != NULL);
 
 	// command
-	if (!dtm_send_command(dtm, CMD_BEGIN, 0)) goto failure;
+	if (!dtm_send_command(dtm, CMD_BEGIN, 1, nParticipants)) goto failure;
 
 	// results
 	reslen = dtm_recv_results(dtm, RESULTS_SIZE, results);

contrib/multimaster/libdtm.h

@@ -22,7 +22,7 @@ void DtmInitSnapshot(Snapshot snapshot);
  * smallest xmin among all snapshots known to arbiter. Returns INVALID_XID
  * otherwise.
  */
-TransactionId DtmGlobalStartTransaction(Snapshot snapshot, TransactionId *gxmin);
+TransactionId DtmGlobalStartTransaction(Snapshot snapshot, TransactionId *gxmin, int nParticipants);
 
 /**
  * Asks the arbiter for a fresh snapshot. Fills the 'snapshot' and 'gxmin' on