Line data Source code
1 : /*
2 : * pgbench.c
3 : *
4 : * A simple benchmark program for PostgreSQL
5 : * Originally written by Tatsuo Ishii and enhanced by many contributors.
6 : *
7 : * src/bin/pgbench/pgbench.c
8 : * Copyright (c) 2000-2017, PostgreSQL Global Development Group
9 : * ALL RIGHTS RESERVED;
10 : *
11 : * Permission to use, copy, modify, and distribute this software and its
12 : * documentation for any purpose, without fee, and without a written agreement
13 : * is hereby granted, provided that the above copyright notice and this
14 : * paragraph and the following two paragraphs appear in all copies.
15 : *
16 : * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
17 : * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
18 : * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
19 : * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF THE
20 : * POSSIBILITY OF SUCH DAMAGE.
21 : *
22 : * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
23 : * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
24 : * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
25 : * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS TO
26 : * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
27 : *
28 : */
29 :
30 : #ifdef WIN32
31 : #define FD_SETSIZE 1024 /* set before winsock2.h is included */
32 : #endif /* ! WIN32 */
33 :
34 : #include "postgres_fe.h"
35 :
36 : #include "getopt_long.h"
37 : #include "libpq-fe.h"
38 : #include "portability/instr_time.h"
39 :
40 : #include <ctype.h>
41 : #include <float.h>
42 : #include <limits.h>
43 : #include <math.h>
44 : #include <signal.h>
45 : #include <time.h>
46 : #include <sys/time.h>
47 : #ifdef HAVE_SYS_SELECT_H
48 : #include <sys/select.h>
49 : #endif
50 :
51 : #ifdef HAVE_SYS_RESOURCE_H
52 : #include <sys/resource.h> /* for getrlimit */
53 : #endif
54 :
55 : #ifndef M_PI
56 : #define M_PI 3.14159265358979323846
57 : #endif
58 :
59 : #include "pgbench.h"
60 :
61 : #define ERRCODE_UNDEFINED_TABLE "42P01"
62 :
63 : /*
64 : * Multi-platform pthread implementations
65 : */
66 :
67 : #ifdef WIN32
68 : /* Use native win32 threads on Windows */
69 : typedef struct win32_pthread *pthread_t;
70 : typedef int pthread_attr_t;
71 :
72 : static int pthread_create(pthread_t *thread, pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
73 : static int pthread_join(pthread_t th, void **thread_return);
74 : #elif defined(ENABLE_THREAD_SAFETY)
75 : /* Use platform-dependent pthread capability */
76 : #include <pthread.h>
77 : #else
78 : /* No threads implementation, use none (-j 1) */
79 : #define pthread_t void *
80 : #endif
81 :
82 :
83 : /********************************************************************
84 : * some configurable parameters */
85 :
86 : /* max number of clients allowed */
87 : #ifdef FD_SETSIZE
88 : #define MAXCLIENTS (FD_SETSIZE - 10)
89 : #else
90 : #define MAXCLIENTS 1024
91 : #endif
92 :
93 : #define LOG_STEP_SECONDS 5 /* seconds between log messages */
94 : #define DEFAULT_NXACTS 10 /* default nxacts */
95 :
96 : #define MIN_GAUSSIAN_PARAM 2.0 /* minimum parameter for gauss */
97 :
98 : int nxacts = 0; /* number of transactions per client */
99 : int duration = 0; /* duration in seconds */
100 : int64 end_time = 0; /* when to stop in micro seconds, under -T */
101 :
102 : /*
103 : * scaling factor. for example, scale = 10 will make 1000000 tuples in
104 : * pgbench_accounts table.
105 : */
106 : int scale = 1;
107 :
108 : /*
109 : * fillfactor. for example, fillfactor = 90 will use only 90 percent
110 : * space during inserts and leave 10 percent free.
111 : */
112 : int fillfactor = 100;
113 :
114 : /*
115 : * create foreign key constraints on the tables?
116 : */
117 : int foreign_keys = 0;
118 :
119 : /*
120 : * use unlogged tables?
121 : */
122 : int unlogged_tables = 0;
123 :
124 : /*
125 : * log sampling rate (1.0 = log everything, 0.0 = option not given)
126 : */
127 : double sample_rate = 0.0;
128 :
129 : /*
130 : * When threads are throttled to a given rate limit, this is the target delay
131 : * to reach that rate in usec. 0 is the default and means no throttling.
132 : */
133 : int64 throttle_delay = 0;
134 :
135 : /*
136 : * Transactions which take longer than this limit (in usec) are counted as
137 : * late, and reported as such, although they are completed anyway. When
138 : * throttling is enabled, execution time slots that are more than this late
139 : * are skipped altogether, and counted separately.
140 : */
141 : int64 latency_limit = 0;
142 :
143 : /*
144 : * tablespace selection
145 : */
146 : char *tablespace = NULL;
147 : char *index_tablespace = NULL;
148 :
149 : /*
150 : * end of configurable parameters
151 : *********************************************************************/
152 :
153 : #define nbranches 1 /* Makes little sense to change this. Change
154 : * -s instead */
155 : #define ntellers 10
156 : #define naccounts 100000
157 :
158 : /*
159 : * The scale factor at/beyond which 32bit integers are incapable of storing
160 : * 64bit values.
161 : *
162 : * Although the actual threshold is 21474, we use 20000 because it is easier to
163 : * document and remember, and isn't that far away from the real threshold.
164 : */
165 : #define SCALE_32BIT_THRESHOLD 20000
166 :
167 : bool use_log; /* log transaction latencies to a file */
168 : bool use_quiet; /* quiet logging onto stderr */
169 : int agg_interval; /* log aggregates instead of individual
170 : * transactions */
171 : bool per_script_stats = false; /* whether to collect stats per script */
172 : int progress = 0; /* thread progress report every this seconds */
173 : bool progress_timestamp = false; /* progress report with Unix time */
174 : int nclients = 1; /* number of clients */
175 : int nthreads = 1; /* number of threads */
176 : bool is_connect; /* establish connection for each transaction */
177 : bool is_latencies; /* report per-command latencies */
178 : int main_pid; /* main process id used in log filename */
179 :
180 : char *pghost = "";
181 : char *pgport = "";
182 : char *login = NULL;
183 : char *dbName;
184 : char *logfile_prefix = NULL;
185 : const char *progname;
186 :
187 : #define WSEP '@' /* weight separator */
188 :
189 : volatile bool timer_exceeded = false; /* flag from signal handler */
190 :
191 : /*
192 : * Variable definitions. If a variable has a string value, "value" is that
193 : * value, is_numeric is false, and num_value is undefined. If the value is
194 : * known to be numeric, is_numeric is true and num_value contains the value
195 : * (in any permitted numeric variant). In this case "value" contains the
196 : * string equivalent of the number, if we've had occasion to compute that,
197 : * or NULL if we haven't.
198 : */
199 : typedef struct
200 : {
201 : char *name; /* variable's name */
202 : char *value; /* its value in string form, if known */
203 : bool is_numeric; /* is numeric value known? */
204 : PgBenchValue num_value; /* variable's value in numeric form */
205 : } Variable;
206 :
207 : #define MAX_SCRIPTS 128 /* max number of SQL scripts allowed */
208 : #define SHELL_COMMAND_SIZE 256 /* maximum size allowed for shell command */
209 :
210 : /*
211 : * Simple data structure to keep stats about something.
212 : *
213 : * XXX probably the first value should be kept and used as an offset for
214 : * better numerical stability...
215 : */
216 : typedef struct SimpleStats
217 : {
218 : int64 count; /* how many values were encountered */
219 : double min; /* the minimum seen */
220 : double max; /* the maximum seen */
221 : double sum; /* sum of values */
222 : double sum2; /* sum of squared values */
223 : } SimpleStats;
224 :
225 : /*
226 : * Data structure to hold various statistics: per-thread and per-script stats
227 : * are maintained and merged together.
228 : */
229 : typedef struct StatsData
230 : {
231 : time_t start_time; /* interval start time, for aggregates */
232 : int64 cnt; /* number of transactions, including skipped */
233 : int64 skipped; /* number of transactions skipped under --rate
234 : * and --latency-limit */
235 : SimpleStats latency;
236 : SimpleStats lag;
237 : } StatsData;
238 :
239 : /*
240 : * Connection state machine states.
241 : */
242 : typedef enum
243 : {
244 : /*
245 : * The client must first choose a script to execute. Once chosen, it can
246 : * either be throttled (state CSTATE_START_THROTTLE under --rate) or start
247 : * right away (state CSTATE_START_TX).
248 : */
249 : CSTATE_CHOOSE_SCRIPT,
250 :
251 : /*
252 : * In CSTATE_START_THROTTLE state, we calculate when to begin the next
253 : * transaction, and advance to CSTATE_THROTTLE. CSTATE_THROTTLE state
254 : * sleeps until that moment. (If throttling is not enabled, doCustom()
255 : * falls directly through from CSTATE_START_THROTTLE to CSTATE_START_TX.)
256 : */
257 : CSTATE_START_THROTTLE,
258 : CSTATE_THROTTLE,
259 :
260 : /*
261 : * CSTATE_START_TX performs start-of-transaction processing. Establishes
262 : * a new connection for the transaction, in --connect mode, and records
263 : * the transaction start time.
264 : */
265 : CSTATE_START_TX,
266 :
267 : /*
268 : * We loop through these states, to process each command in the script:
269 : *
270 : * CSTATE_START_COMMAND starts the execution of a command. On a SQL
271 : * command, the command is sent to the server, and we move to
272 : * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set,
273 : * and we enter the CSTATE_SLEEP state to wait for it to expire. Other
274 : * meta-commands are executed immediately.
275 : *
276 : * CSTATE_WAIT_RESULT waits until we get a result set back from the server
277 : * for the current command.
278 : *
279 : * CSTATE_SLEEP waits until the end of \sleep.
280 : *
281 : * CSTATE_END_COMMAND records the end-of-command timestamp, increments the
282 : * command counter, and loops back to CSTATE_START_COMMAND state.
283 : */
284 : CSTATE_START_COMMAND,
285 : CSTATE_WAIT_RESULT,
286 : CSTATE_SLEEP,
287 : CSTATE_END_COMMAND,
288 :
289 : /*
290 : * CSTATE_END_TX performs end-of-transaction processing. Calculates
291 : * latency, and logs the transaction. In --connect mode, closes the
292 : * current connection. Chooses the next script to execute and starts over
293 : * in CSTATE_START_THROTTLE state, or enters CSTATE_FINISHED if we have no
294 : * more work to do.
295 : */
296 : CSTATE_END_TX,
297 :
298 : /*
299 : * Final states. CSTATE_ABORTED means that the script execution was
300 : * aborted because a command failed, CSTATE_FINISHED means success.
301 : */
302 : CSTATE_ABORTED,
303 : CSTATE_FINISHED
304 : } ConnectionStateEnum;
305 :
306 : /*
307 : * Connection state.
308 : */
309 : typedef struct
310 : {
311 : PGconn *con; /* connection handle to DB */
312 : int id; /* client No. */
313 : ConnectionStateEnum state; /* state machine's current state. */
314 :
315 : int use_file; /* index in sql_script for this client */
316 : int command; /* command number in script */
317 :
318 : /* client variables */
319 : Variable *variables; /* array of variable definitions */
320 : int nvariables; /* number of variables */
321 : bool vars_sorted; /* are variables sorted by name? */
322 :
323 : /* various times about current transaction */
324 : int64 txn_scheduled; /* scheduled start time of transaction (usec) */
325 : int64 sleep_until; /* scheduled start time of next cmd (usec) */
326 : instr_time txn_begin; /* used for measuring schedule lag times */
327 : instr_time stmt_begin; /* used for measuring statement latencies */
328 :
329 : bool prepared[MAX_SCRIPTS]; /* whether client prepared the script */
330 :
331 : /* per client collected stats */
332 : int64 cnt; /* client transaction count, for -t */
333 : int ecnt; /* error count */
334 : } CState;
335 :
336 : /*
337 : * Thread state
338 : */
339 : typedef struct
340 : {
341 : int tid; /* thread id */
342 : pthread_t thread; /* thread handle */
343 : CState *state; /* array of CState */
344 : int nstate; /* length of state[] */
345 : unsigned short random_state[3]; /* separate randomness for each thread */
346 : int64 throttle_trigger; /* previous/next throttling (us) */
347 : FILE *logfile; /* where to log, or NULL */
348 :
349 : /* per thread collected stats */
350 : instr_time start_time; /* thread start time */
351 : instr_time conn_time;
352 : StatsData stats;
353 : int64 latency_late; /* executed but late transactions */
354 : } TState;
355 :
356 : #define INVALID_THREAD ((pthread_t) 0)
357 :
358 : /*
359 : * queries read from files
360 : */
361 : #define SQL_COMMAND 1
362 : #define META_COMMAND 2
363 : #define MAX_ARGS 10
364 :
365 : typedef enum QueryMode
366 : {
367 : QUERY_SIMPLE, /* simple query */
368 : QUERY_EXTENDED, /* extended query */
369 : QUERY_PREPARED, /* extended query with prepared statements */
370 : NUM_QUERYMODE
371 : } QueryMode;
372 :
373 : static QueryMode querymode = QUERY_SIMPLE;
374 : static const char *QUERYMODE[] = {"simple", "extended", "prepared"};
375 :
376 : typedef struct
377 : {
378 : char *line; /* text of command line */
379 : int command_num; /* unique index of this Command struct */
380 : int type; /* command type (SQL_COMMAND or META_COMMAND) */
381 : int argc; /* number of command words */
382 : char *argv[MAX_ARGS]; /* command word list */
383 : PgBenchExpr *expr; /* parsed expression, if needed */
384 : SimpleStats stats; /* time spent in this command */
385 : } Command;
386 :
387 : typedef struct ParsedScript
388 : {
389 : const char *desc; /* script descriptor (eg, file name) */
390 : int weight; /* selection weight */
391 : Command **commands; /* NULL-terminated array of Commands */
392 : StatsData stats; /* total time spent in script */
393 : } ParsedScript;
394 :
395 : static ParsedScript sql_script[MAX_SCRIPTS]; /* SQL script files */
396 : static int num_scripts; /* number of scripts in sql_script[] */
397 : static int num_commands = 0; /* total number of Command structs */
398 : static int64 total_weight = 0;
399 :
400 : static int debug = 0; /* debug flag */
401 :
402 : /* Builtin test scripts */
403 : typedef struct BuiltinScript
404 : {
405 : const char *name; /* very short name for -b ... */
406 : const char *desc; /* short description */
407 : const char *script; /* actual pgbench script */
408 : } BuiltinScript;
409 :
410 : static const BuiltinScript builtin_script[] =
411 : {
412 : {
413 : "tpcb-like",
414 : "<builtin: TPC-B (sort of)>",
415 : "\\set aid random(1, " CppAsString2(naccounts) " * :scale)\n"
416 : "\\set bid random(1, " CppAsString2(nbranches) " * :scale)\n"
417 : "\\set tid random(1, " CppAsString2(ntellers) " * :scale)\n"
418 : "\\set delta random(-5000, 5000)\n"
419 : "BEGIN;\n"
420 : "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
421 : "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
422 : "UPDATE pgbench_tellers SET tbalance = tbalance + :delta WHERE tid = :tid;\n"
423 : "UPDATE pgbench_branches SET bbalance = bbalance + :delta WHERE bid = :bid;\n"
424 : "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
425 : "END;\n"
426 : },
427 : {
428 : "simple-update",
429 : "<builtin: simple update>",
430 : "\\set aid random(1, " CppAsString2(naccounts) " * :scale)\n"
431 : "\\set bid random(1, " CppAsString2(nbranches) " * :scale)\n"
432 : "\\set tid random(1, " CppAsString2(ntellers) " * :scale)\n"
433 : "\\set delta random(-5000, 5000)\n"
434 : "BEGIN;\n"
435 : "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
436 : "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
437 : "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
438 : "END;\n"
439 : },
440 : {
441 : "select-only",
442 : "<builtin: select only>",
443 : "\\set aid random(1, " CppAsString2(naccounts) " * :scale)\n"
444 : "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
445 : }
446 : };
447 :
448 :
449 : /* Function prototypes */
450 : static void setIntValue(PgBenchValue *pv, int64 ival);
451 : static void setDoubleValue(PgBenchValue *pv, double dval);
452 : static bool evaluateExpr(TState *, CState *, PgBenchExpr *, PgBenchValue *);
453 : static void doLog(TState *thread, CState *st,
454 : StatsData *agg, bool skipped, double latency, double lag);
455 : static void processXactStats(TState *thread, CState *st, instr_time *now,
456 : bool skipped, StatsData *agg);
457 : static void pgbench_error(const char *fmt,...) pg_attribute_printf(1, 2);
458 : static void addScript(ParsedScript script);
459 : static void *threadRun(void *arg);
460 : static void setalarm(int seconds);
461 :
462 :
463 : /* callback functions for our flex lexer */
464 : static const PsqlScanCallbacks pgbench_callbacks = {
465 : NULL, /* don't need get_variable functionality */
466 : pgbench_error
467 : };
468 :
469 :
470 : static void
471 1 : usage(void)
472 : {
473 1 : printf("%s is a benchmarking tool for PostgreSQL.\n\n"
474 : "Usage:\n"
475 : " %s [OPTION]... [DBNAME]\n"
476 : "\nInitialization options:\n"
477 : " -i, --initialize invokes initialization mode\n"
478 : " -F, --fillfactor=NUM set fill factor\n"
479 : " -n, --no-vacuum do not run VACUUM after initialization\n"
480 : " -q, --quiet quiet logging (one message each 5 seconds)\n"
481 : " -s, --scale=NUM scaling factor\n"
482 : " --foreign-keys create foreign key constraints between tables\n"
483 : " --index-tablespace=TABLESPACE\n"
484 : " create indexes in the specified tablespace\n"
485 : " --tablespace=TABLESPACE create tables in the specified tablespace\n"
486 : " --unlogged-tables create tables as unlogged tables\n"
487 : "\nOptions to select what to run:\n"
488 : " -b, --builtin=NAME[@W] add builtin script NAME weighted at W (default: 1)\n"
489 : " (use \"-b list\" to list available scripts)\n"
490 : " -f, --file=FILENAME[@W] add script FILENAME weighted at W (default: 1)\n"
491 : " -N, --skip-some-updates skip updates of pgbench_tellers and pgbench_branches\n"
492 : " (same as \"-b simple-update\")\n"
493 : " -S, --select-only perform SELECT-only transactions\n"
494 : " (same as \"-b select-only\")\n"
495 : "\nBenchmarking options:\n"
496 : " -c, --client=NUM number of concurrent database clients (default: 1)\n"
497 : " -C, --connect establish new connection for each transaction\n"
498 : " -D, --define=VARNAME=VALUE\n"
499 : " define variable for use by custom script\n"
500 : " -j, --jobs=NUM number of threads (default: 1)\n"
501 : " -l, --log write transaction times to log file\n"
502 : " -L, --latency-limit=NUM count transactions lasting more than NUM ms as late\n"
503 : " -M, --protocol=simple|extended|prepared\n"
504 : " protocol for submitting queries (default: simple)\n"
505 : " -n, --no-vacuum do not run VACUUM before tests\n"
506 : " -P, --progress=NUM show thread progress report every NUM seconds\n"
507 : " -r, --report-latencies report average latency per command\n"
508 : " -R, --rate=NUM target rate in transactions per second\n"
509 : " -s, --scale=NUM report this scale factor in output\n"
510 : " -t, --transactions=NUM number of transactions each client runs (default: 10)\n"
511 : " -T, --time=NUM duration of benchmark test in seconds\n"
512 : " -v, --vacuum-all vacuum all four standard tables before tests\n"
513 : " --aggregate-interval=NUM aggregate data over NUM seconds\n"
514 : " --log-prefix=PREFIX prefix for transaction time log file\n"
515 : " (default: \"pgbench_log\")\n"
516 : " --progress-timestamp use Unix epoch timestamps for progress\n"
517 : " --sampling-rate=NUM fraction of transactions to log (e.g., 0.01 for 1%%)\n"
518 : "\nCommon options:\n"
519 : " -d, --debug print debugging output\n"
520 : " -h, --host=HOSTNAME database server host or socket directory\n"
521 : " -p, --port=PORT database server port number\n"
522 : " -U, --username=USERNAME connect as specified database user\n"
523 : " -V, --version output version information, then exit\n"
524 : " -?, --help show this help, then exit\n"
525 : "\n"
526 : "Report bugs to <pgsql-bugs@postgresql.org>.\n",
527 : progname, progname);
528 1 : }
529 :
530 : /* return whether str matches "^\s*[-+]?[0-9]+$" */
531 : static bool
532 3 : is_an_int(const char *str)
533 : {
534 3 : const char *ptr = str;
535 :
536 : /* skip leading spaces; cast is consistent with strtoint64 */
537 6 : while (*ptr && isspace((unsigned char) *ptr))
538 0 : ptr++;
539 :
540 : /* skip sign */
541 3 : if (*ptr == '+' || *ptr == '-')
542 2 : ptr++;
543 :
544 : /* at least one digit */
545 3 : if (*ptr && !isdigit((unsigned char) *ptr))
546 1 : return false;
547 :
548 : /* eat all digits */
549 25 : while (*ptr && isdigit((unsigned char) *ptr))
550 21 : ptr++;
551 :
552 : /* must have reached end of string */
553 2 : return *ptr == '\0';
554 : }
555 :
556 :
557 : /*
558 : * strtoint64 -- convert a string to 64-bit integer
559 : *
560 : * This function is a modified version of scanint8() from
561 : * src/backend/utils/adt/int8.c.
562 : */
563 : int64
564 467 : strtoint64(const char *str)
565 : {
566 467 : const char *ptr = str;
567 467 : int64 result = 0;
568 467 : int sign = 1;
569 :
570 : /*
571 : * Do our own scan, rather than relying on sscanf which might be broken
572 : * for long long.
573 : */
574 :
575 : /* skip leading spaces */
576 934 : while (*ptr && isspace((unsigned char) *ptr))
577 0 : ptr++;
578 :
579 : /* handle sign */
580 467 : if (*ptr == '-')
581 : {
582 1 : ptr++;
583 :
584 : /*
585 : * Do an explicit check for INT64_MIN. Ugly though this is, it's
586 : * cleaner than trying to get the loop below to handle it portably.
587 : */
588 1 : if (strncmp(ptr, "9223372036854775808", 19) == 0)
589 : {
590 1 : result = PG_INT64_MIN;
591 1 : ptr += 19;
592 1 : goto gotdigits;
593 : }
594 0 : sign = -1;
595 : }
596 466 : else if (*ptr == '+')
597 0 : ptr++;
598 :
599 : /* require at least one digit */
600 466 : if (!isdigit((unsigned char) *ptr))
601 0 : fprintf(stderr, "invalid input syntax for integer: \"%s\"\n", str);
602 :
603 : /* process digits */
604 2339 : while (*ptr && isdigit((unsigned char) *ptr))
605 : {
606 1407 : int64 tmp = result * 10 + (*ptr++ - '0');
607 :
608 1407 : if ((tmp / 10) != result) /* overflow? */
609 2 : fprintf(stderr, "value \"%s\" is out of range for type bigint\n", str);
610 1407 : result = tmp;
611 : }
612 :
613 : gotdigits:
614 :
615 : /* allow trailing whitespace, but not other trailing chars */
616 934 : while (*ptr != '\0' && isspace((unsigned char) *ptr))
617 0 : ptr++;
618 :
619 467 : if (*ptr != '\0')
620 0 : fprintf(stderr, "invalid input syntax for integer: \"%s\"\n", str);
621 :
622 467 : return ((sign < 0) ? -result : result);
623 : }
624 :
625 : /* random number generator: uniform distribution from min to max inclusive */
626 : static int64
627 1700 : getrand(TState *thread, int64 min, int64 max)
628 : {
629 : /*
630 : * Odd coding is so that min and max have approximately the same chance of
631 : * being selected as do numbers between them.
632 : *
633 : * pg_erand48() is thread-safe and concurrent, which is why we use it
634 : * rather than random(), which in glibc is non-reentrant, and therefore
635 : * protected by a mutex, and therefore a bottleneck on machines with many
636 : * CPUs.
637 : */
638 1700 : return min + (int64) ((max - min + 1) * pg_erand48(thread->random_state));
639 : }
640 :
641 : /*
642 : * random number generator: exponential distribution from min to max inclusive.
643 : * the parameter is so that the density of probability for the last cut-off max
644 : * value is exp(-parameter).
645 : */
646 : static int64
647 1 : getExponentialRand(TState *thread, int64 min, int64 max, double parameter)
648 : {
649 : double cut,
650 : uniform,
651 : rand;
652 :
653 : /* abort if wrong parameter, but must really be checked beforehand */
654 1 : Assert(parameter > 0.0);
655 1 : cut = exp(-parameter);
656 : /* erand in [0, 1), uniform in (0, 1] */
657 1 : uniform = 1.0 - pg_erand48(thread->random_state);
658 :
659 : /*
660 : * inner expression in (cut, 1] (if parameter > 0), rand in [0, 1)
661 : */
662 1 : Assert((1.0 - cut) != 0.0);
663 1 : rand = -log(cut + (1.0 - cut) * uniform) / parameter;
664 : /* return int64 random number within between min and max */
665 1 : return min + (int64) ((max - min + 1) * rand);
666 : }
667 :
668 : /* random number generator: gaussian distribution from min to max inclusive */
669 : static int64
670 1 : getGaussianRand(TState *thread, int64 min, int64 max, double parameter)
671 : {
672 : double stdev;
673 : double rand;
674 :
675 : /* abort if parameter is too low, but must really be checked beforehand */
676 1 : Assert(parameter >= MIN_GAUSSIAN_PARAM);
677 :
678 : /*
679 : * Get user specified random number from this loop, with -parameter <
680 : * stdev <= parameter
681 : *
682 : * This loop is executed until the number is in the expected range.
683 : *
684 : * As the minimum parameter is 2.0, the probability of looping is low:
685 : * sqrt(-2 ln(r)) <= 2 => r >= e^{-2} ~ 0.135, then when taking the
686 : * average sinus multiplier as 2/pi, we have a 8.6% looping probability in
687 : * the worst case. For a parameter value of 5.0, the looping probability
688 : * is about e^{-5} * 2 / pi ~ 0.43%.
689 : */
690 : do
691 : {
692 : /*
693 : * pg_erand48 generates [0,1), but for the basic version of the
694 : * Box-Muller transform the two uniformly distributed random numbers
695 : * are expected in (0, 1] (see
696 : * http://en.wikipedia.org/wiki/Box_muller)
697 : */
698 1 : double rand1 = 1.0 - pg_erand48(thread->random_state);
699 1 : double rand2 = 1.0 - pg_erand48(thread->random_state);
700 :
701 : /* Box-Muller basic form transform */
702 1 : double var_sqrt = sqrt(-2.0 * log(rand1));
703 :
704 1 : stdev = var_sqrt * sin(2.0 * M_PI * rand2);
705 :
706 : /*
707 : * we may try with cos, but there may be a bias induced if the
708 : * previous value fails the test. To be on the safe side, let us try
709 : * over.
710 : */
711 : }
712 1 : while (stdev < -parameter || stdev >= parameter);
713 :
714 : /* stdev is in [-parameter, parameter), normalization to [0,1) */
715 1 : rand = (stdev + parameter) / (parameter * 2.0);
716 :
717 : /* return int64 random number within between min and max */
718 1 : return min + (int64) ((max - min + 1) * rand);
719 : }
720 :
721 : /*
722 : * random number generator: generate a value, such that the series of values
723 : * will approximate a Poisson distribution centered on the given value.
724 : */
725 : static int64
726 254 : getPoissonRand(TState *thread, int64 center)
727 : {
728 : /*
729 : * Use inverse transform sampling to generate a value > 0, such that the
730 : * expected (i.e. average) value is the given argument.
731 : */
732 : double uniform;
733 :
734 : /* erand in [0, 1), uniform in (0, 1] */
735 254 : uniform = 1.0 - pg_erand48(thread->random_state);
736 :
737 254 : return (int64) (-log(uniform) * ((double) center) + 0.5);
738 : }
739 :
740 : /*
741 : * Initialize the given SimpleStats struct to all zeroes
742 : */
743 : static void
744 1075 : initSimpleStats(SimpleStats *ss)
745 : {
746 1075 : memset(ss, 0, sizeof(SimpleStats));
747 1075 : }
748 :
749 : /*
750 : * Accumulate one value into a SimpleStats struct.
751 : */
752 : static void
753 1624 : addToSimpleStats(SimpleStats *ss, double val)
754 : {
755 1624 : if (ss->count == 0 || val < ss->min)
756 113 : ss->min = val;
757 1624 : if (ss->count == 0 || val > ss->max)
758 828 : ss->max = val;
759 1624 : ss->count++;
760 1624 : ss->sum += val;
761 1624 : ss->sum2 += val * val;
762 1624 : }
763 :
764 : /*
765 : * Merge two SimpleStats objects
766 : */
767 : static void
768 68 : mergeSimpleStats(SimpleStats *acc, SimpleStats *ss)
769 : {
770 68 : if (acc->count == 0 || ss->min < acc->min)
771 66 : acc->min = ss->min;
772 68 : if (acc->count == 0 || ss->max > acc->max)
773 64 : acc->max = ss->max;
774 68 : acc->count += ss->count;
775 68 : acc->sum += ss->sum;
776 68 : acc->sum2 += ss->sum2;
777 68 : }
778 :
779 : /*
780 : * Initialize a StatsData struct to mostly zeroes, with its start time set to
781 : * the given value.
782 : */
783 : static void
784 291 : initStats(StatsData *sd, time_t start_time)
785 : {
786 291 : sd->start_time = start_time;
787 291 : sd->cnt = 0;
788 291 : sd->skipped = 0;
789 291 : initSimpleStats(&sd->latency);
790 291 : initSimpleStats(&sd->lag);
791 291 : }
792 :
793 : /*
794 : * Accumulate one additional item into the given stats object.
795 : */
796 : static void
797 642 : accumStats(StatsData *stats, bool skipped, double lat, double lag)
798 : {
799 642 : stats->cnt++;
800 :
801 642 : if (skipped)
802 : {
803 : /* no latency to record on skipped transactions */
804 20 : stats->skipped++;
805 : }
806 : else
807 : {
808 622 : addToSimpleStats(&stats->latency, lat);
809 :
810 : /* and possibly the same for schedule lag */
811 622 : if (throttle_delay)
812 522 : addToSimpleStats(&stats->lag, lag);
813 : }
814 642 : }
815 :
816 : /* call PQexec() and exit() on failure */
817 : static void
818 67 : executeStatement(PGconn *con, const char *sql)
819 : {
820 : PGresult *res;
821 :
822 67 : res = PQexec(con, sql);
823 67 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
824 : {
825 0 : fprintf(stderr, "%s", PQerrorMessage(con));
826 0 : exit(1);
827 : }
828 67 : PQclear(res);
829 67 : }
830 :
831 : /* call PQexec() and complain, but without exiting, on failure */
832 : static void
833 12 : tryExecuteStatement(PGconn *con, const char *sql)
834 : {
835 : PGresult *res;
836 :
837 12 : res = PQexec(con, sql);
838 12 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
839 : {
840 0 : fprintf(stderr, "%s", PQerrorMessage(con));
841 0 : fprintf(stderr, "(ignoring this error and continuing anyway)\n");
842 : }
843 12 : PQclear(res);
844 12 : }
845 :
846 : /* set up a connection to the backend */
847 : static PGconn *
848 189 : doConnect(void)
849 : {
850 : PGconn *conn;
851 : bool new_pass;
852 : static bool have_password = false;
853 : static char password[100];
854 :
855 : /*
856 : * Start the connection. Loop until we have a password if requested by
857 : * backend.
858 : */
859 : do
860 : {
861 : #define PARAMS_ARRAY_SIZE 7
862 :
863 : const char *keywords[PARAMS_ARRAY_SIZE];
864 : const char *values[PARAMS_ARRAY_SIZE];
865 :
866 189 : keywords[0] = "host";
867 189 : values[0] = pghost;
868 189 : keywords[1] = "port";
869 189 : values[1] = pgport;
870 189 : keywords[2] = "user";
871 189 : values[2] = login;
872 189 : keywords[3] = "password";
873 189 : values[3] = have_password ? password : NULL;
874 189 : keywords[4] = "dbname";
875 189 : values[4] = dbName;
876 189 : keywords[5] = "fallback_application_name";
877 189 : values[5] = progname;
878 189 : keywords[6] = NULL;
879 189 : values[6] = NULL;
880 :
881 189 : new_pass = false;
882 :
883 189 : conn = PQconnectdbParams(keywords, values, true);
884 :
885 189 : if (!conn)
886 : {
887 0 : fprintf(stderr, "connection to database \"%s\" failed\n",
888 : dbName);
889 0 : return NULL;
890 : }
891 :
892 190 : if (PQstatus(conn) == CONNECTION_BAD &&
893 1 : PQconnectionNeedsPassword(conn) &&
894 0 : !have_password)
895 : {
896 0 : PQfinish(conn);
897 0 : simple_prompt("Password: ", password, sizeof(password), false);
898 0 : have_password = true;
899 0 : new_pass = true;
900 : }
901 189 : } while (new_pass);
902 :
903 : /* check to see that the backend connection was successfully made */
904 189 : if (PQstatus(conn) == CONNECTION_BAD)
905 : {
906 1 : fprintf(stderr, "connection to database \"%s\" failed:\n%s",
907 : dbName, PQerrorMessage(conn));
908 1 : PQfinish(conn);
909 1 : return NULL;
910 : }
911 :
912 188 : return conn;
913 : }
914 :
915 : /* throw away response from backend */
916 : static void
917 2063 : discard_response(CState *state)
918 : {
919 : PGresult *res;
920 :
921 : do
922 : {
923 2063 : res = PQgetResult(state->con);
924 2063 : if (res)
925 0 : PQclear(res);
926 2063 : } while (res);
927 2063 : }
928 :
929 : /* qsort comparator for Variable array */
930 : static int
931 13437 : compareVariableNames(const void *v1, const void *v2)
932 : {
933 13437 : return strcmp(((const Variable *) v1)->name,
934 13437 : ((const Variable *) v2)->name);
935 : }
936 :
937 : /* Locate a variable by name; returns NULL if unknown */
938 : static Variable *
939 5424 : lookupVariable(CState *st, char *name)
940 : {
941 : Variable key;
942 :
943 : /* On some versions of Solaris, bsearch of zero items dumps core */
944 5424 : if (st->nvariables <= 0)
945 77 : return NULL;
946 :
947 : /* Sort if we have to */
948 5347 : if (!st->vars_sorted)
949 : {
950 179 : qsort((void *) st->variables, st->nvariables, sizeof(Variable),
951 : compareVariableNames);
952 179 : st->vars_sorted = true;
953 : }
954 :
955 : /* Now we can search */
956 5347 : key.name = name;
957 10694 : return (Variable *) bsearch((void *) &key,
958 5347 : (void *) st->variables,
959 5347 : st->nvariables,
960 : sizeof(Variable),
961 : compareVariableNames);
962 : }
963 :
964 : /* Get the value of a variable, in string form; returns NULL if unknown */
965 : static char *
966 2123 : getVariable(CState *st, char *name)
967 : {
968 : Variable *var;
969 : char stringform[64];
970 :
971 2123 : var = lookupVariable(st, name);
972 2123 : if (var == NULL)
973 2 : return NULL; /* not found */
974 :
975 2121 : if (var->value)
976 497 : return var->value; /* we have it in string form */
977 :
978 : /* We need to produce a string equivalent of the numeric value */
979 1624 : Assert(var->is_numeric);
980 1624 : if (var->num_value.type == PGBT_INT)
981 1623 : snprintf(stringform, sizeof(stringform),
982 : INT64_FORMAT, var->num_value.u.ival);
983 : else
984 : {
985 1 : Assert(var->num_value.type == PGBT_DOUBLE);
986 1 : snprintf(stringform, sizeof(stringform),
987 : "%.*g", DBL_DIG, var->num_value.u.dval);
988 : }
989 1624 : var->value = pg_strdup(stringform);
990 1624 : return var->value;
991 : }
992 :
993 : /* Try to convert variable to numeric form; return false on failure */
994 : static bool
995 1452 : makeVariableNumeric(Variable *var)
996 : {
997 1452 : if (var->is_numeric)
998 1449 : return true; /* no work */
999 :
1000 3 : if (is_an_int(var->value))
1001 : {
1002 1 : setIntValue(&var->num_value, strtoint64(var->value));
1003 1 : var->is_numeric = true;
1004 : }
1005 : else /* type should be double */
1006 : {
1007 : double dv;
1008 : char xs;
1009 :
1010 2 : if (sscanf(var->value, "%lf%c", &dv, &xs) != 1)
1011 : {
1012 1 : fprintf(stderr,
1013 : "malformed variable \"%s\" value: \"%s\"\n",
1014 : var->name, var->value);
1015 1 : return false;
1016 : }
1017 1 : setDoubleValue(&var->num_value, dv);
1018 1 : var->is_numeric = true;
1019 : }
1020 2 : return true;
1021 : }
1022 :
1023 : /*
1024 : * Check whether a variable's name is allowed.
1025 : *
1026 : * We allow any non-ASCII character, as well as ASCII letters, digits, and
1027 : * underscore.
1028 : *
1029 : * Keep this in sync with the definitions of variable name characters in
1030 : * "src/fe_utils/psqlscan.l", "src/bin/psql/psqlscanslash.l" and
1031 : * "src/bin/pgbench/exprscan.l". Also see parseVariable(), below.
1032 : *
1033 : * Note: this static function is copied from "src/bin/psql/variables.c"
1034 : */
1035 : static bool
1036 213 : valid_variable_name(const char *name)
1037 : {
1038 213 : const unsigned char *ptr = (const unsigned char *) name;
1039 :
1040 : /* Mustn't be zero-length */
1041 213 : if (*ptr == '\0')
1042 0 : return false;
1043 :
1044 1467 : while (*ptr)
1045 : {
1046 2084 : if (IS_HIGHBIT_SET(*ptr) ||
1047 1042 : strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
1048 1042 : "_0123456789", *ptr) != NULL)
1049 1041 : ptr++;
1050 : else
1051 1 : return false;
1052 : }
1053 :
1054 212 : return true;
1055 : }
1056 :
1057 : /*
1058 : * Lookup a variable by name, creating it if need be.
1059 : * Caller is expected to assign a value to the variable.
1060 : * Returns NULL on failure (bad name).
1061 : */
1062 : static Variable *
1063 1786 : lookupCreateVariable(CState *st, const char *context, char *name)
1064 : {
1065 : Variable *var;
1066 :
1067 1786 : var = lookupVariable(st, name);
1068 1786 : if (var == NULL)
1069 : {
1070 : Variable *newvars;
1071 :
1072 : /*
1073 : * Check for the name only when declaring a new variable to avoid
1074 : * overhead.
1075 : */
1076 213 : if (!valid_variable_name(name))
1077 : {
1078 1 : fprintf(stderr, "%s: invalid variable name: \"%s\"\n",
1079 : context, name);
1080 1 : return NULL;
1081 : }
1082 :
1083 : /* Create variable at the end of the array */
1084 212 : if (st->variables)
1085 147 : newvars = (Variable *) pg_realloc(st->variables,
1086 147 : (st->nvariables + 1) * sizeof(Variable));
1087 : else
1088 65 : newvars = (Variable *) pg_malloc(sizeof(Variable));
1089 :
1090 212 : st->variables = newvars;
1091 :
1092 212 : var = &newvars[st->nvariables];
1093 :
1094 212 : var->name = pg_strdup(name);
1095 212 : var->value = NULL;
1096 : /* caller is expected to initialize remaining fields */
1097 :
1098 212 : st->nvariables++;
1099 : /* we don't re-sort the array till we have to */
1100 212 : st->vars_sorted = false;
1101 : }
1102 :
1103 1785 : return var;
1104 : }
1105 :
1106 : /* Assign a string value to a variable, creating it if need be */
1107 : /* Returns false on failure (bad name) */
1108 : static bool
1109 37 : putVariable(CState *st, const char *context, char *name, const char *value)
1110 : {
1111 : Variable *var;
1112 : char *val;
1113 :
1114 37 : var = lookupCreateVariable(st, context, name);
1115 37 : if (!var)
1116 0 : return false;
1117 :
1118 : /* dup then free, in case value is pointing at this variable */
1119 37 : val = pg_strdup(value);
1120 :
1121 37 : if (var->value)
1122 0 : free(var->value);
1123 37 : var->value = val;
1124 37 : var->is_numeric = false;
1125 :
1126 37 : return true;
1127 : }
1128 :
1129 : /* Assign a numeric value to a variable, creating it if need be */
1130 : /* Returns false on failure (bad name) */
1131 : static bool
1132 1749 : putVariableNumber(CState *st, const char *context, char *name,
1133 : const PgBenchValue *value)
1134 : {
1135 : Variable *var;
1136 :
1137 1749 : var = lookupCreateVariable(st, context, name);
1138 1749 : if (!var)
1139 1 : return false;
1140 :
1141 1748 : if (var->value)
1142 1572 : free(var->value);
1143 1748 : var->value = NULL;
1144 1748 : var->is_numeric = true;
1145 1748 : var->num_value = *value;
1146 :
1147 1748 : return true;
1148 : }
1149 :
1150 : /* Assign an integer value to a variable, creating it if need be */
1151 : /* Returns false on failure (bad name) */
1152 : static bool
1153 103 : putVariableInt(CState *st, const char *context, char *name, int64 value)
1154 : {
1155 : PgBenchValue val;
1156 :
1157 103 : setIntValue(&val, value);
1158 103 : return putVariableNumber(st, context, name, &val);
1159 : }
1160 :
1161 : /*
1162 : * Parse a possible variable reference (:varname).
1163 : *
1164 : * "sql" points at a colon. If what follows it looks like a valid
1165 : * variable name, return a malloc'd string containing the variable name,
1166 : * and set *eaten to the number of characters consumed.
1167 : * Otherwise, return NULL.
1168 : */
1169 : static char *
1170 508 : parseVariable(const char *sql, int *eaten)
1171 : {
1172 508 : int i = 0;
1173 : char *name;
1174 :
1175 : do
1176 : {
1177 2059 : i++;
1178 4118 : } while (IS_HIGHBIT_SET(sql[i]) ||
1179 2059 : strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
1180 2059 : "_0123456789", sql[i]) != NULL);
1181 508 : if (i == 1)
1182 15 : return NULL; /* no valid variable name chars */
1183 :
1184 493 : name = pg_malloc(i);
1185 493 : memcpy(name, &sql[1], i - 1);
1186 493 : name[i - 1] = '\0';
1187 :
1188 493 : *eaten = i;
1189 493 : return name;
1190 : }
1191 :
1192 : static char *
1193 492 : replaceVariable(char **sql, char *param, int len, char *value)
1194 : {
1195 492 : int valueln = strlen(value);
1196 :
1197 492 : if (valueln > len)
1198 : {
1199 354 : size_t offset = param - *sql;
1200 :
1201 354 : *sql = pg_realloc(*sql, strlen(*sql) - len + valueln + 1);
1202 354 : param = *sql + offset;
1203 : }
1204 :
1205 492 : if (valueln != len)
1206 451 : memmove(param + valueln, param + len, strlen(param + len) + 1);
1207 492 : memcpy(param, value, valueln);
1208 :
1209 492 : return param + valueln;
1210 : }
1211 :
1212 : static char *
1213 450 : assignVariables(CState *st, char *sql)
1214 : {
1215 : char *p,
1216 : *name,
1217 : *val;
1218 :
1219 450 : p = sql;
1220 1380 : while ((p = strchr(p, ':')) != NULL)
1221 : {
1222 : int eaten;
1223 :
1224 480 : name = parseVariable(p, &eaten);
1225 480 : if (name == NULL)
1226 : {
1227 40 : while (*p == ':')
1228 : {
1229 20 : p++;
1230 : }
1231 20 : continue;
1232 : }
1233 :
1234 470 : val = getVariable(st, name);
1235 470 : free(name);
1236 470 : if (val == NULL)
1237 : {
1238 0 : p++;
1239 0 : continue;
1240 : }
1241 :
1242 470 : p = replaceVariable(&sql, p, eaten, val);
1243 : }
1244 :
1245 450 : return sql;
1246 : }
1247 :
1248 : static void
1249 1614 : getQueryParams(CState *st, const Command *command, const char **params)
1250 : {
1251 : int i;
1252 :
1253 3262 : for (i = 0; i < command->argc - 1; i++)
1254 1648 : params[i] = getVariable(st, command->argv[i + 1]);
1255 1614 : }
1256 :
1257 : /* get a value as an int, tell if there is a problem */
1258 : static bool
1259 6308 : coerceToInt(PgBenchValue *pval, int64 *ival)
1260 : {
1261 6308 : if (pval->type == PGBT_INT)
1262 : {
1263 6306 : *ival = pval->u.ival;
1264 6306 : return true;
1265 : }
1266 : else
1267 : {
1268 2 : double dval = pval->u.dval;
1269 :
1270 2 : Assert(pval->type == PGBT_DOUBLE);
1271 2 : if (dval < PG_INT64_MIN || PG_INT64_MAX < dval)
1272 : {
1273 1 : fprintf(stderr, "double to int overflow for %f\n", dval);
1274 1 : return false;
1275 : }
1276 1 : *ival = (int64) dval;
1277 1 : return true;
1278 : }
1279 : }
1280 :
1281 : /* get a value as a double, or tell if there is a problem */
1282 : static bool
1283 44 : coerceToDouble(PgBenchValue *pval, double *dval)
1284 : {
1285 44 : if (pval->type == PGBT_DOUBLE)
1286 : {
1287 33 : *dval = pval->u.dval;
1288 33 : return true;
1289 : }
1290 : else
1291 : {
1292 11 : Assert(pval->type == PGBT_INT);
1293 11 : *dval = (double) pval->u.ival;
1294 11 : return true;
1295 : }
1296 : }
1297 :
1298 : /* assign an integer value */
1299 : static void
1300 3251 : setIntValue(PgBenchValue *pv, int64 ival)
1301 : {
1302 3251 : pv->type = PGBT_INT;
1303 3251 : pv->u.ival = ival;
1304 3251 : }
1305 :
1306 : /* assign a double value */
1307 : static void
1308 23 : setDoubleValue(PgBenchValue *pv, double dval)
1309 : {
1310 23 : pv->type = PGBT_DOUBLE;
1311 23 : pv->u.dval = dval;
1312 23 : }
1313 :
1314 : /* maximum number of function arguments */
1315 : #define MAX_FARGS 16
1316 :
1317 : /*
1318 : * Recursive evaluation of functions
1319 : */
1320 : static bool
1321 3200 : evalFunc(TState *thread, CState *st,
1322 : PgBenchFunction func, PgBenchExprLink *args, PgBenchValue *retval)
1323 : {
1324 : /* evaluate all function arguments */
1325 3200 : int nargs = 0;
1326 : PgBenchValue vargs[MAX_FARGS];
1327 3200 : PgBenchExprLink *l = args;
1328 :
1329 9592 : for (nargs = 0; nargs < MAX_FARGS && l != NULL; nargs++, l = l->next)
1330 6393 : if (!evaluateExpr(thread, st, l->expr, &vargs[nargs]))
1331 1 : return false;
1332 :
1333 3199 : if (l != NULL)
1334 : {
1335 1 : fprintf(stderr,
1336 : "too many function arguments, maximum is %d\n", MAX_FARGS);
1337 1 : return false;
1338 : }
1339 :
1340 : /* then evaluate function */
1341 3198 : switch (func)
1342 : {
1343 : /* overloaded operators */
1344 : case PGBENCH_ADD:
1345 : case PGBENCH_SUB:
1346 : case PGBENCH_MUL:
1347 : case PGBENCH_DIV:
1348 : case PGBENCH_MOD:
1349 : {
1350 1602 : PgBenchValue *lval = &vargs[0],
1351 1602 : *rval = &vargs[1];
1352 :
1353 1602 : Assert(nargs == 2);
1354 :
1355 : /* overloaded type management, double if some double */
1356 3192 : if ((lval->type == PGBT_DOUBLE ||
1357 1606 : rval->type == PGBT_DOUBLE) && func != PGBENCH_MOD)
1358 : {
1359 : double ld,
1360 : rd;
1361 :
1362 32 : if (!coerceToDouble(lval, &ld) ||
1363 16 : !coerceToDouble(rval, &rd))
1364 0 : return false;
1365 :
1366 16 : switch (func)
1367 : {
1368 : case PGBENCH_ADD:
1369 1 : setDoubleValue(retval, ld + rd);
1370 1 : return true;
1371 :
1372 : case PGBENCH_SUB:
1373 5 : setDoubleValue(retval, ld - rd);
1374 5 : return true;
1375 :
1376 : case PGBENCH_MUL:
1377 8 : setDoubleValue(retval, ld * rd);
1378 8 : return true;
1379 :
1380 : case PGBENCH_DIV:
1381 2 : setDoubleValue(retval, ld / rd);
1382 2 : return true;
1383 :
1384 : default:
1385 : /* cannot get here */
1386 0 : Assert(0);
1387 : }
1388 : }
1389 : else /* we have integer operands, or % */
1390 : {
1391 : int64 li,
1392 : ri;
1393 :
1394 3172 : if (!coerceToInt(lval, &li) ||
1395 1586 : !coerceToInt(rval, &ri))
1396 0 : return false;
1397 :
1398 1586 : switch (func)
1399 : {
1400 : case PGBENCH_ADD:
1401 7 : setIntValue(retval, li + ri);
1402 7 : return true;
1403 :
1404 : case PGBENCH_SUB:
1405 124 : setIntValue(retval, li - ri);
1406 124 : return true;
1407 :
1408 : case PGBENCH_MUL:
1409 1448 : setIntValue(retval, li * ri);
1410 1448 : return true;
1411 :
1412 : case PGBENCH_DIV:
1413 : case PGBENCH_MOD:
1414 7 : if (ri == 0)
1415 : {
1416 1 : fprintf(stderr, "division by zero\n");
1417 1 : return false;
1418 : }
1419 : /* special handling of -1 divisor */
1420 6 : if (ri == -1)
1421 : {
1422 3 : if (func == PGBENCH_DIV)
1423 : {
1424 : /* overflow check (needed for INT64_MIN) */
1425 2 : if (li == PG_INT64_MIN)
1426 : {
1427 1 : fprintf(stderr, "bigint out of range\n");
1428 1 : return false;
1429 : }
1430 : else
1431 1 : setIntValue(retval, -li);
1432 : }
1433 : else
1434 1 : setIntValue(retval, 0);
1435 2 : return true;
1436 : }
1437 : /* else divisor is not -1 */
1438 3 : if (func == PGBENCH_DIV)
1439 2 : setIntValue(retval, li / ri);
1440 : else /* func == PGBENCH_MOD */
1441 1 : setIntValue(retval, li % ri);
1442 :
1443 3 : return true;
1444 :
1445 : default:
1446 : /* cannot get here */
1447 0 : Assert(0);
1448 : }
1449 : }
1450 : }
1451 :
1452 : /* no arguments */
1453 : case PGBENCH_PI:
1454 1 : setDoubleValue(retval, M_PI);
1455 1 : return true;
1456 :
1457 : /* 1 overloaded argument */
1458 : case PGBENCH_ABS:
1459 : {
1460 2 : PgBenchValue *varg = &vargs[0];
1461 :
1462 2 : Assert(nargs == 1);
1463 :
1464 2 : if (varg->type == PGBT_INT)
1465 : {
1466 1 : int64 i = varg->u.ival;
1467 :
1468 1 : setIntValue(retval, i < 0 ? -i : i);
1469 : }
1470 : else
1471 : {
1472 1 : double d = varg->u.dval;
1473 :
1474 1 : Assert(varg->type == PGBT_DOUBLE);
1475 1 : setDoubleValue(retval, d < 0.0 ? -d : d);
1476 : }
1477 :
1478 2 : return true;
1479 : }
1480 :
1481 : case PGBENCH_DEBUG:
1482 : {
1483 22 : PgBenchValue *varg = &vargs[0];
1484 :
1485 22 : Assert(nargs == 1);
1486 :
1487 22 : fprintf(stderr, "debug(script=%d,command=%d): ",
1488 22 : st->use_file, st->command + 1);
1489 :
1490 22 : if (varg->type == PGBT_INT)
1491 14 : fprintf(stderr, "int " INT64_FORMAT "\n", varg->u.ival);
1492 : else
1493 : {
1494 8 : Assert(varg->type == PGBT_DOUBLE);
1495 8 : fprintf(stderr, "double %.*g\n", DBL_DIG, varg->u.dval);
1496 : }
1497 :
1498 22 : *retval = *varg;
1499 :
1500 22 : return true;
1501 : }
1502 :
1503 : /* 1 double argument */
1504 : case PGBENCH_DOUBLE:
1505 : case PGBENCH_SQRT:
1506 : {
1507 : double dval;
1508 :
1509 2 : Assert(nargs == 1);
1510 :
1511 2 : if (!coerceToDouble(&vargs[0], &dval))
1512 0 : return false;
1513 :
1514 2 : if (func == PGBENCH_SQRT)
1515 1 : dval = sqrt(dval);
1516 :
1517 2 : setDoubleValue(retval, dval);
1518 2 : return true;
1519 : }
1520 :
1521 : /* 1 int argument */
1522 : case PGBENCH_INT:
1523 : {
1524 : int64 ival;
1525 :
1526 2 : Assert(nargs == 1);
1527 :
1528 2 : if (!coerceToInt(&vargs[0], &ival))
1529 1 : return false;
1530 :
1531 1 : setIntValue(retval, ival);
1532 1 : return true;
1533 : }
1534 :
1535 : /* variable number of arguments */
1536 : case PGBENCH_LEAST:
1537 : case PGBENCH_GREATEST:
1538 : {
1539 : bool havedouble;
1540 : int i;
1541 :
1542 4 : Assert(nargs >= 1);
1543 :
1544 : /* need double result if any input is double */
1545 4 : havedouble = false;
1546 14 : for (i = 0; i < nargs; i++)
1547 : {
1548 12 : if (vargs[i].type == PGBT_DOUBLE)
1549 : {
1550 2 : havedouble = true;
1551 2 : break;
1552 : }
1553 : }
1554 4 : if (havedouble)
1555 : {
1556 : double extremum;
1557 :
1558 2 : if (!coerceToDouble(&vargs[0], &extremum))
1559 0 : return false;
1560 6 : for (i = 1; i < nargs; i++)
1561 : {
1562 : double dval;
1563 :
1564 4 : if (!coerceToDouble(&vargs[i], &dval))
1565 0 : return false;
1566 4 : if (func == PGBENCH_LEAST)
1567 2 : extremum = Min(extremum, dval);
1568 : else
1569 2 : extremum = Max(extremum, dval);
1570 : }
1571 2 : setDoubleValue(retval, extremum);
1572 : }
1573 : else
1574 : {
1575 : int64 extremum;
1576 :
1577 2 : if (!coerceToInt(&vargs[0], &extremum))
1578 0 : return false;
1579 8 : for (i = 1; i < nargs; i++)
1580 : {
1581 : int64 ival;
1582 :
1583 6 : if (!coerceToInt(&vargs[i], &ival))
1584 0 : return false;
1585 6 : if (func == PGBENCH_LEAST)
1586 3 : extremum = Min(extremum, ival);
1587 : else
1588 3 : extremum = Max(extremum, ival);
1589 : }
1590 2 : setIntValue(retval, extremum);
1591 : }
1592 4 : return true;
1593 : }
1594 :
1595 : /* random functions */
1596 : case PGBENCH_RANDOM:
1597 : case PGBENCH_RANDOM_EXPONENTIAL:
1598 : case PGBENCH_RANDOM_GAUSSIAN:
1599 : {
1600 : int64 imin,
1601 : imax;
1602 :
1603 1563 : Assert(nargs >= 2);
1604 :
1605 3126 : if (!coerceToInt(&vargs[0], &imin) ||
1606 1563 : !coerceToInt(&vargs[1], &imax))
1607 0 : return false;
1608 :
1609 : /* check random range */
1610 1563 : if (imin > imax)
1611 : {
1612 1 : fprintf(stderr, "empty range given to random\n");
1613 1 : return false;
1614 : }
1615 1562 : else if (imax - imin < 0 || (imax - imin) + 1 < 0)
1616 : {
1617 : /* prevent int overflows in random functions */
1618 1 : fprintf(stderr, "random range is too large\n");
1619 1 : return false;
1620 : }
1621 :
1622 1561 : if (func == PGBENCH_RANDOM)
1623 : {
1624 1557 : Assert(nargs == 2);
1625 1557 : setIntValue(retval, getrand(thread, imin, imax));
1626 : }
1627 : else /* gaussian & exponential */
1628 : {
1629 : double param;
1630 :
1631 4 : Assert(nargs == 3);
1632 :
1633 4 : if (!coerceToDouble(&vargs[2], ¶m))
1634 2 : return false;
1635 :
1636 4 : if (func == PGBENCH_RANDOM_GAUSSIAN)
1637 : {
1638 2 : if (param < MIN_GAUSSIAN_PARAM)
1639 : {
1640 1 : fprintf(stderr,
1641 : "gaussian parameter must be at least %f "
1642 : "(not %f)\n", MIN_GAUSSIAN_PARAM, param);
1643 1 : return false;
1644 : }
1645 :
1646 1 : setIntValue(retval,
1647 : getGaussianRand(thread, imin, imax, param));
1648 : }
1649 : else /* exponential */
1650 : {
1651 2 : if (param <= 0.0)
1652 : {
1653 1 : fprintf(stderr,
1654 : "exponential parameter must be greater than zero"
1655 : " (got %f)\n", param);
1656 1 : return false;
1657 : }
1658 :
1659 1 : setIntValue(retval,
1660 : getExponentialRand(thread, imin, imax, param));
1661 : }
1662 : }
1663 :
1664 1559 : return true;
1665 : }
1666 :
1667 : default:
1668 : /* cannot get here */
1669 0 : Assert(0);
1670 : /* dead code to avoid a compiler warning */
1671 : return false;
1672 : }
1673 : }
1674 :
1675 : /*
1676 : * Recursive evaluation of an expression in a pgbench script
1677 : * using the current state of variables.
1678 : * Returns whether the evaluation was ok,
1679 : * the value itself is returned through the retval pointer.
1680 : */
1681 : static bool
1682 8049 : evaluateExpr(TState *thread, CState *st, PgBenchExpr *expr, PgBenchValue *retval)
1683 : {
1684 8049 : switch (expr->etype)
1685 : {
1686 : case ENODE_CONSTANT:
1687 : {
1688 3396 : *retval = expr->u.constant;
1689 3396 : return true;
1690 : }
1691 :
1692 : case ENODE_VARIABLE:
1693 : {
1694 : Variable *var;
1695 :
1696 1453 : if ((var = lookupVariable(st, expr->u.variable.varname)) == NULL)
1697 : {
1698 1 : fprintf(stderr, "undefined variable \"%s\"\n",
1699 : expr->u.variable.varname);
1700 1 : return false;
1701 : }
1702 :
1703 1452 : if (!makeVariableNumeric(var))
1704 1 : return false;
1705 :
1706 1451 : *retval = var->num_value;
1707 1451 : return true;
1708 : }
1709 :
1710 : case ENODE_FUNCTION:
1711 3200 : return evalFunc(thread, st,
1712 : expr->u.function.function,
1713 : expr->u.function.args,
1714 : retval);
1715 :
1716 : default:
1717 : /* internal error which should never occur */
1718 0 : fprintf(stderr, "unexpected enode type in evaluation: %d\n",
1719 0 : expr->etype);
1720 0 : exit(1);
1721 : }
1722 : }
1723 :
1724 : /*
1725 : * Run a shell command. The result is assigned to the variable if not NULL.
1726 : * Return true if succeeded, or false on error.
1727 : */
1728 : static bool
1729 6 : runShellCommand(CState *st, char *variable, char **argv, int argc)
1730 : {
1731 : char command[SHELL_COMMAND_SIZE];
1732 : int i,
1733 6 : len = 0;
1734 : FILE *fp;
1735 : char res[64];
1736 : char *endptr;
1737 : int retval;
1738 :
1739 : /*----------
1740 : * Join arguments with whitespace separators. Arguments starting with
1741 : * exactly one colon are treated as variables:
1742 : * name - append a string "name"
1743 : * :var - append a variable named 'var'
1744 : * ::name - append a string ":name"
1745 : *----------
1746 : */
1747 19 : for (i = 0; i < argc; i++)
1748 : {
1749 : char *arg;
1750 : int arglen;
1751 :
1752 14 : if (argv[i][0] != ':')
1753 : {
1754 11 : arg = argv[i]; /* a string literal */
1755 : }
1756 3 : else if (argv[i][1] == ':')
1757 : {
1758 1 : arg = argv[i] + 1; /* a string literal starting with colons */
1759 : }
1760 2 : else if ((arg = getVariable(st, argv[i] + 1)) == NULL)
1761 : {
1762 1 : fprintf(stderr, "%s: undefined variable \"%s\"\n",
1763 1 : argv[0], argv[i]);
1764 1 : return false;
1765 : }
1766 :
1767 13 : arglen = strlen(arg);
1768 13 : if (len + arglen + (i > 0 ? 1 : 0) >= SHELL_COMMAND_SIZE - 1)
1769 : {
1770 0 : fprintf(stderr, "%s: shell command is too long\n", argv[0]);
1771 0 : return false;
1772 : }
1773 :
1774 13 : if (i > 0)
1775 7 : command[len++] = ' ';
1776 13 : memcpy(command + len, arg, arglen);
1777 13 : len += arglen;
1778 : }
1779 :
1780 5 : command[len] = '\0';
1781 :
1782 : /* Fast path for non-assignment case */
1783 5 : if (variable == NULL)
1784 : {
1785 2 : if (system(command))
1786 : {
1787 1 : if (!timer_exceeded)
1788 1 : fprintf(stderr, "%s: could not launch shell command\n", argv[0]);
1789 1 : return false;
1790 : }
1791 1 : return true;
1792 : }
1793 :
1794 : /* Execute the command with pipe and read the standard output. */
1795 3 : if ((fp = popen(command, "r")) == NULL)
1796 : {
1797 0 : fprintf(stderr, "%s: could not launch shell command\n", argv[0]);
1798 0 : return false;
1799 : }
1800 3 : if (fgets(res, sizeof(res), fp) == NULL)
1801 : {
1802 1 : if (!timer_exceeded)
1803 1 : fprintf(stderr, "%s: could not read result of shell command\n", argv[0]);
1804 1 : (void) pclose(fp);
1805 1 : return false;
1806 : }
1807 2 : if (pclose(fp) < 0)
1808 : {
1809 0 : fprintf(stderr, "%s: could not close shell command\n", argv[0]);
1810 0 : return false;
1811 : }
1812 :
1813 : /* Check whether the result is an integer and assign it to the variable */
1814 2 : retval = (int) strtol(res, &endptr, 10);
1815 5 : while (*endptr != '\0' && isspace((unsigned char) *endptr))
1816 1 : endptr++;
1817 2 : if (*res == '\0' || *endptr != '\0')
1818 : {
1819 1 : fprintf(stderr, "%s: shell command must return an integer (not \"%s\")\n",
1820 : argv[0], res);
1821 1 : return false;
1822 : }
1823 1 : if (!putVariableInt(st, "setshell", variable, retval))
1824 0 : return false;
1825 :
1826 : #ifdef DEBUG
1827 : printf("shell parameter name: \"%s\", value: \"%s\"\n", argv[1], res);
1828 : #endif
1829 1 : return true;
1830 : }
1831 :
1832 : #define MAX_PREPARE_NAME 32
1833 : static void
1834 1071 : preparedStatementName(char *buffer, int file, int state)
1835 : {
1836 1071 : sprintf(buffer, "P%d_%d", file, state);
1837 1071 : }
1838 :
1839 : static void
1840 17 : commandFailed(CState *st, char *message)
1841 : {
1842 17 : fprintf(stderr,
1843 : "client %d aborted in command %d of script %d; %s\n",
1844 : st->id, st->command, st->use_file, message);
1845 17 : }
1846 :
1847 : /* return a script number with a weighted choice. */
1848 : static int
1849 1438 : chooseScript(TState *thread)
1850 : {
1851 1438 : int i = 0;
1852 : int64 w;
1853 :
1854 1438 : if (num_scripts == 1)
1855 1295 : return 0;
1856 :
1857 143 : w = getrand(thread, 0, total_weight - 1);
1858 : do
1859 : {
1860 236 : w -= sql_script[i++].weight;
1861 236 : } while (w >= 0);
1862 :
1863 143 : return i - 1;
1864 : }
1865 :
1866 : /* Send a SQL command, using the chosen querymode */
1867 : static bool
1868 2064 : sendCommand(CState *st, Command *command)
1869 : {
1870 : int r;
1871 :
1872 2064 : if (querymode == QUERY_SIMPLE)
1873 : {
1874 : char *sql;
1875 :
1876 450 : sql = pg_strdup(command->argv[0]);
1877 450 : sql = assignVariables(st, sql);
1878 :
1879 450 : if (debug)
1880 0 : fprintf(stderr, "client %d sending %s\n", st->id, sql);
1881 450 : r = PQsendQuery(st->con, sql);
1882 450 : free(sql);
1883 : }
1884 1614 : else if (querymode == QUERY_EXTENDED)
1885 : {
1886 560 : const char *sql = command->argv[0];
1887 : const char *params[MAX_ARGS];
1888 :
1889 560 : getQueryParams(st, command, params);
1890 :
1891 560 : if (debug)
1892 0 : fprintf(stderr, "client %d sending %s\n", st->id, sql);
1893 560 : r = PQsendQueryParams(st->con, sql, command->argc - 1,
1894 : NULL, params, NULL, NULL, 0);
1895 : }
1896 1054 : else if (querymode == QUERY_PREPARED)
1897 : {
1898 : char name[MAX_PREPARE_NAME];
1899 : const char *params[MAX_ARGS];
1900 :
1901 1054 : if (!st->prepared[st->use_file])
1902 : {
1903 : int j;
1904 15 : Command **commands = sql_script[st->use_file].commands;
1905 :
1906 41 : for (j = 0; commands[j] != NULL; j++)
1907 : {
1908 : PGresult *res;
1909 : char name[MAX_PREPARE_NAME];
1910 :
1911 26 : if (commands[j]->type != SQL_COMMAND)
1912 9 : continue;
1913 17 : preparedStatementName(name, st->use_file, j);
1914 34 : res = PQprepare(st->con, name,
1915 34 : commands[j]->argv[0], commands[j]->argc - 1, NULL);
1916 17 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1917 1 : fprintf(stderr, "%s", PQerrorMessage(st->con));
1918 17 : PQclear(res);
1919 : }
1920 15 : st->prepared[st->use_file] = true;
1921 : }
1922 :
1923 1054 : getQueryParams(st, command, params);
1924 1054 : preparedStatementName(name, st->use_file, st->command);
1925 :
1926 1054 : if (debug)
1927 700 : fprintf(stderr, "client %d sending %s\n", st->id, name);
1928 1054 : r = PQsendQueryPrepared(st->con, name, command->argc - 1,
1929 : params, NULL, NULL, 0);
1930 : }
1931 : else /* unknown sql mode */
1932 0 : r = 0;
1933 :
1934 2064 : if (r == 0)
1935 : {
1936 0 : if (debug)
1937 0 : fprintf(stderr, "client %d could not send %s\n",
1938 : st->id, command->argv[0]);
1939 0 : st->ecnt++;
1940 0 : return false;
1941 : }
1942 : else
1943 2064 : return true;
1944 : }
1945 :
1946 : /*
1947 : * Parse the argument to a \sleep command, and return the requested amount
1948 : * of delay, in microseconds. Returns true on success, false on error.
1949 : */
1950 : static bool
1951 5 : evaluateSleep(CState *st, int argc, char **argv, int *usecs)
1952 : {
1953 : char *var;
1954 : int usec;
1955 :
1956 5 : if (*argv[1] == ':')
1957 : {
1958 3 : if ((var = getVariable(st, argv[1] + 1)) == NULL)
1959 : {
1960 1 : fprintf(stderr, "%s: undefined variable \"%s\"\n",
1961 1 : argv[0], argv[1]);
1962 1 : return false;
1963 : }
1964 2 : usec = atoi(var);
1965 : }
1966 : else
1967 2 : usec = atoi(argv[1]);
1968 :
1969 4 : if (argc > 2)
1970 : {
1971 3 : if (pg_strcasecmp(argv[2], "ms") == 0)
1972 1 : usec *= 1000;
1973 2 : else if (pg_strcasecmp(argv[2], "s") == 0)
1974 1 : usec *= 1000000;
1975 : }
1976 : else
1977 1 : usec *= 1000000;
1978 :
1979 4 : *usecs = usec;
1980 4 : return true;
1981 : }
1982 :
1983 : /*
1984 : * Advance the state machine of a connection, if possible.
1985 : */
1986 : static void
1987 7232709 : doCustom(TState *thread, CState *st, StatsData *agg)
1988 : {
1989 : PGresult *res;
1990 : Command *command;
1991 : instr_time now;
1992 7232709 : bool end_tx_processed = false;
1993 : int64 wait;
1994 :
1995 : /*
1996 : * gettimeofday() isn't free, so we get the current timestamp lazily the
1997 : * first time it's needed, and reuse the same value throughout this
1998 : * function after that. This also ensures that e.g. the calculated
1999 : * latency reported in the log file and in the totals are the same. Zero
2000 : * means "not set yet". Reset "now" when we execute shell commands or
2001 : * expressions, which might take a non-negligible amount of time, though.
2002 : */
2003 7232709 : INSTR_TIME_SET_ZERO(now);
2004 :
2005 : /*
2006 : * Loop in the state machine, until we have to wait for a result from the
2007 : * server (or have to sleep, for throttling or for \sleep).
2008 : *
2009 : * Note: In the switch-statement below, 'break' will loop back here,
2010 : * meaning "continue in the state machine". Return is used to return to
2011 : * the caller.
2012 : */
2013 : for (;;)
2014 : {
2015 7248412 : switch (st->state)
2016 : {
2017 : /*
2018 : * Select transaction to run.
2019 : */
2020 : case CSTATE_CHOOSE_SCRIPT:
2021 :
2022 1438 : st->use_file = chooseScript(thread);
2023 :
2024 1438 : if (debug)
2025 700 : fprintf(stderr, "client %d executing script \"%s\"\n", st->id,
2026 700 : sql_script[st->use_file].desc);
2027 :
2028 1438 : if (throttle_delay > 0)
2029 244 : st->state = CSTATE_START_THROTTLE;
2030 : else
2031 1194 : st->state = CSTATE_START_TX;
2032 1438 : break;
2033 :
2034 : /*
2035 : * Handle throttling once per transaction by sleeping.
2036 : */
2037 : case CSTATE_START_THROTTLE:
2038 :
2039 : /*
2040 : * Generate a delay such that the series of delays will
2041 : * approximate a Poisson distribution centered on the
2042 : * throttle_delay time.
2043 : *
2044 : * If transactions are too slow or a given wait is shorter
2045 : * than a transaction, the next transaction will start right
2046 : * away.
2047 : */
2048 244 : Assert(throttle_delay > 0);
2049 244 : wait = getPoissonRand(thread, throttle_delay);
2050 :
2051 244 : thread->throttle_trigger += wait;
2052 244 : st->txn_scheduled = thread->throttle_trigger;
2053 :
2054 : /*
2055 : * stop client if next transaction is beyond pgbench end of
2056 : * execution
2057 : */
2058 244 : if (duration > 0 && st->txn_scheduled > end_time)
2059 : {
2060 3 : st->state = CSTATE_FINISHED;
2061 3 : break;
2062 : }
2063 :
2064 : /*
2065 : * If --latency-limit is used, and this slot is already late
2066 : * so that the transaction will miss the latency limit even if
2067 : * it completed immediately, we skip this time slot and
2068 : * iterate till the next slot that isn't late yet. But don't
2069 : * iterate beyond the -t limit, if one is given.
2070 : */
2071 241 : if (latency_limit)
2072 : {
2073 : int64 now_us;
2074 :
2075 241 : if (INSTR_TIME_IS_ZERO(now))
2076 6 : INSTR_TIME_SET_CURRENT(now);
2077 241 : now_us = INSTR_TIME_GET_MICROSEC(now);
2078 503 : while (thread->throttle_trigger < now_us - latency_limit &&
2079 22 : (nxacts <= 0 || st->cnt < nxacts))
2080 : {
2081 10 : processXactStats(thread, st, &now, true, agg);
2082 : /* next rendez-vous */
2083 10 : wait = getPoissonRand(thread, throttle_delay);
2084 10 : thread->throttle_trigger += wait;
2085 10 : st->txn_scheduled = thread->throttle_trigger;
2086 : }
2087 : /* stop client if -t exceeded */
2088 241 : if (nxacts > 0 && st->cnt >= nxacts)
2089 : {
2090 1 : st->state = CSTATE_FINISHED;
2091 1 : break;
2092 : }
2093 : }
2094 :
2095 240 : st->state = CSTATE_THROTTLE;
2096 240 : if (debug)
2097 0 : fprintf(stderr, "client %d throttling " INT64_FORMAT " us\n",
2098 : st->id, wait);
2099 240 : break;
2100 :
2101 : /*
2102 : * Wait until it's time to start next transaction.
2103 : */
2104 : case CSTATE_THROTTLE:
2105 7226522 : if (INSTR_TIME_IS_ZERO(now))
2106 7209840 : INSTR_TIME_SET_CURRENT(now);
2107 7226522 : if (INSTR_TIME_GET_MICROSEC(now) < st->txn_scheduled)
2108 7226282 : return; /* Still sleeping, nothing to do here */
2109 :
2110 : /* Else done sleeping, start the transaction */
2111 240 : st->state = CSTATE_START_TX;
2112 240 : break;
2113 :
2114 : /* Start new transaction */
2115 : case CSTATE_START_TX:
2116 :
2117 : /*
2118 : * Establish connection on first call, or if is_connect is
2119 : * true.
2120 : */
2121 1434 : if (st->con == NULL)
2122 : {
2123 : instr_time start;
2124 :
2125 110 : if (INSTR_TIME_IS_ZERO(now))
2126 110 : INSTR_TIME_SET_CURRENT(now);
2127 110 : start = now;
2128 110 : if ((st->con = doConnect()) == NULL)
2129 : {
2130 0 : fprintf(stderr, "client %d aborted while establishing connection\n",
2131 : st->id);
2132 0 : st->state = CSTATE_ABORTED;
2133 0 : break;
2134 : }
2135 110 : INSTR_TIME_SET_CURRENT(now);
2136 110 : INSTR_TIME_ACCUM_DIFF(thread->conn_time, now, start);
2137 :
2138 : /* Reset session-local state */
2139 110 : memset(st->prepared, 0, sizeof(st->prepared));
2140 : }
2141 :
2142 : /*
2143 : * Record transaction start time under logging, progress or
2144 : * throttling.
2145 : */
2146 1434 : if (use_log || progress || throttle_delay || latency_limit ||
2147 : per_script_stats)
2148 : {
2149 450 : if (INSTR_TIME_IS_ZERO(now))
2150 4 : INSTR_TIME_SET_CURRENT(now);
2151 450 : st->txn_begin = now;
2152 :
2153 : /*
2154 : * When not throttling, this is also the transaction's
2155 : * scheduled start time.
2156 : */
2157 450 : if (!throttle_delay)
2158 210 : st->txn_scheduled = INSTR_TIME_GET_MICROSEC(now);
2159 : }
2160 :
2161 : /* Begin with the first command */
2162 1434 : st->command = 0;
2163 1434 : st->state = CSTATE_START_COMMAND;
2164 1434 : break;
2165 :
2166 : /*
2167 : * Send a command to server (or execute a meta-command)
2168 : */
2169 : case CSTATE_START_COMMAND:
2170 5148 : command = sql_script[st->use_file].commands[st->command];
2171 :
2172 : /*
2173 : * If we reached the end of the script, move to end-of-xact
2174 : * processing.
2175 : */
2176 5148 : if (command == NULL)
2177 : {
2178 1417 : st->state = CSTATE_END_TX;
2179 1417 : break;
2180 : }
2181 :
2182 : /*
2183 : * Record statement start time if per-command latencies are
2184 : * requested
2185 : */
2186 3731 : if (is_latencies)
2187 : {
2188 480 : if (INSTR_TIME_IS_ZERO(now))
2189 0 : INSTR_TIME_SET_CURRENT(now);
2190 480 : st->stmt_begin = now;
2191 : }
2192 :
2193 3731 : if (command->type == SQL_COMMAND)
2194 : {
2195 2064 : if (!sendCommand(st, command))
2196 : {
2197 : /*
2198 : * Failed. Stay in CSTATE_START_COMMAND state, to
2199 : * retry. ??? What the point or retrying? Should
2200 : * rather abort?
2201 : */
2202 0 : return;
2203 : }
2204 : else
2205 2064 : st->state = CSTATE_WAIT_RESULT;
2206 : }
2207 1667 : else if (command->type == META_COMMAND)
2208 : {
2209 1667 : int argc = command->argc,
2210 : i;
2211 1667 : char **argv = command->argv;
2212 :
2213 1667 : if (debug)
2214 : {
2215 700 : fprintf(stderr, "client %d executing \\%s", st->id, argv[0]);
2216 1400 : for (i = 1; i < argc; i++)
2217 700 : fprintf(stderr, " %s", argv[i]);
2218 700 : fprintf(stderr, "\n");
2219 : }
2220 :
2221 1667 : if (pg_strcasecmp(argv[0], "sleep") == 0)
2222 : {
2223 : /*
2224 : * A \sleep doesn't execute anything, we just get the
2225 : * delay from the argument, and enter the CSTATE_SLEEP
2226 : * state. (The per-command latency will be recorded
2227 : * in CSTATE_SLEEP state, not here, after the delay
2228 : * has elapsed.)
2229 : */
2230 : int usec;
2231 :
2232 5 : if (!evaluateSleep(st, argc, argv, &usec))
2233 : {
2234 1 : commandFailed(st, "execution of meta-command 'sleep' failed");
2235 1 : st->state = CSTATE_ABORTED;
2236 1 : break;
2237 : }
2238 :
2239 4 : if (INSTR_TIME_IS_ZERO(now))
2240 1 : INSTR_TIME_SET_CURRENT(now);
2241 4 : st->sleep_until = INSTR_TIME_GET_MICROSEC(now) + usec;
2242 4 : st->state = CSTATE_SLEEP;
2243 4 : break;
2244 : }
2245 : else
2246 : {
2247 1662 : if (pg_strcasecmp(argv[0], "set") == 0)
2248 : {
2249 1656 : PgBenchExpr *expr = command->expr;
2250 : PgBenchValue result;
2251 :
2252 1656 : if (!evaluateExpr(thread, st, expr, &result))
2253 : {
2254 10 : commandFailed(st, "evaluation of meta-command 'set' failed");
2255 10 : st->state = CSTATE_ABORTED;
2256 21 : break;
2257 : }
2258 :
2259 1646 : if (!putVariableNumber(st, argv[0], argv[1], &result))
2260 : {
2261 1 : commandFailed(st, "assignment of meta-command 'set' failed");
2262 1 : st->state = CSTATE_ABORTED;
2263 1 : break;
2264 : }
2265 : }
2266 6 : else if (pg_strcasecmp(argv[0], "setshell") == 0)
2267 : {
2268 3 : bool ret = runShellCommand(st, argv[1], argv + 2, argc - 2);
2269 :
2270 3 : if (timer_exceeded) /* timeout */
2271 : {
2272 0 : st->state = CSTATE_FINISHED;
2273 0 : break;
2274 : }
2275 3 : else if (!ret) /* on error */
2276 : {
2277 2 : commandFailed(st, "execution of meta-command 'setshell' failed");
2278 2 : st->state = CSTATE_ABORTED;
2279 2 : break;
2280 : }
2281 : else
2282 : {
2283 : /* succeeded */
2284 : }
2285 : }
2286 3 : else if (pg_strcasecmp(argv[0], "shell") == 0)
2287 : {
2288 3 : bool ret = runShellCommand(st, NULL, argv + 1, argc - 1);
2289 :
2290 3 : if (timer_exceeded) /* timeout */
2291 : {
2292 0 : st->state = CSTATE_FINISHED;
2293 0 : break;
2294 : }
2295 3 : else if (!ret) /* on error */
2296 : {
2297 2 : commandFailed(st, "execution of meta-command 'shell' failed");
2298 2 : st->state = CSTATE_ABORTED;
2299 2 : break;
2300 : }
2301 : else
2302 : {
2303 : /* succeeded */
2304 : }
2305 : }
2306 :
2307 : /*
2308 : * executing the expression or shell command might
2309 : * take a non-negligible amount of time, so reset
2310 : * 'now'
2311 : */
2312 1647 : INSTR_TIME_SET_ZERO(now);
2313 :
2314 1647 : st->state = CSTATE_END_COMMAND;
2315 : }
2316 : }
2317 3711 : break;
2318 :
2319 : /*
2320 : * Wait for the current SQL command to complete
2321 : */
2322 : case CSTATE_WAIT_RESULT:
2323 4089 : command = sql_script[st->use_file].commands[st->command];
2324 4089 : if (debug)
2325 1400 : fprintf(stderr, "client %d receiving\n", st->id);
2326 4089 : if (!PQconsumeInput(st->con))
2327 : { /* there's something wrong */
2328 0 : commandFailed(st, "perhaps the backend died while processing");
2329 0 : st->state = CSTATE_ABORTED;
2330 0 : break;
2331 : }
2332 4089 : if (PQisBusy(st->con))
2333 2025 : return; /* don't have the whole result yet */
2334 :
2335 : /*
2336 : * Read and discard the query result;
2337 : */
2338 2064 : res = PQgetResult(st->con);
2339 2064 : switch (PQresultStatus(res))
2340 : {
2341 : case PGRES_COMMAND_OK:
2342 : case PGRES_TUPLES_OK:
2343 : case PGRES_EMPTY_QUERY:
2344 : /* OK */
2345 2063 : PQclear(res);
2346 2063 : discard_response(st);
2347 2063 : st->state = CSTATE_END_COMMAND;
2348 2063 : break;
2349 : default:
2350 1 : commandFailed(st, PQerrorMessage(st->con));
2351 1 : PQclear(res);
2352 1 : st->state = CSTATE_ABORTED;
2353 1 : break;
2354 : }
2355 2064 : break;
2356 :
2357 : /*
2358 : * Wait until sleep is done. This state is entered after a
2359 : * \sleep metacommand. The behavior is similar to
2360 : * CSTATE_THROTTLE, but proceeds to CSTATE_START_COMMAND
2361 : * instead of CSTATE_START_TX.
2362 : */
2363 : case CSTATE_SLEEP:
2364 4355 : if (INSTR_TIME_IS_ZERO(now))
2365 4351 : INSTR_TIME_SET_CURRENT(now);
2366 4355 : if (INSTR_TIME_GET_MICROSEC(now) < st->sleep_until)
2367 4351 : return; /* Still sleeping, nothing to do here */
2368 : /* Else done sleeping. */
2369 4 : st->state = CSTATE_END_COMMAND;
2370 4 : break;
2371 :
2372 : /*
2373 : * End of command: record stats and proceed to next command.
2374 : */
2375 : case CSTATE_END_COMMAND:
2376 :
2377 : /*
2378 : * command completed: accumulate per-command execution times
2379 : * in thread-local data structure, if per-command latencies
2380 : * are requested.
2381 : */
2382 3714 : if (is_latencies)
2383 : {
2384 480 : if (INSTR_TIME_IS_ZERO(now))
2385 441 : INSTR_TIME_SET_CURRENT(now);
2386 :
2387 : /* XXX could use a mutex here, but we choose not to */
2388 480 : command = sql_script[st->use_file].commands[st->command];
2389 480 : addToSimpleStats(&command->stats,
2390 480 : INSTR_TIME_GET_DOUBLE(now) -
2391 480 : INSTR_TIME_GET_DOUBLE(st->stmt_begin));
2392 : }
2393 :
2394 : /* Go ahead with next command */
2395 3714 : st->command++;
2396 3714 : st->state = CSTATE_START_COMMAND;
2397 3714 : break;
2398 :
2399 : /*
2400 : * End of transaction.
2401 : */
2402 : case CSTATE_END_TX:
2403 :
2404 : /* transaction finished: calculate latency and do log */
2405 1417 : processXactStats(thread, st, &now, false, agg);
2406 :
2407 1417 : if (is_connect)
2408 : {
2409 110 : PQfinish(st->con);
2410 110 : st->con = NULL;
2411 110 : INSTR_TIME_SET_ZERO(now);
2412 : }
2413 :
2414 1417 : if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
2415 : {
2416 : /* exit success */
2417 30 : st->state = CSTATE_FINISHED;
2418 30 : break;
2419 : }
2420 :
2421 : /*
2422 : * No transaction is underway anymore.
2423 : */
2424 1387 : st->state = CSTATE_CHOOSE_SCRIPT;
2425 :
2426 : /*
2427 : * If we paced through all commands in the script in this
2428 : * loop, without returning to the caller even once, do it now.
2429 : * This gives the thread a chance to process other
2430 : * connections, and to do progress reporting. This can
2431 : * currently only happen if the script consists entirely of
2432 : * meta-commands.
2433 : */
2434 1387 : if (end_tx_processed)
2435 0 : return;
2436 : else
2437 : {
2438 1387 : end_tx_processed = true;
2439 1387 : break;
2440 : }
2441 :
2442 : /*
2443 : * Final states. Close the connection if it's still open.
2444 : */
2445 : case CSTATE_ABORTED:
2446 : case CSTATE_FINISHED:
2447 51 : if (st->con != NULL)
2448 : {
2449 44 : PQfinish(st->con);
2450 44 : st->con = NULL;
2451 : }
2452 51 : return;
2453 : }
2454 15703 : }
2455 : }
2456 :
2457 : /*
2458 : * Print log entry after completing one transaction.
2459 : *
2460 : * We print Unix-epoch timestamps in the log, so that entries can be
2461 : * correlated against other logs. On some platforms this could be obtained
2462 : * from the instr_time reading the caller has, but rather than get entangled
2463 : * with that, we just eat the cost of an extra syscall in all cases.
2464 : */
2465 : static void
2466 152 : doLog(TState *thread, CState *st,
2467 : StatsData *agg, bool skipped, double latency, double lag)
2468 : {
2469 152 : FILE *logfile = thread->logfile;
2470 :
2471 152 : Assert(use_log);
2472 :
2473 : /*
2474 : * Skip the log entry if sampling is enabled and this row doesn't belong
2475 : * to the random sample.
2476 : */
2477 252 : if (sample_rate != 0.0 &&
2478 100 : pg_erand48(thread->random_state) > sample_rate)
2479 198 : return;
2480 :
2481 : /* should we aggregate the results or not? */
2482 106 : if (agg_interval > 0)
2483 : {
2484 : /*
2485 : * Loop until we reach the interval of the current moment, and print
2486 : * any empty intervals in between (this may happen with very low tps,
2487 : * e.g. --rate=0.1).
2488 : */
2489 42 : time_t now = time(NULL);
2490 :
2491 88 : while (agg->start_time + agg_interval <= now)
2492 : {
2493 : /* print aggregated report to logfile */
2494 4 : fprintf(logfile, "%ld " INT64_FORMAT " %.0f %.0f %.0f %.0f",
2495 : (long) agg->start_time,
2496 : agg->cnt,
2497 : agg->latency.sum,
2498 : agg->latency.sum2,
2499 : agg->latency.min,
2500 : agg->latency.max);
2501 4 : if (throttle_delay)
2502 : {
2503 4 : fprintf(logfile, " %.0f %.0f %.0f %.0f",
2504 : agg->lag.sum,
2505 : agg->lag.sum2,
2506 : agg->lag.min,
2507 : agg->lag.max);
2508 4 : if (latency_limit)
2509 4 : fprintf(logfile, " " INT64_FORMAT, agg->skipped);
2510 : }
2511 4 : fputc('\n', logfile);
2512 :
2513 : /* reset data and move to next interval */
2514 4 : initStats(agg, agg->start_time + agg_interval);
2515 : }
2516 :
2517 : /* accumulate the current transaction */
2518 42 : accumStats(agg, skipped, latency, lag);
2519 : }
2520 : else
2521 : {
2522 : /* no, print raw transactions */
2523 : struct timeval tv;
2524 :
2525 64 : gettimeofday(&tv, NULL);
2526 64 : if (skipped)
2527 0 : fprintf(logfile, "%d " INT64_FORMAT " skipped %d %ld %ld",
2528 : st->id, st->cnt, st->use_file,
2529 : (long) tv.tv_sec, (long) tv.tv_usec);
2530 : else
2531 64 : fprintf(logfile, "%d " INT64_FORMAT " %.0f %d %ld %ld",
2532 : st->id, st->cnt, latency, st->use_file,
2533 : (long) tv.tv_sec, (long) tv.tv_usec);
2534 64 : if (throttle_delay)
2535 0 : fprintf(logfile, " %.0f", lag);
2536 64 : fputc('\n', logfile);
2537 : }
2538 : }
2539 :
2540 : /*
2541 : * Accumulate and report statistics at end of a transaction.
2542 : *
2543 : * (This is also called when a transaction is late and thus skipped.
2544 : * Note that even skipped transactions are counted in the "cnt" fields.)
2545 : */
2546 : static void
2547 1427 : processXactStats(TState *thread, CState *st, instr_time *now,
2548 : bool skipped, StatsData *agg)
2549 : {
2550 1427 : double latency = 0.0,
2551 1427 : lag = 0.0;
2552 1427 : bool thread_details = progress || throttle_delay || latency_limit,
2553 1427 : detailed = thread_details || use_log || per_script_stats;
2554 :
2555 1427 : if (detailed && !skipped)
2556 : {
2557 450 : if (INSTR_TIME_IS_ZERO(*now))
2558 210 : INSTR_TIME_SET_CURRENT(*now);
2559 :
2560 : /* compute latency & lag */
2561 450 : latency = INSTR_TIME_GET_MICROSEC(*now) - st->txn_scheduled;
2562 450 : lag = INSTR_TIME_GET_MICROSEC(st->txn_begin) - st->txn_scheduled;
2563 : }
2564 :
2565 1427 : if (thread_details)
2566 : {
2567 : /* keep detailed thread stats */
2568 250 : accumStats(&thread->stats, skipped, latency, lag);
2569 :
2570 : /* count transactions over the latency limit, if needed */
2571 250 : if (latency_limit && latency > latency_limit)
2572 0 : thread->latency_late++;
2573 : }
2574 : else
2575 : {
2576 : /* no detailed stats, just count */
2577 1177 : thread->stats.cnt++;
2578 : }
2579 :
2580 : /* client stat is just counting */
2581 1427 : st->cnt++;
2582 :
2583 1427 : if (use_log)
2584 150 : doLog(thread, st, agg, skipped, latency, lag);
2585 :
2586 : /* XXX could use a mutex here, but we choose not to */
2587 1427 : if (per_script_stats)
2588 350 : accumStats(&sql_script[st->use_file].stats, skipped, latency, lag);
2589 1427 : }
2590 :
2591 :
2592 : /* discard connections */
2593 : static void
2594 63 : disconnect_all(CState *state, int length)
2595 : {
2596 : int i;
2597 :
2598 165 : for (i = 0; i < length; i++)
2599 : {
2600 102 : if (state[i].con)
2601 : {
2602 0 : PQfinish(state[i].con);
2603 0 : state[i].con = NULL;
2604 : }
2605 : }
2606 63 : }
2607 :
2608 : /* create tables and setup data */
2609 : static void
2610 2 : init(bool is_no_vacuum)
2611 : {
2612 : /*
2613 : * The scale factor at/beyond which 32-bit integers are insufficient for
2614 : * storing TPC-B account IDs.
2615 : *
2616 : * Although the actual threshold is 21474, we use 20000 because it is easier to
2617 : * document and remember, and isn't that far away from the real threshold.
2618 : */
2619 : #define SCALE_32BIT_THRESHOLD 20000
2620 :
2621 : /*
2622 : * Note: TPC-B requires at least 100 bytes per row, and the "filler"
2623 : * fields in these table declarations were intended to comply with that.
2624 : * The pgbench_accounts table complies with that because the "filler"
2625 : * column is set to blank-padded empty string. But for all other tables
2626 : * the columns default to NULL and so don't actually take any space. We
2627 : * could fix that by giving them non-null default values. However, that
2628 : * would completely break comparability of pgbench results with prior
2629 : * versions. Since pgbench has never pretended to be fully TPC-B compliant
2630 : * anyway, we stick with the historical behavior.
2631 : */
2632 : struct ddlinfo
2633 : {
2634 : const char *table; /* table name */
2635 : const char *smcols; /* column decls if accountIDs are 32 bits */
2636 : const char *bigcols; /* column decls if accountIDs are 64 bits */
2637 : int declare_fillfactor;
2638 : };
2639 : static const struct ddlinfo DDLs[] = {
2640 : {
2641 : "pgbench_history",
2642 : "tid int,bid int,aid int,delta int,mtime timestamp,filler char(22)",
2643 : "tid int,bid int,aid bigint,delta int,mtime timestamp,filler char(22)",
2644 : 0
2645 : },
2646 : {
2647 : "pgbench_tellers",
2648 : "tid int not null,bid int,tbalance int,filler char(84)",
2649 : "tid int not null,bid int,tbalance int,filler char(84)",
2650 : 1
2651 : },
2652 : {
2653 : "pgbench_accounts",
2654 : "aid int not null,bid int,abalance int,filler char(84)",
2655 : "aid bigint not null,bid int,abalance int,filler char(84)",
2656 : 1
2657 : },
2658 : {
2659 : "pgbench_branches",
2660 : "bid int not null,bbalance int,filler char(88)",
2661 : "bid int not null,bbalance int,filler char(88)",
2662 : 1
2663 : }
2664 : };
2665 : static const char *const DDLINDEXes[] = {
2666 : "alter table pgbench_branches add primary key (bid)",
2667 : "alter table pgbench_tellers add primary key (tid)",
2668 : "alter table pgbench_accounts add primary key (aid)"
2669 : };
2670 : static const char *const DDLKEYs[] = {
2671 : "alter table pgbench_tellers add foreign key (bid) references pgbench_branches",
2672 : "alter table pgbench_accounts add foreign key (bid) references pgbench_branches",
2673 : "alter table pgbench_history add foreign key (bid) references pgbench_branches",
2674 : "alter table pgbench_history add foreign key (tid) references pgbench_tellers",
2675 : "alter table pgbench_history add foreign key (aid) references pgbench_accounts"
2676 : };
2677 :
2678 : PGconn *con;
2679 : PGresult *res;
2680 : char sql[256];
2681 : int i;
2682 : int64 k;
2683 :
2684 : /* used to track elapsed time and estimate of the remaining time */
2685 : instr_time start,
2686 : diff;
2687 : double elapsed_sec,
2688 : remaining_sec;
2689 2 : int log_interval = 1;
2690 :
2691 2 : if ((con = doConnect()) == NULL)
2692 0 : exit(1);
2693 :
2694 10 : for (i = 0; i < lengthof(DDLs); i++)
2695 : {
2696 : char opts[256];
2697 : char buffer[256];
2698 8 : const struct ddlinfo *ddl = &DDLs[i];
2699 : const char *cols;
2700 :
2701 : /* Remove old table, if it exists. */
2702 8 : snprintf(buffer, sizeof(buffer), "drop table if exists %s", ddl->table);
2703 8 : executeStatement(con, buffer);
2704 :
2705 : /* Construct new create table statement. */
2706 8 : opts[0] = '\0';
2707 8 : if (ddl->declare_fillfactor)
2708 6 : snprintf(opts + strlen(opts), sizeof(opts) - strlen(opts),
2709 : " with (fillfactor=%d)", fillfactor);
2710 8 : if (tablespace != NULL)
2711 : {
2712 : char *escape_tablespace;
2713 :
2714 4 : escape_tablespace = PQescapeIdentifier(con, tablespace,
2715 : strlen(tablespace));
2716 4 : snprintf(opts + strlen(opts), sizeof(opts) - strlen(opts),
2717 : " tablespace %s", escape_tablespace);
2718 4 : PQfreemem(escape_tablespace);
2719 : }
2720 :
2721 8 : cols = (scale >= SCALE_32BIT_THRESHOLD) ? ddl->bigcols : ddl->smcols;
2722 :
2723 16 : snprintf(buffer, sizeof(buffer), "create%s table %s(%s)%s",
2724 8 : unlogged_tables ? " unlogged" : "",
2725 : ddl->table, cols, opts);
2726 :
2727 8 : executeStatement(con, buffer);
2728 : }
2729 :
2730 2 : executeStatement(con, "begin");
2731 :
2732 4 : for (i = 0; i < nbranches * scale; i++)
2733 : {
2734 : /* "filler" column defaults to NULL */
2735 2 : snprintf(sql, sizeof(sql),
2736 : "insert into pgbench_branches(bid,bbalance) values(%d,0)",
2737 : i + 1);
2738 2 : executeStatement(con, sql);
2739 : }
2740 :
2741 22 : for (i = 0; i < ntellers * scale; i++)
2742 : {
2743 : /* "filler" column defaults to NULL */
2744 20 : snprintf(sql, sizeof(sql),
2745 : "insert into pgbench_tellers(tid,bid,tbalance) values (%d,%d,0)",
2746 20 : i + 1, i / ntellers + 1);
2747 20 : executeStatement(con, sql);
2748 : }
2749 :
2750 2 : executeStatement(con, "commit");
2751 :
2752 : /*
2753 : * fill the pgbench_accounts table with some data
2754 : */
2755 2 : fprintf(stderr, "creating tables...\n");
2756 :
2757 2 : executeStatement(con, "begin");
2758 2 : executeStatement(con, "truncate pgbench_accounts");
2759 :
2760 2 : res = PQexec(con, "copy pgbench_accounts from stdin");
2761 2 : if (PQresultStatus(res) != PGRES_COPY_IN)
2762 : {
2763 0 : fprintf(stderr, "%s", PQerrorMessage(con));
2764 0 : exit(1);
2765 : }
2766 2 : PQclear(res);
2767 :
2768 2 : INSTR_TIME_SET_CURRENT(start);
2769 :
2770 200002 : for (k = 0; k < (int64) naccounts * scale; k++)
2771 : {
2772 200000 : int64 j = k + 1;
2773 :
2774 : /* "filler" column defaults to blank padded empty string */
2775 200000 : snprintf(sql, sizeof(sql),
2776 : INT64_FORMAT "\t" INT64_FORMAT "\t%d\t\n",
2777 200000 : j, k / naccounts + 1, 0);
2778 200000 : if (PQputline(con, sql))
2779 : {
2780 0 : fprintf(stderr, "PQputline failed\n");
2781 0 : exit(1);
2782 : }
2783 :
2784 : /*
2785 : * If we want to stick with the original logging, print a message each
2786 : * 100k inserted rows.
2787 : */
2788 200000 : if ((!use_quiet) && (j % 100000 == 0))
2789 : {
2790 1 : INSTR_TIME_SET_CURRENT(diff);
2791 1 : INSTR_TIME_SUBTRACT(diff, start);
2792 :
2793 1 : elapsed_sec = INSTR_TIME_GET_DOUBLE(diff);
2794 1 : remaining_sec = ((double) scale * naccounts - j) * elapsed_sec / j;
2795 :
2796 1 : fprintf(stderr, INT64_FORMAT " of " INT64_FORMAT " tuples (%d%%) done (elapsed %.2f s, remaining %.2f s)\n",
2797 : j, (int64) naccounts * scale,
2798 1 : (int) (((int64) j * 100) / (naccounts * (int64) scale)),
2799 : elapsed_sec, remaining_sec);
2800 : }
2801 : /* let's not call the timing for each row, but only each 100 rows */
2802 199999 : else if (use_quiet && (j % 100 == 0))
2803 : {
2804 1000 : INSTR_TIME_SET_CURRENT(diff);
2805 1000 : INSTR_TIME_SUBTRACT(diff, start);
2806 :
2807 1000 : elapsed_sec = INSTR_TIME_GET_DOUBLE(diff);
2808 1000 : remaining_sec = ((double) scale * naccounts - j) * elapsed_sec / j;
2809 :
2810 : /* have we reached the next interval (or end)? */
2811 1000 : if ((j == scale * naccounts) || (elapsed_sec >= log_interval * LOG_STEP_SECONDS))
2812 : {
2813 1 : fprintf(stderr, INT64_FORMAT " of " INT64_FORMAT " tuples (%d%%) done (elapsed %.2f s, remaining %.2f s)\n",
2814 : j, (int64) naccounts * scale,
2815 1 : (int) (((int64) j * 100) / (naccounts * (int64) scale)), elapsed_sec, remaining_sec);
2816 :
2817 : /* skip to the next interval */
2818 1 : log_interval = (int) ceil(elapsed_sec / LOG_STEP_SECONDS);
2819 : }
2820 : }
2821 :
2822 : }
2823 2 : if (PQputline(con, "\\.\n"))
2824 : {
2825 0 : fprintf(stderr, "very last PQputline failed\n");
2826 0 : exit(1);
2827 : }
2828 2 : if (PQendcopy(con))
2829 : {
2830 0 : fprintf(stderr, "PQendcopy failed\n");
2831 0 : exit(1);
2832 : }
2833 2 : executeStatement(con, "commit");
2834 :
2835 : /* vacuum */
2836 2 : if (!is_no_vacuum)
2837 : {
2838 2 : fprintf(stderr, "vacuum...\n");
2839 2 : executeStatement(con, "vacuum analyze pgbench_branches");
2840 2 : executeStatement(con, "vacuum analyze pgbench_tellers");
2841 2 : executeStatement(con, "vacuum analyze pgbench_accounts");
2842 2 : executeStatement(con, "vacuum analyze pgbench_history");
2843 : }
2844 :
2845 : /*
2846 : * create indexes
2847 : */
2848 2 : fprintf(stderr, "set primary keys...\n");
2849 8 : for (i = 0; i < lengthof(DDLINDEXes); i++)
2850 : {
2851 : char buffer[256];
2852 :
2853 6 : strlcpy(buffer, DDLINDEXes[i], sizeof(buffer));
2854 :
2855 6 : if (index_tablespace != NULL)
2856 : {
2857 : char *escape_tablespace;
2858 :
2859 3 : escape_tablespace = PQescapeIdentifier(con, index_tablespace,
2860 : strlen(index_tablespace));
2861 3 : snprintf(buffer + strlen(buffer), sizeof(buffer) - strlen(buffer),
2862 : " using index tablespace %s", escape_tablespace);
2863 3 : PQfreemem(escape_tablespace);
2864 : }
2865 :
2866 6 : executeStatement(con, buffer);
2867 : }
2868 :
2869 : /*
2870 : * create foreign keys
2871 : */
2872 2 : if (foreign_keys)
2873 : {
2874 1 : fprintf(stderr, "set foreign keys...\n");
2875 6 : for (i = 0; i < lengthof(DDLKEYs); i++)
2876 : {
2877 5 : executeStatement(con, DDLKEYs[i]);
2878 : }
2879 : }
2880 :
2881 2 : fprintf(stderr, "done.\n");
2882 2 : PQfinish(con);
2883 2 : }
2884 :
2885 : /*
2886 : * Replace :param with $n throughout the command's SQL text, which
2887 : * is a modifiable string in cmd->argv[0].
2888 : */
2889 : static bool
2890 16 : parseQuery(Command *cmd)
2891 : {
2892 : char *sql,
2893 : *p;
2894 :
2895 : /* We don't want to scribble on cmd->argv[0] until done */
2896 16 : sql = pg_strdup(cmd->argv[0]);
2897 :
2898 16 : cmd->argc = 1;
2899 :
2900 16 : p = sql;
2901 59 : while ((p = strchr(p, ':')) != NULL)
2902 : {
2903 : char var[12];
2904 : char *name;
2905 : int eaten;
2906 :
2907 28 : name = parseVariable(p, &eaten);
2908 28 : if (name == NULL)
2909 : {
2910 20 : while (*p == ':')
2911 : {
2912 10 : p++;
2913 : }
2914 5 : continue;
2915 : }
2916 :
2917 23 : if (cmd->argc >= MAX_ARGS)
2918 : {
2919 1 : fprintf(stderr, "statement has too many arguments (maximum is %d): %s\n",
2920 : MAX_ARGS - 1, cmd->argv[0]);
2921 1 : pg_free(name);
2922 1 : return false;
2923 : }
2924 :
2925 22 : sprintf(var, "$%d", cmd->argc);
2926 22 : p = replaceVariable(&sql, p, eaten, var);
2927 :
2928 22 : cmd->argv[cmd->argc] = name;
2929 22 : cmd->argc++;
2930 : }
2931 :
2932 15 : pg_free(cmd->argv[0]);
2933 15 : cmd->argv[0] = sql;
2934 15 : return true;
2935 : }
2936 :
2937 : /*
2938 : * Simple error-printing function, might be needed by lexer
2939 : */
2940 : static void
2941 0 : pgbench_error(const char *fmt,...)
2942 : {
2943 : va_list ap;
2944 :
2945 0 : fflush(stdout);
2946 0 : va_start(ap, fmt);
2947 0 : vfprintf(stderr, _(fmt), ap);
2948 0 : va_end(ap);
2949 0 : }
2950 :
2951 : /*
2952 : * syntax error while parsing a script (in practice, while parsing a
2953 : * backslash command, because we don't detect syntax errors in SQL)
2954 : *
2955 : * source: source of script (filename or builtin-script ID)
2956 : * lineno: line number within script (count from 1)
2957 : * line: whole line of backslash command, if available
2958 : * command: backslash command name, if available
2959 : * msg: the actual error message
2960 : * more: optional extra message
2961 : * column: zero-based column number, or -1 if unknown
2962 : */
2963 : void
2964 12 : syntax_error(const char *source, int lineno,
2965 : const char *line, const char *command,
2966 : const char *msg, const char *more, int column)
2967 : {
2968 12 : fprintf(stderr, "%s:%d: %s", source, lineno, msg);
2969 12 : if (more != NULL)
2970 3 : fprintf(stderr, " (%s)", more);
2971 12 : if (column >= 0 && line == NULL)
2972 0 : fprintf(stderr, " at column %d", column + 1);
2973 12 : if (command != NULL)
2974 12 : fprintf(stderr, " in command \"%s\"", command);
2975 12 : fprintf(stderr, "\n");
2976 12 : if (line != NULL)
2977 : {
2978 10 : fprintf(stderr, "%s\n", line);
2979 10 : if (column >= 0)
2980 : {
2981 : int i;
2982 :
2983 72 : for (i = 0; i < column; i++)
2984 66 : fprintf(stderr, " ");
2985 6 : fprintf(stderr, "^ error found here\n");
2986 : }
2987 : }
2988 12 : exit(1);
2989 : }
2990 :
2991 : /*
2992 : * Parse a SQL command; return a Command struct, or NULL if it's a comment
2993 : *
2994 : * On entry, psqlscan.l has collected the command into "buf", so we don't
2995 : * really need to do much here except check for comment and set up a
2996 : * Command struct.
2997 : */
2998 : static Command *
2999 672 : process_sql_command(PQExpBuffer buf, const char *source)
3000 : {
3001 : Command *my_command;
3002 : char *p;
3003 : char *nlpos;
3004 :
3005 : /* Skip any leading whitespace, as well as "--" style comments */
3006 672 : p = buf->data;
3007 : for (;;)
3008 : {
3009 672 : if (isspace((unsigned char) *p))
3010 0 : p++;
3011 672 : else if (strncmp(p, "--", 2) == 0)
3012 : {
3013 0 : p = strchr(p, '\n');
3014 0 : if (p == NULL)
3015 0 : return NULL;
3016 0 : p++;
3017 : }
3018 : else
3019 672 : break;
3020 0 : }
3021 :
3022 : /* If there's nothing but whitespace and comments, we're done */
3023 672 : if (*p == '\0')
3024 435 : return NULL;
3025 :
3026 : /* Allocate and initialize Command structure */
3027 237 : my_command = (Command *) pg_malloc0(sizeof(Command));
3028 237 : my_command->command_num = num_commands++;
3029 237 : my_command->type = SQL_COMMAND;
3030 237 : initSimpleStats(&my_command->stats);
3031 :
3032 : /*
3033 : * Install query text as the sole argv string. If we are using a
3034 : * non-simple query mode, we'll extract parameters from it later.
3035 : */
3036 237 : my_command->argv[0] = pg_strdup(p);
3037 237 : my_command->argc = 1;
3038 :
3039 : /*
3040 : * If SQL command is multi-line, we only want to save the first line as
3041 : * the "line" label.
3042 : */
3043 237 : nlpos = strchr(p, '\n');
3044 237 : if (nlpos)
3045 : {
3046 3 : my_command->line = pg_malloc(nlpos - p + 1);
3047 3 : memcpy(my_command->line, p, nlpos - p);
3048 3 : my_command->line[nlpos - p] = '\0';
3049 : }
3050 : else
3051 234 : my_command->line = pg_strdup(p);
3052 :
3053 237 : return my_command;
3054 : }
3055 :
3056 : /*
3057 : * Parse a backslash command; return a Command struct, or NULL if comment
3058 : *
3059 : * At call, we have scanned only the initial backslash.
3060 : */
3061 : static Command *
3062 256 : process_backslash_command(PsqlScanState sstate, const char *source)
3063 : {
3064 : Command *my_command;
3065 : PQExpBufferData word_buf;
3066 : int word_offset;
3067 : int offsets[MAX_ARGS]; /* offsets of argument words */
3068 : int start_offset;
3069 : int lineno;
3070 : int j;
3071 :
3072 256 : initPQExpBuffer(&word_buf);
3073 :
3074 : /* Remember location of the backslash */
3075 256 : start_offset = expr_scanner_offset(sstate) - 1;
3076 256 : lineno = expr_scanner_get_lineno(sstate, start_offset);
3077 :
3078 : /* Collect first word of command */
3079 256 : if (!expr_lex_one_word(sstate, &word_buf, &word_offset))
3080 : {
3081 0 : termPQExpBuffer(&word_buf);
3082 0 : return NULL;
3083 : }
3084 :
3085 : /* Allocate and initialize Command structure */
3086 256 : my_command = (Command *) pg_malloc0(sizeof(Command));
3087 256 : my_command->command_num = num_commands++;
3088 256 : my_command->type = META_COMMAND;
3089 256 : my_command->argc = 0;
3090 256 : initSimpleStats(&my_command->stats);
3091 :
3092 : /* Save first word (command name) */
3093 256 : j = 0;
3094 256 : offsets[j] = word_offset;
3095 256 : my_command->argv[j++] = pg_strdup(word_buf.data);
3096 256 : my_command->argc++;
3097 :
3098 256 : if (pg_strcasecmp(my_command->argv[0], "set") == 0)
3099 : {
3100 : /* For \set, collect var name, then lex the expression. */
3101 : yyscan_t yyscanner;
3102 :
3103 237 : if (!expr_lex_one_word(sstate, &word_buf, &word_offset))
3104 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3105 : "missing argument", NULL, -1);
3106 :
3107 236 : offsets[j] = word_offset;
3108 236 : my_command->argv[j++] = pg_strdup(word_buf.data);
3109 236 : my_command->argc++;
3110 :
3111 236 : yyscanner = expr_scanner_init(sstate, source, lineno, start_offset,
3112 236 : my_command->argv[0]);
3113 :
3114 236 : if (expr_yyparse(yyscanner) != 0)
3115 : {
3116 : /* dead code: exit done from syntax_error called by yyerror */
3117 0 : exit(1);
3118 : }
3119 :
3120 232 : my_command->expr = expr_parse_result;
3121 :
3122 : /* Save line, trimming any trailing newline */
3123 232 : my_command->line = expr_scanner_get_substring(sstate,
3124 : start_offset,
3125 : expr_scanner_offset(sstate),
3126 : true);
3127 :
3128 232 : expr_scanner_finish(yyscanner);
3129 :
3130 232 : termPQExpBuffer(&word_buf);
3131 :
3132 232 : return my_command;
3133 : }
3134 :
3135 : /* For all other commands, collect remaining words. */
3136 79 : while (expr_lex_one_word(sstate, &word_buf, &word_offset))
3137 : {
3138 42 : if (j >= MAX_ARGS)
3139 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3140 : "too many arguments", NULL, -1);
3141 :
3142 41 : offsets[j] = word_offset;
3143 41 : my_command->argv[j++] = pg_strdup(word_buf.data);
3144 41 : my_command->argc++;
3145 : }
3146 :
3147 : /* Save line, trimming any trailing newline */
3148 18 : my_command->line = expr_scanner_get_substring(sstate,
3149 : start_offset,
3150 : expr_scanner_offset(sstate),
3151 : true);
3152 :
3153 18 : if (pg_strcasecmp(my_command->argv[0], "sleep") == 0)
3154 : {
3155 9 : if (my_command->argc < 2)
3156 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3157 : "missing argument", NULL, -1);
3158 :
3159 8 : if (my_command->argc > 3)
3160 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3161 : "too many arguments", NULL,
3162 1 : offsets[3] - start_offset);
3163 :
3164 : /*
3165 : * Split argument into number and unit to allow "sleep 1ms" etc. We
3166 : * don't have to terminate the number argument with null because it
3167 : * will be parsed with atoi, which ignores trailing non-digit
3168 : * characters.
3169 : */
3170 7 : if (my_command->argc == 2 && my_command->argv[1][0] != ':')
3171 : {
3172 1 : char *c = my_command->argv[1];
3173 :
3174 3 : while (isdigit((unsigned char) *c))
3175 1 : c++;
3176 1 : if (*c)
3177 : {
3178 1 : my_command->argv[2] = c;
3179 1 : offsets[2] = offsets[1] + (c - my_command->argv[1]);
3180 1 : my_command->argc = 3;
3181 : }
3182 : }
3183 :
3184 7 : if (my_command->argc == 3)
3185 : {
3186 9 : if (pg_strcasecmp(my_command->argv[2], "us") != 0 &&
3187 6 : pg_strcasecmp(my_command->argv[2], "ms") != 0 &&
3188 2 : pg_strcasecmp(my_command->argv[2], "s") != 0)
3189 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3190 : "unrecognized time unit, must be us, ms or s",
3191 2 : my_command->argv[2], offsets[2] - start_offset);
3192 : }
3193 : }
3194 9 : else if (pg_strcasecmp(my_command->argv[0], "setshell") == 0)
3195 : {
3196 4 : if (my_command->argc < 3)
3197 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3198 : "missing argument", NULL, -1);
3199 : }
3200 5 : else if (pg_strcasecmp(my_command->argv[0], "shell") == 0)
3201 : {
3202 4 : if (my_command->argc < 2)
3203 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3204 : "missing command", NULL, -1);
3205 : }
3206 : else
3207 : {
3208 1 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
3209 : "invalid command", NULL, -1);
3210 : }
3211 :
3212 12 : termPQExpBuffer(&word_buf);
3213 :
3214 12 : return my_command;
3215 : }
3216 :
3217 : /*
3218 : * Parse a script (either the contents of a file, or a built-in script)
3219 : * and add it to the list of scripts.
3220 : */
3221 : static void
3222 191 : ParseScript(const char *script, const char *desc, int weight)
3223 : {
3224 : ParsedScript ps;
3225 : PsqlScanState sstate;
3226 : PQExpBufferData line_buf;
3227 : int alloc_num;
3228 : int index;
3229 :
3230 : #define COMMANDS_ALLOC_NUM 128
3231 191 : alloc_num = COMMANDS_ALLOC_NUM;
3232 :
3233 : /* Initialize all fields of ps */
3234 191 : ps.desc = desc;
3235 191 : ps.weight = weight;
3236 191 : ps.commands = (Command **) pg_malloc(sizeof(Command *) * alloc_num);
3237 191 : initStats(&ps.stats, 0);
3238 :
3239 : /* Prepare to parse script */
3240 191 : sstate = psql_scan_create(&pgbench_callbacks);
3241 :
3242 : /*
3243 : * Ideally, we'd scan scripts using the encoding and stdstrings settings
3244 : * we get from a DB connection. However, without major rearrangement of
3245 : * pgbench's argument parsing, we can't have a DB connection at the time
3246 : * we parse scripts. Using SQL_ASCII (encoding 0) should work well enough
3247 : * with any backend-safe encoding, though conceivably we could be fooled
3248 : * if a script file uses a client-only encoding. We also assume that
3249 : * stdstrings should be true, which is a bit riskier.
3250 : */
3251 191 : psql_scan_setup(sstate, script, strlen(script), 0, true);
3252 :
3253 191 : initPQExpBuffer(&line_buf);
3254 :
3255 191 : index = 0;
3256 :
3257 : for (;;)
3258 : {
3259 : PsqlScanResult sr;
3260 : promptStatus_t prompt;
3261 : Command *command;
3262 :
3263 672 : resetPQExpBuffer(&line_buf);
3264 :
3265 672 : sr = psql_scan(sstate, &line_buf, &prompt);
3266 :
3267 : /* If we collected a SQL command, process that */
3268 672 : command = process_sql_command(&line_buf, desc);
3269 672 : if (command)
3270 : {
3271 237 : ps.commands[index] = command;
3272 237 : index++;
3273 :
3274 237 : if (index >= alloc_num)
3275 : {
3276 0 : alloc_num += COMMANDS_ALLOC_NUM;
3277 0 : ps.commands = (Command **)
3278 0 : pg_realloc(ps.commands, sizeof(Command *) * alloc_num);
3279 : }
3280 : }
3281 :
3282 : /* If we reached a backslash, process that */
3283 672 : if (sr == PSCAN_BACKSLASH)
3284 : {
3285 256 : command = process_backslash_command(sstate, desc);
3286 244 : if (command)
3287 : {
3288 244 : ps.commands[index] = command;
3289 244 : index++;
3290 :
3291 244 : if (index >= alloc_num)
3292 : {
3293 0 : alloc_num += COMMANDS_ALLOC_NUM;
3294 0 : ps.commands = (Command **)
3295 0 : pg_realloc(ps.commands, sizeof(Command *) * alloc_num);
3296 : }
3297 : }
3298 : }
3299 :
3300 : /* Done if we reached EOF */
3301 660 : if (sr == PSCAN_INCOMPLETE || sr == PSCAN_EOL)
3302 : break;
3303 481 : }
3304 :
3305 179 : ps.commands[index] = NULL;
3306 :
3307 179 : addScript(ps);
3308 :
3309 177 : termPQExpBuffer(&line_buf);
3310 177 : psql_scan_finish(sstate);
3311 177 : psql_scan_destroy(sstate);
3312 177 : }
3313 :
3314 : /*
3315 : * Read the entire contents of file fd, and return it in a malloc'd buffer.
3316 : *
3317 : * The buffer will typically be larger than necessary, but we don't care
3318 : * in this program, because we'll free it as soon as we've parsed the script.
3319 : */
3320 : static char *
3321 39 : read_file_contents(FILE *fd)
3322 : {
3323 : char *buf;
3324 39 : size_t buflen = BUFSIZ;
3325 39 : size_t used = 0;
3326 :
3327 39 : buf = (char *) pg_malloc(buflen);
3328 :
3329 : for (;;)
3330 : {
3331 : size_t nread;
3332 :
3333 39 : nread = fread(buf + used, 1, BUFSIZ, fd);
3334 39 : used += nread;
3335 : /* If fread() read less than requested, must be EOF or error */
3336 39 : if (nread < BUFSIZ)
3337 39 : break;
3338 : /* Enlarge buf so we can read some more */
3339 0 : buflen += BUFSIZ;
3340 0 : buf = (char *) pg_realloc(buf, buflen);
3341 0 : }
3342 : /* There is surely room for a terminator */
3343 39 : buf[used] = '\0';
3344 :
3345 39 : return buf;
3346 : }
3347 :
3348 : /*
3349 : * Given a file name, read it and add its script to the list.
3350 : * "-" means to read stdin.
3351 : * NB: filename must be storage that won't disappear.
3352 : */
3353 : static void
3354 40 : process_file(const char *filename, int weight)
3355 : {
3356 : FILE *fd;
3357 : char *buf;
3358 :
3359 : /* Slurp the file contents into "buf" */
3360 40 : if (strcmp(filename, "-") == 0)
3361 0 : fd = stdin;
3362 40 : else if ((fd = fopen(filename, "r")) == NULL)
3363 : {
3364 1 : fprintf(stderr, "could not open file \"%s\": %s\n",
3365 1 : filename, strerror(errno));
3366 1 : exit(1);
3367 : }
3368 :
3369 39 : buf = read_file_contents(fd);
3370 :
3371 39 : if (ferror(fd))
3372 : {
3373 0 : fprintf(stderr, "could not read file \"%s\": %s\n",
3374 0 : filename, strerror(errno));
3375 0 : exit(1);
3376 : }
3377 :
3378 39 : if (fd != stdin)
3379 39 : fclose(fd);
3380 :
3381 39 : ParseScript(buf, filename, weight);
3382 :
3383 26 : free(buf);
3384 26 : }
3385 :
3386 : /* Parse the given builtin script and add it to the list. */
3387 : static void
3388 152 : process_builtin(const BuiltinScript *bi, int weight)
3389 : {
3390 152 : ParseScript(bi->script, bi->desc, weight);
3391 151 : }
3392 :
3393 : /* show available builtin scripts */
3394 : static void
3395 3 : listAvailableScripts(void)
3396 : {
3397 : int i;
3398 :
3399 3 : fprintf(stderr, "Available builtin scripts:\n");
3400 12 : for (i = 0; i < lengthof(builtin_script); i++)
3401 9 : fprintf(stderr, "\t%s\n", builtin_script[i].name);
3402 3 : fprintf(stderr, "\n");
3403 3 : }
3404 :
3405 : /* return builtin script "name" if unambiguous, fails if not found */
3406 : static const BuiltinScript *
3407 154 : findBuiltin(const char *name)
3408 : {
3409 : int i,
3410 154 : found = 0,
3411 154 : len = strlen(name);
3412 154 : const BuiltinScript *result = NULL;
3413 :
3414 616 : for (i = 0; i < lengthof(builtin_script); i++)
3415 : {
3416 462 : if (strncmp(builtin_script[i].name, name, len) == 0)
3417 : {
3418 154 : result = &builtin_script[i];
3419 154 : found++;
3420 : }
3421 : }
3422 :
3423 : /* ok, unambiguous result */
3424 154 : if (found == 1)
3425 304 : return result;
3426 :
3427 : /* error cases */
3428 2 : if (found == 0)
3429 1 : fprintf(stderr, "no builtin script found for name \"%s\"\n", name);
3430 : else /* found > 1 */
3431 1 : fprintf(stderr,
3432 : "ambiguous builtin name: %d builtin scripts found for prefix \"%s\"\n", found, name);
3433 :
3434 2 : listAvailableScripts();
3435 2 : exit(1);
3436 : }
3437 :
3438 : /*
3439 : * Determine the weight specification from a script option (-b, -f), if any,
3440 : * and return it as an integer (1 is returned if there's no weight). The
3441 : * script name is returned in *script as a malloc'd string.
3442 : */
3443 : static int
3444 52 : parseScriptWeight(const char *option, char **script)
3445 : {
3446 : char *sep;
3447 : int weight;
3448 :
3449 52 : if ((sep = strrchr(option, WSEP)))
3450 : {
3451 8 : int namelen = sep - option;
3452 : long wtmp;
3453 : char *badp;
3454 :
3455 : /* generate the script name */
3456 8 : *script = pg_malloc(namelen + 1);
3457 8 : strncpy(*script, option, namelen);
3458 8 : (*script)[namelen] = '\0';
3459 :
3460 : /* process digits of the weight spec */
3461 8 : errno = 0;
3462 8 : wtmp = strtol(sep + 1, &badp, 10);
3463 8 : if (errno != 0 || badp == sep + 1 || *badp != '\0')
3464 : {
3465 1 : fprintf(stderr, "invalid weight specification: %s\n", sep);
3466 1 : exit(1);
3467 : }
3468 7 : if (wtmp > INT_MAX || wtmp < 0)
3469 : {
3470 1 : fprintf(stderr,
3471 : "weight specification out of range (0 .. %u): " INT64_FORMAT "\n",
3472 : INT_MAX, (int64) wtmp);
3473 1 : exit(1);
3474 : }
3475 6 : weight = wtmp;
3476 : }
3477 : else
3478 : {
3479 44 : *script = pg_strdup(option);
3480 44 : weight = 1;
3481 : }
3482 :
3483 50 : return weight;
3484 : }
3485 :
3486 : /* append a script to the list of scripts to process */
3487 : static void
3488 179 : addScript(ParsedScript script)
3489 : {
3490 179 : if (script.commands == NULL || script.commands[0] == NULL)
3491 : {
3492 1 : fprintf(stderr, "empty command list for script \"%s\"\n", script.desc);
3493 1 : exit(1);
3494 : }
3495 :
3496 178 : if (num_scripts >= MAX_SCRIPTS)
3497 : {
3498 1 : fprintf(stderr, "at most %d SQL scripts are allowed\n", MAX_SCRIPTS);
3499 1 : exit(1);
3500 : }
3501 :
3502 177 : sql_script[num_scripts] = script;
3503 177 : num_scripts++;
3504 177 : }
3505 :
3506 : static void
3507 7 : printSimpleStats(char *prefix, SimpleStats *ss)
3508 : {
3509 : /* print NaN if no transactions where executed */
3510 7 : double latency = ss->sum / ss->count;
3511 7 : double stddev = sqrt(ss->sum2 / ss->count - latency * latency);
3512 :
3513 7 : printf("%s average = %.3f ms\n", prefix, 0.001 * latency);
3514 7 : printf("%s stddev = %.3f ms\n", prefix, 0.001 * stddev);
3515 7 : }
3516 :
3517 : /* print out results */
3518 : static void
3519 31 : printResults(TState *threads, StatsData *total, instr_time total_time,
3520 : instr_time conn_total_time, int latency_late)
3521 : {
3522 : double time_include,
3523 : tps_include,
3524 : tps_exclude;
3525 :
3526 31 : time_include = INSTR_TIME_GET_DOUBLE(total_time);
3527 31 : tps_include = total->cnt / time_include;
3528 62 : tps_exclude = total->cnt / (time_include -
3529 31 : (INSTR_TIME_GET_DOUBLE(conn_total_time) / nclients));
3530 :
3531 : /* Report test parameters. */
3532 31 : printf("transaction type: %s\n",
3533 31 : num_scripts == 1 ? sql_script[0].desc : "multiple scripts");
3534 31 : printf("scaling factor: %d\n", scale);
3535 31 : printf("query mode: %s\n", QUERYMODE[querymode]);
3536 31 : printf("number of clients: %d\n", nclients);
3537 31 : printf("number of threads: %d\n", nthreads);
3538 31 : if (duration <= 0)
3539 : {
3540 30 : printf("number of transactions per client: %d\n", nxacts);
3541 60 : printf("number of transactions actually processed: " INT64_FORMAT "/%d\n",
3542 30 : total->cnt - total->skipped, nxacts * nclients);
3543 : }
3544 : else
3545 : {
3546 1 : printf("duration: %d s\n", duration);
3547 1 : printf("number of transactions actually processed: " INT64_FORMAT "\n",
3548 : total->cnt);
3549 : }
3550 :
3551 : /* Remaining stats are nonsensical if we failed to execute any xacts */
3552 31 : if (total->cnt <= 0)
3553 48 : return;
3554 :
3555 14 : if (throttle_delay && latency_limit)
3556 3 : printf("number of transactions skipped: " INT64_FORMAT " (%.3f %%)\n",
3557 : total->skipped,
3558 3 : 100.0 * total->skipped / total->cnt);
3559 :
3560 14 : if (latency_limit)
3561 3 : printf("number of transactions above the %.1f ms latency limit: %d (%.3f %%)\n",
3562 : latency_limit / 1000.0, latency_late,
3563 3 : 100.0 * latency_late / total->cnt);
3564 :
3565 14 : if (throttle_delay || progress || latency_limit)
3566 3 : printSimpleStats("latency", &total->latency);
3567 : else
3568 : {
3569 : /* no measurement, show average latency computed from run time */
3570 11 : printf("latency average = %.3f ms\n",
3571 11 : 1000.0 * time_include * nclients / total->cnt);
3572 : }
3573 :
3574 14 : if (throttle_delay)
3575 : {
3576 : /*
3577 : * Report average transaction lag under rate limit throttling. This
3578 : * is the delay between scheduled and actual start times for the
3579 : * transaction. The measured lag may be caused by thread/client load,
3580 : * the database load, or the Poisson throttling process.
3581 : */
3582 6 : printf("rate limit schedule lag: avg %.3f (max %.3f) ms\n",
3583 6 : 0.001 * total->lag.sum / total->cnt, 0.001 * total->lag.max);
3584 : }
3585 :
3586 14 : printf("tps = %f (including connections establishing)\n", tps_include);
3587 14 : printf("tps = %f (excluding connections establishing)\n", tps_exclude);
3588 :
3589 : /* Report per-script/command statistics */
3590 14 : if (per_script_stats || latency_limit || is_latencies)
3591 : {
3592 : int i;
3593 :
3594 10 : for (i = 0; i < num_scripts; i++)
3595 : {
3596 6 : if (num_scripts > 1)
3597 12 : printf("SQL script %d: %s\n"
3598 : " - weight: %d (targets %.1f%% of total)\n"
3599 : " - " INT64_FORMAT " transactions (%.1f%% of total, tps = %f)\n",
3600 : i + 1, sql_script[i].desc,
3601 : sql_script[i].weight,
3602 4 : 100.0 * sql_script[i].weight / total_weight,
3603 : sql_script[i].stats.cnt,
3604 4 : 100.0 * sql_script[i].stats.cnt / total->cnt,
3605 4 : sql_script[i].stats.cnt / time_include);
3606 : else
3607 2 : printf("script statistics:\n");
3608 :
3609 6 : if (latency_limit)
3610 4 : printf(" - number of transactions skipped: " INT64_FORMAT " (%.3f%%)\n",
3611 : sql_script[i].stats.skipped,
3612 4 : 100.0 * sql_script[i].stats.skipped / sql_script[i].stats.cnt);
3613 :
3614 6 : if (num_scripts > 1)
3615 4 : printSimpleStats(" - latency", &sql_script[i].stats.latency);
3616 :
3617 : /* Report per-command latencies */
3618 6 : if (is_latencies)
3619 : {
3620 : Command **commands;
3621 :
3622 4 : printf(" - statement latencies in milliseconds:\n");
3623 :
3624 15 : for (commands = sql_script[i].commands;
3625 11 : *commands != NULL;
3626 7 : commands++)
3627 14 : printf(" %11.3f %s\n",
3628 7 : 1000.0 * (*commands)->stats.sum /
3629 7 : (*commands)->stats.count,
3630 7 : (*commands)->line);
3631 : }
3632 : }
3633 : }
3634 : }
3635 :
3636 :
3637 : int
3638 85 : main(int argc, char **argv)
3639 : {
3640 : static struct option long_options[] = {
3641 : /* systematic long/short named options */
3642 : {"builtin", required_argument, NULL, 'b'},
3643 : {"client", required_argument, NULL, 'c'},
3644 : {"connect", no_argument, NULL, 'C'},
3645 : {"debug", no_argument, NULL, 'd'},
3646 : {"define", required_argument, NULL, 'D'},
3647 : {"file", required_argument, NULL, 'f'},
3648 : {"fillfactor", required_argument, NULL, 'F'},
3649 : {"host", required_argument, NULL, 'h'},
3650 : {"initialize", no_argument, NULL, 'i'},
3651 : {"jobs", required_argument, NULL, 'j'},
3652 : {"log", no_argument, NULL, 'l'},
3653 : {"latency-limit", required_argument, NULL, 'L'},
3654 : {"no-vacuum", no_argument, NULL, 'n'},
3655 : {"port", required_argument, NULL, 'p'},
3656 : {"progress", required_argument, NULL, 'P'},
3657 : {"protocol", required_argument, NULL, 'M'},
3658 : {"quiet", no_argument, NULL, 'q'},
3659 : {"report-latencies", no_argument, NULL, 'r'},
3660 : {"rate", required_argument, NULL, 'R'},
3661 : {"scale", required_argument, NULL, 's'},
3662 : {"select-only", no_argument, NULL, 'S'},
3663 : {"skip-some-updates", no_argument, NULL, 'N'},
3664 : {"time", required_argument, NULL, 'T'},
3665 : {"transactions", required_argument, NULL, 't'},
3666 : {"username", required_argument, NULL, 'U'},
3667 : {"vacuum-all", no_argument, NULL, 'v'},
3668 : /* long-named only options */
3669 : {"foreign-keys", no_argument, &foreign_keys, 1},
3670 : {"index-tablespace", required_argument, NULL, 3},
3671 : {"tablespace", required_argument, NULL, 2},
3672 : {"unlogged-tables", no_argument, &unlogged_tables, 1},
3673 : {"sampling-rate", required_argument, NULL, 4},
3674 : {"aggregate-interval", required_argument, NULL, 5},
3675 : {"progress-timestamp", no_argument, NULL, 6},
3676 : {"log-prefix", required_argument, NULL, 7},
3677 : {NULL, 0, NULL, 0}
3678 : };
3679 :
3680 : int c;
3681 85 : int is_init_mode = 0; /* initialize mode? */
3682 85 : int is_no_vacuum = 0; /* no vacuum at all before testing? */
3683 85 : int do_vacuum_accounts = 0; /* do vacuum accounts before testing? */
3684 : int optindex;
3685 85 : bool scale_given = false;
3686 :
3687 85 : bool benchmarking_option_set = false;
3688 85 : bool initialization_option_set = false;
3689 85 : bool internal_script_used = false;
3690 :
3691 : CState *state; /* status of clients */
3692 : TState *threads; /* array of thread */
3693 :
3694 : instr_time start_time; /* start up time */
3695 : instr_time total_time;
3696 : instr_time conn_total_time;
3697 85 : int64 latency_late = 0;
3698 : StatsData stats;
3699 : int weight;
3700 :
3701 : int i;
3702 : int nclients_dealt;
3703 :
3704 : #ifdef HAVE_GETRLIMIT
3705 : struct rlimit rlim;
3706 : #endif
3707 :
3708 : PGconn *con;
3709 : PGresult *res;
3710 : char *env;
3711 :
3712 85 : progname = get_progname(argv[0]);
3713 :
3714 85 : if (argc > 1)
3715 : {
3716 85 : if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
3717 : {
3718 1 : usage();
3719 1 : exit(0);
3720 : }
3721 84 : if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
3722 : {
3723 1 : puts("pgbench (PostgreSQL) " PG_VERSION);
3724 1 : exit(0);
3725 : }
3726 : }
3727 :
3728 : #ifdef WIN32
3729 : /* stderr is buffered on Win32. */
3730 : setvbuf(stderr, NULL, _IONBF, 0);
3731 : #endif
3732 :
3733 83 : if ((env = getenv("PGHOST")) != NULL && *env != '\0')
3734 50 : pghost = env;
3735 83 : if ((env = getenv("PGPORT")) != NULL && *env != '\0')
3736 49 : pgport = env;
3737 34 : else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
3738 0 : login = env;
3739 :
3740 83 : state = (CState *) pg_malloc(sizeof(CState));
3741 83 : memset(state, 0, sizeof(CState));
3742 :
3743 571 : while ((c = getopt_long(argc, argv, "ih:nvp:dqb:SNc:j:Crs:t:T:U:lf:D:F:M:P:R:L:", long_options, &optindex)) != -1)
3744 : {
3745 : char *script;
3746 :
3747 442 : switch (c)
3748 : {
3749 : case 'i':
3750 4 : is_init_mode++;
3751 4 : break;
3752 : case 'h':
3753 1 : pghost = pg_strdup(optarg);
3754 1 : break;
3755 : case 'n':
3756 42 : is_no_vacuum++;
3757 42 : break;
3758 : case 'v':
3759 1 : do_vacuum_accounts++;
3760 1 : break;
3761 : case 'p':
3762 1 : pgport = pg_strdup(optarg);
3763 1 : break;
3764 : case 'd':
3765 2 : debug++;
3766 2 : break;
3767 : case 'c':
3768 11 : benchmarking_option_set = true;
3769 11 : nclients = atoi(optarg);
3770 11 : if (nclients <= 0 || nclients > MAXCLIENTS)
3771 : {
3772 1 : fprintf(stderr, "invalid number of clients: \"%s\"\n",
3773 : optarg);
3774 1 : exit(1);
3775 : }
3776 : #ifdef HAVE_GETRLIMIT
3777 : #ifdef RLIMIT_NOFILE /* most platforms use RLIMIT_NOFILE */
3778 10 : if (getrlimit(RLIMIT_NOFILE, &rlim) == -1)
3779 : #else /* but BSD doesn't ... */
3780 : if (getrlimit(RLIMIT_OFILE, &rlim) == -1)
3781 : #endif /* RLIMIT_NOFILE */
3782 : {
3783 0 : fprintf(stderr, "getrlimit failed: %s\n", strerror(errno));
3784 0 : exit(1);
3785 : }
3786 10 : if (rlim.rlim_cur < nclients + 3)
3787 : {
3788 0 : fprintf(stderr, "need at least %d open files, but system limit is %ld\n",
3789 0 : nclients + 3, (long) rlim.rlim_cur);
3790 0 : fprintf(stderr, "Reduce number of clients, or use limit/ulimit to increase the system limit.\n");
3791 0 : exit(1);
3792 : }
3793 : #endif /* HAVE_GETRLIMIT */
3794 10 : break;
3795 : case 'j': /* jobs */
3796 4 : benchmarking_option_set = true;
3797 4 : nthreads = atoi(optarg);
3798 4 : if (nthreads <= 0)
3799 : {
3800 1 : fprintf(stderr, "invalid number of threads: \"%s\"\n",
3801 : optarg);
3802 1 : exit(1);
3803 : }
3804 : #ifndef ENABLE_THREAD_SAFETY
3805 : if (nthreads != 1)
3806 : {
3807 : fprintf(stderr, "threads are not supported on this platform; use -j1\n");
3808 : exit(1);
3809 : }
3810 : #endif /* !ENABLE_THREAD_SAFETY */
3811 3 : break;
3812 : case 'C':
3813 2 : benchmarking_option_set = true;
3814 2 : is_connect = true;
3815 2 : break;
3816 : case 'r':
3817 3 : benchmarking_option_set = true;
3818 3 : per_script_stats = true;
3819 3 : is_latencies = true;
3820 3 : break;
3821 : case 's':
3822 3 : scale_given = true;
3823 3 : scale = atoi(optarg);
3824 3 : if (scale <= 0)
3825 : {
3826 1 : fprintf(stderr, "invalid scaling factor: \"%s\"\n", optarg);
3827 1 : exit(1);
3828 : }
3829 2 : break;
3830 : case 't':
3831 48 : benchmarking_option_set = true;
3832 48 : if (duration > 0)
3833 : {
3834 1 : fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both\n");
3835 1 : exit(1);
3836 : }
3837 47 : nxacts = atoi(optarg);
3838 47 : if (nxacts <= 0)
3839 : {
3840 1 : fprintf(stderr, "invalid number of transactions: \"%s\"\n",
3841 : optarg);
3842 1 : exit(1);
3843 : }
3844 46 : break;
3845 : case 'T':
3846 6 : benchmarking_option_set = true;
3847 6 : if (nxacts > 0)
3848 : {
3849 1 : fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both\n");
3850 1 : exit(1);
3851 : }
3852 5 : duration = atoi(optarg);
3853 5 : if (duration <= 0)
3854 : {
3855 1 : fprintf(stderr, "invalid duration: \"%s\"\n", optarg);
3856 1 : exit(1);
3857 : }
3858 4 : break;
3859 : case 'U':
3860 1 : login = pg_strdup(optarg);
3861 1 : break;
3862 : case 'l':
3863 8 : benchmarking_option_set = true;
3864 8 : use_log = true;
3865 8 : break;
3866 : case 'q':
3867 1 : initialization_option_set = true;
3868 1 : use_quiet = true;
3869 1 : break;
3870 :
3871 : case 'b':
3872 13 : if (strcmp(optarg, "list") == 0)
3873 : {
3874 1 : listAvailableScripts();
3875 1 : exit(0);
3876 : }
3877 :
3878 12 : weight = parseScriptWeight(optarg, &script);
3879 10 : process_builtin(findBuiltin(script), weight);
3880 8 : benchmarking_option_set = true;
3881 8 : internal_script_used = true;
3882 8 : break;
3883 :
3884 : case 'S':
3885 135 : process_builtin(findBuiltin("select-only"), 1);
3886 134 : benchmarking_option_set = true;
3887 134 : internal_script_used = true;
3888 134 : break;
3889 : case 'N':
3890 1 : process_builtin(findBuiltin("simple-update"), 1);
3891 1 : benchmarking_option_set = true;
3892 1 : internal_script_used = true;
3893 1 : break;
3894 : case 'f':
3895 40 : weight = parseScriptWeight(optarg, &script);
3896 40 : process_file(script, weight);
3897 26 : benchmarking_option_set = true;
3898 26 : break;
3899 : case 'D':
3900 : {
3901 : char *p;
3902 :
3903 37 : benchmarking_option_set = true;
3904 :
3905 37 : if ((p = strchr(optarg, '=')) == NULL || p == optarg || *(p + 1) == '\0')
3906 : {
3907 1 : fprintf(stderr, "invalid variable definition: \"%s\"\n",
3908 : optarg);
3909 1 : exit(1);
3910 : }
3911 :
3912 36 : *p++ = '\0';
3913 36 : if (!putVariable(&state[0], "option", optarg, p))
3914 0 : exit(1);
3915 : }
3916 36 : break;
3917 : case 'F':
3918 3 : initialization_option_set = true;
3919 3 : fillfactor = atoi(optarg);
3920 3 : if (fillfactor < 10 || fillfactor > 100)
3921 : {
3922 1 : fprintf(stderr, "invalid fillfactor: \"%s\"\n", optarg);
3923 1 : exit(1);
3924 : }
3925 2 : break;
3926 : case 'M':
3927 39 : benchmarking_option_set = true;
3928 112 : for (querymode = 0; querymode < NUM_QUERYMODE; querymode++)
3929 111 : if (strcmp(optarg, QUERYMODE[querymode]) == 0)
3930 38 : break;
3931 39 : if (querymode >= NUM_QUERYMODE)
3932 : {
3933 1 : fprintf(stderr, "invalid query mode (-M): \"%s\"\n",
3934 : optarg);
3935 1 : exit(1);
3936 : }
3937 38 : break;
3938 : case 'P':
3939 3 : benchmarking_option_set = true;
3940 3 : progress = atoi(optarg);
3941 3 : if (progress <= 0)
3942 : {
3943 1 : fprintf(stderr, "invalid thread progress delay: \"%s\"\n",
3944 : optarg);
3945 1 : exit(1);
3946 : }
3947 2 : break;
3948 : case 'R':
3949 : {
3950 : /* get a double from the beginning of option value */
3951 4 : double throttle_value = atof(optarg);
3952 :
3953 4 : benchmarking_option_set = true;
3954 :
3955 4 : if (throttle_value <= 0.0)
3956 : {
3957 1 : fprintf(stderr, "invalid rate limit: \"%s\"\n", optarg);
3958 1 : exit(1);
3959 : }
3960 : /* Invert rate limit into a time offset */
3961 3 : throttle_delay = (int64) (1000000.0 / throttle_value);
3962 : }
3963 3 : break;
3964 : case 'L':
3965 : {
3966 4 : double limit_ms = atof(optarg);
3967 :
3968 4 : if (limit_ms <= 0.0)
3969 : {
3970 1 : fprintf(stderr, "invalid latency limit: \"%s\"\n",
3971 : optarg);
3972 1 : exit(1);
3973 : }
3974 3 : benchmarking_option_set = true;
3975 3 : latency_limit = (int64) (limit_ms * 1000);
3976 : }
3977 3 : break;
3978 : case 0:
3979 : /* This covers long options which take no argument. */
3980 2 : if (foreign_keys || unlogged_tables)
3981 2 : initialization_option_set = true;
3982 2 : break;
3983 : case 2: /* tablespace */
3984 1 : initialization_option_set = true;
3985 1 : tablespace = pg_strdup(optarg);
3986 1 : break;
3987 : case 3: /* index-tablespace */
3988 1 : initialization_option_set = true;
3989 1 : index_tablespace = pg_strdup(optarg);
3990 1 : break;
3991 : case 4:
3992 5 : benchmarking_option_set = true;
3993 5 : sample_rate = atof(optarg);
3994 5 : if (sample_rate <= 0.0 || sample_rate > 1.0)
3995 : {
3996 1 : fprintf(stderr, "invalid sampling rate: \"%s\"\n", optarg);
3997 1 : exit(1);
3998 : }
3999 4 : break;
4000 : case 5:
4001 7 : benchmarking_option_set = true;
4002 7 : agg_interval = atoi(optarg);
4003 7 : if (agg_interval <= 0)
4004 : {
4005 1 : fprintf(stderr, "invalid number of seconds for aggregation: \"%s\"\n",
4006 : optarg);
4007 1 : exit(1);
4008 : }
4009 6 : break;
4010 : case 6:
4011 2 : progress_timestamp = true;
4012 2 : benchmarking_option_set = true;
4013 2 : break;
4014 : case 7:
4015 5 : benchmarking_option_set = true;
4016 5 : logfile_prefix = pg_strdup(optarg);
4017 5 : break;
4018 : default:
4019 2 : fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
4020 2 : exit(1);
4021 : break;
4022 : }
4023 : }
4024 :
4025 : /* set default script if none */
4026 46 : if (num_scripts == 0 && !is_init_mode)
4027 : {
4028 8 : process_builtin(findBuiltin("tpcb-like"), 1);
4029 8 : benchmarking_option_set = true;
4030 8 : internal_script_used = true;
4031 : }
4032 :
4033 : /* if not simple query mode, parse the script(s) to find parameters */
4034 46 : if (querymode != QUERY_SIMPLE)
4035 : {
4036 46 : for (i = 0; i < num_scripts; i++)
4037 : {
4038 24 : Command **commands = sql_script[i].commands;
4039 : int j;
4040 :
4041 64 : for (j = 0; commands[j] != NULL; j++)
4042 : {
4043 41 : if (commands[j]->type != SQL_COMMAND)
4044 25 : continue;
4045 16 : if (!parseQuery(commands[j]))
4046 1 : exit(1);
4047 : }
4048 : }
4049 : }
4050 :
4051 : /* compute total_weight */
4052 92 : for (i = 0; i < num_scripts; i++)
4053 : /* cannot overflow: weight is 32b, total_weight 64b */
4054 47 : total_weight += sql_script[i].weight;
4055 :
4056 45 : if (total_weight == 0 && !is_init_mode)
4057 : {
4058 1 : fprintf(stderr, "total script weight must not be zero\n");
4059 1 : exit(1);
4060 : }
4061 :
4062 : /* show per script stats if several scripts are used */
4063 44 : if (num_scripts > 1)
4064 2 : per_script_stats = true;
4065 :
4066 : /*
4067 : * Don't need more threads than there are clients. (This is not merely an
4068 : * optimization; throttle_delay is calculated incorrectly below if some
4069 : * threads have no clients assigned to them.)
4070 : */
4071 44 : if (nthreads > nclients)
4072 1 : nthreads = nclients;
4073 :
4074 : /* compute a per thread delay */
4075 44 : throttle_delay *= nthreads;
4076 :
4077 44 : if (argc > optind)
4078 1 : dbName = argv[optind];
4079 : else
4080 : {
4081 43 : if ((env = getenv("PGDATABASE")) != NULL && *env != '\0')
4082 34 : dbName = env;
4083 9 : else if (login != NULL && *login != '\0')
4084 0 : dbName = login;
4085 : else
4086 9 : dbName = "";
4087 : }
4088 :
4089 44 : if (is_init_mode)
4090 : {
4091 3 : if (benchmarking_option_set)
4092 : {
4093 1 : fprintf(stderr, "some of the specified options cannot be used in initialization (-i) mode\n");
4094 1 : exit(1);
4095 : }
4096 :
4097 2 : init(is_no_vacuum);
4098 2 : exit(0);
4099 : }
4100 : else
4101 : {
4102 41 : if (initialization_option_set)
4103 : {
4104 1 : fprintf(stderr, "some of the specified options cannot be used in benchmarking mode\n");
4105 1 : exit(1);
4106 : }
4107 : }
4108 :
4109 : /* Use DEFAULT_NXACTS if neither nxacts nor duration is specified. */
4110 40 : if (nxacts <= 0 && duration <= 0)
4111 6 : nxacts = DEFAULT_NXACTS;
4112 :
4113 : /* --sampling-rate may be used only with -l */
4114 40 : if (sample_rate > 0.0 && !use_log)
4115 : {
4116 1 : fprintf(stderr, "log sampling (--sampling-rate) is allowed only when logging transactions (-l)\n");
4117 1 : exit(1);
4118 : }
4119 :
4120 : /* --sampling-rate may not be used with --aggregate-interval */
4121 39 : if (sample_rate > 0.0 && agg_interval > 0)
4122 : {
4123 1 : fprintf(stderr, "log sampling (--sampling-rate) and aggregation (--aggregate-interval) cannot be used at the same time\n");
4124 1 : exit(1);
4125 : }
4126 :
4127 38 : if (agg_interval > 0 && !use_log)
4128 : {
4129 1 : fprintf(stderr, "log aggregation is allowed only when actually logging transactions\n");
4130 1 : exit(1);
4131 : }
4132 :
4133 37 : if (!use_log && logfile_prefix)
4134 : {
4135 1 : fprintf(stderr, "log file prefix (--log-prefix) is allowed only when logging transactions (-l)\n");
4136 1 : exit(1);
4137 : }
4138 :
4139 36 : if (duration > 0 && agg_interval > duration)
4140 : {
4141 1 : fprintf(stderr, "number of seconds for aggregation (%d) must not be higher than test duration (%d)\n", agg_interval, duration);
4142 1 : exit(1);
4143 : }
4144 :
4145 35 : if (duration > 0 && agg_interval > 0 && duration % agg_interval != 0)
4146 : {
4147 1 : fprintf(stderr, "duration (%d) must be a multiple of aggregation interval (%d)\n", duration, agg_interval);
4148 1 : exit(1);
4149 : }
4150 :
4151 34 : if (progress_timestamp && progress == 0)
4152 : {
4153 1 : fprintf(stderr, "--progress-timestamp is allowed only under --progress\n");
4154 1 : exit(1);
4155 : }
4156 :
4157 : /*
4158 : * save main process id in the global variable because process id will be
4159 : * changed after fork.
4160 : */
4161 33 : main_pid = (int) getpid();
4162 :
4163 33 : if (nclients > 1)
4164 : {
4165 8 : state = (CState *) pg_realloc(state, sizeof(CState) * nclients);
4166 8 : memset(state + 1, 0, sizeof(CState) * (nclients - 1));
4167 :
4168 : /* copy any -D switch values to all clients */
4169 28 : for (i = 1; i < nclients; i++)
4170 : {
4171 : int j;
4172 :
4173 20 : state[i].id = i;
4174 21 : for (j = 0; j < state[0].nvariables; j++)
4175 : {
4176 1 : Variable *var = &state[0].variables[j];
4177 :
4178 1 : if (var->is_numeric)
4179 : {
4180 0 : if (!putVariableNumber(&state[i], "startup",
4181 0 : var->name, &var->num_value))
4182 0 : exit(1);
4183 : }
4184 : else
4185 : {
4186 1 : if (!putVariable(&state[i], "startup",
4187 1 : var->name, var->value))
4188 0 : exit(1);
4189 : }
4190 : }
4191 : }
4192 : }
4193 :
4194 33 : if (debug)
4195 : {
4196 1 : if (duration <= 0)
4197 1 : printf("pghost: %s pgport: %s nclients: %d nxacts: %d dbName: %s\n",
4198 : pghost, pgport, nclients, nxacts, dbName);
4199 : else
4200 0 : printf("pghost: %s pgport: %s nclients: %d duration: %d dbName: %s\n",
4201 : pghost, pgport, nclients, duration, dbName);
4202 : }
4203 :
4204 : /* opening connection... */
4205 33 : con = doConnect();
4206 33 : if (con == NULL)
4207 1 : exit(1);
4208 :
4209 32 : if (PQstatus(con) == CONNECTION_BAD)
4210 : {
4211 0 : fprintf(stderr, "connection to database \"%s\" failed\n", dbName);
4212 0 : fprintf(stderr, "%s", PQerrorMessage(con));
4213 0 : exit(1);
4214 : }
4215 :
4216 32 : if (internal_script_used)
4217 : {
4218 : /*
4219 : * get the scaling factor that should be same as count(*) from
4220 : * pgbench_branches if this is not a custom query
4221 : */
4222 8 : res = PQexec(con, "select count(*) from pgbench_branches");
4223 8 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
4224 : {
4225 1 : char *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
4226 :
4227 1 : fprintf(stderr, "%s", PQerrorMessage(con));
4228 1 : if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) == 0)
4229 : {
4230 1 : fprintf(stderr, "Perhaps you need to do initialization (\"pgbench -i\") in database \"%s\"\n", PQdb(con));
4231 : }
4232 :
4233 1 : exit(1);
4234 : }
4235 7 : scale = atoi(PQgetvalue(res, 0, 0));
4236 7 : if (scale < 0)
4237 : {
4238 0 : fprintf(stderr, "invalid count(*) from pgbench_branches: \"%s\"\n",
4239 : PQgetvalue(res, 0, 0));
4240 0 : exit(1);
4241 : }
4242 7 : PQclear(res);
4243 :
4244 : /* warn if we override user-given -s switch */
4245 7 : if (scale_given)
4246 1 : fprintf(stderr,
4247 : "scale option ignored, using count from pgbench_branches table (%d)\n",
4248 : scale);
4249 : }
4250 :
4251 : /*
4252 : * :scale variables normally get -s or database scale, but don't override
4253 : * an explicit -D switch
4254 : */
4255 31 : if (lookupVariable(&state[0], "scale") == NULL)
4256 : {
4257 82 : for (i = 0; i < nclients; i++)
4258 : {
4259 51 : if (!putVariableInt(&state[i], "startup", "scale", scale))
4260 0 : exit(1);
4261 : }
4262 : }
4263 :
4264 : /*
4265 : * Define a :client_id variable that is unique per connection. But don't
4266 : * override an explicit -D switch.
4267 : */
4268 31 : if (lookupVariable(&state[0], "client_id") == NULL)
4269 : {
4270 82 : for (i = 0; i < nclients; i++)
4271 : {
4272 51 : if (!putVariableInt(&state[i], "startup", "client_id", i))
4273 0 : exit(1);
4274 : }
4275 : }
4276 :
4277 31 : if (!is_no_vacuum)
4278 : {
4279 4 : fprintf(stderr, "starting vacuum...");
4280 4 : tryExecuteStatement(con, "vacuum pgbench_branches");
4281 4 : tryExecuteStatement(con, "vacuum pgbench_tellers");
4282 4 : tryExecuteStatement(con, "truncate pgbench_history");
4283 4 : fprintf(stderr, "end.\n");
4284 :
4285 4 : if (do_vacuum_accounts)
4286 : {
4287 0 : fprintf(stderr, "starting vacuum pgbench_accounts...");
4288 0 : tryExecuteStatement(con, "vacuum analyze pgbench_accounts");
4289 0 : fprintf(stderr, "end.\n");
4290 : }
4291 : }
4292 31 : PQfinish(con);
4293 :
4294 : /* set random seed */
4295 31 : INSTR_TIME_SET_CURRENT(start_time);
4296 31 : srandom((unsigned int) INSTR_TIME_GET_MICROSEC(start_time));
4297 :
4298 : /* set up thread data structures */
4299 31 : threads = (TState *) pg_malloc(sizeof(TState) * nthreads);
4300 31 : nclients_dealt = 0;
4301 :
4302 63 : for (i = 0; i < nthreads; i++)
4303 : {
4304 32 : TState *thread = &threads[i];
4305 :
4306 32 : thread->tid = i;
4307 32 : thread->state = &state[nclients_dealt];
4308 32 : thread->nstate =
4309 32 : (nclients - nclients_dealt + nthreads - i - 1) / (nthreads - i);
4310 32 : thread->random_state[0] = random();
4311 32 : thread->random_state[1] = random();
4312 32 : thread->random_state[2] = random();
4313 32 : thread->logfile = NULL; /* filled in later */
4314 32 : thread->latency_late = 0;
4315 32 : initStats(&thread->stats, 0);
4316 :
4317 32 : nclients_dealt += thread->nstate;
4318 : }
4319 :
4320 : /* all clients must be assigned to a thread */
4321 31 : Assert(nclients_dealt == nclients);
4322 :
4323 : /* get start up time */
4324 31 : INSTR_TIME_SET_CURRENT(start_time);
4325 :
4326 : /* set alarm if duration is specified. */
4327 31 : if (duration > 0)
4328 1 : setalarm(duration);
4329 :
4330 : /* start threads */
4331 : #ifdef ENABLE_THREAD_SAFETY
4332 63 : for (i = 0; i < nthreads; i++)
4333 : {
4334 32 : TState *thread = &threads[i];
4335 :
4336 32 : INSTR_TIME_SET_CURRENT(thread->start_time);
4337 :
4338 : /* compute when to stop */
4339 32 : if (duration > 0)
4340 4 : end_time = INSTR_TIME_GET_MICROSEC(thread->start_time) +
4341 2 : (int64) 1000000 * duration;
4342 :
4343 : /* the first thread (i = 0) is executed by main thread */
4344 32 : if (i > 0)
4345 : {
4346 1 : int err = pthread_create(&thread->thread, NULL, threadRun, thread);
4347 :
4348 1 : if (err != 0 || thread->thread == INVALID_THREAD)
4349 : {
4350 0 : fprintf(stderr, "could not create thread: %s\n", strerror(err));
4351 0 : exit(1);
4352 : }
4353 : }
4354 : else
4355 : {
4356 31 : thread->thread = INVALID_THREAD;
4357 : }
4358 : }
4359 : #else
4360 : INSTR_TIME_SET_CURRENT(threads[0].start_time);
4361 : /* compute when to stop */
4362 : if (duration > 0)
4363 : end_time = INSTR_TIME_GET_MICROSEC(threads[0].start_time) +
4364 : (int64) 1000000 * duration;
4365 : threads[0].thread = INVALID_THREAD;
4366 : #endif /* ENABLE_THREAD_SAFETY */
4367 :
4368 : /* wait for threads and accumulate results */
4369 31 : initStats(&stats, 0);
4370 31 : INSTR_TIME_SET_ZERO(conn_total_time);
4371 63 : for (i = 0; i < nthreads; i++)
4372 : {
4373 32 : TState *thread = &threads[i];
4374 :
4375 : #ifdef ENABLE_THREAD_SAFETY
4376 32 : if (threads[i].thread == INVALID_THREAD)
4377 : /* actually run this thread directly in the main thread */
4378 31 : (void) threadRun(thread);
4379 : else
4380 : /* wait of other threads. should check that 0 is returned? */
4381 1 : pthread_join(thread->thread, NULL);
4382 : #else
4383 : (void) threadRun(thread);
4384 : #endif /* ENABLE_THREAD_SAFETY */
4385 :
4386 : /* aggregate thread level stats */
4387 32 : mergeSimpleStats(&stats.latency, &thread->stats.latency);
4388 32 : mergeSimpleStats(&stats.lag, &thread->stats.lag);
4389 32 : stats.cnt += thread->stats.cnt;
4390 32 : stats.skipped += thread->stats.skipped;
4391 32 : latency_late += thread->latency_late;
4392 32 : INSTR_TIME_ADD(conn_total_time, thread->conn_time);
4393 : }
4394 31 : disconnect_all(state, nclients);
4395 :
4396 : /*
4397 : * XXX We compute results as though every client of every thread started
4398 : * and finished at the same time. That model can diverge noticeably from
4399 : * reality for a short benchmark run involving relatively many threads.
4400 : * The first thread may process notably many transactions before the last
4401 : * thread begins. Improving the model alone would bring limited benefit,
4402 : * because performance during those periods of partial thread count can
4403 : * easily exceed steady state performance. This is one of the many ways
4404 : * short runs convey deceptive performance figures.
4405 : */
4406 31 : INSTR_TIME_SET_CURRENT(total_time);
4407 31 : INSTR_TIME_SUBTRACT(total_time, start_time);
4408 31 : printResults(threads, &stats, total_time, conn_total_time, latency_late);
4409 :
4410 31 : return 0;
4411 : }
4412 :
4413 : static void *
4414 32 : threadRun(void *arg)
4415 : {
4416 32 : TState *thread = (TState *) arg;
4417 32 : CState *state = thread->state;
4418 : instr_time start,
4419 : end;
4420 32 : int nstate = thread->nstate;
4421 32 : int remains = nstate; /* number of remaining clients */
4422 : int i;
4423 :
4424 : /* for reporting progress: */
4425 32 : int64 thread_start = INSTR_TIME_GET_MICROSEC(thread->start_time);
4426 32 : int64 last_report = thread_start;
4427 32 : int64 next_report = last_report + (int64) progress * 1000000;
4428 : StatsData last,
4429 : aggs;
4430 :
4431 : /*
4432 : * Initialize throttling rate target for all of the thread's clients. It
4433 : * might be a little more accurate to reset thread->start_time here too.
4434 : * The possible drift seems too small relative to typical throttle delay
4435 : * times to worry about it.
4436 : */
4437 32 : INSTR_TIME_SET_CURRENT(start);
4438 32 : thread->throttle_trigger = INSTR_TIME_GET_MICROSEC(start);
4439 :
4440 32 : INSTR_TIME_SET_ZERO(thread->conn_time);
4441 :
4442 32 : initStats(&aggs, time(NULL));
4443 32 : last = aggs;
4444 :
4445 : /* open log file if requested */
4446 32 : if (use_log)
4447 : {
4448 : char logpath[MAXPGPATH];
4449 4 : char *prefix = logfile_prefix ? logfile_prefix : "pgbench_log";
4450 :
4451 4 : if (thread->tid == 0)
4452 3 : snprintf(logpath, sizeof(logpath), "%s.%d", prefix, main_pid);
4453 : else
4454 1 : snprintf(logpath, sizeof(logpath), "%s.%d.%d", prefix, main_pid, thread->tid);
4455 :
4456 4 : thread->logfile = fopen(logpath, "w");
4457 :
4458 4 : if (thread->logfile == NULL)
4459 : {
4460 0 : fprintf(stderr, "could not open logfile \"%s\": %s\n",
4461 0 : logpath, strerror(errno));
4462 0 : goto done;
4463 : }
4464 : }
4465 :
4466 32 : if (!is_connect)
4467 : {
4468 : /* make connections to the database */
4469 74 : for (i = 0; i < nstate; i++)
4470 : {
4471 44 : if ((state[i].con = doConnect()) == NULL)
4472 0 : goto done;
4473 : }
4474 : }
4475 :
4476 : /* time after thread and connections set up */
4477 32 : INSTR_TIME_SET_CURRENT(thread->conn_time);
4478 32 : INSTR_TIME_SUBTRACT(thread->conn_time, thread->start_time);
4479 :
4480 : /* explicitly initialize the state machines */
4481 83 : for (i = 0; i < nstate; i++)
4482 : {
4483 51 : state[i].state = CSTATE_CHOOSE_SCRIPT;
4484 : }
4485 :
4486 : /* loop till all clients have terminated */
4487 5503822 : while (remains > 0)
4488 : {
4489 : fd_set input_mask;
4490 : int maxsock; /* max socket number to be waited for */
4491 : int64 min_usec;
4492 5503758 : int64 now_usec = 0; /* set this only if needed */
4493 :
4494 : /* identify which client sockets should be checked for input */
4495 5503758 : FD_ZERO(&input_mask);
4496 5503758 : maxsock = -1;
4497 5503758 : min_usec = PG_INT64_MAX;
4498 12896797 : for (i = 0; i < nstate; i++)
4499 : {
4500 7409217 : CState *st = &state[i];
4501 :
4502 7409217 : if (st->state == CSTATE_THROTTLE && timer_exceeded)
4503 : {
4504 : /* interrupt client that has not started a transaction */
4505 0 : st->state = CSTATE_FINISHED;
4506 0 : PQfinish(st->con);
4507 0 : st->con = NULL;
4508 0 : remains--;
4509 : }
4510 7409217 : else if (st->state == CSTATE_SLEEP || st->state == CSTATE_THROTTLE)
4511 7230633 : {
4512 : /* a nap from the script, or under throttling */
4513 : int64 this_usec;
4514 :
4515 : /* get current time if needed */
4516 7230633 : if (now_usec == 0)
4517 : {
4518 : instr_time now;
4519 :
4520 5501990 : INSTR_TIME_SET_CURRENT(now);
4521 5501990 : now_usec = INSTR_TIME_GET_MICROSEC(now);
4522 : }
4523 :
4524 : /* min_usec should be the minimum delay across all clients */
4525 14461266 : this_usec = (st->state == CSTATE_SLEEP ?
4526 7230633 : st->sleep_until : st->txn_scheduled) - now_usec;
4527 7230633 : if (min_usec > this_usec)
4528 6373564 : min_usec = this_usec;
4529 : }
4530 178584 : else if (st->state == CSTATE_WAIT_RESULT)
4531 : {
4532 : /*
4533 : * waiting for result from server - nothing to do unless the
4534 : * socket is readable
4535 : */
4536 7328 : int sock = PQsocket(st->con);
4537 :
4538 7328 : if (sock < 0)
4539 : {
4540 0 : fprintf(stderr, "invalid socket: %s",
4541 0 : PQerrorMessage(st->con));
4542 0 : goto done;
4543 : }
4544 :
4545 7328 : FD_SET(sock, &input_mask);
4546 7328 : if (maxsock < sock)
4547 7322 : maxsock = sock;
4548 : }
4549 342512 : else if (st->state != CSTATE_ABORTED &&
4550 171256 : st->state != CSTATE_FINISHED)
4551 : {
4552 : /*
4553 : * This client thread is ready to do something, so we don't
4554 : * want to wait. No need to examine additional clients.
4555 : */
4556 32 : min_usec = 0;
4557 32 : break;
4558 : }
4559 : }
4560 :
4561 : /* also wake up to print the next progress report on time */
4562 5487612 : if (progress && min_usec > 0 && thread->tid == 0)
4563 : {
4564 : /* get current time if needed */
4565 1899550 : if (now_usec == 0)
4566 : {
4567 : instr_time now;
4568 :
4569 0 : INSTR_TIME_SET_CURRENT(now);
4570 0 : now_usec = INSTR_TIME_GET_MICROSEC(now);
4571 : }
4572 :
4573 1899550 : if (now_usec >= next_report)
4574 0 : min_usec = 0;
4575 1899550 : else if ((next_report - now_usec) < min_usec)
4576 171715 : min_usec = next_report - now_usec;
4577 : }
4578 :
4579 : /*
4580 : * If no clients are ready to execute actions, sleep until we receive
4581 : * data from the server, or a nap-time specified in the script ends,
4582 : * or it's time to print a progress report. Update input_mask to show
4583 : * which client(s) received data.
4584 : */
4585 5487612 : if (min_usec > 0 && maxsock != -1)
4586 1737 : {
4587 : int nsocks; /* return from select(2) */
4588 :
4589 1737 : if (min_usec != PG_INT64_MAX)
4590 : {
4591 : struct timeval timeout;
4592 :
4593 1 : timeout.tv_sec = min_usec / 1000000;
4594 1 : timeout.tv_usec = min_usec % 1000000;
4595 1 : nsocks = select(maxsock + 1, &input_mask, NULL, NULL, &timeout);
4596 : }
4597 : else
4598 1736 : nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
4599 1737 : if (nsocks < 0)
4600 : {
4601 0 : if (errno == EINTR)
4602 : {
4603 : /* On EINTR, go back to top of loop */
4604 0 : continue;
4605 : }
4606 : /* must be something wrong */
4607 0 : fprintf(stderr, "select() failed: %s\n", strerror(errno));
4608 0 : goto done;
4609 : }
4610 : }
4611 : else
4612 : {
4613 : /* If we didn't call select(), don't try to read any data */
4614 5485875 : FD_ZERO(&input_mask);
4615 : }
4616 :
4617 : /* ok, advance the state machine of each connection */
4618 12912994 : for (i = 0; i < nstate; i++)
4619 : {
4620 7409236 : CState *st = &state[i];
4621 :
4622 7409236 : if (st->state == CSTATE_WAIT_RESULT)
4623 : {
4624 : /* don't call doCustom unless data is available */
4625 7328 : int sock = PQsocket(st->con);
4626 :
4627 7328 : if (sock < 0)
4628 : {
4629 0 : fprintf(stderr, "invalid socket: %s",
4630 0 : PQerrorMessage(st->con));
4631 0 : goto done;
4632 : }
4633 :
4634 7328 : if (!FD_ISSET(sock, &input_mask))
4635 5303 : continue;
4636 : }
4637 14632592 : else if (st->state == CSTATE_FINISHED ||
4638 7230684 : st->state == CSTATE_ABORTED)
4639 : {
4640 : /* this client is done, no need to consider it anymore */
4641 171224 : continue;
4642 : }
4643 :
4644 7232709 : doCustom(thread, st, &aggs);
4645 :
4646 : /* If doCustom changed client to finished state, reduce remains */
4647 7232709 : if (st->state == CSTATE_FINISHED || st->state == CSTATE_ABORTED)
4648 51 : remains--;
4649 : }
4650 :
4651 : /* progress report is made by thread 0 for all threads */
4652 5503758 : if (progress && thread->tid == 0)
4653 : {
4654 : instr_time now_time;
4655 : int64 now;
4656 :
4657 1899560 : INSTR_TIME_SET_CURRENT(now_time);
4658 1899560 : now = INSTR_TIME_GET_MICROSEC(now_time);
4659 1899560 : if (now >= next_report)
4660 : {
4661 : /* generate and show report */
4662 : StatsData cur;
4663 1 : int64 run = now - last_report;
4664 : double tps,
4665 : total_run,
4666 : latency,
4667 : sqlat,
4668 : lag,
4669 : stdev;
4670 : char tbuf[64];
4671 :
4672 : /*
4673 : * Add up the statistics of all threads.
4674 : *
4675 : * XXX: No locking. There is no guarantee that we get an
4676 : * atomic snapshot of the transaction count and latencies, so
4677 : * these figures can well be off by a small amount. The
4678 : * progress is report's purpose is to give a quick overview of
4679 : * how the test is going, so that shouldn't matter too much.
4680 : * (If a read from a 64-bit integer is not atomic, you might
4681 : * get a "torn" read and completely bogus latencies though!)
4682 : */
4683 1 : initStats(&cur, 0);
4684 3 : for (i = 0; i < nthreads; i++)
4685 : {
4686 2 : mergeSimpleStats(&cur.latency, &thread[i].stats.latency);
4687 2 : mergeSimpleStats(&cur.lag, &thread[i].stats.lag);
4688 2 : cur.cnt += thread[i].stats.cnt;
4689 2 : cur.skipped += thread[i].stats.skipped;
4690 : }
4691 :
4692 1 : total_run = (now - thread_start) / 1000000.0;
4693 1 : tps = 1000000.0 * (cur.cnt - last.cnt) / run;
4694 2 : latency = 0.001 * (cur.latency.sum - last.latency.sum) /
4695 1 : (cur.cnt - last.cnt);
4696 2 : sqlat = 1.0 * (cur.latency.sum2 - last.latency.sum2)
4697 1 : / (cur.cnt - last.cnt);
4698 1 : stdev = 0.001 * sqrt(sqlat - 1000000.0 * latency * latency);
4699 2 : lag = 0.001 * (cur.lag.sum - last.lag.sum) /
4700 1 : (cur.cnt - last.cnt);
4701 :
4702 1 : if (progress_timestamp)
4703 : {
4704 : /*
4705 : * On some platforms the current system timestamp is
4706 : * available in now_time, but rather than get entangled
4707 : * with that, we just eat the cost of an extra syscall in
4708 : * all cases.
4709 : */
4710 : struct timeval tv;
4711 :
4712 0 : gettimeofday(&tv, NULL);
4713 0 : snprintf(tbuf, sizeof(tbuf), "%ld.%03ld s",
4714 0 : (long) tv.tv_sec, (long) (tv.tv_usec / 1000));
4715 : }
4716 : else
4717 1 : snprintf(tbuf, sizeof(tbuf), "%.1f s", total_run);
4718 :
4719 1 : fprintf(stderr,
4720 : "progress: %s, %.1f tps, lat %.3f ms stddev %.3f",
4721 : tbuf, tps, latency, stdev);
4722 :
4723 1 : if (throttle_delay)
4724 : {
4725 1 : fprintf(stderr, ", lag %.3f ms", lag);
4726 1 : if (latency_limit)
4727 1 : fprintf(stderr, ", " INT64_FORMAT " skipped",
4728 1 : cur.skipped - last.skipped);
4729 : }
4730 1 : fprintf(stderr, "\n");
4731 :
4732 1 : last = cur;
4733 1 : last_report = now;
4734 :
4735 : /*
4736 : * Ensure that the next report is in the future, in case
4737 : * pgbench/postgres got stuck somewhere.
4738 : */
4739 : do
4740 : {
4741 1 : next_report += (int64) progress * 1000000;
4742 1 : } while (now >= next_report);
4743 : }
4744 : }
4745 : }
4746 :
4747 : done:
4748 32 : INSTR_TIME_SET_CURRENT(start);
4749 32 : disconnect_all(state, nstate);
4750 32 : INSTR_TIME_SET_CURRENT(end);
4751 32 : INSTR_TIME_ACCUM_DIFF(thread->conn_time, end, start);
4752 32 : if (thread->logfile)
4753 : {
4754 4 : if (agg_interval > 0)
4755 : {
4756 : /* log aggregated but not yet reported transactions */
4757 2 : doLog(thread, state, &aggs, false, 0, 0);
4758 : }
4759 4 : fclose(thread->logfile);
4760 4 : thread->logfile = NULL;
4761 : }
4762 32 : return NULL;
4763 : }
4764 :
4765 : /*
4766 : * Support for duration option: set timer_exceeded after so many seconds.
4767 : */
4768 :
4769 : #ifndef WIN32
4770 :
4771 : static void
4772 0 : handle_sig_alarm(SIGNAL_ARGS)
4773 : {
4774 0 : timer_exceeded = true;
4775 0 : }
4776 :
4777 : static void
4778 1 : setalarm(int seconds)
4779 : {
4780 1 : pqsignal(SIGALRM, handle_sig_alarm);
4781 1 : alarm(seconds);
4782 1 : }
4783 :
4784 : #else /* WIN32 */
4785 :
4786 : static VOID CALLBACK
4787 : win32_timer_callback(PVOID lpParameter, BOOLEAN TimerOrWaitFired)
4788 : {
4789 : timer_exceeded = true;
4790 : }
4791 :
4792 : static void
4793 : setalarm(int seconds)
4794 : {
4795 : HANDLE queue;
4796 : HANDLE timer;
4797 :
4798 : /* This function will be called at most once, so we can cheat a bit. */
4799 : queue = CreateTimerQueue();
4800 : if (seconds > ((DWORD) -1) / 1000 ||
4801 : !CreateTimerQueueTimer(&timer, queue,
4802 : win32_timer_callback, NULL, seconds * 1000, 0,
4803 : WT_EXECUTEINTIMERTHREAD | WT_EXECUTEONLYONCE))
4804 : {
4805 : fprintf(stderr, "failed to set timer\n");
4806 : exit(1);
4807 : }
4808 : }
4809 :
4810 : /* partial pthread implementation for Windows */
4811 :
4812 : typedef struct win32_pthread
4813 : {
4814 : HANDLE handle;
4815 : void *(*routine) (void *);
4816 : void *arg;
4817 : void *result;
4818 : } win32_pthread;
4819 :
4820 : static unsigned __stdcall
4821 : win32_pthread_run(void *arg)
4822 : {
4823 : win32_pthread *th = (win32_pthread *) arg;
4824 :
4825 : th->result = th->routine(th->arg);
4826 :
4827 : return 0;
4828 : }
4829 :
4830 : static int
4831 : pthread_create(pthread_t *thread,
4832 : pthread_attr_t *attr,
4833 : void *(*start_routine) (void *),
4834 : void *arg)
4835 : {
4836 : int save_errno;
4837 : win32_pthread *th;
4838 :
4839 : th = (win32_pthread *) pg_malloc(sizeof(win32_pthread));
4840 : th->routine = start_routine;
4841 : th->arg = arg;
4842 : th->result = NULL;
4843 :
4844 : th->handle = (HANDLE) _beginthreadex(NULL, 0, win32_pthread_run, th, 0, NULL);
4845 : if (th->handle == NULL)
4846 : {
4847 : save_errno = errno;
4848 : free(th);
4849 : return save_errno;
4850 : }
4851 :
4852 : *thread = th;
4853 : return 0;
4854 : }
4855 :
4856 : static int
4857 : pthread_join(pthread_t th, void **thread_return)
4858 : {
4859 : if (th == NULL || th->handle == NULL)
4860 : return errno = EINVAL;
4861 :
4862 : if (WaitForSingleObject(th->handle, INFINITE) != WAIT_OBJECT_0)
4863 : {
4864 : _dosmaperr(GetLastError());
4865 : return errno;
4866 : }
4867 :
4868 : if (thread_return)
4869 : *thread_return = th->result;
4870 :
4871 : CloseHandle(th->handle);
4872 : free(th);
4873 : return 0;
4874 : }
4875 :
4876 : #endif /* WIN32 */
|
