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-2018, 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 : #include "fe_utils/conditional.h"
36 :
37 : #include "getopt_long.h"
38 : #include "libpq-fe.h"
39 : #include "portability/instr_time.h"
40 :
41 : #include <ctype.h>
42 : #include <float.h>
43 : #include <limits.h>
44 : #include <math.h>
45 : #include <signal.h>
46 : #include <time.h>
47 : #include <sys/time.h>
48 : #ifdef HAVE_SYS_SELECT_H
49 : #include <sys/select.h>
50 : #endif
51 :
52 : #ifdef HAVE_SYS_RESOURCE_H
53 : #include <sys/resource.h> /* for getrlimit */
54 : #endif
55 :
56 : #ifndef M_PI
57 : #define M_PI 3.14159265358979323846
58 : #endif
59 :
60 : #include "pgbench.h"
61 :
62 : #define ERRCODE_UNDEFINED_TABLE "42P01"
63 :
64 : /*
65 : * Hashing constants
66 : */
67 : #define FNV_PRIME UINT64CONST(0x100000001b3)
68 : #define FNV_OFFSET_BASIS UINT64CONST(0xcbf29ce484222325)
69 : #define MM2_MUL UINT64CONST(0xc6a4a7935bd1e995)
70 : #define MM2_MUL_TIMES_8 UINT64CONST(0x35253c9ade8f4ca8)
71 : #define MM2_ROT 47
72 :
73 : /*
74 : * Multi-platform pthread implementations
75 : */
76 :
77 : #ifdef WIN32
78 : /* Use native win32 threads on Windows */
79 : typedef struct win32_pthread *pthread_t;
80 : typedef int pthread_attr_t;
81 :
82 : static int pthread_create(pthread_t *thread, pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
83 : static int pthread_join(pthread_t th, void **thread_return);
84 : #elif defined(ENABLE_THREAD_SAFETY)
85 : /* Use platform-dependent pthread capability */
86 : #include <pthread.h>
87 : #else
88 : /* No threads implementation, use none (-j 1) */
89 : #define pthread_t void *
90 : #endif
91 :
92 :
93 : /********************************************************************
94 : * some configurable parameters */
95 :
96 : /* max number of clients allowed */
97 : #ifdef FD_SETSIZE
98 : #define MAXCLIENTS (FD_SETSIZE - 10)
99 : #else
100 : #define MAXCLIENTS 1024
101 : #endif
102 :
103 : #define DEFAULT_INIT_STEPS "dtgvp" /* default -I setting */
104 :
105 : #define LOG_STEP_SECONDS 5 /* seconds between log messages */
106 : #define DEFAULT_NXACTS 10 /* default nxacts */
107 :
108 : #define ZIPF_CACHE_SIZE 15 /* cache cells number */
109 :
110 : #define MIN_GAUSSIAN_PARAM 2.0 /* minimum parameter for gauss */
111 : #define MAX_ZIPFIAN_PARAM 1000 /* maximum parameter for zipfian */
112 :
113 : int nxacts = 0; /* number of transactions per client */
114 : int duration = 0; /* duration in seconds */
115 : int64 end_time = 0; /* when to stop in micro seconds, under -T */
116 :
117 : /*
118 : * scaling factor. for example, scale = 10 will make 1000000 tuples in
119 : * pgbench_accounts table.
120 : */
121 : int scale = 1;
122 :
123 : /*
124 : * fillfactor. for example, fillfactor = 90 will use only 90 percent
125 : * space during inserts and leave 10 percent free.
126 : */
127 : int fillfactor = 100;
128 :
129 : /*
130 : * use unlogged tables?
131 : */
132 : bool unlogged_tables = false;
133 :
134 : /*
135 : * log sampling rate (1.0 = log everything, 0.0 = option not given)
136 : */
137 : double sample_rate = 0.0;
138 :
139 : /*
140 : * When threads are throttled to a given rate limit, this is the target delay
141 : * to reach that rate in usec. 0 is the default and means no throttling.
142 : */
143 : int64 throttle_delay = 0;
144 :
145 : /*
146 : * Transactions which take longer than this limit (in usec) are counted as
147 : * late, and reported as such, although they are completed anyway. When
148 : * throttling is enabled, execution time slots that are more than this late
149 : * are skipped altogether, and counted separately.
150 : */
151 : int64 latency_limit = 0;
152 :
153 : /*
154 : * tablespace selection
155 : */
156 : char *tablespace = NULL;
157 : char *index_tablespace = NULL;
158 :
159 : /* random seed used when calling srandom() */
160 : int64 random_seed = -1;
161 :
162 : /*
163 : * end of configurable parameters
164 : *********************************************************************/
165 :
166 : #define nbranches 1 /* Makes little sense to change this. Change
167 : * -s instead */
168 : #define ntellers 10
169 : #define naccounts 100000
170 :
171 : /*
172 : * The scale factor at/beyond which 32bit integers are incapable of storing
173 : * 64bit values.
174 : *
175 : * Although the actual threshold is 21474, we use 20000 because it is easier to
176 : * document and remember, and isn't that far away from the real threshold.
177 : */
178 : #define SCALE_32BIT_THRESHOLD 20000
179 :
180 : bool use_log; /* log transaction latencies to a file */
181 : bool use_quiet; /* quiet logging onto stderr */
182 : int agg_interval; /* log aggregates instead of individual
183 : * transactions */
184 : bool per_script_stats = false; /* whether to collect stats per script */
185 : int progress = 0; /* thread progress report every this seconds */
186 : bool progress_timestamp = false; /* progress report with Unix time */
187 : int nclients = 1; /* number of clients */
188 : int nthreads = 1; /* number of threads */
189 : bool is_connect; /* establish connection for each transaction */
190 : bool is_latencies; /* report per-command latencies */
191 : int main_pid; /* main process id used in log filename */
192 :
193 : char *pghost = "";
194 : char *pgport = "";
195 : char *login = NULL;
196 : char *dbName;
197 : char *logfile_prefix = NULL;
198 : const char *progname;
199 :
200 : #define WSEP '@' /* weight separator */
201 :
202 : volatile bool timer_exceeded = false; /* flag from signal handler */
203 :
204 : /*
205 : * Variable definitions.
206 : *
207 : * If a variable only has a string value, "svalue" is that value, and value is
208 : * "not set". If the value is known, "value" contains the value (in any
209 : * variant).
210 : *
211 : * In this case "svalue" contains the string equivalent of the value, if we've
212 : * had occasion to compute that, or NULL if we haven't.
213 : */
214 : typedef struct
215 : {
216 : char *name; /* variable's name */
217 : char *svalue; /* its value in string form, if known */
218 : PgBenchValue value; /* actual variable's value */
219 : } Variable;
220 :
221 : #define MAX_SCRIPTS 128 /* max number of SQL scripts allowed */
222 : #define SHELL_COMMAND_SIZE 256 /* maximum size allowed for shell command */
223 :
224 : /*
225 : * Simple data structure to keep stats about something.
226 : *
227 : * XXX probably the first value should be kept and used as an offset for
228 : * better numerical stability...
229 : */
230 : typedef struct SimpleStats
231 : {
232 : int64 count; /* how many values were encountered */
233 : double min; /* the minimum seen */
234 : double max; /* the maximum seen */
235 : double sum; /* sum of values */
236 : double sum2; /* sum of squared values */
237 : } SimpleStats;
238 :
239 : /*
240 : * Data structure to hold various statistics: per-thread and per-script stats
241 : * are maintained and merged together.
242 : */
243 : typedef struct StatsData
244 : {
245 : time_t start_time; /* interval start time, for aggregates */
246 : int64 cnt; /* number of transactions, including skipped */
247 : int64 skipped; /* number of transactions skipped under --rate
248 : * and --latency-limit */
249 : SimpleStats latency;
250 : SimpleStats lag;
251 : } StatsData;
252 :
253 : /*
254 : * Connection state machine states.
255 : */
256 : typedef enum
257 : {
258 : /*
259 : * The client must first choose a script to execute. Once chosen, it can
260 : * either be throttled (state CSTATE_START_THROTTLE under --rate) or start
261 : * right away (state CSTATE_START_TX).
262 : */
263 : CSTATE_CHOOSE_SCRIPT,
264 :
265 : /*
266 : * In CSTATE_START_THROTTLE state, we calculate when to begin the next
267 : * transaction, and advance to CSTATE_THROTTLE. CSTATE_THROTTLE state
268 : * sleeps until that moment. (If throttling is not enabled, doCustom()
269 : * falls directly through from CSTATE_START_THROTTLE to CSTATE_START_TX.)
270 : */
271 : CSTATE_START_THROTTLE,
272 : CSTATE_THROTTLE,
273 :
274 : /*
275 : * CSTATE_START_TX performs start-of-transaction processing. Establishes
276 : * a new connection for the transaction, in --connect mode, and records
277 : * the transaction start time.
278 : */
279 : CSTATE_START_TX,
280 :
281 : /*
282 : * We loop through these states, to process each command in the script:
283 : *
284 : * CSTATE_START_COMMAND starts the execution of a command. On a SQL
285 : * command, the command is sent to the server, and we move to
286 : * CSTATE_WAIT_RESULT state. On a \sleep meta-command, the timer is set,
287 : * and we enter the CSTATE_SLEEP state to wait for it to expire. Other
288 : * meta-commands are executed immediately.
289 : *
290 : * CSTATE_SKIP_COMMAND for conditional branches which are not executed,
291 : * quickly skip commands that do not need any evaluation.
292 : *
293 : * CSTATE_WAIT_RESULT waits until we get a result set back from the server
294 : * for the current command.
295 : *
296 : * CSTATE_SLEEP waits until the end of \sleep.
297 : *
298 : * CSTATE_END_COMMAND records the end-of-command timestamp, increments the
299 : * command counter, and loops back to CSTATE_START_COMMAND state.
300 : */
301 : CSTATE_START_COMMAND,
302 : CSTATE_SKIP_COMMAND,
303 : CSTATE_WAIT_RESULT,
304 : CSTATE_SLEEP,
305 : CSTATE_END_COMMAND,
306 :
307 : /*
308 : * CSTATE_END_TX performs end-of-transaction processing. Calculates
309 : * latency, and logs the transaction. In --connect mode, closes the
310 : * current connection. Chooses the next script to execute and starts over
311 : * in CSTATE_START_THROTTLE state, or enters CSTATE_FINISHED if we have no
312 : * more work to do.
313 : */
314 : CSTATE_END_TX,
315 :
316 : /*
317 : * Final states. CSTATE_ABORTED means that the script execution was
318 : * aborted because a command failed, CSTATE_FINISHED means success.
319 : */
320 : CSTATE_ABORTED,
321 : CSTATE_FINISHED
322 : } ConnectionStateEnum;
323 :
324 : /*
325 : * Connection state.
326 : */
327 : typedef struct
328 : {
329 : PGconn *con; /* connection handle to DB */
330 : int id; /* client No. */
331 : ConnectionStateEnum state; /* state machine's current state. */
332 : ConditionalStack cstack; /* enclosing conditionals state */
333 :
334 : int use_file; /* index in sql_script for this client */
335 : int command; /* command number in script */
336 :
337 : /* client variables */
338 : Variable *variables; /* array of variable definitions */
339 : int nvariables; /* number of variables */
340 : bool vars_sorted; /* are variables sorted by name? */
341 :
342 : /* various times about current transaction */
343 : int64 txn_scheduled; /* scheduled start time of transaction (usec) */
344 : int64 sleep_until; /* scheduled start time of next cmd (usec) */
345 : instr_time txn_begin; /* used for measuring schedule lag times */
346 : instr_time stmt_begin; /* used for measuring statement latencies */
347 :
348 : bool prepared[MAX_SCRIPTS]; /* whether client prepared the script */
349 :
350 : /* per client collected stats */
351 : int64 cnt; /* client transaction count, for -t */
352 : int ecnt; /* error count */
353 : } CState;
354 :
355 : /*
356 : * Cache cell for zipfian_random call
357 : */
358 : typedef struct
359 : {
360 : /* cell keys */
361 : double s; /* s - parameter of zipfan_random function */
362 : int64 n; /* number of elements in range (max - min + 1) */
363 :
364 : double harmonicn; /* generalizedHarmonicNumber(n, s) */
365 : double alpha;
366 : double beta;
367 : double eta;
368 :
369 : uint64 last_used; /* last used logical time */
370 : } ZipfCell;
371 :
372 : /*
373 : * Zipf cache for zeta values
374 : */
375 : typedef struct
376 : {
377 : uint64 current; /* counter for LRU cache replacement algorithm */
378 :
379 : int nb_cells; /* number of filled cells */
380 : int overflowCount; /* number of cache overflows */
381 : ZipfCell cells[ZIPF_CACHE_SIZE];
382 : } ZipfCache;
383 :
384 : /*
385 : * Thread state
386 : */
387 : typedef struct
388 : {
389 : int tid; /* thread id */
390 : pthread_t thread; /* thread handle */
391 : CState *state; /* array of CState */
392 : int nstate; /* length of state[] */
393 : unsigned short random_state[3]; /* separate randomness for each thread */
394 : int64 throttle_trigger; /* previous/next throttling (us) */
395 : FILE *logfile; /* where to log, or NULL */
396 : ZipfCache zipf_cache; /* for thread-safe zipfian random number
397 : * generation */
398 :
399 : /* per thread collected stats */
400 : instr_time start_time; /* thread start time */
401 : instr_time conn_time;
402 : StatsData stats;
403 : int64 latency_late; /* executed but late transactions */
404 : } TState;
405 :
406 : #define INVALID_THREAD ((pthread_t) 0)
407 :
408 : /*
409 : * queries read from files
410 : */
411 : #define SQL_COMMAND 1
412 : #define META_COMMAND 2
413 : #define MAX_ARGS 10
414 :
415 : typedef enum MetaCommand
416 : {
417 : META_NONE, /* not a known meta-command */
418 : META_SET, /* \set */
419 : META_SETSHELL, /* \setshell */
420 : META_SHELL, /* \shell */
421 : META_SLEEP, /* \sleep */
422 : META_IF, /* \if */
423 : META_ELIF, /* \elif */
424 : META_ELSE, /* \else */
425 : META_ENDIF /* \endif */
426 : } MetaCommand;
427 :
428 : typedef enum QueryMode
429 : {
430 : QUERY_SIMPLE, /* simple query */
431 : QUERY_EXTENDED, /* extended query */
432 : QUERY_PREPARED, /* extended query with prepared statements */
433 : NUM_QUERYMODE
434 : } QueryMode;
435 :
436 : static QueryMode querymode = QUERY_SIMPLE;
437 : static const char *QUERYMODE[] = {"simple", "extended", "prepared"};
438 :
439 : typedef struct
440 : {
441 : char *line; /* text of command line */
442 : int command_num; /* unique index of this Command struct */
443 : int type; /* command type (SQL_COMMAND or META_COMMAND) */
444 : MetaCommand meta; /* meta command identifier, or META_NONE */
445 : int argc; /* number of command words */
446 : char *argv[MAX_ARGS]; /* command word list */
447 : PgBenchExpr *expr; /* parsed expression, if needed */
448 : SimpleStats stats; /* time spent in this command */
449 : } Command;
450 :
451 : typedef struct ParsedScript
452 : {
453 : const char *desc; /* script descriptor (eg, file name) */
454 : int weight; /* selection weight */
455 : Command **commands; /* NULL-terminated array of Commands */
456 : StatsData stats; /* total time spent in script */
457 : } ParsedScript;
458 :
459 : static ParsedScript sql_script[MAX_SCRIPTS]; /* SQL script files */
460 : static int num_scripts; /* number of scripts in sql_script[] */
461 : static int num_commands = 0; /* total number of Command structs */
462 : static int64 total_weight = 0;
463 :
464 : static int debug = 0; /* debug flag */
465 :
466 : /* Builtin test scripts */
467 : typedef struct BuiltinScript
468 : {
469 : const char *name; /* very short name for -b ... */
470 : const char *desc; /* short description */
471 : const char *script; /* actual pgbench script */
472 : } BuiltinScript;
473 :
474 : static const BuiltinScript builtin_script[] =
475 : {
476 : {
477 : "tpcb-like",
478 : "<builtin: TPC-B (sort of)>",
479 : "\\set aid random(1, " CppAsString2(naccounts) " * :scale)\n"
480 : "\\set bid random(1, " CppAsString2(nbranches) " * :scale)\n"
481 : "\\set tid random(1, " CppAsString2(ntellers) " * :scale)\n"
482 : "\\set delta random(-5000, 5000)\n"
483 : "BEGIN;\n"
484 : "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
485 : "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
486 : "UPDATE pgbench_tellers SET tbalance = tbalance + :delta WHERE tid = :tid;\n"
487 : "UPDATE pgbench_branches SET bbalance = bbalance + :delta WHERE bid = :bid;\n"
488 : "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
489 : "END;\n"
490 : },
491 : {
492 : "simple-update",
493 : "<builtin: simple update>",
494 : "\\set aid random(1, " CppAsString2(naccounts) " * :scale)\n"
495 : "\\set bid random(1, " CppAsString2(nbranches) " * :scale)\n"
496 : "\\set tid random(1, " CppAsString2(ntellers) " * :scale)\n"
497 : "\\set delta random(-5000, 5000)\n"
498 : "BEGIN;\n"
499 : "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
500 : "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
501 : "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
502 : "END;\n"
503 : },
504 : {
505 : "select-only",
506 : "<builtin: select only>",
507 : "\\set aid random(1, " CppAsString2(naccounts) " * :scale)\n"
508 : "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
509 : }
510 : };
511 :
512 :
513 : /* Function prototypes */
514 : static void setNullValue(PgBenchValue *pv);
515 : static void setBoolValue(PgBenchValue *pv, bool bval);
516 : static void setIntValue(PgBenchValue *pv, int64 ival);
517 : static void setDoubleValue(PgBenchValue *pv, double dval);
518 : static bool evaluateExpr(TState *, CState *, PgBenchExpr *, PgBenchValue *);
519 : static void doLog(TState *thread, CState *st,
520 : StatsData *agg, bool skipped, double latency, double lag);
521 : static void processXactStats(TState *thread, CState *st, instr_time *now,
522 : bool skipped, StatsData *agg);
523 : static void pgbench_error(const char *fmt,...) pg_attribute_printf(1, 2);
524 : static void addScript(ParsedScript script);
525 : static void *threadRun(void *arg);
526 : static void setalarm(int seconds);
527 : static void finishCon(CState *st);
528 :
529 :
530 : /* callback functions for our flex lexer */
531 : static const PsqlScanCallbacks pgbench_callbacks = {
532 : NULL, /* don't need get_variable functionality */
533 : pgbench_error
534 : };
535 :
536 :
537 : static void
538 2 : usage(void)
539 : {
540 2 : printf("%s is a benchmarking tool for PostgreSQL.\n\n"
541 : "Usage:\n"
542 : " %s [OPTION]... [DBNAME]\n"
543 : "\nInitialization options:\n"
544 : " -i, --initialize invokes initialization mode\n"
545 : " -I, --init-steps=[dtgvpf]+ (default \"dtgvp\")\n"
546 : " run selected initialization steps\n"
547 : " -F, --fillfactor=NUM set fill factor\n"
548 : " -n, --no-vacuum do not run VACUUM during initialization\n"
549 : " -q, --quiet quiet logging (one message each 5 seconds)\n"
550 : " -s, --scale=NUM scaling factor\n"
551 : " --foreign-keys create foreign key constraints between tables\n"
552 : " --index-tablespace=TABLESPACE\n"
553 : " create indexes in the specified tablespace\n"
554 : " --tablespace=TABLESPACE create tables in the specified tablespace\n"
555 : " --unlogged-tables create tables as unlogged tables\n"
556 : "\nOptions to select what to run:\n"
557 : " -b, --builtin=NAME[@W] add builtin script NAME weighted at W (default: 1)\n"
558 : " (use \"-b list\" to list available scripts)\n"
559 : " -f, --file=FILENAME[@W] add script FILENAME weighted at W (default: 1)\n"
560 : " -N, --skip-some-updates skip updates of pgbench_tellers and pgbench_branches\n"
561 : " (same as \"-b simple-update\")\n"
562 : " -S, --select-only perform SELECT-only transactions\n"
563 : " (same as \"-b select-only\")\n"
564 : "\nBenchmarking options:\n"
565 : " -c, --client=NUM number of concurrent database clients (default: 1)\n"
566 : " -C, --connect establish new connection for each transaction\n"
567 : " -D, --define=VARNAME=VALUE\n"
568 : " define variable for use by custom script\n"
569 : " -j, --jobs=NUM number of threads (default: 1)\n"
570 : " -l, --log write transaction times to log file\n"
571 : " -L, --latency-limit=NUM count transactions lasting more than NUM ms as late\n"
572 : " -M, --protocol=simple|extended|prepared\n"
573 : " protocol for submitting queries (default: simple)\n"
574 : " -n, --no-vacuum do not run VACUUM before tests\n"
575 : " -P, --progress=NUM show thread progress report every NUM seconds\n"
576 : " -r, --report-latencies report average latency per command\n"
577 : " -R, --rate=NUM target rate in transactions per second\n"
578 : " -s, --scale=NUM report this scale factor in output\n"
579 : " -t, --transactions=NUM number of transactions each client runs (default: 10)\n"
580 : " -T, --time=NUM duration of benchmark test in seconds\n"
581 : " -v, --vacuum-all vacuum all four standard tables before tests\n"
582 : " --aggregate-interval=NUM aggregate data over NUM seconds\n"
583 : " --log-prefix=PREFIX prefix for transaction time log file\n"
584 : " (default: \"pgbench_log\")\n"
585 : " --progress-timestamp use Unix epoch timestamps for progress\n"
586 : " --random-seed=SEED set random seed (\"time\", \"rand\", integer)\n"
587 : " --sampling-rate=NUM fraction of transactions to log (e.g., 0.01 for 1%%)\n"
588 : "\nCommon options:\n"
589 : " -d, --debug print debugging output\n"
590 : " -h, --host=HOSTNAME database server host or socket directory\n"
591 : " -p, --port=PORT database server port number\n"
592 : " -U, --username=USERNAME connect as specified database user\n"
593 : " -V, --version output version information, then exit\n"
594 : " -?, --help show this help, then exit\n"
595 : "\n"
596 : "Report bugs to <pgsql-bugs@postgresql.org>.\n",
597 : progname, progname);
598 2 : }
599 :
600 : /* return whether str matches "^\s*[-+]?[0-9]+$" */
601 : static bool
602 8 : is_an_int(const char *str)
603 : {
604 8 : const char *ptr = str;
605 :
606 : /* skip leading spaces; cast is consistent with strtoint64 */
607 16 : while (*ptr && isspace((unsigned char) *ptr))
608 0 : ptr++;
609 :
610 : /* skip sign */
611 8 : if (*ptr == '+' || *ptr == '-')
612 4 : ptr++;
613 :
614 : /* at least one digit */
615 8 : if (*ptr && !isdigit((unsigned char) *ptr))
616 4 : return false;
617 :
618 : /* eat all digits */
619 50 : while (*ptr && isdigit((unsigned char) *ptr))
620 42 : ptr++;
621 :
622 : /* must have reached end of string */
623 4 : return *ptr == '\0';
624 : }
625 :
626 :
627 : /*
628 : * strtoint64 -- convert a string to 64-bit integer
629 : *
630 : * This function is a modified version of scanint8() from
631 : * src/backend/utils/adt/int8.c.
632 : */
633 : int64
634 1364 : strtoint64(const char *str)
635 : {
636 1364 : const char *ptr = str;
637 1364 : int64 result = 0;
638 1364 : int sign = 1;
639 :
640 : /*
641 : * Do our own scan, rather than relying on sscanf which might be broken
642 : * for long long.
643 : */
644 :
645 : /* skip leading spaces */
646 2728 : while (*ptr && isspace((unsigned char) *ptr))
647 0 : ptr++;
648 :
649 : /* handle sign */
650 1364 : if (*ptr == '-')
651 : {
652 2 : ptr++;
653 :
654 : /*
655 : * Do an explicit check for INT64_MIN. Ugly though this is, it's
656 : * cleaner than trying to get the loop below to handle it portably.
657 : */
658 2 : if (strncmp(ptr, "9223372036854775808", 19) == 0)
659 : {
660 2 : result = PG_INT64_MIN;
661 2 : ptr += 19;
662 2 : goto gotdigits;
663 : }
664 0 : sign = -1;
665 : }
666 1362 : else if (*ptr == '+')
667 0 : ptr++;
668 :
669 : /* require at least one digit */
670 1362 : if (!isdigit((unsigned char) *ptr))
671 0 : fprintf(stderr, "invalid input syntax for integer: \"%s\"\n", str);
672 :
673 : /* process digits */
674 6316 : while (*ptr && isdigit((unsigned char) *ptr))
675 : {
676 3592 : int64 tmp = result * 10 + (*ptr++ - '0');
677 :
678 3592 : if ((tmp / 10) != result) /* overflow? */
679 4 : fprintf(stderr, "value \"%s\" is out of range for type bigint\n", str);
680 3592 : result = tmp;
681 : }
682 :
683 : gotdigits:
684 :
685 : /* allow trailing whitespace, but not other trailing chars */
686 2728 : while (*ptr != '\0' && isspace((unsigned char) *ptr))
687 0 : ptr++;
688 :
689 1364 : if (*ptr != '\0')
690 0 : fprintf(stderr, "invalid input syntax for integer: \"%s\"\n", str);
691 :
692 1364 : return ((sign < 0) ? -result : result);
693 : }
694 :
695 : /* random number generator: uniform distribution from min to max inclusive */
696 : static int64
697 3226 : getrand(TState *thread, int64 min, int64 max)
698 : {
699 : /*
700 : * Odd coding is so that min and max have approximately the same chance of
701 : * being selected as do numbers between them.
702 : *
703 : * pg_erand48() is thread-safe and concurrent, which is why we use it
704 : * rather than random(), which in glibc is non-reentrant, and therefore
705 : * protected by a mutex, and therefore a bottleneck on machines with many
706 : * CPUs.
707 : */
708 3226 : return min + (int64) ((max - min + 1) * pg_erand48(thread->random_state));
709 : }
710 :
711 : /*
712 : * random number generator: exponential distribution from min to max inclusive.
713 : * the parameter is so that the density of probability for the last cut-off max
714 : * value is exp(-parameter).
715 : */
716 : static int64
717 6 : getExponentialRand(TState *thread, int64 min, int64 max, double parameter)
718 : {
719 : double cut,
720 : uniform,
721 : rand;
722 :
723 : /* abort if wrong parameter, but must really be checked beforehand */
724 : Assert(parameter > 0.0);
725 6 : cut = exp(-parameter);
726 : /* erand in [0, 1), uniform in (0, 1] */
727 6 : uniform = 1.0 - pg_erand48(thread->random_state);
728 :
729 : /*
730 : * inner expression in (cut, 1] (if parameter > 0), rand in [0, 1)
731 : */
732 : Assert((1.0 - cut) != 0.0);
733 6 : rand = -log(cut + (1.0 - cut) * uniform) / parameter;
734 : /* return int64 random number within between min and max */
735 6 : return min + (int64) ((max - min + 1) * rand);
736 : }
737 :
738 : /* random number generator: gaussian distribution from min to max inclusive */
739 : static int64
740 6 : getGaussianRand(TState *thread, int64 min, int64 max, double parameter)
741 : {
742 : double stdev;
743 : double rand;
744 :
745 : /* abort if parameter is too low, but must really be checked beforehand */
746 : Assert(parameter >= MIN_GAUSSIAN_PARAM);
747 :
748 : /*
749 : * Get user specified random number from this loop, with -parameter <
750 : * stdev <= parameter
751 : *
752 : * This loop is executed until the number is in the expected range.
753 : *
754 : * As the minimum parameter is 2.0, the probability of looping is low:
755 : * sqrt(-2 ln(r)) <= 2 => r >= e^{-2} ~ 0.135, then when taking the
756 : * average sinus multiplier as 2/pi, we have a 8.6% looping probability in
757 : * the worst case. For a parameter value of 5.0, the looping probability
758 : * is about e^{-5} * 2 / pi ~ 0.43%.
759 : */
760 : do
761 : {
762 : /*
763 : * pg_erand48 generates [0,1), but for the basic version of the
764 : * Box-Muller transform the two uniformly distributed random numbers
765 : * are expected in (0, 1] (see
766 : * http://en.wikipedia.org/wiki/Box_muller)
767 : */
768 6 : double rand1 = 1.0 - pg_erand48(thread->random_state);
769 6 : double rand2 = 1.0 - pg_erand48(thread->random_state);
770 :
771 : /* Box-Muller basic form transform */
772 6 : double var_sqrt = sqrt(-2.0 * log(rand1));
773 :
774 6 : stdev = var_sqrt * sin(2.0 * M_PI * rand2);
775 :
776 : /*
777 : * we may try with cos, but there may be a bias induced if the
778 : * previous value fails the test. To be on the safe side, let us try
779 : * over.
780 : */
781 : }
782 6 : while (stdev < -parameter || stdev >= parameter);
783 :
784 : /* stdev is in [-parameter, parameter), normalization to [0,1) */
785 6 : rand = (stdev + parameter) / (parameter * 2.0);
786 :
787 : /* return int64 random number within between min and max */
788 6 : return min + (int64) ((max - min + 1) * rand);
789 : }
790 :
791 : /*
792 : * random number generator: generate a value, such that the series of values
793 : * will approximate a Poisson distribution centered on the given value.
794 : */
795 : static int64
796 422 : getPoissonRand(TState *thread, int64 center)
797 : {
798 : /*
799 : * Use inverse transform sampling to generate a value > 0, such that the
800 : * expected (i.e. average) value is the given argument.
801 : */
802 : double uniform;
803 :
804 : /* erand in [0, 1), uniform in (0, 1] */
805 422 : uniform = 1.0 - pg_erand48(thread->random_state);
806 :
807 422 : return (int64) (-log(uniform) * ((double) center) + 0.5);
808 : }
809 :
810 : /* helper function for getZipfianRand */
811 : static double
812 64 : generalizedHarmonicNumber(int64 n, double s)
813 : {
814 : int i;
815 64 : double ans = 0.0;
816 :
817 3024 : for (i = n; i > 1; i--)
818 2960 : ans += pow(i, -s);
819 64 : return ans + 1.0;
820 : }
821 :
822 : /* set harmonicn and other parameters to cache cell */
823 : static void
824 32 : zipfSetCacheCell(ZipfCell *cell, int64 n, double s)
825 : {
826 : double harmonic2;
827 :
828 32 : cell->n = n;
829 32 : cell->s = s;
830 :
831 32 : harmonic2 = generalizedHarmonicNumber(2, s);
832 32 : cell->harmonicn = generalizedHarmonicNumber(n, s);
833 :
834 32 : cell->alpha = 1.0 / (1.0 - s);
835 32 : cell->beta = pow(0.5, s);
836 32 : cell->eta = (1.0 - pow(2.0 / n, 1.0 - s)) / (1.0 - harmonic2 / cell->harmonicn);
837 32 : }
838 :
839 : /*
840 : * search for cache cell with keys (n, s)
841 : * and create new cell if it does not exist
842 : */
843 : static ZipfCell *
844 32 : zipfFindOrCreateCacheCell(ZipfCache *cache, int64 n, double s)
845 : {
846 : int i,
847 32 : least_recently_used = 0;
848 : ZipfCell *cell;
849 :
850 : /* search cached cell for given parameters */
851 272 : for (i = 0; i < cache->nb_cells; i++)
852 : {
853 240 : cell = &cache->cells[i];
854 240 : if (cell->n == n && cell->s == s)
855 0 : return &cache->cells[i];
856 :
857 240 : if (cell->last_used < cache->cells[least_recently_used].last_used)
858 0 : least_recently_used = i;
859 : }
860 :
861 : /* create new one if it does not exist */
862 32 : if (cache->nb_cells < ZIPF_CACHE_SIZE)
863 30 : i = cache->nb_cells++;
864 : else
865 : {
866 : /* replace LRU cell if cache is full */
867 2 : i = least_recently_used;
868 2 : cache->overflowCount++;
869 : }
870 :
871 32 : zipfSetCacheCell(&cache->cells[i], n, s);
872 :
873 32 : cache->cells[i].last_used = cache->current++;
874 32 : return &cache->cells[i];
875 : }
876 :
877 : /*
878 : * Computing zipfian using rejection method, based on
879 : * "Non-Uniform Random Variate Generation",
880 : * Luc Devroye, p. 550-551, Springer 1986.
881 : */
882 : static int64
883 6 : computeIterativeZipfian(TState *thread, int64 n, double s)
884 : {
885 6 : double b = pow(2.0, s - 1.0);
886 : double x,
887 : t,
888 : u,
889 : v;
890 :
891 : while (true)
892 : {
893 : /* random variates */
894 14 : u = pg_erand48(thread->random_state);
895 10 : v = pg_erand48(thread->random_state);
896 :
897 10 : x = floor(pow(u, -1.0 / (s - 1.0)));
898 :
899 10 : t = pow(1.0 + 1.0 / x, s - 1.0);
900 : /* reject if too large or out of bound */
901 10 : if (v * x * (t - 1.0) / (b - 1.0) <= t / b && x <= n)
902 6 : break;
903 : }
904 6 : return (int64) x;
905 : }
906 :
907 : /*
908 : * Computing zipfian using harmonic numbers, based on algorithm described in
909 : * "Quickly Generating Billion-Record Synthetic Databases",
910 : * Jim Gray et al, SIGMOD 1994
911 : */
912 : static int64
913 32 : computeHarmonicZipfian(TState *thread, int64 n, double s)
914 : {
915 32 : ZipfCell *cell = zipfFindOrCreateCacheCell(&thread->zipf_cache, n, s);
916 32 : double uniform = pg_erand48(thread->random_state);
917 32 : double uz = uniform * cell->harmonicn;
918 :
919 32 : if (uz < 1.0)
920 0 : return 1;
921 32 : if (uz < 1.0 + cell->beta)
922 0 : return 2;
923 32 : return 1 + (int64) (cell->n * pow(cell->eta * uniform - cell->eta + 1.0, cell->alpha));
924 : }
925 :
926 : /* random number generator: zipfian distribution from min to max inclusive */
927 : static int64
928 38 : getZipfianRand(TState *thread, int64 min, int64 max, double s)
929 : {
930 38 : int64 n = max - min + 1;
931 :
932 : /* abort if parameter is invalid */
933 : Assert(s > 0.0 && s != 1.0 && s <= MAX_ZIPFIAN_PARAM);
934 :
935 :
936 38 : return min - 1 + ((s > 1)
937 : ? computeIterativeZipfian(thread, n, s)
938 38 : : computeHarmonicZipfian(thread, n, s));
939 : }
940 :
941 : /*
942 : * FNV-1a hash function
943 : */
944 : static int64
945 2 : getHashFnv1a(int64 val, uint64 seed)
946 : {
947 : int64 result;
948 : int i;
949 :
950 2 : result = FNV_OFFSET_BASIS ^ seed;
951 18 : for (i = 0; i < 8; ++i)
952 : {
953 16 : int32 octet = val & 0xff;
954 :
955 16 : val = val >> 8;
956 16 : result = result ^ octet;
957 16 : result = result * FNV_PRIME;
958 : }
959 :
960 2 : return result;
961 : }
962 :
963 : /*
964 : * Murmur2 hash function
965 : *
966 : * Based on original work of Austin Appleby
967 : * https://github.com/aappleby/smhasher/blob/master/src/MurmurHash2.cpp
968 : */
969 : static int64
970 10 : getHashMurmur2(int64 val, uint64 seed)
971 : {
972 10 : uint64 result = seed ^ MM2_MUL_TIMES_8; /* sizeof(int64) */
973 10 : uint64 k = (uint64) val;
974 :
975 10 : k *= MM2_MUL;
976 10 : k ^= k >> MM2_ROT;
977 10 : k *= MM2_MUL;
978 :
979 10 : result ^= k;
980 10 : result *= MM2_MUL;
981 :
982 10 : result ^= result >> MM2_ROT;
983 10 : result *= MM2_MUL;
984 10 : result ^= result >> MM2_ROT;
985 :
986 10 : return (int64) result;
987 : }
988 :
989 : /*
990 : * Initialize the given SimpleStats struct to all zeroes
991 : */
992 : static void
993 2620 : initSimpleStats(SimpleStats *ss)
994 : {
995 2620 : memset(ss, 0, sizeof(SimpleStats));
996 2620 : }
997 :
998 : /*
999 : * Accumulate one value into a SimpleStats struct.
1000 : */
1001 : static void
1002 1800 : addToSimpleStats(SimpleStats *ss, double val)
1003 : {
1004 1800 : if (ss->count == 0 || val < ss->min)
1005 60 : ss->min = val;
1006 1800 : if (ss->count == 0 || val > ss->max)
1007 802 : ss->max = val;
1008 1800 : ss->count++;
1009 1800 : ss->sum += val;
1010 1800 : ss->sum2 += val * val;
1011 1800 : }
1012 :
1013 : /*
1014 : * Merge two SimpleStats objects
1015 : */
1016 : static void
1017 160 : mergeSimpleStats(SimpleStats *acc, SimpleStats *ss)
1018 : {
1019 160 : if (acc->count == 0 || ss->min < acc->min)
1020 160 : acc->min = ss->min;
1021 160 : if (acc->count == 0 || ss->max > acc->max)
1022 160 : acc->max = ss->max;
1023 160 : acc->count += ss->count;
1024 160 : acc->sum += ss->sum;
1025 160 : acc->sum2 += ss->sum2;
1026 160 : }
1027 :
1028 : /*
1029 : * Initialize a StatsData struct to mostly zeroes, with its start time set to
1030 : * the given value.
1031 : */
1032 : static void
1033 668 : initStats(StatsData *sd, time_t start_time)
1034 : {
1035 668 : sd->start_time = start_time;
1036 668 : sd->cnt = 0;
1037 668 : sd->skipped = 0;
1038 668 : initSimpleStats(&sd->latency);
1039 668 : initSimpleStats(&sd->lag);
1040 668 : }
1041 :
1042 : /*
1043 : * Accumulate one additional item into the given stats object.
1044 : */
1045 : static void
1046 620 : accumStats(StatsData *stats, bool skipped, double lat, double lag)
1047 : {
1048 620 : stats->cnt++;
1049 :
1050 620 : if (skipped)
1051 : {
1052 : /* no latency to record on skipped transactions */
1053 20 : stats->skipped++;
1054 : }
1055 : else
1056 : {
1057 600 : addToSimpleStats(&stats->latency, lat);
1058 :
1059 : /* and possibly the same for schedule lag */
1060 600 : if (throttle_delay)
1061 400 : addToSimpleStats(&stats->lag, lag);
1062 : }
1063 620 : }
1064 :
1065 : /* call PQexec() and exit() on failure */
1066 : static void
1067 168 : executeStatement(PGconn *con, const char *sql)
1068 : {
1069 : PGresult *res;
1070 :
1071 168 : res = PQexec(con, sql);
1072 168 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1073 : {
1074 0 : fprintf(stderr, "%s", PQerrorMessage(con));
1075 0 : exit(1);
1076 : }
1077 168 : PQclear(res);
1078 168 : }
1079 :
1080 : /* call PQexec() and complain, but without exiting, on failure */
1081 : static void
1082 36 : tryExecuteStatement(PGconn *con, const char *sql)
1083 : {
1084 : PGresult *res;
1085 :
1086 36 : res = PQexec(con, sql);
1087 36 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1088 : {
1089 0 : fprintf(stderr, "%s", PQerrorMessage(con));
1090 0 : fprintf(stderr, "(ignoring this error and continuing anyway)\n");
1091 : }
1092 36 : PQclear(res);
1093 36 : }
1094 :
1095 : /* set up a connection to the backend */
1096 : static PGconn *
1097 412 : doConnect(void)
1098 : {
1099 : PGconn *conn;
1100 : bool new_pass;
1101 : static bool have_password = false;
1102 : static char password[100];
1103 :
1104 : /*
1105 : * Start the connection. Loop until we have a password if requested by
1106 : * backend.
1107 : */
1108 : do
1109 : {
1110 : #define PARAMS_ARRAY_SIZE 7
1111 :
1112 : const char *keywords[PARAMS_ARRAY_SIZE];
1113 : const char *values[PARAMS_ARRAY_SIZE];
1114 :
1115 412 : keywords[0] = "host";
1116 412 : values[0] = pghost;
1117 412 : keywords[1] = "port";
1118 412 : values[1] = pgport;
1119 412 : keywords[2] = "user";
1120 412 : values[2] = login;
1121 412 : keywords[3] = "password";
1122 412 : values[3] = have_password ? password : NULL;
1123 412 : keywords[4] = "dbname";
1124 412 : values[4] = dbName;
1125 412 : keywords[5] = "fallback_application_name";
1126 412 : values[5] = progname;
1127 412 : keywords[6] = NULL;
1128 412 : values[6] = NULL;
1129 :
1130 412 : new_pass = false;
1131 :
1132 412 : conn = PQconnectdbParams(keywords, values, true);
1133 :
1134 412 : if (!conn)
1135 : {
1136 0 : fprintf(stderr, "connection to database \"%s\" failed\n",
1137 : dbName);
1138 0 : return NULL;
1139 : }
1140 :
1141 414 : if (PQstatus(conn) == CONNECTION_BAD &&
1142 2 : PQconnectionNeedsPassword(conn) &&
1143 0 : !have_password)
1144 : {
1145 0 : PQfinish(conn);
1146 0 : simple_prompt("Password: ", password, sizeof(password), false);
1147 0 : have_password = true;
1148 0 : new_pass = true;
1149 : }
1150 412 : } while (new_pass);
1151 :
1152 : /* check to see that the backend connection was successfully made */
1153 412 : if (PQstatus(conn) == CONNECTION_BAD)
1154 : {
1155 2 : fprintf(stderr, "connection to database \"%s\" failed:\n%s",
1156 : dbName, PQerrorMessage(conn));
1157 2 : PQfinish(conn);
1158 2 : return NULL;
1159 : }
1160 :
1161 410 : return conn;
1162 : }
1163 :
1164 : /* throw away response from backend */
1165 : static void
1166 4076 : discard_response(CState *state)
1167 : {
1168 : PGresult *res;
1169 :
1170 : do
1171 : {
1172 4076 : res = PQgetResult(state->con);
1173 4076 : if (res)
1174 0 : PQclear(res);
1175 4076 : } while (res);
1176 4076 : }
1177 :
1178 : /* qsort comparator for Variable array */
1179 : static int
1180 83568 : compareVariableNames(const void *v1, const void *v2)
1181 : {
1182 83568 : return strcmp(((const Variable *) v1)->name,
1183 83568 : ((const Variable *) v2)->name);
1184 : }
1185 :
1186 : /* Locate a variable by name; returns NULL if unknown */
1187 : static Variable *
1188 11754 : lookupVariable(CState *st, char *name)
1189 : {
1190 : Variable key;
1191 :
1192 : /* On some versions of Solaris, bsearch of zero items dumps core */
1193 11754 : if (st->nvariables <= 0)
1194 178 : return NULL;
1195 :
1196 : /* Sort if we have to */
1197 11576 : if (!st->vars_sorted)
1198 : {
1199 1138 : qsort((void *) st->variables, st->nvariables, sizeof(Variable),
1200 : compareVariableNames);
1201 1138 : st->vars_sorted = true;
1202 : }
1203 :
1204 : /* Now we can search */
1205 11576 : key.name = name;
1206 23152 : return (Variable *) bsearch((void *) &key,
1207 11576 : (void *) st->variables,
1208 11576 : st->nvariables,
1209 : sizeof(Variable),
1210 : compareVariableNames);
1211 : }
1212 :
1213 : /* Get the value of a variable, in string form; returns NULL if unknown */
1214 : static char *
1215 4230 : getVariable(CState *st, char *name)
1216 : {
1217 : Variable *var;
1218 : char stringform[64];
1219 :
1220 4230 : var = lookupVariable(st, name);
1221 4230 : if (var == NULL)
1222 4 : return NULL; /* not found */
1223 :
1224 4226 : if (var->svalue)
1225 1030 : return var->svalue; /* we have it in string form */
1226 :
1227 : /* We need to produce a string equivalent of the value */
1228 : Assert(var->value.type != PGBT_NO_VALUE);
1229 3196 : if (var->value.type == PGBT_NULL)
1230 2 : snprintf(stringform, sizeof(stringform), "NULL");
1231 3194 : else if (var->value.type == PGBT_BOOLEAN)
1232 2 : snprintf(stringform, sizeof(stringform),
1233 2 : "%s", var->value.u.bval ? "true" : "false");
1234 3192 : else if (var->value.type == PGBT_INT)
1235 3188 : snprintf(stringform, sizeof(stringform),
1236 : INT64_FORMAT, var->value.u.ival);
1237 4 : else if (var->value.type == PGBT_DOUBLE)
1238 4 : snprintf(stringform, sizeof(stringform),
1239 : "%.*g", DBL_DIG, var->value.u.dval);
1240 : else /* internal error, unexpected type */
1241 : Assert(0);
1242 3196 : var->svalue = pg_strdup(stringform);
1243 3196 : return var->svalue;
1244 : }
1245 :
1246 : /* Try to convert variable to a value; return false on failure */
1247 : static bool
1248 2866 : makeVariableValue(Variable *var)
1249 : {
1250 : size_t slen;
1251 :
1252 2866 : if (var->value.type != PGBT_NO_VALUE)
1253 2852 : return true; /* no work */
1254 :
1255 14 : slen = strlen(var->svalue);
1256 :
1257 14 : if (slen == 0)
1258 : /* what should it do on ""? */
1259 0 : return false;
1260 :
1261 14 : if (pg_strcasecmp(var->svalue, "null") == 0)
1262 : {
1263 2 : setNullValue(&var->value);
1264 : }
1265 :
1266 : /*
1267 : * accept prefixes such as y, ye, n, no... but not for "o". 0/1 are
1268 : * recognized later as an int, which is converted to bool if needed.
1269 : */
1270 22 : else if (pg_strncasecmp(var->svalue, "true", slen) == 0 ||
1271 20 : pg_strncasecmp(var->svalue, "yes", slen) == 0 ||
1272 10 : pg_strcasecmp(var->svalue, "on") == 0)
1273 : {
1274 2 : setBoolValue(&var->value, true);
1275 : }
1276 20 : else if (pg_strncasecmp(var->svalue, "false", slen) == 0 ||
1277 20 : pg_strncasecmp(var->svalue, "no", slen) == 0 ||
1278 20 : pg_strcasecmp(var->svalue, "off") == 0 ||
1279 10 : pg_strcasecmp(var->svalue, "of") == 0)
1280 : {
1281 2 : setBoolValue(&var->value, false);
1282 : }
1283 8 : else if (is_an_int(var->svalue))
1284 : {
1285 2 : setIntValue(&var->value, strtoint64(var->svalue));
1286 : }
1287 : else /* type should be double */
1288 : {
1289 : double dv;
1290 : char xs;
1291 :
1292 6 : if (sscanf(var->svalue, "%lf%c", &dv, &xs) != 1)
1293 : {
1294 4 : fprintf(stderr,
1295 : "malformed variable \"%s\" value: \"%s\"\n",
1296 : var->name, var->svalue);
1297 4 : return false;
1298 : }
1299 2 : setDoubleValue(&var->value, dv);
1300 : }
1301 10 : return true;
1302 : }
1303 :
1304 : /*
1305 : * Check whether a variable's name is allowed.
1306 : *
1307 : * We allow any non-ASCII character, as well as ASCII letters, digits, and
1308 : * underscore.
1309 : *
1310 : * Keep this in sync with the definitions of variable name characters in
1311 : * "src/fe_utils/psqlscan.l", "src/bin/psql/psqlscanslash.l" and
1312 : * "src/bin/pgbench/exprscan.l". Also see parseVariable(), below.
1313 : *
1314 : * Note: this static function is copied from "src/bin/psql/variables.c"
1315 : */
1316 : static bool
1317 1226 : valid_variable_name(const char *name)
1318 : {
1319 1226 : const unsigned char *ptr = (const unsigned char *) name;
1320 :
1321 : /* Mustn't be zero-length */
1322 1226 : if (*ptr == '\0')
1323 0 : return false;
1324 :
1325 9572 : while (*ptr)
1326 : {
1327 14244 : if (IS_HIGHBIT_SET(*ptr) ||
1328 7122 : strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
1329 7122 : "_0123456789", *ptr) != NULL)
1330 7120 : ptr++;
1331 : else
1332 2 : return false;
1333 : }
1334 :
1335 1224 : return true;
1336 : }
1337 :
1338 : /*
1339 : * Lookup a variable by name, creating it if need be.
1340 : * Caller is expected to assign a value to the variable.
1341 : * Returns NULL on failure (bad name).
1342 : */
1343 : static Variable *
1344 4336 : lookupCreateVariable(CState *st, const char *context, char *name)
1345 : {
1346 : Variable *var;
1347 :
1348 4336 : var = lookupVariable(st, name);
1349 4336 : if (var == NULL)
1350 : {
1351 : Variable *newvars;
1352 :
1353 : /*
1354 : * Check for the name only when declaring a new variable to avoid
1355 : * overhead.
1356 : */
1357 1226 : if (!valid_variable_name(name))
1358 : {
1359 2 : fprintf(stderr, "%s: invalid variable name: \"%s\"\n",
1360 : context, name);
1361 2 : return NULL;
1362 : }
1363 :
1364 : /* Create variable at the end of the array */
1365 1224 : if (st->variables)
1366 1074 : newvars = (Variable *) pg_realloc(st->variables,
1367 1074 : (st->nvariables + 1) * sizeof(Variable));
1368 : else
1369 150 : newvars = (Variable *) pg_malloc(sizeof(Variable));
1370 :
1371 1224 : st->variables = newvars;
1372 :
1373 1224 : var = &newvars[st->nvariables];
1374 :
1375 1224 : var->name = pg_strdup(name);
1376 1224 : var->svalue = NULL;
1377 : /* caller is expected to initialize remaining fields */
1378 :
1379 1224 : st->nvariables++;
1380 : /* we don't re-sort the array till we have to */
1381 1224 : st->vars_sorted = false;
1382 : }
1383 :
1384 4334 : return var;
1385 : }
1386 :
1387 : /* Assign a string value to a variable, creating it if need be */
1388 : /* Returns false on failure (bad name) */
1389 : static bool
1390 512 : putVariable(CState *st, const char *context, char *name, const char *value)
1391 : {
1392 : Variable *var;
1393 : char *val;
1394 :
1395 512 : var = lookupCreateVariable(st, context, name);
1396 512 : if (!var)
1397 0 : return false;
1398 :
1399 : /* dup then free, in case value is pointing at this variable */
1400 512 : val = pg_strdup(value);
1401 :
1402 512 : if (var->svalue)
1403 0 : free(var->svalue);
1404 512 : var->svalue = val;
1405 512 : var->value.type = PGBT_NO_VALUE;
1406 :
1407 512 : return true;
1408 : }
1409 :
1410 : /* Assign a value to a variable, creating it if need be */
1411 : /* Returns false on failure (bad name) */
1412 : static bool
1413 3824 : putVariableValue(CState *st, const char *context, char *name,
1414 : const PgBenchValue *value)
1415 : {
1416 : Variable *var;
1417 :
1418 3824 : var = lookupCreateVariable(st, context, name);
1419 3824 : if (!var)
1420 2 : return false;
1421 :
1422 3822 : if (var->svalue)
1423 3070 : free(var->svalue);
1424 3822 : var->svalue = NULL;
1425 3822 : var->value = *value;
1426 :
1427 3822 : return true;
1428 : }
1429 :
1430 : /* Assign an integer value to a variable, creating it if need be */
1431 : /* Returns false on failure (bad name) */
1432 : static bool
1433 466 : putVariableInt(CState *st, const char *context, char *name, int64 value)
1434 : {
1435 : PgBenchValue val;
1436 :
1437 466 : setIntValue(&val, value);
1438 466 : return putVariableValue(st, context, name, &val);
1439 : }
1440 :
1441 : /*
1442 : * Parse a possible variable reference (:varname).
1443 : *
1444 : * "sql" points at a colon. If what follows it looks like a valid
1445 : * variable name, return a malloc'd string containing the variable name,
1446 : * and set *eaten to the number of characters consumed.
1447 : * Otherwise, return NULL.
1448 : */
1449 : static char *
1450 976 : parseVariable(const char *sql, int *eaten)
1451 : {
1452 976 : int i = 0;
1453 : char *name;
1454 :
1455 : do
1456 : {
1457 4062 : i++;
1458 8124 : } while (IS_HIGHBIT_SET(sql[i]) ||
1459 4062 : strchr("ABCDEFGHIJKLMNOPQRSTUVWXYZ" "abcdefghijklmnopqrstuvwxyz"
1460 4062 : "_0123456789", sql[i]) != NULL);
1461 976 : if (i == 1)
1462 30 : return NULL; /* no valid variable name chars */
1463 :
1464 946 : name = pg_malloc(i);
1465 946 : memcpy(name, &sql[1], i - 1);
1466 946 : name[i - 1] = '\0';
1467 :
1468 946 : *eaten = i;
1469 946 : return name;
1470 : }
1471 :
1472 : static char *
1473 944 : replaceVariable(char **sql, char *param, int len, char *value)
1474 : {
1475 944 : int valueln = strlen(value);
1476 :
1477 944 : if (valueln > len)
1478 : {
1479 652 : size_t offset = param - *sql;
1480 :
1481 652 : *sql = pg_realloc(*sql, strlen(*sql) - len + valueln + 1);
1482 652 : param = *sql + offset;
1483 : }
1484 :
1485 944 : if (valueln != len)
1486 860 : memmove(param + valueln, param + len, strlen(param + len) + 1);
1487 944 : memcpy(param, value, valueln);
1488 :
1489 944 : return param + valueln;
1490 : }
1491 :
1492 : static char *
1493 826 : assignVariables(CState *st, char *sql)
1494 : {
1495 : char *p,
1496 : *name,
1497 : *val;
1498 :
1499 826 : p = sql;
1500 2572 : while ((p = strchr(p, ':')) != NULL)
1501 : {
1502 : int eaten;
1503 :
1504 920 : name = parseVariable(p, &eaten);
1505 920 : if (name == NULL)
1506 : {
1507 80 : while (*p == ':')
1508 : {
1509 40 : p++;
1510 : }
1511 40 : continue;
1512 : }
1513 :
1514 900 : val = getVariable(st, name);
1515 900 : free(name);
1516 900 : if (val == NULL)
1517 : {
1518 0 : p++;
1519 0 : continue;
1520 : }
1521 :
1522 900 : p = replaceVariable(&sql, p, eaten, val);
1523 : }
1524 :
1525 826 : return sql;
1526 : }
1527 :
1528 : static void
1529 3252 : getQueryParams(CState *st, const Command *command, const char **params)
1530 : {
1531 : int i;
1532 :
1533 6572 : for (i = 0; i < command->argc - 1; i++)
1534 3320 : params[i] = getVariable(st, command->argv[i + 1]);
1535 3252 : }
1536 :
1537 : static char *
1538 8 : valueTypeName(PgBenchValue *pval)
1539 : {
1540 8 : if (pval->type == PGBT_NO_VALUE)
1541 0 : return "none";
1542 8 : else if (pval->type == PGBT_NULL)
1543 0 : return "null";
1544 8 : else if (pval->type == PGBT_INT)
1545 0 : return "int";
1546 8 : else if (pval->type == PGBT_DOUBLE)
1547 2 : return "double";
1548 6 : else if (pval->type == PGBT_BOOLEAN)
1549 6 : return "boolean";
1550 : else
1551 : {
1552 : /* internal error, should never get there */
1553 : Assert(false);
1554 0 : return NULL;
1555 : }
1556 : }
1557 :
1558 : /* get a value as a boolean, or tell if there is a problem */
1559 : static bool
1560 168 : coerceToBool(PgBenchValue *pval, bool *bval)
1561 : {
1562 168 : if (pval->type == PGBT_BOOLEAN)
1563 : {
1564 166 : *bval = pval->u.bval;
1565 166 : return true;
1566 : }
1567 : else /* NULL, INT or DOUBLE */
1568 : {
1569 2 : fprintf(stderr, "cannot coerce %s to boolean\n", valueTypeName(pval));
1570 2 : *bval = false; /* suppress uninitialized-variable warnings */
1571 2 : return false;
1572 : }
1573 : }
1574 :
1575 : /*
1576 : * Return true or false from an expression for conditional purposes.
1577 : * Non zero numerical values are true, zero and NULL are false.
1578 : */
1579 : static bool
1580 48 : valueTruth(PgBenchValue *pval)
1581 : {
1582 48 : switch (pval->type)
1583 : {
1584 : case PGBT_NULL:
1585 2 : return false;
1586 : case PGBT_BOOLEAN:
1587 42 : return pval->u.bval;
1588 : case PGBT_INT:
1589 2 : return pval->u.ival != 0;
1590 : case PGBT_DOUBLE:
1591 2 : return pval->u.dval != 0.0;
1592 : default:
1593 : /* internal error, unexpected type */
1594 : Assert(0);
1595 0 : return false;
1596 : }
1597 : }
1598 :
1599 : /* get a value as an int, tell if there is a problem */
1600 : static bool
1601 12628 : coerceToInt(PgBenchValue *pval, int64 *ival)
1602 : {
1603 12628 : if (pval->type == PGBT_INT)
1604 : {
1605 12620 : *ival = pval->u.ival;
1606 12620 : return true;
1607 : }
1608 8 : else if (pval->type == PGBT_DOUBLE)
1609 : {
1610 4 : double dval = pval->u.dval;
1611 :
1612 4 : if (dval < PG_INT64_MIN || PG_INT64_MAX < dval)
1613 : {
1614 2 : fprintf(stderr, "double to int overflow for %f\n", dval);
1615 2 : return false;
1616 : }
1617 2 : *ival = (int64) dval;
1618 2 : return true;
1619 : }
1620 : else /* BOOLEAN or NULL */
1621 : {
1622 4 : fprintf(stderr, "cannot coerce %s to int\n", valueTypeName(pval));
1623 4 : return false;
1624 : }
1625 : }
1626 :
1627 : /* get a value as a double, or tell if there is a problem */
1628 : static bool
1629 240 : coerceToDouble(PgBenchValue *pval, double *dval)
1630 : {
1631 240 : if (pval->type == PGBT_DOUBLE)
1632 : {
1633 178 : *dval = pval->u.dval;
1634 178 : return true;
1635 : }
1636 62 : else if (pval->type == PGBT_INT)
1637 : {
1638 60 : *dval = (double) pval->u.ival;
1639 60 : return true;
1640 : }
1641 : else /* BOOLEAN or NULL */
1642 : {
1643 2 : fprintf(stderr, "cannot coerce %s to double\n", valueTypeName(pval));
1644 2 : return false;
1645 : }
1646 : }
1647 :
1648 : /* assign a null value */
1649 : static void
1650 8 : setNullValue(PgBenchValue *pv)
1651 : {
1652 8 : pv->type = PGBT_NULL;
1653 8 : pv->u.ival = 0;
1654 8 : }
1655 :
1656 : /* assign a boolean value */
1657 : static void
1658 202 : setBoolValue(PgBenchValue *pv, bool bval)
1659 : {
1660 202 : pv->type = PGBT_BOOLEAN;
1661 202 : pv->u.bval = bval;
1662 202 : }
1663 :
1664 : /* assign an integer value */
1665 : static void
1666 6706 : setIntValue(PgBenchValue *pv, int64 ival)
1667 : {
1668 6706 : pv->type = PGBT_INT;
1669 6706 : pv->u.ival = ival;
1670 6706 : }
1671 :
1672 : /* assign a double value */
1673 : static void
1674 78 : setDoubleValue(PgBenchValue *pv, double dval)
1675 : {
1676 78 : pv->type = PGBT_DOUBLE;
1677 78 : pv->u.dval = dval;
1678 78 : }
1679 :
1680 : static bool
1681 6708 : isLazyFunc(PgBenchFunction func)
1682 : {
1683 6708 : return func == PGBENCH_AND || func == PGBENCH_OR || func == PGBENCH_CASE;
1684 : }
1685 :
1686 : /* lazy evaluation of some functions */
1687 : static bool
1688 106 : evalLazyFunc(TState *thread, CState *st,
1689 : PgBenchFunction func, PgBenchExprLink *args, PgBenchValue *retval)
1690 : {
1691 : PgBenchValue a1,
1692 : a2;
1693 : bool ba1,
1694 : ba2;
1695 :
1696 : Assert(isLazyFunc(func) && args != NULL && args->next != NULL);
1697 :
1698 : /* args points to first condition */
1699 106 : if (!evaluateExpr(thread, st, args->expr, &a1))
1700 2 : return false;
1701 :
1702 : /* second condition for AND/OR and corresponding branch for CASE */
1703 104 : args = args->next;
1704 :
1705 104 : switch (func)
1706 : {
1707 : case PGBENCH_AND:
1708 64 : if (a1.type == PGBT_NULL)
1709 : {
1710 0 : setNullValue(retval);
1711 0 : return true;
1712 : }
1713 :
1714 64 : if (!coerceToBool(&a1, &ba1))
1715 0 : return false;
1716 :
1717 64 : if (!ba1)
1718 : {
1719 6 : setBoolValue(retval, false);
1720 6 : return true;
1721 : }
1722 :
1723 58 : if (!evaluateExpr(thread, st, args->expr, &a2))
1724 0 : return false;
1725 :
1726 58 : if (a2.type == PGBT_NULL)
1727 : {
1728 0 : setNullValue(retval);
1729 0 : return true;
1730 : }
1731 58 : else if (!coerceToBool(&a2, &ba2))
1732 0 : return false;
1733 : else
1734 : {
1735 58 : setBoolValue(retval, ba2);
1736 58 : return true;
1737 : }
1738 :
1739 : return true;
1740 :
1741 : case PGBENCH_OR:
1742 :
1743 8 : if (a1.type == PGBT_NULL)
1744 : {
1745 0 : setNullValue(retval);
1746 0 : return true;
1747 : }
1748 :
1749 8 : if (!coerceToBool(&a1, &ba1))
1750 0 : return false;
1751 :
1752 8 : if (ba1)
1753 : {
1754 2 : setBoolValue(retval, true);
1755 2 : return true;
1756 : }
1757 :
1758 6 : if (!evaluateExpr(thread, st, args->expr, &a2))
1759 0 : return false;
1760 :
1761 6 : if (a2.type == PGBT_NULL)
1762 : {
1763 0 : setNullValue(retval);
1764 0 : return true;
1765 : }
1766 6 : else if (!coerceToBool(&a2, &ba2))
1767 0 : return false;
1768 : else
1769 : {
1770 6 : setBoolValue(retval, ba2);
1771 6 : return true;
1772 : }
1773 :
1774 : case PGBENCH_CASE:
1775 : /* when true, execute branch */
1776 32 : if (valueTruth(&a1))
1777 22 : return evaluateExpr(thread, st, args->expr, retval);
1778 :
1779 : /* now args contains next condition or final else expression */
1780 10 : args = args->next;
1781 :
1782 : /* final else case? */
1783 10 : if (args->next == NULL)
1784 6 : return evaluateExpr(thread, st, args->expr, retval);
1785 :
1786 : /* no, another when, proceed */
1787 4 : return evalLazyFunc(thread, st, PGBENCH_CASE, args, retval);
1788 :
1789 : default:
1790 : /* internal error, cannot get here */
1791 : Assert(0);
1792 0 : break;
1793 : }
1794 0 : return false;
1795 : }
1796 :
1797 : /* maximum number of function arguments */
1798 : #define MAX_FARGS 16
1799 :
1800 : /*
1801 : * Recursive evaluation of standard functions,
1802 : * which do not require lazy evaluation.
1803 : */
1804 : static bool
1805 6606 : evalStandardFunc(TState *thread, CState *st,
1806 : PgBenchFunction func, PgBenchExprLink *args,
1807 : PgBenchValue *retval)
1808 : {
1809 : /* evaluate all function arguments */
1810 6606 : int nargs = 0;
1811 : PgBenchValue vargs[MAX_FARGS];
1812 6606 : PgBenchExprLink *l = args;
1813 6606 : bool has_null = false;
1814 :
1815 19726 : for (nargs = 0; nargs < MAX_FARGS && l != NULL; nargs++, l = l->next)
1816 : {
1817 13122 : if (!evaluateExpr(thread, st, l->expr, &vargs[nargs]))
1818 2 : return false;
1819 13120 : has_null |= vargs[nargs].type == PGBT_NULL;
1820 : }
1821 :
1822 6604 : if (l != NULL)
1823 : {
1824 2 : fprintf(stderr,
1825 : "too many function arguments, maximum is %d\n", MAX_FARGS);
1826 2 : return false;
1827 : }
1828 :
1829 : /* NULL arguments */
1830 6602 : if (has_null && func != PGBENCH_IS && func != PGBENCH_DEBUG)
1831 : {
1832 6 : setNullValue(retval);
1833 6 : return true;
1834 : }
1835 :
1836 : /* then evaluate function */
1837 6596 : switch (func)
1838 : {
1839 : /* overloaded operators */
1840 : case PGBENCH_ADD:
1841 : case PGBENCH_SUB:
1842 : case PGBENCH_MUL:
1843 : case PGBENCH_DIV:
1844 : case PGBENCH_MOD:
1845 : case PGBENCH_EQ:
1846 : case PGBENCH_NE:
1847 : case PGBENCH_LE:
1848 : case PGBENCH_LT:
1849 : {
1850 3250 : PgBenchValue *lval = &vargs[0],
1851 3250 : *rval = &vargs[1];
1852 :
1853 : Assert(nargs == 2);
1854 :
1855 : /* overloaded type management, double if some double */
1856 6462 : if ((lval->type == PGBT_DOUBLE ||
1857 3274 : rval->type == PGBT_DOUBLE) && func != PGBENCH_MOD)
1858 0 : {
1859 : double ld,
1860 : rd;
1861 :
1862 124 : if (!coerceToDouble(lval, &ld) ||
1863 62 : !coerceToDouble(rval, &rd))
1864 62 : return false;
1865 :
1866 62 : switch (func)
1867 : {
1868 : case PGBENCH_ADD:
1869 2 : setDoubleValue(retval, ld + rd);
1870 2 : return true;
1871 :
1872 : case PGBENCH_SUB:
1873 20 : setDoubleValue(retval, ld - rd);
1874 20 : return true;
1875 :
1876 : case PGBENCH_MUL:
1877 16 : setDoubleValue(retval, ld * rd);
1878 16 : return true;
1879 :
1880 : case PGBENCH_DIV:
1881 4 : setDoubleValue(retval, ld / rd);
1882 4 : return true;
1883 :
1884 : case PGBENCH_EQ:
1885 8 : setBoolValue(retval, ld == rd);
1886 8 : return true;
1887 :
1888 : case PGBENCH_NE:
1889 4 : setBoolValue(retval, ld != rd);
1890 4 : return true;
1891 :
1892 : case PGBENCH_LE:
1893 4 : setBoolValue(retval, ld <= rd);
1894 4 : return true;
1895 :
1896 : case PGBENCH_LT:
1897 4 : setBoolValue(retval, ld < rd);
1898 4 : return true;
1899 :
1900 : default:
1901 : /* cannot get here */
1902 : Assert(0);
1903 : }
1904 : }
1905 : else /* we have integer operands, or % */
1906 : {
1907 : int64 li,
1908 : ri;
1909 :
1910 6374 : if (!coerceToInt(lval, &li) ||
1911 3186 : !coerceToInt(rval, &ri))
1912 3190 : return false;
1913 :
1914 3186 : switch (func)
1915 : {
1916 : case PGBENCH_ADD:
1917 30 : setIntValue(retval, li + ri);
1918 30 : return true;
1919 :
1920 : case PGBENCH_SUB:
1921 270 : setIntValue(retval, li - ri);
1922 270 : return true;
1923 :
1924 : case PGBENCH_MUL:
1925 2812 : setIntValue(retval, li * ri);
1926 2812 : return true;
1927 :
1928 : case PGBENCH_EQ:
1929 18 : setBoolValue(retval, li == ri);
1930 18 : return true;
1931 :
1932 : case PGBENCH_NE:
1933 8 : setBoolValue(retval, li != ri);
1934 8 : return true;
1935 :
1936 : case PGBENCH_LE:
1937 8 : setBoolValue(retval, li <= ri);
1938 8 : return true;
1939 :
1940 : case PGBENCH_LT:
1941 22 : setBoolValue(retval, li < ri);
1942 22 : return true;
1943 :
1944 : case PGBENCH_DIV:
1945 : case PGBENCH_MOD:
1946 18 : if (ri == 0)
1947 : {
1948 2 : fprintf(stderr, "division by zero\n");
1949 2 : return false;
1950 : }
1951 : /* special handling of -1 divisor */
1952 16 : if (ri == -1)
1953 : {
1954 6 : if (func == PGBENCH_DIV)
1955 : {
1956 : /* overflow check (needed for INT64_MIN) */
1957 4 : if (li == PG_INT64_MIN)
1958 : {
1959 2 : fprintf(stderr, "bigint out of range\n");
1960 2 : return false;
1961 : }
1962 : else
1963 2 : setIntValue(retval, -li);
1964 : }
1965 : else
1966 2 : setIntValue(retval, 0);
1967 4 : return true;
1968 : }
1969 : /* else divisor is not -1 */
1970 10 : if (func == PGBENCH_DIV)
1971 4 : setIntValue(retval, li / ri);
1972 : else /* func == PGBENCH_MOD */
1973 6 : setIntValue(retval, li % ri);
1974 :
1975 10 : return true;
1976 :
1977 : default:
1978 : /* cannot get here */
1979 : Assert(0);
1980 : }
1981 : }
1982 :
1983 : Assert(0);
1984 0 : return false; /* NOTREACHED */
1985 : }
1986 :
1987 : /* integer bitwise operators */
1988 : case PGBENCH_BITAND:
1989 : case PGBENCH_BITOR:
1990 : case PGBENCH_BITXOR:
1991 : case PGBENCH_LSHIFT:
1992 : case PGBENCH_RSHIFT:
1993 : {
1994 : int64 li,
1995 : ri;
1996 :
1997 16 : if (!coerceToInt(&vargs[0], &li) || !coerceToInt(&vargs[1], &ri))
1998 0 : return false;
1999 :
2000 16 : if (func == PGBENCH_BITAND)
2001 2 : setIntValue(retval, li & ri);
2002 14 : else if (func == PGBENCH_BITOR)
2003 4 : setIntValue(retval, li | ri);
2004 10 : else if (func == PGBENCH_BITXOR)
2005 6 : setIntValue(retval, li ^ ri);
2006 4 : else if (func == PGBENCH_LSHIFT)
2007 2 : setIntValue(retval, li << ri);
2008 2 : else if (func == PGBENCH_RSHIFT)
2009 2 : setIntValue(retval, li >> ri);
2010 : else /* cannot get here */
2011 : Assert(0);
2012 :
2013 16 : return true;
2014 : }
2015 :
2016 : /* logical operators */
2017 : case PGBENCH_NOT:
2018 : {
2019 : bool b;
2020 :
2021 32 : if (!coerceToBool(&vargs[0], &b))
2022 2 : return false;
2023 :
2024 30 : setBoolValue(retval, !b);
2025 30 : return true;
2026 : }
2027 :
2028 : /* no arguments */
2029 : case PGBENCH_PI:
2030 2 : setDoubleValue(retval, M_PI);
2031 2 : return true;
2032 :
2033 : /* 1 overloaded argument */
2034 : case PGBENCH_ABS:
2035 : {
2036 4 : PgBenchValue *varg = &vargs[0];
2037 :
2038 : Assert(nargs == 1);
2039 :
2040 4 : if (varg->type == PGBT_INT)
2041 : {
2042 2 : int64 i = varg->u.ival;
2043 :
2044 2 : setIntValue(retval, i < 0 ? -i : i);
2045 : }
2046 : else
2047 : {
2048 2 : double d = varg->u.dval;
2049 :
2050 : Assert(varg->type == PGBT_DOUBLE);
2051 2 : setDoubleValue(retval, d < 0.0 ? -d : d);
2052 : }
2053 :
2054 4 : return true;
2055 : }
2056 :
2057 : case PGBENCH_DEBUG:
2058 : {
2059 130 : PgBenchValue *varg = &vargs[0];
2060 :
2061 : Assert(nargs == 1);
2062 :
2063 130 : fprintf(stderr, "debug(script=%d,command=%d): ",
2064 130 : st->use_file, st->command + 1);
2065 :
2066 130 : if (varg->type == PGBT_NULL)
2067 4 : fprintf(stderr, "null\n");
2068 126 : else if (varg->type == PGBT_BOOLEAN)
2069 20 : fprintf(stderr, "boolean %s\n", varg->u.bval ? "true" : "false");
2070 106 : else if (varg->type == PGBT_INT)
2071 74 : fprintf(stderr, "int " INT64_FORMAT "\n", varg->u.ival);
2072 32 : else if (varg->type == PGBT_DOUBLE)
2073 32 : fprintf(stderr, "double %.*g\n", DBL_DIG, varg->u.dval);
2074 : else /* internal error, unexpected type */
2075 : Assert(0);
2076 :
2077 130 : *retval = *varg;
2078 :
2079 130 : return true;
2080 : }
2081 :
2082 : /* 1 double argument */
2083 : case PGBENCH_DOUBLE:
2084 : case PGBENCH_SQRT:
2085 : case PGBENCH_LN:
2086 : case PGBENCH_EXP:
2087 : {
2088 : double dval;
2089 :
2090 : Assert(nargs == 1);
2091 :
2092 10 : if (!coerceToDouble(&vargs[0], &dval))
2093 2 : return false;
2094 :
2095 8 : if (func == PGBENCH_SQRT)
2096 2 : dval = sqrt(dval);
2097 6 : else if (func == PGBENCH_LN)
2098 2 : dval = log(dval);
2099 4 : else if (func == PGBENCH_EXP)
2100 2 : dval = exp(dval);
2101 : /* else is cast: do nothing */
2102 :
2103 8 : setDoubleValue(retval, dval);
2104 8 : return true;
2105 : }
2106 :
2107 : /* 1 int argument */
2108 : case PGBENCH_INT:
2109 : {
2110 : int64 ival;
2111 :
2112 : Assert(nargs == 1);
2113 :
2114 4 : if (!coerceToInt(&vargs[0], &ival))
2115 2 : return false;
2116 :
2117 2 : setIntValue(retval, ival);
2118 2 : return true;
2119 : }
2120 :
2121 : /* variable number of arguments */
2122 : case PGBENCH_LEAST:
2123 : case PGBENCH_GREATEST:
2124 : {
2125 : bool havedouble;
2126 : int i;
2127 :
2128 : Assert(nargs >= 1);
2129 :
2130 : /* need double result if any input is double */
2131 8 : havedouble = false;
2132 28 : for (i = 0; i < nargs; i++)
2133 : {
2134 24 : if (vargs[i].type == PGBT_DOUBLE)
2135 : {
2136 4 : havedouble = true;
2137 4 : break;
2138 : }
2139 : }
2140 8 : if (havedouble)
2141 : {
2142 : double extremum;
2143 :
2144 4 : if (!coerceToDouble(&vargs[0], &extremum))
2145 0 : return false;
2146 12 : for (i = 1; i < nargs; i++)
2147 : {
2148 : double dval;
2149 :
2150 8 : if (!coerceToDouble(&vargs[i], &dval))
2151 0 : return false;
2152 8 : if (func == PGBENCH_LEAST)
2153 4 : extremum = Min(extremum, dval);
2154 : else
2155 4 : extremum = Max(extremum, dval);
2156 : }
2157 4 : setDoubleValue(retval, extremum);
2158 : }
2159 : else
2160 : {
2161 : int64 extremum;
2162 :
2163 4 : if (!coerceToInt(&vargs[0], &extremum))
2164 0 : return false;
2165 16 : for (i = 1; i < nargs; i++)
2166 : {
2167 : int64 ival;
2168 :
2169 12 : if (!coerceToInt(&vargs[i], &ival))
2170 0 : return false;
2171 12 : if (func == PGBENCH_LEAST)
2172 6 : extremum = Min(extremum, ival);
2173 : else
2174 6 : extremum = Max(extremum, ival);
2175 : }
2176 4 : setIntValue(retval, extremum);
2177 : }
2178 8 : return true;
2179 : }
2180 :
2181 : /* random functions */
2182 : case PGBENCH_RANDOM:
2183 : case PGBENCH_RANDOM_EXPONENTIAL:
2184 : case PGBENCH_RANDOM_GAUSSIAN:
2185 : case PGBENCH_RANDOM_ZIPFIAN:
2186 : {
2187 : int64 imin,
2188 : imax;
2189 :
2190 : Assert(nargs >= 2);
2191 :
2192 6178 : if (!coerceToInt(&vargs[0], &imin) ||
2193 3088 : !coerceToInt(&vargs[1], &imax))
2194 2 : return false;
2195 :
2196 : /* check random range */
2197 3088 : if (imin > imax)
2198 : {
2199 2 : fprintf(stderr, "empty range given to random\n");
2200 2 : return false;
2201 : }
2202 3086 : else if (imax - imin < 0 || (imax - imin) + 1 < 0)
2203 : {
2204 : /* prevent int overflows in random functions */
2205 2 : fprintf(stderr, "random range is too large\n");
2206 2 : return false;
2207 : }
2208 :
2209 3084 : if (func == PGBENCH_RANDOM)
2210 : {
2211 : Assert(nargs == 2);
2212 3026 : setIntValue(retval, getrand(thread, imin, imax));
2213 : }
2214 : else /* gaussian & exponential */
2215 : {
2216 : double param;
2217 :
2218 : Assert(nargs == 3);
2219 :
2220 58 : if (!coerceToDouble(&vargs[2], ¶m))
2221 8 : return false;
2222 :
2223 58 : if (func == PGBENCH_RANDOM_GAUSSIAN)
2224 : {
2225 8 : if (param < MIN_GAUSSIAN_PARAM)
2226 : {
2227 2 : fprintf(stderr,
2228 : "gaussian parameter must be at least %f "
2229 : "(not %f)\n", MIN_GAUSSIAN_PARAM, param);
2230 2 : return false;
2231 : }
2232 :
2233 6 : setIntValue(retval,
2234 : getGaussianRand(thread, imin, imax, param));
2235 : }
2236 50 : else if (func == PGBENCH_RANDOM_ZIPFIAN)
2237 : {
2238 42 : if (param <= 0.0 || param == 1.0 || param > MAX_ZIPFIAN_PARAM)
2239 : {
2240 4 : fprintf(stderr,
2241 : "zipfian parameter must be in range (0, 1) U (1, %d]"
2242 : " (got %f)\n", MAX_ZIPFIAN_PARAM, param);
2243 4 : return false;
2244 : }
2245 38 : setIntValue(retval,
2246 : getZipfianRand(thread, imin, imax, param));
2247 : }
2248 : else /* exponential */
2249 : {
2250 8 : if (param <= 0.0)
2251 : {
2252 2 : fprintf(stderr,
2253 : "exponential parameter must be greater than zero"
2254 : " (got %f)\n", param);
2255 2 : return false;
2256 : }
2257 :
2258 6 : setIntValue(retval,
2259 : getExponentialRand(thread, imin, imax, param));
2260 : }
2261 : }
2262 :
2263 3076 : return true;
2264 : }
2265 :
2266 : case PGBENCH_POW:
2267 : {
2268 18 : PgBenchValue *lval = &vargs[0];
2269 18 : PgBenchValue *rval = &vargs[1];
2270 : double ld,
2271 : rd;
2272 :
2273 : Assert(nargs == 2);
2274 :
2275 36 : if (!coerceToDouble(lval, &ld) ||
2276 18 : !coerceToDouble(rval, &rd))
2277 0 : return false;
2278 :
2279 18 : setDoubleValue(retval, pow(ld, rd));
2280 :
2281 18 : return true;
2282 : }
2283 :
2284 : case PGBENCH_IS:
2285 : {
2286 : Assert(nargs == 2);
2287 :
2288 : /*
2289 : * note: this simple implementation is more permissive than
2290 : * SQL
2291 : */
2292 20 : setBoolValue(retval,
2293 30 : vargs[0].type == vargs[1].type &&
2294 10 : vargs[0].u.bval == vargs[1].u.bval);
2295 20 : return true;
2296 : }
2297 :
2298 : /* hashing */
2299 : case PGBENCH_HASH_FNV1A:
2300 : case PGBENCH_HASH_MURMUR2:
2301 : {
2302 : int64 val,
2303 : seed;
2304 :
2305 : Assert(nargs == 2);
2306 :
2307 24 : if (!coerceToInt(&vargs[0], &val) ||
2308 12 : !coerceToInt(&vargs[1], &seed))
2309 0 : return false;
2310 :
2311 12 : if (func == PGBENCH_HASH_MURMUR2)
2312 10 : setIntValue(retval, getHashMurmur2(val, seed));
2313 2 : else if (func == PGBENCH_HASH_FNV1A)
2314 2 : setIntValue(retval, getHashFnv1a(val, seed));
2315 : else
2316 : /* cannot get here */
2317 : Assert(0);
2318 :
2319 12 : return true;
2320 : }
2321 :
2322 : default:
2323 : /* cannot get here */
2324 : Assert(0);
2325 : /* dead code to avoid a compiler warning */
2326 0 : return false;
2327 : }
2328 : }
2329 :
2330 : /* evaluate some function */
2331 : static bool
2332 6708 : evalFunc(TState *thread, CState *st,
2333 : PgBenchFunction func, PgBenchExprLink *args, PgBenchValue *retval)
2334 : {
2335 6708 : if (isLazyFunc(func))
2336 102 : return evalLazyFunc(thread, st, func, args, retval);
2337 : else
2338 6606 : return evalStandardFunc(thread, st, func, args, retval);
2339 : }
2340 :
2341 : /*
2342 : * Recursive evaluation of an expression in a pgbench script
2343 : * using the current state of variables.
2344 : * Returns whether the evaluation was ok,
2345 : * the value itself is returned through the retval pointer.
2346 : */
2347 : static bool
2348 16728 : evaluateExpr(TState *thread, CState *st, PgBenchExpr *expr, PgBenchValue *retval)
2349 : {
2350 16728 : switch (expr->etype)
2351 : {
2352 : case ENODE_CONSTANT:
2353 : {
2354 7152 : *retval = expr->u.constant;
2355 7152 : return true;
2356 : }
2357 :
2358 : case ENODE_VARIABLE:
2359 : {
2360 : Variable *var;
2361 :
2362 2868 : if ((var = lookupVariable(st, expr->u.variable.varname)) == NULL)
2363 : {
2364 2 : fprintf(stderr, "undefined variable \"%s\"\n",
2365 : expr->u.variable.varname);
2366 2 : return false;
2367 : }
2368 :
2369 2866 : if (!makeVariableValue(var))
2370 4 : return false;
2371 :
2372 2862 : *retval = var->value;
2373 2862 : return true;
2374 : }
2375 :
2376 : case ENODE_FUNCTION:
2377 6708 : return evalFunc(thread, st,
2378 : expr->u.function.function,
2379 : expr->u.function.args,
2380 : retval);
2381 :
2382 : default:
2383 : /* internal error which should never occur */
2384 0 : fprintf(stderr, "unexpected enode type in evaluation: %d\n",
2385 0 : expr->etype);
2386 0 : exit(1);
2387 : }
2388 : }
2389 :
2390 : /*
2391 : * Convert command name to meta-command enum identifier
2392 : */
2393 : static MetaCommand
2394 780 : getMetaCommand(const char *cmd)
2395 : {
2396 : MetaCommand mc;
2397 :
2398 780 : if (cmd == NULL)
2399 0 : mc = META_NONE;
2400 780 : else if (pg_strcasecmp(cmd, "set") == 0)
2401 668 : mc = META_SET;
2402 112 : else if (pg_strcasecmp(cmd, "setshell") == 0)
2403 8 : mc = META_SETSHELL;
2404 104 : else if (pg_strcasecmp(cmd, "shell") == 0)
2405 10 : mc = META_SHELL;
2406 94 : else if (pg_strcasecmp(cmd, "sleep") == 0)
2407 18 : mc = META_SLEEP;
2408 76 : else if (pg_strcasecmp(cmd, "if") == 0)
2409 24 : mc = META_IF;
2410 52 : else if (pg_strcasecmp(cmd, "elif") == 0)
2411 16 : mc = META_ELIF;
2412 36 : else if (pg_strcasecmp(cmd, "else") == 0)
2413 16 : mc = META_ELSE;
2414 20 : else if (pg_strcasecmp(cmd, "endif") == 0)
2415 18 : mc = META_ENDIF;
2416 : else
2417 2 : mc = META_NONE;
2418 780 : return mc;
2419 : }
2420 :
2421 : /*
2422 : * Run a shell command. The result is assigned to the variable if not NULL.
2423 : * Return true if succeeded, or false on error.
2424 : */
2425 : static bool
2426 12 : runShellCommand(CState *st, char *variable, char **argv, int argc)
2427 : {
2428 : char command[SHELL_COMMAND_SIZE];
2429 : int i,
2430 12 : len = 0;
2431 : FILE *fp;
2432 : char res[64];
2433 : char *endptr;
2434 : int retval;
2435 :
2436 : /*----------
2437 : * Join arguments with whitespace separators. Arguments starting with
2438 : * exactly one colon are treated as variables:
2439 : * name - append a string "name"
2440 : * :var - append a variable named 'var'
2441 : * ::name - append a string ":name"
2442 : *----------
2443 : */
2444 38 : for (i = 0; i < argc; i++)
2445 : {
2446 : char *arg;
2447 : int arglen;
2448 :
2449 28 : if (argv[i][0] != ':')
2450 : {
2451 22 : arg = argv[i]; /* a string literal */
2452 : }
2453 6 : else if (argv[i][1] == ':')
2454 : {
2455 2 : arg = argv[i] + 1; /* a string literal starting with colons */
2456 : }
2457 4 : else if ((arg = getVariable(st, argv[i] + 1)) == NULL)
2458 : {
2459 2 : fprintf(stderr, "%s: undefined variable \"%s\"\n",
2460 2 : argv[0], argv[i]);
2461 2 : return false;
2462 : }
2463 :
2464 26 : arglen = strlen(arg);
2465 26 : if (len + arglen + (i > 0 ? 1 : 0) >= SHELL_COMMAND_SIZE - 1)
2466 : {
2467 0 : fprintf(stderr, "%s: shell command is too long\n", argv[0]);
2468 0 : return false;
2469 : }
2470 :
2471 26 : if (i > 0)
2472 14 : command[len++] = ' ';
2473 26 : memcpy(command + len, arg, arglen);
2474 26 : len += arglen;
2475 : }
2476 :
2477 10 : command[len] = '\0';
2478 :
2479 : /* Fast path for non-assignment case */
2480 10 : if (variable == NULL)
2481 : {
2482 4 : if (system(command))
2483 : {
2484 2 : if (!timer_exceeded)
2485 2 : fprintf(stderr, "%s: could not launch shell command\n", argv[0]);
2486 2 : return false;
2487 : }
2488 2 : return true;
2489 : }
2490 :
2491 : /* Execute the command with pipe and read the standard output. */
2492 6 : if ((fp = popen(command, "r")) == NULL)
2493 : {
2494 0 : fprintf(stderr, "%s: could not launch shell command\n", argv[0]);
2495 0 : return false;
2496 : }
2497 6 : if (fgets(res, sizeof(res), fp) == NULL)
2498 : {
2499 2 : if (!timer_exceeded)
2500 2 : fprintf(stderr, "%s: could not read result of shell command\n", argv[0]);
2501 2 : (void) pclose(fp);
2502 2 : return false;
2503 : }
2504 4 : if (pclose(fp) < 0)
2505 : {
2506 0 : fprintf(stderr, "%s: could not close shell command\n", argv[0]);
2507 0 : return false;
2508 : }
2509 :
2510 : /* Check whether the result is an integer and assign it to the variable */
2511 4 : retval = (int) strtol(res, &endptr, 10);
2512 10 : while (*endptr != '\0' && isspace((unsigned char) *endptr))
2513 2 : endptr++;
2514 4 : if (*res == '\0' || *endptr != '\0')
2515 : {
2516 2 : fprintf(stderr, "%s: shell command must return an integer (not \"%s\")\n",
2517 : argv[0], res);
2518 2 : return false;
2519 : }
2520 2 : if (!putVariableInt(st, "setshell", variable, retval))
2521 0 : return false;
2522 :
2523 : #ifdef DEBUG
2524 : printf("shell parameter name: \"%s\", value: \"%s\"\n", argv[1], res);
2525 : #endif
2526 2 : return true;
2527 : }
2528 :
2529 : #define MAX_PREPARE_NAME 32
2530 : static void
2531 2166 : preparedStatementName(char *buffer, int file, int state)
2532 : {
2533 2166 : sprintf(buffer, "P%d_%d", file, state);
2534 2166 : }
2535 :
2536 : static void
2537 48 : commandFailed(CState *st, const char *cmd, const char *message)
2538 : {
2539 48 : fprintf(stderr,
2540 : "client %d aborted in command %d (%s) of script %d; %s\n",
2541 : st->id, st->command, cmd, st->use_file, message);
2542 48 : }
2543 :
2544 : /* return a script number with a weighted choice. */
2545 : static int
2546 2810 : chooseScript(TState *thread)
2547 : {
2548 2810 : int i = 0;
2549 : int64 w;
2550 :
2551 2810 : if (num_scripts == 1)
2552 2610 : return 0;
2553 :
2554 200 : w = getrand(thread, 0, total_weight - 1);
2555 : do
2556 : {
2557 340 : w -= sql_script[i++].weight;
2558 340 : } while (w >= 0);
2559 :
2560 200 : return i - 1;
2561 : }
2562 :
2563 : /* Send a SQL command, using the chosen querymode */
2564 : static bool
2565 4078 : sendCommand(CState *st, Command *command)
2566 : {
2567 : int r;
2568 :
2569 4078 : if (querymode == QUERY_SIMPLE)
2570 : {
2571 : char *sql;
2572 :
2573 826 : sql = pg_strdup(command->argv[0]);
2574 826 : sql = assignVariables(st, sql);
2575 :
2576 826 : if (debug)
2577 0 : fprintf(stderr, "client %d sending %s\n", st->id, sql);
2578 826 : r = PQsendQuery(st->con, sql);
2579 826 : free(sql);
2580 : }
2581 3252 : else if (querymode == QUERY_EXTENDED)
2582 : {
2583 1120 : const char *sql = command->argv[0];
2584 : const char *params[MAX_ARGS];
2585 :
2586 1120 : getQueryParams(st, command, params);
2587 :
2588 1120 : if (debug)
2589 0 : fprintf(stderr, "client %d sending %s\n", st->id, sql);
2590 1120 : r = PQsendQueryParams(st->con, sql, command->argc - 1,
2591 : NULL, params, NULL, NULL, 0);
2592 : }
2593 2132 : else if (querymode == QUERY_PREPARED)
2594 : {
2595 : char name[MAX_PREPARE_NAME];
2596 : const char *params[MAX_ARGS];
2597 :
2598 2132 : if (!st->prepared[st->use_file])
2599 : {
2600 : int j;
2601 30 : Command **commands = sql_script[st->use_file].commands;
2602 :
2603 82 : for (j = 0; commands[j] != NULL; j++)
2604 : {
2605 : PGresult *res;
2606 : char name[MAX_PREPARE_NAME];
2607 :
2608 52 : if (commands[j]->type != SQL_COMMAND)
2609 18 : continue;
2610 34 : preparedStatementName(name, st->use_file, j);
2611 68 : res = PQprepare(st->con, name,
2612 68 : commands[j]->argv[0], commands[j]->argc - 1, NULL);
2613 34 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
2614 2 : fprintf(stderr, "%s", PQerrorMessage(st->con));
2615 34 : PQclear(res);
2616 : }
2617 30 : st->prepared[st->use_file] = true;
2618 : }
2619 :
2620 2132 : getQueryParams(st, command, params);
2621 2132 : preparedStatementName(name, st->use_file, st->command);
2622 :
2623 2132 : if (debug)
2624 1400 : fprintf(stderr, "client %d sending %s\n", st->id, name);
2625 2132 : r = PQsendQueryPrepared(st->con, name, command->argc - 1,
2626 : params, NULL, NULL, 0);
2627 : }
2628 : else /* unknown sql mode */
2629 0 : r = 0;
2630 :
2631 4078 : if (r == 0)
2632 : {
2633 0 : if (debug)
2634 0 : fprintf(stderr, "client %d could not send %s\n",
2635 : st->id, command->argv[0]);
2636 0 : st->ecnt++;
2637 0 : return false;
2638 : }
2639 : else
2640 4078 : return true;
2641 : }
2642 :
2643 : /*
2644 : * Parse the argument to a \sleep command, and return the requested amount
2645 : * of delay, in microseconds. Returns true on success, false on error.
2646 : */
2647 : static bool
2648 10 : evaluateSleep(CState *st, int argc, char **argv, int *usecs)
2649 : {
2650 : char *var;
2651 : int usec;
2652 :
2653 10 : if (*argv[1] == ':')
2654 : {
2655 6 : if ((var = getVariable(st, argv[1] + 1)) == NULL)
2656 : {
2657 2 : fprintf(stderr, "%s: undefined variable \"%s\"\n",
2658 2 : argv[0], argv[1]);
2659 2 : return false;
2660 : }
2661 4 : usec = atoi(var);
2662 : }
2663 : else
2664 4 : usec = atoi(argv[1]);
2665 :
2666 8 : if (argc > 2)
2667 : {
2668 6 : if (pg_strcasecmp(argv[2], "ms") == 0)
2669 2 : usec *= 1000;
2670 4 : else if (pg_strcasecmp(argv[2], "s") == 0)
2671 2 : usec *= 1000000;
2672 : }
2673 : else
2674 2 : usec *= 1000000;
2675 :
2676 8 : *usecs = usec;
2677 8 : return true;
2678 : }
2679 :
2680 : /*
2681 : * Advance the state machine of a connection, if possible.
2682 : */
2683 : static void
2684 4198 : doCustom(TState *thread, CState *st, StatsData *agg)
2685 : {
2686 : PGresult *res;
2687 : Command *command;
2688 : instr_time now;
2689 4198 : bool end_tx_processed = false;
2690 : int64 wait;
2691 :
2692 : /*
2693 : * gettimeofday() isn't free, so we get the current timestamp lazily the
2694 : * first time it's needed, and reuse the same value throughout this
2695 : * function after that. This also ensures that e.g. the calculated
2696 : * latency reported in the log file and in the totals are the same. Zero
2697 : * means "not set yet". Reset "now" when we execute shell commands or
2698 : * expressions, which might take a non-negligible amount of time, though.
2699 : */
2700 4198 : INSTR_TIME_SET_ZERO(now);
2701 :
2702 : /*
2703 : * Loop in the state machine, until we have to wait for a result from the
2704 : * server (or have to sleep, for throttling or for \sleep).
2705 : *
2706 : * Note: In the switch-statement below, 'break' will loop back here,
2707 : * meaning "continue in the state machine". Return is used to return to
2708 : * the caller.
2709 : */
2710 : for (;;)
2711 : {
2712 66254 : switch (st->state)
2713 : {
2714 : /*
2715 : * Select transaction to run.
2716 : */
2717 : case CSTATE_CHOOSE_SCRIPT:
2718 :
2719 2810 : st->use_file = chooseScript(thread);
2720 :
2721 2810 : if (debug)
2722 1400 : fprintf(stderr, "client %d executing script \"%s\"\n", st->id,
2723 1400 : sql_script[st->use_file].desc);
2724 :
2725 2810 : if (throttle_delay > 0)
2726 402 : st->state = CSTATE_START_THROTTLE;
2727 : else
2728 2408 : st->state = CSTATE_START_TX;
2729 : /* check consistency */
2730 : Assert(conditional_stack_empty(st->cstack));
2731 2810 : break;
2732 :
2733 : /*
2734 : * Handle throttling once per transaction by sleeping.
2735 : */
2736 : case CSTATE_START_THROTTLE:
2737 :
2738 : /*
2739 : * Generate a delay such that the series of delays will
2740 : * approximate a Poisson distribution centered on the
2741 : * throttle_delay time.
2742 : *
2743 : * If transactions are too slow or a given wait is shorter
2744 : * than a transaction, the next transaction will start right
2745 : * away.
2746 : */
2747 : Assert(throttle_delay > 0);
2748 402 : wait = getPoissonRand(thread, throttle_delay);
2749 :
2750 402 : thread->throttle_trigger += wait;
2751 402 : st->txn_scheduled = thread->throttle_trigger;
2752 :
2753 : /*
2754 : * stop client if next transaction is beyond pgbench end of
2755 : * execution
2756 : */
2757 402 : if (duration > 0 && st->txn_scheduled > end_time)
2758 : {
2759 0 : st->state = CSTATE_FINISHED;
2760 0 : break;
2761 : }
2762 :
2763 : /*
2764 : * If --latency-limit is used, and this slot is already late
2765 : * so that the transaction will miss the latency limit even if
2766 : * it completed immediately, we skip this time slot and
2767 : * iterate till the next slot that isn't late yet. But don't
2768 : * iterate beyond the -t limit, if one is given.
2769 : */
2770 402 : if (latency_limit)
2771 : {
2772 : int64 now_us;
2773 :
2774 402 : if (INSTR_TIME_IS_ZERO(now))
2775 6 : INSTR_TIME_SET_CURRENT(now);
2776 402 : now_us = INSTR_TIME_GET_MICROSEC(now);
2777 846 : while (thread->throttle_trigger < now_us - latency_limit &&
2778 44 : (nxacts <= 0 || st->cnt < nxacts))
2779 : {
2780 20 : processXactStats(thread, st, &now, true, agg);
2781 : /* next rendez-vous */
2782 20 : wait = getPoissonRand(thread, throttle_delay);
2783 20 : thread->throttle_trigger += wait;
2784 20 : st->txn_scheduled = thread->throttle_trigger;
2785 : }
2786 : /* stop client if -t exceeded */
2787 402 : if (nxacts > 0 && st->cnt >= nxacts)
2788 : {
2789 2 : st->state = CSTATE_FINISHED;
2790 2 : break;
2791 : }
2792 : }
2793 :
2794 400 : st->state = CSTATE_THROTTLE;
2795 400 : if (debug)
2796 0 : fprintf(stderr, "client %d throttling " INT64_FORMAT " us\n",
2797 : st->id, wait);
2798 400 : break;
2799 :
2800 : /*
2801 : * Wait until it's time to start next transaction.
2802 : */
2803 : case CSTATE_THROTTLE:
2804 400 : if (INSTR_TIME_IS_ZERO(now))
2805 0 : INSTR_TIME_SET_CURRENT(now);
2806 400 : if (INSTR_TIME_GET_MICROSEC(now) < st->txn_scheduled)
2807 0 : return; /* Still sleeping, nothing to do here */
2808 :
2809 : /* Else done sleeping, start the transaction */
2810 400 : st->state = CSTATE_START_TX;
2811 400 : break;
2812 :
2813 : /* Start new transaction */
2814 : case CSTATE_START_TX:
2815 :
2816 : /*
2817 : * Establish connection on first call, or if is_connect is
2818 : * true.
2819 : */
2820 2808 : if (st->con == NULL)
2821 : {
2822 : instr_time start;
2823 :
2824 220 : if (INSTR_TIME_IS_ZERO(now))
2825 220 : INSTR_TIME_SET_CURRENT(now);
2826 220 : start = now;
2827 220 : if ((st->con = doConnect()) == NULL)
2828 : {
2829 0 : fprintf(stderr, "client %d aborted while establishing connection\n",
2830 : st->id);
2831 0 : st->state = CSTATE_ABORTED;
2832 0 : break;
2833 : }
2834 220 : INSTR_TIME_SET_CURRENT(now);
2835 220 : INSTR_TIME_ACCUM_DIFF(thread->conn_time, now, start);
2836 :
2837 : /* Reset session-local state */
2838 220 : memset(st->prepared, 0, sizeof(st->prepared));
2839 : }
2840 :
2841 : /*
2842 : * Record transaction start time under logging, progress or
2843 : * throttling.
2844 : */
2845 2808 : if (use_log || progress || throttle_delay || latency_limit ||
2846 : per_script_stats)
2847 : {
2848 820 : if (INSTR_TIME_IS_ZERO(now))
2849 8 : INSTR_TIME_SET_CURRENT(now);
2850 820 : st->txn_begin = now;
2851 :
2852 : /*
2853 : * When not throttling, this is also the transaction's
2854 : * scheduled start time.
2855 : */
2856 820 : if (!throttle_delay)
2857 420 : st->txn_scheduled = INSTR_TIME_GET_MICROSEC(now);
2858 : }
2859 :
2860 : /* Begin with the first command */
2861 2808 : st->command = 0;
2862 2808 : st->state = CSTATE_START_COMMAND;
2863 2808 : break;
2864 :
2865 : /*
2866 : * Send a command to server (or execute a meta-command)
2867 : */
2868 : case CSTATE_START_COMMAND:
2869 10278 : command = sql_script[st->use_file].commands[st->command];
2870 :
2871 : /*
2872 : * If we reached the end of the script, move to end-of-xact
2873 : * processing.
2874 : */
2875 10278 : if (command == NULL)
2876 : {
2877 2760 : st->state = CSTATE_END_TX;
2878 2760 : break;
2879 : }
2880 :
2881 : /*
2882 : * Record statement start time if per-command latencies are
2883 : * requested
2884 : */
2885 7518 : if (is_latencies)
2886 : {
2887 800 : if (INSTR_TIME_IS_ZERO(now))
2888 0 : INSTR_TIME_SET_CURRENT(now);
2889 800 : st->stmt_begin = now;
2890 : }
2891 :
2892 7518 : if (command->type == SQL_COMMAND)
2893 : {
2894 4078 : if (!sendCommand(st, command))
2895 : {
2896 0 : commandFailed(st, "SQL", "SQL command send failed");
2897 0 : st->state = CSTATE_ABORTED;
2898 : }
2899 : else
2900 4078 : st->state = CSTATE_WAIT_RESULT;
2901 : }
2902 3440 : else if (command->type == META_COMMAND)
2903 : {
2904 3440 : int argc = command->argc,
2905 : i;
2906 3440 : char **argv = command->argv;
2907 :
2908 3440 : if (debug)
2909 : {
2910 1400 : fprintf(stderr, "client %d executing \\%s", st->id, argv[0]);
2911 2800 : for (i = 1; i < argc; i++)
2912 1400 : fprintf(stderr, " %s", argv[i]);
2913 1400 : fprintf(stderr, "\n");
2914 : }
2915 :
2916 3440 : if (command->meta == META_SLEEP)
2917 : {
2918 : /*
2919 : * A \sleep doesn't execute anything, we just get the
2920 : * delay from the argument, and enter the CSTATE_SLEEP
2921 : * state. (The per-command latency will be recorded
2922 : * in CSTATE_SLEEP state, not here, after the delay
2923 : * has elapsed.)
2924 : */
2925 : int usec;
2926 :
2927 10 : if (!evaluateSleep(st, argc, argv, &usec))
2928 : {
2929 2 : commandFailed(st, "sleep", "execution of meta-command failed");
2930 2 : st->state = CSTATE_ABORTED;
2931 2 : break;
2932 : }
2933 :
2934 8 : if (INSTR_TIME_IS_ZERO(now))
2935 2 : INSTR_TIME_SET_CURRENT(now);
2936 8 : st->sleep_until = INSTR_TIME_GET_MICROSEC(now) + usec;
2937 8 : st->state = CSTATE_SLEEP;
2938 8 : break;
2939 : }
2940 3468 : else if (command->meta == META_SET ||
2941 66 : command->meta == META_IF ||
2942 28 : command->meta == META_ELIF)
2943 3372 : {
2944 : /* backslash commands with an expression to evaluate */
2945 3412 : PgBenchExpr *expr = command->expr;
2946 : PgBenchValue result;
2947 :
2948 3422 : if (command->meta == META_ELIF &&
2949 10 : conditional_stack_peek(st->cstack) == IFSTATE_TRUE)
2950 : {
2951 : /*
2952 : * elif after executed block, skip eval and wait
2953 : * for endif
2954 : */
2955 4 : conditional_stack_poke(st->cstack, IFSTATE_IGNORED);
2956 4 : goto move_to_end_command;
2957 : }
2958 :
2959 3408 : if (!evaluateExpr(thread, st, expr, &result))
2960 : {
2961 34 : commandFailed(st, argv[0], "evaluation of meta-command failed");
2962 34 : st->state = CSTATE_ABORTED;
2963 70 : break;
2964 : }
2965 :
2966 3374 : if (command->meta == META_SET)
2967 : {
2968 3358 : if (!putVariableValue(st, argv[0], argv[1], &result))
2969 : {
2970 2 : commandFailed(st, "set", "assignment of meta-command failed");
2971 2 : st->state = CSTATE_ABORTED;
2972 2 : break;
2973 : }
2974 : }
2975 : else /* if and elif evaluated cases */
2976 : {
2977 16 : bool cond = valueTruth(&result);
2978 :
2979 : /* execute or not depending on evaluated condition */
2980 16 : if (command->meta == META_IF)
2981 : {
2982 10 : conditional_stack_push(st->cstack, cond ? IFSTATE_TRUE : IFSTATE_FALSE);
2983 : }
2984 : else /* elif */
2985 : {
2986 : /*
2987 : * we should get here only if the "elif"
2988 : * needed evaluation
2989 : */
2990 : Assert(conditional_stack_peek(st->cstack) == IFSTATE_FALSE);
2991 6 : conditional_stack_poke(st->cstack, cond ? IFSTATE_TRUE : IFSTATE_FALSE);
2992 : }
2993 : }
2994 : }
2995 18 : else if (command->meta == META_ELSE)
2996 : {
2997 2 : switch (conditional_stack_peek(st->cstack))
2998 : {
2999 : case IFSTATE_TRUE:
3000 2 : conditional_stack_poke(st->cstack, IFSTATE_ELSE_FALSE);
3001 2 : break;
3002 : case IFSTATE_FALSE: /* inconsistent if active */
3003 : case IFSTATE_IGNORED: /* inconsistent if active */
3004 : case IFSTATE_NONE: /* else without if */
3005 : case IFSTATE_ELSE_TRUE: /* else after else */
3006 : case IFSTATE_ELSE_FALSE: /* else after else */
3007 : default:
3008 : /* dead code if conditional check is ok */
3009 : Assert(false);
3010 : }
3011 2 : goto move_to_end_command;
3012 : }
3013 16 : else if (command->meta == META_ENDIF)
3014 : {
3015 : Assert(!conditional_stack_empty(st->cstack));
3016 4 : conditional_stack_pop(st->cstack);
3017 4 : goto move_to_end_command;
3018 : }
3019 12 : else if (command->meta == META_SETSHELL)
3020 : {
3021 6 : bool ret = runShellCommand(st, argv[1], argv + 2, argc - 2);
3022 :
3023 6 : if (timer_exceeded) /* timeout */
3024 : {
3025 0 : st->state = CSTATE_FINISHED;
3026 0 : break;
3027 : }
3028 6 : else if (!ret) /* on error */
3029 : {
3030 4 : commandFailed(st, "setshell", "execution of meta-command failed");
3031 4 : st->state = CSTATE_ABORTED;
3032 4 : break;
3033 : }
3034 : else
3035 : {
3036 : /* succeeded */
3037 : }
3038 : }
3039 6 : else if (command->meta == META_SHELL)
3040 : {
3041 6 : bool ret = runShellCommand(st, NULL, argv + 1, argc - 1);
3042 :
3043 6 : if (timer_exceeded) /* timeout */
3044 : {
3045 0 : st->state = CSTATE_FINISHED;
3046 0 : break;
3047 : }
3048 6 : else if (!ret) /* on error */
3049 : {
3050 4 : commandFailed(st, "shell", "execution of meta-command failed");
3051 4 : st->state = CSTATE_ABORTED;
3052 4 : break;
3053 : }
3054 : else
3055 : {
3056 : /* succeeded */
3057 : }
3058 : }
3059 :
3060 : move_to_end_command:
3061 :
3062 : /*
3063 : * executing the expression or shell command might take a
3064 : * non-negligible amount of time, so reset 'now'
3065 : */
3066 3386 : INSTR_TIME_SET_ZERO(now);
3067 :
3068 3386 : st->state = CSTATE_END_COMMAND;
3069 : }
3070 7464 : break;
3071 :
3072 : /*
3073 : * non executed conditional branch
3074 : */
3075 : case CSTATE_SKIP_COMMAND:
3076 : Assert(!conditional_active(st->cstack));
3077 : /* quickly skip commands until something to do... */
3078 : while (true)
3079 : {
3080 50 : command = sql_script[st->use_file].commands[st->command];
3081 :
3082 : /* cannot reach end of script in that state */
3083 : Assert(command != NULL);
3084 :
3085 : /*
3086 : * if this is conditional related, update conditional
3087 : * state
3088 : */
3089 64 : if (command->type == META_COMMAND &&
3090 64 : (command->meta == META_IF ||
3091 58 : command->meta == META_ELIF ||
3092 48 : command->meta == META_ELSE ||
3093 22 : command->meta == META_ENDIF))
3094 : {
3095 16 : switch (conditional_stack_peek(st->cstack))
3096 : {
3097 : case IFSTATE_FALSE:
3098 8 : if (command->meta == META_IF || command->meta == META_ELIF)
3099 : {
3100 : /* we must evaluate the condition */
3101 6 : st->state = CSTATE_START_COMMAND;
3102 : }
3103 2 : else if (command->meta == META_ELSE)
3104 : {
3105 : /* we must execute next command */
3106 2 : conditional_stack_poke(st->cstack, IFSTATE_ELSE_TRUE);
3107 2 : st->state = CSTATE_START_COMMAND;
3108 2 : st->command++;
3109 : }
3110 0 : else if (command->meta == META_ENDIF)
3111 : {
3112 : Assert(!conditional_stack_empty(st->cstack));
3113 0 : conditional_stack_pop(st->cstack);
3114 0 : if (conditional_active(st->cstack))
3115 0 : st->state = CSTATE_START_COMMAND;
3116 :
3117 : /*
3118 : * else state remains in
3119 : * CSTATE_SKIP_COMMAND
3120 : */
3121 0 : st->command++;
3122 : }
3123 8 : break;
3124 :
3125 : case IFSTATE_IGNORED:
3126 : case IFSTATE_ELSE_FALSE:
3127 8 : if (command->meta == META_IF)
3128 0 : conditional_stack_push(st->cstack, IFSTATE_IGNORED);
3129 8 : else if (command->meta == META_ENDIF)
3130 : {
3131 : Assert(!conditional_stack_empty(st->cstack));
3132 6 : conditional_stack_pop(st->cstack);
3133 6 : if (conditional_active(st->cstack))
3134 6 : st->state = CSTATE_START_COMMAND;
3135 : }
3136 : /* could detect "else" & "elif" after "else" */
3137 8 : st->command++;
3138 8 : break;
3139 :
3140 : case IFSTATE_NONE:
3141 : case IFSTATE_TRUE:
3142 : case IFSTATE_ELSE_TRUE:
3143 : default:
3144 :
3145 : /*
3146 : * inconsistent if inactive, unreachable dead
3147 : * code
3148 : */
3149 : Assert(false);
3150 : }
3151 16 : }
3152 : else
3153 : {
3154 : /* skip and consider next */
3155 16 : st->command++;
3156 : }
3157 :
3158 32 : if (st->state != CSTATE_SKIP_COMMAND)
3159 14 : break;
3160 : }
3161 14 : break;
3162 :
3163 : /*
3164 : * Wait for the current SQL command to complete
3165 : */
3166 : case CSTATE_WAIT_RESULT:
3167 8156 : command = sql_script[st->use_file].commands[st->command];
3168 8156 : if (debug)
3169 2800 : fprintf(stderr, "client %d receiving\n", st->id);
3170 8156 : if (!PQconsumeInput(st->con))
3171 : { /* there's something wrong */
3172 0 : commandFailed(st, "SQL", "perhaps the backend died while processing");
3173 0 : st->state = CSTATE_ABORTED;
3174 0 : break;
3175 : }
3176 8156 : if (PQisBusy(st->con))
3177 4078 : return; /* don't have the whole result yet */
3178 :
3179 : /*
3180 : * Read and discard the query result;
3181 : */
3182 4078 : res = PQgetResult(st->con);
3183 4078 : switch (PQresultStatus(res))
3184 : {
3185 : case PGRES_COMMAND_OK:
3186 : case PGRES_TUPLES_OK:
3187 : case PGRES_EMPTY_QUERY:
3188 : /* OK */
3189 4076 : PQclear(res);
3190 4076 : discard_response(st);
3191 4076 : st->state = CSTATE_END_COMMAND;
3192 4076 : break;
3193 : default:
3194 2 : commandFailed(st, "SQL", PQerrorMessage(st->con));
3195 2 : PQclear(res);
3196 2 : st->state = CSTATE_ABORTED;
3197 2 : break;
3198 : }
3199 4078 : break;
3200 :
3201 : /*
3202 : * Wait until sleep is done. This state is entered after a
3203 : * \sleep metacommand. The behavior is similar to
3204 : * CSTATE_THROTTLE, but proceeds to CSTATE_START_COMMAND
3205 : * instead of CSTATE_START_TX.
3206 : */
3207 : case CSTATE_SLEEP:
3208 12 : if (INSTR_TIME_IS_ZERO(now))
3209 4 : INSTR_TIME_SET_CURRENT(now);
3210 12 : if (INSTR_TIME_GET_MICROSEC(now) < st->sleep_until)
3211 4 : return; /* Still sleeping, nothing to do here */
3212 : /* Else done sleeping. */
3213 8 : st->state = CSTATE_END_COMMAND;
3214 8 : break;
3215 :
3216 : /*
3217 : * End of command: record stats and proceed to next command.
3218 : */
3219 : case CSTATE_END_COMMAND:
3220 :
3221 : /*
3222 : * command completed: accumulate per-command execution times
3223 : * in thread-local data structure, if per-command latencies
3224 : * are requested.
3225 : */
3226 7470 : if (is_latencies)
3227 : {
3228 800 : if (INSTR_TIME_IS_ZERO(now))
3229 800 : INSTR_TIME_SET_CURRENT(now);
3230 :
3231 : /* XXX could use a mutex here, but we choose not to */
3232 800 : command = sql_script[st->use_file].commands[st->command];
3233 800 : addToSimpleStats(&command->stats,
3234 800 : INSTR_TIME_GET_DOUBLE(now) -
3235 800 : INSTR_TIME_GET_DOUBLE(st->stmt_begin));
3236 : }
3237 :
3238 : /* Go ahead with next command, to be executed or skipped */
3239 7470 : st->command++;
3240 14940 : st->state = conditional_active(st->cstack) ?
3241 7470 : CSTATE_START_COMMAND : CSTATE_SKIP_COMMAND;
3242 7470 : break;
3243 :
3244 : /*
3245 : * End of transaction.
3246 : */
3247 : case CSTATE_END_TX:
3248 :
3249 : /* transaction finished: calculate latency and do log */
3250 2760 : processXactStats(thread, st, &now, false, agg);
3251 :
3252 : /* conditional stack must be empty */
3253 2760 : if (!conditional_stack_empty(st->cstack))
3254 : {
3255 0 : fprintf(stderr, "end of script reached within a conditional, missing \\endif\n");
3256 0 : exit(1);
3257 : }
3258 :
3259 2760 : if (is_connect)
3260 : {
3261 220 : finishCon(st);
3262 220 : INSTR_TIME_SET_ZERO(now);
3263 : }
3264 :
3265 2760 : if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
3266 : {
3267 : /* exit success */
3268 66 : st->state = CSTATE_FINISHED;
3269 66 : break;
3270 : }
3271 :
3272 : /*
3273 : * No transaction is underway anymore.
3274 : */
3275 2694 : st->state = CSTATE_CHOOSE_SCRIPT;
3276 :
3277 : /*
3278 : * If we paced through all commands in the script in this
3279 : * loop, without returning to the caller even once, do it now.
3280 : * This gives the thread a chance to process other
3281 : * connections, and to do progress reporting. This can
3282 : * currently only happen if the script consists entirely of
3283 : * meta-commands.
3284 : */
3285 2694 : if (end_tx_processed)
3286 0 : return;
3287 : else
3288 : {
3289 2694 : end_tx_processed = true;
3290 2694 : break;
3291 : }
3292 :
3293 : /*
3294 : * Final states. Close the connection if it's still open.
3295 : */
3296 : case CSTATE_ABORTED:
3297 : case CSTATE_FINISHED:
3298 116 : finishCon(st);
3299 116 : return;
3300 : }
3301 : }
3302 : }
3303 :
3304 : /*
3305 : * Print log entry after completing one transaction.
3306 : *
3307 : * We print Unix-epoch timestamps in the log, so that entries can be
3308 : * correlated against other logs. On some platforms this could be obtained
3309 : * from the instr_time reading the caller has, but rather than get entangled
3310 : * with that, we just eat the cost of an extra syscall in all cases.
3311 : */
3312 : static void
3313 220 : doLog(TState *thread, CState *st,
3314 : StatsData *agg, bool skipped, double latency, double lag)
3315 : {
3316 220 : FILE *logfile = thread->logfile;
3317 :
3318 : Assert(use_log);
3319 :
3320 : /*
3321 : * Skip the log entry if sampling is enabled and this row doesn't belong
3322 : * to the random sample.
3323 : */
3324 420 : if (sample_rate != 0.0 &&
3325 200 : pg_erand48(thread->random_state) > sample_rate)
3326 108 : return;
3327 :
3328 : /* should we aggregate the results or not? */
3329 112 : if (agg_interval > 0)
3330 : {
3331 : /*
3332 : * Loop until we reach the interval of the current moment, and print
3333 : * any empty intervals in between (this may happen with very low tps,
3334 : * e.g. --rate=0.1).
3335 : */
3336 0 : time_t now = time(NULL);
3337 :
3338 0 : while (agg->start_time + agg_interval <= now)
3339 : {
3340 : /* print aggregated report to logfile */
3341 0 : fprintf(logfile, "%ld " INT64_FORMAT " %.0f %.0f %.0f %.0f",
3342 0 : (long) agg->start_time,
3343 : agg->cnt,
3344 : agg->latency.sum,
3345 : agg->latency.sum2,
3346 : agg->latency.min,
3347 : agg->latency.max);
3348 0 : if (throttle_delay)
3349 : {
3350 0 : fprintf(logfile, " %.0f %.0f %.0f %.0f",
3351 : agg->lag.sum,
3352 : agg->lag.sum2,
3353 : agg->lag.min,
3354 : agg->lag.max);
3355 0 : if (latency_limit)
3356 0 : fprintf(logfile, " " INT64_FORMAT, agg->skipped);
3357 : }
3358 0 : fputc('\n', logfile);
3359 :
3360 : /* reset data and move to next interval */
3361 0 : initStats(agg, agg->start_time + agg_interval);
3362 : }
3363 :
3364 : /* accumulate the current transaction */
3365 0 : accumStats(agg, skipped, latency, lag);
3366 : }
3367 : else
3368 : {
3369 : /* no, print raw transactions */
3370 : struct timeval tv;
3371 :
3372 112 : gettimeofday(&tv, NULL);
3373 112 : if (skipped)
3374 0 : fprintf(logfile, "%d " INT64_FORMAT " skipped %d %ld %ld",
3375 : st->id, st->cnt, st->use_file,
3376 0 : (long) tv.tv_sec, (long) tv.tv_usec);
3377 : else
3378 112 : fprintf(logfile, "%d " INT64_FORMAT " %.0f %d %ld %ld",
3379 : st->id, st->cnt, latency, st->use_file,
3380 112 : (long) tv.tv_sec, (long) tv.tv_usec);
3381 112 : if (throttle_delay)
3382 0 : fprintf(logfile, " %.0f", lag);
3383 112 : fputc('\n', logfile);
3384 : }
3385 : }
3386 :
3387 : /*
3388 : * Accumulate and report statistics at end of a transaction.
3389 : *
3390 : * (This is also called when a transaction is late and thus skipped.
3391 : * Note that even skipped transactions are counted in the "cnt" fields.)
3392 : */
3393 : static void
3394 2780 : processXactStats(TState *thread, CState *st, instr_time *now,
3395 : bool skipped, StatsData *agg)
3396 : {
3397 2780 : double latency = 0.0,
3398 2780 : lag = 0.0;
3399 2780 : bool thread_details = progress || throttle_delay || latency_limit,
3400 2780 : detailed = thread_details || use_log || per_script_stats;
3401 :
3402 2780 : if (detailed && !skipped)
3403 : {
3404 820 : if (INSTR_TIME_IS_ZERO(*now))
3405 420 : INSTR_TIME_SET_CURRENT(*now);
3406 :
3407 : /* compute latency & lag */
3408 820 : latency = INSTR_TIME_GET_MICROSEC(*now) - st->txn_scheduled;
3409 820 : lag = INSTR_TIME_GET_MICROSEC(st->txn_begin) - st->txn_scheduled;
3410 : }
3411 :
3412 2780 : if (thread_details)
3413 : {
3414 : /* keep detailed thread stats */
3415 420 : accumStats(&thread->stats, skipped, latency, lag);
3416 :
3417 : /* count transactions over the latency limit, if needed */
3418 420 : if (latency_limit && latency > latency_limit)
3419 0 : thread->latency_late++;
3420 : }
3421 : else
3422 : {
3423 : /* no detailed stats, just count */
3424 2360 : thread->stats.cnt++;
3425 : }
3426 :
3427 : /* client stat is just counting */
3428 2780 : st->cnt++;
3429 :
3430 2780 : if (use_log)
3431 220 : doLog(thread, st, agg, skipped, latency, lag);
3432 :
3433 : /* XXX could use a mutex here, but we choose not to */
3434 2780 : if (per_script_stats)
3435 200 : accumStats(&sql_script[st->use_file].stats, skipped, latency, lag);
3436 2780 : }
3437 :
3438 :
3439 : /* discard connections */
3440 : static void
3441 160 : disconnect_all(CState *state, int length)
3442 : {
3443 : int i;
3444 :
3445 392 : for (i = 0; i < length; i++)
3446 232 : finishCon(&state[i]);
3447 160 : }
3448 :
3449 : /*
3450 : * Remove old pgbench tables, if any exist
3451 : */
3452 : static void
3453 6 : initDropTables(PGconn *con)
3454 : {
3455 6 : fprintf(stderr, "dropping old tables...\n");
3456 :
3457 : /*
3458 : * We drop all the tables in one command, so that whether there are
3459 : * foreign key dependencies or not doesn't matter.
3460 : */
3461 6 : executeStatement(con, "drop table if exists "
3462 : "pgbench_accounts, "
3463 : "pgbench_branches, "
3464 : "pgbench_history, "
3465 : "pgbench_tellers");
3466 6 : }
3467 :
3468 : /*
3469 : * Create pgbench's standard tables
3470 : */
3471 : static void
3472 6 : initCreateTables(PGconn *con)
3473 : {
3474 : /*
3475 : * The scale factor at/beyond which 32-bit integers are insufficient for
3476 : * storing TPC-B account IDs.
3477 : *
3478 : * Although the actual threshold is 21474, we use 20000 because it is
3479 : * easier to document and remember, and isn't that far away from the real
3480 : * threshold.
3481 : */
3482 : #define SCALE_32BIT_THRESHOLD 20000
3483 :
3484 : /*
3485 : * Note: TPC-B requires at least 100 bytes per row, and the "filler"
3486 : * fields in these table declarations were intended to comply with that.
3487 : * The pgbench_accounts table complies with that because the "filler"
3488 : * column is set to blank-padded empty string. But for all other tables
3489 : * the columns default to NULL and so don't actually take any space. We
3490 : * could fix that by giving them non-null default values. However, that
3491 : * would completely break comparability of pgbench results with prior
3492 : * versions. Since pgbench has never pretended to be fully TPC-B compliant
3493 : * anyway, we stick with the historical behavior.
3494 : */
3495 : struct ddlinfo
3496 : {
3497 : const char *table; /* table name */
3498 : const char *smcols; /* column decls if accountIDs are 32 bits */
3499 : const char *bigcols; /* column decls if accountIDs are 64 bits */
3500 : int declare_fillfactor;
3501 : };
3502 : static const struct ddlinfo DDLs[] = {
3503 : {
3504 : "pgbench_history",
3505 : "tid int,bid int,aid int,delta int,mtime timestamp,filler char(22)",
3506 : "tid int,bid int,aid bigint,delta int,mtime timestamp,filler char(22)",
3507 : 0
3508 : },
3509 : {
3510 : "pgbench_tellers",
3511 : "tid int not null,bid int,tbalance int,filler char(84)",
3512 : "tid int not null,bid int,tbalance int,filler char(84)",
3513 : 1
3514 : },
3515 : {
3516 : "pgbench_accounts",
3517 : "aid int not null,bid int,abalance int,filler char(84)",
3518 : "aid bigint not null,bid int,abalance int,filler char(84)",
3519 : 1
3520 : },
3521 : {
3522 : "pgbench_branches",
3523 : "bid int not null,bbalance int,filler char(88)",
3524 : "bid int not null,bbalance int,filler char(88)",
3525 : 1
3526 : }
3527 : };
3528 : int i;
3529 :
3530 6 : fprintf(stderr, "creating tables...\n");
3531 :
3532 30 : for (i = 0; i < lengthof(DDLs); i++)
3533 : {
3534 : char opts[256];
3535 : char buffer[256];
3536 24 : const struct ddlinfo *ddl = &DDLs[i];
3537 : const char *cols;
3538 :
3539 : /* Construct new create table statement. */
3540 24 : opts[0] = '\0';
3541 24 : if (ddl->declare_fillfactor)
3542 18 : snprintf(opts + strlen(opts), sizeof(opts) - strlen(opts),
3543 : " with (fillfactor=%d)", fillfactor);
3544 24 : if (tablespace != NULL)
3545 : {
3546 : char *escape_tablespace;
3547 :
3548 8 : escape_tablespace = PQescapeIdentifier(con, tablespace,
3549 : strlen(tablespace));
3550 8 : snprintf(opts + strlen(opts), sizeof(opts) - strlen(opts),
3551 : " tablespace %s", escape_tablespace);
3552 8 : PQfreemem(escape_tablespace);
3553 : }
3554 :
3555 24 : cols = (scale >= SCALE_32BIT_THRESHOLD) ? ddl->bigcols : ddl->smcols;
3556 :
3557 48 : snprintf(buffer, sizeof(buffer), "create%s table %s(%s)%s",
3558 24 : unlogged_tables ? " unlogged" : "",
3559 : ddl->table, cols, opts);
3560 :
3561 24 : executeStatement(con, buffer);
3562 : }
3563 6 : }
3564 :
3565 : /*
3566 : * Fill the standard tables with some data
3567 : */
3568 : static void
3569 6 : initGenerateData(PGconn *con)
3570 : {
3571 : char sql[256];
3572 : PGresult *res;
3573 : int i;
3574 : int64 k;
3575 :
3576 : /* used to track elapsed time and estimate of the remaining time */
3577 : instr_time start,
3578 : diff;
3579 : double elapsed_sec,
3580 : remaining_sec;
3581 6 : int log_interval = 1;
3582 :
3583 6 : fprintf(stderr, "generating data...\n");
3584 :
3585 : /*
3586 : * we do all of this in one transaction to enable the backend's
3587 : * data-loading optimizations
3588 : */
3589 6 : executeStatement(con, "begin");
3590 :
3591 : /*
3592 : * truncate away any old data, in one command in case there are foreign
3593 : * keys
3594 : */
3595 6 : executeStatement(con, "truncate table "
3596 : "pgbench_accounts, "
3597 : "pgbench_branches, "
3598 : "pgbench_history, "
3599 : "pgbench_tellers");
3600 :
3601 : /*
3602 : * fill branches, tellers, accounts in that order in case foreign keys
3603 : * already exist
3604 : */
3605 12 : for (i = 0; i < nbranches * scale; i++)
3606 : {
3607 : /* "filler" column defaults to NULL */
3608 6 : snprintf(sql, sizeof(sql),
3609 : "insert into pgbench_branches(bid,bbalance) values(%d,0)",
3610 : i + 1);
3611 6 : executeStatement(con, sql);
3612 : }
3613 :
3614 66 : for (i = 0; i < ntellers * scale; i++)
3615 : {
3616 : /* "filler" column defaults to NULL */
3617 60 : snprintf(sql, sizeof(sql),
3618 : "insert into pgbench_tellers(tid,bid,tbalance) values (%d,%d,0)",
3619 60 : i + 1, i / ntellers + 1);
3620 60 : executeStatement(con, sql);
3621 : }
3622 :
3623 : /*
3624 : * accounts is big enough to be worth using COPY and tracking runtime
3625 : */
3626 6 : res = PQexec(con, "copy pgbench_accounts from stdin");
3627 6 : if (PQresultStatus(res) != PGRES_COPY_IN)
3628 : {
3629 0 : fprintf(stderr, "%s", PQerrorMessage(con));
3630 0 : exit(1);
3631 : }
3632 6 : PQclear(res);
3633 :
3634 6 : INSTR_TIME_SET_CURRENT(start);
3635 :
3636 600006 : for (k = 0; k < (int64) naccounts * scale; k++)
3637 : {
3638 600000 : int64 j = k + 1;
3639 :
3640 : /* "filler" column defaults to blank padded empty string */
3641 600000 : snprintf(sql, sizeof(sql),
3642 : INT64_FORMAT "\t" INT64_FORMAT "\t%d\t\n",
3643 600000 : j, k / naccounts + 1, 0);
3644 600000 : if (PQputline(con, sql))
3645 : {
3646 0 : fprintf(stderr, "PQputline failed\n");
3647 0 : exit(1);
3648 : }
3649 :
3650 : /*
3651 : * If we want to stick with the original logging, print a message each
3652 : * 100k inserted rows.
3653 : */
3654 600000 : if ((!use_quiet) && (j % 100000 == 0))
3655 : {
3656 4 : INSTR_TIME_SET_CURRENT(diff);
3657 4 : INSTR_TIME_SUBTRACT(diff, start);
3658 :
3659 4 : elapsed_sec = INSTR_TIME_GET_DOUBLE(diff);
3660 4 : remaining_sec = ((double) scale * naccounts - j) * elapsed_sec / j;
3661 :
3662 4 : fprintf(stderr, INT64_FORMAT " of " INT64_FORMAT " tuples (%d%%) done (elapsed %.2f s, remaining %.2f s)\n",
3663 : j, (int64) naccounts * scale,
3664 4 : (int) (((int64) j * 100) / (naccounts * (int64) scale)),
3665 : elapsed_sec, remaining_sec);
3666 : }
3667 : /* let's not call the timing for each row, but only each 100 rows */
3668 599996 : else if (use_quiet && (j % 100 == 0))
3669 : {
3670 2000 : INSTR_TIME_SET_CURRENT(diff);
3671 2000 : INSTR_TIME_SUBTRACT(diff, start);
3672 :
3673 2000 : elapsed_sec = INSTR_TIME_GET_DOUBLE(diff);
3674 2000 : remaining_sec = ((double) scale * naccounts - j) * elapsed_sec / j;
3675 :
3676 : /* have we reached the next interval (or end)? */
3677 2000 : if ((j == scale * naccounts) || (elapsed_sec >= log_interval * LOG_STEP_SECONDS))
3678 : {
3679 2 : fprintf(stderr, INT64_FORMAT " of " INT64_FORMAT " tuples (%d%%) done (elapsed %.2f s, remaining %.2f s)\n",
3680 : j, (int64) naccounts * scale,
3681 2 : (int) (((int64) j * 100) / (naccounts * (int64) scale)), elapsed_sec, remaining_sec);
3682 :
3683 : /* skip to the next interval */
3684 2 : log_interval = (int) ceil(elapsed_sec / LOG_STEP_SECONDS);
3685 : }
3686 : }
3687 :
3688 : }
3689 6 : if (PQputline(con, "\\.\n"))
3690 : {
3691 0 : fprintf(stderr, "very last PQputline failed\n");
3692 0 : exit(1);
3693 : }
3694 6 : if (PQendcopy(con))
3695 : {
3696 0 : fprintf(stderr, "PQendcopy failed\n");
3697 0 : exit(1);
3698 : }
3699 :
3700 6 : executeStatement(con, "commit");
3701 6 : }
3702 :
3703 : /*
3704 : * Invoke vacuum on the standard tables
3705 : */
3706 : static void
3707 4 : initVacuum(PGconn *con)
3708 : {
3709 4 : fprintf(stderr, "vacuuming...\n");
3710 4 : executeStatement(con, "vacuum analyze pgbench_branches");
3711 4 : executeStatement(con, "vacuum analyze pgbench_tellers");
3712 4 : executeStatement(con, "vacuum analyze pgbench_accounts");
3713 4 : executeStatement(con, "vacuum analyze pgbench_history");
3714 4 : }
3715 :
3716 : /*
3717 : * Create primary keys on the standard tables
3718 : */
3719 : static void
3720 6 : initCreatePKeys(PGconn *con)
3721 : {
3722 : static const char *const DDLINDEXes[] = {
3723 : "alter table pgbench_branches add primary key (bid)",
3724 : "alter table pgbench_tellers add primary key (tid)",
3725 : "alter table pgbench_accounts add primary key (aid)"
3726 : };
3727 : int i;
3728 :
3729 6 : fprintf(stderr, "creating primary keys...\n");
3730 24 : for (i = 0; i < lengthof(DDLINDEXes); i++)
3731 : {
3732 : char buffer[256];
3733 :
3734 18 : strlcpy(buffer, DDLINDEXes[i], sizeof(buffer));
3735 :
3736 18 : if (index_tablespace != NULL)
3737 : {
3738 : char *escape_tablespace;
3739 :
3740 6 : escape_tablespace = PQescapeIdentifier(con, index_tablespace,
3741 : strlen(index_tablespace));
3742 6 : snprintf(buffer + strlen(buffer), sizeof(buffer) - strlen(buffer),
3743 : " using index tablespace %s", escape_tablespace);
3744 6 : PQfreemem(escape_tablespace);
3745 : }
3746 :
3747 18 : executeStatement(con, buffer);
3748 : }
3749 6 : }
3750 :
3751 : /*
3752 : * Create foreign key constraints between the standard tables
3753 : */
3754 : static void
3755 4 : initCreateFKeys(PGconn *con)
3756 : {
3757 : static const char *const DDLKEYs[] = {
3758 : "alter table pgbench_tellers add constraint pgbench_tellers_bid_fkey foreign key (bid) references pgbench_branches",
3759 : "alter table pgbench_accounts add constraint pgbench_accounts_bid_fkey foreign key (bid) references pgbench_branches",
3760 : "alter table pgbench_history add constraint pgbench_history_bid_fkey foreign key (bid) references pgbench_branches",
3761 : "alter table pgbench_history add constraint pgbench_history_tid_fkey foreign key (tid) references pgbench_tellers",
3762 : "alter table pgbench_history add constraint pgbench_history_aid_fkey foreign key (aid) references pgbench_accounts"
3763 : };
3764 : int i;
3765 :
3766 4 : fprintf(stderr, "creating foreign keys...\n");
3767 24 : for (i = 0; i < lengthof(DDLKEYs); i++)
3768 : {
3769 20 : executeStatement(con, DDLKEYs[i]);
3770 : }
3771 4 : }
3772 :
3773 : /*
3774 : * Validate an initialization-steps string
3775 : *
3776 : * (We could just leave it to runInitSteps() to fail if there are wrong
3777 : * characters, but since initialization can take awhile, it seems friendlier
3778 : * to check during option parsing.)
3779 : */
3780 : static void
3781 8 : checkInitSteps(const char *initialize_steps)
3782 : {
3783 : const char *step;
3784 :
3785 8 : if (initialize_steps[0] == '\0')
3786 : {
3787 0 : fprintf(stderr, "no initialization steps specified\n");
3788 0 : exit(1);
3789 : }
3790 :
3791 42 : for (step = initialize_steps; *step != '\0'; step++)
3792 : {
3793 36 : if (strchr("dtgvpf ", *step) == NULL)
3794 : {
3795 2 : fprintf(stderr, "unrecognized initialization step \"%c\"\n",
3796 2 : *step);
3797 2 : fprintf(stderr, "allowed steps are: \"d\", \"t\", \"g\", \"v\", \"p\", \"f\"\n");
3798 2 : exit(1);
3799 : }
3800 : }
3801 6 : }
3802 :
3803 : /*
3804 : * Invoke each initialization step in the given string
3805 : */
3806 : static void
3807 6 : runInitSteps(const char *initialize_steps)
3808 : {
3809 : PGconn *con;
3810 : const char *step;
3811 :
3812 6 : if ((con = doConnect()) == NULL)
3813 0 : exit(1);
3814 :
3815 44 : for (step = initialize_steps; *step != '\0'; step++)
3816 : {
3817 38 : switch (*step)
3818 : {
3819 : case 'd':
3820 6 : initDropTables(con);
3821 6 : break;
3822 : case 't':
3823 6 : initCreateTables(con);
3824 6 : break;
3825 : case 'g':
3826 6 : initGenerateData(con);
3827 6 : break;
3828 : case 'v':
3829 4 : initVacuum(con);
3830 4 : break;
3831 : case 'p':
3832 6 : initCreatePKeys(con);
3833 6 : break;
3834 : case 'f':
3835 4 : initCreateFKeys(con);
3836 4 : break;
3837 : case ' ':
3838 6 : break; /* ignore */
3839 : default:
3840 0 : fprintf(stderr, "unrecognized initialization step \"%c\"\n",
3841 0 : *step);
3842 0 : PQfinish(con);
3843 0 : exit(1);
3844 : }
3845 : }
3846 :
3847 6 : fprintf(stderr, "done.\n");
3848 6 : PQfinish(con);
3849 6 : }
3850 :
3851 : /*
3852 : * Replace :param with $n throughout the command's SQL text, which
3853 : * is a modifiable string in cmd->argv[0].
3854 : */
3855 : static bool
3856 32 : parseQuery(Command *cmd)
3857 : {
3858 : char *sql,
3859 : *p;
3860 :
3861 : /* We don't want to scribble on cmd->argv[0] until done */
3862 32 : sql = pg_strdup(cmd->argv[0]);
3863 :
3864 32 : cmd->argc = 1;
3865 :
3866 32 : p = sql;
3867 118 : while ((p = strchr(p, ':')) != NULL)
3868 : {
3869 : char var[13];
3870 : char *name;
3871 : int eaten;
3872 :
3873 56 : name = parseVariable(p, &eaten);
3874 56 : if (name == NULL)
3875 : {
3876 40 : while (*p == ':')
3877 : {
3878 20 : p++;
3879 : }
3880 10 : continue;
3881 : }
3882 :
3883 46 : if (cmd->argc >= MAX_ARGS)
3884 : {
3885 2 : fprintf(stderr, "statement has too many arguments (maximum is %d): %s\n",
3886 : MAX_ARGS - 1, cmd->argv[0]);
3887 2 : pg_free(name);
3888 2 : return false;
3889 : }
3890 :
3891 44 : sprintf(var, "$%d", cmd->argc);
3892 44 : p = replaceVariable(&sql, p, eaten, var);
3893 :
3894 44 : cmd->argv[cmd->argc] = name;
3895 44 : cmd->argc++;
3896 : }
3897 :
3898 30 : pg_free(cmd->argv[0]);
3899 30 : cmd->argv[0] = sql;
3900 30 : return true;
3901 : }
3902 :
3903 : /*
3904 : * Simple error-printing function, might be needed by lexer
3905 : */
3906 : static void
3907 0 : pgbench_error(const char *fmt,...)
3908 : {
3909 : va_list ap;
3910 :
3911 0 : fflush(stdout);
3912 0 : va_start(ap, fmt);
3913 0 : vfprintf(stderr, _(fmt), ap);
3914 0 : va_end(ap);
3915 0 : }
3916 :
3917 : /*
3918 : * syntax error while parsing a script (in practice, while parsing a
3919 : * backslash command, because we don't detect syntax errors in SQL)
3920 : *
3921 : * source: source of script (filename or builtin-script ID)
3922 : * lineno: line number within script (count from 1)
3923 : * line: whole line of backslash command, if available
3924 : * command: backslash command name, if available
3925 : * msg: the actual error message
3926 : * more: optional extra message
3927 : * column: zero-based column number, or -1 if unknown
3928 : */
3929 : void
3930 38 : syntax_error(const char *source, int lineno,
3931 : const char *line, const char *command,
3932 : const char *msg, const char *more, int column)
3933 : {
3934 38 : fprintf(stderr, "%s:%d: %s", source, lineno, msg);
3935 38 : if (more != NULL)
3936 10 : fprintf(stderr, " (%s)", more);
3937 38 : if (column >= 0 && line == NULL)
3938 0 : fprintf(stderr, " at column %d", column + 1);
3939 38 : if (command != NULL)
3940 38 : fprintf(stderr, " in command \"%s\"", command);
3941 38 : fprintf(stderr, "\n");
3942 38 : if (line != NULL)
3943 : {
3944 34 : fprintf(stderr, "%s\n", line);
3945 34 : if (column >= 0)
3946 : {
3947 : int i;
3948 :
3949 272 : for (i = 0; i < column; i++)
3950 250 : fprintf(stderr, " ");
3951 22 : fprintf(stderr, "^ error found here\n");
3952 : }
3953 : }
3954 38 : exit(1);
3955 : }
3956 :
3957 : /*
3958 : * Parse a SQL command; return a Command struct, or NULL if it's a comment
3959 : *
3960 : * On entry, psqlscan.l has collected the command into "buf", so we don't
3961 : * really need to do much here except check for comment and set up a
3962 : * Command struct.
3963 : */
3964 : static Command *
3965 1674 : process_sql_command(PQExpBuffer buf, const char *source)
3966 : {
3967 : Command *my_command;
3968 : char *p;
3969 : char *nlpos;
3970 :
3971 : /* Skip any leading whitespace, as well as "--" style comments */
3972 1674 : p = buf->data;
3973 : for (;;)
3974 : {
3975 1674 : if (isspace((unsigned char) *p))
3976 0 : p++;
3977 1674 : else if (strncmp(p, "--", 2) == 0)
3978 : {
3979 0 : p = strchr(p, '\n');
3980 0 : if (p == NULL)
3981 0 : return NULL;
3982 0 : p++;
3983 : }
3984 : else
3985 1674 : break;
3986 : }
3987 :
3988 : /* If there's nothing but whitespace and comments, we're done */
3989 1674 : if (*p == '\0')
3990 1170 : return NULL;
3991 :
3992 : /* Allocate and initialize Command structure */
3993 504 : my_command = (Command *) pg_malloc0(sizeof(Command));
3994 504 : my_command->command_num = num_commands++;
3995 504 : my_command->type = SQL_COMMAND;
3996 504 : my_command->meta = META_NONE;
3997 504 : initSimpleStats(&my_command->stats);
3998 :
3999 : /*
4000 : * Install query text as the sole argv string. If we are using a
4001 : * non-simple query mode, we'll extract parameters from it later.
4002 : */
4003 504 : my_command->argv[0] = pg_strdup(p);
4004 504 : my_command->argc = 1;
4005 :
4006 : /*
4007 : * If SQL command is multi-line, we only want to save the first line as
4008 : * the "line" label.
4009 : */
4010 504 : nlpos = strchr(p, '\n');
4011 504 : if (nlpos)
4012 : {
4013 10 : my_command->line = pg_malloc(nlpos - p + 1);
4014 10 : memcpy(my_command->line, p, nlpos - p);
4015 10 : my_command->line[nlpos - p] = '\0';
4016 : }
4017 : else
4018 494 : my_command->line = pg_strdup(p);
4019 :
4020 504 : return my_command;
4021 : }
4022 :
4023 : /*
4024 : * Parse a backslash command; return a Command struct, or NULL if comment
4025 : *
4026 : * At call, we have scanned only the initial backslash.
4027 : */
4028 : static Command *
4029 780 : process_backslash_command(PsqlScanState sstate, const char *source)
4030 : {
4031 : Command *my_command;
4032 : PQExpBufferData word_buf;
4033 : int word_offset;
4034 : int offsets[MAX_ARGS]; /* offsets of argument words */
4035 : int start_offset;
4036 : int lineno;
4037 : int j;
4038 :
4039 780 : initPQExpBuffer(&word_buf);
4040 :
4041 : /* Remember location of the backslash */
4042 780 : start_offset = expr_scanner_offset(sstate) - 1;
4043 780 : lineno = expr_scanner_get_lineno(sstate, start_offset);
4044 :
4045 : /* Collect first word of command */
4046 780 : if (!expr_lex_one_word(sstate, &word_buf, &word_offset))
4047 : {
4048 0 : termPQExpBuffer(&word_buf);
4049 0 : return NULL;
4050 : }
4051 :
4052 : /* Allocate and initialize Command structure */
4053 780 : my_command = (Command *) pg_malloc0(sizeof(Command));
4054 780 : my_command->command_num = num_commands++;
4055 780 : my_command->type = META_COMMAND;
4056 780 : my_command->argc = 0;
4057 780 : initSimpleStats(&my_command->stats);
4058 :
4059 : /* Save first word (command name) */
4060 780 : j = 0;
4061 780 : offsets[j] = word_offset;
4062 780 : my_command->argv[j++] = pg_strdup(word_buf.data);
4063 780 : my_command->argc++;
4064 :
4065 : /* ... and convert it to enum form */
4066 780 : my_command->meta = getMetaCommand(my_command->argv[0]);
4067 :
4068 892 : if (my_command->meta == META_SET ||
4069 200 : my_command->meta == META_IF ||
4070 88 : my_command->meta == META_ELIF)
4071 : {
4072 : yyscan_t yyscanner;
4073 :
4074 : /* For \set, collect var name */
4075 708 : if (my_command->meta == META_SET)
4076 : {
4077 668 : if (!expr_lex_one_word(sstate, &word_buf, &word_offset))
4078 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4079 : "missing argument", NULL, -1);
4080 :
4081 666 : offsets[j] = word_offset;
4082 666 : my_command->argv[j++] = pg_strdup(word_buf.data);
4083 666 : my_command->argc++;
4084 : }
4085 :
4086 : /* then for all parse the expression */
4087 706 : yyscanner = expr_scanner_init(sstate, source, lineno, start_offset,
4088 706 : my_command->argv[0]);
4089 :
4090 706 : if (expr_yyparse(yyscanner) != 0)
4091 : {
4092 : /* dead code: exit done from syntax_error called by yyerror */
4093 0 : exit(1);
4094 : }
4095 :
4096 688 : my_command->expr = expr_parse_result;
4097 :
4098 : /* Save line, trimming any trailing newline */
4099 688 : my_command->line = expr_scanner_get_substring(sstate,
4100 : start_offset,
4101 : expr_scanner_offset(sstate),
4102 : true);
4103 :
4104 688 : expr_scanner_finish(yyscanner);
4105 :
4106 688 : termPQExpBuffer(&word_buf);
4107 :
4108 688 : return my_command;
4109 : }
4110 :
4111 : /* For all other commands, collect remaining words. */
4112 230 : while (expr_lex_one_word(sstate, &word_buf, &word_offset))
4113 : {
4114 88 : if (j >= MAX_ARGS)
4115 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4116 : "too many arguments", NULL, -1);
4117 :
4118 86 : offsets[j] = word_offset;
4119 86 : my_command->argv[j++] = pg_strdup(word_buf.data);
4120 86 : my_command->argc++;
4121 : }
4122 :
4123 : /* Save line, trimming any trailing newline */
4124 70 : my_command->line = expr_scanner_get_substring(sstate,
4125 : start_offset,
4126 : expr_scanner_offset(sstate),
4127 : true);
4128 :
4129 70 : if (my_command->meta == META_SLEEP)
4130 : {
4131 18 : if (my_command->argc < 2)
4132 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4133 : "missing argument", NULL, -1);
4134 :
4135 16 : if (my_command->argc > 3)
4136 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4137 : "too many arguments", NULL,
4138 2 : offsets[3] - start_offset);
4139 :
4140 : /*
4141 : * Split argument into number and unit to allow "sleep 1ms" etc. We
4142 : * don't have to terminate the number argument with null because it
4143 : * will be parsed with atoi, which ignores trailing non-digit
4144 : * characters.
4145 : */
4146 14 : if (my_command->argc == 2 && my_command->argv[1][0] != ':')
4147 : {
4148 2 : char *c = my_command->argv[1];
4149 :
4150 6 : while (isdigit((unsigned char) *c))
4151 2 : c++;
4152 2 : if (*c)
4153 : {
4154 2 : my_command->argv[2] = c;
4155 2 : offsets[2] = offsets[1] + (c - my_command->argv[1]);
4156 2 : my_command->argc = 3;
4157 : }
4158 : }
4159 :
4160 14 : if (my_command->argc == 3)
4161 : {
4162 18 : if (pg_strcasecmp(my_command->argv[2], "us") != 0 &&
4163 12 : pg_strcasecmp(my_command->argv[2], "ms") != 0 &&
4164 4 : pg_strcasecmp(my_command->argv[2], "s") != 0)
4165 4 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4166 : "unrecognized time unit, must be us, ms or s",
4167 4 : my_command->argv[2], offsets[2] - start_offset);
4168 : }
4169 : }
4170 52 : else if (my_command->meta == META_SETSHELL)
4171 : {
4172 8 : if (my_command->argc < 3)
4173 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4174 : "missing argument", NULL, -1);
4175 : }
4176 44 : else if (my_command->meta == META_SHELL)
4177 : {
4178 8 : if (my_command->argc < 2)
4179 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4180 : "missing command", NULL, -1);
4181 : }
4182 36 : else if (my_command->meta == META_ELSE || my_command->meta == META_ENDIF)
4183 : {
4184 64 : if (my_command->argc != 1)
4185 4 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4186 : "unexpected argument", NULL, -1);
4187 : }
4188 : else
4189 : {
4190 : /* my_command->meta == META_NONE */
4191 2 : syntax_error(source, lineno, my_command->line, my_command->argv[0],
4192 : "invalid command", NULL, -1);
4193 : }
4194 :
4195 54 : termPQExpBuffer(&word_buf);
4196 :
4197 54 : return my_command;
4198 : }
4199 :
4200 : static void
4201 12 : ConditionError(const char *desc, int cmdn, const char *msg)
4202 : {
4203 12 : fprintf(stderr,
4204 : "condition error in script \"%s\" command %d: %s\n",
4205 : desc, cmdn, msg);
4206 12 : exit(1);
4207 : }
4208 :
4209 : /*
4210 : * Partial evaluation of conditionals before recording and running the script.
4211 : */
4212 : static void
4213 386 : CheckConditional(ParsedScript ps)
4214 : {
4215 : /* statically check conditional structure */
4216 386 : ConditionalStack cs = conditional_stack_create();
4217 : int i;
4218 :
4219 1608 : for (i = 0; ps.commands[i] != NULL; i++)
4220 : {
4221 1232 : Command *cmd = ps.commands[i];
4222 :
4223 1232 : if (cmd->type == META_COMMAND)
4224 : {
4225 730 : switch (cmd->meta)
4226 : {
4227 : case META_IF:
4228 16 : conditional_stack_push(cs, IFSTATE_FALSE);
4229 16 : break;
4230 : case META_ELIF:
4231 14 : if (conditional_stack_empty(cs))
4232 2 : ConditionError(ps.desc, i + 1, "\\elif without matching \\if");
4233 12 : if (conditional_stack_peek(cs) == IFSTATE_ELSE_FALSE)
4234 2 : ConditionError(ps.desc, i + 1, "\\elif after \\else");
4235 10 : break;
4236 : case META_ELSE:
4237 14 : if (conditional_stack_empty(cs))
4238 2 : ConditionError(ps.desc, i + 1, "\\else without matching \\if");
4239 12 : if (conditional_stack_peek(cs) == IFSTATE_ELSE_FALSE)
4240 2 : ConditionError(ps.desc, i + 1, "\\else after \\else");
4241 10 : conditional_stack_poke(cs, IFSTATE_ELSE_FALSE);
4242 10 : break;
4243 : case META_ENDIF:
4244 12 : if (!conditional_stack_pop(cs))
4245 2 : ConditionError(ps.desc, i + 1, "\\endif without matching \\if");
4246 10 : break;
4247 : default:
4248 : /* ignore anything else... */
4249 674 : break;
4250 : }
4251 : }
4252 : }
4253 376 : if (!conditional_stack_empty(cs))
4254 2 : ConditionError(ps.desc, i + 1, "\\if without matching \\endif");
4255 374 : conditional_stack_destroy(cs);
4256 374 : }
4257 :
4258 : /*
4259 : * Parse a script (either the contents of a file, or a built-in script)
4260 : * and add it to the list of scripts.
4261 : */
4262 : static void
4263 428 : ParseScript(const char *script, const char *desc, int weight)
4264 : {
4265 : ParsedScript ps;
4266 : PsqlScanState sstate;
4267 : PQExpBufferData line_buf;
4268 : int alloc_num;
4269 : int index;
4270 :
4271 : #define COMMANDS_ALLOC_NUM 128
4272 428 : alloc_num = COMMANDS_ALLOC_NUM;
4273 :
4274 : /* Initialize all fields of ps */
4275 428 : ps.desc = desc;
4276 428 : ps.weight = weight;
4277 428 : ps.commands = (Command **) pg_malloc(sizeof(Command *) * alloc_num);
4278 428 : initStats(&ps.stats, 0);
4279 :
4280 : /* Prepare to parse script */
4281 428 : sstate = psql_scan_create(&pgbench_callbacks);
4282 :
4283 : /*
4284 : * Ideally, we'd scan scripts using the encoding and stdstrings settings
4285 : * we get from a DB connection. However, without major rearrangement of
4286 : * pgbench's argument parsing, we can't have a DB connection at the time
4287 : * we parse scripts. Using SQL_ASCII (encoding 0) should work well enough
4288 : * with any backend-safe encoding, though conceivably we could be fooled
4289 : * if a script file uses a client-only encoding. We also assume that
4290 : * stdstrings should be true, which is a bit riskier.
4291 : */
4292 428 : psql_scan_setup(sstate, script, strlen(script), 0, true);
4293 :
4294 428 : initPQExpBuffer(&line_buf);
4295 :
4296 428 : index = 0;
4297 :
4298 : for (;;)
4299 1246 : {
4300 : PsqlScanResult sr;
4301 : promptStatus_t prompt;
4302 : Command *command;
4303 :
4304 1674 : resetPQExpBuffer(&line_buf);
4305 :
4306 1674 : sr = psql_scan(sstate, &line_buf, &prompt);
4307 :
4308 : /* If we collected a SQL command, process that */
4309 1674 : command = process_sql_command(&line_buf, desc);
4310 1674 : if (command)
4311 : {
4312 504 : ps.commands[index] = command;
4313 504 : index++;
4314 :
4315 504 : if (index >= alloc_num)
4316 : {
4317 0 : alloc_num += COMMANDS_ALLOC_NUM;
4318 0 : ps.commands = (Command **)
4319 0 : pg_realloc(ps.commands, sizeof(Command *) * alloc_num);
4320 : }
4321 : }
4322 :
4323 : /* If we reached a backslash, process that */
4324 1674 : if (sr == PSCAN_BACKSLASH)
4325 : {
4326 780 : command = process_backslash_command(sstate, desc);
4327 742 : if (command)
4328 : {
4329 742 : ps.commands[index] = command;
4330 742 : index++;
4331 :
4332 742 : if (index >= alloc_num)
4333 : {
4334 0 : alloc_num += COMMANDS_ALLOC_NUM;
4335 0 : ps.commands = (Command **)
4336 0 : pg_realloc(ps.commands, sizeof(Command *) * alloc_num);
4337 : }
4338 : }
4339 : }
4340 :
4341 : /* Done if we reached EOF */
4342 1636 : if (sr == PSCAN_INCOMPLETE || sr == PSCAN_EOL)
4343 : break;
4344 : }
4345 :
4346 390 : ps.commands[index] = NULL;
4347 :
4348 390 : addScript(ps);
4349 :
4350 374 : termPQExpBuffer(&line_buf);
4351 374 : psql_scan_finish(sstate);
4352 374 : psql_scan_destroy(sstate);
4353 374 : }
4354 :
4355 : /*
4356 : * Read the entire contents of file fd, and return it in a malloc'd buffer.
4357 : *
4358 : * The buffer will typically be larger than necessary, but we don't care
4359 : * in this program, because we'll free it as soon as we've parsed the script.
4360 : */
4361 : static char *
4362 124 : read_file_contents(FILE *fd)
4363 : {
4364 : char *buf;
4365 124 : size_t buflen = BUFSIZ;
4366 124 : size_t used = 0;
4367 :
4368 124 : buf = (char *) pg_malloc(buflen);
4369 :
4370 : for (;;)
4371 0 : {
4372 : size_t nread;
4373 :
4374 124 : nread = fread(buf + used, 1, BUFSIZ, fd);
4375 124 : used += nread;
4376 : /* If fread() read less than requested, must be EOF or error */
4377 124 : if (nread < BUFSIZ)
4378 124 : break;
4379 : /* Enlarge buf so we can read some more */
4380 0 : buflen += BUFSIZ;
4381 0 : buf = (char *) pg_realloc(buf, buflen);
4382 : }
4383 : /* There is surely room for a terminator */
4384 124 : buf[used] = '\0';
4385 :
4386 124 : return buf;
4387 : }
4388 :
4389 : /*
4390 : * Given a file name, read it and add its script to the list.
4391 : * "-" means to read stdin.
4392 : * NB: filename must be storage that won't disappear.
4393 : */
4394 : static void
4395 126 : process_file(const char *filename, int weight)
4396 : {
4397 : FILE *fd;
4398 : char *buf;
4399 :
4400 : /* Slurp the file contents into "buf" */
4401 126 : if (strcmp(filename, "-") == 0)
4402 0 : fd = stdin;
4403 126 : else if ((fd = fopen(filename, "r")) == NULL)
4404 : {
4405 2 : fprintf(stderr, "could not open file \"%s\": %s\n",
4406 2 : filename, strerror(errno));
4407 2 : exit(1);
4408 : }
4409 :
4410 124 : buf = read_file_contents(fd);
4411 :
4412 124 : if (ferror(fd))
4413 : {
4414 0 : fprintf(stderr, "could not read file \"%s\": %s\n",
4415 0 : filename, strerror(errno));
4416 0 : exit(1);
4417 : }
4418 :
4419 124 : if (fd != stdin)
4420 124 : fclose(fd);
4421 :
4422 124 : ParseScript(buf, filename, weight);
4423 :
4424 72 : free(buf);
4425 72 : }
4426 :
4427 : /* Parse the given builtin script and add it to the list. */
4428 : static void
4429 304 : process_builtin(const BuiltinScript *bi, int weight)
4430 : {
4431 304 : ParseScript(bi->script, bi->desc, weight);
4432 302 : }
4433 :
4434 : /* show available builtin scripts */
4435 : static void
4436 6 : listAvailableScripts(void)
4437 : {
4438 : int i;
4439 :
4440 6 : fprintf(stderr, "Available builtin scripts:\n");
4441 24 : for (i = 0; i < lengthof(builtin_script); i++)
4442 18 : fprintf(stderr, "\t%s\n", builtin_script[i].name);
4443 6 : fprintf(stderr, "\n");
4444 6 : }
4445 :
4446 : /* return builtin script "name" if unambiguous, fails if not found */
4447 : static const BuiltinScript *
4448 308 : findBuiltin(const char *name)
4449 : {
4450 : int i,
4451 308 : found = 0,
4452 308 : len = strlen(name);
4453 308 : const BuiltinScript *result = NULL;
4454 :
4455 1232 : for (i = 0; i < lengthof(builtin_script); i++)
4456 : {
4457 924 : if (strncmp(builtin_script[i].name, name, len) == 0)
4458 : {
4459 308 : result = &builtin_script[i];
4460 308 : found++;
4461 : }
4462 : }
4463 :
4464 : /* ok, unambiguous result */
4465 308 : if (found == 1)
4466 608 : return result;
4467 :
4468 : /* error cases */
4469 4 : if (found == 0)
4470 2 : fprintf(stderr, "no builtin script found for name \"%s\"\n", name);
4471 : else /* found > 1 */
4472 2 : fprintf(stderr,
4473 : "ambiguous builtin name: %d builtin scripts found for prefix \"%s\"\n", found, name);
4474 :
4475 4 : listAvailableScripts();
4476 4 : exit(1);
4477 : }
4478 :
4479 : /*
4480 : * Determine the weight specification from a script option (-b, -f), if any,
4481 : * and return it as an integer (1 is returned if there's no weight). The
4482 : * script name is returned in *script as a malloc'd string.
4483 : */
4484 : static int
4485 148 : parseScriptWeight(const char *option, char **script)
4486 : {
4487 : char *sep;
4488 : int weight;
4489 :
4490 148 : if ((sep = strrchr(option, WSEP)))
4491 : {
4492 14 : int namelen = sep - option;
4493 : long wtmp;
4494 : char *badp;
4495 :
4496 : /* generate the script name */
4497 14 : *script = pg_malloc(namelen + 1);
4498 14 : strncpy(*script, option, namelen);
4499 14 : (*script)[namelen] = '\0';
4500 :
4501 : /* process digits of the weight spec */
4502 14 : errno = 0;
4503 14 : wtmp = strtol(sep + 1, &badp, 10);
4504 14 : if (errno != 0 || badp == sep + 1 || *badp != '\0')
4505 : {
4506 2 : fprintf(stderr, "invalid weight specification: %s\n", sep);
4507 2 : exit(1);
4508 : }
4509 12 : if (wtmp > INT_MAX || wtmp < 0)
4510 : {
4511 2 : fprintf(stderr,
4512 : "weight specification out of range (0 .. %u): " INT64_FORMAT "\n",
4513 : INT_MAX, (int64) wtmp);
4514 2 : exit(1);
4515 : }
4516 10 : weight = wtmp;
4517 : }
4518 : else
4519 : {
4520 134 : *script = pg_strdup(option);
4521 134 : weight = 1;
4522 : }
4523 :
4524 144 : return weight;
4525 : }
4526 :
4527 : /* append a script to the list of scripts to process */
4528 : static void
4529 390 : addScript(ParsedScript script)
4530 : {
4531 390 : if (script.commands == NULL || script.commands[0] == NULL)
4532 : {
4533 2 : fprintf(stderr, "empty command list for script \"%s\"\n", script.desc);
4534 2 : exit(1);
4535 : }
4536 :
4537 388 : if (num_scripts >= MAX_SCRIPTS)
4538 : {
4539 2 : fprintf(stderr, "at most %d SQL scripts are allowed\n", MAX_SCRIPTS);
4540 2 : exit(1);
4541 : }
4542 :
4543 386 : CheckConditional(script);
4544 :
4545 374 : sql_script[num_scripts] = script;
4546 374 : num_scripts++;
4547 374 : }
4548 :
4549 : static void
4550 8 : printSimpleStats(const char *prefix, SimpleStats *ss)
4551 : {
4552 8 : if (ss->count > 0)
4553 : {
4554 6 : double latency = ss->sum / ss->count;
4555 6 : double stddev = sqrt(ss->sum2 / ss->count - latency * latency);
4556 :
4557 6 : printf("%s average = %.3f ms\n", prefix, 0.001 * latency);
4558 6 : printf("%s stddev = %.3f ms\n", prefix, 0.001 * stddev);
4559 : }
4560 8 : }
4561 :
4562 : /* print out results */
4563 : static void
4564 80 : printResults(TState *threads, StatsData *total, instr_time total_time,
4565 : instr_time conn_total_time, int64 latency_late)
4566 : {
4567 : double time_include,
4568 : tps_include,
4569 : tps_exclude;
4570 80 : int64 ntx = total->cnt - total->skipped;
4571 : int i,
4572 80 : totalCacheOverflows = 0;
4573 :
4574 80 : time_include = INSTR_TIME_GET_DOUBLE(total_time);
4575 :
4576 : /* tps is about actually executed transactions */
4577 80 : tps_include = ntx / time_include;
4578 160 : tps_exclude = ntx /
4579 80 : (time_include - (INSTR_TIME_GET_DOUBLE(conn_total_time) / nclients));
4580 :
4581 : /* Report test parameters. */
4582 80 : printf("transaction type: %s\n",
4583 80 : num_scripts == 1 ? sql_script[0].desc : "multiple scripts");
4584 80 : printf("scaling factor: %d\n", scale);
4585 80 : printf("query mode: %s\n", QUERYMODE[querymode]);
4586 80 : printf("number of clients: %d\n", nclients);
4587 80 : printf("number of threads: %d\n", nthreads);
4588 80 : if (duration <= 0)
4589 : {
4590 80 : printf("number of transactions per client: %d\n", nxacts);
4591 80 : printf("number of transactions actually processed: " INT64_FORMAT "/%d\n",
4592 : ntx, nxacts * nclients);
4593 : }
4594 : else
4595 : {
4596 0 : printf("duration: %d s\n", duration);
4597 0 : printf("number of transactions actually processed: " INT64_FORMAT "\n",
4598 : ntx);
4599 : }
4600 : /* Report zipfian cache overflow */
4601 160 : for (i = 0; i < nthreads; i++)
4602 : {
4603 80 : totalCacheOverflows += threads[i].zipf_cache.overflowCount;
4604 : }
4605 80 : if (totalCacheOverflows > 0)
4606 : {
4607 2 : printf("zipfian cache array overflowed %d time(s)\n", totalCacheOverflows);
4608 : }
4609 :
4610 : /* Remaining stats are nonsensical if we failed to execute any xacts */
4611 80 : if (total->cnt <= 0)
4612 48 : return;
4613 :
4614 32 : if (throttle_delay && latency_limit)
4615 4 : printf("number of transactions skipped: " INT64_FORMAT " (%.3f %%)\n",
4616 : total->skipped,
4617 4 : 100.0 * total->skipped / total->cnt);
4618 :
4619 32 : if (latency_limit)
4620 6 : printf("number of transactions above the %.1f ms latency limit: " INT64_FORMAT "/" INT64_FORMAT " (%.3f %%)\n",
4621 : latency_limit / 1000.0, latency_late, ntx,
4622 2 : (ntx > 0) ? 100.0 * latency_late / ntx : 0.0);
4623 :
4624 32 : if (throttle_delay || progress || latency_limit)
4625 4 : printSimpleStats("latency", &total->latency);
4626 : else
4627 : {
4628 : /* no measurement, show average latency computed from run time */
4629 28 : printf("latency average = %.3f ms\n",
4630 28 : 1000.0 * time_include * nclients / total->cnt);
4631 : }
4632 :
4633 32 : if (throttle_delay)
4634 : {
4635 : /*
4636 : * Report average transaction lag under rate limit throttling. This
4637 : * is the delay between scheduled and actual start times for the
4638 : * transaction. The measured lag may be caused by thread/client load,
4639 : * the database load, or the Poisson throttling process.
4640 : */
4641 8 : printf("rate limit schedule lag: avg %.3f (max %.3f) ms\n",
4642 8 : 0.001 * total->lag.sum / total->cnt, 0.001 * total->lag.max);
4643 : }
4644 :
4645 32 : printf("tps = %f (including connections establishing)\n", tps_include);
4646 32 : printf("tps = %f (excluding connections establishing)\n", tps_exclude);
4647 :
4648 : /* Report per-script/command statistics */
4649 32 : if (per_script_stats || is_latencies)
4650 : {
4651 : int i;
4652 :
4653 14 : for (i = 0; i < num_scripts; i++)
4654 : {
4655 8 : if (per_script_stats)
4656 : {
4657 4 : StatsData *sstats = &sql_script[i].stats;
4658 :
4659 12 : printf("SQL script %d: %s\n"
4660 : " - weight: %d (targets %.1f%% of total)\n"
4661 : " - " INT64_FORMAT " transactions (%.1f%% of total, tps = %f)\n",
4662 : i + 1, sql_script[i].desc,
4663 : sql_script[i].weight,
4664 4 : 100.0 * sql_script[i].weight / total_weight,
4665 : sstats->cnt,
4666 4 : 100.0 * sstats->cnt / total->cnt,
4667 4 : (sstats->cnt - sstats->skipped) / time_include);
4668 :
4669 4 : if (throttle_delay && latency_limit && sstats->cnt > 0)
4670 0 : printf(" - number of transactions skipped: " INT64_FORMAT " (%.3f%%)\n",
4671 : sstats->skipped,
4672 0 : 100.0 * sstats->skipped / sstats->cnt);
4673 :
4674 4 : printSimpleStats(" - latency", &sstats->latency);
4675 : }
4676 :
4677 : /* Report per-command latencies */
4678 8 : if (is_latencies)
4679 : {
4680 : Command **commands;
4681 :
4682 4 : if (per_script_stats)
4683 0 : printf(" - statement latencies in milliseconds:\n");
4684 : else
4685 4 : printf("statement latencies in milliseconds:\n");
4686 :
4687 14 : for (commands = sql_script[i].commands;
4688 10 : *commands != NULL;
4689 6 : commands++)
4690 : {
4691 6 : SimpleStats *cstats = &(*commands)->stats;
4692 :
4693 16 : printf(" %11.3f %s\n",
4694 6 : (cstats->count > 0) ?
4695 4 : 1000.0 * cstats->sum / cstats->count : 0.0,
4696 6 : (*commands)->line);
4697 : }
4698 : }
4699 : }
4700 : }
4701 : }
4702 :
4703 : /* call srandom based on some seed. NULL triggers the default behavior. */
4704 : static bool
4705 226 : set_random_seed(const char *seed)
4706 : {
4707 : /* srandom expects an unsigned int */
4708 : unsigned int iseed;
4709 :
4710 226 : if (seed == NULL || strcmp(seed, "time") == 0)
4711 218 : {
4712 : /* rely on current time */
4713 : instr_time now;
4714 :
4715 218 : INSTR_TIME_SET_CURRENT(now);
4716 218 : iseed = (unsigned int) INSTR_TIME_GET_MICROSEC(now);
4717 : }
4718 8 : else if (strcmp(seed, "rand") == 0)
4719 : {
4720 : /* use some "strong" random source */
4721 : #ifdef HAVE_STRONG_RANDOM
4722 0 : if (!pg_strong_random(&iseed, sizeof(iseed)))
4723 : #endif
4724 : {
4725 0 : fprintf(stderr,
4726 : "cannot seed random from a strong source, none available: "
4727 : "use \"time\" or an unsigned integer value.\n");
4728 0 : return false;
4729 : }
4730 : }
4731 : else
4732 : {
4733 : /* parse seed unsigned int value */
4734 : char garbage;
4735 :
4736 8 : if (sscanf(seed, "%u%c", &iseed, &garbage) != 1)
4737 : {
4738 2 : fprintf(stderr,
4739 : "unrecognized random seed option \"%s\": expecting an unsigned integer, \"time\" or \"rand\"\n",
4740 : seed);
4741 2 : return false;
4742 : }
4743 : }
4744 :
4745 224 : if (seed != NULL)
4746 6 : fprintf(stderr, "setting random seed to %u\n", iseed);
4747 224 : srandom(iseed);
4748 : /* no precision loss: 32 bit unsigned int cast to 64 bit int */
4749 224 : random_seed = iseed;
4750 224 : return true;
4751 : }
4752 :
4753 :
4754 : int
4755 222 : main(int argc, char **argv)
4756 : {
4757 : static struct option long_options[] = {
4758 : /* systematic long/short named options */
4759 : {"builtin", required_argument, NULL, 'b'},
4760 : {"client", required_argument, NULL, 'c'},
4761 : {"connect", no_argument, NULL, 'C'},
4762 : {"debug", no_argument, NULL, 'd'},
4763 : {"define", required_argument, NULL, 'D'},
4764 : {"file", required_argument, NULL, 'f'},
4765 : {"fillfactor", required_argument, NULL, 'F'},
4766 : {"host", required_argument, NULL, 'h'},
4767 : {"initialize", no_argument, NULL, 'i'},
4768 : {"init-steps", required_argument, NULL, 'I'},
4769 : {"jobs", required_argument, NULL, 'j'},
4770 : {"log", no_argument, NULL, 'l'},
4771 : {"latency-limit", required_argument, NULL, 'L'},
4772 : {"no-vacuum", no_argument, NULL, 'n'},
4773 : {"port", required_argument, NULL, 'p'},
4774 : {"progress", required_argument, NULL, 'P'},
4775 : {"protocol", required_argument, NULL, 'M'},
4776 : {"quiet", no_argument, NULL, 'q'},
4777 : {"report-latencies", no_argument, NULL, 'r'},
4778 : {"rate", required_argument, NULL, 'R'},
4779 : {"scale", required_argument, NULL, 's'},
4780 : {"select-only", no_argument, NULL, 'S'},
4781 : {"skip-some-updates", no_argument, NULL, 'N'},
4782 : {"time", required_argument, NULL, 'T'},
4783 : {"transactions", required_argument, NULL, 't'},
4784 : {"username", required_argument, NULL, 'U'},
4785 : {"vacuum-all", no_argument, NULL, 'v'},
4786 : /* long-named only options */
4787 : {"unlogged-tables", no_argument, NULL, 1},
4788 : {"tablespace", required_argument, NULL, 2},
4789 : {"index-tablespace", required_argument, NULL, 3},
4790 : {"sampling-rate", required_argument, NULL, 4},
4791 : {"aggregate-interval", required_argument, NULL, 5},
4792 : {"progress-timestamp", no_argument, NULL, 6},
4793 : {"log-prefix", required_argument, NULL, 7},
4794 : {"foreign-keys", no_argument, NULL, 8},
4795 : {"random-seed", required_argument, NULL, 9},
4796 : {NULL, 0, NULL, 0}
4797 : };
4798 :
4799 : int c;
4800 222 : bool is_init_mode = false; /* initialize mode? */
4801 222 : char *initialize_steps = NULL;
4802 222 : bool foreign_keys = false;
4803 222 : bool is_no_vacuum = false;
4804 222 : bool do_vacuum_accounts = false; /* vacuum accounts table? */
4805 : int optindex;
4806 222 : bool scale_given = false;
4807 :
4808 222 : bool benchmarking_option_set = false;
4809 222 : bool initialization_option_set = false;
4810 222 : bool internal_script_used = false;
4811 :
4812 : CState *state; /* status of clients */
4813 : TState *threads; /* array of thread */
4814 :
4815 : instr_time start_time; /* start up time */
4816 : instr_time total_time;
4817 : instr_time conn_total_time;
4818 222 : int64 latency_late = 0;
4819 : StatsData stats;
4820 : int weight;
4821 :
4822 : int i;
4823 : int nclients_dealt;
4824 :
4825 : #ifdef HAVE_GETRLIMIT
4826 : struct rlimit rlim;
4827 : #endif
4828 :
4829 : PGconn *con;
4830 : PGresult *res;
4831 : char *env;
4832 :
4833 222 : progname = get_progname(argv[0]);
4834 :
4835 222 : if (argc > 1)
4836 : {
4837 222 : if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
4838 : {
4839 2 : usage();
4840 2 : exit(0);
4841 : }
4842 220 : if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
4843 : {
4844 2 : puts("pgbench (PostgreSQL) " PG_VERSION);
4845 2 : exit(0);
4846 : }
4847 : }
4848 :
4849 : #ifdef WIN32
4850 : /* stderr is buffered on Win32. */
4851 : setvbuf(stderr, NULL, _IONBF, 0);
4852 : #endif
4853 :
4854 218 : if ((env = getenv("PGHOST")) != NULL && *env != '\0')
4855 126 : pghost = env;
4856 218 : if ((env = getenv("PGPORT")) != NULL && *env != '\0')
4857 124 : pgport = env;
4858 94 : else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
4859 0 : login = env;
4860 :
4861 218 : state = (CState *) pg_malloc(sizeof(CState));
4862 218 : memset(state, 0, sizeof(CState));
4863 :
4864 : /* set random seed early, because it may be used while parsing scripts. */
4865 218 : if (!set_random_seed(getenv("PGBENCH_RANDOM_SEED")))
4866 : {
4867 0 : fprintf(stderr, "error while setting random seed from PGBENCH_RANDOM_SEED environment variable\n");
4868 0 : exit(1);
4869 : }
4870 :
4871 1772 : while ((c = getopt_long(argc, argv, "iI:h:nvp:dqb:SNc:j:Crs:t:T:U:lf:D:F:M:P:R:L:", long_options, &optindex)) != -1)
4872 : {
4873 : char *script;
4874 :
4875 1436 : switch (c)
4876 : {
4877 : case 'i':
4878 12 : is_init_mode = true;
4879 12 : break;
4880 : case 'I':
4881 8 : if (initialize_steps)
4882 0 : pg_free(initialize_steps);
4883 8 : initialize_steps = pg_strdup(optarg);
4884 8 : checkInitSteps(initialize_steps);
4885 6 : initialization_option_set = true;
4886 6 : break;
4887 : case 'h':
4888 2 : pghost = pg_strdup(optarg);
4889 2 : break;
4890 : case 'n':
4891 106 : is_no_vacuum = true;
4892 106 : break;
4893 : case 'v':
4894 2 : benchmarking_option_set = true;
4895 2 : do_vacuum_accounts = true;
4896 2 : break;
4897 : case 'p':
4898 2 : pgport = pg_strdup(optarg);
4899 2 : break;
4900 : case 'd':
4901 4 : debug++;
4902 4 : break;
4903 : case 'c':
4904 20 : benchmarking_option_set = true;
4905 20 : nclients = atoi(optarg);
4906 20 : if (nclients <= 0 || nclients > MAXCLIENTS)
4907 : {
4908 2 : fprintf(stderr, "invalid number of clients: \"%s\"\n",
4909 : optarg);
4910 2 : exit(1);
4911 : }
4912 : #ifdef HAVE_GETRLIMIT
4913 : #ifdef RLIMIT_NOFILE /* most platforms use RLIMIT_NOFILE */
4914 18 : if (getrlimit(RLIMIT_NOFILE, &rlim) == -1)
4915 : #else /* but BSD doesn't ... */
4916 : if (getrlimit(RLIMIT_OFILE, &rlim) == -1)
4917 : #endif /* RLIMIT_NOFILE */
4918 : {
4919 0 : fprintf(stderr, "getrlimit failed: %s\n", strerror(errno));
4920 0 : exit(1);
4921 : }
4922 18 : if (rlim.rlim_cur < nclients + 3)
4923 : {
4924 0 : fprintf(stderr, "need at least %d open files, but system limit is %ld\n",
4925 0 : nclients + 3, (long) rlim.rlim_cur);
4926 0 : fprintf(stderr, "Reduce number of clients, or use limit/ulimit to increase the system limit.\n");
4927 0 : exit(1);
4928 : }
4929 : #endif /* HAVE_GETRLIMIT */
4930 18 : break;
4931 : case 'j': /* jobs */
4932 6 : benchmarking_option_set = true;
4933 6 : nthreads = atoi(optarg);
4934 6 : if (nthreads <= 0)
4935 : {
4936 2 : fprintf(stderr, "invalid number of threads: \"%s\"\n",
4937 : optarg);
4938 2 : exit(1);
4939 : }
4940 : #ifndef ENABLE_THREAD_SAFETY
4941 : if (nthreads != 1)
4942 : {
4943 : fprintf(stderr, "threads are not supported on this platform; use -j1\n");
4944 : exit(1);
4945 : }
4946 : #endif /* !ENABLE_THREAD_SAFETY */
4947 4 : break;
4948 : case 'C':
4949 4 : benchmarking_option_set = true;
4950 4 : is_connect = true;
4951 4 : break;
4952 : case 'r':
4953 4 : benchmarking_option_set = true;
4954 4 : is_latencies = true;
4955 4 : break;
4956 : case 's':
4957 6 : scale_given = true;
4958 6 : scale = atoi(optarg);
4959 6 : if (scale <= 0)
4960 : {
4961 2 : fprintf(stderr, "invalid scaling factor: \"%s\"\n", optarg);
4962 2 : exit(1);
4963 : }
4964 4 : break;
4965 : case 't':
4966 122 : benchmarking_option_set = true;
4967 122 : nxacts = atoi(optarg);
4968 122 : if (nxacts <= 0)
4969 : {
4970 2 : fprintf(stderr, "invalid number of transactions: \"%s\"\n",
4971 : optarg);
4972 2 : exit(1);
4973 : }
4974 120 : break;
4975 : case 'T':
4976 10 : benchmarking_option_set = true;
4977 10 : duration = atoi(optarg);
4978 10 : if (duration <= 0)
4979 : {
4980 2 : fprintf(stderr, "invalid duration: \"%s\"\n", optarg);
4981 2 : exit(1);
4982 : }
4983 8 : break;
4984 : case 'U':
4985 2 : login = pg_strdup(optarg);
4986 2 : break;
4987 : case 'l':
4988 14 : benchmarking_option_set = true;
4989 14 : use_log = true;
4990 14 : break;
4991 : case 'q':
4992 2 : initialization_option_set = true;
4993 2 : use_quiet = true;
4994 2 : break;
4995 : case 'b':
4996 24 : if (strcmp(optarg, "list") == 0)
4997 : {
4998 2 : listAvailableScripts();
4999 2 : exit(0);
5000 : }
5001 22 : weight = parseScriptWeight(optarg, &script);
5002 18 : process_builtin(findBuiltin(script), weight);
5003 14 : benchmarking_option_set = true;
5004 14 : internal_script_used = true;
5005 14 : break;
5006 : case 'S':
5007 268 : process_builtin(findBuiltin("select-only"), 1);
5008 266 : benchmarking_option_set = true;
5009 266 : internal_script_used = true;
5010 266 : break;
5011 : case 'N':
5012 2 : process_builtin(findBuiltin("simple-update"), 1);
5013 2 : benchmarking_option_set = true;
5014 2 : internal_script_used = true;
5015 2 : break;
5016 : case 'f':
5017 126 : weight = parseScriptWeight(optarg, &script);
5018 126 : process_file(script, weight);
5019 72 : benchmarking_option_set = true;
5020 72 : break;
5021 : case 'D':
5022 : {
5023 : char *p;
5024 :
5025 512 : benchmarking_option_set = true;
5026 :
5027 512 : if ((p = strchr(optarg, '=')) == NULL || p == optarg || *(p + 1) == '\0')
5028 : {
5029 2 : fprintf(stderr, "invalid variable definition: \"%s\"\n",
5030 : optarg);
5031 2 : exit(1);
5032 : }
5033 :
5034 510 : *p++ = '\0';
5035 510 : if (!putVariable(&state[0], "option", optarg, p))
5036 0 : exit(1);
5037 : }
5038 510 : break;
5039 : case 'F':
5040 6 : initialization_option_set = true;
5041 6 : fillfactor = atoi(optarg);
5042 6 : if (fillfactor < 10 || fillfactor > 100)
5043 : {
5044 2 : fprintf(stderr, "invalid fillfactor: \"%s\"\n", optarg);
5045 2 : exit(1);
5046 : }
5047 4 : break;
5048 : case 'M':
5049 98 : benchmarking_option_set = true;
5050 284 : for (querymode = 0; querymode < NUM_QUERYMODE; querymode++)
5051 282 : if (strcmp(optarg, QUERYMODE[querymode]) == 0)
5052 96 : break;
5053 98 : if (querymode >= NUM_QUERYMODE)
5054 : {
5055 2 : fprintf(stderr, "invalid query mode (-M): \"%s\"\n",
5056 : optarg);
5057 2 : exit(1);
5058 : }
5059 96 : break;
5060 : case 'P':
5061 4 : benchmarking_option_set = true;
5062 4 : progress = atoi(optarg);
5063 4 : if (progress <= 0)
5064 : {
5065 2 : fprintf(stderr, "invalid thread progress delay: \"%s\"\n",
5066 : optarg);
5067 2 : exit(1);
5068 : }
5069 2 : break;
5070 : case 'R':
5071 : {
5072 : /* get a double from the beginning of option value */
5073 6 : double throttle_value = atof(optarg);
5074 :
5075 6 : benchmarking_option_set = true;
5076 :
5077 6 : if (throttle_value <= 0.0)
5078 : {
5079 2 : fprintf(stderr, "invalid rate limit: \"%s\"\n", optarg);
5080 2 : exit(1);
5081 : }
5082 : /* Invert rate limit into a time offset */
5083 4 : throttle_delay = (int64) (1000000.0 / throttle_value);
5084 : }
5085 4 : break;
5086 : case 'L':
5087 : {
5088 6 : double limit_ms = atof(optarg);
5089 :
5090 6 : if (limit_ms <= 0.0)
5091 : {
5092 2 : fprintf(stderr, "invalid latency limit: \"%s\"\n",
5093 : optarg);
5094 2 : exit(1);
5095 : }
5096 4 : benchmarking_option_set = true;
5097 4 : latency_limit = (int64) (limit_ms * 1000);
5098 : }
5099 4 : break;
5100 : case 1: /* unlogged-tables */
5101 4 : initialization_option_set = true;
5102 4 : unlogged_tables = true;
5103 4 : break;
5104 : case 2: /* tablespace */
5105 2 : initialization_option_set = true;
5106 2 : tablespace = pg_strdup(optarg);
5107 2 : break;
5108 : case 3: /* index-tablespace */
5109 2 : initialization_option_set = true;
5110 2 : index_tablespace = pg_strdup(optarg);
5111 2 : break;
5112 : case 4: /* sampling-rate */
5113 10 : benchmarking_option_set = true;
5114 10 : sample_rate = atof(optarg);
5115 10 : if (sample_rate <= 0.0 || sample_rate > 1.0)
5116 : {
5117 2 : fprintf(stderr, "invalid sampling rate: \"%s\"\n", optarg);
5118 2 : exit(1);
5119 : }
5120 8 : break;
5121 : case 5: /* aggregate-interval */
5122 12 : benchmarking_option_set = true;
5123 12 : agg_interval = atoi(optarg);
5124 12 : if (agg_interval <= 0)
5125 : {
5126 2 : fprintf(stderr, "invalid number of seconds for aggregation: \"%s\"\n",
5127 : optarg);
5128 2 : exit(1);
5129 : }
5130 10 : break;
5131 : case 6: /* progress-timestamp */
5132 4 : progress_timestamp = true;
5133 4 : benchmarking_option_set = true;
5134 4 : break;
5135 : case 7: /* log-prefix */
5136 8 : benchmarking_option_set = true;
5137 8 : logfile_prefix = pg_strdup(optarg);
5138 8 : break;
5139 : case 8: /* foreign-keys */
5140 4 : initialization_option_set = true;
5141 4 : foreign_keys = true;
5142 4 : break;
5143 : case 9: /* random-seed */
5144 8 : benchmarking_option_set = true;
5145 8 : if (!set_random_seed(optarg))
5146 : {
5147 2 : fprintf(stderr, "error while setting random seed from --random-seed option\n");
5148 2 : exit(1);
5149 : }
5150 6 : break;
5151 : default:
5152 4 : fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
5153 4 : exit(1);
5154 : break;
5155 : }
5156 : }
5157 :
5158 : /* set default script if none */
5159 118 : if (num_scripts == 0 && !is_init_mode)
5160 : {
5161 20 : process_builtin(findBuiltin("tpcb-like"), 1);
5162 20 : benchmarking_option_set = true;
5163 20 : internal_script_used = true;
5164 : }
5165 :
5166 : /* if not simple query mode, parse the script(s) to find parameters */
5167 118 : if (querymode != QUERY_SIMPLE)
5168 : {
5169 120 : for (i = 0; i < num_scripts; i++)
5170 : {
5171 62 : Command **commands = sql_script[i].commands;
5172 : int j;
5173 :
5174 156 : for (j = 0; commands[j] != NULL; j++)
5175 : {
5176 96 : if (commands[j]->type != SQL_COMMAND)
5177 64 : continue;
5178 32 : if (!parseQuery(commands[j]))
5179 2 : exit(1);
5180 : }
5181 : }
5182 : }
5183 :
5184 : /* compute total_weight */
5185 232 : for (i = 0; i < num_scripts; i++)
5186 : /* cannot overflow: weight is 32b, total_weight 64b */
5187 116 : total_weight += sql_script[i].weight;
5188 :
5189 116 : if (total_weight == 0 && !is_init_mode)
5190 : {
5191 2 : fprintf(stderr, "total script weight must not be zero\n");
5192 2 : exit(1);
5193 : }
5194 :
5195 : /* show per script stats if several scripts are used */
5196 114 : if (num_scripts > 1)
5197 2 : per_script_stats = true;
5198 :
5199 : /*
5200 : * Don't need more threads than there are clients. (This is not merely an
5201 : * optimization; throttle_delay is calculated incorrectly below if some
5202 : * threads have no clients assigned to them.)
5203 : */
5204 114 : if (nthreads > nclients)
5205 2 : nthreads = nclients;
5206 :
5207 : /* compute a per thread delay */
5208 114 : throttle_delay *= nthreads;
5209 :
5210 114 : if (argc > optind)
5211 2 : dbName = argv[optind];
5212 : else
5213 : {
5214 112 : if ((env = getenv("PGDATABASE")) != NULL && *env != '\0')
5215 88 : dbName = env;
5216 24 : else if (login != NULL && *login != '\0')
5217 0 : dbName = login;
5218 : else
5219 24 : dbName = "";
5220 : }
5221 :
5222 114 : if (is_init_mode)
5223 : {
5224 8 : if (benchmarking_option_set)
5225 : {
5226 2 : fprintf(stderr, "some of the specified options cannot be used in initialization (-i) mode\n");
5227 2 : exit(1);
5228 : }
5229 :
5230 6 : if (initialize_steps == NULL)
5231 2 : initialize_steps = pg_strdup(DEFAULT_INIT_STEPS);
5232 :
5233 6 : if (is_no_vacuum)
5234 : {
5235 : /* Remove any vacuum step in initialize_steps */
5236 : char *p;
5237 :
5238 10 : while ((p = strchr(initialize_steps, 'v')) != NULL)
5239 6 : *p = ' ';
5240 : }
5241 :
5242 6 : if (foreign_keys)
5243 : {
5244 : /* Add 'f' to end of initialize_steps, if not already there */
5245 4 : if (strchr(initialize_steps, 'f') == NULL)
5246 : {
5247 4 : initialize_steps = (char *)
5248 4 : pg_realloc(initialize_steps,
5249 4 : strlen(initialize_steps) + 2);
5250 4 : strcat(initialize_steps, "f");
5251 : }
5252 : }
5253 :
5254 6 : runInitSteps(initialize_steps);
5255 6 : exit(0);
5256 : }
5257 : else
5258 : {
5259 106 : if (initialization_option_set)
5260 : {
5261 4 : fprintf(stderr, "some of the specified options cannot be used in benchmarking mode\n");
5262 4 : exit(1);
5263 : }
5264 : }
5265 :
5266 102 : if (nxacts > 0 && duration > 0)
5267 : {
5268 4 : fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both\n");
5269 4 : exit(1);
5270 : }
5271 :
5272 : /* Use DEFAULT_NXACTS if neither nxacts nor duration is specified. */
5273 98 : if (nxacts <= 0 && duration <= 0)
5274 12 : nxacts = DEFAULT_NXACTS;
5275 :
5276 : /* --sampling-rate may be used only with -l */
5277 98 : if (sample_rate > 0.0 && !use_log)
5278 : {
5279 2 : fprintf(stderr, "log sampling (--sampling-rate) is allowed only when logging transactions (-l)\n");
5280 2 : exit(1);
5281 : }
5282 :
5283 : /* --sampling-rate may not be used with --aggregate-interval */
5284 96 : if (sample_rate > 0.0 && agg_interval > 0)
5285 : {
5286 2 : fprintf(stderr, "log sampling (--sampling-rate) and aggregation (--aggregate-interval) cannot be used at the same time\n");
5287 2 : exit(1);
5288 : }
5289 :
5290 94 : if (agg_interval > 0 && !use_log)
5291 : {
5292 2 : fprintf(stderr, "log aggregation is allowed only when actually logging transactions\n");
5293 2 : exit(1);
5294 : }
5295 :
5296 92 : if (!use_log && logfile_prefix)
5297 : {
5298 2 : fprintf(stderr, "log file prefix (--log-prefix) is allowed only when logging transactions (-l)\n");
5299 2 : exit(1);
5300 : }
5301 :
5302 90 : if (duration > 0 && agg_interval > duration)
5303 : {
5304 2 : fprintf(stderr, "number of seconds for aggregation (%d) must not be higher than test duration (%d)\n", agg_interval, duration);
5305 2 : exit(1);
5306 : }
5307 :
5308 88 : if (duration > 0 && agg_interval > 0 && duration % agg_interval != 0)
5309 : {
5310 2 : fprintf(stderr, "duration (%d) must be a multiple of aggregation interval (%d)\n", duration, agg_interval);
5311 2 : exit(1);
5312 : }
5313 :
5314 86 : if (progress_timestamp && progress == 0)
5315 : {
5316 2 : fprintf(stderr, "--progress-timestamp is allowed only under --progress\n");
5317 2 : exit(1);
5318 : }
5319 :
5320 : /*
5321 : * save main process id in the global variable because process id will be
5322 : * changed after fork.
5323 : */
5324 84 : main_pid = (int) getpid();
5325 :
5326 84 : if (nclients > 1)
5327 : {
5328 14 : state = (CState *) pg_realloc(state, sizeof(CState) * nclients);
5329 14 : memset(state + 1, 0, sizeof(CState) * (nclients - 1));
5330 :
5331 : /* copy any -D switch values to all clients */
5332 50 : for (i = 1; i < nclients; i++)
5333 : {
5334 : int j;
5335 :
5336 36 : state[i].id = i;
5337 38 : for (j = 0; j < state[0].nvariables; j++)
5338 : {
5339 2 : Variable *var = &state[0].variables[j];
5340 :
5341 2 : if (var->value.type != PGBT_NO_VALUE)
5342 : {
5343 0 : if (!putVariableValue(&state[i], "startup",
5344 0 : var->name, &var->value))
5345 0 : exit(1);
5346 : }
5347 : else
5348 : {
5349 2 : if (!putVariable(&state[i], "startup",
5350 2 : var->name, var->svalue))
5351 0 : exit(1);
5352 : }
5353 : }
5354 : }
5355 : }
5356 :
5357 : /* other CState initializations */
5358 204 : for (i = 0; i < nclients; i++)
5359 : {
5360 120 : state[i].cstack = conditional_stack_create();
5361 : }
5362 :
5363 84 : if (debug)
5364 : {
5365 2 : if (duration <= 0)
5366 2 : printf("pghost: %s pgport: %s nclients: %d nxacts: %d dbName: %s\n",
5367 : pghost, pgport, nclients, nxacts, dbName);
5368 : else
5369 0 : printf("pghost: %s pgport: %s nclients: %d duration: %d dbName: %s\n",
5370 : pghost, pgport, nclients, duration, dbName);
5371 : }
5372 :
5373 : /* opening connection... */
5374 84 : con = doConnect();
5375 84 : if (con == NULL)
5376 2 : exit(1);
5377 :
5378 82 : if (PQstatus(con) == CONNECTION_BAD)
5379 : {
5380 0 : fprintf(stderr, "connection to database \"%s\" failed\n", dbName);
5381 0 : fprintf(stderr, "%s", PQerrorMessage(con));
5382 0 : exit(1);
5383 : }
5384 :
5385 82 : if (internal_script_used)
5386 : {
5387 : /*
5388 : * get the scaling factor that should be same as count(*) from
5389 : * pgbench_branches if this is not a custom query
5390 : */
5391 14 : res = PQexec(con, "select count(*) from pgbench_branches");
5392 14 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
5393 : {
5394 2 : char *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
5395 :
5396 2 : fprintf(stderr, "%s", PQerrorMessage(con));
5397 2 : if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) == 0)
5398 : {
5399 2 : fprintf(stderr, "Perhaps you need to do initialization (\"pgbench -i\") in database \"%s\"\n", PQdb(con));
5400 : }
5401 :
5402 2 : exit(1);
5403 : }
5404 12 : scale = atoi(PQgetvalue(res, 0, 0));
5405 12 : if (scale < 0)
5406 : {
5407 0 : fprintf(stderr, "invalid count(*) from pgbench_branches: \"%s\"\n",
5408 : PQgetvalue(res, 0, 0));
5409 0 : exit(1);
5410 : }
5411 12 : PQclear(res);
5412 :
5413 : /* warn if we override user-given -s switch */
5414 12 : if (scale_given)
5415 2 : fprintf(stderr,
5416 : "scale option ignored, using count from pgbench_branches table (%d)\n",
5417 : scale);
5418 : }
5419 :
5420 : /*
5421 : * :scale variables normally get -s or database scale, but don't override
5422 : * an explicit -D switch
5423 : */
5424 80 : if (lookupVariable(&state[0], "scale") == NULL)
5425 : {
5426 196 : for (i = 0; i < nclients; i++)
5427 : {
5428 116 : if (!putVariableInt(&state[i], "startup", "scale", scale))
5429 0 : exit(1);
5430 : }
5431 : }
5432 :
5433 : /*
5434 : * Define a :client_id variable that is unique per connection. But don't
5435 : * override an explicit -D switch.
5436 : */
5437 80 : if (lookupVariable(&state[0], "client_id") == NULL)
5438 : {
5439 196 : for (i = 0; i < nclients; i++)
5440 116 : if (!putVariableInt(&state[i], "startup", "client_id", i))
5441 0 : exit(1);
5442 : }
5443 :
5444 : /* set default seed for hash functions */
5445 80 : if (lookupVariable(&state[0], "default_seed") == NULL)
5446 : {
5447 240 : uint64 seed = ((uint64) (random() & 0xFFFF) << 48) |
5448 160 : ((uint64) (random() & 0xFFFF) << 32) |
5449 80 : ((uint64) (random() & 0xFFFF) << 16) |
5450 80 : (uint64) (random() & 0xFFFF);
5451 :
5452 196 : for (i = 0; i < nclients; i++)
5453 116 : if (!putVariableInt(&state[i], "startup", "default_seed", (int64) seed))
5454 0 : exit(1);
5455 : }
5456 :
5457 : /* set random seed unless overwritten */
5458 80 : if (lookupVariable(&state[0], "random_seed") == NULL)
5459 : {
5460 196 : for (i = 0; i < nclients; i++)
5461 116 : if (!putVariableInt(&state[i], "startup", "random_seed", random_seed))
5462 0 : exit(1);
5463 : }
5464 :
5465 80 : if (!is_no_vacuum)
5466 : {
5467 12 : fprintf(stderr, "starting vacuum...");
5468 12 : tryExecuteStatement(con, "vacuum pgbench_branches");
5469 12 : tryExecuteStatement(con, "vacuum pgbench_tellers");
5470 12 : tryExecuteStatement(con, "truncate pgbench_history");
5471 12 : fprintf(stderr, "end.\n");
5472 :
5473 12 : if (do_vacuum_accounts)
5474 : {
5475 0 : fprintf(stderr, "starting vacuum pgbench_accounts...");
5476 0 : tryExecuteStatement(con, "vacuum analyze pgbench_accounts");
5477 0 : fprintf(stderr, "end.\n");
5478 : }
5479 : }
5480 80 : PQfinish(con);
5481 :
5482 : /* set up thread data structures */
5483 80 : threads = (TState *) pg_malloc(sizeof(TState) * nthreads);
5484 80 : nclients_dealt = 0;
5485 :
5486 160 : for (i = 0; i < nthreads; i++)
5487 : {
5488 80 : TState *thread = &threads[i];
5489 :
5490 80 : thread->tid = i;
5491 80 : thread->state = &state[nclients_dealt];
5492 80 : thread->nstate =
5493 80 : (nclients - nclients_dealt + nthreads - i - 1) / (nthreads - i);
5494 80 : thread->random_state[0] = random();
5495 80 : thread->random_state[1] = random();
5496 80 : thread->random_state[2] = random();
5497 80 : thread->logfile = NULL; /* filled in later */
5498 80 : thread->latency_late = 0;
5499 80 : thread->zipf_cache.nb_cells = 0;
5500 80 : thread->zipf_cache.current = 0;
5501 80 : thread->zipf_cache.overflowCount = 0;
5502 80 : initStats(&thread->stats, 0);
5503 :
5504 80 : nclients_dealt += thread->nstate;
5505 : }
5506 :
5507 : /* all clients must be assigned to a thread */
5508 : Assert(nclients_dealt == nclients);
5509 :
5510 : /* get start up time */
5511 80 : INSTR_TIME_SET_CURRENT(start_time);
5512 :
5513 : /* set alarm if duration is specified. */
5514 80 : if (duration > 0)
5515 0 : setalarm(duration);
5516 :
5517 : /* start threads */
5518 : #ifdef ENABLE_THREAD_SAFETY
5519 160 : for (i = 0; i < nthreads; i++)
5520 : {
5521 80 : TState *thread = &threads[i];
5522 :
5523 80 : INSTR_TIME_SET_CURRENT(thread->start_time);
5524 :
5525 : /* compute when to stop */
5526 80 : if (duration > 0)
5527 0 : end_time = INSTR_TIME_GET_MICROSEC(thread->start_time) +
5528 0 : (int64) 1000000 * duration;
5529 :
5530 : /* the first thread (i = 0) is executed by main thread */
5531 80 : if (i > 0)
5532 : {
5533 0 : int err = pthread_create(&thread->thread, NULL, threadRun, thread);
5534 :
5535 0 : if (err != 0 || thread->thread == INVALID_THREAD)
5536 : {
5537 0 : fprintf(stderr, "could not create thread: %s\n", strerror(err));
5538 0 : exit(1);
5539 : }
5540 : }
5541 : else
5542 : {
5543 80 : thread->thread = INVALID_THREAD;
5544 : }
5545 : }
5546 : #else
5547 : INSTR_TIME_SET_CURRENT(threads[0].start_time);
5548 : /* compute when to stop */
5549 : if (duration > 0)
5550 : end_time = INSTR_TIME_GET_MICROSEC(threads[0].start_time) +
5551 : (int64) 1000000 * duration;
5552 : threads[0].thread = INVALID_THREAD;
5553 : #endif /* ENABLE_THREAD_SAFETY */
5554 :
5555 : /* wait for threads and accumulate results */
5556 80 : initStats(&stats, 0);
5557 80 : INSTR_TIME_SET_ZERO(conn_total_time);
5558 160 : for (i = 0; i < nthreads; i++)
5559 : {
5560 80 : TState *thread = &threads[i];
5561 :
5562 : #ifdef ENABLE_THREAD_SAFETY
5563 80 : if (threads[i].thread == INVALID_THREAD)
5564 : /* actually run this thread directly in the main thread */
5565 80 : (void) threadRun(thread);
5566 : else
5567 : /* wait of other threads. should check that 0 is returned? */
5568 0 : pthread_join(thread->thread, NULL);
5569 : #else
5570 : (void) threadRun(thread);
5571 : #endif /* ENABLE_THREAD_SAFETY */
5572 :
5573 : /* aggregate thread level stats */
5574 80 : mergeSimpleStats(&stats.latency, &thread->stats.latency);
5575 80 : mergeSimpleStats(&stats.lag, &thread->stats.lag);
5576 80 : stats.cnt += thread->stats.cnt;
5577 80 : stats.skipped += thread->stats.skipped;
5578 80 : latency_late += thread->latency_late;
5579 80 : INSTR_TIME_ADD(conn_total_time, thread->conn_time);
5580 : }
5581 80 : disconnect_all(state, nclients);
5582 :
5583 : /*
5584 : * XXX We compute results as though every client of every thread started
5585 : * and finished at the same time. That model can diverge noticeably from
5586 : * reality for a short benchmark run involving relatively many threads.
5587 : * The first thread may process notably many transactions before the last
5588 : * thread begins. Improving the model alone would bring limited benefit,
5589 : * because performance during those periods of partial thread count can
5590 : * easily exceed steady state performance. This is one of the many ways
5591 : * short runs convey deceptive performance figures.
5592 : */
5593 80 : INSTR_TIME_SET_CURRENT(total_time);
5594 80 : INSTR_TIME_SUBTRACT(total_time, start_time);
5595 80 : printResults(threads, &stats, total_time, conn_total_time, latency_late);
5596 :
5597 80 : return 0;
5598 : }
5599 :
5600 : static void *
5601 80 : threadRun(void *arg)
5602 : {
5603 80 : TState *thread = (TState *) arg;
5604 80 : CState *state = thread->state;
5605 : instr_time start,
5606 : end;
5607 80 : int nstate = thread->nstate;
5608 80 : int remains = nstate; /* number of remaining clients */
5609 : int i;
5610 :
5611 : /* for reporting progress: */
5612 80 : int64 thread_start = INSTR_TIME_GET_MICROSEC(thread->start_time);
5613 80 : int64 last_report = thread_start;
5614 80 : int64 next_report = last_report + (int64) progress * 1000000;
5615 : StatsData last,
5616 : aggs;
5617 :
5618 : /*
5619 : * Initialize throttling rate target for all of the thread's clients. It
5620 : * might be a little more accurate to reset thread->start_time here too.
5621 : * The possible drift seems too small relative to typical throttle delay
5622 : * times to worry about it.
5623 : */
5624 80 : INSTR_TIME_SET_CURRENT(start);
5625 80 : thread->throttle_trigger = INSTR_TIME_GET_MICROSEC(start);
5626 :
5627 80 : INSTR_TIME_SET_ZERO(thread->conn_time);
5628 :
5629 80 : initStats(&aggs, time(NULL));
5630 80 : last = aggs;
5631 :
5632 : /* open log file if requested */
5633 80 : if (use_log)
5634 : {
5635 : char logpath[MAXPGPATH];
5636 4 : char *prefix = logfile_prefix ? logfile_prefix : "pgbench_log";
5637 :
5638 4 : if (thread->tid == 0)
5639 4 : snprintf(logpath, sizeof(logpath), "%s.%d", prefix, main_pid);
5640 : else
5641 0 : snprintf(logpath, sizeof(logpath), "%s.%d.%d", prefix, main_pid, thread->tid);
5642 :
5643 4 : thread->logfile = fopen(logpath, "w");
5644 :
5645 4 : if (thread->logfile == NULL)
5646 : {
5647 0 : fprintf(stderr, "could not open logfile \"%s\": %s\n",
5648 0 : logpath, strerror(errno));
5649 0 : goto done;
5650 : }
5651 : }
5652 :
5653 80 : if (!is_connect)
5654 : {
5655 : /* make connections to the database */
5656 178 : for (i = 0; i < nstate; i++)
5657 : {
5658 102 : if ((state[i].con = doConnect()) == NULL)
5659 0 : goto done;
5660 : }
5661 : }
5662 :
5663 : /* time after thread and connections set up */
5664 80 : INSTR_TIME_SET_CURRENT(thread->conn_time);
5665 80 : INSTR_TIME_SUBTRACT(thread->conn_time, thread->start_time);
5666 :
5667 : /* explicitly initialize the state machines */
5668 196 : for (i = 0; i < nstate; i++)
5669 : {
5670 116 : state[i].state = CSTATE_CHOOSE_SCRIPT;
5671 : }
5672 :
5673 : /* loop till all clients have terminated */
5674 2886 : while (remains > 0)
5675 : {
5676 : fd_set input_mask;
5677 : int maxsock; /* max socket number to be waited for */
5678 : int64 min_usec;
5679 2726 : int64 now_usec = 0; /* set this only if needed */
5680 :
5681 : /* identify which client sockets should be checked for input */
5682 2726 : FD_ZERO(&input_mask);
5683 2726 : maxsock = -1;
5684 2726 : min_usec = PG_INT64_MAX;
5685 11856 : for (i = 0; i < nstate; i++)
5686 : {
5687 9210 : CState *st = &state[i];
5688 :
5689 9210 : if (st->state == CSTATE_THROTTLE && timer_exceeded)
5690 : {
5691 : /* interrupt client that has not started a transaction */
5692 0 : st->state = CSTATE_FINISHED;
5693 0 : finishCon(st);
5694 0 : remains--;
5695 : }
5696 9210 : else if (st->state == CSTATE_SLEEP || st->state == CSTATE_THROTTLE)
5697 4 : {
5698 : /* a nap from the script, or under throttling */
5699 : int64 this_usec;
5700 :
5701 : /* get current time if needed */
5702 4 : if (now_usec == 0)
5703 : {
5704 : instr_time now;
5705 :
5706 4 : INSTR_TIME_SET_CURRENT(now);
5707 4 : now_usec = INSTR_TIME_GET_MICROSEC(now);
5708 : }
5709 :
5710 : /* min_usec should be the minimum delay across all clients */
5711 8 : this_usec = (st->state == CSTATE_SLEEP ?
5712 4 : st->sleep_until : st->txn_scheduled) - now_usec;
5713 4 : if (min_usec > this_usec)
5714 4 : min_usec = this_usec;
5715 : }
5716 9206 : else if (st->state == CSTATE_WAIT_RESULT)
5717 : {
5718 : /*
5719 : * waiting for result from server - nothing to do unless the
5720 : * socket is readable
5721 : */
5722 8330 : int sock = PQsocket(st->con);
5723 :
5724 8330 : if (sock < 0)
5725 : {
5726 0 : fprintf(stderr, "invalid socket: %s",
5727 0 : PQerrorMessage(st->con));
5728 0 : goto done;
5729 : }
5730 :
5731 8330 : FD_SET(sock, &input_mask);
5732 8330 : if (maxsock < sock)
5733 8244 : maxsock = sock;
5734 : }
5735 1752 : else if (st->state != CSTATE_ABORTED &&
5736 876 : st->state != CSTATE_FINISHED)
5737 : {
5738 : /*
5739 : * This client thread is ready to do something, so we don't
5740 : * want to wait. No need to examine additional clients.
5741 : */
5742 80 : min_usec = 0;
5743 80 : break;
5744 : }
5745 : }
5746 :
5747 : /* also wake up to print the next progress report on time */
5748 2726 : if (progress && min_usec > 0 && thread->tid == 0)
5749 : {
5750 : /* get current time if needed */
5751 0 : if (now_usec == 0)
5752 : {
5753 : instr_time now;
5754 :
5755 0 : INSTR_TIME_SET_CURRENT(now);
5756 0 : now_usec = INSTR_TIME_GET_MICROSEC(now);
5757 : }
5758 :
5759 0 : if (now_usec >= next_report)
5760 0 : min_usec = 0;
5761 0 : else if ((next_report - now_usec) < min_usec)
5762 0 : min_usec = next_report - now_usec;
5763 : }
5764 :
5765 : /*
5766 : * If no clients are ready to execute actions, sleep until we receive
5767 : * data from the server, or a nap-time specified in the script ends,
5768 : * or it's time to print a progress report. Update input_mask to show
5769 : * which client(s) received data.
5770 : */
5771 2726 : if (min_usec > 0)
5772 : {
5773 2646 : int nsocks = 0; /* return from select(2) if called */
5774 :
5775 2646 : if (min_usec != PG_INT64_MAX)
5776 : {
5777 4 : if (maxsock != -1)
5778 : {
5779 : struct timeval timeout;
5780 :
5781 0 : timeout.tv_sec = min_usec / 1000000;
5782 0 : timeout.tv_usec = min_usec % 1000000;
5783 0 : nsocks = select(maxsock + 1, &input_mask, NULL, NULL, &timeout);
5784 : }
5785 : else /* nothing active, simple sleep */
5786 : {
5787 4 : pg_usleep(min_usec);
5788 : }
5789 : }
5790 : else /* no explicit delay, select without timeout */
5791 : {
5792 2642 : nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
5793 : }
5794 :
5795 2646 : if (nsocks < 0)
5796 : {
5797 0 : if (errno == EINTR)
5798 : {
5799 : /* On EINTR, go back to top of loop */
5800 0 : continue;
5801 : }
5802 : /* must be something wrong */
5803 0 : fprintf(stderr, "select() failed: %s\n", strerror(errno));
5804 0 : goto done;
5805 : }
5806 : }
5807 : else
5808 : {
5809 : /* min_usec == 0, i.e. something needs to be executed */
5810 :
5811 : /* If we didn't call select(), don't try to read any data */
5812 80 : FD_ZERO(&input_mask);
5813 : }
5814 :
5815 : /* ok, advance the state machine of each connection */
5816 11972 : for (i = 0; i < nstate; i++)
5817 : {
5818 9246 : CState *st = &state[i];
5819 :
5820 9246 : if (st->state == CSTATE_WAIT_RESULT)
5821 : {
5822 : /* don't call doCustom unless data is available */
5823 8330 : int sock = PQsocket(st->con);
5824 :
5825 8330 : if (sock < 0)
5826 : {
5827 0 : fprintf(stderr, "invalid socket: %s",
5828 0 : PQerrorMessage(st->con));
5829 0 : goto done;
5830 : }
5831 :
5832 8330 : if (!FD_ISSET(sock, &input_mask))
5833 4252 : continue;
5834 : }
5835 1036 : else if (st->state == CSTATE_FINISHED ||
5836 120 : st->state == CSTATE_ABORTED)
5837 : {
5838 : /* this client is done, no need to consider it anymore */
5839 796 : continue;
5840 : }
5841 :
5842 4198 : doCustom(thread, st, &aggs);
5843 :
5844 : /* If doCustom changed client to finished state, reduce remains */
5845 4198 : if (st->state == CSTATE_FINISHED || st->state == CSTATE_ABORTED)
5846 116 : remains--;
5847 : }
5848 :
5849 : /* progress report is made by thread 0 for all threads */
5850 2726 : if (progress && thread->tid == 0)
5851 : {
5852 : instr_time now_time;
5853 : int64 now;
5854 :
5855 0 : INSTR_TIME_SET_CURRENT(now_time);
5856 0 : now = INSTR_TIME_GET_MICROSEC(now_time);
5857 0 : if (now >= next_report)
5858 : {
5859 : /* generate and show report */
5860 : StatsData cur;
5861 0 : int64 run = now - last_report,
5862 : ntx;
5863 : double tps,
5864 : total_run,
5865 : latency,
5866 : sqlat,
5867 : lag,
5868 : stdev;
5869 : char tbuf[315];
5870 :
5871 : /*
5872 : * Add up the statistics of all threads.
5873 : *
5874 : * XXX: No locking. There is no guarantee that we get an
5875 : * atomic snapshot of the transaction count and latencies, so
5876 : * these figures can well be off by a small amount. The
5877 : * progress report's purpose is to give a quick overview of
5878 : * how the test is going, so that shouldn't matter too much.
5879 : * (If a read from a 64-bit integer is not atomic, you might
5880 : * get a "torn" read and completely bogus latencies though!)
5881 : */
5882 0 : initStats(&cur, 0);
5883 0 : for (i = 0; i < nthreads; i++)
5884 : {
5885 0 : mergeSimpleStats(&cur.latency, &thread[i].stats.latency);
5886 0 : mergeSimpleStats(&cur.lag, &thread[i].stats.lag);
5887 0 : cur.cnt += thread[i].stats.cnt;
5888 0 : cur.skipped += thread[i].stats.skipped;
5889 : }
5890 :
5891 : /* we count only actually executed transactions */
5892 0 : ntx = (cur.cnt - cur.skipped) - (last.cnt - last.skipped);
5893 0 : total_run = (now - thread_start) / 1000000.0;
5894 0 : tps = 1000000.0 * ntx / run;
5895 0 : if (ntx > 0)
5896 : {
5897 0 : latency = 0.001 * (cur.latency.sum - last.latency.sum) / ntx;
5898 0 : sqlat = 1.0 * (cur.latency.sum2 - last.latency.sum2) / ntx;
5899 0 : stdev = 0.001 * sqrt(sqlat - 1000000.0 * latency * latency);
5900 0 : lag = 0.001 * (cur.lag.sum - last.lag.sum) / ntx;
5901 : }
5902 : else
5903 : {
5904 0 : latency = sqlat = stdev = lag = 0;
5905 : }
5906 :
5907 0 : if (progress_timestamp)
5908 : {
5909 : /*
5910 : * On some platforms the current system timestamp is
5911 : * available in now_time, but rather than get entangled
5912 : * with that, we just eat the cost of an extra syscall in
5913 : * all cases.
5914 : */
5915 : struct timeval tv;
5916 :
5917 0 : gettimeofday(&tv, NULL);
5918 0 : snprintf(tbuf, sizeof(tbuf), "%ld.%03ld s",
5919 0 : (long) tv.tv_sec, (long) (tv.tv_usec / 1000));
5920 : }
5921 : else
5922 : {
5923 : /* round seconds are expected, but the thread may be late */
5924 0 : snprintf(tbuf, sizeof(tbuf), "%.1f s", total_run);
5925 : }
5926 :
5927 0 : fprintf(stderr,
5928 : "progress: %s, %.1f tps, lat %.3f ms stddev %.3f",
5929 : tbuf, tps, latency, stdev);
5930 :
5931 0 : if (throttle_delay)
5932 : {
5933 0 : fprintf(stderr, ", lag %.3f ms", lag);
5934 0 : if (latency_limit)
5935 0 : fprintf(stderr, ", " INT64_FORMAT " skipped",
5936 0 : cur.skipped - last.skipped);
5937 : }
5938 0 : fprintf(stderr, "\n");
5939 :
5940 0 : last = cur;
5941 0 : last_report = now;
5942 :
5943 : /*
5944 : * Ensure that the next report is in the future, in case
5945 : * pgbench/postgres got stuck somewhere.
5946 : */
5947 : do
5948 : {
5949 0 : next_report += (int64) progress * 1000000;
5950 0 : } while (now >= next_report);
5951 : }
5952 : }
5953 : }
5954 :
5955 : done:
5956 80 : INSTR_TIME_SET_CURRENT(start);
5957 80 : disconnect_all(state, nstate);
5958 80 : INSTR_TIME_SET_CURRENT(end);
5959 80 : INSTR_TIME_ACCUM_DIFF(thread->conn_time, end, start);
5960 80 : if (thread->logfile)
5961 : {
5962 4 : if (agg_interval > 0)
5963 : {
5964 : /* log aggregated but not yet reported transactions */
5965 0 : doLog(thread, state, &aggs, false, 0, 0);
5966 : }
5967 4 : fclose(thread->logfile);
5968 4 : thread->logfile = NULL;
5969 : }
5970 80 : return NULL;
5971 : }
5972 :
5973 : static void
5974 568 : finishCon(CState *st)
5975 : {
5976 568 : if (st->con != NULL)
5977 : {
5978 322 : PQfinish(st->con);
5979 322 : st->con = NULL;
5980 : }
5981 568 : }
5982 :
5983 : /*
5984 : * Support for duration option: set timer_exceeded after so many seconds.
5985 : */
5986 :
5987 : #ifndef WIN32
5988 :
5989 : static void
5990 0 : handle_sig_alarm(SIGNAL_ARGS)
5991 : {
5992 0 : timer_exceeded = true;
5993 0 : }
5994 :
5995 : static void
5996 0 : setalarm(int seconds)
5997 : {
5998 0 : pqsignal(SIGALRM, handle_sig_alarm);
5999 0 : alarm(seconds);
6000 0 : }
6001 :
6002 : #else /* WIN32 */
6003 :
6004 : static VOID CALLBACK
6005 : win32_timer_callback(PVOID lpParameter, BOOLEAN TimerOrWaitFired)
6006 : {
6007 : timer_exceeded = true;
6008 : }
6009 :
6010 : static void
6011 : setalarm(int seconds)
6012 : {
6013 : HANDLE queue;
6014 : HANDLE timer;
6015 :
6016 : /* This function will be called at most once, so we can cheat a bit. */
6017 : queue = CreateTimerQueue();
6018 : if (seconds > ((DWORD) -1) / 1000 ||
6019 : !CreateTimerQueueTimer(&timer, queue,
6020 : win32_timer_callback, NULL, seconds * 1000, 0,
6021 : WT_EXECUTEINTIMERTHREAD | WT_EXECUTEONLYONCE))
6022 : {
6023 : fprintf(stderr, "failed to set timer\n");
6024 : exit(1);
6025 : }
6026 : }
6027 :
6028 : /* partial pthread implementation for Windows */
6029 :
6030 : typedef struct win32_pthread
6031 : {
6032 : HANDLE handle;
6033 : void *(*routine) (void *);
6034 : void *arg;
6035 : void *result;
6036 : } win32_pthread;
6037 :
6038 : static unsigned __stdcall
6039 : win32_pthread_run(void *arg)
6040 : {
6041 : win32_pthread *th = (win32_pthread *) arg;
6042 :
6043 : th->result = th->routine(th->arg);
6044 :
6045 : return 0;
6046 : }
6047 :
6048 : static int
6049 : pthread_create(pthread_t *thread,
6050 : pthread_attr_t *attr,
6051 : void *(*start_routine) (void *),
6052 : void *arg)
6053 : {
6054 : int save_errno;
6055 : win32_pthread *th;
6056 :
6057 : th = (win32_pthread *) pg_malloc(sizeof(win32_pthread));
6058 : th->routine = start_routine;
6059 : th->arg = arg;
6060 : th->result = NULL;
6061 :
6062 : th->handle = (HANDLE) _beginthreadex(NULL, 0, win32_pthread_run, th, 0, NULL);
6063 : if (th->handle == NULL)
6064 : {
6065 : save_errno = errno;
6066 : free(th);
6067 : return save_errno;
6068 : }
6069 :
6070 : *thread = th;
6071 : return 0;
6072 : }
6073 :
6074 : static int
6075 : pthread_join(pthread_t th, void **thread_return)
6076 : {
6077 : if (th == NULL || th->handle == NULL)
6078 : return errno = EINVAL;
6079 :
6080 : if (WaitForSingleObject(th->handle, INFINITE) != WAIT_OBJECT_0)
6081 : {
6082 : _dosmaperr(GetLastError());
6083 : return errno;
6084 : }
6085 :
6086 : if (thread_return)
6087 : *thread_return = th->result;
6088 :
6089 : CloseHandle(th->handle);
6090 : free(th);
6091 : return 0;
6092 : }
6093 :
6094 : #endif /* WIN32 */
|
