Support table copy between multimaster nodes

Author: knizhnik

contrib/mmts/arbitrator/arbitrator.cpp

@@ -2,6 +2,7 @@
 #include <stdio.h>
 #include <stdarg.h>
 #include <stdlib.h>
+#include <unistd.h>
 #include <inttypes.h>
 #include <sys/time.h>
 #include <pthread.h>
@@ -34,18 +35,34 @@ class my_unique_ptr
   public:
     my_unique_ptr(T* p = NULL) : ptr(p) {}
     ~my_unique_ptr() { delete ptr; }
-    T& operator*() { return *ptr; }
-    T* operator->() { return ptr; }
-    void operator=(T* p) { ptr = p; }
+    T& operator*() const  { return *ptr; }
+    T* operator->() const { return ptr; }
+	bool operator ==(T const* p) const { return ptr == p; }
+	bool operator !=(T const* p) const { return ptr != p; }
+    void operator=(T* p) { 
+		delete ptr;
+		ptr = p; 
+	}
     void operator=(my_unique_ptr& other) {
+		delete ptr;
         ptr = other.ptr;
         other.ptr = NULL;
     }        
 };
 
+struct config
+{
+	int timeout;
+	vector<string> connections;
+
+	config() {
+		timeout = 1000000; // 1 second
+	}	
+};
+
 int main (int argc, char* argv[])
 {
-	vector<string> connection_strings;
+	config cfg;
 
 	if (argc == 1){
         printf("Use -h to show usage options\n");
@@ -56,26 +73,27 @@ int main (int argc, char* argv[])
         if (argv[i][0] == '-') { 
             switch (argv[i][1]) {
 			  case 't':
-				cfs.timeout = atoi(argv[++i]);
+				cfg.timeout = atoi(argv[++i]);
 				continue;
 			  case 'c':
                 cfg.connections.push_back(string(argv[++i]));
                 continue;
 			}
         }
 		printf("Options:\n"
-                "\t-t TIMEOUT\ttimeout in seconds of waiting database connection string\n"
+                "\t-t TIMEOUT\ttimeout in microseconds of waiting database connection string (default: 1 second)\n"
 			   "\t-c STR\tdatabase connection string\n");
         return 1;
     }
 
-	size_t nConns = connection_strings.size();
+	size_t nConns = cfg.connections.size();
     vector< my_unique_ptr<connection> > conns(nConns);
 	for (size_t i = 0; i < nConns; i++) {
-		conns[i] = new connection(connection_strings[i]);
+		conns[i] = new connection(cfg.connections[i]);
     }
 	nodemask_t disabledMask = 0;
-	nodemask_t enabledMask = 0;
+	nodemask_t newEnabledMask = 0;
+	nodemask_t oldEnabledMask = 0;
 
 	while (true) { 
         vector< my_unique_ptr<nontransaction> > txns(conns.size());
@@ -84,46 +102,63 @@ int main (int argc, char* argv[])
 		char sql[128];
 		sprintf(sql, "select mtm.arbitrator_poll(%lld)", disabledMask);
 
+		// Initiate queries to all live nodes
         for (size_t i = 0; i < nConns; i++) {        
-			if (BIT_CHECK(disabledMask, i)) { 
-				if (BIT_CHECK(enabledMask, i)) { 
+			// Some of live node reestablished connection with dead node, so arbitrator should also try to connect to this node
+			if (conns[i] == NULL) { 
+				if (BIT_CHECK(newEnabledMask, i)) { 
 					try { 
-						delete conns[i];
-						conns[i] = new connection(connection_strings[i]);
+						conns[i] = new connection(cfg.connections[i]);
 						BIT_CLEAR(disabledMask, i);
+						fprintf(stdout, "Reestablish connection with node %d\n", (int)i+1);
 					} catch (pqxx_exception const& x) { 
-						conns[i] = NULL;
-						fprintf(stderr, "Failed to connect to node %d: %s\n", (int)i+1, x.base().what());
+						if (conns[i] == NULL) { 
+							conns[i] = NULL;
+							fprintf(stderr, "Failed to connect to node %d: %s\n", (int)i+1, x.base().what());
+						}
 					}
 				}
-			}
-			if (!BIT_CHECK(disabledMask, i)) { 
+			} else { 
 				txns[i] = new nontransaction(*conns[i]);
 				pipes[i] = new pipeline(*txns[i]);
 				queries[i] = pipes[i]->insert(sql);
 			}
-			sleep(cfg.timeout);
-			enabledMask = 0;
-			for (size_t i = 0; i < nConns; i++) {        
-				if (!BIT_CHECK(didsabledMask, i)) { 
-					if (!pipes[i]->is_finished(queries[i])) 
-					{ 
-						fprintf(stderr, "Doesn't receive response from node %d within %d seconds\n", (int)i+1, cfs.timeout);
-						BIT_SET(disabledMask, i);
-						delete conns[i];
+		}
+		// Wait some time
+		usleep(cfg.timeout);
+		oldEnabledMask = newEnabledMask;
+		newEnabledMask = ~0;
+		for (size_t i = 0; i < nConns; i++) {        
+			if (!BIT_CHECK(disabledMask, i)) { 
+				if (!pipes[i]->is_finished(queries[i])) { 
+					fprintf(stderr, "Doesn't receive response from node %d within %d microseconds\n", (int)i+1, cfg.timeout);
+					BIT_SET(disabledMask, i);
+					conns[i] = NULL;
+				} else {
+					try { 
+						result r = pipes[i]->retrieve(queries[i]);
+						newEnabledMask &= r[0][0].as(nodemask_t());
+					} catch (pqxx_exception const& x) { 
 						conns[i] = NULL;
-					} else {
-						try { 
-							result r = pipes[i]->retrieve(results[i]);
-							enabledMask |= r[0][0].as(nodemask_t());
-						} catch (pqxx_exception const& x) { 
-							delete conns[i];
-							conns[i] = NULL;
-							fprintf(stderr, "Failed to retrieve result from node %d: %s\n", (int)i+1, x.base().what());
-						}							
-					}
+						fprintf(stderr, "Failed to retrieve result from node %d: %s\n", (int)i+1, x.base().what());
+					}							
 				}
 			}
 		}
+		if (newEnabledMask == ~0) { 
+			if (oldEnabledMask != ~0) { 
+				fprintf(stdout, "There are no more live nodes\n");
+			}				
+			// No live nodes: 
+			disabledNodeMask = 0;
+		} else { 
+			if (newEnabledMask != oldEnabledMask) { 
+				for (size_t i = 0; i < nConns; i++) {        
+					if (BIT_CHECK(newEnabledMask ^ oldEnabledMask, i)) { 					
+						fprintf(stdout, "Node %d is %s\n", (int)i+1, BIT_CHECK(newEnabledMask, i) ? "enabled" : "disabled");
+					}
+				}
+			}	   
+		}
 	}
 }

contrib/mmts/multimaster--1.0.sql

@@ -75,6 +75,14 @@ CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_make_table_local'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.broadcast_table(srcTable regclass, dstNodesMask bigint) RETURNS void
+AS 'MODULE_PATHNAME','mtm_broadcast_table'
+LANGUAGE C;
+
+CREATE FUNCTION mtm.copy_table(srcTable regclass, dstNode integer) RETURNS void
+AS 'MODULE_PATHNAME','mtm_copy_table'
+LANGUAGE C;
+
 CREATE FUNCTION mtm.dump_lock_graph() RETURNS text
 AS 'MODULE_PATHNAME','mtm_dump_lock_graph'
 LANGUAGE C;

contrib/mmts/multimaster.c

@@ -129,6 +129,8 @@ PG_FUNCTION_INFO_V1(mtm_dump_lock_graph);
 PG_FUNCTION_INFO_V1(mtm_inject_2pc_error);
 PG_FUNCTION_INFO_V1(mtm_check_deadlock);
 PG_FUNCTION_INFO_V1(mtm_arbitrator_poll);
+PG_FUNCTION_INFO_V1(mtm_broadcast_table);
+PG_FUNCTION_INFO_V1(mtm_copy_table);
 
 static Snapshot MtmGetSnapshot(Snapshot snapshot);
 static void MtmInitialize(void);
@@ -4368,6 +4370,24 @@ mtm_collect_cluster_info(PG_FUNCTION_ARGS)
 	}
 }
 
+Datum mtm_broadcast_table(PG_FUNCTION_ARGS)
+{
+	MtmCopyRequest copy;
+	copy.sourceTable = PG_GETARG_OID(0);
+	copy.targetNodes = PG_GETARG_INT64(1);
+	LogLogicalMessage("B", (char*)&copy, sizeof(copy), false);
+	PG_RETURN_VOID();
+}
+
+Datum mtm_copy_table(PG_FUNCTION_ARGS)
+{
+	MtmCopyRequest copy;
+	copy.sourceTable = PG_GETARG_OID(0);
+	copy.targetNodes = (nodemask_t)1 << (PG_GETARG_INT32(1) - 1);
+	LogLogicalMessage("B", (char*)&copy, sizeof(copy), false);
+	PG_RETURN_VOID();
+}
+
 
 Datum mtm_make_table_local(PG_FUNCTION_ARGS)
 {

contrib/mmts/multimaster.h

@@ -207,6 +207,13 @@ typedef struct
 } MtmConnectionInfo;
 
 
+typedef struct
+{
+	Oid sourceTable;
+	nodemask_t targetNodes;
+} MtmCopyRequest;
+
+
 typedef struct
 {
 	MtmConnectionInfo con;

contrib/mmts/pglogical_apply.c

@@ -14,6 +14,7 @@
 #include "catalog/catversion.h"
 #include "catalog/dependency.h"
 #include "catalog/index.h"
+#include "catalog/heap.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_type.h"
 
@@ -367,6 +368,49 @@ process_remote_begin(StringInfo s)
 	return true;
 }
 
+static void 
+process_broadcast_table(StringInfo s)
+{
+	Relation rel;
+	char ch;
+	EState *estate;
+	TupleData new_tuple;
+	TupleTableSlot *newslot;
+	TupleTableSlot *oldslot;
+	HeapTuple tup;
+
+	StartTransactionCommand();
+
+    ch = pq_getmsgbyte(s);
+	Assert(ch == 'R');
+	rel = read_rel(s, AccessExclusiveLock);
+
+	heap_truncate_one_rel(rel);
+
+	estate = create_rel_estate(rel);
+	newslot = ExecInitExtraTupleSlot(estate);
+	oldslot = ExecInitExtraTupleSlot(estate);
+	ExecSetSlotDescriptor(newslot, RelationGetDescr(rel));
+	ExecSetSlotDescriptor(oldslot, RelationGetDescr(rel));
+
+	ExecOpenIndices(estate->es_result_relation_info, false);
+	
+	while (s->cursor != s->len) { 
+		read_tuple_parts(s, rel, &new_tuple);
+		tup = heap_form_tuple(RelationGetDescr(rel),
+							  new_tuple.values, new_tuple.isnull);
+		ExecStoreTuple(tup, newslot, InvalidBuffer, true);
+		simple_heap_insert(rel, newslot->tts_tuple);
+		UserTableUpdateOpenIndexes(estate, newslot);
+	}
+
+	ExecCloseIndices(estate->es_result_relation_info);
+    ExecResetTupleTable(estate->es_tupleTable, true); 
+    FreeExecutorState(estate);
+
+	CommitTransactionCommand();
+}
+
 static bool
 process_remote_message(StringInfo s)
 {
@@ -377,6 +421,12 @@ process_remote_message(StringInfo s)
 
 	switch (action)
 	{
+  	    case 'B':
+		{
+			process_broadcast_table(s);
+			standalone = true;
+			break;
+		}
 		case 'C':
 		{
 			MTM_LOG1("%d: Executing non-tx utility statement %s", MyProcPid, messageBody);

contrib/mmts/pglogical_output.c

@@ -557,7 +557,7 @@ pg_decode_message(LogicalDecodingContext *ctx,
 	PGLogicalOutputData* data = (PGLogicalOutputData*)ctx->output_plugin_private;
 
 	MtmOutputPluginPrepareWrite(ctx, true, !transactional);
-	data->api->write_message(ctx->out, prefix, sz, message);
+	data->api->write_message(ctx->out, data, prefix, sz, message);
 	MtmOutputPluginWrite(ctx, true, !transactional);
 }
 

contrib/mmts/pglogical_proto.c

@@ -34,6 +34,7 @@
 #include "utils/syscache.h"
 #include "utils/timestamp.h"
 #include "utils/typcache.h"
+#include "utils/snapmgr.h"
 
 #include "multimaster.h"
 #include "pglogical_relid_map.h"
@@ -61,7 +62,7 @@ static void pglogical_write_delete(StringInfo out, PGLogicalOutputData *data,
 							Relation rel, HeapTuple oldtuple);
 
 static void pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
-								   Relation rel, HeapTuple tuple);
+								  Relation rel, HeapTuple tuple);
 static char decide_datum_transfer(Form_pg_attribute att,
 								  Form_pg_type typclass,
 								  bool allow_internal_basetypes,
@@ -167,8 +168,38 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 	}
 }
 
+static void pglogical_broadcast_table(StringInfo out, PGLogicalOutputData *data, MtmCopyRequest* copy)
+{
+	if (BIT_CHECK(copy->targetNodes, MtmReplicationNodeId-1)) { 
+		HeapScanDesc scandesc;
+		HeapTuple	 tuple;
+		Relation     rel;
+		
+		StartTransactionCommand();
+
+		rel = heap_open(copy->sourceTable, ShareLock);
+		
+		pq_sendbyte(out, 'M');
+		pq_sendbyte(out, 'B');
+		pq_sendint(out, sizeof(*copy), 4);
+		pq_sendbytes(out, (char*)copy, sizeof(*copy));
+		
+		pglogical_write_rel(out, data, rel);
+
+		scandesc = heap_beginscan(rel, GetTransactionSnapshot(), 0, NULL);
+		while ((tuple = heap_getnext(scandesc, ForwardScanDirection)) != NULL)
+		{
+			pglogical_write_tuple(out, data, rel, tuple);
+		}
+		heap_endscan(scandesc);
+		heap_close(rel, ShareLock);
+
+		CommitTransactionCommand();
+	}
+}
+
 static void
-pglogical_write_message(StringInfo out,
+pglogical_write_message(StringInfo out, PGLogicalOutputData *data,
 						const char *prefix, Size sz, const char *message)
 {
 	MtmLastRelId = InvalidOid;
@@ -199,6 +230,9 @@ pglogical_write_message(StringInfo out,
 		 * so no need to send that to replicas.
 		 */
 		return;
+	  case 'B':
+		pglogical_broadcast_table(out, data, (MtmCopyRequest*)message);
+		return;
 	}
 	pq_sendbyte(out, 'M');
 	pq_sendbyte(out, *prefix);

contrib/mmts/pglogical_proto.h

@@ -21,7 +21,7 @@ typedef void (*pglogical_write_rel_fn)(StringInfo out, struct PGLogicalOutputDat
 
 typedef void (*pglogical_write_begin_fn)(StringInfo out, struct PGLogicalOutputData *data,
 							 ReorderBufferTXN *txn);
-typedef void (*pglogical_write_message_fn)(StringInfo out,
+typedef void (*pglogical_write_message_fn)(StringInfo out, struct PGLogicalOutputData *data,
 					const char *prefix, Size sz, const char *message);
 typedef void (*pglogical_write_commit_fn)(StringInfo out, struct PGLogicalOutputData *data,
 							 ReorderBufferTXN *txn, XLogRecPtr commit_lsn);