Add started hook to pglogical

Author: knizhnik

contrib/mmts/multimaster.c

@@ -245,6 +245,7 @@ static int   MtmMaxRecoveryLag;
 static int   MtmGcPeriod;
 static bool  MtmIgnoreTablesWithoutPk;
 static int   MtmLockCount;
+static int   MtmSenderStarted;
 
 static ExecutorStart_hook_type PreviousExecutorStartHook;
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
@@ -1667,8 +1668,8 @@ void MtmRecoveryCompleted(void)
 		Mtm->nodes[i].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	}
 	/* Mode will be changed to online once all logical receiver are connected */
-	elog(LOG, "Recovery completed with %d active receivers from %d", Mtm->nReceivers, Mtm->nLiveNodes-1);
-	MtmSwitchClusterMode(Mtm->nReceivers == Mtm->nLiveNodes-1 ? MTM_ONLINE : MTM_CONNECTED);
+	elog(LOG, "Recovery completed with %d active receivers and %d started senders from %d", Mtm->nReceivers, Mtm->nSenders, Mtm->nLiveNodes-1);
+	MtmSwitchClusterMode(Mtm->nReceivers == Mtm->nLiveNodes-1 && Mtm->nSenders == Mtm->nLiveNodes-1 ? MTM_ONLINE : MTM_CONNECTED);
 	MtmUnlock();
 }
 
@@ -2198,6 +2199,7 @@ static void MtmInitialize()
         Mtm->transListHead = NULL;
         Mtm->transListTail = &Mtm->transListHead;		
         Mtm->nReceivers = 0;
+        Mtm->nSenders = 0;
 		Mtm->timeShift = 0;		
 		Mtm->transCount = 0;
 		Mtm->gcCount = 0;
@@ -2906,11 +2908,9 @@ void MtmReceiverStarted(int nodeId)
 			MtmEnableNode(nodeId);
 			MtmCheckQuorum();
 		}
-		elog(LOG, "Start %d receivers from %d cluster status %s", Mtm->nReceivers+1, Mtm->nLiveNodes-1, MtmNodeStatusMnem[Mtm->status]);
-		if (++Mtm->nReceivers == Mtm->nLiveNodes-1) {
-			if (Mtm->status == MTM_CONNECTED) { 
-				MtmSwitchClusterMode(MTM_ONLINE);
-			}
+		elog(LOG, "Start %d receivers and %d senders from %d cluster status %s", Mtm->nReceivers+1, Mtm->nSenders, Mtm->nLiveNodes-1, MtmNodeStatusMnem[Mtm->status]);
+		if (++Mtm->nReceivers == Mtm->nLiveNodes-1 && Mtm->nSenders == Mtm->nLiveNodes-1 && Mtm->status == MTM_CONNECTED) { 
+			MtmSwitchClusterMode(MTM_ONLINE);
 		}
 	}
 	MtmUnlock();
@@ -2997,6 +2997,7 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 				elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
 				Mtm->recoverySlot = nodeId;
 				Mtm->nReceivers = 0;
+				Mtm->nSenders = 0;
 				Mtm->recoveryCount += 1;
 				Mtm->pglogicalNodeMask = 0;
 				MtmUnlock();
@@ -3076,6 +3077,18 @@ MtmOnProcExit(int code, Datum arg)
 	}
 }
 
+static void 
+MtmReplicationStartedHook(struct PGLogicalStartedHookArgs* args)
+{
+	MtmLock(LW_EXCLUSIVE);	
+	MtmSenderStarted = 1;
+	if (++Mtm->nSenders == Mtm->nLiveNodes-1 && Mtm->nReceivers == Mtm->nLiveNodes-1 && Mtm->status == MTM_CONNECTED) { 
+		MtmSwitchClusterMode(MTM_ONLINE);
+	}
+	MtmUnlock();	
+}
+
+
 static void 
 MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 {
@@ -3192,6 +3205,9 @@ static void
 MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
 {
 	if (MtmReplicationNodeId >= 0) { 
+		MtmLock(LW_EXCLUSIVE);
+		Mtm->nSenders -= MtmSenderStarted;
+		MtmUnlock();
 		MTM_LOG1("Logical replication to node %d is stopped", MtmReplicationNodeId); 
 		/* MtmOnNodeDisconnect(MtmReplicationNodeId); */
 		MtmReplicationNodeId = -1; /* defuse on_proc_exit hook */
@@ -3303,6 +3319,7 @@ bool MtmFilterTransaction(char* record, int size)
 void MtmSetupReplicationHooks(struct PGLogicalHooks* hooks)
 {
 	hooks->startup_hook = MtmReplicationStartupHook;
+	hooks->started_hook = MtmReplicationStartedHook;
 	hooks->shutdown_hook = MtmReplicationShutdownHook;
 	hooks->txn_filter_hook = MtmReplicationTxnFilterHook;
 	hooks->row_filter_hook = MtmReplicationRowFilterHook;

contrib/mmts/multimaster.h

@@ -273,7 +273,8 @@ typedef struct
 	int    inject2PCError;             /* Simulate error during 2PC commit at this node */
     int    nLiveNodes;                 /* Number of active nodes */
     int    nAllNodes;                  /* Total numbber of nodes */
-    int    nReceivers;                 /* Number of initialized logical receivers (used to determine moment when Mtm intialization is completed */
+    int    nReceivers;                 /* Number of initialized logical receivers (used to determine moment when intialization/recovery is completed) */
+    int    nSenders;                   /* Number of started WAL senders (used to determine moment when recovery) */
 	int    nLockers;                   /* Number of lockers */
 	int    nActiveTransactions;        /* Nunmber of active 2PC transactions */
 	int    nConfigChanges;             /* Number of cluster configuration changes */

contrib/mmts/pglogical_config.c

@@ -477,6 +477,8 @@ prepare_startup_message(PGLogicalOutputData *data)
 	 */
 	l = add_startup_msg_b(l, "hooks.startup_hook_enabled",
 			data->hooks.startup_hook != NULL);
+	l = add_startup_msg_b(l, "hooks.started_hook_enabled",
+			data->hooks.started_hook != NULL);
 	l = add_startup_msg_b(l, "hooks.shutdown_hook_enabled",
 			data->hooks.shutdown_hook != NULL);
 	l = add_startup_msg_b(l, "hooks.row_filter_enabled",

contrib/mmts/pglogical_hooks.c

@@ -96,12 +96,14 @@ load_hooks(PGLogicalOutputData *data)
 
 		elog(DEBUG3, "pglogical_output: Loaded hooks from function %u. Hooks are: \n"
 				"\tstartup_hook: %p\n"
+				"\tstarted_hook: %p\n"
 				"\tshutdown_hook: %p\n"
 				"\trow_filter_hook: %p\n"
 				"\ttxn_filter_hook: %p\n"
 				"\thooks_private_data: %p\n",
 				hooks_func,
 				data->hooks.startup_hook,
+				data->hooks.started_hook,
 				data->hooks.shutdown_hook,
 				data->hooks.row_filter_hook,
 				data->hooks.txn_filter_hook,
@@ -118,6 +120,21 @@ load_hooks(PGLogicalOutputData *data)
 		CommitTransactionCommand();
 }
 
+void
+call_started_hook(PGLogicalOutputData *data)
+{
+	struct PGLogicalStartedHookArgs args;
+	MemoryContext old_ctxt;
+
+	if (data->hooks.started_hook != NULL)
+	{
+		args.private_data = data->hooks.hooks_private_data;
+		old_ctxt = MemoryContextSwitchTo(data->hooks_mctxt);
+		(void) (*data->hooks.started_hook)(&args);
+		MemoryContextSwitchTo(old_ctxt);
+	}
+}
+	
 void
 call_startup_hook(PGLogicalOutputData *data, List *plugin_params)
 {

contrib/mmts/pglogical_hooks.h

@@ -11,6 +11,8 @@ extern void load_hooks(PGLogicalOutputData *data);
 
 extern void call_startup_hook(PGLogicalOutputData *data, List *plugin_params);
 
+extern void call_started_hook(PGLogicalOutputData *data);
+
 extern void call_shutdown_hook(PGLogicalOutputData *data);
 
 extern bool call_row_filter_hook(PGLogicalOutputData *data,

contrib/mmts/pglogical_output.c

@@ -54,6 +54,7 @@ extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
 /* These must be available to pg_dlsym() */
 static void pg_decode_startup(LogicalDecodingContext * ctx,
 							  OutputPluginOptions *opt, bool is_init);
+static void pg_decode_started(LogicalDecodingContext * ctx);
 static void pg_decode_shutdown(LogicalDecodingContext * ctx);
 static void pg_decode_begin_txn(LogicalDecodingContext *ctx,
 					ReorderBufferTXN *txn);
@@ -83,6 +84,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 	AssertVariableIsOfType(&_PG_output_plugin_init, LogicalOutputPluginInit);
 
 	cb->startup_cb = pg_decode_startup;
+	cb->started_cb = pg_decode_started;
 	cb->begin_cb = pg_decode_begin_txn;
 	cb->change_cb = pg_decode_change;
 	cb->commit_cb = pg_decode_commit_txn;
@@ -490,6 +492,15 @@ pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	MemoryContextReset(data->context);
 }
 
+static void
+pg_decode_started(LogicalDecodingContext * ctx)
+{
+	PGLogicalOutputData *data = ctx->output_plugin_private;
+	call_started_hook(data);
+}
+	
+
+
 /*
  * Decide if the whole transaction with specific origin should be filtered out.
  */

contrib/mmts/pglogical_output/hooks.h

@@ -27,6 +27,12 @@ struct PGLogicalStartupHookArgs
 
 typedef void (*pglogical_startup_hook_fn)(struct PGLogicalStartupHookArgs *args);
 
+struct PGLogicalStartedHookArgs
+{
+	void	   *private_data;
+};
+
+typedef void (*pglogical_started_hook_fn)(struct PGLogicalStartedHookArgs *args);
 
 struct PGLogicalTxnFilterArgs
 {
@@ -63,6 +69,7 @@ typedef void (*pglogical_shutdown_hook_fn)(struct PGLogicalShutdownHookArgs *arg
 struct PGLogicalHooks
 {
 	pglogical_startup_hook_fn startup_hook;
+	pglogical_started_hook_fn started_hook;
 	pglogical_shutdown_hook_fn shutdown_hook;
 	pglogical_txn_filter_hook_fn txn_filter_hook;
 	pglogical_row_filter_hook_fn row_filter_hook;

contrib/multimaster/pglogical_output.c

@@ -76,6 +76,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 	AssertVariableIsOfType(&_PG_output_plugin_init, LogicalOutputPluginInit);
 
 	cb->startup_cb = pg_decode_startup;
+	cb->started_cb = pg_decode_started;
 	cb->begin_cb = pg_decode_begin_txn;
 	cb->change_cb = pg_decode_change;
 	cb->commit_cb = pg_decode_commit_txn;

src/backend/replication/logical/logical.c

@@ -54,6 +54,7 @@ typedef struct LogicalErrorCallbackState
 
 /* wrappers around output plugin callbacks */
 static void output_plugin_error_callback(void *arg);
+static void started_cb_wrapper(LogicalDecodingContext *ctx);
 static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
 				   bool is_init);
 static void shutdown_cb_wrapper(LogicalDecodingContext *ctx);
@@ -413,6 +414,7 @@ DecodingContextReady(LogicalDecodingContext *ctx)
 void
 DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
 {
+	MemoryContext old_context;
 	XLogRecPtr	startptr;
 
 	/* Initialize from where to start reading WAL. */
@@ -447,6 +449,11 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
 	}
 
 	ctx->slot->data.confirmed_flush = ctx->reader->EndRecPtr;
+
+	old_context = MemoryContextSwitchTo(ctx->context);
+	if (ctx->callbacks.started_cb != NULL)
+		started_cb_wrapper(ctx);
+	MemoryContextSwitchTo(old_context);
 }
 
 /*
@@ -562,6 +569,31 @@ startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool i
 	error_context_stack = errcallback.previous;
 }
 
+static void
+started_cb_wrapper(LogicalDecodingContext *ctx)
+{
+	LogicalErrorCallbackState state;
+	ErrorContextCallback errcallback;
+
+	/* Push callback + info on the error context stack */
+	state.ctx = ctx;
+	state.callback_name = "startup";
+	state.report_location = InvalidXLogRecPtr;
+	errcallback.callback = output_plugin_error_callback;
+	errcallback.arg = (void *) &state;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
+	/* set output state */
+	ctx->accept_writes = false;
+
+	/* do the actual work: call callback */
+	ctx->callbacks.started_cb(ctx);
+
+	/* Pop the error context stack */
+	error_context_stack = errcallback.previous;
+}
+
 static void
 shutdown_cb_wrapper(LogicalDecodingContext *ctx)
 {

src/include/replication/output_plugin.h

@@ -47,6 +47,15 @@ typedef void (*LogicalDecodeStartupCB) (
 													bool is_init
 );
 
+/*
+ * Callback that gets called when WAL-sender is started. ctx->private_data can
+ * be set to some private data.
+ *
+ */
+typedef void (*LogicalDecodeStartedCB) (
+										  struct LogicalDecodingContext *ctx
+);
+
 /*
  * Callback called for every (explicit or implicit) BEGIN of a successful
  * transaction.
@@ -105,6 +114,7 @@ typedef void (*LogicalDecodeShutdownCB) (
 typedef struct OutputPluginCallbacks
 {
 	LogicalDecodeStartupCB startup_cb;
+	LogicalDecodeStartedCB started_cb;
 	LogicalDecodeBeginCB begin_cb;
 	LogicalDecodeChangeCB change_cb;
 	LogicalDecodeCommitCB commit_cb;