contrib/postgres_fdw/connection.c
static void
pgfdw_xact_callback(XactEvent event, void *arg)
{
	HASH_SEQ_STATUS scan;
	ConnCacheEntry *entry;

	/* Quick exit if no connections were touched in this transaction. */
	if (!xact_got_connection)
		return;

	/*
	 * Scan all connection cache entries to find open remote transactions, and
	 * close them.
	 */
	hash_seq_init(&scan, ConnectionHash);
	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
	{
		PGresult   *res;

		/* Ignore cache entry if no open connection right now */
		if (entry->conn == NULL)
			continue;

		/* If it has an open remote transaction, try to close it */
		if (entry->xact_depth > 0)
		{
			elog(DEBUG3, "closing remote transaction on connection %p",
				 entry->conn);

			switch (event)
			{
				case XACT_EVENT_PARALLEL_PRE_COMMIT:
				case XACT_EVENT_PRE_COMMIT:
					/* Commit all remote transactions during pre-commit */
					do_sql_command(entry->conn, "COMMIT TRANSACTION");

					/*
					 * If there were any errors in subtransactions, and we
					 * made prepared statements, do a DEALLOCATE ALL to make
					 * sure we get rid of all prepared statements. This is
					 * annoying and not terribly bulletproof, but it's
					 * probably not worth trying harder.
					 *
					 * DEALLOCATE ALL only exists in 8.3 and later, so this
					 * constrains how old a server postgres_fdw can
					 * communicate with.  We intentionally ignore errors in
					 * the DEALLOCATE, so that we can hobble along to some
					 * extent with older servers (leaking prepared statements
					 * as we go; but we don't really support update operations
					 * pre-8.3 anyway).
					 */
					if (entry->have_prep_stmt && entry->have_error)
					{
						res = PQexec(entry->conn, "DEALLOCATE ALL");
						PQclear(res);
					}
					entry->have_prep_stmt = false;
					entry->have_error = false;
					break;
				case XACT_EVENT_PRE_PREPARE:

					/*
					 * We disallow remote transactions that modified anything,
					 * since it's not very reasonable to hold them open until
					 * the prepared transaction is committed.  For the moment,
					 * throw error unconditionally; later we might allow
					 * read-only cases.  Note that the error will cause us to
					 * come right back here with event == XACT_EVENT_ABORT, so
					 * we'll clean up the connection state at that point.
					 */
					ereport(ERROR,
							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
							 errmsg("cannot prepare a transaction that modified remote tables")));
					break;
				case XACT_EVENT_PARALLEL_COMMIT:
				case XACT_EVENT_COMMIT:
				case XACT_EVENT_PREPARE:
					/* Pre-commit should have closed the open transaction */
					elog(ERROR, "missed cleaning up connection during pre-commit");
					break;
				case XACT_EVENT_PARALLEL_ABORT:
				case XACT_EVENT_ABORT:
					/* Assume we might have lost track of prepared statements */
					entry->have_error = true;

					/*
					 * If a command has been submitted to the remote server by
					 * using an asynchronous execution function, the command
					 * might not have yet completed.  Check to see if a
					 * command is still being processed by the remote server,
					 * and if so, request cancellation of the command.
					 */
					if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE)
					{
						PGcancel   *cancel;
						char		errbuf[256];

						if ((cancel = PQgetCancel(entry->conn)))
						{
							if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
								ereport(WARNING,
										(errcode(ERRCODE_CONNECTION_FAILURE),
								  errmsg("could not send cancel request: %s",
										 errbuf)));
							PQfreeCancel(cancel);
						}
					}

					/* If we're aborting, abort all remote transactions too */
					res = PQexec(entry->conn, "ABORT TRANSACTION");
					/* Note: can't throw ERROR, it would be infinite loop */
					if (PQresultStatus(res) != PGRES_COMMAND_OK)
						pgfdw_report_error(WARNING, res, entry->conn, true,
										   "ABORT TRANSACTION");
					else
					{
						PQclear(res);
						/* As above, make sure to clear any prepared stmts */
						if (entry->have_prep_stmt && entry->have_error)
						{
							res = PQexec(entry->conn, "DEALLOCATE ALL");
							PQclear(res);
						}
						entry->have_prep_stmt = false;
						entry->have_error = false;
					}
					break;
			}
		}

		/* Reset state to show we're out of a transaction */
		entry->xact_depth = 0;

		/*
		 * If the connection isn't in a good idle state, discard it to
		 * recover. Next GetConnection will open a new connection.
		 */
		if (PQstatus(entry->conn) != CONNECTION_OK ||
			PQtransactionStatus(entry->conn) != PQTRANS_IDLE)
		{
			elog(DEBUG3, "discarding connection %p", entry->conn);
			PQfinish(entry->conn);
			entry->conn = NULL;
		}
	}

	/*
	 * Regardless of the event type, we can now mark ourselves as out of the
	 * transaction.  (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
	 * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
	 */
	xact_got_connection = false;

	/* Also reset cursor numbering for next transaction */
	cursor_number = 0;
}
/*
 * pgfdw_xact_callback --- cleanup at main-transaction end.
 */
cleanup at main-transaction end.
cleanup at main - transaction end .
static void
pgfdw_xact_callback(XactEvent event, void *arg)
{
	HASH_SEQ_STATUS scan;
	ConnCacheEntry *entry;
	if (!xact_got_connection)
		return;
	hash_seq_init(&scan, ConnectionHash);
	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
	{
		PGresult   *res;
		if (entry->conn == NULL)
			continue;
		if (entry->xact_depth > 0)
		{
			elog(DEBUG3, "closing remote transaction on connection %p",
				 entry->conn);
			switch (event)
			{
				case XACT_EVENT_PARALLEL_PRE_COMMIT:
				case XACT_EVENT_PRE_COMMIT:
					do_sql_command(entry->conn, "COMMIT TRANSACTION");
					if (entry->have_prep_stmt && entry->have_error)
					{
						res = PQexec(entry->conn, "DEALLOCATE ALL");
						PQclear(res);
					}
					entry->have_prep_stmt = false;
					entry->have_error = false;
					break;
				case XACT_EVENT_PRE_PREPARE:
					ereport(ERROR,
							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
							 errmsg("cannot prepare a transaction that modified remote tables")));
					break;
				case XACT_EVENT_PARALLEL_COMMIT:
				case XACT_EVENT_COMMIT:
				case XACT_EVENT_PREPARE:
					elog(ERROR, "missed cleaning up connection during pre-commit");
					break;
				case XACT_EVENT_PARALLEL_ABORT:
				case XACT_EVENT_ABORT:
					entry->have_error = true;
					if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE)
					{
						PGcancel   *cancel;
						char		errbuf[256];
						if ((cancel = PQgetCancel(entry->conn)))
						{
							if (!PQcancel(cancel, errbuf, sizeof(errbuf)))
								ereport(WARNING,
										(errcode(ERRCODE_CONNECTION_FAILURE),
								  errmsg("could not send cancel request: %s",
										 errbuf)));
							PQfreeCancel(cancel);
						}
					}
					res = PQexec(entry->conn, "ABORT TRANSACTION");
					if (PQresultStatus(res) != PGRES_COMMAND_OK)
						pgfdw_report_error(WARNING, res, entry->conn, true,
										   "ABORT TRANSACTION");
					else
					{
						PQclear(res);
						if (entry->have_prep_stmt && entry->have_error)
						{
							res = PQexec(entry->conn, "DEALLOCATE ALL");
							PQclear(res);
						}
						entry->have_prep_stmt = false;
						entry->have_error = false;
					}
					break;
			}
		}
		entry->xact_depth = 0;
		if (PQstatus(entry->conn) != CONNECTION_OK ||
			PQtransactionStatus(entry->conn) != PQTRANS_IDLE)
		{
			elog(DEBUG3, "discarding connection %p", entry->conn);
			PQfinish(entry->conn);
			entry->conn = NULL;
		}
	}
	xact_got_connection = false;
	cursor_number = 0;
}
static void pgfdw_xact_callback ( XactEvent event , void * arg ) { HASH_SEQ_STATUS scan ; ConnCacheEntry * entry ; if ( ! xact_got_connection ) return ; hash_seq_init ( & scan , ConnectionHash ) ; while ( ( entry = ( ConnCacheEntry * ) hash_seq_search ( & scan ) ) ) { PGresult * res ; if ( entry -> conn == NULL ) continue ; if ( entry -> xact_depth > 0 ) { elog ( DEBUG3 , " " , entry -> conn ) ; switch ( event ) { case XACT_EVENT_PARALLEL_PRE_COMMIT : case XACT_EVENT_PRE_COMMIT : do_sql_command ( entry -> conn , " " ) ; if ( entry -> have_prep_stmt && entry -> have_error ) { res = PQexec ( entry -> conn , " " ) ; PQclear ( res ) ; } entry -> have_prep_stmt = false ; entry -> have_error = false ; break ; case XACT_EVENT_PRE_PREPARE : ereport ( ERROR , ( errcode ( ERRCODE_FEATURE_NOT_SUPPORTED ) , errmsg ( " " ) ) ) ; break ; case XACT_EVENT_PARALLEL_COMMIT : case XACT_EVENT_COMMIT : case XACT_EVENT_PREPARE : elog ( ERROR , " " ) ; break ; case XACT_EVENT_PARALLEL_ABORT : case XACT_EVENT_ABORT : entry -> have_error = true ; if ( PQtransactionStatus ( entry -> conn ) == PQTRANS_ACTIVE ) { PGcancel * cancel ; char errbuf [ 256 ] ; if ( ( cancel = PQgetCancel ( entry -> conn ) ) ) { if ( ! PQcancel ( cancel , errbuf , sizeof ( errbuf ) ) ) ereport ( WARNING , ( errcode ( ERRCODE_CONNECTION_FAILURE ) , errmsg ( " " , errbuf ) ) ) ; PQfreeCancel ( cancel ) ; } } res = PQexec ( entry -> conn , " " ) ; if ( PQresultStatus ( res ) != PGRES_COMMAND_OK ) pgfdw_report_error ( WARNING , res , entry -> conn , true , " " ) ; else { PQclear ( res ) ; if ( entry -> have_prep_stmt && entry -> have_error ) { res = PQexec ( entry -> conn , " " ) ; PQclear ( res ) ; } entry -> have_prep_stmt = false ; entry -> have_error = false ; } break ; } } entry -> xact_depth = 0 ; if ( PQstatus ( entry -> conn ) != CONNECTION_OK || PQtransactionStatus ( entry -> conn ) != PQTRANS_IDLE ) { elog ( DEBUG3 , " " , entry -> conn ) ; PQfinish ( entry -> conn ) ; entry -> conn = NULL ; } } xact_got_connection = false ; cursor_number = 0 ; }
pgfdw_xact_callback --- cleanup at main-transaction end.
pgfdw_xact_callback --- cleanup at main - transaction end .
/* Quick exit if no connections were touched in this transaction. */ /*
	 * Scan all connection cache entries to find open remote transactions, and
	 * close them.
	 */ /* Ignore cache entry if no open connection right now */ /* If it has an open remote transaction, try to close it */ /* Commit all remote transactions during pre-commit */ /*
					 * If there were any errors in subtransactions, and we
					 * made prepared statements, do a DEALLOCATE ALL to make
					 * sure we get rid of all prepared statements. This is
					 * annoying and not terribly bulletproof, but it's
					 * probably not worth trying harder.
					 *
					 * DEALLOCATE ALL only exists in 8.3 and later, so this
					 * constrains how old a server postgres_fdw can
					 * communicate with.  We intentionally ignore errors in
					 * the DEALLOCATE, so that we can hobble along to some
					 * extent with older servers (leaking prepared statements
					 * as we go; but we don't really support update operations
					 * pre-8.3 anyway).
					 */ /*
					 * We disallow remote transactions that modified anything,
					 * since it's not very reasonable to hold them open until
					 * the prepared transaction is committed.  For the moment,
					 * throw error unconditionally; later we might allow
					 * read-only cases.  Note that the error will cause us to
					 * come right back here with event == XACT_EVENT_ABORT, so
					 * we'll clean up the connection state at that point.
					 */ /* Pre-commit should have closed the open transaction */ /* Assume we might have lost track of prepared statements */ /*
					 * If a command has been submitted to the remote server by
					 * using an asynchronous execution function, the command
					 * might not have yet completed.  Check to see if a
					 * command is still being processed by the remote server,
					 * and if so, request cancellation of the command.
					 */ /* If we're aborting, abort all remote transactions too */ /* Note: can't throw ERROR, it would be infinite loop */ /* As above, make sure to clear any prepared stmts */ /* Reset state to show we're out of a transaction */ /*
		 * If the connection isn't in a good idle state, discard it to
		 * recover. Next GetConnection will open a new connection.
		 */ /*
	 * Regardless of the event type, we can now mark ourselves as out of the
	 * transaction.  (Note: if we are here during PRE_COMMIT or PRE_PREPARE,
	 * this saves a useless scan of the hashtable during COMMIT or PREPARE.)
	 */ /* Also reset cursor numbering for next transaction */