use-main-command-loop-in-walsender-1.patch

*** a/src/backend/replication/basebackup.c
--- b/src/backend/replication/basebackup.c
***************
*** 22,27 ****
--- 22,28 ----
  #include "lib/stringinfo.h"
  #include "libpq/libpq.h"
  #include "libpq/pqformat.h"
+ #include "miscadmin.h"
  #include "nodes/pg_list.h"
  #include "replication/basebackup.h"
  #include "replication/walsender.h"
***************
*** 30,36 ****
  #include "storage/ipc.h"
  #include "utils/builtins.h"
  #include "utils/elog.h"
- #include "utils/memutils.h"
  #include "utils/ps_status.h"
  
  typedef struct
--- 31,36 ----
***************
*** 370,388 **** void
  SendBaseBackup(BaseBackupCmd *cmd)
  {
  	DIR		   *dir;
- 	MemoryContext backup_context;
- 	MemoryContext old_context;
  	basebackup_options opt;
  
  	parse_basebackup_options(cmd->options, &opt);
  
- 	backup_context = AllocSetContextCreate(CurrentMemoryContext,
- 										   "Streaming base backup context",
- 										   ALLOCSET_DEFAULT_MINSIZE,
- 										   ALLOCSET_DEFAULT_INITSIZE,
- 										   ALLOCSET_DEFAULT_MAXSIZE);
- 	old_context = MemoryContextSwitchTo(backup_context);
- 
  	WalSndSetState(WALSNDSTATE_BACKUP);
  
  	if (update_process_title)
--- 370,379 ----
***************
*** 403,411 **** SendBaseBackup(BaseBackupCmd *cmd)
  	perform_base_backup(&opt, dir);
  
  	FreeDir(dir);
- 
- 	MemoryContextSwitchTo(old_context);
- 	MemoryContextDelete(backup_context);
  }
  
  static void
--- 394,399 ----
***************
*** 606,612 **** sendDir(char *path, int basepathlen, bool sizeonly)
  		 * error in that case. The error handler further up will call
  		 * do_pg_abort_backup() for us.
  		 */
! 		if (walsender_shutdown_requested || walsender_ready_to_stop)
  			ereport(ERROR,
  				(errmsg("shutdown requested, aborting active base backup")));
  
--- 594,600 ----
  		 * error in that case. The error handler further up will call
  		 * do_pg_abort_backup() for us.
  		 */
! 		if (ProcDiePending || walsender_ready_to_stop)
  			ereport(ERROR,
  				(errmsg("shutdown requested, aborting active base backup")));
  
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 113,133 **** static TimestampTz last_reply_timestamp;
  
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
- volatile sig_atomic_t walsender_shutdown_requested = false;
  volatile sig_atomic_t walsender_ready_to_stop = false;
  
  /* Signal handlers */
  static void WalSndSigHupHandler(SIGNAL_ARGS);
- static void WalSndShutdownHandler(SIGNAL_ARGS);
- static void WalSndQuickDieHandler(SIGNAL_ARGS);
  static void WalSndXLogSendHandler(SIGNAL_ARGS);
  static void WalSndLastCycleHandler(SIGNAL_ARGS);
  
  /* Prototypes for private functions */
- static bool HandleReplicationCommand(const char *cmd_string);
  static void WalSndLoop(void) __attribute__((noreturn));
! static void InitWalSnd(void);
! static void WalSndHandshake(void);
  static void WalSndKill(int code, Datum arg);
  static void XLogSend(char *msgbuf, bool *caughtup);
  static void IdentifySystem(void);
--- 113,128 ----
  
  /* Flags set by signal handlers for later service in main loop */
  static volatile sig_atomic_t got_SIGHUP = false;
  volatile sig_atomic_t walsender_ready_to_stop = false;
  
  /* Signal handlers */
  static void WalSndSigHupHandler(SIGNAL_ARGS);
  static void WalSndXLogSendHandler(SIGNAL_ARGS);
  static void WalSndLastCycleHandler(SIGNAL_ARGS);
  
  /* Prototypes for private functions */
  static void WalSndLoop(void) __attribute__((noreturn));
! static void InitWalSenderSlot(void);
  static void WalSndKill(int code, Datum arg);
  static void XLogSend(char *msgbuf, bool *caughtup);
  static void IdentifySystem(void);
***************
*** 139,284 **** static void ProcessRepliesIfAny(void);
  static void WalSndKeepalive(char *msgbuf);
  
  
! /* Main entry point for walsender process */
  void
! WalSenderMain(void)
  {
- 	MemoryContext walsnd_context;
- 
  	am_cascading_walsender = RecoveryInProgress();
  
  	/* Create a per-walsender data structure in shared memory */
! 	InitWalSnd();
! 
! 	/*
! 	 * Create a memory context that we will do all our work in.  We do this so
! 	 * that we can reset the context during error recovery and thereby avoid
! 	 * possible memory leaks.  Formerly this code just ran in
! 	 * TopMemoryContext, but resetting that would be a really bad idea.
! 	 *
! 	 * XXX: we don't actually attempt error recovery in walsender, we just
! 	 * close the connection and exit.
! 	 */
! 	walsnd_context = AllocSetContextCreate(TopMemoryContext,
! 										   "Wal Sender",
! 										   ALLOCSET_DEFAULT_MINSIZE,
! 										   ALLOCSET_DEFAULT_INITSIZE,
! 										   ALLOCSET_DEFAULT_MAXSIZE);
! 	MemoryContextSwitchTo(walsnd_context);
  
  	/* Set up resource owner */
  	CurrentResourceOwner = ResourceOwnerCreate(NULL, "walsender top-level resource owner");
  
- 	/* Unblock signals (they were blocked when the postmaster forked us) */
- 	PG_SETMASK(&UnBlockSig);
- 
  	/*
  	 * Use the recovery target timeline ID during recovery
  	 */
  	if (am_cascading_walsender)
  		ThisTimeLineID = GetRecoveryTargetTLI();
- 
- 	/* Tell the standby that walsender is ready for receiving commands */
- 	ReadyForQuery(DestRemote);
- 
- 	/* Handle handshake messages before streaming */
- 	WalSndHandshake();
- 
- 	/* Initialize shared memory status */
- 	{
- 		/* use volatile pointer to prevent code rearrangement */
- 		volatile WalSnd *walsnd = MyWalSnd;
- 
- 		SpinLockAcquire(&walsnd->mutex);
- 		walsnd->sentPtr = sentPtr;
- 		SpinLockRelease(&walsnd->mutex);
- 	}
- 
- 	SyncRepInitConfig();
- 
- 	/* Main loop of walsender */
- 	WalSndLoop();
  }
  
  /*
!  * Execute commands from walreceiver, until we enter streaming mode.
   */
! static void
! WalSndHandshake(void)
  {
! 	StringInfoData input_message;
! 	bool		replication_started = false;
! 
! 	initStringInfo(&input_message);
! 
! 	while (!replication_started)
  	{
! 		int			firstchar;
! 
! 		WalSndSetState(WALSNDSTATE_STARTUP);
! 		set_ps_display("idle", false);
! 
! 		/* Wait for a command to arrive */
! 		firstchar = pq_getbyte();
! 
! 		/*
! 		 * Emergency bailout if postmaster has died.  This is to avoid the
! 		 * necessity for manual cleanup of all postmaster children.
! 		 */
! 		if (!PostmasterIsAlive())
! 			exit(1);
! 
! 		/*
! 		 * Check for any other interesting events that happened while we
! 		 * slept.
! 		 */
! 		if (got_SIGHUP)
! 		{
! 			got_SIGHUP = false;
! 			ProcessConfigFile(PGC_SIGHUP);
! 		}
! 
! 		if (firstchar != EOF)
! 		{
! 			/*
! 			 * Read the message contents. This is expected to be done without
! 			 * blocking because we've been able to get message type code.
! 			 */
! 			if (pq_getmessage(&input_message, 0))
! 				firstchar = EOF;	/* suitable message already logged */
! 		}
! 
! 		/* Handle the very limited subset of commands expected in this phase */
! 		switch (firstchar)
! 		{
! 			case 'Q':			/* Query message */
! 				{
! 					const char *query_string;
! 
! 					query_string = pq_getmsgstring(&input_message);
! 					pq_getmsgend(&input_message);
! 
! 					if (HandleReplicationCommand(query_string))
! 						replication_started = true;
! 				}
! 				break;
! 
! 			case 'X':
! 				/* standby is closing the connection */
! 				proc_exit(0);
! 
! 			case EOF:
! 				/* standby disconnected unexpectedly */
! 				ereport(COMMERROR,
! 						(errcode(ERRCODE_PROTOCOL_VIOLATION),
! 						 errmsg("unexpected EOF on standby connection")));
! 				proc_exit(0);
! 
! 			default:
! 				ereport(FATAL,
! 						(errcode(ERRCODE_PROTOCOL_VIOLATION),
! 						 errmsg("invalid standby handshake message type %d", firstchar)));
! 		}
  	}
  }
  
--- 134,172 ----
  static void WalSndKeepalive(char *msgbuf);
  
  
! /* Initialize walsender process before entering the main command loop */
  void
! InitWalSender(void)
  {
  	am_cascading_walsender = RecoveryInProgress();
  
  	/* Create a per-walsender data structure in shared memory */
! 	InitWalSenderSlot();
  
  	/* Set up resource owner */
  	CurrentResourceOwner = ResourceOwnerCreate(NULL, "walsender top-level resource owner");
  
  	/*
  	 * Use the recovery target timeline ID during recovery
  	 */
  	if (am_cascading_walsender)
  		ThisTimeLineID = GetRecoveryTargetTLI();
  }
  
  /*
!  * Clean up after an error.
!  *
!  * WAL sender processes don't use transactions like regular backends do.
!  * This should do any cleanup required in a WAL sender process, similar to
!  * what transaction abort does in a regular backend.
   */
! void
! WalSndErrorCleanup()
  {
! 	if (sendFile >= 0)
  	{
! 		close(sendFile);
! 		sendFile = -1;
  	}
  }
  
***************
*** 350,364 **** IdentifySystem(void)
  	pq_sendbytes(&buf, (char *) xpos, strlen(xpos));
  
  	pq_endmessage(&buf);
- 
- 	/* Send CommandComplete and ReadyForQuery messages */
- 	EndCommand("SELECT", DestRemote);
- 	ReadyForQuery(DestRemote);
- 	/* ReadyForQuery did pq_flush for us */
  }
  
  /*
!  * START_REPLICATION
   */
  static void
  StartReplication(StartReplicationCmd *cmd)
--- 238,250 ----
  	pq_sendbytes(&buf, (char *) xpos, strlen(xpos));
  
  	pq_endmessage(&buf);
  }
  
  /*
!  * Handle START_REPLICATION command.
!  *
!  * At the moment, this never returns, but an ereport(ERROR) will take us back
!  * to the main loop.
   */
  static void
  StartReplication(StartReplicationCmd *cmd)
***************
*** 435,449 **** StartReplication(StartReplicationCmd *cmd)
  	 * be shipped from that position
  	 */
  	sentPtr = cmd->startpoint;
  }
  
  /*
   * Execute an incoming replication command.
   */
! static bool
! HandleReplicationCommand(const char *cmd_string)
  {
- 	bool		replication_started = false;
  	int			parse_rc;
  	Node	   *cmd_node;
  	MemoryContext cmd_context;
--- 321,349 ----
  	 * be shipped from that position
  	 */
  	sentPtr = cmd->startpoint;
+ 
+ 	/* Also update the start position status in shared memory */
+ 	{
+ 		/* use volatile pointer to prevent code rearrangement */
+ 		volatile WalSnd *walsnd = MyWalSnd;
+ 
+ 		SpinLockAcquire(&walsnd->mutex);
+ 		walsnd->sentPtr = sentPtr;
+ 		SpinLockRelease(&walsnd->mutex);
+ 	}
+ 
+ 	SyncRepInitConfig();
+ 
+ 	/* Main loop of walsender */
+ 	WalSndLoop();
  }
  
  /*
   * Execute an incoming replication command.
   */
! void
! exec_replication_command(const char *cmd_string)
  {
  	int			parse_rc;
  	Node	   *cmd_node;
  	MemoryContext cmd_context;
***************
*** 451,456 **** HandleReplicationCommand(const char *cmd_string)
--- 351,358 ----
  
  	elog(DEBUG1, "received replication command: %s", cmd_string);
  
+ 	CHECK_FOR_INTERRUPTS();
+ 
  	cmd_context = AllocSetContextCreate(CurrentMemoryContext,
  										"Replication command context",
  										ALLOCSET_DEFAULT_MINSIZE,
***************
*** 476,493 **** HandleReplicationCommand(const char *cmd_string)
  
  		case T_StartReplicationCmd:
  			StartReplication((StartReplicationCmd *) cmd_node);
- 
- 			/* break out of the loop */
- 			replication_started = true;
  			break;
  
  		case T_BaseBackupCmd:
  			SendBaseBackup((BaseBackupCmd *) cmd_node);
- 
- 			/* Send CommandComplete and ReadyForQuery messages */
- 			EndCommand("SELECT", DestRemote);
- 			ReadyForQuery(DestRemote);
- 			/* ReadyForQuery did pq_flush for us */
  			break;
  
  		default:
--- 378,387 ----
***************
*** 500,506 **** HandleReplicationCommand(const char *cmd_string)
  	MemoryContextSwitchTo(old_context);
  	MemoryContextDelete(cmd_context);
  
! 	return replication_started;
  }
  
  /*
--- 394,401 ----
  	MemoryContextSwitchTo(old_context);
  	MemoryContextDelete(cmd_context);
  
! 	/* Send CommandComplete message */
! 	EndCommand("SELECT", DestRemote);
  }
  
  /*
***************
*** 754,768 **** WalSndLoop(void)
  			SyncRepInitConfig();
  		}
  
! 		/* Normal exit from the walsender is here */
! 		if (walsender_shutdown_requested)
! 		{
! 			/* Inform the standby that XLOG streaming is done */
! 			pq_puttextmessage('C', "COPY 0");
! 			pq_flush();
! 
! 			proc_exit(0);
! 		}
  
  		/* Check for input from the client */
  		ProcessRepliesIfAny();
--- 649,655 ----
  			SyncRepInitConfig();
  		}
  
! 		CHECK_FOR_INTERRUPTS();
  
  		/* Check for input from the client */
  		ProcessRepliesIfAny();
***************
*** 813,819 **** WalSndLoop(void)
  				XLogSend(output_message, &caughtup);
  				if (caughtup && !pq_is_send_pending())
  				{
! 					walsender_shutdown_requested = true;
  					continue;	/* don't want to wait more */
  				}
  			}
--- 700,706 ----
  				XLogSend(output_message, &caughtup);
  				if (caughtup && !pq_is_send_pending())
  				{
! 					ProcDiePending = true;
  					continue;	/* don't want to wait more */
  				}
  			}
***************
*** 854,861 **** WalSndLoop(void)
--- 741,751 ----
  			}
  
  			/* Sleep until something happens or replication timeout */
+ 			ImmediateInterruptOK = true;
+ 			CHECK_FOR_INTERRUPTS();
  			WaitLatchOrSocket(&MyWalSnd->latch, wakeEvents,
  							  MyProcPort->sock, sleeptime);
+ 			ImmediateInterruptOK = false;
  
  			/*
  			 * Check for replication timeout.  Note we ignore the corner case
***************
*** 892,898 **** WalSndLoop(void)
  
  /* Initialize a per-walsender data structure for this walsender process */
  static void
! InitWalSnd(void)
  {
  	int			i;
  
--- 782,788 ----
  
  /* Initialize a per-walsender data structure for this walsender process */
  static void
! InitWalSenderSlot(void)
  {
  	int			i;
  
***************
*** 1284,1341 **** WalSndSigHupHandler(SIGNAL_ARGS)
  	errno = save_errno;
  }
  
- /* SIGTERM: set flag to shut down */
- static void
- WalSndShutdownHandler(SIGNAL_ARGS)
- {
- 	int			save_errno = errno;
- 
- 	walsender_shutdown_requested = true;
- 	if (MyWalSnd)
- 		SetLatch(&MyWalSnd->latch);
- 
- 	/*
- 	 * Set the standard (non-walsender) state as well, so that we can abort
- 	 * things like do_pg_stop_backup().
- 	 */
- 	InterruptPending = true;
- 	ProcDiePending = true;
- 
- 	errno = save_errno;
- }
- 
- /*
-  * WalSndQuickDieHandler() occurs when signalled SIGQUIT by the postmaster.
-  *
-  * Some backend has bought the farm,
-  * so we need to stop what we're doing and exit.
-  */
- static void
- WalSndQuickDieHandler(SIGNAL_ARGS)
- {
- 	PG_SETMASK(&BlockSig);
- 
- 	/*
- 	 * We DO NOT want to run proc_exit() callbacks -- we're here because
- 	 * shared memory may be corrupted, so we don't want to try to clean up our
- 	 * transaction.  Just nail the windows shut and get out of town.  Now that
- 	 * there's an atexit callback to prevent third-party code from breaking
- 	 * things by calling exit() directly, we have to reset the callbacks
- 	 * explicitly to make this work as intended.
- 	 */
- 	on_exit_reset();
- 
- 	/*
- 	 * Note we do exit(2) not exit(0).	This is to force the postmaster into a
- 	 * system reset cycle if some idiot DBA sends a manual SIGQUIT to a random
- 	 * backend.  This is necessary precisely because we don't clean up our
- 	 * shared memory state.  (The "dead man switch" mechanism in pmsignal.c
- 	 * should ensure the postmaster sees this as a crash, too, but no harm in
- 	 * being doubly sure.)
- 	 */
- 	exit(2);
- }
- 
  /* SIGUSR1: set flag to send WAL records */
  static void
  WalSndXLogSendHandler(SIGNAL_ARGS)
--- 1174,1179 ----
***************
*** 1368,1375 **** WalSndSignals(void)
  	pqsignal(SIGHUP, WalSndSigHupHandler);		/* set flag to read config
  												 * file */
  	pqsignal(SIGINT, SIG_IGN);	/* not used */
! 	pqsignal(SIGTERM, WalSndShutdownHandler);	/* request shutdown */
! 	pqsignal(SIGQUIT, WalSndQuickDieHandler);	/* hard crash time */
  	InitializeTimeouts();		/* establishes SIGALRM handler */
  	pqsignal(SIGPIPE, SIG_IGN);
  	pqsignal(SIGUSR1, WalSndXLogSendHandler);	/* request WAL sending */
--- 1206,1213 ----
  	pqsignal(SIGHUP, WalSndSigHupHandler);		/* set flag to read config
  												 * file */
  	pqsignal(SIGINT, SIG_IGN);	/* not used */
! 	pqsignal(SIGTERM, die);						/* request shutdown */
! 	pqsignal(SIGQUIT, quickdie);				/* hard crash time */
  	InitializeTimeouts();		/* establishes SIGALRM handler */
  	pqsignal(SIGPIPE, SIG_IGN);
  	pqsignal(SIGUSR1, WalSndXLogSendHandler);	/* request WAL sending */
*** a/src/backend/tcop/postgres.c
--- b/src/backend/tcop/postgres.c
***************
*** 192,197 **** static int	InteractiveBackend(StringInfo inBuf);
--- 192,198 ----
  static int	interactive_getc(void);
  static int	SocketBackend(StringInfo inBuf);
  static int	ReadCommand(StringInfo inBuf);
+ static void forbidden_in_wal_sender(char firstchar);
  static List *pg_rewrite_query(Query *query);
  static bool check_log_statement(List *stmt_list);
  static int	errdetail_execute(List *raw_parsetree_list);
***************
*** 3720,3731 **** PostgresMain(int argc, char *argv[], const char *username)
  	if (IsUnderPostmaster && Log_disconnections)
  		on_proc_exit(log_disconnections, 0);
  
! 	/* If this is a WAL sender process, we're done with initialization. */
  	if (am_walsender)
! 	{
! 		WalSenderMain();		/* does not return */
! 		abort();
! 	}
  
  	/*
  	 * process any libraries that should be preloaded at backend start (this
--- 3721,3729 ----
  	if (IsUnderPostmaster && Log_disconnections)
  		on_proc_exit(log_disconnections, 0);
  
! 	/* Perform initialization specific to a WAL sender process. */
  	if (am_walsender)
! 		InitWalSender();
  
  	/*
  	 * process any libraries that should be preloaded at backend start (this
***************
*** 3835,3840 **** PostgresMain(int argc, char *argv[], const char *username)
--- 3833,3841 ----
  		 */
  		AbortCurrentTransaction();
  
+ 		if (am_walsender)
+ 			WalSndErrorCleanup();
+ 
  		/*
  		 * Now return to normal top-level context and clear ErrorContext for
  		 * next time.
***************
*** 3969,3975 **** PostgresMain(int argc, char *argv[], const char *username)
  					query_string = pq_getmsgstring(&input_message);
  					pq_getmsgend(&input_message);
  
! 					exec_simple_query(query_string);
  
  					send_ready_for_query = true;
  				}
--- 3970,3979 ----
  					query_string = pq_getmsgstring(&input_message);
  					pq_getmsgend(&input_message);
  
! 					if (am_walsender)
! 						exec_replication_command(query_string);
! 					else
! 						exec_simple_query(query_string);
  
  					send_ready_for_query = true;
  				}
***************
*** 3982,3987 **** PostgresMain(int argc, char *argv[], const char *username)
--- 3986,3993 ----
  					int			numParams;
  					Oid		   *paramTypes = NULL;
  
+ 					forbidden_in_wal_sender(firstchar);
+ 
  					/* Set statement_timestamp() */
  					SetCurrentStatementStartTimestamp();
  
***************
*** 4004,4009 **** PostgresMain(int argc, char *argv[], const char *username)
--- 4010,4017 ----
  				break;
  
  			case 'B':			/* bind */
+ 				forbidden_in_wal_sender(firstchar);
+ 
  				/* Set statement_timestamp() */
  				SetCurrentStatementStartTimestamp();
  
***************
*** 4019,4024 **** PostgresMain(int argc, char *argv[], const char *username)
--- 4027,4034 ----
  					const char *portal_name;
  					int			max_rows;
  
+ 					forbidden_in_wal_sender(firstchar);
+ 
  					/* Set statement_timestamp() */
  					SetCurrentStatementStartTimestamp();
  
***************
*** 4031,4036 **** PostgresMain(int argc, char *argv[], const char *username)
--- 4041,4048 ----
  				break;
  
  			case 'F':			/* fastpath function call */
+ 				forbidden_in_wal_sender(firstchar);
+ 
  				/* Set statement_timestamp() */
  				SetCurrentStatementStartTimestamp();
  
***************
*** 4078,4083 **** PostgresMain(int argc, char *argv[], const char *username)
--- 4090,4097 ----
  					int			close_type;
  					const char *close_target;
  
+ 					forbidden_in_wal_sender(firstchar);
+ 
  					close_type = pq_getmsgbyte(&input_message);
  					close_target = pq_getmsgstring(&input_message);
  					pq_getmsgend(&input_message);
***************
*** 4120,4125 **** PostgresMain(int argc, char *argv[], const char *username)
--- 4134,4141 ----
  					int			describe_type;
  					const char *describe_target;
  
+ 					forbidden_in_wal_sender(firstchar);
+ 
  					/* Set statement_timestamp() (needed for xact) */
  					SetCurrentStatementStartTimestamp();
  
***************
*** 4201,4206 **** PostgresMain(int argc, char *argv[], const char *username)
--- 4217,4245 ----
  	}							/* end of input-reading loop */
  }
  
+ /*
+  * Throw an error if we're a WAL sender process.
+  *
+  * This is used to forbid anything else than simple query protocol messages
+  * in a WAL sender process. 'firstchar' specifies what kind of a forbidden
+  * message was received, and is used to construct the error message.
+  */
+ static void
+ forbidden_in_wal_sender(char firstchar)
+ {
+ 	if (am_walsender)
+ 	{
+ 		if (firstchar == 'F')
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
+ 					 errmsg("fastpath function calls not supported in a replication connection")));
+ 		else
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
+ 					 errmsg("extended query protocol not supported in a replication connection")));
+ 	}
+ }
+ 
  
  /*
   * Obtain platform stack depth limit (in bytes)
*** a/src/include/replication/walsender.h
--- b/src/include/replication/walsender.h
***************
*** 19,25 ****
  /* global state */
  extern bool am_walsender;
  extern bool am_cascading_walsender;
- extern volatile sig_atomic_t walsender_shutdown_requested;
  extern volatile sig_atomic_t walsender_ready_to_stop;
  extern bool wake_wal_senders;
  
--- 19,24 ----
***************
*** 27,33 **** extern bool wake_wal_senders;
  extern int	max_wal_senders;
  extern int	replication_timeout;
  
! extern void WalSenderMain(void) __attribute__((noreturn));
  extern void WalSndSignals(void);
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);
--- 26,34 ----
  extern int	max_wal_senders;
  extern int	replication_timeout;
  
! extern void InitWalSender(void);
! extern void exec_replication_command(const char *query_string);
! extern void WalSndErrorCleanup(void);
  extern void WalSndSignals(void);
  extern Size WalSndShmemSize(void);
  extern void WalSndShmemInit(void);