Eliminate recursion

Author: knizhnik

contrib/multimaster/decoder_raw.c

@@ -41,7 +41,7 @@ extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
 typedef struct
 {
 	MemoryContext context;
-	bool		include_transaction;
+    bool isExternal;
 } DecoderRawData;
 
 static void decoder_raw_startup(LogicalDecodingContext *ctx,
@@ -79,67 +79,16 @@ decoder_raw_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 	ListCell   *option;
 	DecoderRawData *data;
 
-	data = palloc(sizeof(DecoderRawData));
+	data = (DecoderRawData*)palloc(sizeof(DecoderRawData));
 	data->context = AllocSetContextCreate(ctx->context,
 										  "Raw decoder context",
 										  ALLOCSET_DEFAULT_MINSIZE,
 										  ALLOCSET_DEFAULT_INITSIZE,
 										  ALLOCSET_DEFAULT_MAXSIZE);
-	data->include_transaction = false;
-
+    data->isExternal = false;
 	ctx->output_plugin_private = data;
 
-	/* Default output format */
 	opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
-
-	foreach(option, ctx->output_plugin_options)
-	{
-		DefElem    *elem = lfirst(option);
-
-		Assert(elem->arg == NULL || IsA(elem->arg, String));
-
-		if (strcmp(elem->defname, "include_transaction") == 0)
-		{
-			/* if option does not provide a value, it means its value is true */
-			if (elem->arg == NULL)
-				data->include_transaction = true;
-			else if (!parse_bool(strVal(elem->arg), &data->include_transaction))
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						 errmsg("could not parse value \"%s\" for parameter \"%s\"",
-								strVal(elem->arg), elem->defname)));
-		}
-		else if (strcmp(elem->defname, "output_format") == 0)
-		{
-			char	   *format = NULL;
-
-			if (elem->arg == NULL)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						 errmsg("No value specified for parameter \"%s\"",
-								elem->defname)));
-
-			format = strVal(elem->arg);
-
-			if (strcmp(format, "textual") == 0)
-				opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
-			else if (strcmp(format, "binary") == 0)
-				opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
-			else
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						 errmsg("Incorrect value \"%s\" for parameter \"%s\"",
-								format, elem->defname)));
-		}
-		else
-		{
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					 errmsg("option \"%s\" = \"%s\" is unknown",
-							elem->defname,
-							elem->arg ? strVal(elem->arg) : "(null)")));
-		}
-	}
 }
 
 /* cleanup this plugin's resources */
@@ -155,16 +104,16 @@ decoder_raw_shutdown(LogicalDecodingContext *ctx)
 /* BEGIN callback */
 static void
 decoder_raw_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
-{
+{   
 	DecoderRawData *data = ctx->output_plugin_private;
-
-	/* Write to the plugin only if there is */
-	if (data->include_transaction)
-	{
-		OutputPluginPrepareWrite(ctx, true);
-		appendStringInfoString(ctx->out, "BEGIN;");
-		OutputPluginWrite(ctx, true);
-	}
+    
+    if (MultimasterIsExternalTransaction(txn->xid)) {
+        data->isExternal = true;
+    } else { 
+        OutputPluginPrepareWrite(ctx, true);
+        appendStringInfoString(ctx->out, "BEGIN %u;", txn->xid);
+        OutputPluginWrite(ctx, true);
+    }
 }
 
 /* COMMIT callback */
@@ -173,14 +122,11 @@ decoder_raw_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 					 XLogRecPtr commit_lsn)
 {
 	DecoderRawData *data = ctx->output_plugin_private;
-
-	/* Write to the plugin only if there is */
-	if (data->include_transaction)
-	{
-		OutputPluginPrepareWrite(ctx, true);
-		appendStringInfoString(ctx->out, "COMMIT;");
-		OutputPluginWrite(ctx, true);
-	}
+    if (!data->isExternal) { 
+        OutputPluginPrepareWrite(ctx, true);
+        appendStringInfoString(ctx->out, "COMMIT;");
+        OutputPluginWrite(ctx, true);
+    }
 }
 
 /*
@@ -530,7 +476,9 @@ decoder_raw_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	bool			is_rel_non_selective;
 
 	data = ctx->output_plugin_private;
-
+    if (data->isExternal) { 
+        return;
+    }
 	/* Avoid leaking memory by using and resetting our own context */
 	old = MemoryContextSwitchTo(data->context);
 

contrib/multimaster/multimaster.c

@@ -44,6 +44,7 @@
 #include "sockhub/sockhub.h"
 
 #include "libdtm.h"
+#include "multimaster.h"
 
 typedef struct
 {
@@ -52,6 +53,7 @@ typedef struct
 	TransactionId minXid;  /* XID of oldest transaction visible by any active transaction (local or global) */
 	TransactionId nextXid; /* next XID for local transaction */
 	size_t nReservedXids;  /* number of XIDs reserved for local transactions */
+    int    nNodes;
 } DtmState;
 
 typedef struct
@@ -61,19 +63,18 @@ typedef struct
     int used;
 } ByteBuffer;
 
+typedef struct
+{
+    TransactionId xid;
+    int count;
+} ExternalTransaction;
 
 #define DTM_SHMEM_SIZE (1024*1024)
 #define DTM_HASH_SIZE  1003
 
 void _PG_init(void);
 void _PG_fini(void);
 
-extern void LogicalReplicationStartReceivers(char* nodes, int node_id);
-extern void LogicalReplicationBroadcastXid(TransactonId Xid);
-
-void MultimasterBeginTransaction(void);
-void MultimasterJoinTransaction(TransactionId xid);
-
 static Snapshot DtmGetSnapshot(Snapshot snapshot);
 static void DtmMergeWithGlobalSnapshot(Snapshot snapshot);
 static XidStatus DtmGetTransactionStatus(TransactionId xid, XLogRecPtr *lsn);
@@ -103,6 +104,7 @@ static void ByteBufferFree(ByteBuffer* buf);
 
 static shmem_startup_hook_type prev_shmem_startup_hook;
 static HTAB* xid_in_doubt;
+static HTAB* external_trans;
 static DtmState* dtm;
 static Snapshot CurrentTransactionSnapshot;
 
@@ -126,11 +128,14 @@ static TransactionManager DtmTM = {
 
 static char* MultimasterConnStrs;
 static int MultimasterNodeId;
+static int MultimasterNodes;
 
 static char* DtmHost;
 static int DtmPort;
 static int DtmBufferSize;
 
+bool isBackgroundWorker;
+
 static BackgroundWorker DtmWorker = {
 	"DtmWorker",
 	0, /* do not need connection to the database */
@@ -694,6 +699,7 @@ static void DtmInitialize()
 		dtm->xidLock = LWLockAssign();
 		dtm->nReservedXids = 0;
 		dtm->minXid = InvalidTransactionId;
+        dtm->nNodes = MultimasterNodes;
 		RegisterXactCallback(DtmXactCallback, NULL);
 	}
 	LWLockRelease(AddinShmemInitLock);
@@ -709,6 +715,17 @@ static void DtmInitialize()
 		HASH_ELEM | HASH_FUNCTION | HASH_COMPARE
 	);
 
+	info.keysize = sizeof(TransactionId);
+	info.entrysize = sizeof(ExternalTransaction);
+	info.hash = dtm_xid_hash_fn;
+	info.match = dtm_xid_match_fn;
+	external_trans = ShmemInitHash(
+		"external_trans",
+		DTM_HASH_SIZE, DTM_HASH_SIZE,
+		&info,
+		HASH_ELEM | HASH_FUNCTION | HASH_COMPARE
+	);
+
 
 	TM = &DtmTM;
 }
@@ -720,7 +737,9 @@ DtmXactCallback(XactEvent event, void *arg)
     switch (event) 
     {
       case XACT_EVENT_BEGIN:
-          MultimasterBeginTransaction();
+          if (!isBackgroundWorker) {
+              MultimasterBeginTransaction();
+          }
           break;
       case XACT_EVENT_COMMIT:
       case XACT_EVENT_ABORT:
@@ -865,7 +884,7 @@ _PG_init(void)
 		NULL
 	);
 
-    LogicalReplicationStartReceivers(MultimasterConnStrs, MultimasterNodeId);
+    MultimasterNodes = LogicalReplicationStartReceivers(MultimasterConnStrs, MultimasterNodeId);
 
 	if (DtmBufferSize != 0)
 	{
@@ -931,10 +950,8 @@ dtm_get_current_snapshot_xcnt(PG_FUNCTION_ARGS)
 
 void MultimasterBeginTransaction(void)
 {
-	if (TransactionIdIsValid(DtmNextXid)) {
-        /* slave transaction */
-        return;
-    }
+	if (TransactionIdIsValid(DtmNextXid))
+		elog(ERROR, "MultimasterBeginTransaction 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);
@@ -944,12 +961,12 @@ void MultimasterBeginTransaction(void)
 
 	DtmHasGlobalSnapshot = true;
 	DtmLastSnapshot = NULL;
-
-    LogicalReplicationBroadcastXid(DtmNextXid);
 }
 
 void MultimasterJoinTransaction(TransactionId xid)
 {
+    ExternalTrans* et;
+
 	if (TransactionIdIsValid(DtmNextXid))
 		elog(ERROR, "dtm_begin/join_transaction should be called only once for global transaction");
 	DtmNextXid = xid;
@@ -961,14 +978,39 @@ void MultimasterJoinTransaction(TransactionId xid)
 
 	DtmHasGlobalSnapshot = true;
 	DtmLastSnapshot = NULL;
+
+    LWLockAcquire(dtm->hashLock, LW_EXCLUSIVE);
+    et = hash_search(external_trans, &DtmNextXid, HASH_ENTER, NULL);
+    et->count = dtm->nNodes-1;
+    LWLockRelease(dtm->hashLock);
 }
 
+bool MultimasterIsExternalTransaction(TransactionId xid)
+{
+    ExternalTrans* et;
+    bool result = false;
+    LWLockAcquire(dtm->hashLock, LW_EXCLUSIVE);
+    et = hash_search(external_trans, &DtmNextXid, HASH_FIND, NULL);
+    if (et != NULL) { 
+        result = true;
+        if (--et->count == 0) { 
+            hash_search(external_trans, &DtmNextXid, HASH_REMOVE, NULL);
+        }
+    }
+    LWLockRelease(dtm->hashLock);
+    return result;
+}
+
+
+
 void DtmBackgroundWorker(Datum arg)
 {
 	Shub shub;
 	ShubParams params;
 	char unix_sock_path[MAXPGPATH];
 
+    isBackgroundWorker = true;
+    
 	snprintf(unix_sock_path, sizeof(unix_sock_path), "%s/p%d", Unix_socket_directories, DtmPort);
 
 	ShubInitParams(&params);

contrib/multimaster/multimaster.h

@@ -0,0 +1,11 @@
+#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 bool isBackgroundWorker;
+
+#endif