Throttle new async

README.md

@@ -85,11 +85,11 @@ target/benchmarks -h 127.0.0.1 -p 3000 -n test -k 1000000 -o B1400 -w RU,80 -g 2
 ```
 
 To:
- * Benchmark asynchronous methods using 1 event loop.
+ * Benchmark asynchronous methods using 4 event loops.
  * Limit the maximum number of concurrent commands to 50.
  * Use and 50% read 50% write pattern.
 ```sh
-target/benchmarks -h 127.0.0.1 -p 3000 -n test -k 1000000 -o S:50 -w RU,50 --async --asyncMaxCommands 50 --eventLoops 1
+target/benchmarks -h 127.0.0.1 -p 3000 -n test -k 1000000 -o S:50 -w RU,50 --async --asyncMaxCommands 50 --threads 4
 ```
 
 Command line usage can be read with:

src/include/benchmark.h

@@ -96,7 +96,6 @@ typedef struct args_s {
 	int async_max_conns_per_node;
 	bool durable_deletes;
 	int async_max_commands;
-	int event_loop_capacity;
 	as_config_tls tls;
 	char* tls_name;
 	as_auth_mode auth_mode;
@@ -166,6 +165,12 @@ typedef struct threaddata_s {
 	// which workload stage we're currrently on
 	_Atomic(uint32_t) stage_idx;
 
+	// For async linear workloads
+	_Atomic(uint64_t) current_key;
+
+	// the stopping point for async linear workloads
+	uint64_t end_key;
+
 	/*
 	 * note: to stop threads, tdata->finished must be set before tdata->do_work
 	 * to prevent deadlocking

src/main/benchmark.c

@@ -184,7 +184,7 @@ connect_to_server(args_t* args, aerospike* client)
 {
 	if (stages_contain_async(&args->stages)) {
 #if AS_EVENT_LIB_DEFINED
-		if (! as_event_create_loops(args->event_loop_capacity)) {
+		if (! as_event_create_loops(args->transaction_worker_threads)) {
 			blog_error("Failed to create asynchronous event loops\n");
 			return 2;
 		}
@@ -322,10 +322,13 @@ init_tdata(const args_t* args, cdata_t* cdata, thr_coord_t* coord,
 	tdata->t_idx = t_idx;
 	// always start on the first stage
 	atomic_init(&tdata->stage_idx, 0);
+	atomic_init(&tdata->current_key, args->start_key);
 
 	atomic_init(&tdata->do_work, true);
 	atomic_init(&tdata->finished, false);
 
+	tdata->end_key = UINT64_MAX;
+
 	as_policies* p = &tdata->policies;
 	as_policies_init(p);
 

src/main/benchmark_init.c

@@ -43,7 +43,7 @@
 // Typedefs & constants.
 //
 
-static const char* short_options = "vVh:p:U:P::n:s:b:K:k:o:Re:t:w:z:g:T:dLSC:N:B:M:Y:Dac:W:";
+static const char* short_options = "vVh:p:U:P::n:s:b:K:k:o:Re:t:w:z:g:T:dLSC:N:B:M:Y:Dac:";
 
 #define WARN_MSG 0x40000000
 
@@ -544,6 +544,7 @@ print_usage(const char* program)
 
 	printf("-z --threads <count> # Default: 16\n");
 	printf("   Load generating thread count.\n");
+	printf("   This also sets the number of event loops used with --async.\n");
 	printf("\n");
 
 	printf("-g --throughput <tps> # Default: 0\n");
@@ -745,17 +746,14 @@ print_usage(const char* program)
 
 	printf("-a --async # Default: synchronous mode\n");
 	printf("   Enable asynchronous mode.\n");
+	printf("   Use --threads and --async-max-commands to tune async performance.\n");
 	printf("\n");
 
 	printf("-c --async-max-commands <command count> # Default: 50\n");
 	printf("   Maximum number of concurrent asynchronous commands that are active at any point\n");
 	printf("   in time.\n");
 	printf("\n");
 
-	printf("-W --event-loops <thread count> # Default: 1\n");
-	printf("   Number of event loops (or selector threads) when running in asynchronous mode.\n");
-	printf("\n");
-
 	printf("   --tls-enable         # Default: TLS disabled\n");
 	printf("   Enable TLS.\n");
 	printf("\n");
@@ -956,7 +954,7 @@ print_args(args_t* args)
 	printf("async min conns per node: %d\n", args->async_min_conns_per_node);
 	printf("async max conns per node: %d\n", args->async_max_conns_per_node);
 	printf("async max commands:       %d\n", args->async_max_commands);
-	printf("event loops:              %d\n", args->event_loop_capacity);
+	printf("event loops:              %d\n", args->transaction_worker_threads);
 
 	if (args->tls.enable) {
 		printf("TLS:                    enabled\n");
@@ -1156,9 +1154,9 @@ validate_args(args_t* args)
 		return 1;
 	}
 
-	if (args->event_loop_capacity <= 0 || args->event_loop_capacity > 1000) {
-		printf("Invalid event-loops: %d  Valid values: [1-1000]\n",
-				args->event_loop_capacity);
+	if (args->transaction_worker_threads <= 0 || args->transaction_worker_threads > 1000) {
+		printf("Invalid threads/event_loops: %d  Valid values: [1-1000]\n",
+				args->transaction_worker_threads);
 		return 1;
 	}
 	return 0;
@@ -1711,10 +1709,6 @@ set_args(int argc, char * const* argv, args_t* args)
 				args->async_max_commands = atoi(optarg);
 				break;
 
-			case 'W':
-				args->event_loop_capacity = atoi(optarg);
-				break;
-
 			case BENCH_OPT_SEND_KEY:
 				args->key = AS_POLICY_KEY_SEND;
 				break;
@@ -1876,7 +1870,6 @@ _load_defaults(args_t* args)
 	args->async_min_conns_per_node = 0;
 	args->async_max_conns_per_node = 300;
 	args->async_max_commands = 50;
-	args->event_loop_capacity = 1;
 	memset(&args->tls, 0, sizeof(as_config_tls));
 	args->tls_name = NULL;
 	args->auth_mode = AS_AUTH_INTERNAL;

src/main/coordinator.c

@@ -135,7 +135,7 @@ coordinator_worker(void* udata)
 		stage_t* stage = &cdata->stages.stages[stage_idx];
 		fprint_stage(stdout, &cdata->stages, stage_idx);
 
-		if (stage->workload.type == WORKLOAD_TYPE_I) {
+		if (stage->workload.type == WORKLOAD_TYPE_I && stage->batch_write_size > 1) {
 			uint64_t nkeys = stage->key_end - stage->key_start;
 			if (nkeys % stage->batch_write_size != 0) {
 				blog_warn("--keys is not divisible by --batch-write-size so more than "
@@ -155,7 +155,7 @@ coordinator_worker(void* udata)
 			}
 		}
 
-		if (stage->workload.type == WORKLOAD_TYPE_D) {
+		if (stage->workload.type == WORKLOAD_TYPE_D && stage->batch_delete_size > 1) {
 			uint64_t nkeys = stage->key_end - stage->key_start;
 			if (nkeys % stage->batch_delete_size != 0) {
 				blog_warn("--keys is not divisible by --batch-delete-size so some records "