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