granicus.if.org Git - postgresql/blob - contrib/pgbench/pgbench.c

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