Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * libpq_pipeline.c
4 : : * Verify libpq pipeline execution functionality
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : *
10 : : * IDENTIFICATION
11 : : * src/test/modules/libpq_pipeline/libpq_pipeline.c
12 : : *
13 : : *-------------------------------------------------------------------------
14 : : */
15 : :
16 : : #include "postgres_fe.h"
17 : :
18 : : #include <sys/select.h>
19 : : #include <sys/time.h>
20 : :
21 : : #include "catalog/pg_type_d.h"
22 : : #include "libpq-fe.h"
23 : : #include "pg_getopt.h"
24 : :
25 : :
26 : : static void exit_nicely(PGconn *conn);
27 : : pg_noreturn static void pg_fatal_impl(int line, const char *fmt, ...)
28 : : pg_attribute_printf(2, 3);
29 : : static bool process_result(PGconn *conn, PGresult *res, int results,
30 : : int numsent);
31 : :
32 : : static const char *const progname = "libpq_pipeline";
33 : :
34 : : /* Options and defaults */
35 : : static char *tracefile = NULL; /* path to PQtrace() file */
36 : :
37 : :
38 : : #ifdef DEBUG_OUTPUT
39 : : #define pg_debug(...) do { fprintf(stderr, __VA_ARGS__); } while (0)
40 : : #else
41 : : #define pg_debug(...)
42 : : #endif
43 : :
44 : : static const char *const drop_table_sql =
45 : : "DROP TABLE IF EXISTS pq_pipeline_demo";
46 : : static const char *const create_table_sql =
47 : : "CREATE UNLOGGED TABLE pq_pipeline_demo(id serial primary key, itemno integer,"
48 : : "int8filler int8);";
49 : : static const char *const insert_sql =
50 : : "INSERT INTO pq_pipeline_demo(itemno) VALUES ($1)";
51 : : static const char *const insert_sql2 =
52 : : "INSERT INTO pq_pipeline_demo(itemno,int8filler) VALUES ($1, $2)";
53 : :
54 : : /* max char length of an int32/64, plus sign and null terminator */
55 : : #define MAXINTLEN 12
56 : : #define MAXINT8LEN 20
57 : :
58 : : static void
59 : 0 : exit_nicely(PGconn *conn)
60 : : {
61 : 0 : PQfinish(conn);
62 : 0 : exit(1);
63 : : }
64 : :
65 : : /*
66 : : * The following few functions are wrapped in macros to make the reported line
67 : : * number in an error match the line number of the invocation.
68 : : */
69 : :
70 : : /*
71 : : * Print an error to stderr and terminate the program.
72 : : */
73 : : #define pg_fatal(...) pg_fatal_impl(__LINE__, __VA_ARGS__)
74 : : pg_noreturn static void
75 : 0 : pg_fatal_impl(int line, const char *fmt, ...)
76 : : {
77 : : va_list args;
78 : :
79 : 0 : fflush(stdout);
80 : :
81 : 0 : fprintf(stderr, "\n%s:%d: ", progname, line);
82 : 0 : va_start(args, fmt);
83 : 0 : vfprintf(stderr, fmt, args);
84 : 0 : va_end(args);
85 : : Assert(fmt[strlen(fmt) - 1] != '\n');
86 : 0 : fprintf(stderr, "\n");
87 : 0 : exit(1);
88 : : }
89 : :
90 : : /*
91 : : * Check that libpq next returns a PGresult with the specified status,
92 : : * returning the PGresult so that caller can perform additional checks.
93 : : */
94 : : #define confirm_result_status(conn, status) confirm_result_status_impl(__LINE__, conn, status)
95 : : static PGresult *
96 : 56 : confirm_result_status_impl(int line, PGconn *conn, ExecStatusType status)
97 : : {
98 : : PGresult *res;
99 : :
100 : 56 : res = PQgetResult(conn);
101 [ - + ]: 56 : if (res == NULL)
102 : 0 : pg_fatal_impl(line, "PQgetResult returned null unexpectedly: %s",
103 : : PQerrorMessage(conn));
104 [ - + ]: 56 : if (PQresultStatus(res) != status)
105 : 0 : pg_fatal_impl(line, "PQgetResult returned status %s, expected %s: %s",
106 : : PQresStatus(PQresultStatus(res)),
107 : : PQresStatus(status),
108 : : PQerrorMessage(conn));
109 : 56 : return res;
110 : : }
111 : :
112 : : /*
113 : : * Check that libpq next returns a PGresult with the specified status,
114 : : * then free the PGresult.
115 : : */
116 : : #define consume_result_status(conn, status) consume_result_status_impl(__LINE__, conn, status)
117 : : static void
118 : 39 : consume_result_status_impl(int line, PGconn *conn, ExecStatusType status)
119 : : {
120 : : PGresult *res;
121 : :
122 : 39 : res = confirm_result_status_impl(line, conn, status);
123 : 39 : PQclear(res);
124 : 39 : }
125 : :
126 : : /*
127 : : * Check that libpq next returns a null PGresult.
128 : : */
129 : : #define consume_null_result(conn) consume_null_result_impl(__LINE__, conn)
130 : : static void
131 : 631 : consume_null_result_impl(int line, PGconn *conn)
132 : : {
133 : : PGresult *res;
134 : :
135 : 631 : res = PQgetResult(conn);
136 [ - + ]: 631 : if (res != NULL)
137 : 0 : pg_fatal_impl(line, "expected NULL PGresult, got %s: %s",
138 : : PQresStatus(PQresultStatus(res)),
139 : : PQerrorMessage(conn));
140 : 631 : }
141 : :
142 : : /*
143 : : * Check that the query on the given connection got canceled.
144 : : */
145 : : #define consume_query_cancel(conn) consume_query_cancel_impl(__LINE__, conn)
146 : : static void
147 : 12 : consume_query_cancel_impl(int line, PGconn *conn)
148 : : {
149 : : PGresult *res;
150 : :
151 : 12 : res = confirm_result_status_impl(line, conn, PGRES_FATAL_ERROR);
152 [ - + ]: 12 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "57014") != 0)
153 : 0 : pg_fatal_impl(line, "query failed with a different error than cancellation: %s",
154 : : PQerrorMessage(conn));
155 : 12 : PQclear(res);
156 : :
157 [ - + ]: 12 : while (PQisBusy(conn))
158 : 0 : PQconsumeInput(conn);
159 : 12 : }
160 : :
161 : : /*
162 : : * Using monitorConn, query pg_stat_activity to see that the connection with
163 : : * the given PID is either in the given state, or waiting on the given event
164 : : * (only one of them can be given).
165 : : */
166 : : static void
167 : 24 : wait_for_connection_state(int line, PGconn *monitorConn, int procpid,
168 : : char *state, char *event)
169 : : {
170 : 24 : const Oid paramTypes[] = {INT4OID, TEXTOID};
171 : : const char *paramValues[2];
172 : 24 : char *pidstr = psprintf("%d", procpid);
173 : :
174 : : Assert((state == NULL) ^ (event == NULL));
175 : :
176 : 24 : paramValues[0] = pidstr;
177 [ + + ]: 24 : paramValues[1] = state ? state : event;
178 : :
179 : : while (true)
180 : 1 : {
181 : : PGresult *res;
182 : : char *value;
183 : :
184 [ + + ]: 25 : if (state != NULL)
185 : 12 : res = PQexecParams(monitorConn,
186 : : "SELECT count(*) FROM pg_stat_activity WHERE "
187 : : "pid = $1 AND state = $2",
188 : : 2, paramTypes, paramValues, NULL, NULL, 0);
189 : : else
190 : 13 : res = PQexecParams(monitorConn,
191 : : "SELECT count(*) FROM pg_stat_activity WHERE "
192 : : "pid = $1 AND wait_event = $2",
193 : : 2, paramTypes, paramValues, NULL, NULL, 0);
194 : :
195 [ - + ]: 25 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
196 : 0 : pg_fatal_impl(line, "could not query pg_stat_activity: %s", PQerrorMessage(monitorConn));
197 [ - + ]: 25 : if (PQntuples(res) != 1)
198 : 0 : pg_fatal_impl(line, "unexpected number of rows received: %d", PQntuples(res));
199 [ - + ]: 25 : if (PQnfields(res) != 1)
200 : 0 : pg_fatal_impl(line, "unexpected number of columns received: %d", PQnfields(res));
201 : 25 : value = PQgetvalue(res, 0, 0);
202 [ + + ]: 25 : if (strcmp(value, "0") != 0)
203 : : {
204 : 24 : PQclear(res);
205 : 24 : break;
206 : : }
207 : 1 : PQclear(res);
208 : :
209 : : /* wait 10ms before polling again */
210 : 1 : pg_usleep(10000);
211 : : }
212 : :
213 : 24 : pfree(pidstr);
214 : 24 : }
215 : :
216 : : #define send_cancellable_query(conn, monitorConn) \
217 : : send_cancellable_query_impl(__LINE__, conn, monitorConn)
218 : : static void
219 : 12 : send_cancellable_query_impl(int line, PGconn *conn, PGconn *monitorConn)
220 : : {
221 : : const char *env_wait;
222 : 12 : const Oid paramTypes[1] = {INT4OID};
223 : :
224 : : /*
225 : : * Wait for the connection to be idle, so that our check for an active
226 : : * connection below is reliable, instead of possibly seeing an outdated
227 : : * state.
228 : : */
229 : 12 : wait_for_connection_state(line, monitorConn, PQbackendPID(conn), "idle", NULL);
230 : :
231 : 12 : env_wait = getenv("PG_TEST_TIMEOUT_DEFAULT");
232 [ + - ]: 12 : if (env_wait == NULL)
233 : 12 : env_wait = "180";
234 : :
235 [ - + ]: 12 : if (PQsendQueryParams(conn, "SELECT pg_sleep($1)", 1, paramTypes,
236 : : &env_wait, NULL, NULL, 0) != 1)
237 : 0 : pg_fatal_impl(line, "failed to send query: %s", PQerrorMessage(conn));
238 : :
239 : : /*
240 : : * Wait for the sleep to be active, because if the query is not running
241 : : * yet, the cancel request that we send won't have any effect.
242 : : */
243 : 12 : wait_for_connection_state(line, monitorConn, PQbackendPID(conn), NULL, "PgSleep");
244 : 12 : }
245 : :
246 : : /*
247 : : * Create a new connection with the same conninfo as the given one.
248 : : */
249 : : static PGconn *
250 : 2 : copy_connection(PGconn *conn)
251 : : {
252 : : PGconn *copyConn;
253 : 2 : PQconninfoOption *opts = PQconninfo(conn);
254 : : const char **keywords;
255 : : const char **vals;
256 : 2 : int nopts = 0;
257 : : int i;
258 : :
259 [ + + ]: 104 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
260 : 102 : nopts++;
261 : 2 : nopts++; /* for the NULL terminator */
262 : :
263 : 2 : keywords = pg_malloc_array(const char *, nopts);
264 : 2 : vals = pg_malloc_array(const char *, nopts);
265 : :
266 : 2 : i = 0;
267 [ + + ]: 104 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
268 : : {
269 [ + + ]: 102 : if (opt->val)
270 : : {
271 : 40 : keywords[i] = opt->keyword;
272 : 40 : vals[i] = opt->val;
273 : 40 : i++;
274 : : }
275 : : }
276 : 2 : keywords[i] = vals[i] = NULL;
277 : :
278 : 2 : copyConn = PQconnectdbParams(keywords, vals, false);
279 : :
280 [ - + ]: 2 : if (PQstatus(copyConn) != CONNECTION_OK)
281 : 0 : pg_fatal("Connection to database failed: %s",
282 : : PQerrorMessage(copyConn));
283 : :
284 : 2 : pg_free(keywords);
285 : 2 : pg_free(vals);
286 : 2 : PQconninfoFree(opts);
287 : :
288 : 2 : return copyConn;
289 : : }
290 : :
291 : : /*
292 : : * Test query cancellation routines
293 : : */
294 : : static void
295 : 2 : test_cancel(PGconn *conn)
296 : : {
297 : : PGcancel *cancel;
298 : : PGcancelConn *cancelConn;
299 : : PGconn *monitorConn;
300 : : char errorbuf[256];
301 : :
302 : 2 : fprintf(stderr, "test cancellations... ");
303 : :
304 [ - + ]: 2 : if (PQsetnonblocking(conn, 1) != 0)
305 : 0 : pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn));
306 : :
307 : : /*
308 : : * Make a separate connection to the database to monitor the query on the
309 : : * main connection.
310 : : */
311 : 2 : monitorConn = copy_connection(conn);
312 : : Assert(PQstatus(monitorConn) == CONNECTION_OK);
313 : :
314 : : /* test PQcancel */
315 : 2 : send_cancellable_query(conn, monitorConn);
316 : 2 : cancel = PQgetCancel(conn);
317 [ - + ]: 2 : if (!PQcancel(cancel, errorbuf, sizeof(errorbuf)))
318 : 0 : pg_fatal("failed to run PQcancel: %s", errorbuf);
319 : 2 : consume_query_cancel(conn);
320 : :
321 : : /* PGcancel object can be reused for the next query */
322 : 2 : send_cancellable_query(conn, monitorConn);
323 [ - + ]: 2 : if (!PQcancel(cancel, errorbuf, sizeof(errorbuf)))
324 : 0 : pg_fatal("failed to run PQcancel: %s", errorbuf);
325 : 2 : consume_query_cancel(conn);
326 : :
327 : 2 : PQfreeCancel(cancel);
328 : :
329 : : /* test PQrequestCancel */
330 : 2 : send_cancellable_query(conn, monitorConn);
331 [ - + ]: 2 : if (!PQrequestCancel(conn))
332 : 0 : pg_fatal("failed to run PQrequestCancel: %s", PQerrorMessage(conn));
333 : 2 : consume_query_cancel(conn);
334 : :
335 : : /* test PQcancelBlocking */
336 : 2 : send_cancellable_query(conn, monitorConn);
337 : 2 : cancelConn = PQcancelCreate(conn);
338 [ - + ]: 2 : if (!PQcancelBlocking(cancelConn))
339 : 0 : pg_fatal("failed to run PQcancelBlocking: %s", PQcancelErrorMessage(cancelConn));
340 : 2 : consume_query_cancel(conn);
341 : 2 : PQcancelFinish(cancelConn);
342 : :
343 : : /* test PQcancelCreate and then polling with PQcancelPoll */
344 : 2 : send_cancellable_query(conn, monitorConn);
345 : 2 : cancelConn = PQcancelCreate(conn);
346 [ - + ]: 2 : if (!PQcancelStart(cancelConn))
347 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
348 : : while (true)
349 : 2 : {
350 : : struct timeval tv;
351 : : fd_set input_mask;
352 : : fd_set output_mask;
353 : 4 : PostgresPollingStatusType pollres = PQcancelPoll(cancelConn);
354 : 4 : int sock = PQcancelSocket(cancelConn);
355 : :
356 [ + + ]: 4 : if (pollres == PGRES_POLLING_OK)
357 : 2 : break;
358 : :
359 [ + + ]: 34 : FD_ZERO(&input_mask);
360 [ + + ]: 34 : FD_ZERO(&output_mask);
361 [ + - - ]: 2 : switch (pollres)
362 : : {
363 : 2 : case PGRES_POLLING_READING:
364 : : pg_debug("polling for reads\n");
365 : 2 : FD_SET(sock, &input_mask);
366 : 2 : break;
367 : 0 : case PGRES_POLLING_WRITING:
368 : : pg_debug("polling for writes\n");
369 : 0 : FD_SET(sock, &output_mask);
370 : 0 : break;
371 : 0 : default:
372 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
373 : : }
374 : :
375 [ - + ]: 2 : if (sock < 0)
376 : 0 : pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn));
377 : :
378 : 2 : tv.tv_sec = 3;
379 : 2 : tv.tv_usec = 0;
380 : :
381 : : while (true)
382 : : {
383 [ - + ]: 2 : if (select(sock + 1, &input_mask, &output_mask, NULL, &tv) < 0)
384 : : {
385 [ # # ]: 0 : if (errno == EINTR)
386 : 0 : continue;
387 : 0 : pg_fatal("select() failed: %m");
388 : : }
389 : 2 : break;
390 : : }
391 : : }
392 [ - + ]: 2 : if (PQcancelStatus(cancelConn) != CONNECTION_OK)
393 : 0 : pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn));
394 : 2 : consume_query_cancel(conn);
395 : :
396 : : /*
397 : : * test PQcancelReset works on the cancel connection and it can be reused
398 : : * afterwards
399 : : */
400 : 2 : PQcancelReset(cancelConn);
401 : :
402 : 2 : send_cancellable_query(conn, monitorConn);
403 [ - + ]: 2 : if (!PQcancelStart(cancelConn))
404 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
405 : : while (true)
406 : 2 : {
407 : : struct timeval tv;
408 : : fd_set input_mask;
409 : : fd_set output_mask;
410 : 4 : PostgresPollingStatusType pollres = PQcancelPoll(cancelConn);
411 : 4 : int sock = PQcancelSocket(cancelConn);
412 : :
413 [ + + ]: 4 : if (pollres == PGRES_POLLING_OK)
414 : 2 : break;
415 : :
416 [ + + ]: 34 : FD_ZERO(&input_mask);
417 [ + + ]: 34 : FD_ZERO(&output_mask);
418 [ + - - ]: 2 : switch (pollres)
419 : : {
420 : 2 : case PGRES_POLLING_READING:
421 : : pg_debug("polling for reads\n");
422 : 2 : FD_SET(sock, &input_mask);
423 : 2 : break;
424 : 0 : case PGRES_POLLING_WRITING:
425 : : pg_debug("polling for writes\n");
426 : 0 : FD_SET(sock, &output_mask);
427 : 0 : break;
428 : 0 : default:
429 : 0 : pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn));
430 : : }
431 : :
432 [ - + ]: 2 : if (sock < 0)
433 : 0 : pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn));
434 : :
435 : 2 : tv.tv_sec = 3;
436 : 2 : tv.tv_usec = 0;
437 : :
438 : : while (true)
439 : : {
440 [ - + ]: 2 : if (select(sock + 1, &input_mask, &output_mask, NULL, &tv) < 0)
441 : : {
442 [ # # ]: 0 : if (errno == EINTR)
443 : 0 : continue;
444 : 0 : pg_fatal("select() failed: %m");
445 : : }
446 : 2 : break;
447 : : }
448 : : }
449 [ - + ]: 2 : if (PQcancelStatus(cancelConn) != CONNECTION_OK)
450 : 0 : pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn));
451 : 2 : consume_query_cancel(conn);
452 : :
453 : 2 : PQcancelFinish(cancelConn);
454 : 2 : PQfinish(monitorConn);
455 : :
456 : 2 : fprintf(stderr, "ok\n");
457 : 2 : }
458 : :
459 : : /*
460 : : * Test Describe of a prepared FETCH statement after the cursor it
461 : : * references has been closed.
462 : : */
463 : : static void
464 : 1 : test_describe_fetch(PGconn *conn)
465 : : {
466 : : PGresult *res;
467 : :
468 : 1 : fprintf(stderr, "test cursor describe and fetch... ");
469 : :
470 : 1 : res = PQexec(conn, "BEGIN");
471 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
472 : 0 : pg_fatal("BEGIN failed: %s", PQerrorMessage(conn));
473 : 1 : PQclear(res);
474 : :
475 : 1 : res = PQexec(conn, "DECLARE fetch_cursor CURSOR FOR SELECT 1");
476 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
477 : 0 : pg_fatal("DECLARE CURSOR failed: %s", PQerrorMessage(conn));
478 : 1 : PQclear(res);
479 : :
480 : : /* prepare while the cursor exists, so that it caches a result desc */
481 : 1 : res = PQprepare(conn, "fetch_one", "FETCH 1 FROM fetch_cursor", 0, NULL);
482 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
483 : 0 : pg_fatal("PQprepare failed: %s", PQerrorMessage(conn));
484 : 1 : PQclear(res);
485 : :
486 : 1 : res = PQexec(conn, "CLOSE fetch_cursor");
487 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
488 : 0 : pg_fatal("CLOSE failed: %s", PQerrorMessage(conn));
489 : 1 : PQclear(res);
490 : :
491 : : /* describe fails after the cursor has been closed */
492 : 1 : res = PQdescribePrepared(conn, "fetch_one");
493 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
494 : 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
495 : 1 : PQclear(res);
496 : :
497 : 1 : res = PQexec(conn, "ROLLBACK");
498 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
499 : 0 : pg_fatal("ROLLBACK failed: %s", PQerrorMessage(conn));
500 : 1 : PQclear(res);
501 : :
502 : 1 : fprintf(stderr, "ok\n");
503 : 1 : }
504 : :
505 : : static void
506 : 1 : test_disallowed_in_pipeline(PGconn *conn)
507 : : {
508 : 1 : PGresult *res = NULL;
509 : :
510 : 1 : fprintf(stderr, "test error cases... ");
511 : :
512 [ - + ]: 1 : if (PQisnonblocking(conn))
513 : 0 : pg_fatal("Expected blocking connection mode");
514 : :
515 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
516 : 0 : pg_fatal("Unable to enter pipeline mode");
517 : :
518 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
519 : 0 : pg_fatal("Pipeline mode not activated properly");
520 : :
521 : : /* PQexec should fail in pipeline mode */
522 : 1 : res = PQexec(conn, "SELECT 1");
523 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
524 : 0 : pg_fatal("PQexec should fail in pipeline mode but succeeded");
525 [ - + ]: 1 : if (strcmp(PQerrorMessage(conn),
526 : : "synchronous command execution functions are not allowed in pipeline mode\n") != 0)
527 : 0 : pg_fatal("did not get expected error message; got: \"%s\"",
528 : : PQerrorMessage(conn));
529 : 1 : PQclear(res);
530 : :
531 : : /* PQsendQuery should fail in pipeline mode */
532 [ - + ]: 1 : if (PQsendQuery(conn, "SELECT 1") != 0)
533 : 0 : pg_fatal("PQsendQuery should fail in pipeline mode but succeeded");
534 [ - + ]: 1 : if (strcmp(PQerrorMessage(conn),
535 : : "PQsendQuery not allowed in pipeline mode\n") != 0)
536 : 0 : pg_fatal("did not get expected error message; got: \"%s\"",
537 : : PQerrorMessage(conn));
538 : :
539 : : /* Entering pipeline mode when already in pipeline mode is OK */
540 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
541 : 0 : pg_fatal("re-entering pipeline mode should be a no-op but failed");
542 : :
543 [ - + ]: 1 : if (PQisBusy(conn) != 0)
544 : 0 : pg_fatal("PQisBusy should return 0 when idle in pipeline mode, returned 1");
545 : :
546 : : /* ok, back to normal command mode */
547 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
548 : 0 : pg_fatal("couldn't exit idle empty pipeline mode");
549 : :
550 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
551 : 0 : pg_fatal("Pipeline mode not terminated properly");
552 : :
553 : : /* exiting pipeline mode when not in pipeline mode should be a no-op */
554 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
555 : 0 : pg_fatal("pipeline mode exit when not in pipeline mode should succeed but failed");
556 : :
557 : : /* can now PQexec again */
558 : 1 : res = PQexec(conn, "SELECT 1");
559 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
560 : 0 : pg_fatal("PQexec should succeed after exiting pipeline mode but failed with: %s",
561 : : PQerrorMessage(conn));
562 : 1 : PQclear(res);
563 : :
564 : 1 : fprintf(stderr, "ok\n");
565 : 1 : }
566 : :
567 : : static void
568 : 1 : test_multi_pipelines(PGconn *conn)
569 : : {
570 : 1 : const char *dummy_params[1] = {"1"};
571 : 1 : Oid dummy_param_oids[1] = {INT4OID};
572 : :
573 : 1 : fprintf(stderr, "multi pipeline... ");
574 : :
575 : : /*
576 : : * Queue up a couple of small pipelines and process each without returning
577 : : * to command mode first.
578 : : */
579 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
580 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
581 : :
582 : : /* first pipeline */
583 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
584 : : dummy_params, NULL, NULL, 0) != 1)
585 : 0 : pg_fatal("dispatching first SELECT failed: %s", PQerrorMessage(conn));
586 : :
587 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
588 : 0 : pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
589 : :
590 : : /* second pipeline */
591 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
592 : : dummy_params, NULL, NULL, 0) != 1)
593 : 0 : pg_fatal("dispatching second SELECT failed: %s", PQerrorMessage(conn));
594 : :
595 : : /* Skip flushing once. */
596 [ - + ]: 1 : if (PQsendPipelineSync(conn) != 1)
597 : 0 : pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn));
598 : :
599 : : /* third pipeline */
600 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids,
601 : : dummy_params, NULL, NULL, 0) != 1)
602 : 0 : pg_fatal("dispatching third SELECT failed: %s", PQerrorMessage(conn));
603 : :
604 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
605 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
606 : :
607 : : /* OK, start processing the results */
608 : :
609 : : /* first pipeline */
610 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
611 : :
612 : 1 : consume_null_result(conn);
613 : :
614 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
615 : 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
616 : :
617 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
618 : :
619 : : /* second pipeline */
620 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
621 : :
622 : 1 : consume_null_result(conn);
623 : :
624 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
625 : 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
626 : :
627 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
628 : :
629 : : /* third pipeline */
630 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
631 : :
632 : 1 : consume_null_result(conn);
633 : :
634 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
635 : :
636 : : /* We're still in pipeline mode ... */
637 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
638 : 0 : pg_fatal("Fell out of pipeline mode somehow");
639 : :
640 : : /* until we end it, which we can safely do now */
641 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
642 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
643 : : PQerrorMessage(conn));
644 : :
645 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
646 : 0 : pg_fatal("exiting pipeline mode didn't seem to work");
647 : :
648 : 1 : fprintf(stderr, "ok\n");
649 : 1 : }
650 : :
651 : : /*
652 : : * Test behavior when a pipeline dispatches a number of commands that are
653 : : * not flushed by a sync point.
654 : : */
655 : : static void
656 : 1 : test_nosync(PGconn *conn)
657 : : {
658 : 1 : int numqueries = 10;
659 : 1 : int results = 0;
660 : 1 : int sock = PQsocket(conn);
661 : :
662 : 1 : fprintf(stderr, "nosync... ");
663 : :
664 [ - + ]: 1 : if (sock < 0)
665 : 0 : pg_fatal("invalid socket");
666 : :
667 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
668 : 0 : pg_fatal("could not enter pipeline mode");
669 [ + + ]: 11 : for (int i = 0; i < numqueries; i++)
670 : : {
671 : : fd_set input_mask;
672 : : struct timeval tv;
673 : :
674 [ - + ]: 10 : if (PQsendQueryParams(conn, "SELECT repeat('xyzxz', 12)",
675 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
676 : 0 : pg_fatal("error sending select: %s", PQerrorMessage(conn));
677 : 10 : PQflush(conn);
678 : :
679 : : /*
680 : : * If the server has written anything to us, read (some of) it now.
681 : : */
682 [ + + ]: 170 : FD_ZERO(&input_mask);
683 : 10 : FD_SET(sock, &input_mask);
684 : 10 : tv.tv_sec = 0;
685 : 10 : tv.tv_usec = 0;
686 [ - + ]: 10 : if (select(sock + 1, &input_mask, NULL, NULL, &tv) < 0)
687 : : {
688 : 0 : fprintf(stderr, "select() failed: %m\n");
689 : 0 : exit_nicely(conn);
690 : : }
691 [ - + - - ]: 10 : if (FD_ISSET(sock, &input_mask) && PQconsumeInput(conn) != 1)
692 : 0 : pg_fatal("failed to read from server: %s", PQerrorMessage(conn));
693 : : }
694 : :
695 : : /* tell server to flush its output buffer */
696 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
697 : 0 : pg_fatal("failed to send flush request");
698 : 1 : PQflush(conn);
699 : :
700 : : /* Now read all results */
701 : : for (;;)
702 : : {
703 : : /* We expect exactly one TUPLES_OK result for each query we sent */
704 : 10 : consume_result_status(conn, PGRES_TUPLES_OK);
705 : :
706 : : /* and one NULL result should follow each */
707 : 10 : consume_null_result(conn);
708 : :
709 : 10 : results++;
710 : :
711 : : /* if we're done, we're done */
712 [ + + ]: 10 : if (results == numqueries)
713 : 1 : break;
714 : : }
715 : :
716 : 1 : fprintf(stderr, "ok\n");
717 : 1 : }
718 : :
719 : : /*
720 : : * When an operation in a pipeline fails the rest of the pipeline is flushed. We
721 : : * still have to get results for each pipeline item, but the item will just be
722 : : * a PGRES_PIPELINE_ABORTED code.
723 : : *
724 : : * This intentionally doesn't use a transaction to wrap the pipeline. You should
725 : : * usually use an xact, but in this case we want to observe the effects of each
726 : : * statement.
727 : : */
728 : : static void
729 : 1 : test_pipeline_abort(PGconn *conn)
730 : : {
731 : 1 : PGresult *res = NULL;
732 : 1 : const char *dummy_params[1] = {"1"};
733 : 1 : Oid dummy_param_oids[1] = {INT4OID};
734 : : int i;
735 : : int gotrows;
736 : : bool goterror;
737 : :
738 : 1 : fprintf(stderr, "aborted pipeline... ");
739 : :
740 : 1 : res = PQexec(conn, drop_table_sql);
741 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
742 : 0 : pg_fatal("dispatching DROP TABLE failed: %s", PQerrorMessage(conn));
743 : 1 : PQclear(res);
744 : :
745 : 1 : res = PQexec(conn, create_table_sql);
746 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
747 : 0 : pg_fatal("dispatching CREATE TABLE failed: %s", PQerrorMessage(conn));
748 : 1 : PQclear(res);
749 : :
750 : : /*
751 : : * Queue up a couple of small pipelines and process each without returning
752 : : * to command mode first. Make sure the second operation in the first
753 : : * pipeline ERRORs.
754 : : */
755 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
756 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
757 : :
758 : 1 : dummy_params[0] = "1";
759 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
760 : : dummy_params, NULL, NULL, 0) != 1)
761 : 0 : pg_fatal("dispatching first insert failed: %s", PQerrorMessage(conn));
762 : :
763 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT no_such_function($1)",
764 : : 1, dummy_param_oids, dummy_params,
765 : : NULL, NULL, 0) != 1)
766 : 0 : pg_fatal("dispatching error select failed: %s", PQerrorMessage(conn));
767 : :
768 : 1 : dummy_params[0] = "2";
769 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
770 : : dummy_params, NULL, NULL, 0) != 1)
771 : 0 : pg_fatal("dispatching second insert failed: %s", PQerrorMessage(conn));
772 : :
773 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
774 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
775 : :
776 : 1 : dummy_params[0] = "3";
777 [ - + ]: 1 : if (PQsendQueryParams(conn, insert_sql, 1, dummy_param_oids,
778 : : dummy_params, NULL, NULL, 0) != 1)
779 : 0 : pg_fatal("dispatching second-pipeline insert failed: %s",
780 : : PQerrorMessage(conn));
781 : :
782 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
783 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
784 : :
785 : : /*
786 : : * OK, start processing the pipeline results.
787 : : *
788 : : * We should get a command-ok for the first query, then a fatal error and
789 : : * a pipeline aborted message for the second insert, a pipeline-end, then
790 : : * a command-ok and a pipeline-ok for the second pipeline operation.
791 : : */
792 : 1 : consume_result_status(conn, PGRES_COMMAND_OK);
793 : :
794 : : /* NULL result to signal end-of-results for this command */
795 : 1 : consume_null_result(conn);
796 : :
797 : : /* Second query caused error, so we expect an error next */
798 : 1 : consume_result_status(conn, PGRES_FATAL_ERROR);
799 : :
800 : : /* NULL result to signal end-of-results for this command */
801 : 1 : consume_null_result(conn);
802 : :
803 : : /*
804 : : * pipeline should now be aborted.
805 : : *
806 : : * Note that we could still queue more queries at this point if we wanted;
807 : : * they'd get added to a new third pipeline since we've already sent a
808 : : * second. The aborted flag relates only to the pipeline being received.
809 : : */
810 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
811 : 0 : pg_fatal("pipeline should be flagged as aborted but isn't");
812 : :
813 : : /* third query in pipeline, the second insert */
814 : 1 : consume_result_status(conn, PGRES_PIPELINE_ABORTED);
815 : :
816 : : /* NULL result to signal end-of-results for this command */
817 : 1 : consume_null_result(conn);
818 : :
819 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_ABORTED)
820 : 0 : pg_fatal("pipeline should be flagged as aborted but isn't");
821 : :
822 : : /* Ensure we're still in pipeline */
823 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
824 : 0 : pg_fatal("Fell out of pipeline mode somehow");
825 : :
826 : : /*
827 : : * The end of a failed pipeline is a PGRES_PIPELINE_SYNC.
828 : : *
829 : : * (This is so clients know to start processing results normally again and
830 : : * can tell the difference between skipped commands and the sync.)
831 : : */
832 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
833 : :
834 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_ABORTED)
835 : 0 : pg_fatal("sync should've cleared the aborted flag but didn't");
836 : :
837 : : /* We're still in pipeline mode... */
838 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
839 : 0 : pg_fatal("Fell out of pipeline mode somehow");
840 : :
841 : : /* the insert from the second pipeline */
842 : 1 : consume_result_status(conn, PGRES_COMMAND_OK);
843 : :
844 : : /* Read the NULL result at the end of the command */
845 : 1 : consume_null_result(conn);
846 : :
847 : : /* the second pipeline sync */
848 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
849 : :
850 : : /* Read the NULL result at the end of the command */
851 : 1 : consume_null_result(conn);
852 : :
853 : : /* Try to send two queries in one command */
854 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1; SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
855 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
856 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
857 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
858 : 1 : goterror = false;
859 [ + + ]: 2 : while ((res = PQgetResult(conn)) != NULL)
860 : : {
861 [ + - ]: 1 : switch (PQresultStatus(res))
862 : : {
863 : 1 : case PGRES_FATAL_ERROR:
864 [ - + ]: 1 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "42601") != 0)
865 : 0 : pg_fatal("expected error about multiple commands, got %s",
866 : : PQerrorMessage(conn));
867 : 1 : printf("got expected %s", PQerrorMessage(conn));
868 : 1 : goterror = true;
869 : 1 : break;
870 : 0 : default:
871 : 0 : pg_fatal("got unexpected status %s", PQresStatus(PQresultStatus(res)));
872 : : break;
873 : : }
874 : 1 : PQclear(res);
875 : : }
876 [ - + ]: 1 : if (!goterror)
877 : 0 : pg_fatal("did not get cannot-insert-multiple-commands error");
878 : :
879 : : /* the second pipeline sync */
880 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
881 : :
882 : 1 : fprintf(stderr, "ok\n");
883 : :
884 : : /* Test single-row mode with an error partways */
885 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1.0/g FROM generate_series(3, -1, -1) g",
886 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
887 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
888 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
889 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
890 : 1 : PQsetSingleRowMode(conn);
891 : 1 : goterror = false;
892 : 1 : gotrows = 0;
893 [ + + ]: 5 : while ((res = PQgetResult(conn)) != NULL)
894 : : {
895 [ + + - ]: 4 : switch (PQresultStatus(res))
896 : : {
897 : 3 : case PGRES_SINGLE_TUPLE:
898 : 3 : printf("got row: %s\n", PQgetvalue(res, 0, 0));
899 : 3 : gotrows++;
900 : 3 : break;
901 : 1 : case PGRES_FATAL_ERROR:
902 [ - + ]: 1 : if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "22012") != 0)
903 : 0 : pg_fatal("expected division-by-zero, got: %s (%s)",
904 : : PQerrorMessage(conn),
905 : : PQresultErrorField(res, PG_DIAG_SQLSTATE));
906 : 1 : printf("got expected division-by-zero\n");
907 : 1 : goterror = true;
908 : 1 : break;
909 : 0 : default:
910 : 0 : pg_fatal("got unexpected result %s", PQresStatus(PQresultStatus(res)));
911 : : }
912 : 4 : PQclear(res);
913 : : }
914 [ - + ]: 1 : if (!goterror)
915 : 0 : pg_fatal("did not get division-by-zero error");
916 [ - + ]: 1 : if (gotrows != 3)
917 : 0 : pg_fatal("did not get three rows");
918 : :
919 : : /* the third pipeline sync */
920 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
921 : :
922 : : /* We're still in pipeline mode... */
923 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
924 : 0 : pg_fatal("Fell out of pipeline mode somehow");
925 : :
926 : : /* until we end it, which we can safely do now */
927 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
928 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
929 : : PQerrorMessage(conn));
930 : :
931 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
932 : 0 : pg_fatal("exiting pipeline mode didn't seem to work");
933 : :
934 : : /*-
935 : : * Since we fired the pipelines off without a surrounding xact, the results
936 : : * should be:
937 : : *
938 : : * - Implicit xact started by server around 1st pipeline
939 : : * - First insert applied
940 : : * - Second statement aborted xact
941 : : * - Third insert skipped
942 : : * - Sync rolled back first implicit xact
943 : : * - Implicit xact created by server around 2nd pipeline
944 : : * - insert applied from 2nd pipeline
945 : : * - Sync commits 2nd xact
946 : : *
947 : : * So we should only have the value 3 that we inserted.
948 : : */
949 : 1 : res = PQexec(conn, "SELECT itemno FROM pq_pipeline_demo");
950 : :
951 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
952 : 0 : pg_fatal("Expected tuples, got %s: %s",
953 : : PQresStatus(PQresultStatus(res)), PQerrorMessage(conn));
954 [ - + ]: 1 : if (PQntuples(res) != 1)
955 : 0 : pg_fatal("expected 1 result, got %d", PQntuples(res));
956 [ + + ]: 2 : for (i = 0; i < PQntuples(res); i++)
957 : : {
958 : 1 : const char *val = PQgetvalue(res, i, 0);
959 : :
960 [ - + ]: 1 : if (strcmp(val, "3") != 0)
961 : 0 : pg_fatal("expected only insert with value 3, got %s", val);
962 : : }
963 : :
964 : 1 : PQclear(res);
965 : :
966 : 1 : fprintf(stderr, "ok\n");
967 : 1 : }
968 : :
969 : : /* State machine enum for test_pipelined_insert */
970 : : enum PipelineInsertStep
971 : : {
972 : : BI_BEGIN_TX,
973 : : BI_DROP_TABLE,
974 : : BI_CREATE_TABLE,
975 : : BI_PREPARE,
976 : : BI_INSERT_ROWS,
977 : : BI_COMMIT_TX,
978 : : BI_SYNC,
979 : : BI_DONE,
980 : : };
981 : :
982 : : static void
983 : 1 : test_pipelined_insert(PGconn *conn, int n_rows)
984 : : {
985 : 1 : Oid insert_param_oids[2] = {INT4OID, INT8OID};
986 : : const char *insert_params[2];
987 : : char insert_param_0[MAXINTLEN];
988 : : char insert_param_1[MAXINT8LEN];
989 : 1 : enum PipelineInsertStep send_step = BI_BEGIN_TX,
990 : 1 : recv_step = BI_BEGIN_TX;
991 : : int rows_to_send,
992 : : rows_to_receive;
993 : :
994 : 1 : insert_params[0] = insert_param_0;
995 : 1 : insert_params[1] = insert_param_1;
996 : :
997 : 1 : rows_to_send = rows_to_receive = n_rows;
998 : :
999 : : /*
1000 : : * Do a pipelined insert into a table created at the start of the pipeline
1001 : : */
1002 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1003 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1004 : :
1005 [ + + ]: 4 : while (send_step != BI_PREPARE)
1006 : : {
1007 : : const char *sql;
1008 : :
1009 [ + + + - ]: 3 : switch (send_step)
1010 : : {
1011 : 1 : case BI_BEGIN_TX:
1012 : 1 : sql = "BEGIN TRANSACTION";
1013 : 1 : send_step = BI_DROP_TABLE;
1014 : 1 : break;
1015 : :
1016 : 1 : case BI_DROP_TABLE:
1017 : 1 : sql = drop_table_sql;
1018 : 1 : send_step = BI_CREATE_TABLE;
1019 : 1 : break;
1020 : :
1021 : 1 : case BI_CREATE_TABLE:
1022 : 1 : sql = create_table_sql;
1023 : 1 : send_step = BI_PREPARE;
1024 : 1 : break;
1025 : :
1026 : 0 : default:
1027 : 0 : pg_fatal("invalid state");
1028 : : sql = NULL; /* keep compiler quiet */
1029 : : }
1030 : :
1031 : : pg_debug("sending: %s\n", sql);
1032 [ - + ]: 3 : if (PQsendQueryParams(conn, sql,
1033 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1034 : 0 : pg_fatal("dispatching %s failed: %s", sql, PQerrorMessage(conn));
1035 : : }
1036 : :
1037 : : Assert(send_step == BI_PREPARE);
1038 : : pg_debug("sending: %s\n", insert_sql2);
1039 [ - + ]: 1 : if (PQsendPrepare(conn, "my_insert", insert_sql2, 2, insert_param_oids) != 1)
1040 : 0 : pg_fatal("dispatching PREPARE failed: %s", PQerrorMessage(conn));
1041 : 1 : send_step = BI_INSERT_ROWS;
1042 : :
1043 : : /*
1044 : : * Now we start inserting. We'll be sending enough data that we could fill
1045 : : * our output buffer, so to avoid deadlocking we need to enter nonblocking
1046 : : * mode and consume input while we send more output. As results of each
1047 : : * query are processed we should pop them to allow processing of the next
1048 : : * query. There's no need to finish the pipeline before processing
1049 : : * results.
1050 : : */
1051 [ - + ]: 1 : if (PQsetnonblocking(conn, 1) != 0)
1052 : 0 : pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn));
1053 : :
1054 [ + + ]: 734 : while (recv_step != BI_DONE)
1055 : : {
1056 : : int sock;
1057 : : fd_set input_mask;
1058 : : fd_set output_mask;
1059 : :
1060 : 733 : sock = PQsocket(conn);
1061 : :
1062 [ - + ]: 733 : if (sock < 0)
1063 : 0 : break; /* shouldn't happen */
1064 : :
1065 [ + + ]: 12461 : FD_ZERO(&input_mask);
1066 : 733 : FD_SET(sock, &input_mask);
1067 [ + + ]: 12461 : FD_ZERO(&output_mask);
1068 : 733 : FD_SET(sock, &output_mask);
1069 : :
1070 [ - + ]: 733 : if (select(sock + 1, &input_mask, &output_mask, NULL, NULL) < 0)
1071 : : {
1072 : 0 : fprintf(stderr, "select() failed: %m\n");
1073 : 0 : exit_nicely(conn);
1074 : : }
1075 : :
1076 : : /*
1077 : : * Process any results, so we keep the server's output buffer free
1078 : : * flowing and it can continue to process input
1079 : : */
1080 [ + + ]: 733 : if (FD_ISSET(sock, &input_mask))
1081 : : {
1082 : 3 : PQconsumeInput(conn);
1083 : :
1084 : : /* Read until we'd block if we tried to read */
1085 [ + + + + ]: 1414 : while (!PQisBusy(conn) && recv_step < BI_DONE)
1086 : : {
1087 : : PGresult *res;
1088 : 1411 : const char *cmdtag = "";
1089 : 1411 : const char *description = "";
1090 : : int status;
1091 : :
1092 : : /*
1093 : : * Read next result. If no more results from this query,
1094 : : * advance to the next query
1095 : : */
1096 : 1411 : res = PQgetResult(conn);
1097 [ + + ]: 1411 : if (res == NULL)
1098 : 705 : continue;
1099 : :
1100 : 706 : status = PGRES_COMMAND_OK;
1101 [ + + + + : 706 : switch (recv_step)
+ + + -
- ]
1102 : : {
1103 : 1 : case BI_BEGIN_TX:
1104 : 1 : cmdtag = "BEGIN";
1105 : 1 : recv_step++;
1106 : 1 : break;
1107 : 1 : case BI_DROP_TABLE:
1108 : 1 : cmdtag = "DROP TABLE";
1109 : 1 : recv_step++;
1110 : 1 : break;
1111 : 1 : case BI_CREATE_TABLE:
1112 : 1 : cmdtag = "CREATE TABLE";
1113 : 1 : recv_step++;
1114 : 1 : break;
1115 : 1 : case BI_PREPARE:
1116 : 1 : cmdtag = "";
1117 : 1 : description = "PREPARE";
1118 : 1 : recv_step++;
1119 : 1 : break;
1120 : 700 : case BI_INSERT_ROWS:
1121 : 700 : cmdtag = "INSERT";
1122 : 700 : rows_to_receive--;
1123 [ + + ]: 700 : if (rows_to_receive == 0)
1124 : 1 : recv_step++;
1125 : 700 : break;
1126 : 1 : case BI_COMMIT_TX:
1127 : 1 : cmdtag = "COMMIT";
1128 : 1 : recv_step++;
1129 : 1 : break;
1130 : 1 : case BI_SYNC:
1131 : 1 : cmdtag = "";
1132 : 1 : description = "SYNC";
1133 : 1 : status = PGRES_PIPELINE_SYNC;
1134 : 1 : recv_step++;
1135 : 1 : break;
1136 : 0 : case BI_DONE:
1137 : : /* unreachable */
1138 : 0 : pg_fatal("unreachable state");
1139 : : }
1140 : :
1141 [ - + ]: 706 : if (PQresultStatus(res) != status)
1142 : 0 : pg_fatal("%s reported status %s, expected %s\n"
1143 : : "Error message: \"%s\"",
1144 : : description, PQresStatus(PQresultStatus(res)),
1145 : : PQresStatus(status), PQerrorMessage(conn));
1146 : :
1147 [ - + ]: 706 : if (strncmp(PQcmdStatus(res), cmdtag, strlen(cmdtag)) != 0)
1148 : 0 : pg_fatal("%s expected command tag '%s', got '%s'",
1149 : : description, cmdtag, PQcmdStatus(res));
1150 : :
1151 : : pg_debug("Got %s OK\n", cmdtag[0] != '\0' ? cmdtag : description);
1152 : :
1153 : 706 : PQclear(res);
1154 : : }
1155 : : }
1156 : :
1157 : : /* Write more rows and/or the end pipeline message, if needed */
1158 [ + - ]: 733 : if (FD_ISSET(sock, &output_mask))
1159 : : {
1160 : 733 : PQflush(conn);
1161 : :
1162 [ + + ]: 733 : if (send_step == BI_INSERT_ROWS)
1163 : : {
1164 : 700 : snprintf(insert_param_0, MAXINTLEN, "%d", rows_to_send);
1165 : : /* use up some buffer space with a wide value */
1166 : 700 : snprintf(insert_param_1, MAXINT8LEN, "%lld", 1LL << 62);
1167 : :
1168 [ + - ]: 700 : if (PQsendQueryPrepared(conn, "my_insert",
1169 : : 2, insert_params, NULL, NULL, 0) == 1)
1170 : : {
1171 : : pg_debug("sent row %d\n", rows_to_send);
1172 : :
1173 : 700 : rows_to_send--;
1174 [ + + ]: 700 : if (rows_to_send == 0)
1175 : 1 : send_step++;
1176 : : }
1177 : : else
1178 : : {
1179 : : /*
1180 : : * in nonblocking mode, so it's OK for an insert to fail
1181 : : * to send
1182 : : */
1183 : 0 : fprintf(stderr, "WARNING: failed to send insert #%d: %s\n",
1184 : : rows_to_send, PQerrorMessage(conn));
1185 : : }
1186 : : }
1187 [ + + ]: 33 : else if (send_step == BI_COMMIT_TX)
1188 : : {
1189 [ + - ]: 1 : if (PQsendQueryParams(conn, "COMMIT",
1190 : : 0, NULL, NULL, NULL, NULL, 0) == 1)
1191 : : {
1192 : : pg_debug("sent COMMIT\n");
1193 : 1 : send_step++;
1194 : : }
1195 : : else
1196 : : {
1197 : 0 : fprintf(stderr, "WARNING: failed to send commit: %s\n",
1198 : : PQerrorMessage(conn));
1199 : : }
1200 : : }
1201 [ + + ]: 32 : else if (send_step == BI_SYNC)
1202 : : {
1203 [ + - ]: 1 : if (PQpipelineSync(conn) == 1)
1204 : : {
1205 : 1 : fprintf(stdout, "pipeline sync sent\n");
1206 : 1 : send_step++;
1207 : : }
1208 : : else
1209 : : {
1210 : 0 : fprintf(stderr, "WARNING: pipeline sync failed: %s\n",
1211 : : PQerrorMessage(conn));
1212 : : }
1213 : : }
1214 : : }
1215 : : }
1216 : :
1217 : : /* We've got the sync message and the pipeline should be done */
1218 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1219 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
1220 : : PQerrorMessage(conn));
1221 : :
1222 [ - + ]: 1 : if (PQsetnonblocking(conn, 0) != 0)
1223 : 0 : pg_fatal("failed to clear nonblocking mode: %s", PQerrorMessage(conn));
1224 : :
1225 : 1 : fprintf(stderr, "ok\n");
1226 : 1 : }
1227 : :
1228 : : static void
1229 : 1 : test_prepared(PGconn *conn)
1230 : : {
1231 : 1 : PGresult *res = NULL;
1232 : 1 : Oid param_oids[1] = {INT4OID};
1233 : : Oid expected_oids[4];
1234 : : Oid typ;
1235 : :
1236 : 1 : fprintf(stderr, "prepared... ");
1237 : :
1238 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1239 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1240 [ - + ]: 1 : if (PQsendPrepare(conn, "select_one", "SELECT $1, '42', $1::numeric, "
1241 : : "interval '1 sec'",
1242 : : 1, param_oids) != 1)
1243 : 0 : pg_fatal("preparing query failed: %s", PQerrorMessage(conn));
1244 : 1 : expected_oids[0] = INT4OID;
1245 : 1 : expected_oids[1] = TEXTOID;
1246 : 1 : expected_oids[2] = NUMERICOID;
1247 : 1 : expected_oids[3] = INTERVALOID;
1248 [ - + ]: 1 : if (PQsendDescribePrepared(conn, "select_one") != 1)
1249 : 0 : pg_fatal("failed to send describePrepared: %s", PQerrorMessage(conn));
1250 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1251 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1252 : :
1253 : 1 : consume_result_status(conn, PGRES_COMMAND_OK);
1254 : :
1255 : 1 : consume_null_result(conn);
1256 : :
1257 : 1 : res = confirm_result_status(conn, PGRES_COMMAND_OK);
1258 [ - + ]: 1 : if (PQnfields(res) != lengthof(expected_oids))
1259 : 0 : pg_fatal("expected %zu columns, got %d",
1260 : : lengthof(expected_oids), PQnfields(res));
1261 [ + + ]: 5 : for (int i = 0; i < PQnfields(res); i++)
1262 : : {
1263 : 4 : typ = PQftype(res, i);
1264 [ - + ]: 4 : if (typ != expected_oids[i])
1265 : 0 : pg_fatal("field %d: expected type %u, got %u",
1266 : : i, expected_oids[i], typ);
1267 : : }
1268 : 1 : PQclear(res);
1269 : :
1270 : 1 : consume_null_result(conn);
1271 : :
1272 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1273 : :
1274 : 1 : fprintf(stderr, "closing statement..");
1275 [ - + ]: 1 : if (PQsendClosePrepared(conn, "select_one") != 1)
1276 : 0 : pg_fatal("PQsendClosePrepared failed: %s", PQerrorMessage(conn));
1277 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1278 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1279 : :
1280 : 1 : consume_result_status(conn, PGRES_COMMAND_OK);
1281 : :
1282 : 1 : consume_null_result(conn);
1283 : :
1284 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1285 : :
1286 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1287 : 0 : pg_fatal("could not exit pipeline mode: %s", PQerrorMessage(conn));
1288 : :
1289 : : /* Now that it's closed we should get an error when describing */
1290 : 1 : res = PQdescribePrepared(conn, "select_one");
1291 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
1292 : 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
1293 : 1 : PQclear(res);
1294 : :
1295 : : /*
1296 : : * Also test the blocking close, this should not fail since closing a
1297 : : * non-existent prepared statement is a no-op
1298 : : */
1299 : 1 : res = PQclosePrepared(conn, "select_one");
1300 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1301 : 0 : pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
1302 : 1 : PQclear(res);
1303 : :
1304 : 1 : fprintf(stderr, "creating portal... ");
1305 : :
1306 : 1 : res = PQexec(conn, "BEGIN");
1307 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1308 : 0 : pg_fatal("BEGIN failed: %s", PQerrorMessage(conn));
1309 : 1 : PQclear(res);
1310 : :
1311 : 1 : res = PQexec(conn, "DECLARE cursor_one CURSOR FOR SELECT 1");
1312 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1313 : 0 : pg_fatal("DECLARE CURSOR failed: %s", PQerrorMessage(conn));
1314 : 1 : PQclear(res);
1315 : :
1316 : 1 : PQenterPipelineMode(conn);
1317 [ - + ]: 1 : if (PQsendDescribePortal(conn, "cursor_one") != 1)
1318 : 0 : pg_fatal("PQsendDescribePortal failed: %s", PQerrorMessage(conn));
1319 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1320 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1321 : :
1322 : 1 : res = confirm_result_status(conn, PGRES_COMMAND_OK);
1323 : 1 : typ = PQftype(res, 0);
1324 [ - + ]: 1 : if (typ != INT4OID)
1325 : 0 : pg_fatal("portal: expected type %u, got %u",
1326 : : INT4OID, typ);
1327 : 1 : PQclear(res);
1328 : :
1329 : 1 : consume_null_result(conn);
1330 : :
1331 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1332 : :
1333 : 1 : fprintf(stderr, "closing portal... ");
1334 [ - + ]: 1 : if (PQsendClosePortal(conn, "cursor_one") != 1)
1335 : 0 : pg_fatal("PQsendClosePortal failed: %s", PQerrorMessage(conn));
1336 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1337 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1338 : :
1339 : 1 : consume_result_status(conn, PGRES_COMMAND_OK);
1340 : :
1341 : 1 : consume_null_result(conn);
1342 : :
1343 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1344 : :
1345 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1346 : 0 : pg_fatal("could not exit pipeline mode: %s", PQerrorMessage(conn));
1347 : :
1348 : : /* Now that it's closed we should get an error when describing */
1349 : 1 : res = PQdescribePortal(conn, "cursor_one");
1350 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_FATAL_ERROR)
1351 : 0 : pg_fatal("expected FATAL_ERROR, got %s", PQresStatus(PQresultStatus(res)));
1352 : 1 : PQclear(res);
1353 : :
1354 : : /*
1355 : : * Also test the blocking close, this should not fail since closing a
1356 : : * non-existent portal is a no-op
1357 : : */
1358 : 1 : res = PQclosePortal(conn, "cursor_one");
1359 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1360 : 0 : pg_fatal("expected COMMAND_OK, got %s", PQresStatus(PQresultStatus(res)));
1361 : 1 : PQclear(res);
1362 : :
1363 : 1 : fprintf(stderr, "ok\n");
1364 : 1 : }
1365 : :
1366 : : /*
1367 : : * Test max_protocol_version options.
1368 : : */
1369 : : static void
1370 : 1 : test_protocol_version(PGconn *conn)
1371 : : {
1372 : : const char **keywords;
1373 : : const char **vals;
1374 : : int nopts;
1375 : 1 : PQconninfoOption *opts = PQconninfo(conn);
1376 : : int protocol_version;
1377 : 1 : int max_protocol_version_index = -1;
1378 : : int i;
1379 : :
1380 : : /* Prepare keywords/vals arrays, copied from the existing connection. */
1381 : 1 : nopts = 0;
1382 [ + + ]: 52 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1383 : 51 : nopts++;
1384 : 1 : nopts++; /* NULL terminator */
1385 : :
1386 : 1 : keywords = pg_malloc0_array(const char *, nopts);
1387 : 1 : vals = pg_malloc0_array(const char *, nopts);
1388 : :
1389 : 1 : i = 0;
1390 [ + + ]: 52 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
1391 : : {
1392 : : /*
1393 : : * If the test already specified max_protocol_version, we want to
1394 : : * replace it rather than attempting to override it. This matters when
1395 : : * testing defaults, because empty option values at the end of the
1396 : : * connection string won't replace earlier settings.
1397 : : */
1398 [ + + ]: 51 : if (strcmp(opt->keyword, "max_protocol_version") == 0)
1399 : 1 : max_protocol_version_index = i;
1400 [ + + ]: 50 : else if (!opt->val)
1401 : 31 : continue;
1402 : :
1403 : 20 : keywords[i] = opt->keyword;
1404 : 20 : vals[i] = opt->val;
1405 : :
1406 : 20 : i++;
1407 : : }
1408 : :
1409 : : Assert(max_protocol_version_index >= 0);
1410 : :
1411 : : /*
1412 : : * Test default protocol_version
1413 : : */
1414 : 1 : vals[max_protocol_version_index] = "";
1415 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1416 : :
1417 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1418 : 0 : pg_fatal("Connection to database failed: %s",
1419 : : PQerrorMessage(conn));
1420 : :
1421 : 1 : protocol_version = PQfullProtocolVersion(conn);
1422 [ - + ]: 1 : if (protocol_version != 30000)
1423 : 0 : pg_fatal("expected 30000, got %d", protocol_version);
1424 : :
1425 : 1 : PQfinish(conn);
1426 : :
1427 : : /*
1428 : : * Test max_protocol_version=3.0
1429 : : */
1430 : 1 : vals[max_protocol_version_index] = "3.0";
1431 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1432 : :
1433 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1434 : 0 : pg_fatal("Connection to database failed: %s",
1435 : : PQerrorMessage(conn));
1436 : :
1437 : 1 : protocol_version = PQfullProtocolVersion(conn);
1438 [ - + ]: 1 : if (protocol_version != 30000)
1439 : 0 : pg_fatal("expected 30000, got %d", protocol_version);
1440 : :
1441 : 1 : PQfinish(conn);
1442 : :
1443 : : /*
1444 : : * Test max_protocol_version=3.1. It's not valid, we went straight from
1445 : : * 3.0 to 3.2.
1446 : : */
1447 : 1 : vals[max_protocol_version_index] = "3.1";
1448 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1449 : :
1450 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_BAD)
1451 : 0 : pg_fatal("Connecting with max_protocol_version 3.1 should have failed.");
1452 : :
1453 : 1 : PQfinish(conn);
1454 : :
1455 : : /*
1456 : : * Test max_protocol_version=3.2
1457 : : */
1458 : 1 : vals[max_protocol_version_index] = "3.2";
1459 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1460 : :
1461 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1462 : 0 : pg_fatal("Connection to database failed: %s",
1463 : : PQerrorMessage(conn));
1464 : :
1465 : 1 : protocol_version = PQfullProtocolVersion(conn);
1466 [ - + ]: 1 : if (protocol_version != 30002)
1467 : 0 : pg_fatal("expected 30002, got %d", protocol_version);
1468 : :
1469 : 1 : PQfinish(conn);
1470 : :
1471 : : /*
1472 : : * Test max_protocol_version=latest. 'latest' currently means '3.2'.
1473 : : */
1474 : 1 : vals[max_protocol_version_index] = "latest";
1475 : 1 : conn = PQconnectdbParams(keywords, vals, false);
1476 : :
1477 [ - + ]: 1 : if (PQstatus(conn) != CONNECTION_OK)
1478 : 0 : pg_fatal("Connection to database failed: %s",
1479 : : PQerrorMessage(conn));
1480 : :
1481 : 1 : protocol_version = PQfullProtocolVersion(conn);
1482 [ - + ]: 1 : if (protocol_version != 30002)
1483 : 0 : pg_fatal("expected 30002, got %d", protocol_version);
1484 : :
1485 : 1 : PQfinish(conn);
1486 : :
1487 : 1 : pg_free(keywords);
1488 : 1 : pg_free(vals);
1489 : 1 : PQconninfoFree(opts);
1490 : 1 : }
1491 : :
1492 : : /* Notice processor: print notices, and count how many we got */
1493 : : static void
1494 : 1 : notice_processor(void *arg, const char *message)
1495 : : {
1496 : 1 : int *n_notices = (int *) arg;
1497 : :
1498 : 1 : (*n_notices)++;
1499 : 1 : fprintf(stderr, "NOTICE %d: %s", *n_notices, message);
1500 : 1 : }
1501 : :
1502 : : /* Verify behavior in "idle" state */
1503 : : static void
1504 : 1 : test_pipeline_idle(PGconn *conn)
1505 : : {
1506 : 1 : int n_notices = 0;
1507 : :
1508 : 1 : fprintf(stderr, "\npipeline idle...\n");
1509 : :
1510 : 1 : PQsetNoticeProcessor(conn, notice_processor, &n_notices);
1511 : :
1512 : : /* Try to exit pipeline mode in pipeline-idle state */
1513 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1514 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1515 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1", 0, NULL, NULL, NULL, NULL, 0) != 1)
1516 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1517 : 1 : PQsendFlushRequest(conn);
1518 : :
1519 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1520 : :
1521 : 1 : consume_null_result(conn);
1522 : :
1523 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 2", 0, NULL, NULL, NULL, NULL, 0) != 1)
1524 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1525 [ - + ]: 1 : if (PQexitPipelineMode(conn) == 1)
1526 : 0 : pg_fatal("exiting pipeline succeeded when it shouldn't");
1527 [ - + ]: 1 : if (strncmp(PQerrorMessage(conn), "cannot exit pipeline mode",
1528 : : strlen("cannot exit pipeline mode")) != 0)
1529 : 0 : pg_fatal("did not get expected error; got: %s",
1530 : : PQerrorMessage(conn));
1531 : 1 : PQsendFlushRequest(conn);
1532 : :
1533 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1534 : :
1535 : 1 : consume_null_result(conn);
1536 : :
1537 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1538 : 0 : pg_fatal("exiting pipeline failed: %s", PQerrorMessage(conn));
1539 : :
1540 [ - + ]: 1 : if (n_notices > 0)
1541 : 0 : pg_fatal("got %d notice(s)", n_notices);
1542 : 1 : fprintf(stderr, "ok - 1\n");
1543 : :
1544 : : /* Have a WARNING in the middle of a resultset */
1545 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1546 : 0 : pg_fatal("entering pipeline mode failed: %s", PQerrorMessage(conn));
1547 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT pg_catalog.pg_advisory_unlock(1,1)", 0, NULL, NULL, NULL, NULL, 0) != 1)
1548 : 0 : pg_fatal("failed to send query: %s", PQerrorMessage(conn));
1549 : 1 : PQsendFlushRequest(conn);
1550 : :
1551 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1552 : :
1553 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1554 : 0 : pg_fatal("failed to exit pipeline mode: %s", PQerrorMessage(conn));
1555 : 1 : fprintf(stderr, "ok - 2\n");
1556 : 1 : }
1557 : :
1558 : : static void
1559 : 1 : test_simple_pipeline(PGconn *conn)
1560 : : {
1561 : 1 : const char *dummy_params[1] = {"1"};
1562 : 1 : Oid dummy_param_oids[1] = {INT4OID};
1563 : :
1564 : 1 : fprintf(stderr, "simple pipeline... ");
1565 : :
1566 : : /*
1567 : : * Enter pipeline mode and dispatch a set of operations, which we'll then
1568 : : * process the results of as they come in.
1569 : : *
1570 : : * For a simple case we should be able to do this without interim
1571 : : * processing of results since our output buffer will give us enough slush
1572 : : * to work with and we won't block on sending. So blocking mode is fine.
1573 : : */
1574 [ - + ]: 1 : if (PQisnonblocking(conn))
1575 : 0 : pg_fatal("Expected blocking connection mode");
1576 : :
1577 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1578 : 0 : pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn));
1579 : :
1580 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT $1",
1581 : : 1, dummy_param_oids, dummy_params,
1582 : : NULL, NULL, 0) != 1)
1583 : 0 : pg_fatal("dispatching SELECT failed: %s", PQerrorMessage(conn));
1584 : :
1585 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
1586 : 0 : pg_fatal("exiting pipeline mode with work in progress should fail, but succeeded");
1587 : :
1588 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1589 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1590 : :
1591 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1592 : :
1593 : 1 : consume_null_result(conn);
1594 : :
1595 : : /*
1596 : : * Even though we've processed the result there's still a sync to come and
1597 : : * we can't exit pipeline mode yet
1598 : : */
1599 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 0)
1600 : 0 : pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
1601 : :
1602 : 1 : consume_result_status(conn, PGRES_PIPELINE_SYNC);
1603 : :
1604 : 1 : consume_null_result(conn);
1605 : :
1606 : : /* We're still in pipeline mode... */
1607 [ - + ]: 1 : if (PQpipelineStatus(conn) == PQ_PIPELINE_OFF)
1608 : 0 : pg_fatal("Fell out of pipeline mode somehow");
1609 : :
1610 : : /* ... until we end it, which we can safely do now */
1611 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1612 : 0 : pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s",
1613 : : PQerrorMessage(conn));
1614 : :
1615 [ - + ]: 1 : if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF)
1616 : 0 : pg_fatal("Exiting pipeline mode didn't seem to work");
1617 : :
1618 : 1 : fprintf(stderr, "ok\n");
1619 : 1 : }
1620 : :
1621 : : static void
1622 : 1 : test_singlerowmode(PGconn *conn)
1623 : : {
1624 : : PGresult *res;
1625 : : int i;
1626 : 1 : bool pipeline_ended = false;
1627 : :
1628 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1629 : 0 : pg_fatal("failed to enter pipeline mode: %s",
1630 : : PQerrorMessage(conn));
1631 : :
1632 : : /* One series of three commands, using single-row mode for the first two. */
1633 [ + + ]: 4 : for (i = 0; i < 3; i++)
1634 : : {
1635 : : char *param[1];
1636 : :
1637 : 3 : param[0] = psprintf("%d", 44 + i);
1638 : :
1639 [ - + ]: 3 : if (PQsendQueryParams(conn,
1640 : : "SELECT generate_series(42, $1)",
1641 : : 1,
1642 : : NULL,
1643 : : (const char *const *) param,
1644 : : NULL,
1645 : : NULL,
1646 : : 0) != 1)
1647 : 0 : pg_fatal("failed to send query: %s",
1648 : : PQerrorMessage(conn));
1649 : 3 : pfree(param[0]);
1650 : : }
1651 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1652 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1653 : :
1654 [ + + ]: 5 : for (i = 0; !pipeline_ended; i++)
1655 : : {
1656 : 4 : bool first = true;
1657 : : bool saw_ending_tuplesok;
1658 : 4 : bool isSingleTuple = false;
1659 : :
1660 : : /* Set single row mode for only first 2 SELECT queries */
1661 [ + + ]: 4 : if (i < 2)
1662 : : {
1663 [ - + ]: 2 : if (PQsetSingleRowMode(conn) != 1)
1664 : 0 : pg_fatal("PQsetSingleRowMode() failed for i=%d", i);
1665 : : }
1666 : :
1667 : : /* Consume rows for this query */
1668 : 4 : saw_ending_tuplesok = false;
1669 [ + + ]: 14 : while ((res = PQgetResult(conn)) != NULL)
1670 : : {
1671 : 11 : ExecStatusType est = PQresultStatus(res);
1672 : :
1673 [ + + ]: 11 : if (est == PGRES_PIPELINE_SYNC)
1674 : : {
1675 : 1 : fprintf(stderr, "end of pipeline reached\n");
1676 : 1 : pipeline_ended = true;
1677 : 1 : PQclear(res);
1678 [ - + ]: 1 : if (i != 3)
1679 : 0 : pg_fatal("Expected three results, got %d", i);
1680 : 1 : break;
1681 : : }
1682 : :
1683 : : /* Expect SINGLE_TUPLE for queries 0 and 1, TUPLES_OK for 2 */
1684 [ + + ]: 10 : if (first)
1685 : : {
1686 [ + + - + ]: 3 : if (i <= 1 && est != PGRES_SINGLE_TUPLE)
1687 : 0 : pg_fatal("Expected PGRES_SINGLE_TUPLE for query %d, got %s",
1688 : : i, PQresStatus(est));
1689 [ + + - + ]: 3 : if (i >= 2 && est != PGRES_TUPLES_OK)
1690 : 0 : pg_fatal("Expected PGRES_TUPLES_OK for query %d, got %s",
1691 : : i, PQresStatus(est));
1692 : 3 : first = false;
1693 : : }
1694 : :
1695 : 10 : fprintf(stderr, "Result status %s for query %d", PQresStatus(est), i);
1696 [ + + - ]: 10 : switch (est)
1697 : : {
1698 : 3 : case PGRES_TUPLES_OK:
1699 : 3 : fprintf(stderr, ", tuples: %d\n", PQntuples(res));
1700 : 3 : saw_ending_tuplesok = true;
1701 [ + + ]: 3 : if (isSingleTuple)
1702 : : {
1703 [ + - ]: 2 : if (PQntuples(res) == 0)
1704 : 2 : fprintf(stderr, "all tuples received in query %d\n", i);
1705 : : else
1706 : 0 : pg_fatal("Expected to follow PGRES_SINGLE_TUPLE, but received PGRES_TUPLES_OK directly instead");
1707 : : }
1708 : 3 : break;
1709 : :
1710 : 7 : case PGRES_SINGLE_TUPLE:
1711 : 7 : isSingleTuple = true;
1712 : 7 : fprintf(stderr, ", %d tuple: %s\n", PQntuples(res), PQgetvalue(res, 0, 0));
1713 : 7 : break;
1714 : :
1715 : 0 : default:
1716 : 0 : pg_fatal("unexpected");
1717 : : }
1718 : 10 : PQclear(res);
1719 : : }
1720 [ + + - + ]: 4 : if (!pipeline_ended && !saw_ending_tuplesok)
1721 : 0 : pg_fatal("didn't get expected terminating TUPLES_OK");
1722 : : }
1723 : :
1724 : : /*
1725 : : * Now issue one command, get its results in with single-row mode, then
1726 : : * issue another command, and get its results in normal mode; make sure
1727 : : * the single-row mode flag is reset as expected.
1728 : : */
1729 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT generate_series(0, 0)",
1730 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1731 : 0 : pg_fatal("failed to send query: %s",
1732 : : PQerrorMessage(conn));
1733 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1734 : 0 : pg_fatal("failed to send flush request");
1735 [ - + ]: 1 : if (PQsetSingleRowMode(conn) != 1)
1736 : 0 : pg_fatal("PQsetSingleRowMode() failed");
1737 : :
1738 : 1 : consume_result_status(conn, PGRES_SINGLE_TUPLE);
1739 : :
1740 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1741 : :
1742 : 1 : consume_null_result(conn);
1743 : :
1744 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT 1",
1745 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1746 : 0 : pg_fatal("failed to send query: %s",
1747 : : PQerrorMessage(conn));
1748 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1749 : 0 : pg_fatal("failed to send flush request");
1750 : :
1751 : 1 : consume_result_status(conn, PGRES_TUPLES_OK);
1752 : :
1753 : 1 : consume_null_result(conn);
1754 : :
1755 : : /*
1756 : : * Try chunked mode as well; make sure that it correctly delivers a
1757 : : * partial final chunk.
1758 : : */
1759 [ - + ]: 1 : if (PQsendQueryParams(conn, "SELECT generate_series(1, 5)",
1760 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1761 : 0 : pg_fatal("failed to send query: %s",
1762 : : PQerrorMessage(conn));
1763 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
1764 : 0 : pg_fatal("failed to send flush request");
1765 [ - + ]: 1 : if (PQsetChunkedRowsMode(conn, 3) != 1)
1766 : 0 : pg_fatal("PQsetChunkedRowsMode() failed");
1767 : :
1768 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_CHUNK);
1769 [ - + ]: 1 : if (PQntuples(res) != 3)
1770 : 0 : pg_fatal("Expected 3 rows, got %d", PQntuples(res));
1771 : 1 : PQclear(res);
1772 : :
1773 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_CHUNK);
1774 [ - + ]: 1 : if (PQntuples(res) != 2)
1775 : 0 : pg_fatal("Expected 2 rows, got %d", PQntuples(res));
1776 : 1 : PQclear(res);
1777 : :
1778 : 1 : res = confirm_result_status(conn, PGRES_TUPLES_OK);
1779 [ - + ]: 1 : if (PQntuples(res) != 0)
1780 : 0 : pg_fatal("Expected 0 rows, got %d", PQntuples(res));
1781 : 1 : PQclear(res);
1782 : :
1783 : 1 : consume_null_result(conn);
1784 : :
1785 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1786 : 0 : pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
1787 : :
1788 : 1 : fprintf(stderr, "ok\n");
1789 : 1 : }
1790 : :
1791 : : /*
1792 : : * Simple test to verify that a pipeline is discarded as a whole when there's
1793 : : * an error, ignoring transaction commands.
1794 : : */
1795 : : static void
1796 : 1 : test_transaction(PGconn *conn)
1797 : : {
1798 : : PGresult *res;
1799 : : bool expect_null;
1800 : 1 : int num_syncs = 0;
1801 : :
1802 : 1 : res = PQexec(conn, "DROP TABLE IF EXISTS pq_pipeline_tst;"
1803 : : "CREATE TABLE pq_pipeline_tst (id int)");
1804 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1805 : 0 : pg_fatal("failed to create test table: %s",
1806 : : PQerrorMessage(conn));
1807 : 1 : PQclear(res);
1808 : :
1809 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1810 : 0 : pg_fatal("failed to enter pipeline mode: %s",
1811 : : PQerrorMessage(conn));
1812 [ - + ]: 1 : if (PQsendPrepare(conn, "rollback", "ROLLBACK", 0, NULL) != 1)
1813 : 0 : pg_fatal("could not send prepare on pipeline: %s",
1814 : : PQerrorMessage(conn));
1815 : :
1816 [ - + ]: 1 : if (PQsendQueryParams(conn,
1817 : : "BEGIN",
1818 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1819 : 0 : pg_fatal("failed to send query: %s",
1820 : : PQerrorMessage(conn));
1821 [ - + ]: 1 : if (PQsendQueryParams(conn,
1822 : : "SELECT 0/0",
1823 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1824 : 0 : pg_fatal("failed to send query: %s",
1825 : : PQerrorMessage(conn));
1826 : :
1827 : : /*
1828 : : * send a ROLLBACK using a prepared stmt. Doesn't work because we need to
1829 : : * get out of the pipeline-aborted state first.
1830 : : */
1831 [ - + ]: 1 : if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
1832 : 0 : pg_fatal("failed to execute prepared: %s",
1833 : : PQerrorMessage(conn));
1834 : :
1835 : : /* This insert fails because we're in pipeline-aborted state */
1836 [ - + ]: 1 : if (PQsendQueryParams(conn,
1837 : : "INSERT INTO pq_pipeline_tst VALUES (1)",
1838 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1839 : 0 : pg_fatal("failed to send query: %s",
1840 : : PQerrorMessage(conn));
1841 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1842 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1843 : 1 : num_syncs++;
1844 : :
1845 : : /*
1846 : : * This insert fails even though the pipeline got a SYNC, because we're in
1847 : : * an aborted transaction
1848 : : */
1849 [ - + ]: 1 : if (PQsendQueryParams(conn,
1850 : : "INSERT INTO pq_pipeline_tst VALUES (2)",
1851 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1852 : 0 : pg_fatal("failed to send query: %s",
1853 : : PQerrorMessage(conn));
1854 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1855 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1856 : 1 : num_syncs++;
1857 : :
1858 : : /*
1859 : : * Send ROLLBACK using prepared stmt. This one works because we just did
1860 : : * PQpipelineSync above.
1861 : : */
1862 [ - + ]: 1 : if (PQsendQueryPrepared(conn, "rollback", 0, NULL, NULL, NULL, 1) != 1)
1863 : 0 : pg_fatal("failed to execute prepared: %s",
1864 : : PQerrorMessage(conn));
1865 : :
1866 : : /*
1867 : : * Now that we're out of a transaction and in pipeline-good mode, this
1868 : : * insert works
1869 : : */
1870 [ - + ]: 1 : if (PQsendQueryParams(conn,
1871 : : "INSERT INTO pq_pipeline_tst VALUES (3)",
1872 : : 0, NULL, NULL, NULL, NULL, 0) != 1)
1873 : 0 : pg_fatal("failed to send query: %s",
1874 : : PQerrorMessage(conn));
1875 : : /* Send two syncs now -- match up to SYNC messages below */
1876 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1877 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1878 : 1 : num_syncs++;
1879 [ - + ]: 1 : if (PQpipelineSync(conn) != 1)
1880 : 0 : pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn));
1881 : 1 : num_syncs++;
1882 : :
1883 : 1 : expect_null = false;
1884 : 1 : for (int i = 0;; i++)
1885 : 19 : {
1886 : : ExecStatusType restype;
1887 : :
1888 : 20 : res = PQgetResult(conn);
1889 [ + + ]: 20 : if (res == NULL)
1890 : : {
1891 : 8 : printf("%d: got NULL result\n", i);
1892 [ - + ]: 8 : if (!expect_null)
1893 : 0 : pg_fatal("did not expect NULL here");
1894 : 8 : expect_null = false;
1895 : 8 : continue;
1896 : : }
1897 : 12 : restype = PQresultStatus(res);
1898 : 12 : printf("%d: got status %s", i, PQresStatus(restype));
1899 [ - + ]: 12 : if (expect_null)
1900 : 0 : pg_fatal("expected NULL");
1901 [ + + ]: 12 : if (restype == PGRES_FATAL_ERROR)
1902 : 2 : printf("; error: %s", PQerrorMessage(conn));
1903 [ + + ]: 10 : else if (restype == PGRES_PIPELINE_ABORTED)
1904 : : {
1905 : 2 : printf(": command didn't run because pipeline aborted\n");
1906 : : }
1907 : : else
1908 : 8 : printf("\n");
1909 : 12 : PQclear(res);
1910 : :
1911 [ + + ]: 12 : if (restype == PGRES_PIPELINE_SYNC)
1912 : 4 : num_syncs--;
1913 : : else
1914 : 8 : expect_null = true;
1915 [ + + ]: 12 : if (num_syncs <= 0)
1916 : 1 : break;
1917 : : }
1918 : :
1919 : 1 : consume_null_result(conn);
1920 : :
1921 [ - + ]: 1 : if (PQexitPipelineMode(conn) != 1)
1922 : 0 : pg_fatal("failed to end pipeline mode: %s", PQerrorMessage(conn));
1923 : :
1924 : : /* We expect to find one tuple containing the value "3" */
1925 : 1 : res = PQexec(conn, "SELECT * FROM pq_pipeline_tst");
1926 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
1927 : 0 : pg_fatal("failed to obtain result: %s", PQerrorMessage(conn));
1928 [ - + ]: 1 : if (PQntuples(res) != 1)
1929 : 0 : pg_fatal("did not get 1 tuple");
1930 [ - + ]: 1 : if (strcmp(PQgetvalue(res, 0, 0), "3") != 0)
1931 : 0 : pg_fatal("did not get expected tuple");
1932 : 1 : PQclear(res);
1933 : :
1934 : 1 : fprintf(stderr, "ok\n");
1935 : 1 : }
1936 : :
1937 : : /*
1938 : : * In this test mode we send a stream of queries, with one in the middle
1939 : : * causing an error. Verify that we can still send some more after the
1940 : : * error and have libpq work properly.
1941 : : */
1942 : : static void
1943 : 1 : test_uniqviol(PGconn *conn)
1944 : : {
1945 : 1 : int sock = PQsocket(conn);
1946 : : PGresult *res;
1947 : 1 : Oid paramTypes[2] = {INT8OID, INT8OID};
1948 : : const char *paramValues[2];
1949 : : char paramValue0[MAXINT8LEN];
1950 : : char paramValue1[MAXINT8LEN];
1951 : 1 : int ctr = 0;
1952 : 1 : int numsent = 0;
1953 : 1 : int results = 0;
1954 : 1 : bool read_done = false;
1955 : 1 : bool write_done = false;
1956 : 1 : bool error_sent = false;
1957 : 1 : bool got_error = false;
1958 : 1 : int switched = 0;
1959 : 1 : int socketful = 0;
1960 : : fd_set in_fds;
1961 : : fd_set out_fds;
1962 : :
1963 : 1 : fprintf(stderr, "uniqviol ...");
1964 : :
1965 : 1 : PQsetnonblocking(conn, 1);
1966 : :
1967 : 1 : paramValues[0] = paramValue0;
1968 : 1 : paramValues[1] = paramValue1;
1969 : 1 : sprintf(paramValue1, "42");
1970 : :
1971 : 1 : res = PQexec(conn, "drop table if exists ppln_uniqviol;"
1972 : : "create table ppln_uniqviol(id bigint primary key, idata bigint)");
1973 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1974 : 0 : pg_fatal("failed to create table: %s", PQerrorMessage(conn));
1975 : 1 : PQclear(res);
1976 : :
1977 : 1 : res = PQexec(conn, "begin");
1978 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1979 : 0 : pg_fatal("failed to begin transaction: %s", PQerrorMessage(conn));
1980 : 1 : PQclear(res);
1981 : :
1982 : 1 : res = PQprepare(conn, "insertion",
1983 : : "insert into ppln_uniqviol values ($1, $2) returning id",
1984 : : 2, paramTypes);
1985 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1986 : 0 : pg_fatal("failed to prepare query: %s", PQerrorMessage(conn));
1987 : 1 : PQclear(res);
1988 : :
1989 [ - + ]: 1 : if (PQenterPipelineMode(conn) != 1)
1990 : 0 : pg_fatal("failed to enter pipeline mode");
1991 : :
1992 [ + - ]: 7 : while (!read_done)
1993 : : {
1994 : : /*
1995 : : * Avoid deadlocks by reading everything the server has sent before
1996 : : * sending anything. (Special precaution is needed here to process
1997 : : * PQisBusy before testing the socket for read-readiness, because the
1998 : : * socket does not turn read-ready after "sending" queries in aborted
1999 : : * pipeline mode.)
2000 : : */
2001 [ + + ]: 606 : while (PQisBusy(conn) == 0)
2002 : : {
2003 : : bool new_error;
2004 : :
2005 [ + + ]: 601 : if (results >= numsent)
2006 : : {
2007 [ - + ]: 1 : if (write_done)
2008 : 0 : read_done = true;
2009 : 1 : break;
2010 : : }
2011 : :
2012 : 600 : res = PQgetResult(conn);
2013 : 600 : new_error = process_result(conn, res, results, numsent);
2014 [ + + - + ]: 600 : if (new_error && got_error)
2015 : 0 : pg_fatal("got two errors");
2016 : 600 : got_error |= new_error;
2017 [ + + ]: 600 : if (results++ >= numsent - 1)
2018 : : {
2019 [ + - ]: 1 : if (write_done)
2020 : 1 : read_done = true;
2021 : 1 : break;
2022 : : }
2023 : : }
2024 : :
2025 [ + + ]: 7 : if (read_done)
2026 : 1 : break;
2027 : :
2028 [ + + ]: 102 : FD_ZERO(&out_fds);
2029 : 6 : FD_SET(sock, &out_fds);
2030 : :
2031 [ + + ]: 102 : FD_ZERO(&in_fds);
2032 : 6 : FD_SET(sock, &in_fds);
2033 : :
2034 [ - + - + ]: 6 : if (select(sock + 1, &in_fds, write_done ? NULL : &out_fds, NULL, NULL) == -1)
2035 : : {
2036 [ # # ]: 0 : if (errno == EINTR)
2037 : 0 : continue;
2038 : 0 : pg_fatal("select() failed: %m");
2039 : : }
2040 : :
2041 [ + + - + ]: 6 : if (FD_ISSET(sock, &in_fds) && PQconsumeInput(conn) == 0)
2042 : 0 : pg_fatal("PQconsumeInput failed: %s", PQerrorMessage(conn));
2043 : :
2044 : : /*
2045 : : * If the socket is writable and we haven't finished sending queries,
2046 : : * send some.
2047 : : */
2048 [ + - + + ]: 6 : if (!write_done && FD_ISSET(sock, &out_fds))
2049 : : {
2050 : : for (;;)
2051 : 597 : {
2052 : : int flush;
2053 : :
2054 : : /*
2055 : : * provoke uniqueness violation exactly once after having
2056 : : * switched to read mode.
2057 : : */
2058 [ + + + + : 600 : if (switched >= 1 && !error_sent && ctr % socketful >= socketful / 2)
+ + ]
2059 : : {
2060 : 1 : sprintf(paramValue0, "%d", numsent / 2);
2061 : 1 : fprintf(stderr, "E");
2062 : 1 : error_sent = true;
2063 : : }
2064 : : else
2065 : : {
2066 : 599 : fprintf(stderr, ".");
2067 : 599 : sprintf(paramValue0, "%d", ctr++);
2068 : : }
2069 : :
2070 [ - + ]: 600 : if (PQsendQueryPrepared(conn, "insertion", 2, paramValues, NULL, NULL, 0) != 1)
2071 : 0 : pg_fatal("failed to execute prepared query: %s", PQerrorMessage(conn));
2072 : 600 : numsent++;
2073 : :
2074 : : /* Are we done writing? */
2075 [ + + + + : 600 : if (socketful != 0 && numsent % socketful == 42 && error_sent)
+ + ]
2076 : : {
2077 [ - + ]: 1 : if (PQsendFlushRequest(conn) != 1)
2078 : 0 : pg_fatal("failed to send flush request");
2079 : 1 : write_done = true;
2080 : 1 : fprintf(stderr, "\ndone writing\n");
2081 : 1 : PQflush(conn);
2082 : 1 : break;
2083 : : }
2084 : :
2085 : : /* is the outgoing socket full? */
2086 : 599 : flush = PQflush(conn);
2087 [ - + ]: 599 : if (flush == -1)
2088 : 0 : pg_fatal("failed to flush: %s", PQerrorMessage(conn));
2089 [ + + ]: 599 : if (flush == 1)
2090 : : {
2091 [ + + ]: 2 : if (socketful == 0)
2092 : 1 : socketful = numsent;
2093 : 2 : fprintf(stderr, "\nswitch to reading\n");
2094 : 2 : switched++;
2095 : 2 : break;
2096 : : }
2097 : : }
2098 : : }
2099 : : }
2100 : :
2101 [ - + ]: 1 : if (!got_error)
2102 : 0 : pg_fatal("did not get expected error");
2103 : :
2104 : 1 : fprintf(stderr, "ok\n");
2105 : 1 : }
2106 : :
2107 : : /*
2108 : : * Subroutine for test_uniqviol; given a PGresult, print it out and consume
2109 : : * the expected NULL that should follow it.
2110 : : *
2111 : : * Returns true if we read a fatal error message, otherwise false.
2112 : : */
2113 : : static bool
2114 : 600 : process_result(PGconn *conn, PGresult *res, int results, int numsent)
2115 : : {
2116 : 600 : bool got_error = false;
2117 : :
2118 [ - + ]: 600 : if (res == NULL)
2119 : 0 : pg_fatal("got unexpected NULL");
2120 : :
2121 [ + + + - ]: 600 : switch (PQresultStatus(res))
2122 : : {
2123 : 1 : case PGRES_FATAL_ERROR:
2124 : 1 : got_error = true;
2125 : 1 : fprintf(stderr, "result %d/%d (error): %s\n", results, numsent, PQerrorMessage(conn));
2126 : 1 : PQclear(res);
2127 : 1 : consume_null_result(conn);
2128 : 1 : break;
2129 : :
2130 : 418 : case PGRES_TUPLES_OK:
2131 : 418 : fprintf(stderr, "result %d/%d: %s\n", results, numsent, PQgetvalue(res, 0, 0));
2132 : 418 : PQclear(res);
2133 : 418 : consume_null_result(conn);
2134 : 418 : break;
2135 : :
2136 : 181 : case PGRES_PIPELINE_ABORTED:
2137 : 181 : fprintf(stderr, "result %d/%d: pipeline aborted\n", results, numsent);
2138 : 181 : PQclear(res);
2139 : 181 : consume_null_result(conn);
2140 : 181 : break;
2141 : :
2142 : 0 : default:
2143 : 0 : pg_fatal("got unexpected %s", PQresStatus(PQresultStatus(res)));
2144 : : }
2145 : :
2146 : 600 : return got_error;
2147 : : }
2148 : :
2149 : :
2150 : : static void
2151 : 0 : usage(const char *progname)
2152 : : {
2153 : 0 : fprintf(stderr, "%s tests libpq's pipeline mode.\n\n", progname);
2154 : 0 : fprintf(stderr, "Usage:\n");
2155 : 0 : fprintf(stderr, " %s [OPTION] tests\n", progname);
2156 : 0 : fprintf(stderr, " %s [OPTION] TESTNAME [CONNINFO]\n", progname);
2157 : 0 : fprintf(stderr, "\nOptions:\n");
2158 : 0 : fprintf(stderr, " -t TRACEFILE generate a libpq trace to TRACEFILE\n");
2159 : 0 : fprintf(stderr, " -r NUMROWS use NUMROWS as the test size\n");
2160 : 0 : }
2161 : :
2162 : : static void
2163 : 1 : print_test_list(void)
2164 : : {
2165 : 1 : printf("cancel\n");
2166 : 1 : printf("describe_fetch\n");
2167 : 1 : printf("disallowed_in_pipeline\n");
2168 : 1 : printf("multi_pipelines\n");
2169 : 1 : printf("nosync\n");
2170 : 1 : printf("pipeline_abort\n");
2171 : 1 : printf("pipeline_idle\n");
2172 : 1 : printf("pipelined_insert\n");
2173 : 1 : printf("prepared\n");
2174 : 1 : printf("protocol_version\n");
2175 : 1 : printf("simple_pipeline\n");
2176 : 1 : printf("singlerow\n");
2177 : 1 : printf("transaction\n");
2178 : 1 : printf("uniqviol\n");
2179 : 1 : }
2180 : :
2181 : : int
2182 : 16 : main(int argc, char **argv)
2183 : : {
2184 : 16 : const char *conninfo = "";
2185 : : PGconn *conn;
2186 : 16 : FILE *trace = NULL;
2187 : : char *testname;
2188 : 16 : int numrows = 10000;
2189 : : PGresult *res;
2190 : : int c;
2191 : :
2192 [ + + ]: 55 : while ((c = getopt(argc, argv, "r:t:")) != -1)
2193 : : {
2194 [ + + - ]: 23 : switch (c)
2195 : : {
2196 : 14 : case 'r': /* numrows */
2197 : 14 : errno = 0;
2198 : 14 : numrows = strtol(optarg, NULL, 10);
2199 [ + - - + ]: 14 : if (errno != 0 || numrows <= 0)
2200 : : {
2201 : 0 : fprintf(stderr, "couldn't parse \"%s\" as a positive integer\n",
2202 : : optarg);
2203 : 0 : exit(1);
2204 : : }
2205 : 14 : break;
2206 : 9 : case 't': /* trace file */
2207 : 9 : tracefile = pg_strdup(optarg);
2208 : 9 : break;
2209 : : }
2210 : : }
2211 : :
2212 [ + - ]: 16 : if (optind < argc)
2213 : : {
2214 : 16 : testname = pg_strdup(argv[optind]);
2215 : 16 : optind++;
2216 : : }
2217 : : else
2218 : : {
2219 : 0 : usage(argv[0]);
2220 : 0 : exit(1);
2221 : : }
2222 : :
2223 [ + + ]: 16 : if (strcmp(testname, "tests") == 0)
2224 : : {
2225 : 1 : print_test_list();
2226 : 1 : exit(0);
2227 : : }
2228 : :
2229 [ + - ]: 15 : if (optind < argc)
2230 : : {
2231 : 15 : conninfo = pg_strdup(argv[optind]);
2232 : 15 : optind++;
2233 : : }
2234 : :
2235 : : /* Make a connection to the database */
2236 : 15 : conn = PQconnectdb(conninfo);
2237 [ - + ]: 15 : if (PQstatus(conn) != CONNECTION_OK)
2238 : : {
2239 : 0 : fprintf(stderr, "Connection to database failed: %s\n",
2240 : : PQerrorMessage(conn));
2241 : 0 : exit_nicely(conn);
2242 : : }
2243 : :
2244 : 15 : res = PQexec(conn, "SET lc_messages TO \"C\"");
2245 [ - + ]: 15 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
2246 : 0 : pg_fatal("failed to set \"lc_messages\": %s", PQerrorMessage(conn));
2247 : 15 : PQclear(res);
2248 : 15 : res = PQexec(conn, "SET debug_parallel_query = off");
2249 [ - + ]: 15 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
2250 : 0 : pg_fatal("failed to set \"debug_parallel_query\": %s", PQerrorMessage(conn));
2251 : 15 : PQclear(res);
2252 : :
2253 : : /* Set the trace file, if requested */
2254 [ + + ]: 15 : if (tracefile != NULL)
2255 : : {
2256 [ - + ]: 9 : if (strcmp(tracefile, "-") == 0)
2257 : 0 : trace = stdout;
2258 : : else
2259 : 9 : trace = fopen(tracefile, "w");
2260 [ - + ]: 9 : if (trace == NULL)
2261 : 0 : pg_fatal("could not open file \"%s\": %m", tracefile);
2262 : :
2263 : : /* Make it line-buffered */
2264 : 9 : setvbuf(trace, NULL, PG_IOLBF, 0);
2265 : :
2266 : 9 : PQtrace(conn, trace);
2267 : 9 : PQsetTraceFlags(conn,
2268 : : PQTRACE_SUPPRESS_TIMESTAMPS | PQTRACE_REGRESS_MODE);
2269 : : }
2270 : :
2271 [ + + ]: 15 : if (strcmp(testname, "cancel") == 0)
2272 : 2 : test_cancel(conn);
2273 [ + + ]: 13 : else if (strcmp(testname, "describe_fetch") == 0)
2274 : 1 : test_describe_fetch(conn);
2275 [ + + ]: 12 : else if (strcmp(testname, "disallowed_in_pipeline") == 0)
2276 : 1 : test_disallowed_in_pipeline(conn);
2277 [ + + ]: 11 : else if (strcmp(testname, "multi_pipelines") == 0)
2278 : 1 : test_multi_pipelines(conn);
2279 [ + + ]: 10 : else if (strcmp(testname, "nosync") == 0)
2280 : 1 : test_nosync(conn);
2281 [ + + ]: 9 : else if (strcmp(testname, "pipeline_abort") == 0)
2282 : 1 : test_pipeline_abort(conn);
2283 [ + + ]: 8 : else if (strcmp(testname, "pipeline_idle") == 0)
2284 : 1 : test_pipeline_idle(conn);
2285 [ + + ]: 7 : else if (strcmp(testname, "pipelined_insert") == 0)
2286 : 1 : test_pipelined_insert(conn, numrows);
2287 [ + + ]: 6 : else if (strcmp(testname, "prepared") == 0)
2288 : 1 : test_prepared(conn);
2289 [ + + ]: 5 : else if (strcmp(testname, "protocol_version") == 0)
2290 : 1 : test_protocol_version(conn);
2291 [ + + ]: 4 : else if (strcmp(testname, "simple_pipeline") == 0)
2292 : 1 : test_simple_pipeline(conn);
2293 [ + + ]: 3 : else if (strcmp(testname, "singlerow") == 0)
2294 : 1 : test_singlerowmode(conn);
2295 [ + + ]: 2 : else if (strcmp(testname, "transaction") == 0)
2296 : 1 : test_transaction(conn);
2297 [ + - ]: 1 : else if (strcmp(testname, "uniqviol") == 0)
2298 : 1 : test_uniqviol(conn);
2299 : : else
2300 : : {
2301 : 0 : fprintf(stderr, "\"%s\" is not a recognized test name\n", testname);
2302 : 0 : exit(1);
2303 : : }
2304 : :
2305 : : /* close the connection to the database and cleanup */
2306 : 15 : PQfinish(conn);
2307 : :
2308 [ + + + - ]: 15 : if (trace && trace != stdout)
2309 : 9 : fclose(trace);
2310 : :
2311 : 15 : return 0;
2312 : : }
|
