Intoduce syncpoints instead of replication session

Author: kelvich

Makefile

@@ -4,7 +4,7 @@ OBJS = src/multimaster.o src/dmq.o src/commit.o src/bytebuf.o src/bgwpool.o \
 src/pglogical_output.o src/pglogical_proto.o src/pglogical_receiver.o \
 src/pglogical_apply.o src/pglogical_hooks.o src/pglogical_config.o \
 src/pglogical_relid_map.o src/ddd.o src/bkb.o src/spill.o src/state.o \
-src/resolver.o src/ddl.o
+src/resolver.o src/ddl.o src/syncpoint.o
 MODULE_big = multimaster
 
 ifdef USE_PGXS

multimaster--1.0.sql

@@ -137,9 +137,23 @@ CREATE FUNCTION mtm.check_deadlock(xid bigint) RETURNS boolean
 AS 'MODULE_PATHNAME','mtm_check_deadlock'
 LANGUAGE C;
 
-CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema name, rel_name name, primary key(rel_schema, rel_name));
+CREATE TABLE mtm.local_tables(
+    rel_schema name,
+    rel_name name,
+    primary key(rel_schema, rel_name)
+);
+
+CREATE TABLE mtm.referee_decision(
+    key text primary key not null,
+    node_id int
+);
 
-CREATE TABLE mtm.referee_decision(key text primary key not null, node_id int);
+CREATE TABLE mtm.syncpoints(
+    node_id int not null,
+    origin_lsn bigint not null,
+    local_lsn  bigint not null,
+    primary key(node_id, origin_lsn)
+);
 
 CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS
 $$

src/bgwpool.c

@@ -18,6 +18,8 @@
 #include "mm.h"
 #include "logger.h"
 
+bool MtmIsPoolWorker;
+
 bool MtmIsLogicalReceiver;
 int  MtmMaxWorkers;
 
@@ -50,6 +52,9 @@ BgwPoolMainLoop(BgwPool* pool)
 
 	mtm_log(BgwPoolEvent, "Start background worker %d, shutdown=%d", MyProcPid, pool->shutdown);
 
+	MtmIsPoolWorker = true;
+
+	// XXX: get rid of that
 	MtmBackgroundWorker = true;
 	MtmIsLogicalReceiver = true;
 	MtmPool = pool;
@@ -74,6 +79,8 @@ BgwPoolMainLoop(BgwPool* pool)
 		}
 
 		PGSemaphoreLock(pool->available);
+
+		// XXX: change to LWLock
 		SpinLockAcquire(&pool->lock);
 		if (pool->shutdown)
 		{
@@ -123,6 +130,8 @@ BgwPoolMainLoop(BgwPool* pool)
 		pool->active -= 1;
 		pool->lastPeakTime = 0;
 		SpinLockRelease(&pool->lock);
+
+		ConditionVariableBroadcast(&pool->syncpoint_cv);
 	}
 
 	SpinLockRelease(&pool->lock);
@@ -144,6 +153,7 @@ void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  c
 	PGSemaphoreReset(pool->available);
 	PGSemaphoreReset(pool->overflow);
     SpinLockInit(&pool->lock);
+	ConditionVariableInit(&pool->syncpoint_cv);
 	pool->shutdown = false;
     pool->producerBlocked = false;
     pool->head = 0;

src/commit.c

@@ -22,6 +22,7 @@
 #include "logger.h"
 #include "ddl.h"
 #include "state.h"
+#include "syncpoint.h"
 
 static Oid		MtmDatabaseId;
 static bool		DmqSubscribed;
@@ -210,6 +211,8 @@ MtmTwoPhaseCommit(MtmCurrentTrans* x)
 	dmq_stream_unsubscribe(stream);
 	mtm_log(MtmTxTrace, "%s unsubscribed for %s", gid, stream);
 
+	MaybeLogSyncpoint();
+
 	return true;
 }
 

src/include/bgwpool.h

@@ -5,6 +5,7 @@
 #include "storage/spin.h"
 #include "storage/pg_sema.h"
 #include "postmaster/bgworker.h"
+#include "storage/condition_variable.h"
 #include "bkb.h" // XXX
 
 #include "mm.h"
@@ -30,6 +31,7 @@ typedef struct
     volatile slock_t lock;
 	PGSemaphore available;
 	PGSemaphore overflow;
+	ConditionVariable syncpoint_cv;
     size_t head;
     size_t tail;
     size_t size;

src/include/logger.h

@@ -60,16 +60,20 @@ typedef enum MtmLogTag
 	ProtoTraceState			= LOG,
 
 	/* receiver */
-	MtmReceiverMode			= LOG,
-	MtmReceiverFilter		= LOG,
+	MtmReceiverStart		= LOG,
+	MtmReceiverFilter		= DEBUG2,
 	MtmApplyMessage			= LOG,
 	MtmApplyTrace			= DEBUG2,
 	MtmApplyError			= LOG,
 	MtmApplyBgwFinish		= LOG,
 
 	/* state */
 	MtmStateSwitch			= LOG,
-	MtmStateMessage			= LOG
+	MtmStateMessage			= LOG,
+
+	/* syncpoints */
+	SyncpointCreated		= LOG,
+	SyncpointApply			= LOG,
 } MtmLogTag;
 
 // XXX: also meaningful process name would be cool

src/include/mm.h

@@ -51,6 +51,8 @@ typedef struct
 	bool is_recovery;
 	bool parallel_allowed;
 	TimestampTz session_id;
+	// XXX
+	XLogRecPtr end_lsn;
 } MtmReceiverContext;
 
 /* XXX: drop that */
@@ -66,6 +68,9 @@ typedef ulong64 nodemask_t;
 #define ALL_BITS ((nodemask_t)~0)
 
 extern bool				MtmIsLogicalReceiver;
+extern bool				MtmIsReceiver;
+extern bool				MtmIsPoolWorker;
+
 extern bool				MtmBackgroundWorker;
 extern int				MtmMaxNodes;
 extern int				MtmNodeId;

src/include/multimaster.h

@@ -15,6 +15,7 @@
 
 #define MULTIMASTER_NAME                 "multimaster"
 #define MULTIMASTER_SLOT_PATTERN         "mtm_slot_%d"
+#define MULTIMASTER_RECOVERY_SLOT_PATTERN "mtm_recovery_slot_%d"
 #define MULTIMASTER_MIN_PROTO_VERSION    1
 #define MULTIMASTER_MAX_PROTO_VERSION    1
 #define MULTIMASTER_MAX_GID_SIZE         42
@@ -27,6 +28,7 @@
 #define MULTIMASTER_ADMIN                "mtm_admin"
 #define MULTIMASTER_PRECOMMITTED         "precommitted"
 #define MULTIMASTER_PREABORTED           "preaborted"
+#define MULTIMASTER_SYNCPOINT_INTERVAL    10*1024*1024
 
 #define MULTIMASTER_DEFAULT_ARBITER_PORT 5433
 
@@ -128,6 +130,7 @@ typedef struct
 	bool extension_created;
 	bool stop_new_commits;
 	bool recovered;
+	XLogRecPtr latestSyncpoint;
 	MtmNodeStatus status;              /* Status of this node */
 	char *statusReason;                /* A human-readable description of why the current status was set */
 	int recoverySlot;                  /* NodeId of recovery slot or 0 if none */
@@ -164,6 +167,7 @@ extern char* MtmRefereeConnStr;
 
 extern LWLock *MtmCommitBarrier;
 extern LWLock *MtmReceiverBarrier;
+extern LWLock *MtmSyncpointLock;
 
 extern void MtmXactCallback2(XactEvent event, void *arg);
 extern bool MtmIsUserTransaction(void);

src/include/pglogical_proto.h

@@ -22,6 +22,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, LogicalDecodingContext *ctx,
+					XLogRecPtr end_lsn,
 					const char *prefix, Size sz, const char *message);
 typedef void (*pglogical_write_commit_fn)(StringInfo out, struct PGLogicalOutputData *data,
 							 ReorderBufferTXN *txn, XLogRecPtr commit_lsn);

src/include/syncpoint.h

@@ -0,0 +1,38 @@
+/*-------------------------------------------------------------------------
+ *
+ * syncpoint.h
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SYNCPOINT_H
+#define SYNCPOINT_H
+
+#include "access/xlogdefs.h"
+#include "libpq-fe.h"
+#include "utils/hsearch.h"
+#include "replication/walsender.h"
+
+typedef struct
+{
+	XLogRecPtr	origin_lsn;
+	XLogRecPtr	local_lsn;
+} Syncpoint;
+
+typedef struct
+{
+	int64		node_id;
+	XLogRecPtr	origin_lsn;
+} FilterEntry;
+
+extern void MaybeLogSyncpoint(void);
+extern void SyncpointRegister(int node_id, XLogRecPtr origin_lsn,
+							  XLogRecPtr local_lsn, XLogRecPtr trim_lsn);
+extern Syncpoint SyncpointGetLatest(int node_id);
+extern Syncpoint *SyncpointGetAllLatest(void);
+extern XLogRecPtr QueryRecoveryHorizon(PGconn *conn, int node_id, Syncpoint *local_spvector);
+extern HTAB *RecoveryFilterLoad(int filter_node_id, Syncpoint *spvector);
+
+#endif							/* SYNCPOINT_H */

src/multimaster.c

@@ -125,6 +125,7 @@ static TransactionManager MtmTM =
 
 LWLock *MtmCommitBarrier;
 LWLock *MtmReceiverBarrier;
+LWLock *MtmSyncpointLock;
 
 bool  MtmDoReplication;
 char* MtmDatabaseName;
@@ -581,6 +582,7 @@ static void MtmInitialize()
 		Mtm->pglogicalSenderMask = 0;
 		Mtm->recoveryCount = 0;
 		Mtm->localTablesHashLoaded = false;
+		Mtm->latestSyncpoint = InvalidXLogRecPtr;
 
 		for (i = 0; i < MtmMaxNodes; i++)
 		{
@@ -595,6 +597,9 @@ static void MtmInitialize()
 		}
 		Mtm->nodes[MtmNodeId-1].originId = DoNotReplicateId;
 		/* All transaction originated from the current node should be ignored during recovery */
+
+
+		// XXX: not used anymore
 		Mtm->nodes[MtmNodeId-1].restartLSN = (lsn_t)PG_UINT64_MAX;
 
 		MtmTx.xid = InvalidTransactionId;
@@ -604,6 +609,7 @@ static void MtmInitialize()
 
 	MtmCommitBarrier = &(GetNamedLWLockTranche(MULTIMASTER_NAME)[MtmMaxNodes*2+1].lock);
 	MtmReceiverBarrier = &(GetNamedLWLockTranche(MULTIMASTER_NAME)[MtmMaxNodes*2+2].lock);
+	MtmSyncpointLock = &(GetNamedLWLockTranche(MULTIMASTER_NAME)[MtmMaxNodes*2+3].lock);
 
 	MtmDoReplication = true;
 	TM = &MtmTM;
@@ -1191,7 +1197,7 @@ _PG_init(void)
 	 * resources in mtm_shmem_startup().
 	 */
 	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmMaxNodes*MtmQueueSize + sizeof(MtmTime));
-	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2 + 2);
+	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2 + 3);
 
 	MtmMonitorInitialize();