自学教程：C++ CommitTransactionCommand函数代码示例

finish_sync_worker(void){	/*	 * Commit any outstanding transaction. This is the usual case, unless	 * there was nothing to do for the table.	 */	if (IsTransactionState())	{		CommitTransactionCommand();		pgstat_report_stat(false);	}	/* And flush all writes. */	XLogFlush(GetXLogWriteRecPtr());	StartTransactionCommand();	ereport(LOG,			(errmsg("logical replication table synchronization worker for subscription /"%s/", table /"%s/" has finished",					MySubscription->name,					get_rel_name(MyLogicalRepWorker->relid))));	CommitTransactionCommand();	/* Find the main apply worker and signal it. */	logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);	/* Stop gracefully */	proc_exit(0);}

/* *  Ensure that the environment is sane.   *    This involves checking the Postgresql version, and if in network mode *      also establishing a connection to a receiver.*/int ensure_valid_environment(void) {  StringInfoData buf;  int     retval;	char* pgversion;	  SPITupleTable *coltuptable;   SetCurrentStatementStartTimestamp();  StartTransactionCommand();  SPI_connect();  PushActiveSnapshot(GetTransactionSnapshot());    /* Ensure compatible version */  pgstat_report_activity(STATE_RUNNING, "verifying compatible postgres version");  initStringInfo(&buf);  appendStringInfo(&buf,     "select version();"  );  retval = SPI_execute(buf.data, false, 0);  if (retval != SPI_OK_SELECT) {    elog(FATAL, "Unable to query postgres version %d", retval);    SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();  	return 1;    }  	coltuptable = SPI_tuptable;	pgversion = SPI_getvalue(coltuptable->vals[0], coltuptable->tupdesc, 1);	  if(strstr(pgversion, "PostgreSQL 9.3") == NULL) {    elog(FATAL, "Unsupported Postgresql version");    SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();  	return 1;	}  	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();    /*   * Attempt to establish a connection if the output mode is network.   */  if (strcmp(output_mode, "network") == 0) {		retval = establish_connection();		if (retval == 2) {			elog(LOG, "Error : Failed to connect to antenna please check domain is available from host.");		}	}		//TODO verify logging directory is accessible when csv mode.	  elog(LOG, "Pgsampler Initialized");  return 0;}

/* * Returns a count of the number of non-template databases from the catalog. */int get_database_count(void) {  int retval, processed;	StringInfoData buf;	SPITupleTable *coltuptable;	int database_count = 0;    SetCurrentStatementStartTimestamp();  StartTransactionCommand();  SPI_connect();  PushActiveSnapshot(GetTransactionSnapshot());    initStringInfo(&buf);  appendStringInfo(&buf, "SELECT count(*) FROM pg_database WHERE datname NOT IN ('template0', 'template1') AND datallowconn IS TRUE;");    retval = SPI_execute(buf.data, false, 0);  if (retval != SPI_OK_SELECT) {    elog(FATAL, "Database information collection failed");    // FAIL RETURN 1  }    processed = SPI_processed;    if (processed > 0) {    coltuptable = SPI_tuptable;      database_count = atoi(SPI_getvalue(coltuptable->vals[0], coltuptable->tupdesc, 1));  }    SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();     return database_count;}

/* * ProcessCatchupInterrupt * * The portion of catchup interrupt handling that runs outside of the signal * handler, which allows it to actually process pending invalidations. */voidProcessCatchupInterrupt(void){	while (catchupInterruptPending)	{		/*		 * What we need to do here is cause ReceiveSharedInvalidMessages() to		 * run, which will do the necessary work and also reset the		 * catchupInterruptPending flag.  If we are inside a transaction we		 * can just call AcceptInvalidationMessages() to do this.  If we		 * aren't, we start and immediately end a transaction; the call to		 * AcceptInvalidationMessages() happens down inside transaction start.		 *		 * It is awfully tempting to just call AcceptInvalidationMessages()		 * without the rest of the xact start/stop overhead, and I think that		 * would actually work in the normal case; but I am not sure that things		 * would clean up nicely if we got an error partway through.		 */		if (IsTransactionOrTransactionBlock())		{			elog(DEBUG4, "ProcessCatchupEvent inside transaction");			AcceptInvalidationMessages();		}		else		{			elog(DEBUG4, "ProcessCatchupEvent outside transaction");			StartTransactionCommand();			CommitTransactionCommand();		}	}}

/* * Main worker routine. Accepts dsm_handle as an argument */static voidbg_worker_main(Datum main_arg){	PartitionArgs  *args;	dsm_handle		handle = DatumGetInt32(main_arg);	/* Create resource owner */	CurrentResourceOwner = ResourceOwnerCreate(NULL, "CreatePartitionsWorker");	/* Attach to dynamic shared memory */	if (!handle)	{		ereport(WARNING,                (errmsg("pg_pathman worker: invalid dsm_handle")));	}	segment = dsm_attach(handle);	args = dsm_segment_address(segment);	/* Establish connection and start transaction */	BackgroundWorkerInitializeConnectionByOid(args->dbid, InvalidOid);	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	/* Create partitions */	args->result = create_partitions(args->relid, PATHMAN_GET_DATUM(args->value, args->by_val), args->value_type, &args->crashed);	/* Cleanup */	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	dsm_detach(segment);}

/* * We keep some resources across transactions, so we attach everything to a * long-lived ResourceOwner, which prevents the below commit from thinking that * there are reference leaks */static voidstart_executor(QueryDesc *queryDesc, MemoryContext context, ResourceOwner owner){	MemoryContext old;	ResourceOwner save;	StartTransactionCommand();	old = MemoryContextSwitchTo(context);	save = CurrentResourceOwner;	CurrentResourceOwner = owner;	queryDesc->snapshot = GetTransactionSnapshot();	queryDesc->snapshot->copied = true;	RegisterSnapshotOnOwner(queryDesc->snapshot, owner);	ExecutorStart(queryDesc, 0);	queryDesc->snapshot->active_count++;	UnregisterSnapshotFromOwner(queryDesc->snapshot, owner);	UnregisterSnapshotFromOwner(queryDesc->estate->es_snapshot, owner);	CurrentResourceOwner = TopTransactionResourceOwner;	MemoryContextSwitchTo(old);	CommitTransactionCommand();	CurrentResourceOwner = save;}

/* * get_database_oids * * Returns a list of all database OIDs found in pg_database. */static List *get_database_list(void){	List *dbs = NIL;	Relation rel;	HeapScanDesc scan;	HeapTuple tup;	MemoryContext resultcxt;	/* This is the context that we will allocate our output data in */	resultcxt = CurrentMemoryContext;	/*	 * Start a transaction so we can access pg_database, and get a snapshot.	 * We don't have a use for the snapshot itself, but we're interested in	 * the secondary effect that it sets RecentGlobalXmin.  (This is critical	 * for anything that reads heap pages, because HOT may decide to prune	 * them even if the process doesn't attempt to modify any tuples.)	 */	StartTransactionCommand();	(void) GetTransactionSnapshot();	/* We take a AccessExclusiveLock so we don't conflict with any DATABASE commands */	rel = heap_open(DatabaseRelationId, AccessExclusiveLock);	scan = heap_beginscan_catalog(rel, 0, NULL);	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))	{		MemoryContext oldcxt;		Form_pg_database pgdatabase = (Form_pg_database) GETSTRUCT(tup);		DatabaseEntry *db_entry;		/* Ignore template databases or ones that don't allow connections. */		if (pgdatabase->datistemplate || !pgdatabase->datallowconn)			continue;		/*		 * Allocate our results in the caller's context, not the		 * transaction's. We do this inside the loop, and restore the original		 * context at the end, so that leaky things like heap_getnext() are		 * not called in a potentially long-lived context.		 */		oldcxt = MemoryContextSwitchTo(resultcxt);		db_entry = palloc0(sizeof(DatabaseEntry));		db_entry->oid = HeapTupleGetOid(tup);		StrNCpy(NameStr(db_entry->name), NameStr(pgdatabase->datname), NAMEDATALEN);		dbs = lappend(dbs, db_entry);		MemoryContextSwitchTo(oldcxt);	}	heap_endscan(scan);	heap_close(rel, AccessExclusiveLock);	CommitTransactionCommand();	return dbs;}

/* * Load the list of subscriptions. * * Only the fields interesting for worker start/stop functions are filled for * each subscription. */static List *get_subscription_list(void){	List	   *res = NIL;	Relation	rel;	TableScanDesc scan;	HeapTuple	tup;	MemoryContext resultcxt;	/* This is the context that we will allocate our output data in */	resultcxt = CurrentMemoryContext;	/*	 * Start a transaction so we can access pg_database, and get a snapshot.	 * We don't have a use for the snapshot itself, but we're interested in	 * the secondary effect that it sets RecentGlobalXmin.  (This is critical	 * for anything that reads heap pages, because HOT may decide to prune	 * them even if the process doesn't attempt to modify any tuples.)	 */	StartTransactionCommand();	(void) GetTransactionSnapshot();	rel = table_open(SubscriptionRelationId, AccessShareLock);	scan = table_beginscan_catalog(rel, 0, NULL);	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))	{		Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup);		Subscription *sub;		MemoryContext oldcxt;		/*		 * Allocate our results in the caller's context, not the		 * transaction's. We do this inside the loop, and restore the original		 * context at the end, so that leaky things like heap_getnext() are		 * not called in a potentially long-lived context.		 */		oldcxt = MemoryContextSwitchTo(resultcxt);		sub = (Subscription *) palloc0(sizeof(Subscription));		sub->oid = subform->oid;		sub->dbid = subform->subdbid;		sub->owner = subform->subowner;		sub->enabled = subform->subenabled;		sub->name = pstrdup(NameStr(subform->subname));		/* We don't fill fields we are not interested in. */		res = lappend(res, sub);		MemoryContextSwitchTo(oldcxt);	}	table_endscan(scan);	table_close(rel, AccessShareLock);	CommitTransactionCommand();	return res;}

/* * Initialize workspace for a worker process: create the schema if it doesn't * already exist. */static voidinitialize_worker_spi(worktable *table){	int			ret;	int			ntup;	bool		isnull;	StringInfoData buf;	SetCurrentStatementStartTimestamp();	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	pgstat_report_activity(STATE_RUNNING, "initializing spi_worker schema");	/* XXX could we use CREATE SCHEMA IF NOT EXISTS? */	initStringInfo(&buf);	appendStringInfo(&buf, "select count(*) from pg_namespace where nspname = '%s'",					 table->schema);	ret = SPI_execute(buf.data, true, 0);	if (ret != SPI_OK_SELECT)		elog(FATAL, "SPI_execute failed: error code %d", ret);	if (SPI_processed != 1)		elog(FATAL, "not a singleton result");	ntup = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[0],									   SPI_tuptable->tupdesc,									   1, &isnull));	if (isnull)		elog(FATAL, "null result");	if (ntup == 0)	{		resetStringInfo(&buf);		appendStringInfo(&buf,						 "CREATE SCHEMA /"%s/" "						 "CREATE TABLE /"%s/" ("			   "		type text CHECK (type IN ('total', 'delta')), "						 "		value	integer)"				  "CREATE UNIQUE INDEX /"%s_unique_total/" ON /"%s/" (type) "						 "WHERE type = 'total'",					   table->schema, table->name, table->name, table->name);		/* set statement start time */		SetCurrentStatementStartTimestamp();		ret = SPI_execute(buf.data, false, 0);		if (ret != SPI_OK_UTILITY)			elog(FATAL, "failed to create my schema");	}	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	pgstat_report_activity(STATE_IDLE, NULL);}

static voiddo_end(void){	CommitTransactionCommand();	elog(DEBUG4, "commit transaction");	CHECK_FOR_INTERRUPTS();		/* allow SIGINT to kill bootstrap run */	if (isatty(0))	{		printf("bootstrap> ");		fflush(stdout);	}}

static voidexecute_pg_settings_logger(config_log_objects *objects) {	int		ret;	bool	isnull;	StringInfoData	buf;	SetCurrentStatementStartTimestamp();	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	pgstat_report_activity(STATE_RUNNING, "executing configuration logger function");	initStringInfo(&buf);	appendStringInfo(		&buf,		"SELECT %s.%s()",		config_log_schema,		objects->function_name		);	ret = SPI_execute(buf.data, false, 0);	if (ret != SPI_OK_SELECT)	{		elog(FATAL, "SPI_execute failed: error code %d", ret);	}	if (SPI_processed != 1)	{		elog(FATAL, "not a singleton result");	}	log_info("pg_settings_logger() executed");	if(DatumGetBool(SPI_getbinval(SPI_tuptable->vals[0],				  SPI_tuptable->tupdesc,				  1, &isnull)))	{		log_info("Configuration changes recorded");	}	else	{		log_info("No configuration changes detected");	}   	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	pgstat_report_activity(STATE_IDLE, NULL);}

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

/* * This function will set a string in shared memory which is the name of the database to connect to *  the next time the background worker restarts.  Because a bgworker can only connect to one database *  at a time, and some catalogs and stats are scoped to the current database, the bg worker *  periodically restarts to collect latest stats from another database. **/int set_next_db_target(void) {	int retval, processed;	StringInfoData buf;	SPITupleTable *coltuptable;	char* next_db_target;    SetCurrentStatementStartTimestamp();  StartTransactionCommand();  SPI_connect();  PushActiveSnapshot(GetTransactionSnapshot());    /* get sorted list of databases, find one after target_db*/  initStringInfo(&buf);  appendStringInfo(&buf,     "SELECT datname FROM pg_database WHERE datname NOT IN ('template0', 'template1') AND datallowconn IS TRUE AND datname > '%s' ORDER BY datname ASC LIMIT 1;", target_db  );    retval = SPI_execute(buf.data, false, 0);  if (retval != SPI_OK_SELECT) {    elog(FATAL, "Database information collection failed");    // FAIL RETURN 1  }    processed = SPI_processed;    if(processed == 0) {    //No matching records so pick first database.    resetStringInfo(&buf);    appendStringInfoString(&buf,       "SELECT datname FROM pg_database WHERE datname NOT IN ('template0', 'template1') AND datallowconn IS TRUE ORDER BY datname ASC LIMIT 1;"    );    retval = SPI_execute(buf.data, false, 0);    if (retval != SPI_OK_SELECT) {      elog(FATAL, "Database information collection failed");      // FAIL RETURN 1    }  }    coltuptable = SPI_tuptable;    next_db_target = SPI_getvalue(coltuptable->vals[0], coltuptable->tupdesc, 1);  // elog(LOG, "NEXTDB TARGET: %s", next_db_target); //print next target db  strcpy(pgsampler_state->next_db, next_db_target);  	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();     return 0;}

/* *-------------------------------------------------------------- * Async_UnlistenOnExit * *		Clean up the pg_listener table at backend exit. * *		This is executed if we have done any LISTENs in this backend. *		It might not be necessary anymore, if the user UNLISTENed everything, *		but we don't try to detect that case. * * Results: *		XXX * * Side effects: *		pg_listener is updated if necessary. * *-------------------------------------------------------------- */static voidAsync_UnlistenOnExit(int code, Datum arg){	/*	 * We need to start/commit a transaction for the unlisten, but if there is	 * already an active transaction we had better abort that one first.	 * Otherwise we'd end up committing changes that probably ought to be	 * discarded.	 */	AbortOutOfAnyTransaction();	/* Now we can do the unlisten */	StartTransactionCommand();	Async_UnlistenAll();	CommitTransactionCommand();}

/* * Handle COMMIT message. * * TODO, support tracking of multiple origins */static voidapply_handle_commit(StringInfo s){	LogicalRepCommitData commit_data;	logicalrep_read_commit(s, &commit_data);	Assert(commit_data.commit_lsn == remote_final_lsn);	/* The synchronization worker runs in single transaction. */	if (IsTransactionState() && !am_tablesync_worker())	{		/*		 * Update origin state so we can restart streaming from correct		 * position in case of crash.		 */		replorigin_session_origin_lsn = commit_data.end_lsn;		replorigin_session_origin_timestamp = commit_data.committime;		CommitTransactionCommand();		pgstat_report_stat(false);		store_flush_position(commit_data.end_lsn);	}	else	{		/* Process any invalidation messages that might have accumulated. */		AcceptInvalidationMessages();		maybe_reread_subscription();	}	in_remote_transaction = false;	/* Process any tables that are being synchronized in parallel. */	process_syncing_tables(commit_data.end_lsn);	pgstat_report_activity(STATE_IDLE, NULL);}

/* * ProcessCatchupEvent * * Respond to a catchup event (SIGUSR1) from another backend. * * This is called either directly from the SIGUSR1 signal handler, * or the next time control reaches the outer idle loop (assuming * there's still anything to do by then). */static voidProcessCatchupEvent(void){	bool		notify_enabled;	/* Must prevent SIGUSR2 interrupt while I am running */	notify_enabled = DisableNotifyInterrupt();	/*	 * What we need to do here is cause ReceiveSharedInvalidMessages() to run,	 * which will do the necessary work and also reset the	 * catchupInterruptOccurred flag.  If we are inside a transaction we can	 * just call AcceptInvalidationMessages() to do this.  If we aren't, we	 * start and immediately end a transaction; the call to	 * AcceptInvalidationMessages() happens down inside transaction start.	 *	 * It is awfully tempting to just call AcceptInvalidationMessages()	 * without the rest of the xact start/stop overhead, and I think that	 * would actually work in the normal case; but I am not sure that things	 * would clean up nicely if we got an error partway through.	 */	if (IsTransactionOrTransactionBlock())	{		elog(DEBUG4, "ProcessCatchupEvent inside transaction");		AcceptInvalidationMessages();	}	else	{		elog(DEBUG4, "ProcessCatchupEvent outside transaction");		StartTransactionCommand();		CommitTransactionCommand();	}	if (notify_enabled)		EnableNotifyInterrupt();}

//.........这里部分代码省略.........		ereport(FATAL,				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),		   errmsg("must be superuser to connect during database shutdown")));	/*	 * Check a normal user hasn't connected to a superuser reserved slot.	 */	if (!am_superuser &&		ReservedBackends > 0 &&		!HaveNFreeProcs(ReservedBackends))		ereport(FATAL,				(errcode(ERRCODE_TOO_MANY_CONNECTIONS),				 errmsg("connection limit exceeded for non-superusers")));	/*	 * Now process any command-line switches that were included in the startup	 * packet, if we are in a regular backend.	We couldn't do this before	 * because we didn't know if client is a superuser.	 */	gucctx = am_superuser ? PGC_SUSET : PGC_BACKEND;	if (MyProcPort != NULL &&		MyProcPort->cmdline_options != NULL)	{		/*		 * The maximum possible number of commandline arguments that could		 * come from MyProcPort->cmdline_options is (strlen + 1) / 2; see		 * pg_split_opts().		 */		char	  **av;		int			maxac;		int			ac;		maxac = 2 + (strlen(MyProcPort->cmdline_options) + 1) / 2;		av = (char **) palloc(maxac * sizeof(char *));		ac = 0;		av[ac++] = "postgres";		/* Note this mangles MyProcPort->cmdline_options */		pg_split_opts(av, &ac, MyProcPort->cmdline_options);		av[ac] = NULL;		Assert(ac < maxac);		(void) process_postgres_switches(ac, av, gucctx);	}	/*	 * Process any additional GUC variable settings passed in startup packet.	 * These are handled exactly like command-line variables.	 */	if (MyProcPort != NULL)	{		ListCell   *gucopts = list_head(MyProcPort->guc_options);		while (gucopts)		{			char	   *name;			char	   *value;			name = lfirst(gucopts);			gucopts = lnext(gucopts);			value = lfirst(gucopts);			gucopts = lnext(gucopts);			SetConfigOption(name, value, gucctx, PGC_S_CLIENT);		}	}	/* Apply PostAuthDelay as soon as we've read all options */	if (PostAuthDelay > 0)		pg_usleep(PostAuthDelay * 1000000L);	/*	 * Initialize various default states that can't be set up until we've	 * selected the active user and gotten the right GUC settings.	 */	/* set default namespace search path */	InitializeSearchPath();	/* initialize client encoding */	InitializeClientEncoding();	/* reset the database for walsender */	if (am_walsender)		MyProc->databaseId = MyDatabaseId = InvalidOid;	/* report this backend in the PgBackendStatus array */	if (!bootstrap)		pgstat_bestart();	/* close the transaction we started above */	if (!bootstrap)		CommitTransactionCommand();}

/* * kafka_consume_main * * Main function for Kafka consumers running as background workers */voidkafka_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	*///.........这里部分代码省略.........

//.........这里部分代码省略.........	/*	 * If walsender, we don't want to connect to any particular database. Just	 * finish the backend startup by processing any options from the startup	 * packet, and we're done.	 */	if (am_walsender)	{		Assert(!bootstrap);		/*		 * We don't have replication role, which existed in postgres.		 */		if (!superuser())			ereport(FATAL,					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),					 errmsg("must be superuser role to start walsender")));		/* process any options passed in the startup packet */		if (MyProcPort != NULL)			process_startup_options(MyProcPort, am_superuser);		/* Apply PostAuthDelay as soon as we've read all options */		if (PostAuthDelay > 0)			pg_usleep(PostAuthDelay * 1000000L);		/* initialize client encoding */		InitializeClientEncoding();		/* report this backend in the PgBackendStatus array */		pgstat_bestart();		/* close the transaction we started above */		CommitTransactionCommand();		return;	}	/*	 * Set up the global variables holding database id and path.  But note we	 * won't actually try to touch the database just yet.	 *	 * We take a shortcut in the bootstrap case, otherwise we have to look up	 * the db name in pg_database.	 */	if (bootstrap)	{		MyDatabaseId = TemplateDbOid;		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;	}	else if (in_dbname != NULL)	{		HeapTuple	tuple;		Form_pg_database dbform;		tuple = GetDatabaseTuple(in_dbname);		if (!HeapTupleIsValid(tuple))			ereport(FATAL,					(errcode(ERRCODE_UNDEFINED_DATABASE),					 errmsg("database /"%s/" does not exist", in_dbname)));		dbform = (Form_pg_database) GETSTRUCT(tuple);		MyDatabaseId = HeapTupleGetOid(tuple);		MyDatabaseTableSpace = dbform->dattablespace;		/* take database name from the caller, just for paranoia */		strlcpy(dbname, in_dbname, sizeof(dbname));		pfree(tuple);

//.........这里部分代码省略.........            else                ereport(FATAL,                        (errcode_for_file_access(),                         errmsg("could not access directory /"%s/": %m",                                fullpath)));        }        ValidatePgVersion(fullpath);    }    /*     * It's now possible to do real access to the system catalogs.     *     * Load relcache entries for the system catalogs.  This must create at     * least the minimum set of "nailed-in" cache entries.     */    RelationCacheInitializePhase2();    /*     * Figure out our postgres user id, and see if we are a superuser.     *     * In standalone mode and in the autovacuum process, we use a fixed id,     * otherwise we figure it out from the authenticated user name.     */    if (bootstrap || autovacuum)    {        InitializeSessionUserIdStandalone();        am_superuser = true;    }    else if (!IsUnderPostmaster)    {        InitializeSessionUserIdStandalone();        am_superuser = true;        if (!ThereIsAtLeastOneRole())            ereport(WARNING,                    (errcode(ERRCODE_UNDEFINED_OBJECT),                     errmsg("no roles are defined in this database system"),                     errhint("You should immediately run CREATE USER /"%s/" CREATEUSER;.",                             username)));    }    else    {        /* normal multiuser case */        InitializeSessionUserId(username);        am_superuser = superuser();    }    /* set up ACL framework (so CheckMyDatabase can check permissions) */    initialize_acl();    /*     * Read the real pg_database row for our database, check permissions and     * set up database-specific GUC settings.  We can't do this until all the     * database-access infrastructure is up.  (Also, it wants to know if the     * user is a superuser, so the above stuff has to happen first.)     */    if (!bootstrap)        CheckMyDatabase(dbname, am_superuser);    /*     * If we're trying to shut down, only superusers can connect.     */    if (!am_superuser &&            MyProcPort != NULL &&            MyProcPort->canAcceptConnections == CAC_WAITBACKUP)        ereport(FATAL,                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),                 errmsg("must be superuser to connect during database shutdown")));    /*     * Check a normal user hasn't connected to a superuser reserved slot.     */    if (!am_superuser &&            ReservedBackends > 0 &&            !HaveNFreeProcs(ReservedBackends))        ereport(FATAL,                (errcode(ERRCODE_TOO_MANY_CONNECTIONS),                 errmsg("connection limit exceeded for non-superusers")));    /*     * Initialize various default states that can't be set up until we've     * selected the active user and gotten the right GUC settings.     */    /* set default namespace search path */    InitializeSearchPath();    /* initialize client encoding */    InitializeClientEncoding();    /* report this backend in the PgBackendStatus array */    if (!bootstrap)        pgstat_bestart();    /* close the transaction we started above */    if (!bootstrap)        CommitTransactionCommand();    return am_superuser;}

/* * worker logic */voidwed_worker_main(Datum main_arg){	StringInfoData buf;        	/* Establish signal handlers before unblocking signals. */	pqsignal(SIGHUP, wed_worker_sighup);	pqsignal(SIGTERM, wed_worker_sigterm);	/* We're now ready to receive signals */	BackgroundWorkerUnblockSignals();	/* Connect to our database */	BackgroundWorkerInitializeConnection(wed_worker_db_name, NULL);	elog(LOG, "%s initialized in: %s",		 MyBgworkerEntry->bgw_name, wed_worker_db_name);	initStringInfo(&buf);	appendStringInfo(&buf, "SELECT trcheck()");	/*	 * 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,					   wed_worker_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.		 */		SetCurrentStatementStartTimestamp();		StartTransactionCommand();		SPI_connect();		PushActiveSnapshot(GetTransactionSnapshot());		pgstat_report_activity(STATE_RUNNING, buf.data);		/* We can now execute queries via SPI */		ret = SPI_execute(buf.data, false, 0);		if (ret != SPI_OK_SELECT)			elog(FATAL, "stored procedure trcheck() not found: error code %d", ret);        elog(LOG, "%s : trcheck() done !", MyBgworkerEntry->bgw_name);		/*		 * And finish our transaction.		 */		SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();		pgstat_report_activity(STATE_IDLE, NULL);	}	proc_exit(1);}

/* * -------------------------------------------------------------- * ProcessIncomingNotify * *		Deal with arriving NOTIFYs from other backends. *		This is called either directly from the SIGUSR2 signal handler, *		or the next time control reaches the outer idle loop. *		Scan pg_listener for arriving notifies, report them to my front end, *		and clear the notification field in pg_listener until next time. * *		NOTE: since we are outside any transaction, we must create our own. * -------------------------------------------------------------- */static voidProcessIncomingNotify(void){	Relation	lRel;	TupleDesc	tdesc;	ScanKeyData key[1];	HeapScanDesc scan;	HeapTuple	lTuple,				rTuple;	Datum		value[Natts_pg_listener];	char		repl[Natts_pg_listener],				nulls[Natts_pg_listener];	bool		catchup_enabled;	/* Must prevent SIGUSR1 interrupt while I am running */	catchup_enabled = DisableCatchupInterrupt();	if (Trace_notify)		elog(DEBUG1, "ProcessIncomingNotify");	set_ps_display("notify interrupt", false);	notifyInterruptOccurred = 0;	StartTransactionCommand();	lRel = heap_open(ListenerRelationId, ExclusiveLock);	tdesc = RelationGetDescr(lRel);	/* Scan only entries with my listenerPID */	ScanKeyInit(&key[0],				Anum_pg_listener_pid,				BTEqualStrategyNumber, F_INT4EQ,				Int32GetDatum(MyProcPid));	scan = heap_beginscan(lRel, SnapshotNow, 1, key);	/* Prepare data for rewriting 0 into notification field */	nulls[0] = nulls[1] = nulls[2] = ' ';	repl[0] = repl[1] = repl[2] = ' ';	repl[Anum_pg_listener_notify - 1] = 'r';	value[0] = value[1] = value[2] = (Datum) 0;	value[Anum_pg_listener_notify - 1] = Int32GetDatum(0);	while ((lTuple = heap_getnext(scan, ForwardScanDirection)) != NULL)	{		Form_pg_listener listener = (Form_pg_listener) GETSTRUCT(lTuple);		char	   *relname = NameStr(listener->relname);		int32		sourcePID = listener->notification;		if (sourcePID != 0)		{			/* Notify the frontend */			if (Trace_notify)				elog(DEBUG1, "ProcessIncomingNotify: received %s from %d",					 relname, (int) sourcePID);			NotifyMyFrontEnd(relname, sourcePID);			/*			 * Rewrite the tuple with 0 in notification column.			 *			 * simple_heap_update is safe here because no one else would have			 * tried to UNLISTEN us, so there can be no uncommitted changes.			 */			rTuple = heap_modifytuple(lTuple, tdesc, value, nulls, repl);			simple_heap_update(lRel, &lTuple->t_self, rTuple);#ifdef NOT_USED					/* currently there are no indexes */			CatalogUpdateIndexes(lRel, rTuple);#endif		}	}	heap_endscan(scan);	/*	 * We do NOT release the lock on pg_listener here; we need to hold it	 * until end of transaction (which is about to happen, anyway) to ensure	 * that other backends see our tuple updates when they look. Otherwise, a	 * transaction started after this one might mistakenly think it doesn't	 * need to send this backend a new NOTIFY.	 */	heap_close(lRel, NoLock);	CommitTransactionCommand();	/*//.........这里部分代码省略.........

static voidworker_spi_main(Datum main_arg){	/* Register functions for SIGTERM/SIGHUP management */	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);	while (!got_sigterm)	{		int		ret;		int		rc;		StringInfoData	buf;		/*		 * 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,					   1000L);		ResetLatch(&MyProc->procLatch);		/* emergency bailout if postmaster has died */		if (rc & WL_POSTMASTER_DEATH)			proc_exit(1);		StartTransactionCommand();		SPI_connect();		PushActiveSnapshot(GetTransactionSnapshot());		initStringInfo(&buf);		/* Build the query string */		appendStringInfo(&buf,						 "SELECT count(*) FROM pg_class;");		ret = SPI_execute(buf.data, true, 0);		/* Some error messages in case of incorrect handling */		if (ret != SPI_OK_SELECT)			elog(FATAL, "SPI_execute failed: error code %d", ret);		if (SPI_processed > 0)		{			int32	count;			bool	isnull;			count = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],												 SPI_tuptable->tupdesc,												 1, &isnull));			elog(LOG, "Currently %d relations in database",				 count);		}		SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();	}	proc_exit(0);}

/* * cluster * *   Check that the relation is a relation in the appropriate user *   ACL. I will use the same security that limits users on the *   renamerel() function. * *   Check that the index specified is appropriate for the task *   ( ie it's an index over this relation ). This is trickier. * *   Create a list of all the other indicies on this relation. Because *   the cluster will wreck all the tids, I'll need to destroy bogus *   indicies. The user will have to re-create them. Not nice, but *   I'm not a nice guy. The alternative is to try some kind of post *   destroy re-build. This may be possible. I'll check out what the *   index create functiond want in the way of paramaters. On the other *   hand, re-creating n indicies may blow out the space.  * *   Create new (temporary) relations for the base heap and the new  *   index.  *   *   Exclusively lock the relations. *  *   Create new clustered index and base heap relation. * */voidcluster(char oldrelname[], char oldindexname[]){    Oid OIDOldHeap, OIDOldIndex, OIDNewHeap;        Relation OldHeap, OldIndex;    Relation NewHeap;        char *NewIndexName;    char *szNewHeapName;        /*     *     * I'm going to force all checking back into the commands.c function.     *     * Get the list if indicies for this relation. If the index we want     * is among them, do not add it to the 'kill' list, as it will be     * handled by the 'clean up' code which commits this transaction.     *     * I'm not using the SysCache, because this will happen but     * once, and the slow way is the sure way in this case.     *     */    /*     * Like vacuum, cluster spans transactions, so I'm going to handle it in     * the same way.     */        /* matches the StartTransaction in PostgresMain() */        OldHeap = heap_openr(oldrelname);    if (!RelationIsValid(OldHeap)) {	elog(WARN, "cluster: unknown relation: /"%-.*s/"",	     NAMEDATALEN, oldrelname);    }    OIDOldHeap = OldHeap->rd_id; /* Get OID for the index scan   */        OldIndex=index_openr(oldindexname);/* Open old index relation  */    if (!RelationIsValid(OldIndex)) {	elog(WARN, "cluster: unknown index: /"%-.*s/"",	     NAMEDATALEN, oldindexname);    }    OIDOldIndex = OldIndex->rd_id;     /* OID for the index scan         */        heap_close(OldHeap);    index_close(OldIndex);        /*     * I need to build the copies of the heap and the index. The Commit()     * between here is *very* bogus. If someone is appending stuff, they will     * get the lock after being blocked and add rows which won't be present in     * the new table. Bleagh! I'd be best to try and ensure that no-one's     * in the tables for the entire duration of this process with a pg_vlock.     */    NewHeap    = copy_heap(OIDOldHeap);    OIDNewHeap = NewHeap->rd_id;    szNewHeapName = pstrdup(NewHeap->rd_rel->relname.data);     /* Need to do this to make the new heap visible. */    CommandCounterIncrement();        rebuildheap(OIDNewHeap, OIDOldHeap, OIDOldIndex);        /* Need to do this to make the new heap visible. */    CommandCounterIncrement();    /* can't be found in the SysCache. */    copy_index(OIDOldIndex, OIDNewHeap); /* No contention with the old */        /*      * make this really happen. Flush all the buffers.     */    CommitTransactionCommand();    StartTransactionCommand();//.........这里部分代码省略.........

//.........这里部分代码省略.........	while (!got_sigterm)	{		int rc, ret, i;		/* Wait necessary amount of time */		rc = WaitLatch(&MyProc->procLatch,					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,					   kill_max_idle_time * 1000L,					   PG_WAIT_EXTENSION);		ResetLatch(&MyProc->procLatch);		/* Emergency bailout if postmaster has died */		if (rc & WL_POSTMASTER_DEATH)			proc_exit(1);		/* Process signals */		if (got_sighup)		{			int old_interval;			/* Save old value of kill interval */			old_interval = kill_max_idle_time;			/* Process config file */			ProcessConfigFile(PGC_SIGHUP);			got_sighup = false;			ereport(LOG, (errmsg("bgworker kill_idle signal: processed SIGHUP")));			/* Rebuild query if necessary */			if (old_interval != kill_max_idle_time)			{				resetStringInfo(&buf);				initStringInfo(&buf);				kill_idle_build_query(&buf);			}		}		if (got_sigterm)		{			/* Simply exit */			ereport(LOG, (errmsg("bgworker kill_idle signal: processed SIGTERM")));			proc_exit(0);		}		/* Process idle connection kill */		SetCurrentStatementStartTimestamp();		StartTransactionCommand();		SPI_connect();		PushActiveSnapshot(GetTransactionSnapshot());		pgstat_report_activity(STATE_RUNNING, buf.data);		/* Statement start time */		SetCurrentStatementStartTimestamp();		/* Execute query */		ret = SPI_execute(buf.data, false, 0);		/* Some error handling */		if (ret != SPI_OK_SELECT)			elog(FATAL, "Error when trying to kill idle connections");		/* Do some processing and log stuff disconnected */		for (i = 0; i < SPI_processed; i++)		{			int32 pidValue;			bool isnull;			char *datname = NULL;			char *usename = NULL;			char *client_addr = NULL;			/* Fetch values */			pidValue = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[i],												   SPI_tuptable->tupdesc,												   1, &isnull));			usename = DatumGetCString(SPI_getbinval(SPI_tuptable->vals[i],													SPI_tuptable->tupdesc,													3, &isnull));			datname = DatumGetCString(SPI_getbinval(SPI_tuptable->vals[i],													SPI_tuptable->tupdesc,													4, &isnull));			client_addr = DatumGetCString(SPI_getbinval(SPI_tuptable->vals[i],														SPI_tuptable->tupdesc,														5, &isnull));			/* Log what has been disconnected */			elog(LOG, "Disconnected idle connection: PID %d %s/%s/%s",				 pidValue, datname ? datname : "none",				 usename ? usename : "none",				 client_addr ? client_addr : "none");		}		SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();		pgstat_report_activity(STATE_IDLE, NULL);	}	/* No problems, so clean exit */	proc_exit(0);}

//.........这里部分代码省略.........			TupleBufferResetNotify(WorkerTupleBuffer, MyCQId, MyWorkerId);			if (xact_commit)				StartTransactionCommand();			set_snapshot(estate, cqowner);			CurrentResourceOwner = cqowner;			MemoryContextSwitchTo(estate->es_query_cxt);			estate->es_processed = 0;			estate->es_filtered = 0;			/*			 * Run plan on a microbatch			 */			ExecutePlan(estate, queryDesc->planstate, operation,					true, 0, timeoutms, ForwardScanDirection, dest);			IncrementCQExecutions(1);			TupleBufferClearPinnedSlots();			if (state->long_xact)			{				if (TimestampDifferenceExceeds(last_commit, GetCurrentTimestamp(), LONG_RUNNING_XACT_DURATION))					xact_commit = true;				else					xact_commit = false;			}			unset_snapshot(estate, cqowner);			if (xact_commit)			{				CommitTransactionCommand();				last_commit = GetCurrentTimestamp();			}			MemoryContextResetAndDeleteChildren(CQExecutionContext);			MemoryContextSwitchTo(runcontext);			CurrentResourceOwner = cqowner;			if (estate->es_processed || estate->es_filtered)			{				/*				 * If the CV query is such that the select does not return any tuples				 * ex: select id where id=99; and id=99 does not exist, then this reset				 * will fail. What will happen is that the worker will block at the latch for every				 * allocated slot, TILL a cv returns a non-zero tuple, at which point				 * the worker will resume a simple sleep for the threshold time.				 */				last_process = GetCurrentTimestamp();				/*				 * Send stats to the collector				 */				cq_stat_report(false);			}			/* Has the CQ been deactivated? */			if (!entry->active)			{				if (ActiveSnapshotSet())					unset_snapshot(estate, cqowner);				if (IsTransactionState())					CommitTransactionCommand();				break;

//.........这里部分代码省略.........					 "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.		 */		SetCurrentStatementStartTimestamp();		StartTransactionCommand();		SPI_connect();		PushActiveSnapshot(GetTransactionSnapshot());		pgstat_report_activity(STATE_RUNNING, buf.data);		/* We can now execute queries via SPI */		ret = SPI_execute(buf.data, false, 0);		if (ret != SPI_OK_UPDATE_RETURNING)			elog(FATAL, "cannot select from table %s.%s: error code %d",				 table->schema, table->name, ret);		if (SPI_processed > 0)		{			bool		isnull;			int32		val;			val = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],											  SPI_tuptable->tupdesc,											  1, &isnull));			if (!isnull)				elog(LOG, "%s: count in %s.%s is now %d",					 MyBgworkerEntry->bgw_name,					 table->schema, table->name, val);		}		/*		 * And finish our transaction.		 */		SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();		pgstat_report_activity(STATE_IDLE, NULL);	}	proc_exit(1);}

//.........这里部分代码省略.........	/* Remember whether we are processing everything in the DB */	all_rels = (!OidIsValid(relid) && vacstmt->relation == NULL);	/*	 * Build list of relations to process, unless caller gave us one. (If we	 * build one, we put it in vac_context for safekeeping.)	 */	relations = get_rel_oids(relid, vacstmt->relation);	/*	 * Decide whether we need to start/commit our own transactions.	 *	 * For VACUUM (with or without ANALYZE): always do so, so that we can	 * release locks as soon as possible.  (We could possibly use the outer	 * transaction for a one-table VACUUM, but handling TOAST tables would be	 * problematic.)	 *	 * For ANALYZE (no VACUUM): if inside a transaction block, we cannot	 * start/commit our own transactions.  Also, there's no need to do so if	 * only processing one relation.  For multiple relations when not within a	 * transaction block, and also in an autovacuum worker, use own	 * transactions so we can release locks sooner.	 */	if (vacstmt->options & VACOPT_VACUUM)		use_own_xacts = true;	else	{		Assert(vacstmt->options & VACOPT_ANALYZE);		if (IsAutoVacuumWorkerProcess())			use_own_xacts = true;		else if (in_outer_xact)			use_own_xacts = false;		else if (list_length(relations) > 1)			use_own_xacts = true;		else			use_own_xacts = false;	}	/*	 * vacuum_rel expects to be entered with no transaction active; it will	 * start and commit its own transaction.  But we are called by an SQL	 * command, and so we are executing inside a transaction already. We	 * commit the transaction started in PostgresMain() here, and start	 * another one before exiting to match the commit waiting for us back in	 * PostgresMain().	 */	if (use_own_xacts)	{		/* ActiveSnapshot is not set by autovacuum */		if (ActiveSnapshotSet())			PopActiveSnapshot();		/* matches the StartTransaction in PostgresMain() */		CommitTransactionCommand();	}	/* Turn vacuum cost accounting on or off */	PG_TRY();	{		ListCell   *cur;		VacuumCostActive = (VacuumCostDelay > 0);		VacuumCostBalance = 0;		/*		 * Loop to process each selected relation.		 */		foreach(cur, relations)		{			Oid			relid = lfirst_oid(cur);			bool		scanned_all = false;			if (vacstmt->options & VACOPT_VACUUM)				vacuum_rel(relid, vacstmt, do_toast, for_wraparound,						   &scanned_all);			if (vacstmt->options & VACOPT_ANALYZE)			{				/*				 * If using separate xacts, start one for analyze. Otherwise,				 * we can use the outer transaction.				 */				if (use_own_xacts)				{					StartTransactionCommand();					/* functions in indexes may want a snapshot set */					PushActiveSnapshot(GetTransactionSnapshot());				}				analyze_rel(relid, vacstmt, vac_strategy, !scanned_all);				if (use_own_xacts)				{					PopActiveSnapshot();					CommitTransactionCommand();				}			}		}	}

//.........这里部分代码省略.........    if (am_walsender)    {        Assert(!bootstrap);        if (!superuser() && !has_rolreplication(GetUserId()))            ereport(FATAL,                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),                     errmsg("must be superuser or replication role to start walsender")));    }    /*     * If this is a plain walsender only supporting physical replication, we     * don't want to connect to any particular database. Just finish the     * backend startup by processing any options from the startup packet, and     * we're done.     */    if (am_walsender && !am_db_walsender)    {        /* process any options passed in the startup packet */        if (MyProcPort != NULL)            process_startup_options(MyProcPort, am_superuser);        /* Apply PostAuthDelay as soon as we've read all options */        if (PostAuthDelay > 0)            pg_usleep(PostAuthDelay * 1000000L);        /* initialize client encoding */        InitializeClientEncoding();        /* report this backend in the PgBackendStatus array */        pgstat_bestart();        /* close the transaction we started above */        CommitTransactionCommand();        return;    }    /*     * Set up the global variables holding database id and default tablespace.     * But note we won't actually try to touch the database just yet.     *     * We take a shortcut in the bootstrap case, otherwise we have to look up     * the db's entry in pg_database.     */    if (bootstrap)    {        MyDatabaseId = TemplateDbOid;        MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;    }    else if (in_dbname != NULL)    {        HeapTuple	tuple;        Form_pg_database dbform;        tuple = GetDatabaseTuple(in_dbname);        if (!HeapTupleIsValid(tuple))            ereport(FATAL,                    (errcode(ERRCODE_UNDEFINED_DATABASE),                     errmsg("database /"%s/" does not exist", in_dbname)));        dbform = (Form_pg_database) GETSTRUCT(tuple);        MyDatabaseId = HeapTupleGetOid(tuple);        MyDatabaseTableSpace = dbform->dattablespace;        /* take database name from the caller, just for paranoia */        strlcpy(dbname, in_dbname, sizeof(dbname));    }

voidworker_test_main(Datum main_arg){	dsm_segment *seg;	volatile test_shm_mq_header *hdr;	PGPROC *registrant;	pqsignal(SIGHUP,  handle_sighup);	pqsignal(SIGTERM, handle_sigterm);	BackgroundWorkerUnblockSignals();	printf("worker_test_main: %d/n", DatumGetInt32(main_arg));	CurrentResourceOwner = ResourceOwnerCreate(NULL, "worker test");	seg = dsm_attach(DatumGetInt32(main_arg));	if (seg == NULL)		ereport(ERROR,				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),				 errmsg("unable to map dynamic shared memory segment")));	hdr = dsm_segment_address(seg);	/* 
C++ CommonInit函数代码示例
C++ Commit函数代码示例