dmq integration: add resolver worker

Author: kelvich

Makefile

@@ -3,7 +3,7 @@ EXTENSION = multimaster
 DATA = multimaster--1.0.sql
 OBJS = multimaster.o dmq.o commit.o bytebuf.o bgwpool.o pglogical_output.o \
 pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o \
-pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o state.o
+pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o state.o resolver.o
 MODULE_big = multimaster
 
 PG_CPPFLAGS = -I$(libpq_srcdir)

commit.c

@@ -7,7 +7,7 @@
 
 static Oid		MtmDatabaseId;
 static bool		DmqSubscribed;
-static int		sender_to_node[64];
+static int		sender_to_node[MTM_MAX_NODES];
 
 MtmCurrentTrans MtmTx;
 
@@ -97,17 +97,16 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 
 	if (!DmqSubscribed)
 	{
-		int i, j;
+		int i, sender_id = 0;
 
-		for (j = 0, i = 0; i < Mtm->nAllNodes; i++)
+		for (i = 0; i < Mtm->nAllNodes; i++)
 		{
-			if (i + 1 == MtmNodeId)
-				continue;
-
-			dmq_stream_subscribe(psprintf("node%d", i + 1),
+			if (i + 1 != MtmNodeId)
+			{
+				dmq_stream_subscribe(psprintf("node%d", i + 1),
 								psprintf("be%d", MyProc->pgprocno));
-
-			sender_to_node[j++] = i + 1;
+				sender_to_node[sender_id++] = i + 1;
+			}
 		}
 		DmqSubscribed = true;
 	}

dmq.c

@@ -39,7 +39,7 @@
 #include "tcop/tcopprot.h"
 #include "utils/dynahash.h"
 
-#define DMQ_MQ_SIZE ((Size)1024)
+#define DMQ_MQ_SIZE  ((Size) 65536)
 #define DMQ_MQ_MAGIC 0x646d71
 
 /*
@@ -110,7 +110,8 @@ static DmqBackendState dmq_local;
 
 static shmem_startup_hook_type PreviousShmemStartupHook;
 
-dmq_receiver_start_hook_type dmq_receiver_start_hook;
+dmq_receiver_hook_type dmq_receiver_start_hook;
+dmq_receiver_hook_type dmq_receiver_stop_hook;
 
 void dmq_sender_main(Datum main_arg);
 
@@ -231,11 +232,11 @@ dmq_init(void)
 	worker.bgw_main_arg = 0;
 	sprintf(worker.bgw_library_name, "multimaster");
 	sprintf(worker.bgw_function_name, "dmq_sender_main");
-	snprintf(worker.bgw_name, BGW_MAXLEN, "dmq_sender");
-	snprintf(worker.bgw_type, BGW_MAXLEN, "dmq_sender");
+	snprintf(worker.bgw_name, BGW_MAXLEN, "mtm-dmq-sender");
+	snprintf(worker.bgw_type, BGW_MAXLEN, "mtm-dmq-sender");
 	RegisterBackgroundWorker(&worker);
 
-	/* Register shmem hook for all backends */
+	/* Register shmem hooks */
 	PreviousShmemStartupHook = shmem_startup_hook;
 	shmem_startup_hook = dmq_shmem_startup_hook;
 
@@ -838,12 +839,15 @@ _pq_getbyte_if_available(unsigned char *c)
 static void
 dmq_receiver_at_exit(int status, Datum sender)
 {
-	int sender_id = DatumGetInt32(sender);
+	int		sender_id = DatumGetInt32(sender);
+	char	sender_name[DMQ_NAME_MAXLEN];
 
 	LWLockAcquire(dmq_state->lock, LW_EXCLUSIVE);
-	dmq_state->receiver[DatumGetInt32(sender_id)][0] = '\0';
+	strncmp(sender_name, dmq_state->receiver[sender_id], DMQ_NAME_MAXLEN);
+	dmq_state->receiver[sender_id][0] = '\0';
 	LWLockRelease(dmq_state->lock);
 
+	dmq_receiver_stop_hook(sender_name);
 }
 
 
@@ -1162,6 +1166,36 @@ dmq_pop(DmqSenderId *sender_id, StringInfo msg)
 	}
 }
 
+bool
+dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg)
+{
+	shm_mq_result res;
+	int i;
+
+	for (i = 0; i < dmq_local.n_inhandles; i++)
+	{
+		Size	len;
+		void   *data;
+
+		res = shm_mq_receive(dmq_local.mq_inh[i], &len, &data, true);
+		if (res == SHM_MQ_SUCCESS)
+		{
+			msg->data = data;
+			msg->len = len;
+			msg->maxlen = len;
+			msg->cursor = 0;
+			*sender_id = i;
+			// elog(LOG, "pop message %s, from %d", (char *) *dataptr, i);
+			return true;
+		}
+		else if (res == SHM_MQ_DETACHED)
+		{
+			elog(ERROR, "dmq_pop: queue detached");
+		}
+	}
+
+	return false;
+}
 
 DmqDestinationId
 dmq_destination_add(char *connstr, char *sender_name, int ping_period)

dmq.h

@@ -14,9 +14,11 @@ extern DmqDestinationId dmq_destination_add(char *connstr, char *sender_name, in
 extern void dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg);
 extern void dmq_stream_subscribe(char *sender_name, char *stream_name);
 extern void dmq_pop(DmqSenderId *sender_id, StringInfo msg);
+extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg);
 extern void dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *buffer, size_t len);
 
-typedef void (*dmq_receiver_start_hook_type) (char *);
-extern dmq_receiver_start_hook_type dmq_receiver_start_hook;
+typedef void (*dmq_receiver_hook_type) (char *);
+extern dmq_receiver_hook_type dmq_receiver_start_hook;
+extern dmq_receiver_hook_type dmq_receiver_stop_hook;
 
 #endif

multimaster.c

@@ -161,7 +161,7 @@ static bool MtmRunUtilityStmt(PGconn* conn, char const* sql, char **errmsg);
 static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError, int forceOnNode);
 static void MtmProcessDDLCommand(char const* queryString, bool transactional);
 
-static int  MtmGidParseNodeId(const char* gid);
+// static int  MtmGidParseNodeId(const char* gid);
 
 // static void MtmLockCluster(void);
 // static void MtmUnlockCluster(void);
@@ -214,6 +214,18 @@ char const* const MtmTxnStatusMnem[] =
 	"Unknown"
 };
 
+char const* const MtmTxStateMnem[] =
+{
+	"MtmTxUnknown",
+	"MtmTxNotFound",
+	"MtmTxInProgress",
+	"MtmTxPrepared",
+	"MtmTxPreCommited",
+	"MtmTxPreAborted",
+	"MtmTxCommited",
+	"MtmTxAborted"
+};
+
 bool  MtmDoReplication;
 char* MtmDatabaseName;
 char* MtmDatabaseUser;
@@ -1013,7 +1025,7 @@ bool MtmWatchdog(timestamp_t now)
 			{
 				MTM_LOG1("[STATE] Node %i: Disconnect due to heartbeat timeout (%d msec)",
 					 i+1, (int)USEC_TO_MSEC(now - Mtm->nodes[i].lastHeartbeat));
-				MtmOnNodeDisconnect(i+1);
+				// MtmOnNodeDisconnect(i+1);
 				allAlive = false;
 			}
 		}
@@ -2400,7 +2412,7 @@ static void MtmInitialize()
 		Mtm->nLiveNodes = 0; //MtmNodes;
 		Mtm->nAllNodes = MtmNodes;
 		Mtm->disabledNodeMask =  (((nodemask_t)1 << MtmNodes) - 1);
-		Mtm->clique = 0;
+		Mtm->clique = (((nodemask_t)1 << Mtm->nAllNodes) - 1); //0;
 		Mtm->refereeGrant = false;
 		Mtm->refereeWinnerId = 0;
 		Mtm->stalledNodeMask = 0;
@@ -2802,14 +2814,6 @@ static bool ConfigIsSane(void)
 	return ok;
 }
 
-static void
-MtmDmqStartup(char *sender)
-{
-	int node_id;
-	sscanf(sender, "node%d", &node_id);
-	MtmOnNodeConnect(node_id);
-}
-
 void
 _PG_init(void)
 {
@@ -3273,7 +3277,10 @@ _PG_init(void)
 	MtmMonitorInitialize();
 
 	dmq_init();
-	dmq_receiver_start_hook = MtmDmqStartup;
+	dmq_receiver_start_hook = MtmOnNodeConnect;
+	dmq_receiver_stop_hook = MtmOnNodeDisconnect;
+
+	ResolverInit();
 
 	/*
 	 * Install hooks.
@@ -3422,7 +3429,8 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 		{
 			/* Lock on us */
 			Mtm->recoverySlot = nodeId;
-			MtmPollStatusOfPreparedTransactions(false);
+			// MtmPollStatusOfPreparedTransactions(false);
+			ResolveAllTransactions();
 			MtmUnlock();
 			return REPLMODE_RECOVERY;
 		}
@@ -4651,7 +4659,7 @@ MtmGenerateGid(char* gid)
 	// MTM_LOG1("MtmGenerateGid: %s", gid);
 }
 
-static int
+int
 MtmGidParseNodeId(const char* gid)
 {
 	int MtmNodeId;

multimaster.h

@@ -79,6 +79,7 @@
 #define MULTIMASTER_BROADCAST_SERVICE    "mtm_broadcast"
 #define MULTIMASTER_ADMIN                "mtm_admin"
 #define MULTIMASTER_PRECOMMITTED         "precommitted"
+#define MULTIMASTER_PREABORTED           "preaborted"
 
 #define MULTIMASTER_DEFAULT_ARBITER_PORT 5433
 
@@ -107,9 +108,28 @@ typedef ulong64 lsn_t;
 
 typedef char pgid_t[GIDSIZE];
 
+// XXX! need rename: that's actually a disconnectivity mask
 #define SELF_CONNECTIVITY_MASK  (Mtm->nodes[MtmNodeId-1].connectivityMask)
 #define EFFECTIVE_CONNECTIVITY_MASK  ( SELF_CONNECTIVITY_MASK | Mtm->stoppedNodeMask | ~Mtm->clique )
 
+#define MTM_MAX_NODES 16
+
+typedef enum
+{
+	MtmTxUnknown		= (0<<0),
+	MtmTxNotFound		= (0<<1),
+	MtmTxInProgress		= (0<<2),
+	MtmTxPrepared		= (0<<3),
+	MtmTxPreCommited	= (0<<4),
+	MtmTxPreAborted		= (0<<5),
+	MtmTxCommited		= (0<<6),
+	MtmTxAborted		= (0<<7)
+} MtmTxState;
+
+
+
+typedef int MtmTxStateMask;
+
 typedef enum
 {
 	PGLOGICAL_COMMIT,
@@ -164,6 +184,7 @@ typedef enum
 typedef struct
 {
 	MtmMessageCode code;   /* Message code: MSG_PREPARE, MSG_PRECOMMIT, MSG_COMMIT, MSG_ABORT,... */
+	MtmTxState     state;
     int            node;   /* Sender node ID */
 	bool           lockReq;/* Whether sender node needs to lock cluster to let wal-sender caught-up and complete recovery */
 	bool           locked; /* Whether sender node is locked */
@@ -387,6 +408,7 @@ typedef struct {
 
 extern char const* const MtmNodeStatusMnem[];
 extern char const* const MtmTxnStatusMnem[];
+extern char const* const MtmTxStateMnem[];
 extern char const* const MtmMessageKindMnem[];
 
 extern MtmState* Mtm;
@@ -429,6 +451,12 @@ extern void  MtmMonitorInitialize(void);
 extern bool MtmTwoPhaseCommit(MtmCurrentTrans* x);
 extern bool MtmIsUserTransaction(void);
 extern void MtmGenerateGid(char* gid);
+extern int  MtmGidParseNodeId(const char* gid);
+
+extern void ResolverMain(void);
+extern void ResolverInit(void);
+extern void ResolveTransactionsForNode(int node_id);
+extern void ResolveAllTransactions(void);
 
 extern void  MtmStartReceivers(void);
 extern void  MtmStartReceiver(int nodeId, bool dynamic);