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