WT-2876 Add an oplog-like ability to wtperf utility. (#2999)

This allows us to partially simulate secondary replica nodes.

5 files changed, 72 insertions, 5 deletions

diff --git a/bench/wtperf/runners/mongodb-secondary-apply.wtperf b/bench/wtperf/runners/mongodb-secondary-apply.wtperf
new file mode 100644
index 00000000000..f9e41184f95
--- /dev/null
+++ b/bench/wtperf/runners/mongodb-secondary-apply.wtperf
@@ -0,0 +1,21 @@
+# Simulate the MongoDB oplog apply threads on a secondary.
+conn_config="cache_size=10GB,session_max=1000,eviction=(threads_min=4,threads_max=4),log=(enabled=false),transaction_sync=(enabled=false),checkpoint_sync=true,checkpoint=(wait=60),statistics=(fast),statistics_log=(json,wait=1)"
+table_config="allocation_size=4k,memory_page_max=5MB,prefix_compression=false,split_pct=75,leaf_page_max=32k,internal_page_max=16k,type=file"
+# Spread the workload out over several tables.
+table_count=4
+# We like compression.
+compression=snappy
+icount=1000
+populate_threads=1
+reopen_connection=true
+log_like_table=true
+report_interval=5
+run_time=360
+# Configure multiple threads doing a limited number of operations each. Enclose
+# a few operations in an explicit transaction to simulate MongoDB apply on a
+# secondary.
+threads=((count=16,throttle=1000,inserts=1,ops_per_txn=3))
+# Configure a moderately large value size
+value_sz=1800
+sample_interval=5
+sample_rate=1
diff --git a/bench/wtperf/wtperf.c b/bench/wtperf/wtperf.c
index c689fca80f7..bef5f6b22b5 100644
--- a/bench/wtperf/wtperf.c
+++ b/bench/wtperf/wtperf.c
@@ -37,6 +37,7 @@ static const CONFIG default_cfg = {
 	NULL,				/* partial logging */
 	NULL,				/* reopen config */
 	NULL,				/* base_uri */
+	NULL,				/* log_table_uri */
 	NULL,				/* uris */
 	NULL,				/* conn */
 	NULL,				/* logf */
@@ -55,6 +56,7 @@ static const CONFIG default_cfg = {
 	0,				/* truncate operations */
 	0,				/* update operations */
 	0,				/* insert key */
+	0,				/* log like table key */
 	0,				/* checkpoint in progress */
 	0,				/* thread error */
 	0,				/* notify threads to stop */
@@ -512,11 +514,11 @@ worker(void *arg)
 	CONFIG_THREAD *thread;
 	TRACK *trk;
 	WT_CONNECTION *conn;
-	WT_CURSOR **cursors, *cursor, *tmp_cursor;
+	WT_CURSOR **cursors, *cursor, *log_table_cursor, *tmp_cursor;
 	WT_SESSION *session;
 	size_t i;
 	int64_t ops, ops_per_txn;
-	uint64_t next_val, usecs;
+	uint64_t log_id, next_val, usecs;
 	uint8_t *op, *op_end;
 	int measure_latency, ret, truncated;
 	char *value_buf, *key_buf, *value;
@@ -560,6 +562,16 @@ worker(void *arg)
 			goto err;
 		}
 	}
+	if (cfg->log_like_table) {
+		if ((ret = session->open_cursor(session,
+		    cfg->log_table_uri, NULL, NULL, &log_table_cursor)) != 0) {
+			lprintf(cfg, ret, 0,
+			    "worker: WT_SESSION.open_cursor: %s",
+			    cfg->log_table_uri);
+			goto err;
+		}
+	}
+
 	/* Setup the timer for throttling. */
 	if (thread->workload->throttle != 0)
 		setup_throttle(thread);
@@ -575,7 +587,7 @@ worker(void *arg)
 	op = thread->workload->ops;
 	op_end = op + sizeof(thread->workload->ops);
 
-	if (ops_per_txn != 0 &&
+	if ((ops_per_txn != 0 || cfg->log_like_table) &&
 		(ret = session->begin_transaction(session, NULL)) != 0) {
 		lprintf(cfg, ret, 0, "First transaction begin failed");
 		goto err;
@@ -768,6 +780,20 @@ op_err:			if (ret == WT_ROLLBACK && ops_per_txn != 0) {
 			goto err;		/* can't happen */
 		}
 
+		/* Update the log-like table. */
+		if (cfg->log_like_table &&
+		    (*op != WORKER_READ && *op != WORKER_TRUNCATE)) {
+			log_id = __wt_atomic_add64(&cfg->log_like_table_key, 1);
+			log_table_cursor->set_key(log_table_cursor, log_id);
+			log_table_cursor->set_value(
+			    log_table_cursor, value_buf);
+			if ((ret =
+			    log_table_cursor->insert(log_table_cursor)) != 0) {
+				lprintf(cfg, ret, 0, "Cursor insert failed");
+				goto err;
+			}
+		}
+
 		/* Release the cursor, if we have multiple tables. */
 		if (cfg->table_count > 1 && ret == 0 &&
 		    *op != WORKER_INSERT && *op != WORKER_INSERT_RMW) {
@@ -793,8 +819,12 @@ op_err:			if (ret == WT_ROLLBACK && ops_per_txn != 0) {
 			++trk->ops;
 		}
 
-		/* Commit our work if configured for explicit transactions */
-		if (ops_per_txn != 0 && ops++ % ops_per_txn == 0) {
+		/*
+		 * Commit the transaction if grouping operations together
+		 * or tracking changes in our log table.
+		 */
+		if ((cfg->log_like_table && ops_per_txn == 0) ||
+		    (ops_per_txn != 0 && ops++ % ops_per_txn == 0)) {
 			if ((ret = session->commit_transaction(
 			    session, NULL)) != 0) {
 				lprintf(cfg, ret, 0,
@@ -819,6 +849,7 @@ op_err:			if (ret == WT_ROLLBACK && ops_per_txn != 0) {
 		 */
 		if (--thread->throttle_cfg.ops_count == 0)
 			worker_throttle(thread);
+
 	}
 
 	if ((ret = session->close(session, NULL)) != 0) {
@@ -1879,6 +1910,10 @@ create_uris(CONFIG *cfg)
 		else
 			sprintf(uri, "%s%05d", cfg->base_uri, i);
 	}
+
+	/* Create the log-like-table URI. */
+	cfg->log_table_uri = dcalloc(base_uri_len + 11, 1);
+	sprintf(cfg->log_table_uri, "%s_log_table", cfg->base_uri);
 }
 
 static int
@@ -1905,6 +1940,11 @@ create_tables(CONFIG *cfg)
 			return (ret);
 		}
 	}
+	if (cfg->log_like_table && (ret = session->create(session,
+	    cfg->log_table_uri, "key_format=Q,value_format=S")) != 0) {
+		lprintf(cfg, ret, 0, "Error creating log table %s", buf);
+		return (ret);
+	}
 
 	for (i = 0; i < cfg->table_count; i++) {
 		if (cfg->log_partial && i > 0) {
diff --git a/bench/wtperf/wtperf.h b/bench/wtperf/wtperf.h
index 9d6ef2f56ea..85a821b45eb 100644
--- a/bench/wtperf/wtperf.h
+++ b/bench/wtperf/wtperf.h
@@ -121,6 +121,7 @@ struct __config {			/* Configuration structure */
 	char *partial_config;		/* Config string for partial logging */
 	char *reopen_config;		/* Config string for conn reopen */
 	char *base_uri;			/* Object URI */
+	char *log_table_uri;		/* URI for log table */
 	char **uris;			/* URIs if multiple tables */
 
 	WT_CONNECTION *conn;		/* Database connection */
@@ -151,6 +152,7 @@ struct __config {			/* Configuration structure */
 	uint64_t update_ops;		/* update operations */
 
 	uint64_t insert_key;		/* insert key */
+	uint64_t log_like_table_key;	/* used to allocate IDs for log table */
 
 	volatile int ckpt;		/* checkpoint in progress */
 	volatile int error;		/* thread error */
diff --git a/bench/wtperf/wtperf_opt.i b/bench/wtperf/wtperf_opt.i
index f1f26bd0d01..17517ffe477 100644
--- a/bench/wtperf/wtperf_opt.i
+++ b/bench/wtperf/wtperf_opt.i
@@ -120,6 +120,8 @@ DEF_OPT_AS_BOOL(insert_rmw, 0,
     "execute a read prior to each insert in workload phase")
 DEF_OPT_AS_UINT32(key_sz, 20, "key size")
 DEF_OPT_AS_BOOL(log_partial, 0, "perform partial logging on first table only.")
+DEF_OPT_AS_BOOL(log_like_table, 0,
+    "Append all modification operations to another shared table.")
 DEF_OPT_AS_UINT32(min_throughput, 0,
     "notify if any throughput measured is less than this amount. "
     "Aborts or prints warning based on min_throughput_fatal setting. "
diff --git a/src/docs/wtperf.dox b/src/docs/wtperf.dox
index c794ee2f2d7..a49d0d9f871 100644
--- a/src/docs/wtperf.dox
+++ b/src/docs/wtperf.dox
@@ -185,6 +185,8 @@ execute a read prior to each insert in workload phase
 key size
 @par log_partial (boolean, default=false)
 perform partial logging on first table only.
+@par log_like_table (boolean, default=false)
+Append all modification operations to another shared table.
 @par min_throughput (unsigned int, default=0)
 notify if any throughput measured is less than this amount. Aborts or
 prints warning based on min_throughput_fatal setting. Requires