#include "postgres_fe.h"
#include <sys/select.h>
#include <sys/time.h>
#include "catalog/pg_type_d.h"
#include "common/fe_memutils.h"
#include "libpq-fe.h"
#include "pg_getopt.h"
#include "portability/instr_time.h"

Include dependency graph for libpq_pipeline.c:

Macros
#define	pg_debug(...)

#define	MAXINTLEN 12

#define	MAXINT8LEN 20

#define	pg_fatal(...) pg_fatal_impl(__LINE__, __VA_ARGS__)

Enumerations
enum	PipelineInsertStep { BI_BEGIN_TX , BI_DROP_TABLE , BI_CREATE_TABLE , BI_PREPARE , BI_INSERT_ROWS , BI_COMMIT_TX , BI_SYNC , BI_DONE }

Functions
static void	exit_nicely (PGconn *conn)

static void	pg_attribute_noreturn () pg_fatal_impl(int line

static void const char	pg_attribute_printf (2, 3)

static bool	process_result (PGconn conn, PGresult res, int results, int numsent)

static void const char	fflush (stdout)

	fprintf (stderr, "\n%s:%d: ", progname, line)

	va_start (args, fmt)

	vfprintf (stderr, fmt, args)

	va_end (args)

	Assert (fmt[strlen(fmt) - 1] !='\n')

	fprintf (stderr, "\n")

	exit (1)

static void	test_disallowed_in_pipeline (PGconn *conn)

static void	test_multi_pipelines (PGconn *conn)

static void	test_nosync (PGconn *conn)

static void	test_pipeline_abort (PGconn *conn)

static void	test_pipelined_insert (PGconn *conn, int n_rows)

static void	test_prepared (PGconn *conn)

static void	notice_processor (void arg, const char message)

static void	test_pipeline_idle (PGconn *conn)

static void	test_simple_pipeline (PGconn *conn)

static void	test_singlerowmode (PGconn *conn)

static void	test_transaction (PGconn *conn)

static void	test_uniqviol (PGconn *conn)

static void	usage (const char *progname)

static void	print_test_list (void)

int	main (int argc, char **argv)

Variables
static void const char *	fmt

const char *const	progname = "libpq_pipeline"

char *	tracefile = NULL

static const char *const	drop_table_sql

static const char *const	create_table_sql

static const char *const	insert_sql

static const char *const	insert_sql2

Macro Definition Documentation

◆ MAXINT8LEN

#define MAXINT8LEN 20

Definition at line 58 of file libpq_pipeline.c.

◆ MAXINTLEN

#define MAXINTLEN 12

Definition at line 57 of file libpq_pipeline.c.

◆ pg_debug

#define pg_debug ( ... )

Definition at line 43 of file libpq_pipeline.c.

◆ pg_fatal

#define pg_fatal ( ... ) pg_fatal_impl(__LINE__, __VA_ARGS__)

Definition at line 70 of file libpq_pipeline.c.

Enumeration Type Documentation

◆ PipelineInsertStep

enum PipelineInsertStep

Enumerator
BI_BEGIN_TX
BI_DROP_TABLE
BI_CREATE_TABLE
BI_PREPARE
BI_INSERT_ROWS
BI_COMMIT_TX
BI_SYNC
BI_DONE

Definition at line 633 of file libpq_pipeline.c.

 {
     BI_BEGIN_TX,
     BI_DROP_TABLE,
     BI_CREATE_TABLE,
     BI_PREPARE,
     BI_INSERT_ROWS,
     BI_COMMIT_TX,
     BI_SYNC,
     BI_DONE
 };

Function Documentation

◆ Assert()

Assert ( fmt ! [strlen(fmt) - 1] = '\n' )

◆ exit()

exit ( 1 )

Referenced by _check_database_version(), add_typedefs_from_file(), addtype(), adjleap(), adjust_data_dir(), append_database_pattern(), append_relation_pattern_helper(), append_schema_pattern(), appendPsqlMetaConnect(), appendQualifiedRelation(), appendShellString(), associate(), bail_out(), BaseBackup(), bbstreamer_zstd_compressor_new(), bootstrap_template1(), change_directory(), check_input(), check_locale_name(), check_ok(), check_prepare_conn(), check_root(), check_testspec(), CheckConnection(), CheckDataVersion(), checkInitSteps(), close_cur1(), close_file(), cluster_one_database(), compile_database_list(), compile_relation_list_one_db(), connectToServer(), create_data_directory(), create_xlog_or_symlink(), CreateBackupStreamer(), die_on_query_failure(), digestControlFile(), do_init(), do_kill(), do_logrotate(), do_promote(), do_reload(), do_restart(), do_start(), do_start_bgworker(), do_status(), do_stop(), dolink(), ecpg_filter_source(), ecpg_filter_stderr(), ecpg_start_test(), enlargeStringInfo(), ensureCleanShutdown(), err(), error(), errstart(), errx(), executeCommand(), executeQuery(), executeQueryOrDie(), executeStatement(), exit_nicely(), find_other_exec_or_die(), findBuiltin(), FindStreamingStart(), get_control_dbstate(), get_id(), get_opts(), get_pgpid(), get_record1(), get_restricted_token(), get_su_pwd(), get_user_name_or_exit(), get_var1(), GetConnection(), getfields(), getRestoreCommand(), GetTableInfo(), GucInfoMain(), handle_args(), handle_help_version_opts(), HandleStartupProcInterrupts(), infile(), Initialize(), isolation_init(), isolation_start_test(), leapadd(), main(), memory_exhausted(), mkdirs(), newabbr(), open_cur1(), open_walfile(), ParallelBackupStart(), parse_psql_options(), parseCommandLine(), parseServiceFile(), pg_fatal(), pg_log_v(), pg_malloc_internal(), pg_realloc(), pg_strdup(), pg_wcsformat(), pgfdw_get_cleanup_result(), pgwin32_is_admin(), pnstrdup(), postprocess_sql_command(), PQprint(), printTable(), printTableAddCell(), printTableAddHeader(), proc_exit(), process_backslash_command(), psql_start_test(), pvsnprintf(), read_controlfile(), read_post_opts(), regression_main(), reindex_one_database(), replace_percent_placeholders(), report_backup_error(), report_clusters_compatible(), report_fatal_error(), report_manifest_error(), rewind_parseTimeLineHistory(), rpytime(), rulesub(), run_command(), run_permutation(), run_reindex_command(), runInitSteps(), s_lock_stuck(), save_ps_display_args(), search_directory(), set_mode(), set_option(), set_sig(), setup_data_file_paths(), setup_locale_encoding(), setup_pgdata(), SetWALFileNameForCleanup(), signal_cleanup(), spawn_process(), sql_check(), sql_conn(), sql_exec(), start_postmaster(), StartLogStreamer(), startup_hacks(), StreamLog(), syntax_error(), test_timing(), threadRun(), time_overflow(), TransferPredicateLocksToNewTarget(), try_complete_step(), usage(), vacuum_delay_point(), vacuum_one_database(), WalRcvWaitForStartPosition(), and writezone().

◆ exit_nicely()

static void exit_nicely ( PGconn * conn )

static

Definition at line 61 of file libpq_pipeline.c.

 {
     PQfinish(conn);
     exit(1);
 }

References conn, exit(), and PQfinish().

Referenced by main(), test_nosync(), and test_pipelined_insert().

◆ fflush()

static void const char fflush ( stdout )

Referenced by adjust_data_dir(), bootstrap_template1(), check_ok(), cluster_all_databases(), create_data_directory(), create_xlog_or_symlink(), do_copy(), do_init(), do_pg_backup_stop(), do_shell(), do_watch(), dumpnfa(), echo_hidden_command(), ecpg_log(), editFile(), emit_tap_output_v(), ensureCleanShutdown(), errfinish(), ExceptionalCondition(), exec_command_print(), exec_command_prompt(), exec_command_write(), exec_prog(), ExecQueryTuples(), ExecQueryUsingCursor(), ExecuteRecoveryCommand(), fork_process(), get_bin_version(), get_control_data(), get_prompt(), get_su_pwd(), gets_interactive(), handleCopyIn(), handleCopyOut(), HandleSlashCmds(), initialize_data_directory(), InteractiveBackend(), log_pre_callback(), main(), MainLoop(), OpenPipeStream(), openQueryOutputFile(), PageOutput(), parallel_exec_prog(), parallel_transfer_all_new_dbs(), ParallelBackupStart(), parse_required_wal(), perform_spin_delay(), pg_log_generic_v(), pg_log_v(), pg_regcomp(), pg_regexec(), pipe_read_line(), plpgsql_dumptree(), popen_check(), pprint(), PQdisplayTuples(), pqFlush(), PQprint(), PQuntrace(), print(), print_filemap(), print_msg(), PrintNotifications(), PrintQueryStatus(), PrintQueryTuples(), printVersion(), psql_end_command(), PSQLexec(), regression_main(), reindex_all_databases(), RestoreArchivedFile(), run_diff(), runPgDump(), runShellCommand(), SendQuery(), setup_config(), shell_archive_file(), simple_prompt_extended(), spawn_process(), start_postmaster(), stop_postmaster(), SysLogger_Start(), test_config_settings(), test_file_descriptor_sync(), test_non_sync(), test_open_sync(), test_specific_config_settings(), test_sync(), vacuum_one_database(), vacuumlo(), and write_stderr().

◆ fprintf() [1/2]

fprintf	(	stderr	,
		"\n"
	)

◆ fprintf() [2/2]

fprintf	(	stderr	,
		"\n%s:%d: "	,
		progname	,
		line
	)

Referenced by main(), notice_processor(), process_result(), test_disallowed_in_pipeline(), test_multi_pipelines(), test_nosync(), test_pipeline_abort(), test_pipeline_idle(), test_pipelined_insert(), test_prepared(), test_simple_pipeline(), test_singlerowmode(), test_transaction(), test_uniqviol(), and usage().

◆ main()

int main	(	int	argc,
		char **	argv
	)

Definition at line 1763 of file libpq_pipeline.c.

 {
     const char *conninfo = "";
     PGconn     *conn;
     FILE       *trace;
     char       *testname;
     int         numrows = 10000;
     PGresult   *res;
     int         c;
  
     while ((c = getopt(argc, argv, "r:t:")) != -1)
     {
         switch (c)
         {
             case 'r':           /* numrows */
                 errno = 0;
                 numrows = strtol(optarg, NULL, 10);
                 if (errno != 0 || numrows <= 0)
                 {
                     fprintf(stderr, "couldn't parse \"%s\" as a positive integer\n",
                             optarg);
                     exit(1);
                 }
                 break;
             case 't':           /* trace file */
                 tracefile = pg_strdup(optarg);
                 break;
         }
     }
  
     if (optind < argc)
     {
         testname = pg_strdup(argv[optind]);
         optind++;
     }
     else
     {
         usage(argv[0]);
         exit(1);
     }
  
     if (strcmp(testname, "tests") == 0)
     {
         print_test_list();
         exit(0);
     }
  
     if (optind < argc)
     {
         conninfo = pg_strdup(argv[optind]);
         optind++;
     }
  
     /* Make a connection to the database */
     conn = PQconnectdb(conninfo);
     if (PQstatus(conn) != CONNECTION_OK)
     {
         fprintf(stderr, "Connection to database failed: %s\n",
                 PQerrorMessage(conn));
         exit_nicely(conn);
     }
  
     res = PQexec(conn, "SET lc_messages TO \"C\"");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("failed to set lc_messages: %s", PQerrorMessage(conn));
     res = PQexec(conn, "SET debug_parallel_query = off");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("failed to set debug_parallel_query: %s", PQerrorMessage(conn));
  
     /* Set the trace file, if requested */
     if (tracefile != NULL)
     {
         if (strcmp(tracefile, "-") == 0)
             trace = stdout;
         else
             trace = fopen(tracefile, "w");
         if (trace == NULL)
             pg_fatal("could not open file \"%s\": %m", tracefile);
  
         /* Make it line-buffered */
         setvbuf(trace, NULL, PG_IOLBF, 0);
  
         PQtrace(conn, trace);
         PQsetTraceFlags(conn,
                         PQTRACE_SUPPRESS_TIMESTAMPS | PQTRACE_REGRESS_MODE);
     }
  
     if (strcmp(testname, "disallowed_in_pipeline") == 0)
         test_disallowed_in_pipeline(conn);
     else if (strcmp(testname, "multi_pipelines") == 0)
         test_multi_pipelines(conn);
     else if (strcmp(testname, "nosync") == 0)
         test_nosync(conn);
     else if (strcmp(testname, "pipeline_abort") == 0)
         test_pipeline_abort(conn);
     else if (strcmp(testname, "pipeline_idle") == 0)
         test_pipeline_idle(conn);
     else if (strcmp(testname, "pipelined_insert") == 0)
         test_pipelined_insert(conn, numrows);
     else if (strcmp(testname, "prepared") == 0)
         test_prepared(conn);
     else if (strcmp(testname, "simple_pipeline") == 0)
         test_simple_pipeline(conn);
     else if (strcmp(testname, "singlerow") == 0)
         test_singlerowmode(conn);
     else if (strcmp(testname, "transaction") == 0)
         test_transaction(conn);
     else if (strcmp(testname, "uniqviol") == 0)
         test_uniqviol(conn);
     else
     {
         fprintf(stderr, "\"%s\" is not a recognized test name\n", testname);
         exit(1);
     }
  
     /* close the connection to the database and cleanup */
     PQfinish(conn);
     return 0;
 }

References conn, CONNECTION_OK, exit(), exit_nicely(), fprintf(), getopt(), optarg, optind, pg_fatal, PG_IOLBF, pg_strdup(), PGRES_COMMAND_OK, PQconnectdb(), PQerrorMessage(), PQexec(), PQfinish(), PQresultStatus(), PQsetTraceFlags(), PQstatus(), PQtrace(), PQTRACE_REGRESS_MODE, PQTRACE_SUPPRESS_TIMESTAMPS, print_test_list(), res, generate_unaccent_rules::stdout, test_disallowed_in_pipeline(), test_multi_pipelines(), test_nosync(), test_pipeline_abort(), test_pipeline_idle(), test_pipelined_insert(), test_prepared(), test_simple_pipeline(), test_singlerowmode(), test_transaction(), test_uniqviol(), tracefile, and usage().

◆ notice_processor()

static void notice_processor	(	void *	arg,
		const char *	message
	)

static

Definition at line 1051 of file libpq_pipeline.c.

 {
     int        *n_notices = (int *) arg;
  
     (*n_notices)++;
     fprintf(stderr, "NOTICE %d: %s", *n_notices, message);
 }

References arg, and fprintf().

Referenced by test_pipeline_idle().

◆ pg_attribute_noreturn()

static void pg_attribute_noreturn ( )

static

◆ pg_attribute_printf()

static void const char pg_attribute_printf	(	2	,
		3
	)

◆ print_test_list()

static void print_test_list ( void )

static

Definition at line 1747 of file libpq_pipeline.c.

 {
     printf("disallowed_in_pipeline\n");
     printf("multi_pipelines\n");
     printf("nosync\n");
     printf("pipeline_abort\n");
     printf("pipeline_idle\n");
     printf("pipelined_insert\n");
     printf("prepared\n");
     printf("simple_pipeline\n");
     printf("singlerow\n");
     printf("transaction\n");
     printf("uniqviol\n");
 }

References printf.

Referenced by main().

◆ process_result()

static bool process_result	(	PGconn *	conn,
		PGresult *	res,
		int	results,
		int	numsent
	)

static

Definition at line 1687 of file libpq_pipeline.c.

 {
     PGresult   *res2;
     bool        got_error = false;
  
     if (res == NULL)
         pg_fatal("got unexpected NULL");
  
     switch (PQresultStatus(res))
     {
         case PGRES_FATAL_ERROR:
             got_error = true;
             fprintf(stderr, "result %d/%d (error): %s\n", results, numsent, PQerrorMessage(conn));
             PQclear(res);
  
             res2 = PQgetResult(conn);
             if (res2 != NULL)
                 pg_fatal("expected NULL, got %s",
                          PQresStatus(PQresultStatus(res2)));
             break;
  
         case PGRES_TUPLES_OK:
             fprintf(stderr, "result %d/%d: %s\n", results, numsent, PQgetvalue(res, 0, 0));
             PQclear(res);
  
             res2 = PQgetResult(conn);
             if (res2 != NULL)
                 pg_fatal("expected NULL, got %s",
                          PQresStatus(PQresultStatus(res2)));
             break;
  
         case PGRES_PIPELINE_ABORTED:
             fprintf(stderr, "result %d/%d: pipeline aborted\n", results, numsent);
             res2 = PQgetResult(conn);
             if (res2 != NULL)
                 pg_fatal("expected NULL, got %s",
                          PQresStatus(PQresultStatus(res2)));
             break;
  
         default:
             pg_fatal("got unexpected %s", PQresStatus(PQresultStatus(res)));
     }
  
     return got_error;
 }

References conn, fprintf(), pg_fatal, PGRES_FATAL_ERROR, PGRES_PIPELINE_ABORTED, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetResult(), PQgetvalue(), PQresStatus(), PQresultStatus(), and res.

Referenced by test_uniqviol().

◆ test_disallowed_in_pipeline()

static void test_disallowed_in_pipeline ( PGconn * conn )

static

Definition at line 90 of file libpq_pipeline.c.

 {
     PGresult   *res = NULL;
  
     fprintf(stderr, "test error cases... ");
  
     if (PQisnonblocking(conn))
         pg_fatal("Expected blocking connection mode");
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("Unable to enter pipeline mode");
  
     if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
         pg_fatal("Pipeline mode not activated properly");
  
     /* PQexec should fail in pipeline mode */
     res = PQexec(conn, "SELECT 1");
     if (PQresultStatus(res) != PGRES_FATAL_ERROR)
         pg_fatal("PQexec should fail in pipeline mode but succeeded");
     if (strcmp(PQerrorMessage(conn),
                "synchronous command execution functions are not allowed in pipeline mode\n") != 0)
         pg_fatal("did not get expected error message; got: \"%s\"",
                  PQerrorMessage(conn));
  
     /* PQsendQuery should fail in pipeline mode */
     if (PQsendQuery(conn, "SELECT 1") != 0)
         pg_fatal("PQsendQuery should fail in pipeline mode but succeeded");
     if (strcmp(PQerrorMessage(conn),
                "PQsendQuery not allowed in pipeline mode\n") != 0)
         pg_fatal("did not get expected error message; got: \"%s\"",
                  PQerrorMessage(conn));
  
     /* Entering pipeline mode when already in pipeline mode is OK */
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("re-entering pipeline mode should be a no-op but failed");
  
     if (PQisBusy(conn) != 0)
         pg_fatal("PQisBusy should return 0 when idle in pipeline mode, returned 1");
  
     /* ok, back to normal command mode */
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("couldn't exit idle empty pipeline mode");
  
     if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
         pg_fatal("Pipeline mode not terminated properly");
  
     /* exiting pipeline mode when not in pipeline mode should be a no-op */
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("pipeline mode exit when not in pipeline mode should succeed but failed");
  
     /* can now PQexec again */
     res = PQexec(conn, "SELECT 1");
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("PQexec should succeed after exiting pipeline mode but failed with: %s",
                  PQerrorMessage(conn));
  
     fprintf(stderr, "ok\n");
 }

References conn, fprintf(), pg_fatal, PGRES_FATAL_ERROR, PGRES_TUPLES_OK, PQ_PIPELINE_OFF, PQenterPipelineMode(), PQerrorMessage(), PQexec(), PQexitPipelineMode(), PQisBusy(), PQisnonblocking(), PQpipelineStatus(), PQresultStatus(), PQsendQuery(), and res.

Referenced by main().

◆ test_multi_pipelines()

static void test_multi_pipelines ( PGconn * conn )

static

Definition at line 150 of file libpq_pipeline.c.

 {
     PGresult   *res = NULL;
     const char *dummy_params[1] = {"1"};
     Oid         dummy_param_oids[1] = {INT4OID};
  
     fprintf(stderr, "multi pipeline... ");
  
     /*
      * Queue up a couple of small pipelines and process each without returning
      * to command mode first.
      */
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
  
     if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
                           dummy_params, NULL, NULL, 0) != 1)
         pg_fatal("dispatching first SELECT failed: %s", PQerrorMessage(conn));
  
     if (PQpipelineSync(conn) != 1)
         pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
  
     if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
                           dummy_params, NULL, NULL, 0) != 1)
         pg_fatal("dispatching second SELECT failed: %s", PQerrorMessage(conn));
  
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     /* OK, start processing the results */
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
                  PQerrorMessage(conn));
  
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("Unexpected result code %s from first pipeline item",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
     res = NULL;
  
     if (PQgetResult(conn) != NULL)
         pg_fatal("PQgetResult returned something extra after first result");
  
     if (PQexitPipelineMode(conn) != 0)
         pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when sync result expected: %s",
                  PQerrorMessage(conn));
  
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code %s instead of sync result, error: %s",
                  PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
     PQclear(res);
  
     /* second pipeline */
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
                  PQerrorMessage(conn));
  
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("Unexpected result code %s from second pipeline item",
                  PQresStatus(PQresultStatus(res)));
  
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("Expected null result, got %s",
                  PQresStatus(PQresultStatus(res)));
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
                  PQerrorMessage(conn));
  
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code %s from second pipeline sync",
                  PQresStatus(PQresultStatus(res)));
  
     /* We're still in pipeline mode ... */
     if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
         pg_fatal("Fell out of pipeline mode somehow");
  
     /* until we end it, which we can safely do now */
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
                  PQerrorMessage(conn));
  
     if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
         pg_fatal("exiting pipeline mode didn't seem to work");
  
     fprintf(stderr, "ok\n");
 }

References conn, fprintf(), pg_fatal, PGRES_PIPELINE_SYNC, PGRES_TUPLES_OK, PQ_PIPELINE_OFF, PQclear(), PQenterPipelineMode(), PQerrorMessage(), PQexitPipelineMode(), PQgetResult(), PQpipelineStatus(), PQpipelineSync(), PQresStatus(), PQresultStatus(), PQsendQueryParams(), and res.

Referenced by main().

◆ test_nosync()

static void test_nosync ( PGconn * conn )

static

Definition at line 252 of file libpq_pipeline.c.

 {
     int         numqueries = 10;
     int         results = 0;
     int         sock = PQsocket(conn);
  
     fprintf(stderr, "nosync... ");
  
     if (sock < 0)
         pg_fatal("invalid socket");
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("could not enter pipeline mode");
     for (int i = 0; i < numqueries; i++)
     {
         fd_set      input_mask;
         struct timeval tv;
  
         if (PQsendQueryParams(conn, "SELECT repeat('xyzxz', 12)",
                               0, NULL, NULL, NULL, NULL, 0) != 1)
             pg_fatal("error sending select: %s", PQerrorMessage(conn));
         PQflush(conn);
  
         /*
          * If the server has written anything to us, read (some of) it now.
          */
         FD_ZERO(&input_mask);
         FD_SET(sock, &input_mask);
         tv.tv_sec = 0;
         tv.tv_usec = 0;
         if (select(sock + 1, &input_mask, NULL, NULL, &tv) < 0)
         {
             fprintf(stderr, "select() failed: %s\n", strerror(errno));
             exit_nicely(conn);
         }
         if (FD_ISSET(sock, &input_mask) && PQconsumeInput(conn) != 1)
             pg_fatal("failed to read from server: %s", PQerrorMessage(conn));
     }
  
     /* tell server to flush its output buffer */
     if (PQsendFlushRequest(conn) != 1)
         pg_fatal("failed to send flush request");
     PQflush(conn);
  
     /* Now read all results */
     for (;;)
     {
         PGresult   *res;
  
         res = PQgetResult(conn);
  
         /* NULL results are only expected after TUPLES_OK */
         if (res == NULL)
             pg_fatal("got unexpected NULL result after %d results", results);
  
         /* We expect exactly one TUPLES_OK result for each query we sent */
         if (PQresultStatus(res) == PGRES_TUPLES_OK)
         {
             PGresult   *res2;
  
             /* and one NULL result should follow each */
             res2 = PQgetResult(conn);
             if (res2 != NULL)
                 pg_fatal("expected NULL, got %s",
                          PQresStatus(PQresultStatus(res2)));
             PQclear(res);
             results++;
  
             /* if we're done, we're done */
             if (results == numqueries)
                 break;
  
             continue;
         }
  
         /* anything else is unexpected */
         pg_fatal("got unexpected %s\n", PQresStatus(PQresultStatus(res)));
     }
  
     fprintf(stderr, "ok\n");
 }

References conn, exit_nicely(), fprintf(), i, pg_fatal, PGRES_TUPLES_OK, PQclear(), PQconsumeInput(), PQenterPipelineMode(), PQerrorMessage(), PQflush(), PQgetResult(), PQresStatus(), PQresultStatus(), PQsendFlushRequest(), PQsendQueryParams(), PQsocket(), res, select, and strerror.

Referenced by main().

◆ test_pipeline_abort()

static void test_pipeline_abort ( PGconn * conn )

static

Definition at line 344 of file libpq_pipeline.c.

 {
     PGresult   *res = NULL;
     const char *dummy_params[1] = {"1"};
     Oid         dummy_param_oids[1] = {INT4OID};
     int         i;
     int         gotrows;
     bool        goterror;
  
     fprintf(stderr, "aborted pipeline... ");
  
     res = PQexec(conn, drop_table_sql);
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("dispatching DROP TABLE failed: %s", PQerrorMessage(conn));
  
     res = PQexec(conn, create_table_sql);
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("dispatching CREATE TABLE failed: %s", PQerrorMessage(conn));
  
     /*
      * Queue up a couple of small pipelines and process each without returning
      * to command mode first. Make sure the second operation in the first
      * pipeline ERRORs.
      */
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
  
     dummy_params[0] = "1";
     if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
                           dummy_params, NULL, NULL, 0) != 1)
         pg_fatal("dispatching first insert failed: %s", PQerrorMessage(conn));
  
     if (PQsendQueryParams(conn, "SELECT no_such_function($1)",
                           1, dummy_param_oids, dummy_params,
                           NULL, NULL, 0) != 1)
         pg_fatal("dispatching error select failed: %s", PQerrorMessage(conn));
  
     dummy_params[0] = "2";
     if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
                           dummy_params, NULL, NULL, 0) != 1)
         pg_fatal("dispatching second insert failed: %s", PQerrorMessage(conn));
  
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     dummy_params[0] = "3";
     if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
                           dummy_params, NULL, NULL, 0) != 1)
         pg_fatal("dispatching second-pipeline insert failed: %s",
                  PQerrorMessage(conn));
  
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     /*
      * OK, start processing the pipeline results.
      *
      * We should get a command-ok for the first query, then a fatal error and
      * a pipeline aborted message for the second insert, a pipeline-end, then
      * a command-ok and a pipeline-ok for the second pipeline operation.
      */
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("Unexpected result status %s: %s",
                  PQresStatus(PQresultStatus(res)),
                  PQresultErrorMessage(res));
     PQclear(res);
  
     /* NULL result to signal end-of-results for this command */
     if ((res = PQgetResult(conn)) != NULL)
         pg_fatal("Expected null result, got %s",
                  PQresStatus(PQresultStatus(res)));
  
     /* Second query caused error, so we expect an error next */
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_FATAL_ERROR)
         pg_fatal("Unexpected result code -- expected PGRES_FATAL_ERROR, got %s",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
  
     /* NULL result to signal end-of-results for this command */
     if ((res = PQgetResult(conn)) != NULL)
         pg_fatal("Expected null result, got %s",
                  PQresStatus(PQresultStatus(res)));
  
     /*
      * pipeline should now be aborted.
      *
      * Note that we could still queue more queries at this point if we wanted;
      * they'd get added to a new third pipeline since we've already sent a
      * second. The aborted flag relates only to the pipeline being received.
      */
     if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
         pg_fatal("pipeline should be flagged as aborted but isn't");
  
     /* third query in pipeline, the second insert */
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_PIPELINE_ABORTED)
         pg_fatal("Unexpected result code -- expected PGRES_PIPELINE_ABORTED, got %s",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
  
     /* NULL result to signal end-of-results for this command */
     if ((res = PQgetResult(conn)) != NULL)
         pg_fatal("Expected null result, got %s", PQresStatus(PQresultStatus(res)));
  
     if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
         pg_fatal("pipeline should be flagged as aborted but isn't");
  
     /* Ensure we're still in pipeline */
     if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
         pg_fatal("Fell out of pipeline mode somehow");
  
     /*
      * The end of a failed pipeline is a PGRES_PIPELINE_SYNC.
      *
      * (This is so clients know to start processing results normally again and
      * can tell the difference between skipped commands and the sync.)
      */
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code from first pipeline sync\n"
                  "Expected PGRES_PIPELINE_SYNC, got %s",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
  
     if (PQpipelineStatus(conn) == PQ_PIPELINE_ABORTED)
         pg_fatal("sync should've cleared the aborted flag but didn't");
  
     /* We're still in pipeline mode... */
     if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
         pg_fatal("Fell out of pipeline mode somehow");
  
     /* the insert from the second pipeline */
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("Unexpected result code %s from first item in second pipeline",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
  
     /* Read the NULL result at the end of the command */
     if ((res = PQgetResult(conn)) != NULL)
         pg_fatal("Expected null result, got %s", PQresStatus(PQresultStatus(res)));
  
     /* the second pipeline sync */
     if ((res = PQgetResult(conn)) == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code %s from second pipeline sync",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
  
     if ((res = PQgetResult(conn)) != NULL)
         pg_fatal("Expected null result, got %s: %s",
                  PQresStatus(PQresultStatus(res)),
                  PQerrorMessage(conn));
  
     /* Try to send two queries in one command */
     if (PQsendQueryParams(conn, "SELECT 1; SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s", PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     goterror = false;
     while ((res = PQgetResult(conn)) != NULL)
     {
         switch (PQresultStatus(res))
         {
             case PGRES_FATAL_ERROR:
                 if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "42601") != 0)
                     pg_fatal("expected error about multiple commands, got %s",
                              PQerrorMessage(conn));
                 printf("got expected %s", PQerrorMessage(conn));
                 goterror = true;
                 break;
             default:
                 pg_fatal("got unexpected status %s", PQresStatus(PQresultStatus(res)));
                 break;
         }
     }
     if (!goterror)
         pg_fatal("did not get cannot-insert-multiple-commands error");
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("got NULL result");
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code %s from pipeline sync",
                  PQresStatus(PQresultStatus(res)));
     fprintf(stderr, "ok\n");
  
     /* Test single-row mode with an error partways */
     if (PQsendQueryParams(conn, "SELECT 1.0/g FROM generate_series(3, -1, -1) g",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s", PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     PQsetSingleRowMode(conn);
     goterror = false;
     gotrows = 0;
     while ((res = PQgetResult(conn)) != NULL)
     {
         switch (PQresultStatus(res))
         {
             case PGRES_SINGLE_TUPLE:
                 printf("got row: %s\n", PQgetvalue(res, 0, 0));
                 gotrows++;
                 break;
             case PGRES_FATAL_ERROR:
                 if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "22012") != 0)
                     pg_fatal("expected division-by-zero, got: %s (%s)",
                              PQerrorMessage(conn),
                              PQresultErrorField(res, PG_DIAG_SQLSTATE));
                 printf("got expected division-by-zero\n");
                 goterror = true;
                 break;
             default:
                 pg_fatal("got unexpected result %s", PQresStatus(PQresultStatus(res)));
         }
         PQclear(res);
     }
     if (!goterror)
         pg_fatal("did not get division-by-zero error");
     if (gotrows != 3)
         pg_fatal("did not get three rows");
     /* the third pipeline sync */
     if ((res = PQgetResult(conn)) == NULL)
         pg_fatal("Unexpected NULL result: %s", PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code %s from third pipeline sync",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
  
     /* We're still in pipeline mode... */
     if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
         pg_fatal("Fell out of pipeline mode somehow");
  
     /* until we end it, which we can safely do now */
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
                  PQerrorMessage(conn));
  
     if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
         pg_fatal("exiting pipeline mode didn't seem to work");
  
     /*-
      * Since we fired the pipelines off without a surrounding xact, the results
      * should be:
      *
      * - Implicit xact started by server around 1st pipeline
      * - First insert applied
      * - Second statement aborted xact
      * - Third insert skipped
      * - Sync rolled back first implicit xact
      * - Implicit xact created by server around 2nd pipeline
      * - insert applied from 2nd pipeline
      * - Sync commits 2nd xact
      *
      * So we should only have the value 3 that we inserted.
      */
     res = PQexec(conn, "SELECT itemno FROM pq_pipeline_demo");
  
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("Expected tuples, got %s: %s",
                  PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
     if (PQntuples(res) != 1)
         pg_fatal("expected 1 result, got %d", PQntuples(res));
     for (i = 0; i < PQntuples(res); i++)
     {
         const char *val = PQgetvalue(res, i, 0);
  
         if (strcmp(val, "3") != 0)
             pg_fatal("expected only insert with value 3, got %s", val);
     }
  
     PQclear(res);
  
     fprintf(stderr, "ok\n");
 }

References conn, create_table_sql, drop_table_sql, fprintf(), i, insert_sql, PG_DIAG_SQLSTATE, pg_fatal, PGRES_COMMAND_OK, PGRES_FATAL_ERROR, PGRES_PIPELINE_ABORTED, PGRES_PIPELINE_SYNC, PGRES_SINGLE_TUPLE, PGRES_TUPLES_OK, PQ_PIPELINE_ABORTED, PQ_PIPELINE_OFF, PQclear(), PQenterPipelineMode(), PQerrorMessage(), PQexec(), PQexitPipelineMode(), PQgetResult(), PQgetvalue(), PQntuples(), PQpipelineStatus(), PQpipelineSync(), PQresStatus(), PQresultErrorField(), PQresultErrorMessage(), PQresultStatus(), PQsendQueryParams(), PQsetSingleRowMode(), printf, res, and val.

Referenced by main().

◆ test_pipeline_idle()

static void test_pipeline_idle ( PGconn * conn )

static

Definition at line 1061 of file libpq_pipeline.c.

 {
     PGresult   *res;
     int         n_notices = 0;
  
     fprintf(stderr, "\npipeline idle...\n");
  
     PQsetNoticeProcessor(conn, notice_processor, &n_notices);
  
     /* Try to exit pipeline mode in pipeline-idle state */
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
     if (PQsendQueryParams(conn, "SELECT 1", 0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s", PQerrorMessage(conn));
     PQsendFlushRequest(conn);
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
                  PQerrorMessage(conn));
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("unexpected result code %s from first pipeline item",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("did not receive terminating NULL");
     if (PQsendQueryParams(conn, "SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s", PQerrorMessage(conn));
     if (PQexitPipelineMode(conn) == 1)
         pg_fatal("exiting pipeline succeeded when it shouldn't");
     if (strncmp(PQerrorMessage(conn), "cannot exit pipeline mode",
                 strlen("cannot exit pipeline mode")) != 0)
         pg_fatal("did not get expected error; got: %s",
                  PQerrorMessage(conn));
     PQsendFlushRequest(conn);
     res = PQgetResult(conn);
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("unexpected result code %s from second pipeline item",
                  PQresStatus(PQresultStatus(res)));
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("did not receive terminating NULL");
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("exiting pipeline failed: %s", PQerrorMessage(conn));
  
     if (n_notices > 0)
         pg_fatal("got %d notice(s)", n_notices);
     fprintf(stderr, "ok - 1\n");
  
     /* Have a WARNING in the middle of a resultset */
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("entering pipeline mode failed: %s", PQerrorMessage(conn));
     if (PQsendQueryParams(conn, "SELECT pg_catalog.pg_advisory_unlock(1,1)", 0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s", PQerrorMessage(conn));
     PQsendFlushRequest(conn);
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("unexpected NULL result received");
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("unexpected result code %s", PQresStatus(PQresultStatus(res)));
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("failed to exit pipeline mode: %s", PQerrorMessage(conn));
     fprintf(stderr, "ok - 2\n");
 }

References conn, fprintf(), notice_processor(), pg_fatal, PGRES_TUPLES_OK, PQclear(), PQenterPipelineMode(), PQerrorMessage(), PQexitPipelineMode(), PQgetResult(), PQresStatus(), PQresultStatus(), PQsendFlushRequest(), PQsendQueryParams(), PQsetNoticeProcessor(), and res.

Referenced by main().

◆ test_pipelined_insert()

static void test_pipelined_insert	(	PGconn *	conn,
		int	n_rows
	)

static

Definition at line 646 of file libpq_pipeline.c.

 {
     Oid         insert_param_oids[2] = {INT4OID, INT8OID};
     const char *insert_params[2];
     char        insert_param_0[MAXINTLEN];
     char        insert_param_1[MAXINT8LEN];
     enum PipelineInsertStep send_step = BI_BEGIN_TX,
                 recv_step = BI_BEGIN_TX;
     int         rows_to_send,
                 rows_to_receive;
  
     insert_params[0] = insert_param_0;
     insert_params[1] = insert_param_1;
  
     rows_to_send = rows_to_receive = n_rows;
  
     /*
      * Do a pipelined insert into a table created at the start of the pipeline
      */
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
  
     while (send_step != BI_PREPARE)
     {
         const char *sql;
  
         switch (send_step)
         {
             case BI_BEGIN_TX:
                 sql = "BEGIN TRANSACTION";
                 send_step = BI_DROP_TABLE;
                 break;
  
             case BI_DROP_TABLE:
                 sql = drop_table_sql;
                 send_step = BI_CREATE_TABLE;
                 break;
  
             case BI_CREATE_TABLE:
                 sql = create_table_sql;
                 send_step = BI_PREPARE;
                 break;
  
             default:
                 pg_fatal("invalid state");
                 sql = NULL;     /* keep compiler quiet */
         }
  
         pg_debug("sending: %s\n", sql);
         if (PQsendQueryParams(conn, sql,
                               0, NULL, NULL, NULL, NULL, 0) != 1)
             pg_fatal("dispatching %s failed: %s", sql, PQerrorMessage(conn));
     }
  
     Assert(send_step == BI_PREPARE);
     pg_debug("sending: %s\n", insert_sql2);
     if (PQsendPrepare(conn, "my_insert", insert_sql2, 2, insert_param_oids) != 1)
         pg_fatal("dispatching PREPARE failed: %s", PQerrorMessage(conn));
     send_step = BI_INSERT_ROWS;
  
     /*
      * Now we start inserting. We'll be sending enough data that we could fill
      * our output buffer, so to avoid deadlocking we need to enter nonblocking
      * mode and consume input while we send more output. As results of each
      * query are processed we should pop them to allow processing of the next
      * query. There's no need to finish the pipeline before processing
      * results.
      */
     if (PQsetnonblocking(conn, 1) != 0)
         pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn));
  
     while (recv_step != BI_DONE)
     {
         int         sock;
         fd_set      input_mask;
         fd_set      output_mask;
  
         sock = PQsocket(conn);
  
         if (sock < 0)
             break;              /* shouldn't happen */
  
         FD_ZERO(&input_mask);
         FD_SET(sock, &input_mask);
         FD_ZERO(&output_mask);
         FD_SET(sock, &output_mask);
  
         if (select(sock + 1, &input_mask, &output_mask, NULL, NULL) < 0)
         {
             fprintf(stderr, "select() failed: %s\n", strerror(errno));
             exit_nicely(conn);
         }
  
         /*
          * Process any results, so we keep the server's output buffer free
          * flowing and it can continue to process input
          */
         if (FD_ISSET(sock, &input_mask))
         {
             PQconsumeInput(conn);
  
             /* Read until we'd block if we tried to read */
             while (!PQisBusy(conn) && recv_step < BI_DONE)
             {
                 PGresult   *res;
                 const char *cmdtag = "";
                 const char *description = "";
                 int         status;
  
                 /*
                  * Read next result.  If no more results from this query,
                  * advance to the next query
                  */
                 res = PQgetResult(conn);
                 if (res == NULL)
                     continue;
  
                 status = PGRES_COMMAND_OK;
                 switch (recv_step)
                 {
                     case BI_BEGIN_TX:
                         cmdtag = "BEGIN";
                         recv_step++;
                         break;
                     case BI_DROP_TABLE:
                         cmdtag = "DROP TABLE";
                         recv_step++;
                         break;
                     case BI_CREATE_TABLE:
                         cmdtag = "CREATE TABLE";
                         recv_step++;
                         break;
                     case BI_PREPARE:
                         cmdtag = "";
                         description = "PREPARE";
                         recv_step++;
                         break;
                     case BI_INSERT_ROWS:
                         cmdtag = "INSERT";
                         rows_to_receive--;
                         if (rows_to_receive == 0)
                             recv_step++;
                         break;
                     case BI_COMMIT_TX:
                         cmdtag = "COMMIT";
                         recv_step++;
                         break;
                     case BI_SYNC:
                         cmdtag = "";
                         description = "SYNC";
                         status = PGRES_PIPELINE_SYNC;
                         recv_step++;
                         break;
                     case BI_DONE:
                         /* unreachable */
                         pg_fatal("unreachable state");
                 }
  
                 if (PQresultStatus(res) != status)
                     pg_fatal("%s reported status %s, expected %s\n"
                              "Error message: \"%s\"",
                              description, PQresStatus(PQresultStatus(res)),
                              PQresStatus(status), PQerrorMessage(conn));
  
                 if (strncmp(PQcmdStatus(res), cmdtag, strlen(cmdtag)) != 0)
                     pg_fatal("%s expected command tag '%s', got '%s'",
                              description, cmdtag, PQcmdStatus(res));
  
                 pg_debug("Got %s OK\n", cmdtag[0] != '\0' ? cmdtag : description);
  
                 PQclear(res);
             }
         }
  
         /* Write more rows and/or the end pipeline message, if needed */
         if (FD_ISSET(sock, &output_mask))
         {
             PQflush(conn);
  
             if (send_step == BI_INSERT_ROWS)
             {
                 snprintf(insert_param_0, MAXINTLEN, "%d", rows_to_send);
                 /* use up some buffer space with a wide value */
                 snprintf(insert_param_1, MAXINT8LEN, "%lld", 1LL << 62);
  
                 if (PQsendQueryPrepared(conn, "my_insert",
                                         2, insert_params, NULL, NULL, 0) == 1)
                 {
                     pg_debug("sent row %d\n", rows_to_send);
  
                     rows_to_send--;
                     if (rows_to_send == 0)
                         send_step++;
                 }
                 else
                 {
                     /*
                      * in nonblocking mode, so it's OK for an insert to fail
                      * to send
                      */
                     fprintf(stderr, "WARNING: failed to send insert #%d: %s\n",
                             rows_to_send, PQerrorMessage(conn));
                 }
             }
             else if (send_step == BI_COMMIT_TX)
             {
                 if (PQsendQueryParams(conn, "COMMIT",
                                       0, NULL, NULL, NULL, NULL, 0) == 1)
                 {
                     pg_debug("sent COMMIT\n");
                     send_step++;
                 }
                 else
                 {
                     fprintf(stderr, "WARNING: failed to send commit: %s\n",
                             PQerrorMessage(conn));
                 }
             }
             else if (send_step == BI_SYNC)
             {
                 if (PQpipelineSync(conn) == 1)
                 {
                     fprintf(stdout, "pipeline sync sent\n");
                     send_step++;
                 }
                 else
                 {
                     fprintf(stderr, "WARNING: pipeline sync failed: %s\n",
                             PQerrorMessage(conn));
                 }
             }
         }
     }
  
     /* We've got the sync message and the pipeline should be done */
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
                  PQerrorMessage(conn));
  
     if (PQsetnonblocking(conn, 0) != 0)
         pg_fatal("failed to clear nonblocking mode: %s", PQerrorMessage(conn));
  
     fprintf(stderr, "ok\n");
 }

References Assert(), BI_BEGIN_TX, BI_COMMIT_TX, BI_CREATE_TABLE, BI_DONE, BI_DROP_TABLE, BI_INSERT_ROWS, BI_PREPARE, BI_SYNC, conn, create_table_sql, description, drop_table_sql, exit_nicely(), fprintf(), insert_sql2, MAXINT8LEN, MAXINTLEN, pg_debug, pg_fatal, PGRES_COMMAND_OK, PGRES_PIPELINE_SYNC, PQclear(), PQcmdStatus(), PQconsumeInput(), PQenterPipelineMode(), PQerrorMessage(), PQexitPipelineMode(), PQflush(), PQgetResult(), PQisBusy(), PQpipelineSync(), PQresStatus(), PQresultStatus(), PQsendPrepare(), PQsendQueryParams(), PQsendQueryPrepared(), PQsetnonblocking(), PQsocket(), res, select, snprintf, generate_unaccent_rules::stdout, and strerror.

Referenced by main().

◆ test_prepared()

static void test_prepared ( PGconn * conn )

static

Definition at line 892 of file libpq_pipeline.c.

 {
     PGresult   *res = NULL;
     Oid         param_oids[1] = {INT4OID};
     Oid         expected_oids[4];
     Oid         typ;
  
     fprintf(stderr, "prepared... ");
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
     if (PQsendPrepare(conn, "select_one", "SELECT $1, '42', $1::numeric, "
                       "interval '1 sec'",
                       1, param_oids) != 1)
         pg_fatal("preparing query failed: %s", PQerrorMessage(conn));
     expected_oids[0] = INT4OID;
     expected_oids[1] = TEXTOID;
     expected_oids[2] = NUMERICOID;
     expected_oids[3] = INTERVALOID;
     if (PQsendDescribePrepared(conn, "select_one") != 1)
         pg_fatal("failed to send describePrepared: %s", PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("expected NULL result");
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned NULL");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
     if (PQnfields(res) != lengthof(expected_oids))
         pg_fatal("expected %zd columns, got %d",
                  lengthof(expected_oids), PQnfields(res));
     for (int i = 0; i < PQnfields(res); i++)
     {
         typ = PQftype(res, i);
         if (typ != expected_oids[i])
             pg_fatal("field %d: expected type %u, got %u",
                      i, expected_oids[i], typ);
     }
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("expected NULL result");
  
     res = PQgetResult(conn);
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("expected PGRES_PIPELINE_SYNC, got %s", PQresStatus(PQresultStatus(res)));
  
     fprintf(stderr, "closing statement..");
     if (PQsendClosePrepared(conn, "select_one") != 1)
         pg_fatal("PQsendClosePrepared failed: %s", PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("expected non-NULL result");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("expected NULL result");
     res = PQgetResult(conn);
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("expected PGRES_PIPELINE_SYNC, got %s", PQresStatus(PQresultStatus(res)));
  
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("could not exit pipeline mode: %s", PQerrorMessage(conn));
  
     /* Now that it's closed we should get an error when describing */
     res = PQdescribePrepared(conn, "select_one");
     if (PQresultStatus(res) != PGRES_FATAL_ERROR)
         pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
  
     /*
      * Also test the blocking close, this should not fail since closing a
      * non-existent prepared statement is a no-op
      */
     res = PQclosePrepared(conn, "select_one");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
  
     fprintf(stderr, "creating portal... ");
     PQexec(conn, "BEGIN");
     PQexec(conn, "DECLARE cursor_one CURSOR FOR SELECT 1");
     PQenterPipelineMode(conn);
     if (PQsendDescribePortal(conn, "cursor_one") != 1)
         pg_fatal("PQsendDescribePortal failed: %s", PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
  
     typ = PQftype(res, 0);
     if (typ != INT4OID)
         pg_fatal("portal: expected type %u, got %u",
                  INT4OID, typ);
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("expected NULL result");
     res = PQgetResult(conn);
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("expected PGRES_PIPELINE_SYNC, got %s", PQresStatus(PQresultStatus(res)));
  
     fprintf(stderr, "closing portal... ");
     if (PQsendClosePortal(conn, "cursor_one") != 1)
         pg_fatal("PQsendClosePortal failed: %s", PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("expected non-NULL result");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
     PQclear(res);
     res = PQgetResult(conn);
     if (res != NULL)
         pg_fatal("expected NULL result");
     res = PQgetResult(conn);
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("expected PGRES_PIPELINE_SYNC, got %s", PQresStatus(PQresultStatus(res)));
  
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("could not exit pipeline mode: %s", PQerrorMessage(conn));
  
     /* Now that it's closed we should get an error when describing */
     res = PQdescribePortal(conn, "cursor_one");
     if (PQresultStatus(res) != PGRES_FATAL_ERROR)
         pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
  
     /*
      * Also test the blocking close, this should not fail since closing a
      * non-existent portal is a no-op
      */
     res = PQclosePortal(conn, "cursor_one");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
  
     fprintf(stderr, "ok\n");
 }

References conn, fprintf(), i, lengthof, pg_fatal, PGRES_COMMAND_OK, PGRES_FATAL_ERROR, PGRES_PIPELINE_SYNC, PQclear(), PQclosePortal(), PQclosePrepared(), PQdescribePortal(), PQdescribePrepared(), PQenterPipelineMode(), PQerrorMessage(), PQexec(), PQexitPipelineMode(), PQftype(), PQgetResult(), PQnfields(), PQpipelineSync(), PQresStatus(), PQresultStatus(), PQsendClosePortal(), PQsendClosePrepared(), PQsendDescribePortal(), PQsendDescribePrepared(), PQsendPrepare(), and res.

Referenced by main().

◆ test_simple_pipeline()

static void test_simple_pipeline ( PGconn * conn )

static

Definition at line 1128 of file libpq_pipeline.c.

 {
     PGresult   *res = NULL;
     const char *dummy_params[1] = {"1"};
     Oid         dummy_param_oids[1] = {INT4OID};
  
     fprintf(stderr, "simple pipeline... ");
  
     /*
      * Enter pipeline mode and dispatch a set of operations, which we'll then
      * process the results of as they come in.
      *
      * For a simple case we should be able to do this without interim
      * processing of results since our output buffer will give us enough slush
      * to work with and we won't block on sending. So blocking mode is fine.
      */
     if (PQisnonblocking(conn))
         pg_fatal("Expected blocking connection mode");
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
  
     if (PQsendQueryParams(conn, "SELECT $1",
                           1, dummy_param_oids, dummy_params,
                           NULL, NULL, 0) != 1)
         pg_fatal("dispatching SELECT failed: %s", PQerrorMessage(conn));
  
     if (PQexitPipelineMode(conn) != 0)
         pg_fatal("exiting pipeline mode with work in progress should fail, but succeeded");
  
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
                  PQerrorMessage(conn));
  
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("Unexpected result code %s from first pipeline item",
                  PQresStatus(PQresultStatus(res)));
  
     PQclear(res);
     res = NULL;
  
     if (PQgetResult(conn) != NULL)
         pg_fatal("PQgetResult returned something extra after first query result.");
  
     /*
      * Even though we've processed the result there's still a sync to come and
      * we can't exit pipeline mode yet
      */
     if (PQexitPipelineMode(conn) != 0)
         pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
  
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("PQgetResult returned null when sync result PGRES_PIPELINE_SYNC expected: %s",
                  PQerrorMessage(conn));
  
     if (PQresultStatus(res) != PGRES_PIPELINE_SYNC)
         pg_fatal("Unexpected result code %s instead of PGRES_PIPELINE_SYNC, error: %s",
                  PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
  
     PQclear(res);
     res = NULL;
  
     if (PQgetResult(conn) != NULL)
         pg_fatal("PQgetResult returned something extra after pipeline end: %s",
                  PQresStatus(PQresultStatus(res)));
  
     /* We're still in pipeline mode... */
     if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
         pg_fatal("Fell out of pipeline mode somehow");
  
     /* ... until we end it, which we can safely do now */
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
                  PQerrorMessage(conn));
  
     if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
         pg_fatal("Exiting pipeline mode didn't seem to work");
  
     fprintf(stderr, "ok\n");
 }

References conn, fprintf(), pg_fatal, PGRES_PIPELINE_SYNC, PGRES_TUPLES_OK, PQ_PIPELINE_OFF, PQclear(), PQenterPipelineMode(), PQerrorMessage(), PQexitPipelineMode(), PQgetResult(), PQisnonblocking(), PQpipelineStatus(), PQpipelineSync(), PQresStatus(), PQresultStatus(), PQsendQueryParams(), and res.

Referenced by main().

◆ test_singlerowmode()

static void test_singlerowmode ( PGconn * conn )

static

Definition at line 1215 of file libpq_pipeline.c.

 {
     PGresult   *res;
     int         i;
     bool        pipeline_ended = false;
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s",
                  PQerrorMessage(conn));
  
     /* One series of three commands, using single-row mode for the first two. */
     for (i = 0; i < 3; i++)
     {
         char       *param[1];
  
         param[0] = psprintf("%d", 44 + i);
  
         if (PQsendQueryParams(conn,
                               "SELECT generate_series(42, $1)",
                               1,
                               NULL,
                               (const char **) param,
                               NULL,
                               NULL,
                               0) != 1)
             pg_fatal("failed to send query: %s",
                      PQerrorMessage(conn));
         pfree(param[0]);
     }
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
  
     for (i = 0; !pipeline_ended; i++)
     {
         bool        first = true;
         bool        saw_ending_tuplesok;
         bool        isSingleTuple = false;
  
         /* Set single row mode for only first 2 SELECT queries */
         if (i < 2)
         {
             if (PQsetSingleRowMode(conn) != 1)
                 pg_fatal("PQsetSingleRowMode() failed for i=%d", i);
         }
  
         /* Consume rows for this query */
         saw_ending_tuplesok = false;
         while ((res = PQgetResult(conn)) != NULL)
         {
             ExecStatusType est = PQresultStatus(res);
  
             if (est == PGRES_PIPELINE_SYNC)
             {
                 fprintf(stderr, "end of pipeline reached\n");
                 pipeline_ended = true;
                 PQclear(res);
                 if (i != 3)
                     pg_fatal("Expected three results, got %d", i);
                 break;
             }
  
             /* Expect SINGLE_TUPLE for queries 0 and 1, TUPLES_OK for 2 */
             if (first)
             {
                 if (i <= 1 && est != PGRES_SINGLE_TUPLE)
                     pg_fatal("Expected PGRES_SINGLE_TUPLE for query %d, got %s",
                              i, PQresStatus(est));
                 if (i >= 2 && est != PGRES_TUPLES_OK)
                     pg_fatal("Expected PGRES_TUPLES_OK for query %d, got %s",
                              i, PQresStatus(est));
                 first = false;
             }
  
             fprintf(stderr, "Result status %s for query %d", PQresStatus(est), i);
             switch (est)
             {
                 case PGRES_TUPLES_OK:
                     fprintf(stderr, ", tuples: %d\n", PQntuples(res));
                     saw_ending_tuplesok = true;
                     if (isSingleTuple)
                     {
                         if (PQntuples(res) == 0)
                             fprintf(stderr, "all tuples received in query %d\n", i);
                         else
                             pg_fatal("Expected to follow PGRES_SINGLE_TUPLE, but received PGRES_TUPLES_OK directly instead");
                     }
                     break;
  
                 case PGRES_SINGLE_TUPLE:
                     isSingleTuple = true;
                     fprintf(stderr, ", %d tuple: %s\n", PQntuples(res), PQgetvalue(res, 0, 0));
                     break;
  
                 default:
                     pg_fatal("unexpected");
             }
             PQclear(res);
         }
         if (!pipeline_ended && !saw_ending_tuplesok)
             pg_fatal("didn't get expected terminating TUPLES_OK");
     }
  
     /*
      * Now issue one command, get its results in with single-row mode, then
      * issue another command, and get its results in normal mode; make sure
      * the single-row mode flag is reset as expected.
      */
     if (PQsendQueryParams(conn, "SELECT generate_series(0, 0)",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
     if (PQsendFlushRequest(conn) != 1)
         pg_fatal("failed to send flush request");
     if (PQsetSingleRowMode(conn) != 1)
         pg_fatal("PQsetSingleRowMode() failed");
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("unexpected NULL");
     if (PQresultStatus(res) != PGRES_SINGLE_TUPLE)
         pg_fatal("Expected PGRES_SINGLE_TUPLE, got %s",
                  PQresStatus(PQresultStatus(res)));
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("unexpected NULL");
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("Expected PGRES_TUPLES_OK, got %s",
                  PQresStatus(PQresultStatus(res)));
     if (PQgetResult(conn) != NULL)
         pg_fatal("expected NULL result");
  
     if (PQsendQueryParams(conn, "SELECT 1",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
     if (PQsendFlushRequest(conn) != 1)
         pg_fatal("failed to send flush request");
     res = PQgetResult(conn);
     if (res == NULL)
         pg_fatal("unexpected NULL");
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("Expected PGRES_TUPLES_OK, got %s",
                  PQresStatus(PQresultStatus(res)));
     if (PQgetResult(conn) != NULL)
         pg_fatal("expected NULL result");
  
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
  
     fprintf(stderr, "ok\n");
 }

References conn, fprintf(), i, pfree(), pg_fatal, PGRES_PIPELINE_SYNC, PGRES_SINGLE_TUPLE, PGRES_TUPLES_OK, PQclear(), PQenterPipelineMode(), PQerrorMessage(), PQexitPipelineMode(), PQgetResult(), PQgetvalue(), PQntuples(), PQpipelineSync(), PQresStatus(), PQresultStatus(), PQsendFlushRequest(), PQsendQueryParams(), PQsetSingleRowMode(), psprintf(), and res.

Referenced by main().

◆ test_transaction()

static void test_transaction ( PGconn * conn )

static

Definition at line 1371 of file libpq_pipeline.c.

 {
     PGresult   *res;
     bool        expect_null;
     int         num_syncs = 0;
  
     res = PQexec(conn, "DROP TABLE IF EXISTS pq_pipeline_tst;"
                  "CREATE TABLE pq_pipeline_tst (id int)");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("failed to create test table: %s",
                  PQerrorMessage(conn));
     PQclear(res);
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode: %s",
                  PQerrorMessage(conn));
     if (PQsendPrepare(conn, "rollback", "ROLLBACK", 0, NULL) != 1)
         pg_fatal("could not send prepare on pipeline: %s",
                  PQerrorMessage(conn));
  
     if (PQsendQueryParams(conn,
                           "BEGIN",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
     if (PQsendQueryParams(conn,
                           "SELECT 0/0",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
  
     /*
      * send a ROLLBACK using a prepared stmt. Doesn't work because we need to
      * get out of the pipeline-aborted state first.
      */
     if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
         pg_fatal("failed to execute prepared: %s",
                  PQerrorMessage(conn));
  
     /* This insert fails because we're in pipeline-aborted state */
     if (PQsendQueryParams(conn,
                           "INSERT INTO pq_pipeline_tst VALUES (1)",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     num_syncs++;
  
     /*
      * This insert fails even though the pipeline got a SYNC, because we're in
      * an aborted transaction
      */
     if (PQsendQueryParams(conn,
                           "INSERT INTO pq_pipeline_tst VALUES (2)",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     num_syncs++;
  
     /*
      * Send ROLLBACK using prepared stmt. This one works because we just did
      * PQpipelineSync above.
      */
     if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
         pg_fatal("failed to execute prepared: %s",
                  PQerrorMessage(conn));
  
     /*
      * Now that we're out of a transaction and in pipeline-good mode, this
      * insert works
      */
     if (PQsendQueryParams(conn,
                           "INSERT INTO pq_pipeline_tst VALUES (3)",
                           0, NULL, NULL, NULL, NULL, 0) != 1)
         pg_fatal("failed to send query: %s",
                  PQerrorMessage(conn));
     /* Send two syncs now -- match up to SYNC messages below */
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     num_syncs++;
     if (PQpipelineSync(conn) != 1)
         pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
     num_syncs++;
  
     expect_null = false;
     for (int i = 0;; i++)
     {
         ExecStatusType restype;
  
         res = PQgetResult(conn);
         if (res == NULL)
         {
             printf("%d: got NULL result\n", i);
             if (!expect_null)
                 pg_fatal("did not expect NULL here");
             expect_null = false;
             continue;
         }
         restype = PQresultStatus(res);
         printf("%d: got status %s", i, PQresStatus(restype));
         if (expect_null)
             pg_fatal("expected NULL");
         if (restype == PGRES_FATAL_ERROR)
             printf("; error: %s", PQerrorMessage(conn));
         else if (restype == PGRES_PIPELINE_ABORTED)
         {
             printf(": command didn't run because pipeline aborted\n");
         }
         else
             printf("\n");
         PQclear(res);
  
         if (restype == PGRES_PIPELINE_SYNC)
             num_syncs--;
         else
             expect_null = true;
         if (num_syncs <= 0)
             break;
     }
     if (PQgetResult(conn) != NULL)
         pg_fatal("returned something extra after all the syncs: %s",
                  PQresStatus(PQresultStatus(res)));
  
     if (PQexitPipelineMode(conn) != 1)
         pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
  
     /* We expect to find one tuple containing the value "3" */
     res = PQexec(conn, "SELECT * FROM pq_pipeline_tst");
     if (PQresultStatus(res) != PGRES_TUPLES_OK)
         pg_fatal("failed to obtain result: %s", PQerrorMessage(conn));
     if (PQntuples(res) != 1)
         pg_fatal("did not get 1 tuple");
     if (strcmp(PQgetvalue(res, 0, 0), "3") != 0)
         pg_fatal("did not get expected tuple");
     PQclear(res);
  
     fprintf(stderr, "ok\n");
 }

References conn, fprintf(), i, pg_fatal, PGRES_COMMAND_OK, PGRES_FATAL_ERROR, PGRES_PIPELINE_ABORTED, PGRES_PIPELINE_SYNC, PGRES_TUPLES_OK, PQclear(), PQenterPipelineMode(), PQerrorMessage(), PQexec(), PQexitPipelineMode(), PQgetResult(), PQgetvalue(), PQntuples(), PQpipelineSync(), PQresStatus(), PQresultStatus(), PQsendPrepare(), PQsendQueryParams(), PQsendQueryPrepared(), printf, and res.

Referenced by main().

◆ test_uniqviol()

static void test_uniqviol ( PGconn * conn )

static

Definition at line 1519 of file libpq_pipeline.c.

 {
     int         sock = PQsocket(conn);
     PGresult   *res;
     Oid         paramTypes[2] = {INT8OID, INT8OID};
     const char *paramValues[2];
     char        paramValue0[MAXINT8LEN];
     char        paramValue1[MAXINT8LEN];
     int         ctr = 0;
     int         numsent = 0;
     int         results = 0;
     bool        read_done = false;
     bool        write_done = false;
     bool        error_sent = false;
     bool        got_error = false;
     int         switched = 0;
     int         socketful = 0;
     fd_set      in_fds;
     fd_set      out_fds;
  
     fprintf(stderr, "uniqviol ...");
  
     PQsetnonblocking(conn, 1);
  
     paramValues[0] = paramValue0;
     paramValues[1] = paramValue1;
     sprintf(paramValue1, "42");
  
     res = PQexec(conn, "drop table if exists ppln_uniqviol;"
                  "create table ppln_uniqviol(id bigint primary key, idata bigint)");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("failed to create table: %s", PQerrorMessage(conn));
  
     res = PQexec(conn, "begin");
     if (PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("failed to begin transaction: %s", PQerrorMessage(conn));
  
     res = PQprepare(conn, "insertion",
                     "insert into ppln_uniqviol values ($1, $2) returning id",
                     2, paramTypes);
     if (res == NULL || PQresultStatus(res) != PGRES_COMMAND_OK)
         pg_fatal("failed to prepare query: %s", PQerrorMessage(conn));
  
     if (PQenterPipelineMode(conn) != 1)
         pg_fatal("failed to enter pipeline mode");
  
     while (!read_done)
     {
         /*
          * Avoid deadlocks by reading everything the server has sent before
          * sending anything.  (Special precaution is needed here to process
          * PQisBusy before testing the socket for read-readiness, because the
          * socket does not turn read-ready after "sending" queries in aborted
          * pipeline mode.)
          */
         while (PQisBusy(conn) == 0)
         {
             bool        new_error;
  
             if (results >= numsent)
             {
                 if (write_done)
                     read_done = true;
                 break;
             }
  
             res = PQgetResult(conn);
             new_error = process_result(conn, res, results, numsent);
             if (new_error && got_error)
                 pg_fatal("got two errors");
             got_error |= new_error;
             if (results++ >= numsent - 1)
             {
                 if (write_done)
                     read_done = true;
                 break;
             }
         }
  
         if (read_done)
             break;
  
         FD_ZERO(&out_fds);
         FD_SET(sock, &out_fds);
  
         FD_ZERO(&in_fds);
         FD_SET(sock, &in_fds);
  
         if (select(sock + 1, &in_fds, write_done ? NULL : &out_fds, NULL, NULL) == -1)
         {
             if (errno == EINTR)
                 continue;
             pg_fatal("select() failed: %m");
         }
  
         if (FD_ISSET(sock, &in_fds) && PQconsumeInput(conn) == 0)
             pg_fatal("PQconsumeInput failed: %s", PQerrorMessage(conn));
  
         /*
          * If the socket is writable and we haven't finished sending queries,
          * send some.
          */
         if (!write_done && FD_ISSET(sock, &out_fds))
         {
             for (;;)
             {
                 int         flush;
  
                 /*
                  * provoke uniqueness violation exactly once after having
                  * switched to read mode.
                  */
                 if (switched >= 1 && !error_sent && ctr % socketful >= socketful / 2)
                 {
                     sprintf(paramValue0, "%d", numsent / 2);
                     fprintf(stderr, "E");
                     error_sent = true;
                 }
                 else
                 {
                     fprintf(stderr, ".");
                     sprintf(paramValue0, "%d", ctr++);
                 }
  
                 if (PQsendQueryPrepared(conn, "insertion", 2, paramValues, NULL, NULL, 0) != 1)
                     pg_fatal("failed to execute prepared query: %s", PQerrorMessage(conn));
                 numsent++;
  
                 /* Are we done writing? */
                 if (socketful != 0 && numsent % socketful == 42 && error_sent)
                 {
                     if (PQsendFlushRequest(conn) != 1)
                         pg_fatal("failed to send flush request");
                     write_done = true;
                     fprintf(stderr, "\ndone writing\n");
                     PQflush(conn);
                     break;
                 }
  
                 /* is the outgoing socket full? */
                 flush = PQflush(conn);
                 if (flush == -1)
                     pg_fatal("failed to flush: %s", PQerrorMessage(conn));
                 if (flush == 1)
                 {
                     if (socketful == 0)
                         socketful = numsent;
                     fprintf(stderr, "\nswitch to reading\n");
                     switched++;
                     break;
                 }
             }
         }
     }
  
     if (!got_error)
         pg_fatal("did not get expected error");
  
     fprintf(stderr, "ok\n");
 }

References conn, EINTR, fprintf(), MAXINT8LEN, pg_fatal, PGRES_COMMAND_OK, PQconsumeInput(), PQenterPipelineMode(), PQerrorMessage(), PQexec(), PQflush(), PQgetResult(), PQisBusy(), PQprepare(), PQresultStatus(), PQsendFlushRequest(), PQsendQueryPrepared(), PQsetnonblocking(), PQsocket(), process_result(), res, select, and sprintf.

Referenced by main().

◆ usage()

static void usage ( const char * progname )

static

Definition at line 1735 of file libpq_pipeline.c.

 {
     fprintf(stderr, "%s tests libpq's pipeline mode.\n\n", progname);
     fprintf(stderr, "Usage:\n");
     fprintf(stderr, "  %s [OPTION] tests\n", progname);
     fprintf(stderr, "  %s [OPTION] TESTNAME [CONNINFO]\n", progname);
     fprintf(stderr, "\nOptions:\n");
     fprintf(stderr, "  -t TRACEFILE       generate a libpq trace to TRACEFILE\n");
     fprintf(stderr, "  -r NUMROWS         use NUMROWS as the test size\n");
 }

References fprintf(), and progname.

Referenced by main().

◆ va_end()

va_end ( args )

Referenced by ahprintf(), appendJSONKeyValueFmt(), appendPQExpBuffer(), appendStringInfo(), archprintf(), bail_out(), check_amproc_signature(), dblink_res_error(), do_serialize(), ecpg_log(), ECPGdescribe(), ECPGdo(), ECPGget_desc(), ECPGset_desc(), emit_tap_output(), emit_tap_output_v(), err(), error(), errx(), exec_prog(), executeQueryOrDie(), libpq_append_conn_error(), libpq_append_error(), log_error(), parallel_exec_prog(), pg_fatal(), pg_fprintf(), pg_log(), pg_log_generic(), pg_log_generic_v(), pg_printf(), pg_snprintf(), pg_sprintf(), PLy_elog_impl(), PLy_exception_set(), PLy_exception_set_plural(), pqInternalNotice(), prep_status(), prep_status_progress(), print_lo_result(), printfPQExpBuffer(), psprintf(), psql_add_command(), px_debug(), replace_percent_placeholders(), report_backup_error(), report_fatal_error(), report_invalid_record(), report_manifest_error(), report_status(), StringAt(), tarPrintf(), warn_or_exit_horribly(), warning(), and write_stderr().

◆ va_start()

va_start	(	args	,
		fmt
	)

Referenced by ahprintf(), appendJSONKeyValueFmt(), appendPQExpBuffer(), appendStringInfo(), archprintf(), bail_out(), check_amproc_signature(), dblink_res_error(), do_serialize(), ecpg_log(), ECPGdescribe(), ECPGdo(), ECPGget_desc(), ECPGset_desc(), emit_tap_output(), err(), error(), errx(), exec_prog(), executeQueryOrDie(), libpq_append_conn_error(), libpq_append_error(), log_error(), parallel_exec_prog(), pg_fatal(), pg_fprintf(), pg_log(), pg_log_generic(), pg_printf(), pg_snprintf(), pg_sprintf(), PLy_elog_impl(), PLy_exception_set(), PLy_exception_set_plural(), pqInternalNotice(), prep_status(), prep_status_progress(), print_lo_result(), printfPQExpBuffer(), psprintf(), psql_add_command(), px_debug(), replace_percent_placeholders(), report_backup_error(), report_fatal_error(), report_invalid_record(), report_manifest_error(), report_status(), StringAt(), tarPrintf(), warn_or_exit_horribly(), warning(), and write_stderr().

◆ vfprintf()

vfprintf	(	stderr	,
		fmt	,
		args
	)

Referenced by ecpg_log(), emit_tap_output_v(), err(), errx(), pg_log_generic_v(), print_lo_result(), verror(), and write_stderr().

Variable Documentation

◆ create_table_sql

const char* const create_table_sql

static

Initial value:

=
"CREATE UNLOGGED TABLE pq_pipeline_demo(id serial primary key, itemno integer,"
"int8filler int8);"

Definition at line 48 of file libpq_pipeline.c.

Referenced by test_pipeline_abort(), and test_pipelined_insert().

◆ drop_table_sql

const char* const drop_table_sql

static

Initial value:

=

"DROP TABLE IF EXISTS pq_pipeline_demo"

Definition at line 46 of file libpq_pipeline.c.

Referenced by test_pipeline_abort(), and test_pipelined_insert().

◆ fmt

static void const char * fmt

Definition at line 29 of file libpq_pipeline.c.

Referenced by _allocAH(), ahprintf(), appendJSONKeyValueFmt(), appendPQExpBuffer(), appendPQExpBufferVA(), appendStringInfo(), appendStringInfoVA(), archprintf(), bail_out(), CreateArchive(), date_test_defmt(), date_test_fmt(), datetime_to_char_body(), dblink_res_error(), do_pset(), do_serialize(), do_to_timestamp(), dttofmtasc_replace(), ecpg_log(), emit_tap_output(), emit_tap_output_v(), err(), errmsg(), errmsg_internal(), errx(), exec_prog(), executeQueryOrDie(), find_end_token(), float4_to_char(), float8_to_char(), fmtfloat(), fmtlong(), format_elog_string(), int4_to_char(), int8_to_char(), interval_to_char(), libpq_append_conn_error(), libpq_append_error(), log_error(), main(), my_strftime(), numeric_to_char(), numeric_to_number(), OpenArchive(), parallel_exec_prog(), parse_datetime(), pg_fatal(), pg_fprintf(), pg_log(), pg_log_generic(), pg_log_generic_v(), pg_log_v(), pg_printf(), pg_snprintf(), pg_sprintf(), pg_strfromd(), pg_vfprintf(), pg_vprintf(), pg_vsnprintf(), pg_vsprintf(), PGTYPESdate_defmt_asc(), PGTYPEStimestamp_defmt_asc(), PGTYPEStimestamp_defmt_scan(), PLy_elog_impl(), PLy_exception_set(), pqInternalNotice(), prep_status(), prep_status_progress(), print_lo_result(), printfPQExpBuffer(), ProcessCopyOptions(), psprintf(), pvsnprintf(), px_debug(), rdefmtdate(), ReadHead(), report_backup_error(), report_fatal_error(), report_invalid_record(), report_manifest_error(), report_status(), rfmtdate(), rfmtlong(), tarPrintf(), text_format(), timestamp_to_char(), timestamptz_to_char(), to_date(), to_timestamp(), warn_or_exit_horribly(), and write_stderr().

◆ insert_sql

const char* const insert_sql

static

Initial value:

=

"INSERT INTO pq_pipeline_demo(itemno) VALUES ($1)"

Definition at line 51 of file libpq_pipeline.c.

Referenced by test_pipeline_abort().

◆ insert_sql2

const char* const insert_sql2

static

Initial value:

=

"INSERT INTO pq_pipeline_demo(itemno,int8filler) VALUES ($1, $2)"

Definition at line 53 of file libpq_pipeline.c.

Referenced by test_pipelined_insert().

◆ progname

const char* const progname = "libpq_pipeline"

Definition at line 34 of file libpq_pipeline.c.

Referenced by usage().

◆ tracefile

char* tracefile = NULL

Definition at line 37 of file libpq_pipeline.c.

Referenced by main().

Macros

Enumerations

Functions

Variables

Macro Definition Documentation

◆ MAXINT8LEN

◆ MAXINTLEN

◆ pg_debug

◆ pg_fatal

Enumeration Type Documentation

◆ PipelineInsertStep

Function Documentation

◆ Assert()

◆ exit()

◆ exit_nicely()

◆ fflush()

◆ fprintf() [1/2]

◆ fprintf() [2/2]

◆ main()

◆ notice_processor()

◆ pg_attribute_noreturn()

◆ pg_attribute_printf()

◆ print_test_list()

◆ process_result()

◆ test_disallowed_in_pipeline()

◆ test_multi_pipelines()

◆ test_nosync()

◆ test_pipeline_abort()

◆ test_pipeline_idle()

◆ test_pipelined_insert()

◆ test_prepared()

◆ test_simple_pipeline()

◆ test_singlerowmode()

◆ test_transaction()

◆ test_uniqviol()

◆ usage()

◆ va_end()

◆ va_start()

◆ vfprintf()

Variable Documentation

◆ create_table_sql

◆ drop_table_sql

◆ fmt

◆ insert_sql

◆ insert_sql2

◆ progname

◆ tracefile