?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Change pgbench to use the table names pgbench_accounts, pgbench_branches,
[PostgreSQL.git] / contrib / pgbench / pgbench.c
blob5fb561b9163e90653828c94ae56ae6422ff9dad7
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 * $PostgreSQL$
8 * Copyright (c) 2000-2009, 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 #include "postgres_fe.h"
30
31 #include "libpq-fe.h"
32 #include "pqsignal.h"
33
34 #include <ctype.h>
35
36 #ifdef WIN32
37 #undef FD_SETSIZE
38 #define FD_SETSIZE 1024
39 #include <win32.h>
40 #else
41 #include <signal.h>
42 #include <sys/time.h>
43 #include <unistd.h>
44 #endif /* ! WIN32 */
45
46 #ifdef HAVE_GETOPT_H
47 #include <getopt.h>
48 #endif
49
50 #ifdef HAVE_SYS_SELECT_H
51 #include <sys/select.h>
52 #endif
53
54 #ifdef HAVE_SYS_RESOURCE_H
55 #include <sys/resource.h> /* for getrlimit */
56 #endif
57
58 extern char *optarg;
59 extern int optind;
60
61
62 /********************************************************************
63 * some configurable parameters */
64
65 /* max number of clients allowed */
66 #ifdef FD_SETSIZE
67 #define MAXCLIENTS (FD_SETSIZE - 10)
68 #else
69 #define MAXCLIENTS 1024
70 #endif
71
72 #define DEFAULT_NXACTS 10 /* default nxacts */
73
74 int nclients = 1; /* default number of simulated clients */
75 int nxacts = 0; /* number of transactions per client */
76 int duration = 0; /* duration in seconds */
77
78 /*
79 * scaling factor. for example, scale = 10 will make 1000000 tuples in
80 * pgbench_accounts table.
81 */
82 int scale = 1;
83
84 /*
85 * fillfactor. for example, fillfactor = 90 will use only 90 percent
86 * space during inserts and leave 10 percent free.
87 */
88 int fillfactor = 100;
89
90 /*
91 * end of configurable parameters
92 *********************************************************************/
93
94 #define nbranches 1
95 #define ntellers 10
96 #define naccounts 100000
97
98 FILE *LOGFILE = NULL;
99
100 bool use_log; /* log transaction latencies to a file */
101
102 int remains; /* number of remaining clients */
103
104 int is_connect; /* establish connection for each transaction */
105
106 char *pghost = "";
107 char *pgport = "";
108 char *pgoptions = NULL;
109 char *pgtty = NULL;
110 char *login = NULL;
111 char *dbName;
112
113 volatile bool timer_exceeded = false; /* flag from signal handler */
114
115 /* variable definitions */
116 typedef struct
117 {
118 char *name; /* variable name */
119 char *value; /* its value */
120 } Variable;
121
122 #define MAX_FILES 128 /* max number of SQL script files allowed */
123
124 /*
125 * structures used in custom query mode
126 */
127
128 typedef struct
129 {
130 PGconn *con; /* connection handle to DB */
131 int id; /* client No. */
132 int state; /* state No. */
133 int cnt; /* xacts count */
134 int ecnt; /* error count */
135 int listen; /* 0 indicates that an async query has been
136 * sent */
137 int sleeping; /* 1 indicates that the client is napping */
138 struct timeval until; /* napping until */
139 Variable *variables; /* array of variable definitions */
140 int nvariables;
141 struct timeval txn_begin; /* used for measuring latencies */
142 int use_file; /* index in sql_files for this client */
143 bool prepared[MAX_FILES];
144 } CState;
145
146 /*
147 * queries read from files
148 */
149 #define SQL_COMMAND 1
150 #define META_COMMAND 2
151 #define MAX_ARGS 10
152
153 typedef enum QueryMode
154 {
155 QUERY_SIMPLE, /* simple query */
156 QUERY_EXTENDED, /* extended query */
157 QUERY_PREPARED, /* extended query with prepared statements */
158 NUM_QUERYMODE
159 } QueryMode;
160
161 static QueryMode querymode = QUERY_SIMPLE;
162 static const char *QUERYMODE[] = { "simple", "extended", "prepared" };
163
164 typedef struct
165 {
166 int type; /* command type (SQL_COMMAND or META_COMMAND) */
167 int argc; /* number of commands */
168 char *argv[MAX_ARGS]; /* command list */
169 } Command;
170
171 Command **sql_files[MAX_FILES]; /* SQL script files */
172 int num_files; /* number of script files */
173
174 /* default scenario */
175 static char *tpc_b = {
176 "\\set nbranches :scale\n"
177 "\\set ntellers 10 * :scale\n"
178 "\\set naccounts 100000 * :scale\n"
179 "\\setrandom aid 1 :naccounts\n"
180 "\\setrandom bid 1 :nbranches\n"
181 "\\setrandom tid 1 :ntellers\n"
182 "\\setrandom delta -5000 5000\n"
183 "BEGIN;\n"
184 "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
185 "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
186 "UPDATE pgbench_tellers SET tbalance = tbalance + :delta WHERE tid = :tid;\n"
187 "UPDATE pgbench_branches SET bbalance = bbalance + :delta WHERE bid = :bid;\n"
188 "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
189 "END;\n"
190 };
191
192 /* -N case */
193 static char *simple_update = {
194 "\\set nbranches :scale\n"
195 "\\set ntellers 10 * :scale\n"
196 "\\set naccounts 100000 * :scale\n"
197 "\\setrandom aid 1 :naccounts\n"
198 "\\setrandom bid 1 :nbranches\n"
199 "\\setrandom tid 1 :ntellers\n"
200 "\\setrandom delta -5000 5000\n"
201 "BEGIN;\n"
202 "UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
203 "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
204 "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
205 "END;\n"
206 };
207
208 /* -S case */
209 static char *select_only = {
210 "\\set naccounts 100000 * :scale\n"
211 "\\setrandom aid 1 :naccounts\n"
212 "SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
213 };
214
215 /* Connection overhead time */
216 static struct timeval conn_total_time = {0, 0};
217
218 /* Function prototypes */
219 static void setalarm(int seconds);
220
221
222 /* Calculate total time */
223 static void
224 addTime(struct timeval *t1, struct timeval *t2, struct timeval *result)
225 {
226 int sec = t1->tv_sec + t2->tv_sec;
227 int usec = t1->tv_usec + t2->tv_usec;
228 if (usec >= 1000000)
229 {
230 usec -= 1000000;
231 sec++;
232 }
233 result->tv_sec = sec;
234 result->tv_usec = usec;
235 }
236
237 /* Calculate time difference */
238 static void
239 diffTime(struct timeval *t1, struct timeval *t2, struct timeval *result)
240 {
241 int sec = t1->tv_sec - t2->tv_sec;
242 int usec = t1->tv_usec - t2->tv_usec;
243 if (usec < 0)
244 {
245 usec += 1000000;
246 sec--;
247 }
248 result->tv_sec = sec;
249 result->tv_usec = usec;
250 }
251
252 static void
253 usage(const char *progname)
254 {
255 printf("%s is a benchmarking tool for PostgreSQL.\n\n"
256 "Usage:\n"
257 " %s [OPTIONS]... [DBNAME]\n"
258 "\nInitialization options:\n"
259 " -i invokes initialization mode\n"
260 " -F NUM fill factor\n"
261 " -s NUM scaling factor\n"
262 "\nBenchmarking options:\n"
263 " -c NUM number of concurrent database clients (default: 1)\n"
264 " -C establish new connection for each transaction\n"
265 " -D VARNAME=VALUE\n"
266 " define variable for use by custom script\n"
267 " -f FILENAME read transaction script from FILENAME\n"
268 " -l write transaction times to log file\n"
269 " -M {simple|extended|prepared}\n"
270 " protocol for submitting queries to server (default: simple)\n"
271 " -n do not run VACUUM before tests\n"
272 " -N do not update tables \"pgbench_tellers\" and \"pgbench_branches\"\n"
273 " -s NUM report this scale factor in output\n"
274 " -S perform SELECT-only transactions\n"
275 " -t NUM number of transactions each client runs (default: 10)\n"
276 " -T NUM duration of benchmark test in seconds\n"
277 " -v vacuum all four standard tables before tests\n"
278 "\nCommon options:\n"
279 " -d print debugging output\n"
280 " -h HOSTNAME database server host or socket directory\n"
281 " -p PORT database server port number\n"
282 " -U USERNAME connect as specified database user\n"
283 " --help show this help, then exit\n"
284 " --version output version information, then exit\n"
285 "\n"
286 "Report bugs to <pgsql-bugs@postgresql.org>.\n",
287 progname, progname);
288 }
289
290 /* random number generator: uniform distribution from min to max inclusive */
291 static int
292 getrand(int min, int max)
293 {
294 /*
295 * Odd coding is so that min and max have approximately the same chance of
296 * being selected as do numbers between them.
297 */
298 return min + (int) (((max - min + 1) * (double) random()) / (MAX_RANDOM_VALUE + 1.0));
299 }
300
301 /* call PQexec() and exit() on failure */
302 static void
303 executeStatement(PGconn *con, const char *sql)
304 {
305 PGresult *res;
306
307 res = PQexec(con, sql);
308 if (PQresultStatus(res) != PGRES_COMMAND_OK)
309 {
310 fprintf(stderr, "%s", PQerrorMessage(con));
311 exit(1);
312 }
313 PQclear(res);
314 }
315
316 /* set up a connection to the backend */
317 static PGconn *
318 doConnect(void)
319 {
320 PGconn *conn;
321 static char *password = NULL;
322 bool new_pass;
323
324 /*
325 * Start the connection. Loop until we have a password if requested by
326 * backend.
327 */
328 do
329 {
330 new_pass = false;
331
332 conn = PQsetdbLogin(pghost, pgport, pgoptions, pgtty, dbName,
333 login, password);
334 if (!conn)
335 {
336 fprintf(stderr, "Connection to database \"%s\" failed\n",
337 dbName);
338 return NULL;
339 }
340
341 if (PQstatus(conn) == CONNECTION_BAD &&
342 PQconnectionNeedsPassword(conn) &&
343 password == NULL)
344 {
345 PQfinish(conn);
346 password = simple_prompt("Password: ", 100, false);
347 new_pass = true;
348 }
349 } while (new_pass);
350
351 /* check to see that the backend connection was successfully made */
352 if (PQstatus(conn) == CONNECTION_BAD)
353 {
354 fprintf(stderr, "Connection to database \"%s\" failed:\n%s",
355 dbName, PQerrorMessage(conn));
356 PQfinish(conn);
357 return NULL;
358 }
359
360 return conn;
361 }
362
363 /* throw away response from backend */
364 static void
365 discard_response(CState * state)
366 {
367 PGresult *res;
368
369 do
370 {
371 res = PQgetResult(state->con);
372 if (res)
373 PQclear(res);
374 } while (res);
375 }
376
377 /* check to see if the SQL result was good */
378 static int
379 check(CState * state, PGresult *res, int n)
380 {
381 CState *st = &state[n];
382
383 switch (PQresultStatus(res))
384 {
385 case PGRES_COMMAND_OK:
386 case PGRES_TUPLES_OK:
387 /* OK */
388 break;
389 default:
390 fprintf(stderr, "Client %d aborted in state %d: %s",
391 n, st->state, PQerrorMessage(st->con));
392 remains--; /* I've aborted */
393 PQfinish(st->con);
394 st->con = NULL;
395 return (-1);
396 }
397 return (0); /* OK */
398 }
399
400 static int
401 compareVariables(const void *v1, const void *v2)
402 {
403 return strcmp(((const Variable *) v1)->name,
404 ((const Variable *) v2)->name);
405 }
406
407 static char *
408 getVariable(CState * st, char *name)
409 {
410 Variable key,
411 *var;
412
413 /* On some versions of Solaris, bsearch of zero items dumps core */
414 if (st->nvariables <= 0)
415 return NULL;
416
417 key.name = name;
418 var = (Variable *) bsearch((void *) &key,
419 (void *) st->variables,
420 st->nvariables,
421 sizeof(Variable),
422 compareVariables);
423 if (var != NULL)
424 return var->value;
425 else
426 return NULL;
427 }
428
429 static int
430 putVariable(CState * st, char *name, char *value)
431 {
432 Variable key,
433 *var;
434
435 key.name = name;
436 /* On some versions of Solaris, bsearch of zero items dumps core */
437 if (st->nvariables > 0)
438 var = (Variable *) bsearch((void *) &key,
439 (void *) st->variables,
440 st->nvariables,
441 sizeof(Variable),
442 compareVariables);
443 else
444 var = NULL;
445
446 if (var == NULL)
447 {
448 Variable *newvars;
449
450 if (st->variables)
451 newvars = (Variable *) realloc(st->variables,
452 (st->nvariables + 1) * sizeof(Variable));
453 else
454 newvars = (Variable *) malloc(sizeof(Variable));
455
456 if (newvars == NULL)
457 return false;
458
459 st->variables = newvars;
460
461 var = &newvars[st->nvariables];
462
463 var->name = NULL;
464 var->value = NULL;
465
466 if ((var->name = strdup(name)) == NULL
467 || (var->value = strdup(value)) == NULL)
468 {
469 free(var->name);
470 free(var->value);
471 return false;
472 }
473
474 st->nvariables++;
475
476 qsort((void *) st->variables, st->nvariables, sizeof(Variable),
477 compareVariables);
478 }
479 else
480 {
481 char *val;
482
483 if ((val = strdup(value)) == NULL)
484 return false;
485
486 free(var->value);
487 var->value = val;
488 }
489
490 return true;
491 }
492
493 static char *
494 parseVariable(const char *sql, int *eaten)
495 {
496 int i = 0;
497 char *name;
498
499 do
500 {
501 i++;
502 } while (isalnum((unsigned char) sql[i]) || sql[i] == '_');
503 if (i == 1)
504 return NULL;
505
506 name = malloc(i);
507 if (name == NULL)
508 return NULL;
509 memcpy(name, &sql[1], i - 1);
510 name[i - 1] = '\0';
511
512 *eaten = i;
513 return name;
514 }
515
516 static char *
517 replaceVariable(char **sql, char *param, int len, char *value)
518 {
519 int valueln = strlen(value);
520
521 if (valueln > len)
522 {
523 char *tmp;
524 size_t offset = param - *sql;
525
526 tmp = realloc(*sql, strlen(*sql) - len + valueln + 1);
527 if (tmp == NULL)
528 {
529 free(*sql);
530 return NULL;
531 }
532 *sql = tmp;
533 param = *sql + offset;
534 }
535
536 if (valueln != len)
537 memmove(param + valueln, param + len, strlen(param + len) + 1);
538 strncpy(param, value, valueln);
539
540 return param + valueln;
541 }
542
543 static char *
544 assignVariables(CState * st, char *sql)
545 {
546 char *p,
547 *name,
548 *val;
549
550 p = sql;
551 while ((p = strchr(p, ':')) != NULL)
552 {
553 int eaten;
554
555 name = parseVariable(p, &eaten);
556 if (name == NULL)
557 {
558 while (*p == ':') { p++; }
559 continue;
560 }
561
562 val = getVariable(st, name);
563 free(name);
564 if (val == NULL)
565 {
566 p++;
567 continue;
568 }
569
570 if ((p = replaceVariable(&sql, p, eaten, val)) == NULL)
571 return NULL;
572 }
573
574 return sql;
575 }
576
577 static void
578 getQueryParams(CState *st, const Command *command, const char **params)
579 {
580 int i;
581
582 for (i = 0; i < command->argc - 1; i++)
583 params[i] = getVariable(st, command->argv[i+1]);
584 }
585
586 #define MAX_PREPARE_NAME 32
587 static void
588 preparedStatementName(char *buffer, int file, int state)
589 {
590 sprintf(buffer, "P%d_%d", file, state);
591 }
592
593 static void
594 doCustom(CState * state, int n, int debug)
595 {
596 PGresult *res;
597 CState *st = &state[n];
598 Command **commands;
599
600 top:
601 commands = sql_files[st->use_file];
602
603 if (st->sleeping)
604 { /* are we sleeping? */
605 int usec;
606 struct timeval now;
607
608 gettimeofday(&now, NULL);
609 usec = (st->until.tv_sec - now.tv_sec) * 1000000 +
610 st->until.tv_usec - now.tv_usec;
611 if (usec <= 0)
612 st->sleeping = 0; /* Done sleeping, go ahead with next command */
613 else
614 return; /* Still sleeping, nothing to do here */
615 }
616
617 if (st->listen)
618 { /* are we receiver? */
619 if (commands[st->state]->type == SQL_COMMAND)
620 {
621 if (debug)
622 fprintf(stderr, "client %d receiving\n", n);
623 if (!PQconsumeInput(st->con))
624 { /* there's something wrong */
625 fprintf(stderr, "Client %d aborted in state %d. Probably the backend died while processing.\n", n, st->state);
626 remains--; /* I've aborted */
627 PQfinish(st->con);
628 st->con = NULL;
629 return;
630 }
631 if (PQisBusy(st->con))
632 return; /* don't have the whole result yet */
633 }
634
635 /*
636 * transaction finished: record the time it took in the log
637 */
638 if (use_log && commands[st->state + 1] == NULL)
639 {
640 double diff;
641 struct timeval now;
642
643 gettimeofday(&now, NULL);
644 diff = (int) (now.tv_sec - st->txn_begin.tv_sec) * 1000000.0 +
645 (int) (now.tv_usec - st->txn_begin.tv_usec);
646
647 fprintf(LOGFILE, "%d %d %.0f %d %ld %ld\n",
648 st->id, st->cnt, diff, st->use_file,
649 (long) now.tv_sec, (long) now.tv_usec);
650 }
651
652 if (commands[st->state]->type == SQL_COMMAND)
653 {
654 res = PQgetResult(st->con);
655 if (check(state, res, n))
656 {
657 PQclear(res);
658 return;
659 }
660 PQclear(res);
661 discard_response(st);
662 }
663
664 if (commands[st->state + 1] == NULL)
665 {
666 if (is_connect)
667 {
668 PQfinish(st->con);
669 st->con = NULL;
670 }
671
672 ++st->cnt;
673 if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
674 {
675 remains--; /* I've done */
676 if (st->con != NULL)
677 {
678 PQfinish(st->con);
679 st->con = NULL;
680 }
681 return;
682 }
683 }
684
685 /* increment state counter */
686 st->state++;
687 if (commands[st->state] == NULL)
688 {
689 st->state = 0;
690 st->use_file = getrand(0, num_files - 1);
691 commands = sql_files[st->use_file];
692 }
693 }
694
695 if (st->con == NULL)
696 {
697 struct timeval t1, t2, t3;
698
699 gettimeofday(&t1, NULL);
700 if ((st->con = doConnect()) == NULL)
701 {
702 fprintf(stderr, "Client %d aborted in establishing connection.\n",
703 n);
704 remains--; /* I've aborted */
705 PQfinish(st->con);
706 st->con = NULL;
707 return;
708 }
709 gettimeofday(&t2, NULL);
710 diffTime(&t2, &t1, &t3);
711 addTime(&conn_total_time, &t3, &conn_total_time);
712 }
713
714 if (use_log && st->state == 0)
715 gettimeofday(&(st->txn_begin), NULL);
716
717 if (commands[st->state]->type == SQL_COMMAND)
718 {
719 const Command *command = commands[st->state];
720 int r;
721
722 if (querymode == QUERY_SIMPLE)
723 {
724 char *sql;
725
726 if ((sql = strdup(command->argv[0])) == NULL
727 || (sql = assignVariables(st, sql)) == NULL)
728 {
729 fprintf(stderr, "out of memory\n");
730 st->ecnt++;
731 return;
732 }
733
734 if (debug)
735 fprintf(stderr, "client %d sending %s\n", n, sql);
736 r = PQsendQuery(st->con, sql);
737 free(sql);
738 }
739 else if (querymode == QUERY_EXTENDED)
740 {
741 const char *sql = command->argv[0];
742 const char *params[MAX_ARGS];
743
744 getQueryParams(st, command, params);
745
746 if (debug)
747 fprintf(stderr, "client %d sending %s\n", n, sql);
748 r = PQsendQueryParams(st->con, sql, command->argc - 1,
749 NULL, params, NULL, NULL, 0);
750 }
751 else if (querymode == QUERY_PREPARED)
752 {
753 char name[MAX_PREPARE_NAME];
754 const char *params[MAX_ARGS];
755
756 if (!st->prepared[st->use_file])
757 {
758 int j;
759
760 for (j = 0; commands[j] != NULL; j++)
761 {
762 PGresult *res;
763 char name[MAX_PREPARE_NAME];
764
765 if (commands[j]->type != SQL_COMMAND)
766 continue;
767 preparedStatementName(name, st->use_file, j);
768 res = PQprepare(st->con, name,
769 commands[j]->argv[0], commands[j]->argc - 1, NULL);
770 if (PQresultStatus(res) != PGRES_COMMAND_OK)
771 fprintf(stderr, "%s", PQerrorMessage(st->con));
772 PQclear(res);
773 }
774 st->prepared[st->use_file] = true;
775 }
776
777 getQueryParams(st, command, params);
778 preparedStatementName(name, st->use_file, st->state);
779
780 if (debug)
781 fprintf(stderr, "client %d sending %s\n", n, name);
782 r = PQsendQueryPrepared(st->con, name, command->argc - 1,
783 params, NULL, NULL, 0);
784 }
785 else /* unknown sql mode */
786 r = 0;
787
788 if (r == 0)
789 {
790 if (debug)
791 fprintf(stderr, "client %d cannot send %s\n", n, command->argv[0]);
792 st->ecnt++;
793 }
794 else
795 st->listen = 1; /* flags that should be listened */
796 }
797 else if (commands[st->state]->type == META_COMMAND)
798 {
799 int argc = commands[st->state]->argc,
800 i;
801 char **argv = commands[st->state]->argv;
802
803 if (debug)
804 {
805 fprintf(stderr, "client %d executing \\%s", n, argv[0]);
806 for (i = 1; i < argc; i++)
807 fprintf(stderr, " %s", argv[i]);
808 fprintf(stderr, "\n");
809 }
810
811 if (pg_strcasecmp(argv[0], "setrandom") == 0)
812 {
813 char *var;
814 int min,
815 max;
816 char res[64];
817
818 if (*argv[2] == ':')
819 {
820 if ((var = getVariable(st, argv[2] + 1)) == NULL)
821 {
822 fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
823 st->ecnt++;
824 return;
825 }
826 min = atoi(var);
827 }
828 else
829 min = atoi(argv[2]);
830
831 #ifdef NOT_USED
832 if (min < 0)
833 {
834 fprintf(stderr, "%s: invalid minimum number %d\n", argv[0], min);
835 st->ecnt++;
836 return;
837 }
838 #endif
839
840 if (*argv[3] == ':')
841 {
842 if ((var = getVariable(st, argv[3] + 1)) == NULL)
843 {
844 fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[3]);
845 st->ecnt++;
846 return;
847 }
848 max = atoi(var);
849 }
850 else
851 max = atoi(argv[3]);
852
853 if (max < min || max > MAX_RANDOM_VALUE)
854 {
855 fprintf(stderr, "%s: invalid maximum number %d\n", argv[0], max);
856 st->ecnt++;
857 return;
858 }
859
860 #ifdef DEBUG
861 printf("min: %d max: %d random: %d\n", min, max, getrand(min, max));
862 #endif
863 snprintf(res, sizeof(res), "%d", getrand(min, max));
864
865 if (putVariable(st, argv[1], res) == false)
866 {
867 fprintf(stderr, "%s: out of memory\n", argv[0]);
868 st->ecnt++;
869 return;
870 }
871
872 st->listen = 1;
873 }
874 else if (pg_strcasecmp(argv[0], "set") == 0)
875 {
876 char *var;
877 int ope1,
878 ope2;
879 char res[64];
880
881 if (*argv[2] == ':')
882 {
883 if ((var = getVariable(st, argv[2] + 1)) == NULL)
884 {
885 fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
886 st->ecnt++;
887 return;
888 }
889 ope1 = atoi(var);
890 }
891 else
892 ope1 = atoi(argv[2]);
893
894 if (argc < 5)
895 snprintf(res, sizeof(res), "%d", ope1);
896 else
897 {
898 if (*argv[4] == ':')
899 {
900 if ((var = getVariable(st, argv[4] + 1)) == NULL)
901 {
902 fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[4]);
903 st->ecnt++;
904 return;
905 }
906 ope2 = atoi(var);
907 }
908 else
909 ope2 = atoi(argv[4]);
910
911 if (strcmp(argv[3], "+") == 0)
912 snprintf(res, sizeof(res), "%d", ope1 + ope2);
913 else if (strcmp(argv[3], "-") == 0)
914 snprintf(res, sizeof(res), "%d", ope1 - ope2);
915 else if (strcmp(argv[3], "*") == 0)
916 snprintf(res, sizeof(res), "%d", ope1 * ope2);
917 else if (strcmp(argv[3], "/") == 0)
918 {
919 if (ope2 == 0)
920 {
921 fprintf(stderr, "%s: division by zero\n", argv[0]);
922 st->ecnt++;
923 return;
924 }
925 snprintf(res, sizeof(res), "%d", ope1 / ope2);
926 }
927 else
928 {
929 fprintf(stderr, "%s: unsupported operator %s\n", argv[0], argv[3]);
930 st->ecnt++;
931 return;
932 }
933 }
934
935 if (putVariable(st, argv[1], res) == false)
936 {
937 fprintf(stderr, "%s: out of memory\n", argv[0]);
938 st->ecnt++;
939 return;
940 }
941
942 st->listen = 1;
943 }
944 else if (pg_strcasecmp(argv[0], "sleep") == 0)
945 {
946 char *var;
947 int usec;
948 struct timeval now;
949
950 if (*argv[1] == ':')
951 {
952 if ((var = getVariable(st, argv[1] + 1)) == NULL)
953 {
954 fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[1]);
955 st->ecnt++;
956 return;
957 }
958 usec = atoi(var);
959 }
960 else
961 usec = atoi(argv[1]);
962
963 if (argc > 2)
964 {
965 if (pg_strcasecmp(argv[2], "ms") == 0)
966 usec *= 1000;
967 else if (pg_strcasecmp(argv[2], "s") == 0)
968 usec *= 1000000;
969 }
970 else
971 usec *= 1000000;
972
973 gettimeofday(&now, NULL);
974 st->until.tv_sec = now.tv_sec + (now.tv_usec + usec) / 1000000;
975 st->until.tv_usec = (now.tv_usec + usec) % 1000000;
976 st->sleeping = 1;
977
978 st->listen = 1;
979 }
980
981 goto top;
982 }
983 }
984
985 /* discard connections */
986 static void
987 disconnect_all(CState * state)
988 {
989 int i;
990
991 for (i = 0; i < nclients; i++)
992 {
993 if (state[i].con)
994 PQfinish(state[i].con);
995 }
996 }
997
998 /* create tables and setup data */
999 static void
1000 init(void)
1001 {
1002 /*
1003 * Note: TPC-B requires at least 100 bytes per row, and the "filler"
1004 * fields in these table declarations were intended to comply with that.
1005 * But because they default to NULLs, they don't actually take any
1006 * space. We could fix that by giving them non-null default values.
1007 * However, that would completely break comparability of pgbench
1008 * results with prior versions. Since pgbench has never pretended
1009 * to be fully TPC-B compliant anyway, we stick with the historical
1010 * behavior.
1011 */
1012 static char *DDLs[] = {
1013 "drop table if exists pgbench_branches",
1014 "create table pgbench_branches(bid int not null,bbalance int,filler char(88)) with (fillfactor=%d)",
1015 "drop table if exists pgbench_tellers",
1016 "create table pgbench_tellers(tid int not null,bid int,tbalance int,filler char(84)) with (fillfactor=%d)",
1017 "drop table if exists pgbench_accounts",
1018 "create table pgbench_accounts(aid int not null,bid int,abalance int,filler char(84)) with (fillfactor=%d)",
1019 "drop table if exists pgbench_history",
1020 "create table pgbench_history(tid int,bid int,aid int,delta int,mtime timestamp,filler char(22))"
1021 };
1022 static char *DDLAFTERs[] = {
1023 "alter table pgbench_branches add primary key (bid)",
1024 "alter table pgbench_tellers add primary key (tid)",
1025 "alter table pgbench_accounts add primary key (aid)"
1026 };
1027
1028 PGconn *con;
1029 PGresult *res;
1030 char sql[256];
1031 int i;
1032
1033 if ((con = doConnect()) == NULL)
1034 exit(1);
1035
1036 for (i = 0; i < lengthof(DDLs); i++)
1037 {
1038 /*
1039 * set fillfactor for branches, tellers and accounts tables
1040 */
1041 if ((strstr(DDLs[i], "create table pgbench_branches") == DDLs[i]) ||
1042 (strstr(DDLs[i], "create table pgbench_tellers") == DDLs[i]) ||
1043 (strstr(DDLs[i], "create table pgbench_accounts") == DDLs[i]))
1044 {
1045 char ddl_stmt[128];
1046
1047 snprintf(ddl_stmt, 128, DDLs[i], fillfactor);
1048 executeStatement(con, ddl_stmt);
1049 continue;
1050 }
1051 else
1052 executeStatement(con, DDLs[i]);
1053 }
1054
1055 executeStatement(con, "begin");
1056
1057 for (i = 0; i < nbranches * scale; i++)
1058 {
1059 snprintf(sql, 256, "insert into pgbench_branches(bid,bbalance) values(%d,0)", i + 1);
1060 executeStatement(con, sql);
1061 }
1062
1063 for (i = 0; i < ntellers * scale; i++)
1064 {
1065 snprintf(sql, 256, "insert into pgbench_tellers(tid,bid,tbalance) values (%d,%d,0)"
1066 ,i + 1, i / ntellers + 1);
1067 executeStatement(con, sql);
1068 }
1069
1070 executeStatement(con, "commit");
1071
1072 /*
1073 * fill the pgbench_accounts table with some data
1074 */
1075 fprintf(stderr, "creating tables...\n");
1076
1077 executeStatement(con, "begin");
1078 executeStatement(con, "truncate pgbench_accounts");
1079
1080 res = PQexec(con, "copy pgbench_accounts from stdin");
1081 if (PQresultStatus(res) != PGRES_COPY_IN)
1082 {
1083 fprintf(stderr, "%s", PQerrorMessage(con));
1084 exit(1);
1085 }
1086 PQclear(res);
1087
1088 for (i = 0; i < naccounts * scale; i++)
1089 {
1090 int j = i + 1;
1091
1092 snprintf(sql, 256, "%d\t%d\t%d\t\n", j, i / naccounts + 1, 0);
1093 if (PQputline(con, sql))
1094 {
1095 fprintf(stderr, "PQputline failed\n");
1096 exit(1);
1097 }
1098
1099 if (j % 10000 == 0)
1100 fprintf(stderr, "%d tuples done.\n", j);
1101 }
1102 if (PQputline(con, "\\.\n"))
1103 {
1104 fprintf(stderr, "very last PQputline failed\n");
1105 exit(1);
1106 }
1107 if (PQendcopy(con))
1108 {
1109 fprintf(stderr, "PQendcopy failed\n");
1110 exit(1);
1111 }
1112 executeStatement(con, "commit");
1113
1114 /*
1115 * create indexes
1116 */
1117 fprintf(stderr, "set primary key...\n");
1118 for (i = 0; i < lengthof(DDLAFTERs); i++)
1119 executeStatement(con, DDLAFTERs[i]);
1120
1121 /* vacuum */
1122 fprintf(stderr, "vacuum...");
1123 executeStatement(con, "vacuum analyze pgbench_branches");
1124 executeStatement(con, "vacuum analyze pgbench_tellers");
1125 executeStatement(con, "vacuum analyze pgbench_accounts");
1126 executeStatement(con, "vacuum analyze pgbench_history");
1127
1128 fprintf(stderr, "done.\n");
1129 PQfinish(con);
1130 }
1131
1132 /*
1133 * Parse the raw sql and replace :param to $n.
1134 */
1135 static bool
1136 parseQuery(Command *cmd, const char *raw_sql)
1137 {
1138 char *sql,
1139 *p;
1140
1141 sql = strdup(raw_sql);
1142 if (sql == NULL)
1143 return false;
1144 cmd->argc = 1;
1145
1146 p = sql;
1147 while ((p = strchr(p, ':')) != NULL)
1148 {
1149 char var[12];
1150 char *name;
1151 int eaten;
1152
1153 name = parseVariable(p, &eaten);
1154 if (name == NULL)
1155 {
1156 while (*p == ':') { p++; }
1157 continue;
1158 }
1159
1160 if (cmd->argc >= MAX_ARGS)
1161 {
1162 fprintf(stderr, "statement has too many arguments (maximum is %d): %s\n", MAX_ARGS - 1, raw_sql);
1163 return false;
1164 }
1165
1166 sprintf(var, "$%d", cmd->argc);
1167 if ((p = replaceVariable(&sql, p, eaten, var)) == NULL)
1168 return false;
1169
1170 cmd->argv[cmd->argc] = name;
1171 cmd->argc++;
1172 }
1173
1174 cmd->argv[0] = sql;
1175 return true;
1176 }
1177
1178 static Command *
1179 process_commands(char *buf)
1180 {
1181 const char delim[] = " \f\n\r\t\v";
1182
1183 Command *my_commands;
1184 int j;
1185 char *p,
1186 *tok;
1187
1188 if ((p = strchr(buf, '\n')) != NULL)
1189 *p = '\0';
1190
1191 p = buf;
1192 while (isspace((unsigned char) *p))
1193 p++;
1194
1195 if (*p == '\0' || strncmp(p, "--", 2) == 0)
1196 {
1197 return NULL;
1198 }
1199
1200 my_commands = (Command *) malloc(sizeof(Command));
1201 if (my_commands == NULL)
1202 {
1203 return NULL;
1204 }
1205
1206 my_commands->argc = 0;
1207
1208 if (*p == '\\')
1209 {
1210 my_commands->type = META_COMMAND;
1211
1212 j = 0;
1213 tok = strtok(++p, delim);
1214
1215 while (tok != NULL)
1216 {
1217 if ((my_commands->argv[j] = strdup(tok)) == NULL)
1218 return NULL;
1219
1220 my_commands->argc++;
1221
1222 j++;
1223 tok = strtok(NULL, delim);
1224 }
1225
1226 if (pg_strcasecmp(my_commands->argv[0], "setrandom") == 0)
1227 {
1228 if (my_commands->argc < 4)
1229 {
1230 fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1231 return NULL;
1232 }
1233
1234 for (j = 4; j < my_commands->argc; j++)
1235 fprintf(stderr, "%s: extra argument \"%s\" ignored\n",
1236 my_commands->argv[0], my_commands->argv[j]);
1237 }
1238 else if (pg_strcasecmp(my_commands->argv[0], "set") == 0)
1239 {
1240 if (my_commands->argc < 3)
1241 {
1242 fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1243 return NULL;
1244 }
1245
1246 for (j = my_commands->argc < 5 ? 3 : 5; j < my_commands->argc; j++)
1247 fprintf(stderr, "%s: extra argument \"%s\" ignored\n",
1248 my_commands->argv[0], my_commands->argv[j]);
1249 }
1250 else if (pg_strcasecmp(my_commands->argv[0], "sleep") == 0)
1251 {
1252 if (my_commands->argc < 2)
1253 {
1254 fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1255 return NULL;
1256 }
1257
1258 if (my_commands->argc >= 3)
1259 {
1260 if (pg_strcasecmp(my_commands->argv[2], "us") != 0 &&
1261 pg_strcasecmp(my_commands->argv[2], "ms") != 0 &&
1262 pg_strcasecmp(my_commands->argv[2], "s"))
1263 {
1264 fprintf(stderr, "%s: unknown time unit '%s' - must be us, ms or s\n",
1265 my_commands->argv[0], my_commands->argv[2]);
1266 return NULL;
1267 }
1268 }
1269
1270 for (j = 3; j < my_commands->argc; j++)
1271 fprintf(stderr, "%s: extra argument \"%s\" ignored\n",
1272 my_commands->argv[0], my_commands->argv[j]);
1273 }
1274 else
1275 {
1276 fprintf(stderr, "Invalid command %s\n", my_commands->argv[0]);
1277 return NULL;
1278 }
1279 }
1280 else
1281 {
1282 my_commands->type = SQL_COMMAND;
1283
1284 switch (querymode)
1285 {
1286 case QUERY_SIMPLE:
1287 if ((my_commands->argv[0] = strdup(p)) == NULL)
1288 return NULL;
1289 my_commands->argc++;
1290 break;
1291 case QUERY_EXTENDED:
1292 case QUERY_PREPARED:
1293 if (!parseQuery(my_commands, p))
1294 return NULL;
1295 break;
1296 default:
1297 return NULL;
1298 }
1299 }
1300
1301 return my_commands;
1302 }
1303
1304 static int
1305 process_file(char *filename)
1306 {
1307 #define COMMANDS_ALLOC_NUM 128
1308
1309 Command **my_commands;
1310 FILE *fd;
1311 int lineno;
1312 char buf[BUFSIZ];
1313 int alloc_num;
1314
1315 if (num_files >= MAX_FILES)
1316 {
1317 fprintf(stderr, "Up to only %d SQL files are allowed\n", MAX_FILES);
1318 exit(1);
1319 }
1320
1321 alloc_num = COMMANDS_ALLOC_NUM;
1322 my_commands = (Command **) malloc(sizeof(Command *) * alloc_num);
1323 if (my_commands == NULL)
1324 return false;
1325
1326 if (strcmp(filename, "-") == 0)
1327 fd = stdin;
1328 else if ((fd = fopen(filename, "r")) == NULL)
1329 {
1330 fprintf(stderr, "%s: %s\n", filename, strerror(errno));
1331 return false;
1332 }
1333
1334 lineno = 0;
1335
1336 while (fgets(buf, sizeof(buf), fd) != NULL)
1337 {
1338 Command *commands;
1339 int i;
1340
1341 i = 0;
1342 while (isspace((unsigned char) buf[i]))
1343 i++;
1344
1345 if (buf[i] != '\0' && strncmp(&buf[i], "--", 2) != 0)
1346 {
1347 commands = process_commands(&buf[i]);
1348 if (commands == NULL)
1349 {
1350 fclose(fd);
1351 return false;
1352 }
1353 }
1354 else
1355 continue;
1356
1357 my_commands[lineno] = commands;
1358 lineno++;
1359
1360 if (lineno >= alloc_num)
1361 {
1362 alloc_num += COMMANDS_ALLOC_NUM;
1363 my_commands = realloc(my_commands, sizeof(Command *) * alloc_num);
1364 if (my_commands == NULL)
1365 {
1366 fclose(fd);
1367 return false;
1368 }
1369 }
1370 }
1371 fclose(fd);
1372
1373 my_commands[lineno] = NULL;
1374
1375 sql_files[num_files++] = my_commands;
1376
1377 return true;
1378 }
1379
1380 static Command **
1381 process_builtin(char *tb)
1382 {
1383 #define COMMANDS_ALLOC_NUM 128
1384
1385 Command **my_commands;
1386 int lineno;
1387 char buf[BUFSIZ];
1388 int alloc_num;
1389
1390 if (*tb == '\0')
1391 return NULL;
1392
1393 alloc_num = COMMANDS_ALLOC_NUM;
1394 my_commands = (Command **) malloc(sizeof(Command *) * alloc_num);
1395 if (my_commands == NULL)
1396 return NULL;
1397
1398 lineno = 0;
1399
1400 for (;;)
1401 {
1402 char *p;
1403 Command *commands;
1404
1405 p = buf;
1406 while (*tb && *tb != '\n')
1407 *p++ = *tb++;
1408
1409 if (*tb == '\0')
1410 break;
1411
1412 if (*tb == '\n')
1413 tb++;
1414
1415 *p = '\0';
1416
1417 commands = process_commands(buf);
1418 if (commands == NULL)
1419 {
1420 return NULL;
1421 }
1422
1423 my_commands[lineno] = commands;
1424 lineno++;
1425
1426 if (lineno >= alloc_num)
1427 {
1428 alloc_num += COMMANDS_ALLOC_NUM;
1429 my_commands = realloc(my_commands, sizeof(Command *) * alloc_num);
1430 if (my_commands == NULL)
1431 {
1432 return NULL;
1433 }
1434 }
1435 }
1436
1437 my_commands[lineno] = NULL;
1438
1439 return my_commands;
1440 }
1441
1442 /* print out results */
1443 static void
1444 printResults(
1445 int ttype, CState * state,
1446 struct timeval * start_time, struct timeval * end_time)
1447 {
1448 double t1,
1449 t2;
1450 int i;
1451 int normal_xacts = 0;
1452 char *s;
1453
1454 for (i = 0; i < nclients; i++)
1455 normal_xacts += state[i].cnt;
1456
1457 t1 = (end_time->tv_sec - start_time->tv_sec) * 1000000.0 + (end_time->tv_usec - start_time->tv_usec);
1458 t1 = normal_xacts * 1000000.0 / t1;
1459
1460 t2 = (end_time->tv_sec - start_time->tv_sec - conn_total_time.tv_sec) * 1000000.0 +
1461 (end_time->tv_usec - start_time->tv_usec - conn_total_time.tv_usec);
1462 t2 = normal_xacts * 1000000.0 / t2;
1463
1464 if (ttype == 0)
1465 s = "TPC-B (sort of)";
1466 else if (ttype == 2)
1467 s = "Update only pgbench_accounts";
1468 else if (ttype == 1)
1469 s = "SELECT only";
1470 else
1471 s = "Custom query";
1472
1473 printf("transaction type: %s\n", s);
1474 printf("scaling factor: %d\n", scale);
1475 printf("query mode: %s\n", QUERYMODE[querymode]);
1476 printf("number of clients: %d\n", nclients);
1477 if (duration <= 0)
1478 {
1479 printf("number of transactions per client: %d\n", nxacts);
1480 printf("number of transactions actually processed: %d/%d\n",
1481 normal_xacts, nxacts * nclients);
1482 }
1483 else
1484 {
1485 printf("duration: %d s\n", duration);
1486 printf("number of transactions actually processed: %d\n",
1487 normal_xacts);
1488 }
1489 printf("tps = %f (including connections establishing)\n", t1);
1490 printf("tps = %f (excluding connections establishing)\n", t2);
1491 }
1492
1493
1494 int
1495 main(int argc, char **argv)
1496 {
1497 int c;
1498 int is_init_mode = 0; /* initialize mode? */
1499 int is_no_vacuum = 0; /* no vacuum at all before testing? */
1500 int do_vacuum_accounts = 0; /* do vacuum accounts before testing? */
1501 int debug = 0; /* debug flag */
1502 int ttype = 0; /* transaction type. 0: TPC-B, 1: SELECT only,
1503 * 2: skip update of branches and tellers */
1504 char *filename = NULL;
1505 bool scale_given = false;
1506
1507 CState *state; /* status of clients */
1508
1509 struct timeval start_time; /* start up time */
1510 struct timeval end_time; /* end time */
1511
1512 int i;
1513
1514 fd_set input_mask;
1515 int nsocks; /* return from select(2) */
1516 int maxsock; /* max socket number to be waited */
1517 struct timeval now;
1518 struct timeval timeout;
1519 int min_usec;
1520
1521 #ifdef HAVE_GETRLIMIT
1522 struct rlimit rlim;
1523 #endif
1524
1525 PGconn *con;
1526 PGresult *res;
1527 char *env;
1528
1529 char val[64];
1530
1531 const char *progname;
1532
1533 progname = get_progname(argv[0]);
1534
1535 if (argc > 1)
1536 {
1537 if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
1538 {
1539 usage(progname);
1540 exit(0);
1541 }
1542 if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
1543 {
1544 puts("pgbench (PostgreSQL) " PG_VERSION);
1545 exit(0);
1546 }
1547 }
1548
1549 #ifdef WIN32
1550 /* stderr is buffered on Win32. */
1551 setvbuf(stderr, NULL, _IONBF, 0);
1552 #endif
1553
1554 if ((env = getenv("PGHOST")) != NULL && *env != '\0')
1555 pghost = env;
1556 if ((env = getenv("PGPORT")) != NULL && *env != '\0')
1557 pgport = env;
1558 else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
1559 login = env;
1560
1561 state = (CState *) malloc(sizeof(CState));
1562 if (state == NULL)
1563 {
1564 fprintf(stderr, "Couldn't allocate memory for state\n");
1565 exit(1);
1566 }
1567
1568 memset(state, 0, sizeof(*state));
1569
1570 while ((c = getopt(argc, argv, "ih:nvp:dSNc:Cs:t:T:U:lf:D:F:M:")) != -1)
1571 {
1572 switch (c)
1573 {
1574 case 'i':
1575 is_init_mode++;
1576 break;
1577 case 'h':
1578 pghost = optarg;
1579 break;
1580 case 'n':
1581 is_no_vacuum++;
1582 break;
1583 case 'v':
1584 do_vacuum_accounts++;
1585 break;
1586 case 'p':
1587 pgport = optarg;
1588 break;
1589 case 'd':
1590 debug++;
1591 break;
1592 case 'S':
1593 ttype = 1;
1594 break;
1595 case 'N':
1596 ttype = 2;
1597 break;
1598 case 'c':
1599 nclients = atoi(optarg);
1600 if (nclients <= 0 || nclients > MAXCLIENTS)
1601 {
1602 fprintf(stderr, "invalid number of clients: %d\n", nclients);
1603 exit(1);
1604 }
1605 #ifdef HAVE_GETRLIMIT
1606 #ifdef RLIMIT_NOFILE /* most platforms use RLIMIT_NOFILE */
1607 if (getrlimit(RLIMIT_NOFILE, &rlim) == -1)
1608 #else /* but BSD doesn't ... */
1609 if (getrlimit(RLIMIT_OFILE, &rlim) == -1)
1610 #endif /* RLIMIT_NOFILE */
1611 {
1612 fprintf(stderr, "getrlimit failed: %s\n", strerror(errno));
1613 exit(1);
1614 }
1615 if (rlim.rlim_cur <= (nclients + 2))
1616 {
1617 fprintf(stderr, "You need at least %d open files but you are only allowed to use %ld.\n", nclients + 2, (long) rlim.rlim_cur);
1618 fprintf(stderr, "Use limit/ulimit to increase the limit before using pgbench.\n");
1619 exit(1);
1620 }
1621 #endif /* HAVE_GETRLIMIT */
1622 break;
1623 case 'C':
1624 is_connect = 1;
1625 break;
1626 case 's':
1627 scale_given = true;
1628 scale = atoi(optarg);
1629 if (scale <= 0)
1630 {
1631 fprintf(stderr, "invalid scaling factor: %d\n", scale);
1632 exit(1);
1633 }
1634 break;
1635 case 't':
1636 if (duration > 0)
1637 {
1638 fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both.\n");
1639 exit(1);
1640 }
1641 nxacts = atoi(optarg);
1642 if (nxacts <= 0)
1643 {
1644 fprintf(stderr, "invalid number of transactions: %d\n", nxacts);
1645 exit(1);
1646 }
1647 break;
1648 case 'T':
1649 if (nxacts > 0)
1650 {
1651 fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both.\n");
1652 exit(1);
1653 }
1654 duration = atoi(optarg);
1655 if (duration <= 0)
1656 {
1657 fprintf(stderr, "invalid duration: %d\n", duration);
1658 exit(1);
1659 }
1660 break;
1661 case 'U':
1662 login = optarg;
1663 break;
1664 case 'l':
1665 use_log = true;
1666 break;
1667 case 'f':
1668 ttype = 3;
1669 filename = optarg;
1670 if (process_file(filename) == false || *sql_files[num_files - 1] == NULL)
1671 exit(1);
1672 break;
1673 case 'D':
1674 {
1675 char *p;
1676
1677 if ((p = strchr(optarg, '=')) == NULL || p == optarg || *(p + 1) == '\0')
1678 {
1679 fprintf(stderr, "invalid variable definition: %s\n", optarg);
1680 exit(1);
1681 }
1682
1683 *p++ = '\0';
1684 if (putVariable(&state[0], optarg, p) == false)
1685 {
1686 fprintf(stderr, "Couldn't allocate memory for variable\n");
1687 exit(1);
1688 }
1689 }
1690 break;
1691 case 'F':
1692 fillfactor = atoi(optarg);
1693 if ((fillfactor < 10) || (fillfactor > 100))
1694 {
1695 fprintf(stderr, "invalid fillfactor: %d\n", fillfactor);
1696 exit(1);
1697 }
1698 break;
1699 case 'M':
1700 if (num_files > 0)
1701 {
1702 fprintf(stderr, "query mode (-M) should be specifiled before transaction scripts (-f)\n");
1703 exit(1);
1704 }
1705 for (querymode = 0; querymode < NUM_QUERYMODE; querymode++)
1706 if (strcmp(optarg, QUERYMODE[querymode]) == 0)
1707 break;
1708 if (querymode >= NUM_QUERYMODE)
1709 {
1710 fprintf(stderr, "invalid query mode (-M): %s\n", optarg);
1711 exit(1);
1712 }
1713 break;
1714 default:
1715 fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
1716 exit(1);
1717 break;
1718 }
1719 }
1720
1721 if (argc > optind)
1722 dbName = argv[optind];
1723 else
1724 {
1725 if ((env = getenv("PGDATABASE")) != NULL && *env != '\0')
1726 dbName = env;
1727 else if (login != NULL && *login != '\0')
1728 dbName = login;
1729 else
1730 dbName = "";
1731 }
1732
1733 if (is_init_mode)
1734 {
1735 init();
1736 exit(0);
1737 }
1738
1739 /* Use DEFAULT_NXACTS if neither nxacts nor duration is specified. */
1740 if (nxacts <= 0 && duration <= 0)
1741 nxacts = DEFAULT_NXACTS;
1742
1743 remains = nclients;
1744
1745 if (nclients > 1)
1746 {
1747 state = (CState *) realloc(state, sizeof(CState) * nclients);
1748 if (state == NULL)
1749 {
1750 fprintf(stderr, "Couldn't allocate memory for state\n");
1751 exit(1);
1752 }
1753
1754 memset(state + 1, 0, sizeof(*state) * (nclients - 1));
1755
1756 /* copy any -D switch values to all clients */
1757 for (i = 1; i < nclients; i++)
1758 {
1759 int j;
1760
1761 for (j = 0; j < state[0].nvariables; j++)
1762 {
1763 if (putVariable(&state[i], state[0].variables[j].name, state[0].variables[j].value) == false)
1764 {
1765 fprintf(stderr, "Couldn't allocate memory for variable\n");
1766 exit(1);
1767 }
1768 }
1769 }
1770 }
1771
1772 if (use_log)
1773 {
1774 char logpath[64];
1775
1776 snprintf(logpath, 64, "pgbench_log.%d", (int) getpid());
1777 LOGFILE = fopen(logpath, "w");
1778
1779 if (LOGFILE == NULL)
1780 {
1781 fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
1782 exit(1);
1783 }
1784 }
1785
1786 if (debug)
1787 {
1788 if (duration <= 0)
1789 printf("pghost: %s pgport: %s nclients: %d nxacts: %d dbName: %s\n",
1790 pghost, pgport, nclients, nxacts, dbName);
1791 else
1792 printf("pghost: %s pgport: %s nclients: %d duration: %d dbName: %s\n",
1793 pghost, pgport, nclients, duration, dbName);
1794 }
1795
1796 /* opening connection... */
1797 con = doConnect();
1798 if (con == NULL)
1799 exit(1);
1800
1801 if (PQstatus(con) == CONNECTION_BAD)
1802 {
1803 fprintf(stderr, "Connection to database '%s' failed.\n", dbName);
1804 fprintf(stderr, "%s", PQerrorMessage(con));
1805 exit(1);
1806 }
1807
1808 if (ttype != 3)
1809 {
1810 /*
1811 * get the scaling factor that should be same as count(*) from
1812 * pgbench_branches if this is not a custom query
1813 */
1814 res = PQexec(con, "select count(*) from pgbench_branches");
1815 if (PQresultStatus(res) != PGRES_TUPLES_OK)
1816 {
1817 fprintf(stderr, "%s", PQerrorMessage(con));
1818 exit(1);
1819 }
1820 scale = atoi(PQgetvalue(res, 0, 0));
1821 if (scale < 0)
1822 {
1823 fprintf(stderr, "count(*) from pgbench_branches invalid (%d)\n", scale);
1824 exit(1);
1825 }
1826 PQclear(res);
1827
1828 /* warn if we override user-given -s switch */
1829 if (scale_given)
1830 fprintf(stderr,
1831 "Scale option ignored, using pgbench_branches table count = %d\n",
1832 scale);
1833 }
1834
1835 /*
1836 * :scale variables normally get -s or database scale, but don't override
1837 * an explicit -D switch
1838 */
1839 if (getVariable(&state[0], "scale") == NULL)
1840 {
1841 snprintf(val, sizeof(val), "%d", scale);
1842 for (i = 0; i < nclients; i++)
1843 {
1844 if (putVariable(&state[i], "scale", val) == false)
1845 {
1846 fprintf(stderr, "Couldn't allocate memory for variable\n");
1847 exit(1);
1848 }
1849 }
1850 }
1851
1852 if (!is_no_vacuum)
1853 {
1854 fprintf(stderr, "starting vacuum...");
1855 executeStatement(con, "vacuum pgbench_branches");
1856 executeStatement(con, "vacuum pgbench_tellers");
1857 executeStatement(con, "truncate pgbench_history");
1858 fprintf(stderr, "end.\n");
1859
1860 if (do_vacuum_accounts)
1861 {
1862 fprintf(stderr, "starting vacuum pgbench_accounts...");
1863 executeStatement(con, "vacuum analyze pgbench_accounts");
1864 fprintf(stderr, "end.\n");
1865 }
1866 }
1867 PQfinish(con);
1868
1869 /* set random seed */
1870 gettimeofday(&start_time, NULL);
1871 srandom((unsigned int) start_time.tv_usec);
1872
1873 /* get start up time */
1874 gettimeofday(&start_time, NULL);
1875
1876 /* set alarm if duration is specified. */
1877 if (duration > 0)
1878 setalarm(duration);
1879
1880 if (is_connect == 0)
1881 {
1882 struct timeval t, now;
1883
1884 /* make connections to the database */
1885 for (i = 0; i < nclients; i++)
1886 {
1887 state[i].id = i;
1888 if ((state[i].con = doConnect()) == NULL)
1889 exit(1);
1890 }
1891 /* time after connections set up */
1892 gettimeofday(&now, NULL);
1893 diffTime(&now, &start_time, &t);
1894 addTime(&conn_total_time, &t, &conn_total_time);
1895 }
1896
1897 /* process bultin SQL scripts */
1898 switch (ttype)
1899 {
1900 case 0:
1901 sql_files[0] = process_builtin(tpc_b);
1902 num_files = 1;
1903 break;
1904
1905 case 1:
1906 sql_files[0] = process_builtin(select_only);
1907 num_files = 1;
1908 break;
1909
1910 case 2:
1911 sql_files[0] = process_builtin(simple_update);
1912 num_files = 1;
1913 break;
1914
1915 default:
1916 break;
1917 }
1918
1919 /* send start up queries in async manner */
1920 for (i = 0; i < nclients; i++)
1921 {
1922 Command **commands = sql_files[state[i].use_file];
1923 int prev_ecnt = state[i].ecnt;
1924
1925 state[i].use_file = getrand(0, num_files - 1);
1926 doCustom(state, i, debug);
1927
1928 if (state[i].ecnt > prev_ecnt && commands[state[i].state]->type == META_COMMAND)
1929 {
1930 fprintf(stderr, "Client %d aborted in state %d. Execution meta-command failed.\n", i, state[i].state);
1931 remains--; /* I've aborted */
1932 PQfinish(state[i].con);
1933 state[i].con = NULL;
1934 }
1935 }
1936
1937 for (;;)
1938 {
1939 if (remains <= 0)
1940 { /* all done ? */
1941 disconnect_all(state);
1942 /* get end time */
1943 gettimeofday(&end_time, NULL);
1944 printResults(ttype, state, &start_time, &end_time);
1945 if (LOGFILE)
1946 fclose(LOGFILE);
1947 exit(0);
1948 }
1949
1950 FD_ZERO(&input_mask);
1951
1952 maxsock = -1;
1953 min_usec = -1;
1954 for (i = 0; i < nclients; i++)
1955 {
1956 Command **commands = sql_files[state[i].use_file];
1957
1958 if (state[i].sleeping)
1959 {
1960 int this_usec;
1961 int sock = PQsocket(state[i].con);
1962
1963 if (min_usec < 0)
1964 {
1965 gettimeofday(&now, NULL);
1966 min_usec = 0;
1967 }
1968
1969 this_usec = (state[i].until.tv_sec - now.tv_sec) * 1000000 +
1970 state[i].until.tv_usec - now.tv_usec;
1971
1972 if (this_usec > 0 && (min_usec == 0 || this_usec < min_usec))
1973 min_usec = this_usec;
1974
1975 FD_SET(sock, &input_mask);
1976 if (maxsock < sock)
1977 maxsock = sock;
1978 }
1979 else if (state[i].con && commands[state[i].state]->type != META_COMMAND)
1980 {
1981 int sock = PQsocket(state[i].con);
1982
1983 if (sock < 0)
1984 {
1985 disconnect_all(state);
1986 exit(1);
1987 }
1988 FD_SET(sock, &input_mask);
1989 if (maxsock < sock)
1990 maxsock = sock;
1991 }
1992 }
1993
1994 if (maxsock != -1)
1995 {
1996 if (min_usec >= 0)
1997 {
1998 timeout.tv_sec = min_usec / 1000000;
1999 timeout.tv_usec = min_usec % 1000000;
2000
2001 nsocks = select(maxsock + 1, &input_mask, (fd_set *) NULL,
2002 (fd_set *) NULL, &timeout);
2003 }
2004 else
2005 nsocks = select(maxsock + 1, &input_mask, (fd_set *) NULL,
2006 (fd_set *) NULL, (struct timeval *) NULL);
2007 if (nsocks < 0)
2008 {
2009 if (errno == EINTR)
2010 continue;
2011 /* must be something wrong */
2012 disconnect_all(state);
2013 fprintf(stderr, "select failed: %s\n", strerror(errno));
2014 exit(1);
2015 }
2016 #ifdef NOT_USED
2017 else if (nsocks == 0)
2018 { /* timeout */
2019 fprintf(stderr, "select timeout\n");
2020 for (i = 0; i < nclients; i++)
2021 {
2022 fprintf(stderr, "client %d:state %d cnt %d ecnt %d listen %d\n",
2023 i, state[i].state, state[i].cnt, state[i].ecnt, state[i].listen);
2024 }
2025 exit(0);
2026 }
2027 #endif
2028 }
2029
2030 /* ok, backend returns reply */
2031 for (i = 0; i < nclients; i++)
2032 {
2033 Command **commands = sql_files[state[i].use_file];
2034 int prev_ecnt = state[i].ecnt;
2035
2036 if (state[i].con && (FD_ISSET(PQsocket(state[i].con), &input_mask)
2037 || commands[state[i].state]->type == META_COMMAND))
2038 {
2039 doCustom(state, i, debug);
2040 }
2041
2042 if (state[i].ecnt > prev_ecnt && commands[state[i].state]->type == META_COMMAND)
2043 {
2044 fprintf(stderr, "Client %d aborted in state %d. Execution of meta-command failed.\n", i, state[i].state);
2045 remains--; /* I've aborted */
2046 PQfinish(state[i].con);
2047 state[i].con = NULL;
2048 }
2049 }
2050 }
2051 }
2052
2053
2054 /*
2055 * Support for duration option: set timer_exceeded after so many seconds.
2056 */
2057
2058 #ifndef WIN32
2059
2060 static void
2061 handle_sig_alarm(SIGNAL_ARGS)
2062 {
2063 timer_exceeded = true;
2064 }
2065
2066 static void
2067 setalarm(int seconds)
2068 {
2069 pqsignal(SIGALRM, handle_sig_alarm);
2070 alarm(seconds);
2071 }
2072
2073 #else /* WIN32 */
2074
2075 static VOID CALLBACK
2076 win32_timer_callback(PVOID lpParameter, BOOLEAN TimerOrWaitFired)
2077 {
2078 timer_exceeded = true;
2079 }
2080
2081 static void
2082 setalarm(int seconds)
2083 {
2084 HANDLE queue;
2085 HANDLE timer;
2086
2087 /* This function will be called at most once, so we can cheat a bit. */
2088 queue = CreateTimerQueue();
2089 if (seconds > ((DWORD)-1) / 1000 ||
2090 !CreateTimerQueueTimer(&timer, queue,
2091 win32_timer_callback, NULL, seconds * 1000, 0,
2092 WT_EXECUTEINTIMERTHREAD | WT_EXECUTEONLYONCE))
2093 {
2094 fprintf(stderr, "Failed to set timer\n");
2095 exit(1);
2096 }
2097 }
2098
2099 #endif /* WIN32 */
PostgreSQL from CVS