4
* A simple benchmark program for PostgreSQL
5
* Originally written by Tatsuo Ishii and enhanced by many contributors.
7
* contrib/pgbench/pgbench.c
8
* Copyright (c) 2000-2011, PostgreSQL Global Development Group
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.
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.
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.
31
#define FD_SETSIZE 1024 /* set before winsock2.h is included */
34
#include "postgres_fe.h"
37
#include "libpq/pqsignal.h"
38
#include "portability/instr_time.h"
51
#ifdef HAVE_SYS_SELECT_H
52
#include <sys/select.h>
55
#ifdef HAVE_SYS_RESOURCE_H
56
#include <sys/resource.h> /* for getrlimit */
60
#define INT64_MAX INT64CONST(0x7FFFFFFFFFFFFFFF)
64
* Multi-platform pthread implementations
68
/* Use native win32 threads on Windows */
69
typedef struct win32_pthread *pthread_t;
70
typedef int pthread_attr_t;
72
static int pthread_create(pthread_t *thread, pthread_attr_t * attr, void *(*start_routine) (void *), void *arg);
73
static int pthread_join(pthread_t th, void **thread_return);
74
#elif defined(ENABLE_THREAD_SAFETY)
75
/* Use platform-dependent pthread capability */
78
/* Use emulation with fork. Rename pthread identifiers to avoid conflicts */
82
#define pthread_t pg_pthread_t
83
#define pthread_attr_t pg_pthread_attr_t
84
#define pthread_create pg_pthread_create
85
#define pthread_join pg_pthread_join
87
typedef struct fork_pthread *pthread_t;
88
typedef int pthread_attr_t;
90
static int pthread_create(pthread_t *thread, pthread_attr_t * attr, void *(*start_routine) (void *), void *arg);
91
static int pthread_join(pthread_t th, void **thread_return);
98
/********************************************************************
99
* some configurable parameters */
101
/* max number of clients allowed */
103
#define MAXCLIENTS (FD_SETSIZE - 10)
105
#define MAXCLIENTS 1024
108
#define DEFAULT_NXACTS 10 /* default nxacts */
110
int nxacts = 0; /* number of transactions per client */
111
int duration = 0; /* duration in seconds */
114
* scaling factor. for example, scale = 10 will make 1000000 tuples in
115
* pgbench_accounts table.
120
* fillfactor. for example, fillfactor = 90 will use only 90 percent
121
* space during inserts and leave 10 percent free.
123
int fillfactor = 100;
126
* end of configurable parameters
127
*********************************************************************/
129
#define nbranches 1 /* Makes little sense to change this. Change
132
#define naccounts 100000
134
bool use_log; /* log transaction latencies to a file */
135
bool is_connect; /* establish connection for each transaction */
136
bool is_latencies; /* report per-command latencies */
137
int main_pid; /* main process id used in log filename */
141
char *pgoptions = NULL;
146
volatile bool timer_exceeded = false; /* flag from signal handler */
148
/* variable definitions */
151
char *name; /* variable name */
152
char *value; /* its value */
155
#define MAX_FILES 128 /* max number of SQL script files allowed */
156
#define SHELL_COMMAND_SIZE 256 /* maximum size allowed for shell command */
159
* structures used in custom query mode
164
PGconn *con; /* connection handle to DB */
165
int id; /* client No. */
166
int state; /* state No. */
167
int cnt; /* xacts count */
168
int ecnt; /* error count */
169
int listen; /* 0 indicates that an async query has been
171
int sleeping; /* 1 indicates that the client is napping */
172
int64 until; /* napping until (usec) */
173
Variable *variables; /* array of variable definitions */
175
instr_time txn_begin; /* used for measuring transaction latencies */
176
instr_time stmt_begin; /* used for measuring statement latencies */
177
int use_file; /* index in sql_files for this client */
178
bool prepared[MAX_FILES];
182
* Thread state and result
186
int tid; /* thread id */
187
pthread_t thread; /* thread handle */
188
CState *state; /* array of CState */
189
int nstate; /* length of state[] */
190
instr_time start_time; /* thread start time */
191
instr_time *exec_elapsed; /* time spent executing cmds (per Command) */
192
int *exec_count; /* number of cmd executions (per Command) */
195
#define INVALID_THREAD ((pthread_t) 0)
199
instr_time conn_time;
204
* queries read from files
206
#define SQL_COMMAND 1
207
#define META_COMMAND 2
210
typedef enum QueryMode
212
QUERY_SIMPLE, /* simple query */
213
QUERY_EXTENDED, /* extended query */
214
QUERY_PREPARED, /* extended query with prepared statements */
218
static QueryMode querymode = QUERY_SIMPLE;
219
static const char *QUERYMODE[] = {"simple", "extended", "prepared"};
223
char *line; /* full text of command line */
224
int command_num; /* unique index of this Command struct */
225
int type; /* command type (SQL_COMMAND or META_COMMAND) */
226
int argc; /* number of command words */
227
char *argv[MAX_ARGS]; /* command word list */
230
static Command **sql_files[MAX_FILES]; /* SQL script files */
231
static int num_files; /* number of script files */
232
static int num_commands = 0; /* total number of Command structs */
233
static int debug = 0; /* debug flag */
235
/* default scenario */
236
static char *tpc_b = {
237
"\\set nbranches " CppAsString2(nbranches) " * :scale\n"
238
"\\set ntellers " CppAsString2(ntellers) " * :scale\n"
239
"\\set naccounts " CppAsString2(naccounts) " * :scale\n"
240
"\\setrandom aid 1 :naccounts\n"
241
"\\setrandom bid 1 :nbranches\n"
242
"\\setrandom tid 1 :ntellers\n"
243
"\\setrandom delta -5000 5000\n"
245
"UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
246
"SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
247
"UPDATE pgbench_tellers SET tbalance = tbalance + :delta WHERE tid = :tid;\n"
248
"UPDATE pgbench_branches SET bbalance = bbalance + :delta WHERE bid = :bid;\n"
249
"INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
254
static char *simple_update = {
255
"\\set nbranches " CppAsString2(nbranches) " * :scale\n"
256
"\\set ntellers " CppAsString2(ntellers) " * :scale\n"
257
"\\set naccounts " CppAsString2(naccounts) " * :scale\n"
258
"\\setrandom aid 1 :naccounts\n"
259
"\\setrandom bid 1 :nbranches\n"
260
"\\setrandom tid 1 :ntellers\n"
261
"\\setrandom delta -5000 5000\n"
263
"UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid;\n"
264
"SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
265
"INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) VALUES (:tid, :bid, :aid, :delta, CURRENT_TIMESTAMP);\n"
270
static char *select_only = {
271
"\\set naccounts " CppAsString2(naccounts) " * :scale\n"
272
"\\setrandom aid 1 :naccounts\n"
273
"SELECT abalance FROM pgbench_accounts WHERE aid = :aid;\n"
276
/* Function prototypes */
277
static void setalarm(int seconds);
278
static void *threadRun(void *arg);
282
* routines to check mem allocations and fail noisily.
289
result = malloc(size);
292
fprintf(stderr, "out of memory\n");
299
xrealloc(void *ptr, size_t size)
303
result = realloc(ptr, size);
306
fprintf(stderr, "out of memory\n");
313
xstrdup(const char *s)
320
fprintf(stderr, "out of memory\n");
328
usage(const char *progname)
330
printf("%s is a benchmarking tool for PostgreSQL.\n\n"
332
" %s [OPTIONS]... [DBNAME]\n"
333
"\nInitialization options:\n"
334
" -i invokes initialization mode\n"
335
" -F NUM fill factor\n"
336
" -s NUM scaling factor\n"
337
"\nBenchmarking options:\n"
338
" -c NUM number of concurrent database clients (default: 1)\n"
339
" -C establish new connection for each transaction\n"
340
" -D VARNAME=VALUE\n"
341
" define variable for use by custom script\n"
342
" -f FILENAME read transaction script from FILENAME\n"
343
" -j NUM number of threads (default: 1)\n"
344
" -r report average latency per command\n"
345
" -l write transaction times to log file\n"
346
" -M {simple|extended|prepared}\n"
347
" protocol for submitting queries to server (default: simple)\n"
348
" -n do not run VACUUM before tests\n"
349
" -N do not update tables \"pgbench_tellers\" and \"pgbench_branches\"\n"
350
" -s NUM report this scale factor in output\n"
351
" -S perform SELECT-only transactions\n"
352
" -t NUM number of transactions each client runs (default: 10)\n"
353
" -T NUM duration of benchmark test in seconds\n"
354
" -v vacuum all four standard tables before tests\n"
355
"\nCommon options:\n"
356
" -d print debugging output\n"
357
" -h HOSTNAME database server host or socket directory\n"
358
" -p PORT database server port number\n"
359
" -U USERNAME connect as specified database user\n"
360
" --help show this help, then exit\n"
361
" --version output version information, then exit\n"
363
"Report bugs to <pgsql-bugs@postgresql.org>.\n",
367
/* random number generator: uniform distribution from min to max inclusive */
369
getrand(int min, int max)
372
* Odd coding is so that min and max have approximately the same chance of
373
* being selected as do numbers between them.
375
return min + (int) (((max - min + 1) * (double) random()) / (MAX_RANDOM_VALUE + 1.0));
378
/* call PQexec() and exit() on failure */
380
executeStatement(PGconn *con, const char *sql)
384
res = PQexec(con, sql);
385
if (PQresultStatus(res) != PGRES_COMMAND_OK)
387
fprintf(stderr, "%s", PQerrorMessage(con));
393
/* set up a connection to the backend */
398
static char *password = NULL;
402
* Start the connection. Loop until we have a password if requested by
409
conn = PQsetdbLogin(pghost, pgport, pgoptions, pgtty, dbName,
413
fprintf(stderr, "Connection to database \"%s\" failed\n",
418
if (PQstatus(conn) == CONNECTION_BAD &&
419
PQconnectionNeedsPassword(conn) &&
423
password = simple_prompt("Password: ", 100, false);
428
/* check to see that the backend connection was successfully made */
429
if (PQstatus(conn) == CONNECTION_BAD)
431
fprintf(stderr, "Connection to database \"%s\" failed:\n%s",
432
dbName, PQerrorMessage(conn));
440
/* throw away response from backend */
442
discard_response(CState *state)
448
res = PQgetResult(state->con);
455
compareVariables(const void *v1, const void *v2)
457
return strcmp(((const Variable *) v1)->name,
458
((const Variable *) v2)->name);
462
getVariable(CState *st, char *name)
467
/* On some versions of Solaris, bsearch of zero items dumps core */
468
if (st->nvariables <= 0)
472
var = (Variable *) bsearch((void *) &key,
473
(void *) st->variables,
483
/* check whether the name consists of alphabets, numerals and underscores. */
485
isLegalVariableName(const char *name)
489
for (i = 0; name[i] != '\0'; i++)
491
if (!isalnum((unsigned char) name[i]) && name[i] != '_')
499
putVariable(CState *st, const char *context, char *name, char *value)
505
/* On some versions of Solaris, bsearch of zero items dumps core */
506
if (st->nvariables > 0)
507
var = (Variable *) bsearch((void *) &key,
508
(void *) st->variables,
520
* Check for the name only when declaring a new variable to avoid
523
if (!isLegalVariableName(name))
525
fprintf(stderr, "%s: invalid variable name '%s'\n", context, name);
530
newvars = (Variable *) xrealloc(st->variables,
531
(st->nvariables + 1) * sizeof(Variable));
533
newvars = (Variable *) xmalloc(sizeof(Variable));
535
st->variables = newvars;
537
var = &newvars[st->nvariables];
539
var->name = xstrdup(name);
540
var->value = xstrdup(value);
544
qsort((void *) st->variables, st->nvariables, sizeof(Variable),
551
/* dup then free, in case value is pointing at this variable */
552
val = xstrdup(value);
562
parseVariable(const char *sql, int *eaten)
570
} while (isalnum((unsigned char) sql[i]) || sql[i] == '_');
575
memcpy(name, &sql[1], i - 1);
583
replaceVariable(char **sql, char *param, int len, char *value)
585
int valueln = strlen(value);
589
size_t offset = param - *sql;
591
*sql = xrealloc(*sql, strlen(*sql) - len + valueln + 1);
592
param = *sql + offset;
596
memmove(param + valueln, param + len, strlen(param + len) + 1);
597
strncpy(param, value, valueln);
599
return param + valueln;
603
assignVariables(CState *st, char *sql)
610
while ((p = strchr(p, ':')) != NULL)
614
name = parseVariable(p, &eaten);
624
val = getVariable(st, name);
632
p = replaceVariable(&sql, p, eaten, val);
639
getQueryParams(CState *st, const Command *command, const char **params)
643
for (i = 0; i < command->argc - 1; i++)
644
params[i] = getVariable(st, command->argv[i + 1]);
648
* Run a shell command. The result is assigned to the variable if not NULL.
649
* Return true if succeeded, or false on error.
652
runShellCommand(CState *st, char *variable, char **argv, int argc)
654
char command[SHELL_COMMAND_SIZE];
663
* Join arguments with whitespace separators. Arguments starting with
664
* exactly one colon are treated as variables:
665
* name - append a string "name"
666
* :var - append a variable named 'var'
667
* ::name - append a string ":name"
670
for (i = 0; i < argc; i++)
675
if (argv[i][0] != ':')
677
arg = argv[i]; /* a string literal */
679
else if (argv[i][1] == ':')
681
arg = argv[i] + 1; /* a string literal starting with colons */
683
else if ((arg = getVariable(st, argv[i] + 1)) == NULL)
685
fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[i]);
689
arglen = strlen(arg);
690
if (len + arglen + (i > 0 ? 1 : 0) >= SHELL_COMMAND_SIZE - 1)
692
fprintf(stderr, "%s: too long shell command\n", argv[0]);
697
command[len++] = ' ';
698
memcpy(command + len, arg, arglen);
704
/* Fast path for non-assignment case */
705
if (variable == NULL)
710
fprintf(stderr, "%s: cannot launch shell command\n", argv[0]);
716
/* Execute the command with pipe and read the standard output. */
717
if ((fp = popen(command, "r")) == NULL)
719
fprintf(stderr, "%s: cannot launch shell command\n", argv[0]);
722
if (fgets(res, sizeof(res), fp) == NULL)
725
fprintf(stderr, "%s: cannot read the result\n", argv[0]);
730
fprintf(stderr, "%s: cannot close shell command\n", argv[0]);
734
/* Check whether the result is an integer and assign it to the variable */
735
retval = (int) strtol(res, &endptr, 10);
736
while (*endptr != '\0' && isspace((unsigned char) *endptr))
738
if (*res == '\0' || *endptr != '\0')
740
fprintf(stderr, "%s: must return an integer ('%s' returned)\n", argv[0], res);
743
snprintf(res, sizeof(res), "%d", retval);
744
if (!putVariable(st, "setshell", variable, res))
748
printf("shell parameter name: %s, value: %s\n", argv[1], res);
753
#define MAX_PREPARE_NAME 32
755
preparedStatementName(char *buffer, int file, int state)
757
sprintf(buffer, "P%d_%d", file, state);
761
clientDone(CState *st, bool ok)
763
(void) ok; /* unused */
770
return false; /* always false */
773
/* return false iff client should be disconnected */
775
doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile)
781
commands = sql_files[st->use_file];
784
{ /* are we sleeping? */
787
INSTR_TIME_SET_CURRENT(now);
788
if (st->until <= INSTR_TIME_GET_MICROSEC(now))
789
st->sleeping = 0; /* Done sleeping, go ahead with next command */
791
return true; /* Still sleeping, nothing to do here */
795
{ /* are we receiver? */
796
if (commands[st->state]->type == SQL_COMMAND)
799
fprintf(stderr, "client %d receiving\n", st->id);
800
if (!PQconsumeInput(st->con))
801
{ /* there's something wrong */
802
fprintf(stderr, "Client %d aborted in state %d. Probably the backend died while processing.\n", st->id, st->state);
803
return clientDone(st, false);
805
if (PQisBusy(st->con))
806
return true; /* don't have the whole result yet */
810
* command finished: accumulate per-command execution times in
811
* thread-local data structure, if per-command latencies are requested
816
int cnum = commands[st->state]->command_num;
818
INSTR_TIME_SET_CURRENT(now);
819
INSTR_TIME_ACCUM_DIFF(thread->exec_elapsed[cnum],
820
now, st->stmt_begin);
821
thread->exec_count[cnum]++;
825
* if transaction finished, record the time it took in the log
827
if (logfile && commands[st->state + 1] == NULL)
833
INSTR_TIME_SET_CURRENT(now);
835
INSTR_TIME_SUBTRACT(diff, st->txn_begin);
836
usec = (double) INSTR_TIME_GET_MICROSEC(diff);
839
/* This is more than we really ought to know about instr_time */
840
fprintf(logfile, "%d %d %.0f %d %ld %ld\n",
841
st->id, st->cnt, usec, st->use_file,
842
(long) now.tv_sec, (long) now.tv_usec);
844
/* On Windows, instr_time doesn't provide a timestamp anyway */
845
fprintf(logfile, "%d %d %.0f %d 0 0\n",
846
st->id, st->cnt, usec, st->use_file);
850
if (commands[st->state]->type == SQL_COMMAND)
853
* Read and discard the query result; note this is not included in
854
* the statement latency numbers.
856
res = PQgetResult(st->con);
857
switch (PQresultStatus(res))
859
case PGRES_COMMAND_OK:
860
case PGRES_TUPLES_OK:
863
fprintf(stderr, "Client %d aborted in state %d: %s",
864
st->id, st->state, PQerrorMessage(st->con));
866
return clientDone(st, false);
869
discard_response(st);
872
if (commands[st->state + 1] == NULL)
881
if ((st->cnt >= nxacts && duration <= 0) || timer_exceeded)
882
return clientDone(st, true); /* exit success */
885
/* increment state counter */
887
if (commands[st->state] == NULL)
890
st->use_file = getrand(0, num_files - 1);
891
commands = sql_files[st->use_file];
900
INSTR_TIME_SET_CURRENT(start);
901
if ((st->con = doConnect()) == NULL)
903
fprintf(stderr, "Client %d aborted in establishing connection.\n", st->id);
904
return clientDone(st, false);
906
INSTR_TIME_SET_CURRENT(end);
907
INSTR_TIME_ACCUM_DIFF(*conn_time, end, start);
910
/* Record transaction start time if logging is enabled */
911
if (logfile && st->state == 0)
912
INSTR_TIME_SET_CURRENT(st->txn_begin);
914
/* Record statement start time if per-command latencies are requested */
916
INSTR_TIME_SET_CURRENT(st->stmt_begin);
918
if (commands[st->state]->type == SQL_COMMAND)
920
const Command *command = commands[st->state];
923
if (querymode == QUERY_SIMPLE)
927
sql = xstrdup(command->argv[0]);
928
sql = assignVariables(st, sql);
931
fprintf(stderr, "client %d sending %s\n", st->id, sql);
932
r = PQsendQuery(st->con, sql);
935
else if (querymode == QUERY_EXTENDED)
937
const char *sql = command->argv[0];
938
const char *params[MAX_ARGS];
940
getQueryParams(st, command, params);
943
fprintf(stderr, "client %d sending %s\n", st->id, sql);
944
r = PQsendQueryParams(st->con, sql, command->argc - 1,
945
NULL, params, NULL, NULL, 0);
947
else if (querymode == QUERY_PREPARED)
949
char name[MAX_PREPARE_NAME];
950
const char *params[MAX_ARGS];
952
if (!st->prepared[st->use_file])
956
for (j = 0; commands[j] != NULL; j++)
959
char name[MAX_PREPARE_NAME];
961
if (commands[j]->type != SQL_COMMAND)
963
preparedStatementName(name, st->use_file, j);
964
res = PQprepare(st->con, name,
965
commands[j]->argv[0], commands[j]->argc - 1, NULL);
966
if (PQresultStatus(res) != PGRES_COMMAND_OK)
967
fprintf(stderr, "%s", PQerrorMessage(st->con));
970
st->prepared[st->use_file] = true;
973
getQueryParams(st, command, params);
974
preparedStatementName(name, st->use_file, st->state);
977
fprintf(stderr, "client %d sending %s\n", st->id, name);
978
r = PQsendQueryPrepared(st->con, name, command->argc - 1,
979
params, NULL, NULL, 0);
981
else /* unknown sql mode */
987
fprintf(stderr, "client %d cannot send %s\n", st->id, command->argv[0]);
991
st->listen = 1; /* flags that should be listened */
993
else if (commands[st->state]->type == META_COMMAND)
995
int argc = commands[st->state]->argc,
997
char **argv = commands[st->state]->argv;
1001
fprintf(stderr, "client %d executing \\%s", st->id, argv[0]);
1002
for (i = 1; i < argc; i++)
1003
fprintf(stderr, " %s", argv[i]);
1004
fprintf(stderr, "\n");
1007
if (pg_strcasecmp(argv[0], "setrandom") == 0)
1014
if (*argv[2] == ':')
1016
if ((var = getVariable(st, argv[2] + 1)) == NULL)
1018
fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
1025
min = atoi(argv[2]);
1030
fprintf(stderr, "%s: invalid minimum number %d\n", argv[0], min);
1036
if (*argv[3] == ':')
1038
if ((var = getVariable(st, argv[3] + 1)) == NULL)
1040
fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[3]);
1047
max = atoi(argv[3]);
1049
if (max < min || max > MAX_RANDOM_VALUE)
1051
fprintf(stderr, "%s: invalid maximum number %d\n", argv[0], max);
1057
printf("min: %d max: %d random: %d\n", min, max, getrand(min, max));
1059
snprintf(res, sizeof(res), "%d", getrand(min, max));
1061
if (!putVariable(st, argv[0], argv[1], res))
1069
else if (pg_strcasecmp(argv[0], "set") == 0)
1076
if (*argv[2] == ':')
1078
if ((var = getVariable(st, argv[2] + 1)) == NULL)
1080
fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[2]);
1087
ope1 = atoi(argv[2]);
1090
snprintf(res, sizeof(res), "%d", ope1);
1093
if (*argv[4] == ':')
1095
if ((var = getVariable(st, argv[4] + 1)) == NULL)
1097
fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[4]);
1104
ope2 = atoi(argv[4]);
1106
if (strcmp(argv[3], "+") == 0)
1107
snprintf(res, sizeof(res), "%d", ope1 + ope2);
1108
else if (strcmp(argv[3], "-") == 0)
1109
snprintf(res, sizeof(res), "%d", ope1 - ope2);
1110
else if (strcmp(argv[3], "*") == 0)
1111
snprintf(res, sizeof(res), "%d", ope1 * ope2);
1112
else if (strcmp(argv[3], "/") == 0)
1116
fprintf(stderr, "%s: division by zero\n", argv[0]);
1120
snprintf(res, sizeof(res), "%d", ope1 / ope2);
1124
fprintf(stderr, "%s: unsupported operator %s\n", argv[0], argv[3]);
1130
if (!putVariable(st, argv[0], argv[1], res))
1138
else if (pg_strcasecmp(argv[0], "sleep") == 0)
1144
if (*argv[1] == ':')
1146
if ((var = getVariable(st, argv[1] + 1)) == NULL)
1148
fprintf(stderr, "%s: undefined variable %s\n", argv[0], argv[1]);
1155
usec = atoi(argv[1]);
1159
if (pg_strcasecmp(argv[2], "ms") == 0)
1161
else if (pg_strcasecmp(argv[2], "s") == 0)
1167
INSTR_TIME_SET_CURRENT(now);
1168
st->until = INSTR_TIME_GET_MICROSEC(now) + usec;
1173
else if (pg_strcasecmp(argv[0], "setshell") == 0)
1175
bool ret = runShellCommand(st, argv[1], argv + 2, argc - 2);
1177
if (timer_exceeded) /* timeout */
1178
return clientDone(st, true);
1179
else if (!ret) /* on error */
1184
else /* succeeded */
1187
else if (pg_strcasecmp(argv[0], "shell") == 0)
1189
bool ret = runShellCommand(st, NULL, argv + 1, argc - 1);
1191
if (timer_exceeded) /* timeout */
1192
return clientDone(st, true);
1193
else if (!ret) /* on error */
1198
else /* succeeded */
1207
/* discard connections */
1209
disconnect_all(CState *state, int length)
1213
for (i = 0; i < length; i++)
1217
PQfinish(state[i].con);
1218
state[i].con = NULL;
1223
/* create tables and setup data */
1228
* Note: TPC-B requires at least 100 bytes per row, and the "filler"
1229
* fields in these table declarations were intended to comply with that.
1230
* But because they default to NULLs, they don't actually take any space.
1231
* We could fix that by giving them non-null default values. However, that
1232
* would completely break comparability of pgbench results with prior
1233
* versions. Since pgbench has never pretended to be fully TPC-B
1234
* compliant anyway, we stick with the historical behavior.
1236
static char *DDLs[] = {
1237
"drop table if exists pgbench_branches",
1238
"create table pgbench_branches(bid int not null,bbalance int,filler char(88)) with (fillfactor=%d)",
1239
"drop table if exists pgbench_tellers",
1240
"create table pgbench_tellers(tid int not null,bid int,tbalance int,filler char(84)) with (fillfactor=%d)",
1241
"drop table if exists pgbench_accounts",
1242
"create table pgbench_accounts(aid int not null,bid int,abalance int,filler char(84)) with (fillfactor=%d)",
1243
"drop table if exists pgbench_history",
1244
"create table pgbench_history(tid int,bid int,aid int,delta int,mtime timestamp,filler char(22))"
1246
static char *DDLAFTERs[] = {
1247
"alter table pgbench_branches add primary key (bid)",
1248
"alter table pgbench_tellers add primary key (tid)",
1249
"alter table pgbench_accounts add primary key (aid)"
1257
if ((con = doConnect()) == NULL)
1260
for (i = 0; i < lengthof(DDLs); i++)
1263
* set fillfactor for branches, tellers and accounts tables
1265
if ((strstr(DDLs[i], "create table pgbench_branches") == DDLs[i]) ||
1266
(strstr(DDLs[i], "create table pgbench_tellers") == DDLs[i]) ||
1267
(strstr(DDLs[i], "create table pgbench_accounts") == DDLs[i]))
1271
snprintf(ddl_stmt, 128, DDLs[i], fillfactor);
1272
executeStatement(con, ddl_stmt);
1276
executeStatement(con, DDLs[i]);
1279
executeStatement(con, "begin");
1281
for (i = 0; i < nbranches * scale; i++)
1283
snprintf(sql, 256, "insert into pgbench_branches(bid,bbalance) values(%d,0)", i + 1);
1284
executeStatement(con, sql);
1287
for (i = 0; i < ntellers * scale; i++)
1289
snprintf(sql, 256, "insert into pgbench_tellers(tid,bid,tbalance) values (%d,%d,0)",
1290
i + 1, i / ntellers + 1);
1291
executeStatement(con, sql);
1294
executeStatement(con, "commit");
1297
* fill the pgbench_accounts table with some data
1299
fprintf(stderr, "creating tables...\n");
1301
executeStatement(con, "begin");
1302
executeStatement(con, "truncate pgbench_accounts");
1304
res = PQexec(con, "copy pgbench_accounts from stdin");
1305
if (PQresultStatus(res) != PGRES_COPY_IN)
1307
fprintf(stderr, "%s", PQerrorMessage(con));
1312
for (i = 0; i < naccounts * scale; i++)
1316
snprintf(sql, 256, "%d\t%d\t%d\t\n", j, i / naccounts + 1, 0);
1317
if (PQputline(con, sql))
1319
fprintf(stderr, "PQputline failed\n");
1324
fprintf(stderr, "%d tuples done.\n", j);
1326
if (PQputline(con, "\\.\n"))
1328
fprintf(stderr, "very last PQputline failed\n");
1333
fprintf(stderr, "PQendcopy failed\n");
1336
executeStatement(con, "commit");
1341
fprintf(stderr, "set primary key...\n");
1342
for (i = 0; i < lengthof(DDLAFTERs); i++)
1343
executeStatement(con, DDLAFTERs[i]);
1346
fprintf(stderr, "vacuum...");
1347
executeStatement(con, "vacuum analyze pgbench_branches");
1348
executeStatement(con, "vacuum analyze pgbench_tellers");
1349
executeStatement(con, "vacuum analyze pgbench_accounts");
1350
executeStatement(con, "vacuum analyze pgbench_history");
1352
fprintf(stderr, "done.\n");
1357
* Parse the raw sql and replace :param to $n.
1360
parseQuery(Command *cmd, const char *raw_sql)
1365
sql = xstrdup(raw_sql);
1369
while ((p = strchr(p, ':')) != NULL)
1375
name = parseVariable(p, &eaten);
1385
if (cmd->argc >= MAX_ARGS)
1387
fprintf(stderr, "statement has too many arguments (maximum is %d): %s\n", MAX_ARGS - 1, raw_sql);
1391
sprintf(var, "$%d", cmd->argc);
1392
p = replaceVariable(&sql, p, eaten, var);
1394
cmd->argv[cmd->argc] = name;
1402
/* Parse a command; return a Command struct, or NULL if it's a comment */
1404
process_commands(char *buf)
1406
const char delim[] = " \f\n\r\t\v";
1408
Command *my_commands;
1413
/* Make the string buf end at the next newline */
1414
if ((p = strchr(buf, '\n')) != NULL)
1417
/* Skip leading whitespace */
1419
while (isspace((unsigned char) *p))
1422
/* If the line is empty or actually a comment, we're done */
1423
if (*p == '\0' || strncmp(p, "--", 2) == 0)
1426
/* Allocate and initialize Command structure */
1427
my_commands = (Command *) xmalloc(sizeof(Command));
1428
my_commands->line = xstrdup(buf);
1429
my_commands->command_num = num_commands++;
1430
my_commands->type = 0; /* until set */
1431
my_commands->argc = 0;
1435
my_commands->type = META_COMMAND;
1438
tok = strtok(++p, delim);
1442
my_commands->argv[j++] = xstrdup(tok);
1443
my_commands->argc++;
1444
tok = strtok(NULL, delim);
1447
if (pg_strcasecmp(my_commands->argv[0], "setrandom") == 0)
1449
if (my_commands->argc < 4)
1451
fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1455
for (j = 4; j < my_commands->argc; j++)
1456
fprintf(stderr, "%s: extra argument \"%s\" ignored\n",
1457
my_commands->argv[0], my_commands->argv[j]);
1459
else if (pg_strcasecmp(my_commands->argv[0], "set") == 0)
1461
if (my_commands->argc < 3)
1463
fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1467
for (j = my_commands->argc < 5 ? 3 : 5; j < my_commands->argc; j++)
1468
fprintf(stderr, "%s: extra argument \"%s\" ignored\n",
1469
my_commands->argv[0], my_commands->argv[j]);
1471
else if (pg_strcasecmp(my_commands->argv[0], "sleep") == 0)
1473
if (my_commands->argc < 2)
1475
fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1480
* Split argument into number and unit to allow "sleep 1ms" etc.
1481
* We don't have to terminate the number argument with null
1482
* because it will be parsed with atoi, which ignores trailing
1483
* non-digit characters.
1485
if (my_commands->argv[1][0] != ':')
1487
char *c = my_commands->argv[1];
1489
while (isdigit((unsigned char) *c))
1493
my_commands->argv[2] = c;
1494
if (my_commands->argc < 3)
1495
my_commands->argc = 3;
1499
if (my_commands->argc >= 3)
1501
if (pg_strcasecmp(my_commands->argv[2], "us") != 0 &&
1502
pg_strcasecmp(my_commands->argv[2], "ms") != 0 &&
1503
pg_strcasecmp(my_commands->argv[2], "s"))
1505
fprintf(stderr, "%s: unknown time unit '%s' - must be us, ms or s\n",
1506
my_commands->argv[0], my_commands->argv[2]);
1511
for (j = 3; j < my_commands->argc; j++)
1512
fprintf(stderr, "%s: extra argument \"%s\" ignored\n",
1513
my_commands->argv[0], my_commands->argv[j]);
1515
else if (pg_strcasecmp(my_commands->argv[0], "setshell") == 0)
1517
if (my_commands->argc < 3)
1519
fprintf(stderr, "%s: missing argument\n", my_commands->argv[0]);
1523
else if (pg_strcasecmp(my_commands->argv[0], "shell") == 0)
1525
if (my_commands->argc < 1)
1527
fprintf(stderr, "%s: missing command\n", my_commands->argv[0]);
1533
fprintf(stderr, "Invalid command %s\n", my_commands->argv[0]);
1539
my_commands->type = SQL_COMMAND;
1544
my_commands->argv[0] = xstrdup(p);
1545
my_commands->argc++;
1547
case QUERY_EXTENDED:
1548
case QUERY_PREPARED:
1549
if (!parseQuery(my_commands, p))
1561
process_file(char *filename)
1563
#define COMMANDS_ALLOC_NUM 128
1565
Command **my_commands;
1571
if (num_files >= MAX_FILES)
1573
fprintf(stderr, "Up to only %d SQL files are allowed\n", MAX_FILES);
1577
alloc_num = COMMANDS_ALLOC_NUM;
1578
my_commands = (Command **) xmalloc(sizeof(Command *) * alloc_num);
1580
if (strcmp(filename, "-") == 0)
1582
else if ((fd = fopen(filename, "r")) == NULL)
1584
fprintf(stderr, "%s: %s\n", filename, strerror(errno));
1590
while (fgets(buf, sizeof(buf), fd) != NULL)
1594
command = process_commands(buf);
1595
if (command == NULL)
1598
my_commands[lineno] = command;
1601
if (lineno >= alloc_num)
1603
alloc_num += COMMANDS_ALLOC_NUM;
1604
my_commands = xrealloc(my_commands, sizeof(Command *) * alloc_num);
1609
my_commands[lineno] = NULL;
1611
sql_files[num_files++] = my_commands;
1617
process_builtin(char *tb)
1619
#define COMMANDS_ALLOC_NUM 128
1621
Command **my_commands;
1626
alloc_num = COMMANDS_ALLOC_NUM;
1627
my_commands = (Command **) xmalloc(sizeof(Command *) * alloc_num);
1637
while (*tb && *tb != '\n')
1648
command = process_commands(buf);
1649
if (command == NULL)
1652
my_commands[lineno] = command;
1655
if (lineno >= alloc_num)
1657
alloc_num += COMMANDS_ALLOC_NUM;
1658
my_commands = xrealloc(my_commands, sizeof(Command *) * alloc_num);
1662
my_commands[lineno] = NULL;
1667
/* print out results */
1669
printResults(int ttype, int normal_xacts, int nclients,
1670
TState *threads, int nthreads,
1671
instr_time total_time, instr_time conn_total_time)
1673
double time_include,
1678
time_include = INSTR_TIME_GET_DOUBLE(total_time);
1679
tps_include = normal_xacts / time_include;
1680
tps_exclude = normal_xacts / (time_include -
1681
(INSTR_TIME_GET_DOUBLE(conn_total_time) / nthreads));
1684
s = "TPC-B (sort of)";
1685
else if (ttype == 2)
1686
s = "Update only pgbench_accounts";
1687
else if (ttype == 1)
1692
printf("transaction type: %s\n", s);
1693
printf("scaling factor: %d\n", scale);
1694
printf("query mode: %s\n", QUERYMODE[querymode]);
1695
printf("number of clients: %d\n", nclients);
1696
printf("number of threads: %d\n", nthreads);
1699
printf("number of transactions per client: %d\n", nxacts);
1700
printf("number of transactions actually processed: %d/%d\n",
1701
normal_xacts, nxacts * nclients);
1705
printf("duration: %d s\n", duration);
1706
printf("number of transactions actually processed: %d\n",
1709
printf("tps = %f (including connections establishing)\n", tps_include);
1710
printf("tps = %f (excluding connections establishing)\n", tps_exclude);
1712
/* Report per-command latencies */
1717
for (i = 0; i < num_files; i++)
1722
printf("statement latencies in milliseconds, file %d:\n", i + 1);
1724
printf("statement latencies in milliseconds:\n");
1726
for (commands = sql_files[i]; *commands != NULL; commands++)
1728
Command *command = *commands;
1729
int cnum = command->command_num;
1731
instr_time total_exec_elapsed;
1732
int total_exec_count;
1735
/* Accumulate per-thread data for command */
1736
INSTR_TIME_SET_ZERO(total_exec_elapsed);
1737
total_exec_count = 0;
1738
for (t = 0; t < nthreads; t++)
1740
TState *thread = &threads[t];
1742
INSTR_TIME_ADD(total_exec_elapsed,
1743
thread->exec_elapsed[cnum]);
1744
total_exec_count += thread->exec_count[cnum];
1747
if (total_exec_count > 0)
1748
total_time = INSTR_TIME_GET_MILLISEC(total_exec_elapsed) / (double) total_exec_count;
1752
printf("\t%f\t%s\n", total_time, command->line);
1760
main(int argc, char **argv)
1763
int nclients = 1; /* default number of simulated clients */
1764
int nthreads = 1; /* default number of threads */
1765
int is_init_mode = 0; /* initialize mode? */
1766
int is_no_vacuum = 0; /* no vacuum at all before testing? */
1767
int do_vacuum_accounts = 0; /* do vacuum accounts before testing? */
1768
int ttype = 0; /* transaction type. 0: TPC-B, 1: SELECT only,
1769
* 2: skip update of branches and tellers */
1770
char *filename = NULL;
1771
bool scale_given = false;
1773
CState *state; /* status of clients */
1774
TState *threads; /* array of thread */
1776
instr_time start_time; /* start up time */
1777
instr_time total_time;
1778
instr_time conn_total_time;
1783
#ifdef HAVE_GETRLIMIT
1793
const char *progname;
1795
progname = get_progname(argv[0]);
1799
if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
1804
if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
1806
puts("pgbench (PostgreSQL) " PG_VERSION);
1812
/* stderr is buffered on Win32. */
1813
setvbuf(stderr, NULL, _IONBF, 0);
1816
if ((env = getenv("PGHOST")) != NULL && *env != '\0')
1818
if ((env = getenv("PGPORT")) != NULL && *env != '\0')
1820
else if ((env = getenv("PGUSER")) != NULL && *env != '\0')
1823
state = (CState *) xmalloc(sizeof(CState));
1824
memset(state, 0, sizeof(CState));
1826
while ((c = getopt(argc, argv, "ih:nvp:dSNc:j:Crs:t:T:U:lf:D:F:M:")) != -1)
1840
do_vacuum_accounts++;
1855
nclients = atoi(optarg);
1856
if (nclients <= 0 || nclients > MAXCLIENTS)
1858
fprintf(stderr, "invalid number of clients: %d\n", nclients);
1861
#ifdef HAVE_GETRLIMIT
1862
#ifdef RLIMIT_NOFILE /* most platforms use RLIMIT_NOFILE */
1863
if (getrlimit(RLIMIT_NOFILE, &rlim) == -1)
1864
#else /* but BSD doesn't ... */
1865
if (getrlimit(RLIMIT_OFILE, &rlim) == -1)
1866
#endif /* RLIMIT_NOFILE */
1868
fprintf(stderr, "getrlimit failed: %s\n", strerror(errno));
1871
if (rlim.rlim_cur <= (nclients + 2))
1873
fprintf(stderr, "You need at least %d open files but you are only allowed to use %ld.\n", nclients + 2, (long) rlim.rlim_cur);
1874
fprintf(stderr, "Use limit/ulimit to increase the limit before using pgbench.\n");
1877
#endif /* HAVE_GETRLIMIT */
1879
case 'j': /* jobs */
1880
nthreads = atoi(optarg);
1883
fprintf(stderr, "invalid number of threads: %d\n", nthreads);
1891
is_latencies = true;
1895
scale = atoi(optarg);
1898
fprintf(stderr, "invalid scaling factor: %d\n", scale);
1905
fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both.\n");
1908
nxacts = atoi(optarg);
1911
fprintf(stderr, "invalid number of transactions: %d\n", nxacts);
1918
fprintf(stderr, "specify either a number of transactions (-t) or a duration (-T), not both.\n");
1921
duration = atoi(optarg);
1924
fprintf(stderr, "invalid duration: %d\n", duration);
1937
if (process_file(filename) == false || *sql_files[num_files - 1] == NULL)
1944
if ((p = strchr(optarg, '=')) == NULL || p == optarg || *(p + 1) == '\0')
1946
fprintf(stderr, "invalid variable definition: %s\n", optarg);
1951
if (!putVariable(&state[0], "option", optarg, p))
1956
fillfactor = atoi(optarg);
1957
if ((fillfactor < 10) || (fillfactor > 100))
1959
fprintf(stderr, "invalid fillfactor: %d\n", fillfactor);
1966
fprintf(stderr, "query mode (-M) should be specifiled before transaction scripts (-f)\n");
1969
for (querymode = 0; querymode < NUM_QUERYMODE; querymode++)
1970
if (strcmp(optarg, QUERYMODE[querymode]) == 0)
1972
if (querymode >= NUM_QUERYMODE)
1974
fprintf(stderr, "invalid query mode (-M): %s\n", optarg);
1979
fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
1986
dbName = argv[optind];
1989
if ((env = getenv("PGDATABASE")) != NULL && *env != '\0')
1991
else if (login != NULL && *login != '\0')
2003
/* Use DEFAULT_NXACTS if neither nxacts nor duration is specified. */
2004
if (nxacts <= 0 && duration <= 0)
2005
nxacts = DEFAULT_NXACTS;
2007
if (nclients % nthreads != 0)
2009
fprintf(stderr, "number of clients (%d) must be a multiple of number of threads (%d)\n", nclients, nthreads);
2014
* is_latencies only works with multiple threads in thread-based
2015
* implementations, not fork-based ones, because it supposes that the
2016
* parent can see changes made to the per-thread execution stats by child
2017
* threads. It seems useful enough to accept despite this limitation, but
2018
* perhaps we should FIXME someday (by passing the stats data back up
2019
* through the parent-to-child pipes).
2021
#ifndef ENABLE_THREAD_SAFETY
2022
if (is_latencies && nthreads > 1)
2024
fprintf(stderr, "-r does not work with -j larger than 1 on this platform.\n");
2030
* save main process id in the global variable because process id will be
2031
* changed after fork.
2033
main_pid = (int) getpid();
2037
state = (CState *) xrealloc(state, sizeof(CState) * nclients);
2038
memset(state + 1, 0, sizeof(CState) * (nclients - 1));
2040
/* copy any -D switch values to all clients */
2041
for (i = 1; i < nclients; i++)
2046
for (j = 0; j < state[0].nvariables; j++)
2048
if (!putVariable(&state[i], "startup", state[0].variables[j].name, state[0].variables[j].value))
2057
printf("pghost: %s pgport: %s nclients: %d nxacts: %d dbName: %s\n",
2058
pghost, pgport, nclients, nxacts, dbName);
2060
printf("pghost: %s pgport: %s nclients: %d duration: %d dbName: %s\n",
2061
pghost, pgport, nclients, duration, dbName);
2064
/* opening connection... */
2069
if (PQstatus(con) == CONNECTION_BAD)
2071
fprintf(stderr, "Connection to database '%s' failed.\n", dbName);
2072
fprintf(stderr, "%s", PQerrorMessage(con));
2079
* get the scaling factor that should be same as count(*) from
2080
* pgbench_branches if this is not a custom query
2082
res = PQexec(con, "select count(*) from pgbench_branches");
2083
if (PQresultStatus(res) != PGRES_TUPLES_OK)
2085
fprintf(stderr, "%s", PQerrorMessage(con));
2088
scale = atoi(PQgetvalue(res, 0, 0));
2091
fprintf(stderr, "count(*) from pgbench_branches invalid (%d)\n", scale);
2096
/* warn if we override user-given -s switch */
2099
"Scale option ignored, using pgbench_branches table count = %d\n",
2104
* :scale variables normally get -s or database scale, but don't override
2105
* an explicit -D switch
2107
if (getVariable(&state[0], "scale") == NULL)
2109
snprintf(val, sizeof(val), "%d", scale);
2110
for (i = 0; i < nclients; i++)
2112
if (!putVariable(&state[i], "startup", "scale", val))
2119
fprintf(stderr, "starting vacuum...");
2120
executeStatement(con, "vacuum pgbench_branches");
2121
executeStatement(con, "vacuum pgbench_tellers");
2122
executeStatement(con, "truncate pgbench_history");
2123
fprintf(stderr, "end.\n");
2125
if (do_vacuum_accounts)
2127
fprintf(stderr, "starting vacuum pgbench_accounts...");
2128
executeStatement(con, "vacuum analyze pgbench_accounts");
2129
fprintf(stderr, "end.\n");
2134
/* set random seed */
2135
INSTR_TIME_SET_CURRENT(start_time);
2136
srandom((unsigned int) INSTR_TIME_GET_MICROSEC(start_time));
2138
/* process builtin SQL scripts */
2142
sql_files[0] = process_builtin(tpc_b);
2147
sql_files[0] = process_builtin(select_only);
2152
sql_files[0] = process_builtin(simple_update);
2160
/* set up thread data structures */
2161
threads = (TState *) xmalloc(sizeof(TState) * nthreads);
2162
for (i = 0; i < nthreads; i++)
2164
TState *thread = &threads[i];
2167
thread->state = &state[nclients / nthreads * i];
2168
thread->nstate = nclients / nthreads;
2172
/* Reserve memory for the thread to store per-command latencies */
2175
thread->exec_elapsed = (instr_time *)
2176
xmalloc(sizeof(instr_time) * num_commands);
2177
thread->exec_count = (int *)
2178
xmalloc(sizeof(int) * num_commands);
2180
for (t = 0; t < num_commands; t++)
2182
INSTR_TIME_SET_ZERO(thread->exec_elapsed[t]);
2183
thread->exec_count[t] = 0;
2188
thread->exec_elapsed = NULL;
2189
thread->exec_count = NULL;
2193
/* get start up time */
2194
INSTR_TIME_SET_CURRENT(start_time);
2196
/* set alarm if duration is specified. */
2201
for (i = 0; i < nthreads; i++)
2203
TState *thread = &threads[i];
2205
INSTR_TIME_SET_CURRENT(thread->start_time);
2207
/* the first thread (i = 0) is executed by main thread */
2210
int err = pthread_create(&thread->thread, NULL, threadRun, thread);
2212
if (err != 0 || thread->thread == INVALID_THREAD)
2214
fprintf(stderr, "cannot create thread: %s\n", strerror(err));
2220
thread->thread = INVALID_THREAD;
2224
/* wait for threads and accumulate results */
2226
INSTR_TIME_SET_ZERO(conn_total_time);
2227
for (i = 0; i < nthreads; i++)
2231
if (threads[i].thread == INVALID_THREAD)
2232
ret = threadRun(&threads[i]);
2234
pthread_join(threads[i].thread, &ret);
2238
TResult *r = (TResult *) ret;
2240
total_xacts += r->xacts;
2241
INSTR_TIME_ADD(conn_total_time, r->conn_time);
2245
disconnect_all(state, nclients);
2248
INSTR_TIME_SET_CURRENT(total_time);
2249
INSTR_TIME_SUBTRACT(total_time, start_time);
2250
printResults(ttype, total_xacts, nclients, threads, nthreads,
2251
total_time, conn_total_time);
2257
threadRun(void *arg)
2259
TState *thread = (TState *) arg;
2260
CState *state = thread->state;
2262
FILE *logfile = NULL; /* per-thread log file */
2265
int nstate = thread->nstate;
2266
int remains = nstate; /* number of remaining clients */
2269
result = xmalloc(sizeof(TResult));
2270
INSTR_TIME_SET_ZERO(result->conn_time);
2272
/* open log file if requested */
2277
if (thread->tid == 0)
2278
snprintf(logpath, sizeof(logpath), "pgbench_log.%d", main_pid);
2280
snprintf(logpath, sizeof(logpath), "pgbench_log.%d.%d", main_pid, thread->tid);
2281
logfile = fopen(logpath, "w");
2283
if (logfile == NULL)
2285
fprintf(stderr, "Couldn't open logfile \"%s\": %s", logpath, strerror(errno));
2292
/* make connections to the database */
2293
for (i = 0; i < nstate; i++)
2295
if ((state[i].con = doConnect()) == NULL)
2300
/* time after thread and connections set up */
2301
INSTR_TIME_SET_CURRENT(result->conn_time);
2302
INSTR_TIME_SUBTRACT(result->conn_time, thread->start_time);
2304
/* send start up queries in async manner */
2305
for (i = 0; i < nstate; i++)
2307
CState *st = &state[i];
2308
Command **commands = sql_files[st->use_file];
2309
int prev_ecnt = st->ecnt;
2311
st->use_file = getrand(0, num_files - 1);
2312
if (!doCustom(thread, st, &result->conn_time, logfile))
2313
remains--; /* I've aborted */
2315
if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND)
2317
fprintf(stderr, "Client %d aborted in state %d. Execution meta-command failed.\n", i, st->state);
2318
remains--; /* I've aborted */
2327
int maxsock; /* max socket number to be waited */
2331
FD_ZERO(&input_mask);
2334
min_usec = INT64_MAX;
2335
for (i = 0; i < nstate; i++)
2337
CState *st = &state[i];
2338
Command **commands = sql_files[st->use_file];
2345
if (min_usec == INT64_MAX)
2349
INSTR_TIME_SET_CURRENT(now);
2350
now_usec = INSTR_TIME_GET_MICROSEC(now);
2353
this_usec = st->until - now_usec;
2354
if (min_usec > this_usec)
2355
min_usec = this_usec;
2357
else if (st->con == NULL)
2361
else if (commands[st->state]->type == META_COMMAND)
2363
min_usec = 0; /* the connection is ready to run */
2367
sock = PQsocket(st->con);
2370
fprintf(stderr, "bad socket: %s\n", strerror(errno));
2374
FD_SET(sock, &input_mask);
2380
if (min_usec > 0 && maxsock != -1)
2382
int nsocks; /* return from select(2) */
2384
if (min_usec != INT64_MAX)
2386
struct timeval timeout;
2388
timeout.tv_sec = min_usec / 1000000;
2389
timeout.tv_usec = min_usec % 1000000;
2390
nsocks = select(maxsock + 1, &input_mask, NULL, NULL, &timeout);
2393
nsocks = select(maxsock + 1, &input_mask, NULL, NULL, NULL);
2398
/* must be something wrong */
2399
fprintf(stderr, "select failed: %s\n", strerror(errno));
2404
/* ok, backend returns reply */
2405
for (i = 0; i < nstate; i++)
2407
CState *st = &state[i];
2408
Command **commands = sql_files[st->use_file];
2409
int prev_ecnt = st->ecnt;
2411
if (st->con && (FD_ISSET(PQsocket(st->con), &input_mask)
2412
|| commands[st->state]->type == META_COMMAND))
2414
if (!doCustom(thread, st, &result->conn_time, logfile))
2415
remains--; /* I've aborted */
2418
if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND)
2420
fprintf(stderr, "Client %d aborted in state %d. Execution of meta-command failed.\n", i, st->state);
2421
remains--; /* I've aborted */
2429
INSTR_TIME_SET_CURRENT(start);
2430
disconnect_all(state, nstate);
2432
for (i = 0; i < nstate; i++)
2433
result->xacts += state[i].cnt;
2434
INSTR_TIME_SET_CURRENT(end);
2435
INSTR_TIME_ACCUM_DIFF(result->conn_time, end, start);
2443
* Support for duration option: set timer_exceeded after so many seconds.
2449
handle_sig_alarm(SIGNAL_ARGS)
2451
timer_exceeded = true;
2455
setalarm(int seconds)
2457
pqsignal(SIGALRM, handle_sig_alarm);
2461
#ifndef ENABLE_THREAD_SAFETY
2464
* implements pthread using fork.
2467
typedef struct fork_pthread
2474
pthread_create(pthread_t *thread,
2475
pthread_attr_t * attr,
2476
void *(*start_routine) (void *),
2481
instr_time start_time;
2483
th = (fork_pthread *) xmalloc(sizeof(fork_pthread));
2484
if (pipe(th->pipes) < 0)
2491
if (th->pid == -1) /* error */
2496
if (th->pid != 0) /* in parent process */
2498
close(th->pipes[1]);
2503
/* in child process */
2504
close(th->pipes[0]);
2506
/* set alarm again because the child does not inherit timers */
2511
* Set a different random seed in each child process. Otherwise they all
2512
* inherit the parent's state and generate the same "random" sequence. (In
2513
* the threaded case, the different threads will obtain subsets of the
2514
* output of a single random() sequence, which should be okay for our
2517
INSTR_TIME_SET_CURRENT(start_time);
2518
srandom(((unsigned int) INSTR_TIME_GET_MICROSEC(start_time)) +
2519
((unsigned int) getpid()));
2521
ret = start_routine(arg);
2522
write(th->pipes[1], ret, sizeof(TResult));
2523
close(th->pipes[1]);
2529
pthread_join(pthread_t th, void **thread_return)
2533
while (waitpid(th->pid, &status, 0) != th->pid)
2539
if (thread_return != NULL)
2541
/* assume result is TResult */
2542
*thread_return = xmalloc(sizeof(TResult));
2543
if (read(th->pipes[0], *thread_return, sizeof(TResult)) != sizeof(TResult))
2545
free(*thread_return);
2546
*thread_return = NULL;
2549
close(th->pipes[0]);
2557
static VOID CALLBACK
2558
win32_timer_callback(PVOID lpParameter, BOOLEAN TimerOrWaitFired)
2560
timer_exceeded = true;
2564
setalarm(int seconds)
2569
/* This function will be called at most once, so we can cheat a bit. */
2570
queue = CreateTimerQueue();
2571
if (seconds > ((DWORD) -1) / 1000 ||
2572
!CreateTimerQueueTimer(&timer, queue,
2573
win32_timer_callback, NULL, seconds * 1000, 0,
2574
WT_EXECUTEINTIMERTHREAD | WT_EXECUTEONLYONCE))
2576
fprintf(stderr, "Failed to set timer\n");
2581
/* partial pthread implementation for Windows */
2583
typedef struct win32_pthread
2586
void *(*routine) (void *);
2591
static unsigned __stdcall
2592
win32_pthread_run(void *arg)
2594
win32_pthread *th = (win32_pthread *) arg;
2596
th->result = th->routine(th->arg);
2602
pthread_create(pthread_t *thread,
2603
pthread_attr_t * attr,
2604
void *(*start_routine) (void *),
2610
th = (win32_pthread *) xmalloc(sizeof(win32_pthread));
2611
th->routine = start_routine;
2615
th->handle = (HANDLE) _beginthreadex(NULL, 0, win32_pthread_run, th, 0, NULL);
2616
if (th->handle == NULL)
2628
pthread_join(pthread_t th, void **thread_return)
2630
if (th == NULL || th->handle == NULL)
2631
return errno = EINVAL;
2633
if (WaitForSingleObject(th->handle, INFINITE) != WAIT_OBJECT_0)
2635
_dosmaperr(GetLastError());
2640
*thread_return = th->result;
2642
CloseHandle(th->handle);
added
removed
contrib/pgbench/pgbench.c
