/*
 * MPP-220077: real_act_prefix_size should not go beyond ps_buffer_size
 */
void
test__set_ps_display__real_act_prefix_size(void **state)
{
	int		len;

	ps_buffer = (char *) malloc(127 * sizeof(char));
	ps_buffer_fixed_size = 79;
	memset(ps_buffer, 'x', ps_buffer_fixed_size * sizeof(char));
	ps_buffer_size = 127;
	IsUnderPostmaster = true;

	StrNCpy(ps_host_info, "msa4000125.europe.corp.microsoft.com(57193)",
			sizeof(ps_host_info));
	ps_host_info_size = 0;
	gp_session_id = 26351;
	Gp_role = GP_ROLE_DISPATCH;
	Gp_segment = -1;
	gp_command_count = 964;
	currentSliceId = -1;

	set_ps_display("testing activity", true);
	assert_true(real_act_prefix_size <= ps_buffer_size);

	get_real_act_ps_display(&len);
	assert_true(len >= 0);
}

/* Flush the log to disk */
static void
XLogWalRcvFlush(void)
{
	if (XLByteLT(LogstreamResult.Flush, LogstreamResult.Write))
	{
		/* use volatile pointer to prevent code rearrangement */
		volatile WalRcvData *walrcv = WalRcv;

		issue_xlog_fsync(recvFile, recvId, recvSeg);

		LogstreamResult.Flush = LogstreamResult.Write;

		/* Update shared-memory status */
		SpinLockAcquire(&walrcv->mutex);
		walrcv->latestChunkStart = walrcv->receivedUpto;
		walrcv->receivedUpto = LogstreamResult.Flush;
		SpinLockRelease(&walrcv->mutex);

		/* Report XLOG streaming progress in PS display */
		if (update_process_title)
		{
			char		activitymsg[50];

			snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
					 LogstreamResult.Write.xlogid,
					 LogstreamResult.Write.xrecoff);
			set_ps_display(activitymsg, false);
		}
	}
}

/*
 * PerformAuthentication -- authenticate a remote client
 *
 * returns: nothing.  Will not return at all if there's any failure.
 */
static void
PerformAuthentication(Port *port)
{
	/* This should be set already, but let's make sure */
	ClientAuthInProgress = true;	/* limit visibility of log messages */

	/*
	 * In EXEC_BACKEND case, we didn't inherit the contents of pg_hba.conf
	 * etcetera from the postmaster, and have to load them ourselves.  Note we
	 * are loading them into the startup transaction's memory context, not
	 * PostmasterContext, but that shouldn't matter.
	 *
	 * FIXME: [fork/exec] Ugh.	Is there a way around this overhead?
	 */
#ifdef EXEC_BACKEND
	if (!load_hba())
	{
		/*
		 * It makes no sense to continue if we fail to load the HBA file,
		 * since there is no way to connect to the database in this case.
		 */
		ereport(FATAL,
				(errmsg("could not load pg_hba.conf")));
	}
	load_ident();
#endif

	/*
	 * Set up a timeout in case a buggy or malicious client fails to respond
	 * during authentication.  Since we're inside a transaction and might do
	 * database access, we have to use the statement_timeout infrastructure.
	 */
	enable_timeout_after(STATEMENT_TIMEOUT, AuthenticationTimeout * 1000);

	/*
	 * Now perform authentication exchange.
	 */
	ClientAuthentication(port); /* might not return, if failure */

	/*
	 * Done with authentication.  Disable the timeout, and log if needed.
	 */
	disable_timeout(STATEMENT_TIMEOUT, false);

	if (Log_connections)
	{
		if (am_walsender)
			ereport(LOG,
					(errmsg("replication connection authorized: user=%s",
							port->user_name)));
		else
			ereport(LOG,
					(errmsg("connection authorized: user=%s database=%s",
							port->user_name, port->database_name)));
	}

	set_ps_display("startup", false);

	ClientAuthInProgress = false;		/* client_min_messages is active now */
}

/*
 * Call this once during subprocess startup to set the identification
 * values. At this point, the original argv[] array may be overwritten.
 */
void init_ps_display(
	const char *username,
	const char *dbname,
	const char *host_info,
	const char *initial_str)
{
	ASSERT(username);
	ASSERT(dbname);
	ASSERT(host_info);

#ifndef PS_USE_NONE
	/* no ps display for stand-alone backend */
	if (!child)
		return;

	/* no ps display if you didn't call save_args() */
	if (!save_argv)
		return;

#ifdef PS_USE_CLOBBER_ARGV
	/* If ps_buffer is a pointer, it might still be null */
	if (!ps_buffer)
		return;
#endif	// PS_USE_CLOBBER_ARGV

	/*
	 * Overwrite argv[] to point at appropriate space, if needed
	 */
#ifdef PS_USE_CHANGE_ARGV
	save_argv[0] = ps_buffer;
	save_argv[1] = NULL;
#endif   /* PS_USE_CHANGE_ARGV */

#ifdef PS_USE_CLOBBER_ARGV
	int i;

	/* make extra argv slots point at end_of_area (a NUL) */
	for (i = 1; i < save_argc; i++)
		save_argv[i] = ps_buffer + ps_buffer_size;
#endif   /* PS_USE_CLOBBER_ARGV */

	/*
	 * Make fixed prefix of ps display.
	 */
#ifdef PS_USE_SETPROCTITLE
	/*
	 * apparently setproctitle() already adds a `progname:' prefix to the 
	 * ps line
	 */
	snprintf(ps_buffer, ps_buffer_size, "%s %s %s ", username, dbname, host_info);
#else
	snprintf(ps_buffer, ps_buffer_size, "postgres: %s %s %s ", username, dbname, host_info);
#endif	// PS_USE_SETPROCTITLE

	ps_buffer_cur_len = ps_buffer_fixed_size = strlen(ps_buffer);
	set_ps_display(initial_str, true);
#endif   /* not PS_USE_NONE */
}

/*
 * Check it won't crash in case the ps_buffer overflows.
 */
void
test__set_ps_display(void **state)
{
	ps_buffer = (char *) malloc(64 * sizeof(char));
	memset(ps_buffer, 0x7F, 64 * sizeof(char));
	ps_buffer_fixed_size = 25;
	ps_buffer_size = 32;
	IsUnderPostmaster = true;

	gp_session_id = 1024;
	Gp_role = GP_ROLE_DISPATCH;
	Gp_segment = 24;
	gp_command_count = 1024;
	currentSliceId = 40;

	set_ps_display("testing activity", true);
	set_ps_display("testing activity", false);

	assert_true(ps_buffer[32] == 0x7f);
}

/*************************M*A*I*N*************************/
int main(void) {
	turtle_t boo;
	pen_t pen;
	srand(time(NULL));

	set_ps_header(HEIGHT, WIDTH);
	process_commands(&boo, &pen);
	set_ps_display();

	return 0;
}

/*************************M*A*I*N*************************/
int main(void) {
	turtle_t boo;
	pen_t pen;
	srand(time(NULL));

	set_ps_header(HEIGHT, WIDTH);
	turtle_goto(&boo, 297.5, 420.5);
	turtle_random_walk(&boo, &pen, 20000);
	set_ps_display();

	return 0;
}

/*
* worker child main loop
*/
void do_worker_child(void)
{
	pool_debug("I am %d", getpid());

	/* Identify myself via ps */
	init_ps_display("", "", "", "");
	set_ps_display("worker process", false);

	/* set up signal handlers */
	signal(SIGALRM, SIG_DFL);
	signal(SIGTERM, my_signal_handler);
	signal(SIGINT, my_signal_handler);
	signal(SIGHUP, reload_config_handler);
	signal(SIGQUIT, my_signal_handler);
	signal(SIGCHLD, SIG_IGN);
	signal(SIGUSR1, my_signal_handler);
	signal(SIGUSR2, SIG_IGN);
	signal(SIGPIPE, SIG_IGN);

	/* Initialize my backend status */
	pool_initialize_private_backend_status();

	/* Initialize per process context */
	pool_init_process_context();

	for (;;)
	{
		CHECK_REQUEST;

		if (pool_config->sr_check_period <= 0)
		{
			sleep(30);
		}

		/*
		 * If streaming replication mode, do time lag checking
		 */

		if (pool_config->sr_check_period > 0 && MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
		{
			/* Check and establish persistent connections to the backend */
			establish_persistent_connection();

			/* Do replication time lag checking */
			check_replication_time_lag();

			/* Discard persistent connections */
			discard_persistent_connection();
		}
		sleep(pool_config->sr_check_period);
	}
	exit(0);
}

static PyObject *
spt_setproctitle(PyObject *self /* Not used */, PyObject *args)
{
    const char *title;

    if (!PyArg_ParseTuple(args, "s", &title))
        return NULL;

    set_ps_display(title, true);

    Py_INCREF(Py_None);
    return Py_None;
}

/* fork lifecheck process*/
static pid_t
fork_a_lifecheck(int fork_wait_time)
{
	pid_t pid;

	pid = fork();
	if (pid != 0)
	{
		if (pid == -1)
			pool_error("fork_a_lifecheck: fork() failed.");

		return pid;
	}

	if (fork_wait_time > 0) {
		sleep(fork_wait_time);
	}

	myargv = save_ps_display_args(myargc, myargv);

	POOL_SETMASK(&UnBlockSig);

	init_ps_display("", "", "", "");

	signal(SIGTERM, wd_exit);	
	signal(SIGINT, wd_exit);	
	signal(SIGQUIT, wd_exit);	
	signal(SIGCHLD, SIG_DFL);
	signal(SIGHUP, SIG_IGN);	
	signal(SIGPIPE, SIG_IGN);	

	set_ps_display("lifecheck",false);

	/* wait until ready to go */
	while (WD_OK != is_wd_lifecheck_ready())
	{
		sleep(pool_config->wd_interval * 10);
	}

	pool_log("watchdog: lifecheck started");

	/* watchdog loop */
	for (;;)
	{
		/* pgpool life check */
		wd_lifecheck();
		sleep(pool_config->wd_interval);
	}

	return pid;
}

/*
 * Show ps idle status
 */
void
pool_ps_idle_display(POOL_CONNECTION_POOL * backend)
{
	StartupPacket *sp;
	char		psbuf[1024];

	sp = MASTER_CONNECTION(backend)->sp;
	if (MASTER(backend)->tstate == 'T')
		snprintf(psbuf, sizeof(psbuf), "%s %s %s idle in transaction",
				 sp->user, sp->database, remote_ps_data);
	else
		snprintf(psbuf, sizeof(psbuf), "%s %s %s idle",
				 sp->user, sp->database, remote_ps_data);
	set_ps_display(psbuf, false);
}

/*
 * Flush the log to disk.
 *
 * If we're in the midst of dying, it's unwise to do anything that might throw
 * an error, so we skip sending a reply in that case.
 */
static void
XLogWalRcvFlush(bool dying)
{
	if (XLByteLT(LogstreamResult.Flush, LogstreamResult.Write))
	{
		/* use volatile pointer to prevent code rearrangement */
		volatile WalRcvData *walrcv = WalRcv;

		issue_xlog_fsync(recvFile, recvId, recvSeg);

		LogstreamResult.Flush = LogstreamResult.Write;

		/* Update shared-memory status */
		SpinLockAcquire(&walrcv->mutex);
		if (XLByteLT(walrcv->receivedUpto, LogstreamResult.Flush))
		{
			walrcv->latestChunkStart = walrcv->receivedUpto;
			walrcv->receivedUpto = LogstreamResult.Flush;
		}
		SpinLockRelease(&walrcv->mutex);

		/* Signal the startup process and walsender that new WAL has arrived */
		WakeupRecovery();
		if (AllowCascadeReplication())
			WalSndWakeup();

		/* Report XLOG streaming progress in PS display */
		if (update_process_title)
		{
			char		activitymsg[50];

			snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
					 LogstreamResult.Write.xlogid,
					 LogstreamResult.Write.xrecoff);
			set_ps_display(activitymsg, false);
		}

		/* Also let the master know that we made some progress */
		if (!dying)
		{
			XLogWalRcvSendReply();
			XLogWalRcvSendHSFeedback();
		}
	}
}

/*
 * Call this once during subprocess startup to set the identification
 * values.  At this point, the original argv[] array may be overwritten.
 */
void
init_ps_display(const char *initial_str)
{

#ifndef PS_USE_NONE
    /* no ps display if you didn't call save_ps_display_args() */
    if (!save_argv)
        return;
#ifdef PS_USE_CLOBBER_ARGV
    /* If ps_buffer is a pointer, it might still be null */
    if (!ps_buffer)
        return;
#endif

    /*
     * Overwrite argv[] to point at appropriate space, if needed
     */

#ifdef PS_USE_CHANGE_ARGV
    save_argv[0] = ps_buffer;
    save_argv[1] = NULL;
#endif   /* PS_USE_CHANGE_ARGV */

#ifdef PS_USE_CLOBBER_ARGV
    {
        int         i;

        /* make extra argv slots point at end_of_area (a NUL) */
        for (i = 1; i < save_argc; i++)
            save_argv[i] = ps_buffer + ps_buffer_size;
    }
#endif   /* PS_USE_CLOBBER_ARGV */

    /*
     * Make fixed prefix of ps display.
     */

    ps_buffer[0] = '\0';

    ps_buffer_fixed_size = strlen(ps_buffer);

    set_ps_display(initial_str, true);
#endif   /* not PS_USE_NONE */
}

/*
 * PgArchiverMain
 *
 *	The argc/argv parameters are valid only in EXEC_BACKEND case.  However,
 *	since we don't use 'em, it hardly matters...
 */
NON_EXEC_STATIC void
PgArchiverMain(int argc, char *argv[])
{
	IsUnderPostmaster = true;	/* we are a postmaster subprocess now */

	MyProcPid = getpid();		/* reset MyProcPid */

	/* Lose the postmaster's on-exit routines */
	on_exit_reset();

	/*
	 * Ignore all signals usually bound to some action in the postmaster,
	 * except for SIGHUP, SIGUSR1 and SIGQUIT.
	 */
	pqsignal(SIGHUP, ArchSigHupHandler);
	pqsignal(SIGINT, SIG_IGN);
	pqsignal(SIGTERM, SIG_IGN);
	pqsignal(SIGQUIT, pgarch_exit);
	pqsignal(SIGALRM, SIG_IGN);
	pqsignal(SIGPIPE, SIG_IGN);
	pqsignal(SIGUSR1, pgarch_waken);
	pqsignal(SIGUSR2, SIG_IGN);
	pqsignal(SIGCHLD, SIG_DFL);
	pqsignal(SIGTTIN, SIG_DFL);
	pqsignal(SIGTTOU, SIG_DFL);
	pqsignal(SIGCONT, SIG_DFL);
	pqsignal(SIGWINCH, SIG_DFL);
	PG_SETMASK(&UnBlockSig);

	/*
	 * Identify myself via ps
	 */
	init_ps_display("archiver process", "", "");
	set_ps_display("");

	pgarch_MainLoop();

	exit(0);
}

/*
 * 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)));
		}
	}
}

//.........这里部分代码省略.........
	{
		*caughtup = true;
		return;
	}

	/*
	 * Figure out how much to send in one message. If there's no more than
	 * MAX_SEND_SIZE bytes to send, send everything. Otherwise send
	 * MAX_SEND_SIZE bytes, but round back to logfile or page boundary.
	 *
	 * The rounding is not only for performance reasons. Walreceiver relies on
	 * the fact that we never split a WAL record across two messages. Since a
	 * long WAL record is split at page boundary into continuation records,
	 * page boundary is always a safe cut-off point. We also assume that
	 * SendRqstPtr never points to the middle of a WAL record.
	 */
	startptr = sentPtr;
	if (startptr.xrecoff >= XLogFileSize)
	{
		/*
		 * crossing a logid boundary, skip the non-existent last log segment
		 * in previous logical log file.
		 */
		startptr.xlogid += 1;
		startptr.xrecoff = 0;
	}

	endptr = startptr;
	XLByteAdvance(endptr, MAX_SEND_SIZE);
	if (endptr.xlogid != startptr.xlogid)
	{
		/* Don't cross a logfile boundary within one message */
		Assert(endptr.xlogid == startptr.xlogid + 1);
		endptr.xlogid = startptr.xlogid;
		endptr.xrecoff = XLogFileSize;
	}

	/* if we went beyond SendRqstPtr, back off */
	if (XLByteLE(SendRqstPtr, endptr))
	{
		endptr = SendRqstPtr;
		*caughtup = true;
	}
	else
	{
		/* round down to page boundary. */
		endptr.xrecoff -= (endptr.xrecoff % XLOG_BLCKSZ);
		*caughtup = false;
	}

	nbytes = endptr.xrecoff - startptr.xrecoff;
	Assert(nbytes <= MAX_SEND_SIZE);

	/*
	 * OK to read and send the slice.
	 */
	msgbuf[0] = 'w';

	/*
	 * Read the log directly into the output buffer to avoid extra memcpy
	 * calls.
	 */
	XLogRead(msgbuf + 1 + sizeof(WalDataMessageHeader), startptr, nbytes);

	/*
	 * We fill the message header last so that the send timestamp is taken as
	 * late as possible.
	 */
	msghdr.dataStart = startptr;
	msghdr.walEnd = SendRqstPtr;
	msghdr.sendTime = GetCurrentTimestamp();

	memcpy(msgbuf + 1, &msghdr, sizeof(WalDataMessageHeader));

	pq_putmessage_noblock('d', msgbuf, 1 + sizeof(WalDataMessageHeader) + nbytes);

	sentPtr = endptr;

	/* Update shared memory status */
	{
		/* use volatile pointer to prevent code rearrangement */
		volatile WalSnd *walsnd = MyWalSnd;

		SpinLockAcquire(&walsnd->mutex);
		walsnd->sentPtr = sentPtr;
		SpinLockRelease(&walsnd->mutex);
	}

	/* Report progress of XLOG streaming in PS display */
	if (update_process_title)
	{
		char		activitymsg[50];

		snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
				 sentPtr.xlogid, sentPtr.xrecoff);
		set_ps_display(activitymsg, false);
	}

	return;
}

/*
 * Wait for synchronous replication, if requested by user.
 *
 * Initially backends start in state SYNC_REP_NOT_WAITING and then
 * change that state to SYNC_REP_WAITING before adding ourselves
 * to the wait queue. During SyncRepWakeQueue() a WALSender changes
 * the state to SYNC_REP_WAIT_COMPLETE once replication is confirmed.
 * This backend then resets its state to SYNC_REP_NOT_WAITING.
 */
void
SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
{
	char 		*new_status = NULL;
	const char *old_status;

	/*
	 * Fast exit if user has not requested sync replication, or
	 * there are no sync replication standby names defined.
	 * Note that those standbys don't need to be connected.
	 */
	if (!SyncRepRequested() || !SyncStandbysDefined())
		return;

	Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
	Assert(WalSndCtl != NULL);

	/* Reset the latch before adding ourselves to the queue. */
	ResetLatch(&MyProc->waitLatch);

	/*
	 * Set our waitLSN so WALSender will know when to wake us, and add
	 * ourselves to the queue.
	 */
	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
	Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING);
	if (!WalSndCtl->sync_standbys_defined)
	{
		/*
		 * We don't wait for sync rep if WalSndCtl->sync_standbys_defined is
		 * not set.  See SyncRepUpdateSyncStandbysDefined.
		 */
		LWLockRelease(SyncRepLock);
		return;
	}
	MyProc->waitLSN = XactCommitLSN;
	MyProc->syncRepState = SYNC_REP_WAITING;
	SyncRepQueueInsert();
	Assert(SyncRepQueueIsOrderedByLSN());
	LWLockRelease(SyncRepLock);

	/* Alter ps display to show waiting for sync rep. */
	if (update_process_title)
	{
		int			len;

		old_status = get_ps_display(&len);
		new_status = (char *) palloc(len + 32 + 1);
		memcpy(new_status, old_status, len);
		sprintf(new_status + len, " waiting for %X/%X",
				XactCommitLSN.xlogid, XactCommitLSN.xrecoff);
		set_ps_display(new_status, false);
		new_status[len] = '\0'; /* truncate off " waiting ..." */
	}

	/*
	 * Wait for specified LSN to be confirmed.
	 *
	 * Each proc has its own wait latch, so we perform a normal latch
	 * check/wait loop here.
	 */
	for (;;)
	{
		int syncRepState;

		/*
		 * Wait on latch for up to 60 seconds. This allows us to
		 * check for postmaster death regularly while waiting.
		 * Note that timeout here does not necessarily release from loop.
		 */
		WaitLatch(&MyProc->waitLatch, 60000000L);

		/* Must reset the latch before testing state. */
		ResetLatch(&MyProc->waitLatch);

		/*
		 * Try checking the state without the lock first.  There's no guarantee
		 * that we'll read the most up-to-date value, so if it looks like we're
		 * still waiting, recheck while holding the lock.  But if it looks like
		 * we're done, we must really be done, because once walsender changes
		 * the state to SYNC_REP_WAIT_COMPLETE, it will never update it again,
		 * so we can't be seeing a stale value in that case.
		 */
		syncRepState = MyProc->syncRepState;
		if (syncRepState == SYNC_REP_WAITING)
		{
			LWLockAcquire(SyncRepLock, LW_SHARED);
			syncRepState = MyProc->syncRepState;
			LWLockRelease(SyncRepLock);
		}
		if (syncRepState == SYNC_REP_WAIT_COMPLETE)
//.........这里部分代码省略.........

pid_t
wd_child(int fork_wait_time)
{
	int sock;
	int fd;
	int rtn;
	pid_t pid = 0;

	pid = fork();
	if (pid != 0)
	{
		if (pid == -1)
			pool_error("wd_child: fork() failed.");

		return pid;
	}

	if (fork_wait_time > 0)
	{
		sleep(fork_wait_time);
	}

	myargv = save_ps_display_args(myargc, myargv);

	POOL_SETMASK(&UnBlockSig);

	signal(SIGTERM, wd_child_exit);
	signal(SIGINT, wd_child_exit);
	signal(SIGQUIT, wd_child_exit);
	signal(SIGCHLD, SIG_IGN);
	signal(SIGHUP, SIG_IGN);
	signal(SIGUSR1, SIG_IGN);
	signal(SIGUSR2, SIG_IGN);
	signal(SIGPIPE, SIG_IGN);
	signal(SIGALRM, SIG_IGN);

	init_ps_display("", "", "", "");

	if (WD_List == NULL)
	{
		/* memory allocate is not ready */
		wd_child_exit(15);
	}

	sock = wd_create_recv_socket(WD_MYSELF->wd_port);

	if (sock < 0)
	{
		/* socket create failed */
		wd_child_exit(15);
	}

	set_ps_display("watchdog", false);

	/* child loop */
	for(;;)
	{
		WdPacket buf;
		fd = wd_accept(sock);
		if (fd < 0)
		{
			continue;
		}
		rtn = wd_recv_packet(fd, &buf);
		if (rtn == WD_OK)
		{
			wd_send_response(fd, &buf);
		}
		close(fd);
	}
	return pid;
}

/*
 * This is the main executioner for any query backend that conflicts with
 * recovery processing. Judgement has already been passed on it within
 * a specific rmgr. Here we just issue the orders to the procs. The procs
 * then throw the required error as instructed.
 */
static void
ResolveRecoveryConflictWithVirtualXIDs(VirtualTransactionId *waitlist,
									   ProcSignalReason reason)
{
	TimestampTz waitStart;
	char	   *new_status;

	/* Fast exit, to avoid a kernel call if there's no work to be done. */
	if (!VirtualTransactionIdIsValid(*waitlist))
		return;

	waitStart = GetCurrentTimestamp();
	new_status = NULL;			/* we haven't changed the ps display */

	while (VirtualTransactionIdIsValid(*waitlist))
	{
		/* reset standbyWait_us for each xact we wait for */
		standbyWait_us = STANDBY_INITIAL_WAIT_US;

		/* wait until the virtual xid is gone */
		while (!ConditionalVirtualXactLockTableWait(*waitlist))
		{
			/*
			 * Report via ps if we have been waiting for more than 500 msec
			 * (should that be configurable?)
			 */
			if (update_process_title && new_status == NULL &&
				TimestampDifferenceExceeds(waitStart, GetCurrentTimestamp(),
										   500))
			{
				const char *old_status;
				int			len;

				old_status = get_ps_display(&len);
				new_status = (char *) palloc(len + 8 + 1);
				memcpy(new_status, old_status, len);
				strcpy(new_status + len, " waiting");
				set_ps_display(new_status, false);
				new_status[len] = '\0'; /* truncate off " waiting" */
			}

			/* Is it time to kill it? */
			if (WaitExceedsMaxStandbyDelay())
			{
				pid_t		pid;

				/*
				 * Now find out who to throw out of the balloon.
				 */
				Assert(VirtualTransactionIdIsValid(*waitlist));
				pid = CancelVirtualTransaction(*waitlist, reason);

				/*
				 * Wait a little bit for it to die so that we avoid flooding
				 * an unresponsive backend when system is heavily loaded.
				 */
				if (pid != 0)
					pg_usleep(5000L);
			}
		}

		/* The virtual transaction is gone now, wait for the next one */
		waitlist++;
	}

	/* Reset ps display if we changed it */
	if (new_status)
	{
		set_ps_display(new_status, false);
		pfree(new_status);
	}
}

/* pcp command processor */
static void
pcp_process_command(char tos, char *buf, int buf_len)
{

	if (tos == 'C' || tos == 'd' || tos == 'D' || tos == 'j' ||
		tos == 'J' || tos == 'O' || tos == 'T')
	{
		if (Req_info->switching)
		{
			if(Req_info->request_queue_tail != Req_info->request_queue_head)
			{
				POOL_REQUEST_KIND reqkind;
				reqkind = Req_info->request[(Req_info->request_queue_head +1) % MAX_REQUEST_QUEUE_SIZE].kind;
				
				if (reqkind == NODE_UP_REQUEST)
					ereport(ERROR,
							(errmsg("failed to process PCP request at the moment"),
							 errdetail("failback is in progress")));
				else if (reqkind == NODE_DOWN_REQUEST)
					ereport(ERROR,
							(errmsg("failed to process PCP request at the moment"),
							 errdetail("failover is in progress")));
				else if (reqkind == PROMOTE_NODE_REQUEST)
					ereport(ERROR,
							(errmsg("failed to process PCP request at the moment"),
							 errdetail("promote node operation is in progress")));

				ereport(ERROR,
						(errmsg("failed to process PCP request at the moment"),
						 errdetail("operation is in progress")));
			}
		}
	}
	
	switch (tos)
	{
		case 'A':			/* set configuration parameter */
			set_ps_display("PCP: processing set configration parameter request", false);
			process_set_configration_parameter(pcp_frontend,buf,buf_len);
			break;

		case 'L':			/* node count */
			set_ps_display("PCP: processing node count request", false);
			inform_node_count(pcp_frontend);
			break;

		case 'I':			/* node info */
			set_ps_display("PCP: processing node info request", false);
			inform_node_info(pcp_frontend, buf);
			break;

		case 'N':			/* process count */
			set_ps_display("PCP: processing process count request", false);
			inform_process_count(pcp_frontend);
			break;

		case 'P':			/* process info */
			set_ps_display("PCP: processing process info request", false);
			inform_process_info(pcp_frontend, buf);
			break;

		case 'W':			/* watchdog info */
			set_ps_display("PCP: processing watchdog info request", false);
			inform_watchdog_info(pcp_frontend, buf);
			break;

		case 'D':			/* detach node */
		case 'd':			/* detach node gracefully */
			set_ps_display("PCP: processing detach node request", false);
			process_detach_node(pcp_frontend, buf, tos);
			break;

		case 'C':			/* attach node */
			set_ps_display("PCP: processing attach node request", false);
			process_attach_node(pcp_frontend, buf);
			break;

		case 'T':
			set_ps_display("PCP: processing shutdown request", false);
			process_shutown_request(pcp_frontend, buf[0]);
			break;

		case 'O': /* recovery request */
			set_ps_display("PCP: processing recovery request", false);
			process_recovery_request(pcp_frontend, buf);
			break;

		case 'B': /* status request*/
			set_ps_display("PCP: processing status request request", false);
			process_status_request(pcp_frontend);
			break;

		case 'J':			/* promote node */
		case 'j':			/* promote node gracefully */
			set_ps_display("PCP: processing promote node request", false);
			process_promote_node(pcp_frontend,buf,tos);
			break;

		case 'F':
//.........这里部分代码省略.........