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