static void
setup_event_handlers(void)
{
	pqsignal(SIGHUP, handle_sighup);
	pqsignal(SIGINT, handle_sigint);
	pqsignal(SIGTERM, handle_sigint);
}

static void
WorkerCommon(void)
{
	/* Establish signal handlers before unblocking signals. */
	pqsignal(SIGHUP, sighupHandler);
	pqsignal(SIGTERM, sigtermHandler);

	/* We're now ready to receive signals */
	BackgroundWorkerUnblockSignals();
}

void worker_main(Datum arg)
{
	int 			ret;
	StringInfoData 	buf;
	uint32 			segment = UInt32GetDatum(arg);

	/* Setup signal handlers */
	pqsignal(SIGHUP, worker_sighup);
	pqsignal(SIGTERM, worker_sigterm);

	/* Allow signals */
	BackgroundWorkerUnblockSignals();

	initialize_worker(segment);

	/* Connect to the database */
	BackgroundWorkerInitializeConnection(job->datname, job->rolname);

	elog(LOG, "%s initialized running job id %d", MyBgworkerEntry->bgw_name, job->job_id);
	pgstat_report_appname(MyBgworkerEntry->bgw_name);

	/* Initialize the query text */
	initStringInfo(&buf);
	appendStringInfo(&buf,
					"SELECT * FROM %s.%s(%d, NULL)",
					job_run_function.schema,
					job_run_function.name,
					job->job_id);

	/* Initialize the SPI subsystem */
	SetCurrentStatementStartTimestamp();
	StartTransactionCommand();
	SPI_connect();
	PushActiveSnapshot(GetTransactionSnapshot());
	pgstat_report_activity(STATE_RUNNING, buf.data);

	SetCurrentStatementStartTimestamp();

	/* And run the query */
	ret = SPI_execute(buf.data, true, 0);
	if (ret < 0)
		elog(FATAL, "errors while executing %s", buf.data);

	/* Commmit the transaction */
	SPI_finish();
	PopActiveSnapshot();
	CommitTransactionCommand();
	pgstat_report_activity(STATE_IDLE, NULL);

	proc_exit(0);
}

/*
 * proc_exit callback to tear down the JVM
 */
static void _destroyJavaVM(int status, Datum dummy)
{
	if(s_javaVM != 0)
	{
		Invocation ctx;
#ifdef USE_PLJAVA_SIGHANDLERS

#if PG_VERSION_NUM >= 90300
		TimeoutId tid;
#else
		pqsigfunc saveSigAlrm;
#endif

		Invocation_pushInvocation(&ctx, false);
		if(sigsetjmp(recoverBuf, 1) != 0)
		{
			elog(DEBUG2,
				"needed to forcibly shut down the Java virtual machine");
			s_javaVM = 0;
			return;
		}

#if PG_VERSION_NUM >= 90300
		InitializeTimeouts();           /* establishes SIGALRM handler */
		tid = RegisterTimeout(USER_TIMEOUT, terminationTimeoutHandler);
#else
		saveSigAlrm = pqsignal(SIGALRM, terminationTimeoutHandler);
		enable_sig_alarm(5000, false);
#endif

		elog(DEBUG2, "shutting down the Java virtual machine");
		JNI_destroyVM(s_javaVM);

#if PG_VERSION_NUM >= 90300
		disable_timeout(tid, false);
#else
		disable_sig_alarm(false);
		pqsignal(SIGALRM, saveSigAlrm);
#endif

#else
		Invocation_pushInvocation(&ctx, false);
		elog(DEBUG2, "shutting down the Java virtual machine");
		JNI_destroyVM(s_javaVM);
#endif
		elog(DEBUG2, "done shutting down the Java virtual machine");
		s_javaVM = 0;
		currentInvocation = 0;
	}
}

/*
 * hello_main
 *
 * Main loop processing.
 */
void
hello_main(Datum main_arg)
{
	/* Set up the sigterm signal before unblocking them */
	pqsignal(SIGTERM, hello_sigterm);

	/* We're now ready to receive signals */
	BackgroundWorkerUnblockSignals();
	while (!got_sigterm)
	{
		int rc;

		/* Wait 10s */
		rc = WaitLatch(&MyProc->procLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
					   10000L,
					   PG_WAIT_EXTENSION);
		ResetLatch(&MyProc->procLatch);

		/* Emergency bailout if postmaster has died */
		if (rc & WL_POSTMASTER_DEATH)
			proc_exit(1);

		elog(LOG, "Hello World!"); /* Say Hello to the world */
	}
	proc_exit(0);
}

/*
 * Foreign-data wrapper handler function: return a struct with pointers
 * to my callback routines.
 */
Datum
jdbc_fdw_handler(PG_FUNCTION_ARGS)
{
	FdwRoutine 	*fdwroutine = makeNode(FdwRoutine);
	
	#if (PG_VERSION_NUM < 90200)
	fdwroutine->PlanForeignScan = jdbcPlanForeignScan;
	#endif

	#if (PG_VERSION_NUM >= 90200)
	fdwroutine->GetForeignRelSize = jdbcGetForeignRelSize;
	fdwroutine->GetForeignPaths = jdbcGetForeignPaths;
	fdwroutine->GetForeignPlan = jdbcGetForeignPlan;
	#endif

	fdwroutine->ExplainForeignScan = jdbcExplainForeignScan;
	fdwroutine->BeginForeignScan = jdbcBeginForeignScan;
	fdwroutine->IterateForeignScan = jdbcIterateForeignScan;
	fdwroutine->ReScanForeignScan = jdbcReScanForeignScan;
	fdwroutine->EndForeignScan = jdbcEndForeignScan;

	pqsignal(SIGINT, SIGINTInterruptHandler);

	PG_RETURN_POINTER(fdwroutine);
}

/*
 * signal handler in case we are interrupted.
 *
 * The Windows runtime docs at
 * http://msdn.microsoft.com/library/en-us/vclib/html/_crt_signal.asp
 * specifically forbid a number of things being done from a signal handler,
 * including IO, memory allocation and system calls, and only allow jmpbuf
 * if you are handling SIGFPE.
 *
 * I avoided doing the forbidden things by setting a flag instead of calling
 * exit_nicely() directly.
 *
 * Also note the behaviour of Windows with SIGINT, which says this:
 *	 Note	SIGINT is not supported for any Win32 application, including
 *	 Windows 98/Me and Windows NT/2000/XP. When a CTRL+C interrupt occurs,
 *	 Win32 operating systems generate a new thread to specifically handle
 *	 that interrupt. This can cause a single-thread application such as UNIX,
 *	 to become multithreaded, resulting in unexpected behavior.
 *
 * I have no idea how to handle this. (Strange they call UNIX an application!)
 * So this will need some testing on Windows.
 */
static void
trapsig(int signum)
{
	/* handle systems that reset the handler, like Windows (grr) */
	pqsignal(signum, trapsig);
	caught_signal = true;
}

void
_PG_init(void)
{

  coreIntHandler = pqsignal(SIGINT, handleInterrupt); 

#ifdef WIN32
#if POSTGIS_GEOS_VERSION >= 34 
  GEOS_interruptRegisterCallback(interruptCallback);
#endif
  lwgeom_register_interrupt_callback(interruptCallback);
#endif

#if 0
  /* Define custom GUC variables. */
  DefineCustomIntVariable(
    "postgis.debug.level", /* name */
    "Sets the debugging level of PostGIS.", /* short_desc */
    "This is an experimental configuration.", /* long_desc */
    &postgis_debug_level, /* valueAddr */
    0, 8, /* min-max */
    0, /* bootValue */
    PGC_SUSET, /* GucContext context */
    GUC_UNIT_MS, /* int flags */
#if POSTGIS_PGSQL_VERSION >= 91
    NULL, /* GucStringCheckHook check_hook */
#endif
    NULL, /* GucStringAssignHook assign_hook */
    NULL  /* GucShowHook show_hook */
   );
#endif

#if 0
  /* Define custom GUC variables. */
  DefineCustomStringVariable(
    "postgis.greeting.string", /* name */
    "Sets the greeting string used on postgis module load.", /* short_desc */
    "This is an experimental configuration.", /* long_desc */
    &greeting, /* valueAddr */
    "Welcome to PostGIS " POSTGIS_VERSION, /* bootValue */
    PGC_SUSET, /* GucContext context */
    GUC_UNIT_MS, /* int flags */
#if POSTGIS_PGSQL_VERSION >= 91
    NULL, /* GucStringCheckHook check_hook */
#endif
    NULL, /* GucStringAssignHook assign_hook */
    NULL  /* GucShowHook show_hook */
   );
#endif

    /* install PostgreSQL handlers */
    pg_install_lwgeom_handlers();

    /* initialize geometry backend */
    lwgeom_init_backend();
}

static void
config_log_main(Datum main_arg)
{
	config_log_objects	   *objects;

	pqsignal(SIGTERM, config_log_sigterm);
	pqsignal(SIGHUP,  config_log_sighup);

	/* We're now ready to receive signals */
	BackgroundWorkerUnblockSignals();

	/* Connect to database */
	BackgroundWorkerInitializeConnection(config_log_database, NULL);

	/* Verify expected objects exist */
	objects = initialize_objects();

	while (!got_sigterm)
	{
		int		rc;

		rc = WaitLatch(&MyProc->procLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
					   100000L);
		ResetLatch(&MyProc->procLatch);

		/* emergency bailout if postmaster has died */
		if (rc & WL_POSTMASTER_DEATH)
			proc_exit(1);

		/*
		 * In case of a SIGHUP, just reload the configuration.
		 */
		if (got_sighup)
		{
              got_sighup = false;
              ProcessConfigFile(PGC_SIGHUP);
              execute_pg_settings_logger(objects);
		}
    }

	proc_exit(0);
}

void
hello_main(Datum main_arg)
{
	/* Register functions for SIGTERM/SIGHUP management */
	pqsignal(SIGHUP, hello_sighup);
	pqsignal(SIGTERM, hello_sigterm);

	/* We're now ready to receive signals */
	BackgroundWorkerUnblockSignals();

	while (true)
	{
		/* Wait 1s */
		WaitLatch(MyLatch,
				  WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
				  1000L,
				  PG_WAIT_EXTENSION);
		ResetLatch(MyLatch);

		/* Process signals */
		if (got_sighup)
		{
			/* Process config file */
			ProcessConfigFile(PGC_SIGHUP);
			got_sighup = false;
			ereport(LOG, (errmsg("hello signal: processed SIGHUP")));
		}

		if (got_sigterm)
		{
			/* Simply exit */
			ereport(LOG, (errmsg("hello signal: processed SIGTERM")));
			proc_exit(0);
		}
	}

	/* No problems, so clean exit */
	proc_exit(0);
}

int
main(int argc, char *argv[])
{
	set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_test_fsync"));
	progname = get_progname(argv[0]);

	handle_args(argc, argv);

	/* Prevent leaving behind the test file */
	pqsignal(SIGINT, signal_cleanup);
	pqsignal(SIGTERM, signal_cleanup);
#ifndef WIN32
	pqsignal(SIGALRM, process_alarm);
#endif
#ifdef SIGHUP
	/* Not defined on win32 */
	pqsignal(SIGHUP, signal_cleanup);
#endif

	prepare_buf();

	test_open();

	/* Test using 1 XLOG_BLCKSZ write */
	test_sync(1);

	/* Test using 2 XLOG_BLCKSZ writes */
	test_sync(2);

	test_open_syncs();

	test_file_descriptor_sync();

	test_non_sync();

	unlink(filename);

	return 0;
}

/* ----------------------------------------------------------------
 *		proc_exit
 *
 *		this function calls all the callbacks registered
 *		for it (to free resources) and then calls exit.
 *
 *		This should be the only function to call exit().
 *		-cim 2/6/90
 *
 *		Unfortunately, we can't really guarantee that add-on code
 *		obeys the rule of not calling exit() directly.  So, while
 *		this is the preferred way out of the system, we also register
 *		an atexit callback that will make sure cleanup happens.
 * ----------------------------------------------------------------
 */
void
proc_exit(int code)
{
	pqsignal(SIGALRM, SIG_IGN);

	/* Clean up everything that must be cleaned up */
	proc_exit_prepare(code);

#ifdef PROFILE_PID_DIR
	{
		/*
		 * If we are profiling ourself then gprof's mcleanup() is about to
		 * write out a profile to ./gmon.out.  Since mcleanup() always uses a
		 * fixed file name, each backend will overwrite earlier profiles. To
		 * fix that, we create a separate subdirectory for each backend
		 * (./gprof/pid) and 'cd' to that subdirectory before we exit() - that
		 * forces mcleanup() to write each profile into its own directory.  We
		 * end up with something like: $PGDATA/gprof/8829/gmon.out
		 * $PGDATA/gprof/8845/gmon.out ...
		 *
		 * To avoid undesirable disk space bloat, autovacuum workers are
		 * discriminated against: all their gmon.out files go into the same
		 * subdirectory.  Without this, an installation that is "just sitting
		 * there" nonetheless eats megabytes of disk space every few seconds.
		 *
		 * Note that we do this here instead of in an on_proc_exit() callback
		 * because we want to ensure that this code executes last - we don't
		 * want to interfere with any other on_proc_exit() callback.  For the
		 * same reason, we do not include it in proc_exit_prepare ... so if
		 * you are exiting in the "wrong way" you won't drop your profile in a
		 * nice place.
		 */
		char		gprofDirName[32];

		if (IsAutoVacuumWorkerProcess())
			snprintf(gprofDirName, 32, "gprof/avworker");
		else
			snprintf(gprofDirName, 32, "gprof/%d", (int) getpid());

		mkdir("gprof", S_IRWXU | S_IRWXG | S_IRWXO);
		mkdir(gprofDirName, S_IRWXU | S_IRWXG | S_IRWXO);
		chdir(gprofDirName);
	}
#endif

	elog(DEBUG3, "exit(%d)", code);

	exit(code);
}

/*
 * setQFout
 * -- handler for -o command line option and \o command
 *
 * Tries to open file fname (or pipe if fname starts with '|')
 * and stores the file handle in pset)
 * Upon failure, sets stdout and returns false.
 */
bool
setQFout(const char *fname)
{
	bool		status = true;

	/* Close old file/pipe */
	if (pset.queryFout && pset.queryFout != stdout && pset.queryFout != stderr)
	{
		if (pset.queryFoutPipe)
			pclose(pset.queryFout);
		else
			fclose(pset.queryFout);
	}

	/* If no filename, set stdout */
	if (!fname || fname[0] == '\0')
	{
		pset.queryFout = stdout;
		pset.queryFoutPipe = false;
	}
	else if (*fname == '|')
	{
		pset.queryFout = popen(fname + 1, "w");
		pset.queryFoutPipe = true;
	}
	else
	{
		pset.queryFout = fopen(fname, "w");
		pset.queryFoutPipe = false;
	}

	if (!(pset.queryFout))
	{
		psql_error("%s: %s\n", fname, strerror(errno));
		pset.queryFout = stdout;
		pset.queryFoutPipe = false;
		status = false;
	}

	/* Direct signals */
#ifndef WIN32
	pqsignal(SIGPIPE, pset.queryFoutPipe ? SIG_IGN : SIG_DFL);
#endif

	return status;
}

/*
 * Initialize timeout module.
 *
 * This must be called in every process that wants to use timeouts.
 *
 * If the process was forked from another one that was also using this
 * module, be sure to call this before re-enabling signals; else handlers
 * meant to run in the parent process might get invoked in this one.
 */
void
InitializeTimeouts(void)
{
	int			i;

	/* Initialize, or re-initialize, all local state */
	disable_alarm();

	num_active_timeouts = 0;

	for (i = 0; i < MAX_TIMEOUTS; i++)
	{
		all_timeouts[i].index = i;
		all_timeouts[i].indicator = false;
		all_timeouts[i].timeout_handler = NULL;
		all_timeouts[i].start_time = 0;
		all_timeouts[i].fin_time = 0;
	}

	all_timeouts_initialized = true;

	/* Now establish the signal handler */
	pqsignal(SIGALRM, handle_sig_alarm);
}

//.........这里部分代码省略.........
	fprintf(output, "\n");

	fprintf(output, _("Input/Output\n"));
	fprintf(output, _("  \\copy ...              perform SQL COPY with data stream to the client host\n"));
	fprintf(output, _("  \\echo [STRING]         write string to standard output\n"));
	fprintf(output, _("  \\i FILE                execute commands from file\n"));
	fprintf(output, _("  \\o [FILE]              send all query results to file or |pipe\n"));
	fprintf(output, _("  \\qecho [STRING]        write string to query output stream (see \\o)\n"));
	fprintf(output, "\n");

	fprintf(output, _("Informational\n"));
	fprintf(output, _("  (options: S = show system objects, + = additional detail)\n"));
	fprintf(output, _("  \\d[S+]                 list tables, views, and sequences\n"));
	fprintf(output, _("  \\d[S+]  NAME           describe table, view, sequence, or index\n"));
	fprintf(output, _("  \\da[S]  [PATTERN]      list aggregates\n"));
	fprintf(output, _("  \\db[+]  [PATTERN]      list tablespaces\n"));
	fprintf(output, _("  \\dc[S]  [PATTERN]      list conversions\n"));
	fprintf(output, _("  \\dC     [PATTERN]      list casts\n"));
	fprintf(output, _("  \\dd[S]  [PATTERN]      show comments on objects\n"));
	fprintf(output, _("  \\ddp    [PATTERN]      list default privileges\n"));
	fprintf(output, _("  \\dD[S]  [PATTERN]      list domains\n"));
	fprintf(output, _("  \\des[+] [PATTERN]      list foreign servers\n"));
	fprintf(output, _("  \\deu[+] [PATTERN]      list user mappings\n"));
	fprintf(output, _("  \\dew[+] [PATTERN]      list foreign-data wrappers\n"));
	fprintf(output, _("  \\df[antw][S+] [PATRN]  list [only agg/normal/trigger/window] functions\n"));
	fprintf(output, _("  \\dF[+]  [PATTERN]      list text search configurations\n"));
	fprintf(output, _("  \\dFd[+] [PATTERN]      list text search dictionaries\n"));
	fprintf(output, _("  \\dFp[+] [PATTERN]      list text search parsers\n"));
	fprintf(output, _("  \\dFt[+] [PATTERN]      list text search templates\n"));
	fprintf(output, _("  \\dg[+]  [PATTERN]      list roles (groups)\n"));
	fprintf(output, _("  \\dx[+]  [PATTERN]      list extensions\n"));
	fprintf(output, _("  \\di[S+] [PATTERN]      list indexes\n"));
	fprintf(output, _("  \\dl                    list large objects, same as \\lo_list\n"));
	fprintf(output, _("  \\dn[+]  [PATTERN]      list schemas\n"));
	fprintf(output, _("  \\do[S]  [PATTERN]      list operators\n"));
	fprintf(output, _("  \\dp     [PATTERN]      list table, view, and sequence access privileges\n"));
	fprintf(output, _("  \\dr[S+] [PATTERN]      list foreign tables\n"));  /* GPDB Only */
	fprintf(output, _("  \\drds [PATRN1 [PATRN2]] list per-database role settings\n"));
	fprintf(output, _("  \\ds[S+] [PATTERN]      list sequences\n"));
	fprintf(output, _("  \\dt[S+] [PATTERN]      list tables\n"));
	fprintf(output, _("  \\dT[S+] [PATTERN]      list data types\n"));
	fprintf(output, _("  \\du[+]  [PATTERN]      list roles (users)\n"));
	fprintf(output, _("  \\dv[S+] [PATTERN]      list views\n"));
	fprintf(output, _("  \\dE     [PATTERN]      list external tables\n"));
	fprintf(output, _("  \\l[+]                  list all databases\n"));
	fprintf(output, _("  \\z      [PATTERN]      same as \\dp\n"));
	fprintf(output, "\n");

	fprintf(output, _("Formatting\n"));
	fprintf(output, _("  \\a                     toggle between unaligned and aligned output mode\n"));
	fprintf(output, _("  \\C [STRING]            set table title, or unset if none\n"));
	fprintf(output, _("  \\f [STRING]            show or set field separator for unaligned query output\n"));
	fprintf(output, _("  \\H                     toggle HTML output mode (currently %s)\n"),
			ON(pset.popt.topt.format == PRINT_HTML));
	fprintf(output, _("  \\pset NAME [VALUE]     set table output option\n"
					  "                         (NAME := {format|border|expanded|fieldsep|footer|null|\n"
					  "                         numericlocale|recordsep|tuples_only|title|tableattr|pager})\n"));
	fprintf(output, _("  \\t [on|off]            show only rows (currently %s)\n"),
			ON(pset.popt.topt.tuples_only));
	fprintf(output, _("  \\T [STRING]            set HTML <table> tag attributes, or unset if none\n"));
	fprintf(output, _("  \\x [on|off]            toggle expanded output (currently %s)\n"),
			ON(pset.popt.topt.expanded));
	fprintf(output, "\n");

	fprintf(output, _("Connection\n"));
	fprintf(output, _("  \\c[onnect] [DBNAME|- USER|- HOST|- PORT|-]\n"
	"                         connect to new database (currently \"%s\")\n"),
			PQdb(pset.db));
	fprintf(output, _("  \\encoding [ENCODING]   show or set client encoding\n"));
	fprintf(output, _("  \\password [USERNAME]   securely change the password for a user\n"));
	fprintf(output, _("  \\conninfo              display information about current connection\n"));
	fprintf(output, "\n");

	fprintf(output, _("Operating System\n"));
	fprintf(output, _("  \\cd [DIR]              change the current working directory\n"));
	fprintf(output, _("  \\timing [on|off]       toggle timing of commands (currently %s)\n"),
			ON(pset.timing));
	fprintf(output, _("  \\! [COMMAND]           execute command in shell or start interactive shell\n"));
	fprintf(output, "\n");

	fprintf(output, _("Variables\n"));
	fprintf(output, _("  \\prompt [TEXT] NAME    prompt user to set internal variable\n"));
	fprintf(output, _("  \\set [NAME [VALUE]]    set internal variable, or list all if no parameters\n"));
	fprintf(output, _("  \\unset NAME            unset (delete) internal variable\n"));
	fprintf(output, "\n");

	fprintf(output, _("Large Objects\n"));
	fprintf(output, _("  \\lo_export LOBOID FILE\n"
					  "  \\lo_import FILE [COMMENT]\n"
					  "  \\lo_list\n"
					  "  \\lo_unlink LOBOID      large object operations\n"));

	if (output != stdout)
	{
		pclose(output);
#ifndef WIN32
		pqsignal(SIGPIPE, SIG_DFL);
#endif
	}
}

/* Set up signal handlers */
void
WalSndSignals(void)
{
	/* Set up signal handlers */
	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 */
	pqsignal(SIGALRM, SIG_IGN);
	pqsignal(SIGPIPE, SIG_IGN);
	pqsignal(SIGUSR1, WalSndXLogSendHandler);	/* request WAL sending */
	pqsignal(SIGUSR2, WalSndLastCycleHandler);	/* request a last cycle and
												 * shutdown */

	/* Reset some signals that are accepted by postmaster but not here */
	pqsignal(SIGCHLD, SIG_DFL);
	pqsignal(SIGTTIN, SIG_DFL);
	pqsignal(SIGTTOU, SIG_DFL);
	pqsignal(SIGCONT, SIG_DFL);
	pqsignal(SIGWINCH, SIG_DFL);
}

static void
FileRepSubProcess_ConfigureSignals(void)
{
	/* Accept Signals */
	/* emergency shutdown */
	pqsignal(SIGQUIT, FileRepSubProcess_ImmediateShutdownHandler);

	/* graceful shutdown */
	pqsignal(SIGUSR2, FileRepSubProcess_ShutdownHandler);

	/* reload configuration file */
	pqsignal(SIGHUP, FileRepSubProcess_SigHupHandler);

	/* data or segment state changed */
	pqsignal(SIGUSR1, FileRepSubProcess_FileRepStateHandler);

	/* Ignore Signals */
	pqsignal(SIGTERM, SIG_IGN);
	pqsignal(SIGALRM, SIG_IGN);
	pqsignal(SIGPIPE, SIG_IGN);

	/* Use default action */
	pqsignal(SIGCHLD, SIG_DFL);
	pqsignal(SIGINT, SIG_DFL);
	pqsignal(SIGTTIN, SIG_DFL);
	pqsignal(SIGTTOU, SIG_DFL);
	pqsignal(SIGCONT, SIG_DFL);
	pqsignal(SIGWINCH, SIG_DFL);

#ifdef SIGSEGV
	pqsignal(SIGSEGV, FileRepSubProcess_HandleCrash);
#endif

#ifdef SIGILL
	pqsignal(SIGILL, FileRepSubProcess_HandleCrash);
#endif

#ifdef SIGBUS
	pqsignal(SIGBUS, FileRepSubProcess_HandleCrash);
#endif

}

//.........这里部分代码省略.........
				screen_size.ws_col == 0 ||
				screen_size.ws_row == 0)
			{
				screen_size.ws_row = 24;
				screen_size.ws_col = 80;
			}
#else
			screen_size.ws_row = 24;
			screen_size.ws_col = 80;
#endif
			pagerenv = getenv("PAGER");
			if (pagerenv != NULL &&
				pagerenv[0] != '\0' &&
				!po->html3 &&
				((po->expanded &&
				  nTups * (nFields + 1) >= screen_size.ws_row) ||
				 (!po->expanded &&
				  nTups * (total_line_length / screen_size.ws_col + 1) *
				  (1 + (po->standard != 0)) >= screen_size.ws_row -
				  (po->header != 0) *
				  (total_line_length / screen_size.ws_col + 1) * 2
				  - (po->header != 0) * 2		/* row count and newline */
				  )))
			{
				fout = popen(pagerenv, "w");
				if (fout)
				{
					usePipe = 1;
#ifndef WIN32
#ifdef ENABLE_THREAD_SAFETY
					if (pq_block_sigpipe(&osigset, &sigpipe_pending) == 0)
						sigpipe_masked = true;
#else
					oldsigpipehandler = pqsignal(SIGPIPE, SIG_IGN);
#endif   /* ENABLE_THREAD_SAFETY */
#endif   /* WIN32 */
				}
				else
					fout = stdout;
			}
		}

		if (!po->expanded && (po->align || po->html3))
		{
			if (!(fields = (char **) calloc(nFields * (nTups + 1), sizeof(char *))))
			{
				fprintf(stderr, libpq_gettext("out of memory\n"));
				abort();
			}
		}
		else if (po->header && !po->html3)
		{
			if (po->expanded)
			{
				if (po->align)
					fprintf(fout, libpq_gettext("%-*s%s Value\n"),
							fieldMaxLen - fs_len, libpq_gettext("Field"), po->fieldSep);
				else
					fprintf(fout, libpq_gettext("%s%sValue\n"), libpq_gettext("Field"), po->fieldSep);
			}
			else
			{
				int			len = 0;

				for (j = 0; j < nFields; j++)
				{

void
worker_spi_main(Datum main_arg)
{
	int			index = DatumGetInt32(main_arg);
	worktable  *table;
	StringInfoData buf;
	char		name[20];

	table = palloc(sizeof(worktable));
	sprintf(name, "schema%d", index);
	table->schema = pstrdup(name);
	table->name = pstrdup("counted");

	/* Establish signal handlers before unblocking signals. */
	pqsignal(SIGHUP, worker_spi_sighup);
	pqsignal(SIGTERM, worker_spi_sigterm);

	/* We're now ready to receive signals */
	BackgroundWorkerUnblockSignals();

	/* Connect to our database */
	BackgroundWorkerInitializeConnection("postgres", NULL);

	elog(LOG, "%s initialized with %s.%s",
		 MyBgworkerEntry->bgw_name, table->schema, table->name);
	initialize_worker_spi(table);

	/*
	 * Quote identifiers passed to us.  Note that this must be done after
	 * initialize_worker_spi, because that routine assumes the names are not
	 * quoted.
	 *
	 * Note some memory might be leaked here.
	 */
	table->schema = quote_identifier(table->schema);
	table->name = quote_identifier(table->name);

	initStringInfo(&buf);
	appendStringInfo(&buf,
					 "WITH deleted AS (DELETE "
					 "FROM %s.%s "
					 "WHERE type = 'delta' RETURNING value), "
					 "total AS (SELECT coalesce(sum(value), 0) as sum "
					 "FROM deleted) "
					 "UPDATE %s.%s "
					 "SET value = %s.value + total.sum "
					 "FROM total WHERE type = 'total' "
					 "RETURNING %s.value",
					 table->schema, table->name,
					 table->schema, table->name,
					 table->name,
					 table->name);

	/*
	 * Main loop: do this until the SIGTERM handler tells us to terminate
	 */
	while (!got_sigterm)
	{
		int			ret;
		int			rc;

		/*
		 * Background workers mustn't call usleep() or any direct equivalent:
		 * instead, they may wait on their process latch, which sleeps as
		 * necessary, but is awakened if postmaster dies.  That way the
		 * background process goes away immediately in an emergency.
		 */
		rc = WaitLatch(&MyProc->procLatch,
					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
					   worker_spi_naptime * 1000L);
		ResetLatch(&MyProc->procLatch);

		/* emergency bailout if postmaster has died */
		if (rc & WL_POSTMASTER_DEATH)
			proc_exit(1);

		/*
		 * In case of a SIGHUP, just reload the configuration.
		 */
		if (got_sighup)
		{
			got_sighup = false;
			ProcessConfigFile(PGC_SIGHUP);
		}

		/*
		 * Start a transaction on which we can run queries.  Note that each
		 * StartTransactionCommand() call should be preceded by a
		 * SetCurrentStatementStartTimestamp() call, which sets both the time
		 * for the statement we're about the run, and also the transaction
		 * start time.  Also, each other query sent to SPI should probably be
		 * preceded by SetCurrentStatementStartTimestamp(), so that statement
		 * start time is always up to date.
		 *
		 * The SPI_connect() call lets us run queries through the SPI manager,
		 * and the PushActiveSnapshot() call creates an "active" snapshot
		 * which is necessary for queries to have MVCC data to work on.
		 *
		 * The pgstat_report_activity() call makes our activity visible
		 * through the pgstat views.
//.........这里部分代码省略.........

/*
 * kafka_consume_main
 *
 * Main function for Kafka consumers running as background workers
 */
void
kafka_consume_main(Datum arg)
{
	char err_msg[512];
	rd_kafka_topic_conf_t *topic_conf;
	rd_kafka_t *kafka;
	rd_kafka_topic_t *topic;
	rd_kafka_message_t **messages;
	const struct rd_kafka_metadata *meta;
	struct rd_kafka_metadata_topic topic_meta;
	rd_kafka_resp_err_t err;
	bool found;
	Oid id = (Oid) arg;
	ListCell *lc;
	KafkaConsumerProc *proc = hash_search(consumer_procs, &id, HASH_FIND, &found);
	KafkaConsumer consumer;
	CopyStmt *copy;
	int valid_brokers = 0;
	int i;
	int my_partitions = 0;

	if (!found)
		elog(ERROR, "kafka consumer %d not found", id);

	pqsignal(SIGTERM, kafka_consume_main_sigterm);
#define BACKTRACE_SEGFAULTS
#ifdef BACKTRACE_SEGFAULTS
	pqsignal(SIGSEGV, debug_segfault);
#endif

	/* we're now ready to receive signals */
	BackgroundWorkerUnblockSignals();

	/* give this proc access to the database */
	BackgroundWorkerInitializeConnection(NameStr(proc->dbname), NULL);

	/* load saved consumer state */
	StartTransactionCommand();
	load_consumer_state(proc->consumer_id, &consumer);
	copy = get_copy_statement(&consumer);

	topic_conf = rd_kafka_topic_conf_new();
	kafka = rd_kafka_new(RD_KAFKA_CONSUMER, NULL, err_msg, sizeof(err_msg));
	rd_kafka_set_logger(kafka, logger);

	/*
	 * Add all brokers currently in pipeline_kafka_brokers
	 */
	if (consumer.brokers == NIL)
		elog(ERROR, "no valid brokers were found");

	foreach(lc, consumer.brokers)
		valid_brokers += rd_kafka_brokers_add(kafka, lfirst(lc));

	if (!valid_brokers)
		elog(ERROR, "no valid brokers were found");

	/*
	 * Set up our topic to read from
	 */
	topic = rd_kafka_topic_new(kafka, consumer.topic, topic_conf);
	err = rd_kafka_metadata(kafka, false, topic, &meta, CONSUMER_TIMEOUT);

	if (err != RD_KAFKA_RESP_ERR_NO_ERROR)
		elog(ERROR, "failed to acquire metadata: %s", rd_kafka_err2str(err));

	Assert(meta->topic_cnt == 1);
	topic_meta = meta->topics[0];

	load_consumer_offsets(&consumer, &topic_meta, proc->offset);
	CommitTransactionCommand();

	/*
	* Begin consuming all partitions that this process is responsible for
	*/
	for (i = 0; i < topic_meta.partition_cnt; i++)
	{
		int partition = topic_meta.partitions[i].id;

		Assert(partition <= consumer.num_partitions);
		if (partition % consumer.parallelism != proc->partition_group)
			continue;

		elog(LOG, "[kafka consumer] %s <- %s consuming partition %d from offset %ld",
				consumer.rel->relname, consumer.topic, partition, consumer.offsets[partition]);

		if (rd_kafka_consume_start(topic, partition, consumer.offsets[partition]) == -1)
			elog(ERROR, "failed to start consuming: %s", rd_kafka_err2str(rd_kafka_errno2err(errno)));

		my_partitions++;
	}

	/*
	* No point doing anything if we don't have any partitions assigned to us
	*/
//.........这里部分代码省略.........