Reverse started hook commits

Author: knizhnik

multimaster.c

@@ -245,7 +245,6 @@ 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;
@@ -1668,8 +1667,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 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);
+	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);
 	MtmUnlock();
 }
 
@@ -2199,7 +2198,6 @@ 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;
@@ -2908,9 +2906,11 @@ void MtmReceiverStarted(int nodeId)
 			MtmEnableNode(nodeId);
 			MtmCheckQuorum();
 		}
-		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);
+		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);
+			}
 		}
 	}
 	MtmUnlock();
@@ -2997,7 +2997,6 @@ 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();
@@ -3077,19 +3076,6 @@ MtmOnProcExit(int code, Datum arg)
 	}
 }
 
-static void 
-MtmReplicationStartedHook(struct PGLogicalStartedHookArgs* args)
-{
-	MtmLock(LW_EXCLUSIVE);	
-	MtmSenderStarted = 1;
-	elog(LOG, "Start %d senders and %d receivers from %d cluster status %s", Mtm->nSenders+1, Mtm->nReceivers, Mtm->nLiveNodes-1, MtmNodeStatusMnem[Mtm->status]);
-	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)
 {
@@ -3206,9 +3192,6 @@ 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 */
@@ -3320,7 +3303,6 @@ 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;

multimaster.h

@@ -273,8 +273,7 @@ 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 intialization/recovery is completed) */
-    int    nSenders;                   /* Number of started WAL senders (used to determine moment when recovery) */
+    int    nReceivers;                 /* Number of initialized logical receivers (used to determine moment when Mtm intialization is completed */
 	int    nLockers;                   /* Number of lockers */
 	int    nActiveTransactions;        /* Nunmber of active 2PC transactions */
 	int    nConfigChanges;             /* Number of cluster configuration changes */

pglogical_config.c

@@ -477,8 +477,6 @@ 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",

pglogical_hooks.c

@@ -96,14 +96,12 @@ 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,
@@ -120,21 +118,6 @@ 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)
 {

pglogical_hooks.h

@@ -11,8 +11,6 @@ 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,

pglogical_output.c

@@ -54,7 +54,6 @@ 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);
@@ -84,7 +83,6 @@ _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;
@@ -492,15 +490,6 @@ 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.
  */

pglogical_output/hooks.h

@@ -27,12 +27,6 @@ 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
 {
@@ -69,7 +63,6 @@ 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;