From 328bea28c4fc777d483b2c7837fcb8fafcd08923 Mon Sep 17 00:00:00 2001 From: Kyotaro Horiguchi Date: Wed, 19 Aug 2015 12:53:13 +0900 Subject: [PATCH 4/4] Change the way to hold command list. Commands are generated as a linked list and stored into and accessed as an array. This patch unifies the way to store them to linked list. --- src/bin/pgbench/pgbench.c | 189 +++++++++++++++++++++++----------------------- 1 file changed, 95 insertions(+), 94 deletions(-) diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c index b6fd399..285ccca 100644 --- a/src/bin/pgbench/pgbench.c +++ b/src/bin/pgbench/pgbench.c @@ -191,16 +191,29 @@ typedef struct #define MAX_FILES 128 /* max number of SQL script files allowed */ #define SHELL_COMMAND_SIZE 256 /* maximum size allowed for shell command */ +#define MAX_ARGS 10 /* * structures used in custom query mode */ +typedef struct Command_t +{ + char *line; /* full text of command line */ + int command_num; /* unique index of this Command struct */ + int type; /* command type (SQL_COMMAND or META_COMMAND) */ + int argc; /* number of command words */ + char *argv[MAX_ARGS]; /* command word list */ + int cols[MAX_ARGS]; /* corresponding column starting from 1 */ + PgBenchExpr *expr; /* parsed expression */ + struct Command_t *next; /* more command if any, for multistatements */ +} Command; + typedef struct { PGconn *con; /* connection handle to DB */ int id; /* client No. */ - int state; /* state No. */ + Command *curr; /* current command */ int listen; /* 0 indicates that an async query has been * sent */ int sleeping; /* 1 indicates that the client is napping */ @@ -252,7 +265,6 @@ typedef struct */ #define SQL_COMMAND 1 #define META_COMMAND 2 -#define MAX_ARGS 10 typedef enum QueryMode { @@ -265,18 +277,6 @@ typedef enum QueryMode static QueryMode querymode = QUERY_SIMPLE; static const char *QUERYMODE[] = {"simple", "extended", "prepared"}; -typedef struct Command_t -{ - char *line; /* full text of command line */ - int command_num; /* unique index of this Command struct */ - int type; /* command type (SQL_COMMAND or META_COMMAND) */ - int argc; /* number of command words */ - char *argv[MAX_ARGS]; /* command word list */ - int cols[MAX_ARGS]; /* corresponding column starting from 1 */ - PgBenchExpr *expr; /* parsed expression */ - struct Command_t *next; /* more command if any, for multistatements */ -} Command; - typedef struct { @@ -312,7 +312,7 @@ typedef struct ParseInfo } ParseInfoData; typedef ParseInfoData *ParseInfo; -static Command **sql_files[MAX_FILES]; /* SQL script files */ +static Command *sql_files[MAX_FILES]; /* SQL script files */ static int num_files; /* number of script files */ static int num_commands = 0; /* total number of Command structs */ static int debug = 0; /* debug flag */ @@ -1140,12 +1140,27 @@ agg_vals_init(AggVals *aggs, instr_time start) aggs->start_time = INSTR_TIME_GET_DOUBLE(start); } +/* Return the ordinal of a command list item in a list */ +static int +get_command_number(Command *head, Command *curr) +{ + int i; + Command *p = head; + + for (i = 0 ; p && p != curr ; p = p->next, i++); + + /* curr must be in the list */ + Assert(p); + + return i; +} + /* return false iff client should be disconnected */ static bool doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile, AggVals *agg) { PGresult *res; - Command **commands; + Command *commands; bool trans_needs_throttle = false; instr_time now; @@ -1242,13 +1257,14 @@ top: if (st->listen) { /* are we receiver? */ - if (commands[st->state]->type == SQL_COMMAND) + if (st->curr->type == SQL_COMMAND) { if (debug) fprintf(stderr, "client %d receiving\n", st->id); if (!PQconsumeInput(st->con)) { /* there's something wrong */ - fprintf(stderr, "client %d aborted in state %d; perhaps the backend died while processing\n", st->id, st->state); + fprintf(stderr, "client %d aborted in state %d; perhaps the backend died while processing\n", + st->id, get_command_number(commands, st->curr)); return clientDone(st, false); } if (PQisBusy(st->con)) @@ -1261,7 +1277,7 @@ top: */ if (is_latencies) { - int cnum = commands[st->state]->command_num; + int cnum = st->curr->command_num; if (INSTR_TIME_IS_ZERO(now)) INSTR_TIME_SET_CURRENT(now); @@ -1271,7 +1287,7 @@ top: } /* transaction finished: calculate latency and log the transaction */ - if (commands[st->state + 1] == NULL) + if (st->curr->next == NULL) { /* only calculate latency if an option is used that needs it */ if (progress || throttle_delay || latency_limit) @@ -1304,7 +1320,7 @@ top: doLog(thread, st, logfile, &now, agg, false); } - if (commands[st->state]->type == SQL_COMMAND) + if (st->curr->type == SQL_COMMAND) { /* * Read and discard the query result; note this is not included in @@ -1318,7 +1334,8 @@ top: break; /* OK */ default: fprintf(stderr, "client %d aborted in state %d: %s", - st->id, st->state, PQerrorMessage(st->con)); + st->id, get_command_number(commands, st->curr), + PQerrorMessage(st->con)); PQclear(res); return clientDone(st, false); } @@ -1326,7 +1343,7 @@ top: discard_response(st); } - if (commands[st->state + 1] == NULL) + if (st->curr->next == NULL) { if (is_connect) { @@ -1340,12 +1357,12 @@ top: } /* increment state counter */ - st->state++; - if (commands[st->state] == NULL) + st->curr = st->curr->next; + if (st->curr == NULL) { - st->state = 0; st->use_file = (int) getrand(thread, 0, num_files - 1); commands = sql_files[st->use_file]; + st->curr = commands; st->is_throttled = false; /* @@ -1388,7 +1405,8 @@ top: } /* Record transaction start time under logging, progress or throttling */ - if ((logfile || progress || throttle_delay || latency_limit) && st->state == 0) + if ((logfile || progress || throttle_delay || latency_limit) && + st->curr == commands) { INSTR_TIME_SET_CURRENT(st->txn_begin); @@ -1404,9 +1422,9 @@ top: if (is_latencies) INSTR_TIME_SET_CURRENT(st->stmt_begin); - if (commands[st->state]->type == SQL_COMMAND) + if (st->curr->type == SQL_COMMAND) { - const Command *command = commands[st->state]; + const Command *command = st->curr; int r; if (querymode == QUERY_SIMPLE) @@ -1440,18 +1458,19 @@ top: if (!st->prepared[st->use_file]) { - int j; + int j = 0; + Command *pcom = commands; - for (j = 0; commands[j] != NULL; j++) + for (; pcom ; pcom = pcom->next, j++) { PGresult *res; char name[MAX_PREPARE_NAME]; - if (commands[j]->type != SQL_COMMAND) + if (pcom->type != SQL_COMMAND) continue; preparedStatementName(name, st->use_file, j); res = PQprepare(st->con, name, - commands[j]->argv[0], commands[j]->argc - 1, NULL); + pcom->argv[0], pcom->argc - 1, NULL); if (PQresultStatus(res) != PGRES_COMMAND_OK) fprintf(stderr, "%s", PQerrorMessage(st->con)); PQclear(res); @@ -1460,7 +1479,8 @@ top: } getQueryParams(st, command, params); - preparedStatementName(name, st->use_file, st->state); + preparedStatementName(name, st->use_file, + get_command_number(commands, st->curr)); if (debug) fprintf(stderr, "client %d sending %s\n", st->id, name); @@ -1480,11 +1500,11 @@ top: else st->listen = 1; /* flags that should be listened */ } - else if (commands[st->state]->type == META_COMMAND) + else if (st->curr->type == META_COMMAND) { - int argc = commands[st->state]->argc, + int argc = st->curr->argc, i; - char **argv = commands[st->state]->argv; + char **argv = st->curr->argv; if (debug) { @@ -1626,7 +1646,7 @@ top: else if (pg_strcasecmp(argv[0], "set") == 0) { char res[64]; - PgBenchExpr *expr = commands[st->state]->expr; + PgBenchExpr *expr = st->curr->expr; int64 result; if (!evaluateExpr(st, expr, &result)) @@ -2629,14 +2649,11 @@ read_line_from_file(FILE *fd) static int process_file(char *filename) { -#define COMMANDS_ALLOC_NUM 128 - - Command **my_commands; + Command *my_commands = NULL, + *my_commands_tail = NULL; FILE *fd; - int lineno, - index; + int lineno; char *buf; - int alloc_num; ParseInfo proc_state = createParseInfo(); if (num_files >= MAX_FILES) @@ -2645,23 +2662,18 @@ process_file(char *filename) exit(1); } - alloc_num = COMMANDS_ALLOC_NUM; - my_commands = (Command **) pg_malloc(sizeof(Command *) * alloc_num); - if (strcmp(filename, "-") == 0) fd = stdin; else if ((fd = fopen(filename, "r")) == NULL) { fprintf(stderr, "could not open file \"%s\": %s\n", filename, strerror(errno)); - pg_free(my_commands); return false; } proc_state->mode = PS_IDLE; lineno = 0; - index = 0; while ((buf = read_line_from_file(fd)) != NULL) { @@ -2677,52 +2689,42 @@ process_file(char *filename) /* * command is NULL when psql_scan returns PSCAN_EOL or * PSCAN_INCOMPLETE. Immediately ask for the next line for the - * cases. + * case. */ continue; } - while (command) - { - my_commands[index++] = command; - command = command->next; - } + /* Append new commands at the end of the list */ + if (my_commands_tail) + my_commands_tail->next = command; + else + my_commands = my_commands_tail = command; - if (index > alloc_num) - { - alloc_num += COMMANDS_ALLOC_NUM; - my_commands = pg_realloc(my_commands, - sizeof(Command *) * alloc_num); - } + /* Seek to the tail of the list */ + while (my_commands_tail->next) + my_commands_tail = my_commands_tail->next; } fclose(fd); parse_finish_scan(proc_state); - my_commands[index] = NULL; + my_commands_tail->next = NULL; sql_files[num_files++] = my_commands; return true; } -static Command ** +static Command * process_builtin(char *tb, const char *source) { -#define COMMANDS_ALLOC_NUM 128 - - Command **my_commands; - int lineno, - index; + Command *my_commands = NULL, + *my_commands_tail = NULL; + int lineno; char buf[BUFSIZ]; - int alloc_num; ParseInfo proc_state = createParseInfo(); - alloc_num = COMMANDS_ALLOC_NUM; - my_commands = (Command **) pg_malloc(sizeof(Command *) * alloc_num); - lineno = 0; - index = 0; for (;;) { @@ -2747,19 +2749,17 @@ process_builtin(char *tb, const char *source) if (command == NULL) continue; - /* builtin doesn't need multistatements */ + /* For simplisity, inhibit builtin from multistatements */ Assert(command->next == NULL); - my_commands[index] = command; - index++; - - if (index >= alloc_num) + if (my_commands_tail) { - alloc_num += COMMANDS_ALLOC_NUM; - my_commands = pg_realloc(my_commands, sizeof(Command *) * alloc_num); + my_commands_tail->next = command; + my_commands_tail = command; } + else + my_commands = my_commands_tail = command; } - my_commands[index] = NULL; parse_finish_scan(proc_state); return my_commands; @@ -2864,16 +2864,16 @@ printResults(int ttype, int64 normal_xacts, int nclients, for (i = 0; i < num_files; i++) { - Command **commands; + Command *command; if (num_files > 1) printf("statement latencies in milliseconds, file %d:\n", i + 1); else printf("statement latencies in milliseconds:\n"); - for (commands = sql_files[i]; *commands != NULL; commands++) + for (command = sql_files[i]; command ; + command=command->next) { - Command *command = *commands; int cnum = command->command_num; double total_time; instr_time total_exec_elapsed; @@ -3153,7 +3153,7 @@ main(int argc, char **argv) benchmarking_option_set = true; ttype = 3; filename = pg_strdup(optarg); - if (process_file(filename) == false || *sql_files[num_files - 1] == NULL) + if (process_file(filename) == false || sql_files[num_files - 1] == NULL) exit(1); break; case 'D': @@ -3735,17 +3735,19 @@ threadRun(void *arg) for (i = 0; i < nstate; i++) { CState *st = &state[i]; - Command **commands = sql_files[st->use_file]; + Command *commands = sql_files[st->use_file]; int prev_ecnt = st->ecnt; st->use_file = getrand(thread, 0, num_files - 1); + st->curr = sql_files[st->use_file]; + if (!doCustom(thread, st, &thread->conn_time, logfile, &aggs)) remains--; /* I've aborted */ - if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND) + if (st->ecnt > prev_ecnt && st->curr->type == META_COMMAND) { fprintf(stderr, "client %d aborted in state %d; execution of meta-command failed\n", - i, st->state); + i, get_command_number(commands, st->curr)); remains--; /* I've aborted */ PQfinish(st->con); st->con = NULL; @@ -3766,7 +3768,6 @@ threadRun(void *arg) for (i = 0; i < nstate; i++) { CState *st = &state[i]; - Command **commands = sql_files[st->use_file]; int sock; if (st->con == NULL) @@ -3802,7 +3803,7 @@ threadRun(void *arg) min_usec = this_usec; } } - else if (commands[st->state]->type == META_COMMAND) + else if (st->curr->type == META_COMMAND) { min_usec = 0; /* the connection is ready to run */ break; @@ -3872,20 +3873,20 @@ threadRun(void *arg) for (i = 0; i < nstate; i++) { CState *st = &state[i]; - Command **commands = sql_files[st->use_file]; + Command *commands = sql_files[st->use_file]; int prev_ecnt = st->ecnt; if (st->con && (FD_ISSET(PQsocket(st->con), &input_mask) - || commands[st->state]->type == META_COMMAND)) + || st->curr->type == META_COMMAND)) { if (!doCustom(thread, st, &thread->conn_time, logfile, &aggs)) remains--; /* I've aborted */ } - if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND) + if (st->ecnt > prev_ecnt && st->curr->type == META_COMMAND) { fprintf(stderr, "client %d aborted in state %d; execution of meta-command failed\n", - i, st->state); + i, get_command_number(commands, st->curr)); remains--; /* I've aborted */ PQfinish(st->con); st->con = NULL; -- 1.8.3.1