multidb refactor: move receiver-related functions to logical_receiver.c, adopot new logger everywhere

Author: kelvich

src/commit.c

@@ -10,6 +10,8 @@
  */
 
 #include "postgres.h"
+#include "access/twophase.h"
+#include "access/transam.h"
 #include "storage/proc.h"
 #include "utils/guc.h"
 #include "miscadmin.h"
@@ -19,6 +21,7 @@
 #include "multimaster.h"
 #include "logger.h"
 #include "ddl.h"
+#include "state.h"
 
 static Oid		MtmDatabaseId;
 static bool		DmqSubscribed;
@@ -78,7 +81,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 		/* Reject all user's transactions at offline cluster.
 		 * Allow execution of transaction by bg-workers to makeit possible to perform recovery.
 		 */
-		MTM_ELOG(ERROR,
+		mtm_log(ERROR,
 				"Multimaster node is not online: current status %s",
 				MtmNodeStatusMnem[Mtm->status]);
 	}

src/include/logger.h

@@ -36,20 +36,46 @@ typedef enum MtmLogTag
 	/* status worker */
 	StatusRequest			= LOG,
 
+	/* ddd */
 	BgwPoolEvent			= LOG,
 
+	/* ddd */
 	DeadlockCheck			= LOG,
 	DeadlockUpdate			= LOG,
 	DeadlockSerialize		= DEBUG3,
 
+	/* ddl XXX! */
 	DMLStmtOutgoing			= DEBUG1,
 	DMLStmtIncoming			= DEBUG1,
 	DMLProcessingTrace		= DEBUG1,
 
+	/* broadcast service */
+	BroadcastNotice			= DEBUG1,
+
+	/* walsender's proto */
+	ProtoTraceFilter		= DEBUG1,
+	ProtoTraceTx			= DEBUG1,
+	ProtoTraceMode			= LOG,
+	ProtoTraceMessage		= LOG,
+
+	/* receiver */
+	MtmReceiverMode			= LOG,
+	MtmReceiverFilter		= LOG,
+	MtmApplyMessage			= LOG,
+	MtmApplyTrace			= DEBUG2,
+	MtmApplyError			= LOG,
+
+	/* state */
+	MtmStateSwitch			= LOG,
+	MtmStateMessage			= LOG
 } MtmLogTag;
 
 // XXX: also meaningful process name would be cool
 
+#define MTM_TAG "[MTM] "
+
+#define MTM_ERRMSG(fmt,...) errmsg(MTM_TAG fmt, ## __VA_ARGS__)
+
 #define mtm_log(tag, fmt, ...) ereport(tag, \
-								(errmsg("[MTM] " fmt, ## __VA_ARGS__), \
+								(errmsg(MTM_TAG fmt, ## __VA_ARGS__), \
 								errhidestmt(true), errhidecontext(true)))

src/include/multimaster.h

@@ -1,68 +1,17 @@
 #ifndef __MULTIMASTER_H__
 #define __MULTIMASTER_H__
 
+#include "postgres.h"
+
 #include "bytebuf.h"
 #include "bgwpool.h"
 #include "bkb.h"
 
-#include "access/clog.h"
-#include "pglogical_output/hooks.h"
-#include "commands/vacuum.h"
-#include "libpq-fe.h"
+#include "storage/lwlock.h"
 
 #include "dmq.h"
 #include "mm.h"
-
-#ifndef DEBUG_LEVEL
-#define DEBUG_LEVEL 0
-#endif
-
-
-#define MTM_TAG "[MTM] "
-#define MTM_ELOG(level,fmt,...) elog(level, MTM_TAG fmt, ## __VA_ARGS__)
-#define MTM_ERRMSG(fmt,...)     errmsg(MTM_TAG fmt, ## __VA_ARGS__)
-
-#if DEBUG_LEVEL == 0
-#define MTM_LOG1(fmt, ...) ereport(LOG, \
-								(errmsg("[MTM] " fmt, ## __VA_ARGS__), \
-								errhidestmt(true), errhidecontext(true)))
-
-#define MTM_LOG2(fmt, ...)
-#define MTM_LOG3(fmt, ...)
-#define MTM_LOG4(fmt, ...)
-#elif  DEBUG_LEVEL == 1
-#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG3(fmt, ...)
-#define MTM_LOG4(fmt, ...)
-#elif  DEBUG_LEVEL == 2
-#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG3(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG4(fmt, ...)
-#elif  DEBUG_LEVEL >= 3
-#define MTM_LOG1(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG2(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG3(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
-#endif
-
-// #define MTM_TXFINISH 1
-
-#ifndef MTM_TXFINISH
-#define TXFINISH(fmt, ...)
-#else
-#define TXFINISH(fmt, ...) elog(LOG, MTM_TAG "[TXFINISH] " fmt, ## __VA_ARGS__)
-#endif
-
-// #define MTM_TRACE 1
-
-#ifndef MTM_TRACE
-#define MTM_TXTRACE(tx, event, ...)
-#else
-#define MTM_TXTRACE(tx, event, ...) \
-		elog(LOG, MTM_TAG "%s, %lld, %u " event "\n", tx->gid, (long long)MtmGetSystemTime(), MyProcPid, ## __VA_ARGS__)
-#endif
+#include "resolver.h"
 
 #define MULTIMASTER_NAME                 "multimaster"
 #define MULTIMASTER_SLOT_PATTERN         "mtm_slot_%d"
@@ -86,13 +35,6 @@
 #define USEC_TO_MSEC(t) ((t)/1000)
 #define MSEC_TO_USEC(t) ((timestamp_t)(t)*1000)
 
-#define Natts_mtm_ddl_log 2
-#define Anum_mtm_ddl_log_issued		1
-#define Anum_mtm_ddl_log_query		2
-
-#define Natts_mtm_trans_state   15
-#define Natts_mtm_nodes_state   17
-#define Natts_mtm_cluster_state 20
 
 typedef ulong64 lsn_t;
 #define INVALID_LSN  InvalidXLogRecPtr
@@ -103,20 +45,6 @@ typedef ulong64 lsn_t;
 
 #define MTM_MAX_NODES 16
 
-typedef enum
-{
-	MtmTxUnknown		= (1<<0),
-	MtmTxNotFound		= (1<<1),
-	MtmTxInProgress		= (1<<2),
-	MtmTxPrepared		= (1<<3),
-	MtmTxPreCommited	= (1<<4),
-	MtmTxPreAborted		= (1<<5),
-	MtmTxCommited		= (1<<6),
-	MtmTxAborted		= (1<<7)
-} MtmTxState;
-
-typedef int MtmTxStateMask;
-
 typedef enum
 {
 	PGLOGICAL_COMMIT,
@@ -153,23 +81,13 @@ typedef enum
 	MTM_ONLINE          /* Ready to receive client's queries */
 } MtmNodeStatus;
 
-typedef enum
-{
-	REPLMODE_EXIT,         /* receiver should exit */
-	REPLMODE_RECOVERED,    /* recovery of receiver node is completed so drop old slot and restart replication from the current position in WAL */
-	REPLMODE_RECOVERY,     /* perform recovery of the node by applying all data from the slot from specified point */
-	REPLMODE_CREATE_NEW,   /* destination node is recovered: drop old slot and restart from roveredLsn position */
-	REPLMODE_OPEN_EXISTED  /* normal mode: use existed slot or create new one and start receiving data from it from the remembered position */
-} MtmReplicationMode;
-
 typedef struct
 {
 	MtmMessageCode code;   /* Message code: MSG_PREPARE, MSG_PRECOMMIT, MSG_COMMIT, MSG_ABORT,... */
 	MtmTxState     state;
     int            node;   /* Sender node ID */
 	TransactionId  dxid;   /* Transaction ID at destination node */
 	TransactionId  sxid;   /* Transaction ID at sender node */
-    XidStatus      status; /* Transaction status */
 	nodemask_t     connectivityMask; /* Connectivity bitmask at the sender of message */
 	pgid_t         gid;    /* Global transaction identifier */
 } MtmArbiterMessage;
@@ -223,28 +141,9 @@ typedef struct
 	int    nAllNodes;                  /* Total number of nodes */
 	int    recoveryCount;              /* Number of completed recoveries */
 	int    donorNodeId;                /* Cluster node from which this node was populated */
-	lsn_t recoveredLSN;                /* LSN at the moment of recovery completion */
 	MtmNodeInfo nodes[1];              /* [Mtm->nAllNodes]: per-node data */
 } MtmState;
 
-typedef struct MtmFlushPosition
-{
-	dlist_node node;
-	int        node_id;
-	lsn_t      local_end;
-	lsn_t      remote_end;
-} MtmFlushPosition;
-
-typedef struct
-{
-	int magic;
-	bool is_recovery;
-} MtmDecoderPrivate;
-
-extern char const* const MtmNodeStatusMnem[];
-extern char const* const MtmTxnStatusMnem[];
-extern char const* const MtmMessageKindMnem[];
-
 extern MtmState* Mtm;
 
 extern int   MtmReplicationNodeId;
@@ -260,8 +159,7 @@ extern bool MtmMajorNode;
 extern bool MtmBackgroundWorker;
 extern char* MtmRefereeConnStr;
 
-extern void  MtmXactCallback2(XactEvent event, void *arg);
-extern void  MtmMonitorInitialize(void);
+extern void MtmXactCallback2(XactEvent event, void *arg);
 extern bool MtmIsUserTransaction(void);
 extern void MtmGenerateGid(char *gid, TransactionId xid);
 extern int  MtmGidParseNodeId(const char* gid);
@@ -270,55 +168,23 @@ extern TransactionId MtmGidParseXid(const char* gid);
 extern void MtmWaitForExtensionCreation(void);
 extern void erase_option_from_connstr(const char *option, char *connstr);
 
-extern void ResolverMain(void);
-extern void ResolverInit(void);
-extern void ResolveTransactionsForNode(int node_id);
-extern void ResolveAllTransactions(void);
-extern char *MtmTxStateMnem(MtmTxState state);
-
-extern void  MtmStartReceivers(void);
-extern void  MtmStartReceiver(int nodeId, bool dynamic);
-
-extern MtmReplicationMode MtmGetReplicationMode(int nodeId);
-extern void  MtmExecutor(void* work, size_t size);
-
-
-
 extern void  MtmLock(LWLockMode mode);
 extern void  MtmUnlock(void);
 extern void  MtmLockNode(int nodeId, LWLockMode mode);
 extern bool  MtmTryLockNode(int nodeId, LWLockMode mode);
 extern void  MtmUnlockNode(int nodeId);
 
-extern void  MtmStopNode(int nodeId, bool dropSlot);
-extern void  MtmRecoverNode(int nodeId);
-extern void  MtmResumeNode(int nodeId);
-
 extern void  MtmSleep(timestamp_t interval);
 
 extern void  MtmSetCurrentTransactionGID(char const* gid, int node_id);
 extern void  MtmSetCurrentTransactionCSN(void);
 extern TransactionId MtmGetCurrentTransactionId(void);
 extern void MtmResetTransaction(void);
 
-extern void  MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr);
-
-extern void  MtmHandleApplyError(void);
-
-extern void  MtmUpdateLsnMapping(int nodeId, lsn_t endLsn);
-extern lsn_t MtmGetFlushPosition(int nodeId);
-
-
-extern void MtmReleaseRecoverySlot(int nodeId);
 extern PGconn *PQconnectdb_safe(const char *conninfo, int timeout);
-extern void MtmBeginSession(int nodeId);
-extern void MtmEndSession(int nodeId, bool unlock);
-
-extern bool MtmFilterTransaction(char* record, int size);
 
 extern void MtmInitMessage(MtmArbiterMessage* msg, MtmMessageCode code);
 
-extern void MtmRefereeInitialize(void);
 extern void MtmUpdateControlFile(void);
 
 extern void MtmCheckSlots(void);

src/include/pglogical_output.h

@@ -41,6 +41,12 @@ typedef struct HookFuncName
 	char    schema[NAMEDATALEN];
 } HookFuncName;
 
+typedef struct MtmDecoderPrivate
+{
+	int magic;
+	bool is_recovery;
+} MtmDecoderPrivate;
+
 typedef struct PGLogicalOutputData
 {
 	MemoryContext context;

src/include/receiver.h

@@ -0,0 +1,11 @@
+#include "multimaster.h"
+#include "mm.h"
+
+extern void MtmStartReceivers(void);
+extern void MtmStartReceiver(int nodeId, bool dynamic);
+
+extern void MtmExecutor(void* work, size_t size);
+extern void MtmUpdateLsnMapping(int node_id, lsn_t end_lsn);
+
+extern void MtmBeginSession(int nodeId);
+extern void MtmEndSession(int nodeId, bool unlock);

src/include/resolver.h

@@ -0,0 +1,24 @@
+#ifndef RESOLVER_H
+#define RESOLVER_H
+
+typedef enum
+{
+	MtmTxUnknown		= (1<<0),
+	MtmTxNotFound		= (1<<1),
+	MtmTxInProgress		= (1<<2),
+	MtmTxPrepared		= (1<<3),
+	MtmTxPreCommited	= (1<<4),
+	MtmTxPreAborted		= (1<<5),
+	MtmTxCommited		= (1<<6),
+	MtmTxAborted		= (1<<7)
+} MtmTxState;
+
+typedef int MtmTxStateMask;
+
+extern void ResolverMain(void);
+extern void ResolverInit(void);
+extern void ResolveTransactionsForNode(int node_id);
+extern void ResolveAllTransactions(void);
+extern char *MtmTxStateMnem(MtmTxState state);
+
+#endif  /* RESOLVER_H */

src/include/state.h

@@ -21,6 +21,10 @@ typedef enum
 	MTM_NONRECOVERABLE_ERROR
 } MtmEvent;
 
+extern char const* const MtmNodeStatusMnem[];
+
+extern void MtmMonitorInitialize(void);
+
 extern void MtmStateProcessNeighborEvent(int node_id, MtmNeighborEvent ev, bool locked);
 extern void MtmStateProcessEvent(MtmEvent ev, bool locked);
 extern void MtmDisableNode(int nodeId);