diff options
Diffstat (limited to 'bench/wtperf/wtperf.c')
-rw-r--r-- | bench/wtperf/wtperf.c | 346 |
1 files changed, 245 insertions, 101 deletions
diff --git a/bench/wtperf/wtperf.c b/bench/wtperf/wtperf.c index 8c7f0053388..6d79eebe8b2 100644 --- a/bench/wtperf/wtperf.c +++ b/bench/wtperf/wtperf.c @@ -32,9 +32,6 @@ #define DEFAULT_HOME "WT_TEST" #define DEFAULT_MONITOR_DIR "WT_TEST" -static const char * const debug_cconfig = ""; -static const char * const debug_tconfig = ""; - static void *checkpoint_worker(void *); static int drop_all_tables(WTPERF *); static int execute_populate(WTPERF *); @@ -432,19 +429,17 @@ err: wtperf->error = wtperf->stop = true; * search do them. Ensuring the keys we see are always in order. */ static int -do_range_reads(WTPERF *wtperf, WT_CURSOR *cursor) +do_range_reads(WTPERF *wtperf, WT_CURSOR *cursor, int64_t read_range) { - CONFIG_OPTS *opts; - size_t range; uint64_t next_val, prev_val; + int64_t range; char *range_key_buf; char buf[512]; int ret; - opts = wtperf->opts; ret = 0; - if (opts->read_range == 0) + if (read_range == 0) return (0); memset(&buf[0], 0, 512 * sizeof(char)); @@ -454,7 +449,7 @@ do_range_reads(WTPERF *wtperf, WT_CURSOR *cursor) testutil_check(cursor->get_key(cursor, &range_key_buf)); extract_key(range_key_buf, &next_val); - for (range = 0; range < opts->read_range; ++range) { + for (range = 0; range < read_range; ++range) { prev_val = next_val; ret = cursor->next(cursor); /* We are done if we reach the end. */ @@ -475,12 +470,56 @@ do_range_reads(WTPERF *wtperf, WT_CURSOR *cursor) return (0); } +/* pre_load_data -- + * Pull everything into cache before starting the workload phase. + */ +static int +pre_load_data(WTPERF *wtperf) +{ + CONFIG_OPTS *opts; + WT_CONNECTION *conn; + WT_CURSOR *cursor; + WT_SESSION *session; + char *key; + int ret; + size_t i; + + opts = wtperf->opts; + conn = wtperf->conn; + + if ((ret = conn->open_session( + conn, NULL, opts->sess_config, &session)) != 0) { + lprintf(wtperf, ret, 0, "worker: WT_CONNECTION.open_session"); + goto err; + } + for (i = 0; i < opts->table_count; i++) { + if ((ret = session->open_cursor(session, + wtperf->uris[i], NULL, NULL, &cursor)) != 0) { + lprintf(wtperf, ret, 0, + "worker: WT_SESSION.open_cursor: %s", + wtperf->uris[i]); + goto err; + } + while (cursor->next(cursor) == 0) + if ((ret = cursor->get_key(cursor, &key)) != 0) + goto err; + if ((ret = cursor->close(cursor)) != 0) + goto err; + } + if ((ret = session->close(session, NULL)) != 0) + goto err; + if (ret != 0) +err: lprintf(wtperf, ret, 0, "Pre-workload traverse error"); + return (ret); +} + static void * worker(void *arg) { struct timespec start, stop; CONFIG_OPTS *opts; TRACK *trk; + WORKLOAD *workload; WTPERF *wtperf; WTPERF_THREAD *thread; WT_CONNECTION *conn; @@ -495,13 +534,14 @@ worker(void *arg) char buf[512]; thread = (WTPERF_THREAD *)arg; + workload = thread->workload; wtperf = thread->wtperf; opts = wtperf->opts; conn = wtperf->conn; cursors = NULL; - log_table_cursor = NULL; /* -Wconditional-initialized */ + cursor = log_table_cursor = NULL; /* -Wconditional-initialized */ ops = 0; - ops_per_txn = thread->workload->ops_per_txn; + ops_per_txn = workload->ops_per_txn; session = NULL; trk = NULL; @@ -510,9 +550,9 @@ worker(void *arg) lprintf(wtperf, ret, 0, "worker: WT_CONNECTION.open_session"); goto err; } - cursors = dcalloc(opts->table_count, sizeof(WT_CURSOR *)); for (i = 0; i < opts->table_count_idle; i++) { - snprintf(buf, 512, "%s_idle%05d", wtperf->uris[0], (int)i); + testutil_check(__wt_snprintf( + buf, 512, "%s_idle%05d", wtperf->uris[0], (int)i)); if ((ret = session->open_cursor( session, buf, NULL, NULL, &tmp_cursor)) != 0) { lprintf(wtperf, ret, 0, @@ -525,14 +565,34 @@ worker(void *arg) goto err; } } - for (i = 0; i < opts->table_count; i++) { + if (workload->table_index != INT32_MAX) { if ((ret = session->open_cursor(session, - wtperf->uris[i], NULL, NULL, &cursors[i])) != 0) { + wtperf->uris[workload->table_index], + NULL, NULL, &cursor)) != 0) { lprintf(wtperf, ret, 0, "worker: WT_SESSION.open_cursor: %s", - wtperf->uris[i]); + wtperf->uris[workload->table_index]); + goto err; + } + if ((ret = session->open_cursor(session, + wtperf->uris[workload->table_index], + NULL, "next_random=true", &thread->rand_cursor)) != 0) { + lprintf(wtperf, ret, 0, + "worker: WT_SESSION.open_cursor: random %s", + wtperf->uris[workload->table_index]); goto err; } + } else { + cursors = dcalloc(opts->table_count, sizeof(WT_CURSOR *)); + for (i = 0; i < opts->table_count; i++) { + if ((ret = session->open_cursor(session, + wtperf->uris[i], NULL, NULL, &cursors[i])) != 0) { + lprintf(wtperf, ret, 0, + "worker: WT_SESSION.open_cursor: %s", + wtperf->uris[i]); + goto err; + } + } } if (opts->log_like_table && (ret = session->open_cursor(session, wtperf->log_table_uri, NULL, NULL, &log_table_cursor)) != 0) { @@ -543,19 +603,19 @@ worker(void *arg) } /* Setup the timer for throttling. */ - if (thread->workload->throttle != 0) + if (workload->throttle != 0) setup_throttle(thread); /* Setup for truncate */ - if (thread->workload->truncate != 0) + if (workload->truncate != 0) if ((ret = setup_truncate(wtperf, thread, session)) != 0) goto err; key_buf = thread->key_buf; value_buf = thread->value_buf; - op = thread->workload->ops; - op_end = op + sizeof(thread->workload->ops); + op = workload->ops; + op_end = op + sizeof(workload->ops); if ((ops_per_txn != 0 || opts->log_like_table) && (ret = session->begin_transaction(session, NULL)) != 0) { @@ -564,6 +624,8 @@ worker(void *arg) } while (!wtperf->stop) { + if (workload->pause != 0) + (void)sleep((unsigned int)workload->pause); /* * Generate the next key and setup operation specific * statistics tracking objects. @@ -603,10 +665,12 @@ worker(void *arg) generate_key(opts, key_buf, next_val); - /* - * Spread the data out around the multiple databases. - */ - cursor = cursors[map_key_to_table(wtperf->opts, next_val)]; + if (workload->table_index == INT32_MAX) + /* + * Spread the data out around the multiple databases. + */ + cursor = cursors[ + map_key_to_table(wtperf->opts, next_val)]; /* * Skip the first time we do an operation, when trk->ops @@ -642,7 +706,8 @@ worker(void *arg) * for several operations, confirming that the * next key is in the correct order. */ - ret = do_range_reads(wtperf, cursor); + ret = do_range_reads(wtperf, + cursor, workload->read_range); } if (ret == 0 || ret == WT_NOTFOUND) @@ -689,7 +754,7 @@ worker(void *arg) */ strncpy(value_buf, value, opts->value_sz_max - 1); - if (thread->workload->update_delta != 0) + if (workload->update_delta != 0) update_value_delta(thread); if (value_buf[0] == 'a') value_buf[0] = 'b'; @@ -806,7 +871,7 @@ op_err: if (ret == WT_ROLLBACK && ops_per_txn != 0) { /* Schedule the next operation */ if (++op == op_end) - op = thread->workload->ops; + op = workload->ops; /* * Decrement throttle ops and check if we should sleep @@ -843,7 +908,7 @@ run_mix_schedule_op(WORKLOAD *workp, int op, int64_t op_cnt) uint8_t *p, *end; /* Jump around the array to roughly spread out the operations. */ - jump = 100 / op_cnt; + jump = (int)(100 / op_cnt); /* * Find a read operation and replace it with another operation. This @@ -884,17 +949,6 @@ run_mix_schedule(WTPERF *wtperf, WORKLOAD *workp) opts = wtperf->opts; - /* Confirm reads, inserts, truncates and updates cannot all be zero. */ - if (workp->insert == 0 && workp->read == 0 && - workp->truncate == 0 && workp->update == 0) { - lprintf(wtperf, EINVAL, 0, "no operations scheduled"); - return (EINVAL); - } - - /* - * Handle truncate first - it's a special case that can't be used in - * a mixed workload. - */ if (workp->truncate != 0) { if (workp->insert != 0 || workp->read != 0 || workp->update != 0) { @@ -906,6 +960,12 @@ run_mix_schedule(WTPERF *wtperf, WORKLOAD *workp) return (0); } + /* Confirm reads, inserts and updates cannot all be zero. */ + if (workp->insert == 0 && workp->read == 0 && workp->update == 0) { + lprintf(wtperf, EINVAL, 0, "no operations scheduled"); + return (EINVAL); + } + /* * Check for a simple case where the thread is only doing insert or * update operations (because the default operation for a @@ -1238,7 +1298,8 @@ monitor(void *arg) /* Open the logging file. */ len = strlen(wtperf->monitor_dir) + 100; path = dmalloc(len); - snprintf(path, len, "%s/monitor", wtperf->monitor_dir); + testutil_check(__wt_snprintf( + path, len, "%s/monitor", wtperf->monitor_dir)); if ((fp = fopen(path, "w")) == NULL) { lprintf(wtperf, errno, 0, "%s", path); goto err; @@ -1593,6 +1654,9 @@ close_reopen(WTPERF *wtperf) opts = wtperf->opts; + if (opts->in_memory) + return (0); + if (!opts->readonly && !opts->reopen_connection) return (0); /* @@ -1875,19 +1939,19 @@ create_uris(WTPERF *wtperf) /* If there is only one table, just use the base name. */ wtperf->uris[i] = dmalloc(len); if (opts->table_count == 1) - snprintf(wtperf->uris[i], - len, "table:%s", opts->table_name); + testutil_check(__wt_snprintf(wtperf->uris[i], + len, "table:%s", opts->table_name)); else - snprintf(wtperf->uris[i], - len, "table:%s%05d", opts->table_name, i); + testutil_check(__wt_snprintf(wtperf->uris[i], + len, "table:%s%05d", opts->table_name, i)); } /* Create the log-like-table URI. */ len = strlen("table:") + strlen(opts->table_name) + strlen("_log_table") + 1; wtperf->log_table_uri = dmalloc(len); - snprintf( - wtperf->log_table_uri, len, "table:%s_log_table", opts->table_name); + testutil_check(__wt_snprintf(wtperf->log_table_uri, + len, "table:%s_log_table", opts->table_name)); } static int @@ -1909,7 +1973,8 @@ create_tables(WTPERF *wtperf) } for (i = 0; i < opts->table_count_idle; i++) { - snprintf(buf, 512, "%s_idle%05d", wtperf->uris[0], (int)i); + testutil_check(__wt_snprintf( + buf, 512, "%s_idle%05d", wtperf->uris[0], (int)i)); if ((ret = session->create( session, buf, opts->table_config)) != 0) { lprintf(wtperf, ret, 0, @@ -1938,8 +2003,9 @@ create_tables(WTPERF *wtperf) return (ret); } if (opts->index) { - snprintf(buf, 512, "index:%s:val_idx", - wtperf->uris[i] + strlen("table:")); + testutil_check(__wt_snprintf(buf, 512, + "index:%s:val_idx", + wtperf->uris[i] + strlen("table:"))); if ((ret = session->create( session, buf, "columns=(val)")) != 0) { lprintf(wtperf, ret, 0, @@ -2124,15 +2190,15 @@ start_all_runs(WTPERF *wtperf) */ len = strlen(wtperf->home) + 5; next_wtperf->home = dmalloc(len); - snprintf( - next_wtperf->home, len, "%s/D%02d", wtperf->home, (int)i); + testutil_check(__wt_snprintf( + next_wtperf->home, len, "%s/D%02d", wtperf->home, (int)i)); if (opts->create != 0) recreate_dir(next_wtperf->home); len = strlen(wtperf->monitor_dir) + 5; next_wtperf->monitor_dir = dmalloc(len); - snprintf(next_wtperf->monitor_dir, - len, "%s/D%02d", wtperf->monitor_dir, (int)i); + testutil_check(__wt_snprintf(next_wtperf->monitor_dir, + len, "%s/D%02d", wtperf->monitor_dir, (int)i)); if (opts->create != 0 && strcmp(next_wtperf->home, next_wtperf->monitor_dir) != 0) recreate_dir(next_wtperf->monitor_dir); @@ -2244,6 +2310,8 @@ start_run(WTPERF *wtperf) opts->checkpoint_threads, checkpoint_worker) != 0) goto err; } + if (opts->pre_load_data && (ret = pre_load_data(wtperf)) != 0) + goto err; /* Execute the workload. */ if ((ret = execute_workload(wtperf)) != 0) goto err; @@ -2361,11 +2429,11 @@ main(int argc, char *argv[]) { CONFIG_OPTS *opts; WTPERF *wtperf, _wtperf; - size_t req_len, sreq_len; + size_t pos, req_len, sreq_len; bool monitor_set; int ch, ret; const char *cmdflags = "C:h:m:O:o:T:"; - const char *config_opts; + const char *append_comma, *config_opts; char *cc_buf, *path, *sess_cfg, *tc_buf, *user_cconfig, *user_tconfig; /* The first WTPERF structure (from which all others are derived). */ @@ -2479,9 +2547,9 @@ main(int argc, char *argv[]) */ req_len = strlen(",async=(enabled=true,threads=)") + 4; wtperf->async_config = dmalloc(req_len); - snprintf(wtperf->async_config, req_len, + testutil_check(__wt_snprintf(wtperf->async_config, req_len, ",async=(enabled=true,threads=%" PRIu32 ")", - opts->async_threads); + opts->async_threads)); } if ((ret = config_compress(wtperf)) != 0) goto err; @@ -2502,53 +2570,96 @@ main(int argc, char *argv[]) __wt_stream_set_line_buffer(stdout); /* Concatenate non-default configuration strings. */ - if (opts->verbose > 1 || user_cconfig != NULL || - opts->session_count_idle > 0 || wtperf->compress_ext != NULL || - wtperf->async_config != NULL) { - req_len = strlen(debug_cconfig) + 20; - if (user_cconfig != NULL) - req_len += strlen(user_cconfig); - if (wtperf->async_config != NULL) - req_len += strlen(wtperf->async_config); - if (wtperf->compress_ext != NULL) - req_len += strlen(wtperf->compress_ext); + if (user_cconfig != NULL || opts->session_count_idle > 0 || + wtperf->compress_ext != NULL || wtperf->async_config != NULL || + opts->in_memory) { + req_len = 20; + req_len += wtperf->async_config != NULL ? + strlen(wtperf->async_config) : 0; + req_len += wtperf->compress_ext != NULL ? + strlen(wtperf->compress_ext) : 0; if (opts->session_count_idle > 0) { - sreq_len = strlen(",session_max=") + 6; + sreq_len = strlen("session_max=") + 6; req_len += sreq_len; sess_cfg = dmalloc(sreq_len); - snprintf(sess_cfg, sreq_len, - ",session_max=%" PRIu32, + testutil_check(__wt_snprintf(sess_cfg, sreq_len, + "session_max=%" PRIu32, opts->session_count_idle + - wtperf->workers_cnt + opts->populate_threads + 10); + wtperf->workers_cnt + opts->populate_threads + 10)); } + req_len += opts->in_memory ? strlen("in_memory=true") : 0; + req_len += user_cconfig != NULL ? strlen(user_cconfig) : 0; cc_buf = dmalloc(req_len); - snprintf(cc_buf, req_len, "%s,%s,%s,%s,%s", - wtperf->async_config ? wtperf->async_config : "", - wtperf->compress_ext ? wtperf->compress_ext : "", - opts->verbose > 1 ? debug_cconfig : "", - sess_cfg != NULL ? sess_cfg : "", - user_cconfig != NULL ? user_cconfig : ""); - if (strlen(cc_buf) && (ret = + + pos = 0; + append_comma = ""; + if (wtperf->async_config != NULL && + strlen(wtperf->async_config) != 0) { + testutil_check(__wt_snprintf_len_incr( + cc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, wtperf->async_config)); + append_comma = ","; + } + if (wtperf->compress_ext != NULL && + strlen(wtperf->compress_ext) != 0) { + testutil_check(__wt_snprintf_len_incr( + cc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, wtperf->compress_ext)); + append_comma = ","; + } + if (opts->in_memory) { + testutil_check(__wt_snprintf_len_incr( + cc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, "in_memory=true")); + append_comma = ","; + } + if (sess_cfg != NULL && strlen(sess_cfg) != 0) { + testutil_check(__wt_snprintf_len_incr( + cc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, sess_cfg)); + append_comma = ","; + } + if (user_cconfig != NULL && strlen(user_cconfig) != 0) { + testutil_check(__wt_snprintf_len_incr( + cc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, user_cconfig)); + } + + if (strlen(cc_buf) != 0 && (ret = config_opt_name_value(wtperf, "conn_config", cc_buf)) != 0) goto err; } - if (opts->verbose > 1 || opts->index || + if (opts->index || user_tconfig != NULL || wtperf->compress_table != NULL) { - req_len = strlen(debug_tconfig) + 20; - if (user_tconfig != NULL) - req_len += strlen(user_tconfig); - if (wtperf->compress_table != NULL) - req_len += strlen(wtperf->compress_table); - if (opts->index) - req_len += strlen(INDEX_COL_NAMES); + req_len = 20; + req_len += wtperf->compress_table != NULL ? + strlen(wtperf->compress_table) : 0; + req_len += opts->index ? strlen(INDEX_COL_NAMES) : 0; + req_len += user_tconfig != NULL ? strlen(user_tconfig) : 0; tc_buf = dmalloc(req_len); - snprintf(tc_buf, req_len, "%s,%s,%s,%s", - opts->index ? INDEX_COL_NAMES : "", - wtperf->compress_table != NULL ? - wtperf->compress_table : "", - opts->verbose > 1 ? debug_tconfig : "", - user_tconfig ? user_tconfig : ""); - if (strlen(tc_buf) && (ret = + + pos = 0; + append_comma = ""; + if (wtperf->compress_table != NULL && + strlen(wtperf->compress_table) != 0) { + testutil_check(__wt_snprintf_len_incr( + tc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, wtperf->compress_table)); + append_comma = ","; + } + if (opts->index) { + testutil_check(__wt_snprintf_len_incr( + tc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, INDEX_COL_NAMES)); + append_comma = ","; + } + if (user_tconfig != NULL && strlen(user_tconfig) != 0) { + testutil_check(__wt_snprintf_len_incr( + tc_buf + pos, req_len - pos, &pos, "%s%s", + append_comma, user_tconfig)); + } + + if (strlen(tc_buf) != 0 && (ret = config_opt_name_value(wtperf, "table_config", tc_buf)) != 0) goto err; } @@ -2556,8 +2667,9 @@ main(int argc, char *argv[]) req_len = strlen(opts->table_config) + strlen(LOG_PARTIAL_CONFIG) + 1; wtperf->partial_config = dmalloc(req_len); - snprintf(wtperf->partial_config, req_len, "%s%s", - opts->table_config, LOG_PARTIAL_CONFIG); + testutil_check(__wt_snprintf( + wtperf->partial_config, req_len, "%s%s", + opts->table_config, LOG_PARTIAL_CONFIG)); } /* * Set the config for reopen. If readonly add in that string. @@ -2570,11 +2682,12 @@ main(int argc, char *argv[]) req_len = strlen(opts->conn_config) + 1; wtperf->reopen_config = dmalloc(req_len); if (opts->readonly) - snprintf(wtperf->reopen_config, req_len, "%s%s", - opts->conn_config, READONLY_CONFIG); + testutil_check(__wt_snprintf( + wtperf->reopen_config, req_len, "%s%s", + opts->conn_config, READONLY_CONFIG)); else - snprintf(wtperf->reopen_config, - req_len, "%s", opts->conn_config); + testutil_check(__wt_snprintf( + wtperf->reopen_config, req_len, "%s", opts->conn_config)); /* Sanity-check the configuration. */ if ((ret = config_sanity(wtperf)) != 0) @@ -2587,7 +2700,8 @@ main(int argc, char *argv[]) /* Write a copy of the config. */ req_len = strlen(wtperf->home) + strlen("/CONFIG.wtperf") + 1; path = dmalloc(req_len); - snprintf(path, req_len, "%s/CONFIG.wtperf", wtperf->home); + testutil_check(__wt_snprintf( + path, req_len, "%s/CONFIG.wtperf", wtperf->home)); config_opt_log(opts, path); free(path); @@ -2712,7 +2826,8 @@ recreate_dir(const char *name) len = strlen(name) * 2 + 100; buf = dmalloc(len); - (void)snprintf(buf, len, "rm -rf %s && mkdir %s", name, name); + testutil_check(__wt_snprintf( + buf, len, "rm -rf %s && mkdir %s", name, name)); testutil_checkfmt(system(buf), "system: %s", buf); free(buf); } @@ -2779,14 +2894,43 @@ static uint64_t wtperf_rand(WTPERF_THREAD *thread) { CONFIG_OPTS *opts; + WT_CURSOR *rnd_cursor; WTPERF *wtperf; double S1, S2, U; uint64_t rval; + int ret; + char *key_buf; wtperf = thread->wtperf; opts = wtperf->opts; /* + * If we have a random cursor set up then use it. + */ + if ((rnd_cursor = thread->rand_cursor) != NULL) { + if ((ret = rnd_cursor->next(rnd_cursor)) != 0) { + lprintf(wtperf, ret, 0, "worker: rand next failed"); + /* 0 is outside the expected range. */ + return (0); + } + if ((ret = rnd_cursor->get_key(rnd_cursor, &key_buf)) != 0) { + lprintf(wtperf, ret, 0, + "worker: rand next key retrieval"); + return (0); + } + /* + * Resetting the cursor is not fatal. We still return the + * value we retrieved above. We do it so that we don't + * leave a cursor positioned. + */ + if ((ret = rnd_cursor->reset(rnd_cursor)) != 0) + lprintf(wtperf, ret, 0, + "worker: rand cursor reset failed"); + extract_key(key_buf, &rval); + return (rval); + } + + /* * Use WiredTiger's random number routine: it's lock-free and fairly * good. */ |