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