Add start/stop replication functions

knizhnik · knizhnik · commit 23aa4594fc65 · 2015-11-19T23:58:08.000+03:00
diff --git a/contrib/multimaster/decoder_raw.c b/contrib/multimaster/decoder_raw.c
@@ -39,7 +39,7 @@ extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
 typedef struct
 {
 	MemoryContext context;
-    bool isExternal;
+    bool isLocal;
 } DecoderRawData;
 
 static void decoder_raw_startup(LogicalDecodingContext *ctx,
@@ -82,7 +82,7 @@ decoder_raw_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 										  ALLOCSET_DEFAULT_MINSIZE,
 										  ALLOCSET_DEFAULT_INITSIZE,
 										  ALLOCSET_DEFAULT_MAXSIZE);
-    data->isExternal = false;
+    data->isLocal = false;
 	ctx->output_plugin_private = data;
 
 	opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
@@ -104,8 +104,8 @@ decoder_raw_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
 {   
 	DecoderRawData *data = ctx->output_plugin_private;
     
-    if (MultimasterIsExternalTransaction(txn->xid)) {
-        data->isExternal = true;
+    if (MMIsLocalTransaction(txn->xid)) {
+        data->isLocal = true;
     } else { 
         OutputPluginPrepareWrite(ctx, true);
         appendStringInfo(ctx->out, "BEGIN %u;", txn->xid);
@@ -119,7 +119,7 @@ decoder_raw_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 					 XLogRecPtr commit_lsn)
 {
 	DecoderRawData *data = ctx->output_plugin_private;
-    if (!data->isExternal) { 
+    if (!data->isLocal) { 
         OutputPluginPrepareWrite(ctx, true);
         appendStringInfoString(ctx->out, "COMMIT;");
         OutputPluginWrite(ctx, true);
@@ -473,7 +473,7 @@ decoder_raw_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	bool			is_rel_non_selective;
 
 	data = ctx->output_plugin_private;
-    if (data->isExternal) { 
+    if (data->isLocal) { 
         return;
     }
 	/* Avoid leaking memory by using and resetting our own context */
diff --git a/contrib/multimaster/multimaster--1.0.sql b/contrib/multimaster/multimaster--1.0.sql
@@ -1,2 +1,11 @@
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION multimaster" to load this file. \quit
+
+CREATE FUNCTION mm_start_replication() RETURNS void
+AS 'MODULE_PATHNAME','mm_start_replication'
+LANGUAGE C;
+
+CREATE FUNCTION mm_stop_replication() RETURNS void
+AS 'MODULE_PATHNAME','mm_stop_replication'
+LANGUAGE C;
+
diff --git a/contrib/multimaster/multimaster.c b/contrib/multimaster/multimaster.c
@@ -67,7 +67,7 @@ typedef struct
 {
     TransactionId xid;
     int count;
-} ExternalTransaction;
+} LocalTransaction;
 
 #define DTM_SHMEM_SIZE (1024*1024)
 #define DTM_HASH_SIZE  1003
@@ -77,6 +77,9 @@ void _PG_fini(void);
 
 PG_MODULE_MAGIC;
 
+PG_FUNCTION_INFO_V1(mm_start_replication);
+PG_FUNCTION_INFO_V1(mm_stop_replication);
+
 static Snapshot DtmGetSnapshot(Snapshot snapshot);
 static void DtmMergeWithGlobalSnapshot(Snapshot snapshot);
 static XidStatus DtmGetTransactionStatus(TransactionId xid, XLogRecPtr *lsn);
@@ -103,10 +106,11 @@ static void ByteBufferAppend(ByteBuffer* buf, void* data, int len);
 static void ByteBufferAppendInt32(ByteBuffer* buf, int data);
 static void ByteBufferFree(ByteBuffer* buf);
 
+static void MMMarkTransAsLocal(TransactionId xid);
 
 static shmem_startup_hook_type prev_shmem_startup_hook;
 static HTAB* xid_in_doubt;
-static HTAB* external_trans;
+static HTAB* local_trans;
 static DtmState* dtm;
 static Snapshot CurrentTransactionSnapshot;
 
@@ -128,9 +132,10 @@ static TransactionManager DtmTM = {
     DtmDetectGlobalDeadLock
 };
 
-static char* MultimasterConnStrs;
-static int MultimasterNodeId;
-static int MultimasterNodes;
+static char* MMConnStrs;
+static int   MMNodeId;
+static int   MMNodes;
+static bool  MMDoReplication = true;
 
 static char* DtmHost;
 static int DtmPort;
@@ -145,8 +150,8 @@ static BackgroundWorker DtmWorker = {
 };
 
 #define XTM_TRACE(fmt, ...)
-//#define XTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__)
-#define XTM_INFO(fmt, ...)
+#define XTM_INFO(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__)
+//#define XTM_INFO(fmt, ...)
 
 static void DumpSnapshot(Snapshot s, char *name)
 {
@@ -697,7 +702,7 @@ static void DtmInitialize()
 		dtm->xidLock = LWLockAssign();
 		dtm->nReservedXids = 0;
 		dtm->minXid = InvalidTransactionId;
-        dtm->nNodes = MultimasterNodes;
+        dtm->nNodes = MMNodes;
 		RegisterXactCallback(DtmXactCallback, NULL);
 	}
 	LWLockRelease(AddinShmemInitLock);
@@ -714,11 +719,11 @@ static void DtmInitialize()
 	);
 
 	info.keysize = sizeof(TransactionId);
-	info.entrysize = sizeof(ExternalTransaction);
+	info.entrysize = sizeof(LocalTransaction);
 	info.hash = dtm_xid_hash_fn;
 	info.match = dtm_xid_match_fn;
-	external_trans = ShmemInitHash(
-		"external_trans",
+	local_trans = ShmemInitHash(
+		"local_trans",
 		DTM_HASH_SIZE, DTM_HASH_SIZE,
 		&info,
 		HASH_ELEM | HASH_FUNCTION | HASH_COMPARE
@@ -734,14 +739,19 @@ DtmXactCallback(XactEvent event, void *arg)
 	XTM_INFO("%d: DtmXactCallbackevent=%d nextxid=%d\n", getpid(), event, DtmNextXid);
     switch (event) 
     {
-      case XACT_EVENT_START: 
-       if (MyProc && MyProc->backendId != InvalidBackendId) { 
-           printf("getpid=%d, MyProc=%d, MyProc->backendId=%d\n", getpid(), MyProc->pid, MyProc->backendId);
-           MultimasterBeginTransaction();
+    case XACT_EVENT_START: 
+        if (MyBackendId != InvalidBackendId && MMDoReplication) { 
+            printf("getpid=%d, backendId=%d\n", getpid(), MyBackendId);
+            MMBeginTransaction();
+        }
+        break;
+    case XACT_EVENT_PRE_COMMIT:
+    case XACT_EVENT_PARALLEL_PRE_COMMIT:
+        if (!MMDoReplication && TransactionIdIsValid(GetCurrentTransactionIdIfAny())) {
+            MMMarkTransAsLocal(GetCurrentTransactionIdIfAny());               
         }
         break;
-      case XACT_EVENT_COMMIT:
-      case XACT_EVENT_ABORT:
+    case XACT_EVENT_ABORT:
 		if (TransactionIdIsValid(DtmNextXid))
 		{
 			if (event == XACT_EVENT_COMMIT)
@@ -862,7 +872,7 @@ _PG_init(void)
 		"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'",
 		NULL,
-		&MultimasterConnStrs,
+		&MMConnStrs,
 		"",
 		PGC_POSTMASTER, // context
 		0, // flags,
@@ -875,7 +885,7 @@ _PG_init(void)
 		"multimaster.node_id",
 		"Multimaster node ID",
 		NULL,
-		&MultimasterNodeId,
+		&MMNodeId,
 		1,
 		1,
 		INT_MAX,
@@ -886,7 +896,7 @@ _PG_init(void)
 		NULL
 	);
     
-    MultimasterNodes = LogicalReplicationStartReceivers(MultimasterConnStrs, MultimasterNodeId);
+    MMNodes = MMStartReceivers(MMConnStrs, MMNodeId);
 
 	if (DtmBufferSize != 0)
 	{
@@ -924,10 +934,10 @@ static void DtmShmemStartup(void)
  *  ***************************************************************************
  */
 
-void MultimasterBeginTransaction(void)
+void MMBeginTransaction(void)
 {
 	if (TransactionIdIsValid(DtmNextXid))
-		elog(ERROR, "MultimasterBeginTransaction should be called only once for global transaction");
+		elog(ERROR, "MMBeginTransaction should be called only once for global transaction");
 	if (dtm == NULL)
 		elog(ERROR, "DTM is not properly initialized, please check that pg_dtm plugin was added to shared_preload_libraries list in postgresql.conf");
 	DtmNextXid = DtmGlobalStartTransaction(&DtmSnapshot, &dtm->minXid);
@@ -939,10 +949,8 @@ void MultimasterBeginTransaction(void)
 	DtmLastSnapshot = NULL;
 }
 
-void MultimasterJoinTransaction(TransactionId xid)
+void MMJoinTransaction(TransactionId xid)
 {
-    ExternalTransaction* et;
-
 	if (TransactionIdIsValid(DtmNextXid))
 		elog(ERROR, "dtm_begin/join_transaction should be called only once for global transaction");
 	DtmNextXid = xid;
@@ -955,28 +963,52 @@ void MultimasterJoinTransaction(TransactionId xid)
 	DtmHasGlobalSnapshot = true;
 	DtmLastSnapshot = NULL;
 
+    MMMarkTransAsLocal(DtmNextXid);
+}
+ 
+
+void MMMarkTransAsLocal(TransactionId xid)
+{
+    LocalTransaction* lt;
+
+    Assert(TransactionIdIsValid(xid));
     LWLockAcquire(dtm->hashLock, LW_EXCLUSIVE);
-    et = hash_search(external_trans, &DtmNextXid, HASH_ENTER, NULL);
-    et->count = dtm->nNodes-1;
+    lt = hash_search(local_trans, &xid, HASH_ENTER, NULL);
+    lt->count = dtm->nNodes-1;
     LWLockRelease(dtm->hashLock);
 }
 
-bool MultimasterIsExternalTransaction(TransactionId xid)
+bool MMIsLocalTransaction(TransactionId xid)
 {
-    ExternalTransaction* et;
+    LocalTransaction* lt;
     bool result = false;
     LWLockAcquire(dtm->hashLock, LW_EXCLUSIVE);
-    et = hash_search(external_trans, &xid, HASH_FIND, NULL);
-    if (et != NULL) { 
+    lt = hash_search(local_trans, &xid, HASH_FIND, NULL);
+    if (lt != NULL) { 
         result = true;
-        if (--et->count == 0) { 
-            hash_search(external_trans, &xid, HASH_REMOVE, NULL);
+        if (--lt->count == 0) { 
+            hash_search(local_trans, &xid, HASH_REMOVE, NULL);
         }
     }
     LWLockRelease(dtm->hashLock);
     return result;
 }
 
+Datum
+mm_start_replication(PG_FUNCTION_ARGS)
+{
+    MMDoReplication = true;
+    PG_RETURN_VOID();
+}
+
+Datum
+mm_stop_replication(PG_FUNCTION_ARGS)
+{
+    MMDoReplication = false;
+    PG_RETURN_VOID();
+}
+
+
 
 
 void DtmBackgroundWorker(Datum arg)
diff --git a/contrib/multimaster/multimaster.h b/contrib/multimaster/multimaster.h
@@ -1,10 +1,10 @@
 #ifndef __MULTIMASTER_H__
 #define __MULTIMASTER_H__
 
-extern int  LogicalReplicationStartReceivers(char* nodes, int node_id);
-extern void MultimasterBeginTransaction(void);
-extern void MultimasterJoinTransaction(TransactionId xid);
-extern bool MultimasterIsExternalTransaction(TransactionId xid);
+extern int  MMStartReceivers(char* nodes, int node_id);
+extern void MMBeginTransaction(void);
+extern void MMJoinTransaction(TransactionId xid);
+extern bool MMIsLocalTransaction(TransactionId xid);
 
 extern bool isBackgroundWorker;
 
diff --git a/contrib/multimaster/receiver_raw.c b/contrib/multimaster/receiver_raw.c
@@ -402,7 +402,7 @@ receiver_raw_main(Datum main_arg)
                 int rc = sscanf(stmt + 6, "%u", &xid);
                 Assert(rc == 1);
                 Assert(!insideTrans);
-                MultimasterJoinTransaction(xid);
+                MMJoinTransaction(xid);
                 insideTrans = true;
             } else if (strncmp(stmt, "COMMIT;", 7) == 0) { 
                 Assert(insideTrans);
@@ -526,7 +526,7 @@ receiver_raw_main(Datum main_arg)
 }
 
 
-int LogicalReplicationStartReceivers(char* conn_strs, int node_id)
+int MMStartReceivers(char* conn_strs, int node_id)
 {
     int i = 0;
 	BackgroundWorker worker;
diff --git a/contrib/multimaster/sockhub/sockhub.c b/contrib/multimaster/sockhub/sockhub.c
@@ -255,11 +255,11 @@ static void die(int sig) {
 void ShubLoop(Shub* shub)
 {    
     int buffer_size = shub->params->buffer_size;
+    sigset_t sset;
     signal(SIGINT, die);
     signal(SIGQUIT, die);
     signal(SIGTERM, die);
-    // signal(SIGHUP, die);
-    sigset_t sset;
+    /* signal(SIGHUP, die); */
     sigfillset(&sset);
     sigprocmask(SIG_UNBLOCK, &sset, NULL);
 
