bdevperf: do not start tests until RPC is sent

Starting with this patch it is possible to issue
"perform_tests" RPC to bdevperf application.
It will allow to configure the bdevs after startup,
but before starting the tests.

bdevperf in addition to usual cmd line params for tests,
need to be started with '-z' argument.
After that it is possible to start test using
'./bdevperf.py perform_tests'.
Tests can be issued multiple times in the same app run.

At this time the RPC does not take any arguments
for the tests. All are assumed to be set in
the command line.

This is series for adding RPC to bdevperf app.

Change-Id: If71853b1aa742f9cbf3d65c8d694a0437aad4500
Signed-off-by: Tomasz Zawadzki <tomasz.zawadzki@intel.com>
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/458598
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Konrad Sztyber <konrad.sztyber@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Paul Luse <paul.e.luse@intel.com>
Reviewed-by: Darek Stojaczyk <dariusz.stojaczyk@intel.com>

test/bdev/bdevperf/bdevperf.c

@@ -42,6 +42,7 @@
 #include "spdk/util.h"
 #include "spdk/thread.h"
 #include "spdk/string.h"
+#include "spdk/rpc.h"
 
 struct bdevperf_task {
 	struct iovec			iov;
@@ -79,11 +80,14 @@ static unsigned g_master_core;
 static int g_time_in_sec;
 static bool g_mix_specified;
 static const char *g_target_bdev_name;
+static bool g_wait_for_tests = false;
+static struct spdk_jsonrpc_request *g_request = NULL;
 
 static struct spdk_poller *g_perf_timer = NULL;
 
 static void bdevperf_submit_single(struct io_target *target, struct bdevperf_task *task);
 static void performance_dump(uint64_t io_time_in_usec, uint64_t ema_period);
+static void rpc_perform_tests_cb(int rc);
 
 struct io_target {
 	char				*name;
@@ -395,6 +399,7 @@ static void
 end_run(void *arg1, void *arg2)
 {
 	struct io_target *target = arg1;
+	int rc = 0;
 
 	spdk_put_io_channel(target->ch);
 	spdk_bdev_close(target->bdev_desc);
@@ -417,9 +422,13 @@ end_run(void *arg1, void *arg2)
 		}
 
 		if (g_run_failed) {
-			spdk_app_stop(1);
+			rc = 1;
+		}
+
+		if (g_request && !g_shutdown) {
+			rpc_perform_tests_cb(rc);
 		} else {
-			spdk_app_stop(0);
+			spdk_app_stop(rc);
 		}
 	}
 }
@@ -840,6 +849,7 @@ bdevperf_usage(void)
 	printf("\t\t(only valid with -S)\n");
 	printf(" -S <period>               show performance result in real time every <period> seconds\n");
 	printf(" -T <target>               target bdev\n");
+	printf(" -z                        start bdevperf, but wait for RPC to start tests\n");
 }
 
 /*
@@ -999,6 +1009,13 @@ ret:
 static int
 verify_test_params(struct spdk_app_opts *opts)
 {
+	/* When RPC is used for starting tests and
+	 * no rpc_addr was configured for the app,
+	 * use the default address. */
+	if (g_wait_for_tests && opts->rpc_addr == NULL) {
+		opts->rpc_addr = SPDK_DEFAULT_RPC_ADDR;
+	}
+
 	if (g_queue_depth <= 0) {
 		spdk_app_usage();
 		bdevperf_usage();
@@ -1180,6 +1197,11 @@ bdevperf_run(void *arg1)
 
 	g_master_core = spdk_env_get_current_core();
 
+	if (g_wait_for_tests) {
+		/* Do not perform any tests until RPC is received */
+		return;
+	}
+
 	bdevperf_construct_targets();
 
 	if (g_target_count == 0) {
@@ -1247,6 +1269,8 @@ bdevperf_parse_arg(int ch, char *arg)
 		g_workload_type = optarg;
 	} else if (ch == 'T') {
 		g_target_bdev_name = optarg;
+	} else if (ch == 'z') {
+		g_wait_for_tests = true;
 	} else {
 		tmp = spdk_strtoll(optarg, 10);
 		if (tmp < 0) {
@@ -1287,6 +1311,67 @@ bdevperf_parse_arg(int ch, char *arg)
 	return 0;
 }
 
+static void
+rpc_perform_tests_cb(int rc)
+{
+	struct spdk_json_write_ctx *w;
+	struct spdk_jsonrpc_request *request = g_request;
+
+	g_request = NULL;
+
+	if (rc == 0) {
+		w = spdk_jsonrpc_begin_result(request);
+		if (w == NULL) {
+			return;
+		}
+		spdk_json_write_uint32(w, rc);
+		spdk_jsonrpc_end_result(request, w);
+	} else {
+		spdk_jsonrpc_send_error_response_fmt(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						     "bdevperf failed with error %s", spdk_strerror(-rc));
+	}
+
+	bdevperf_free_targets();
+}
+
+static void
+rpc_perform_tests(struct spdk_jsonrpc_request *request, const struct spdk_json_val *params)
+{
+	int rc;
+
+	if (params != NULL) {
+		spdk_jsonrpc_send_error_response(request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
+						 "perform_tests method requires no parameters");
+		return;
+	}
+	if (g_request != NULL) {
+		fprintf(stderr, "Another test is already in progress.\n");
+		spdk_jsonrpc_send_error_response(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						 spdk_strerror(-EINPROGRESS));
+		return;
+	}
+	g_request = request;
+
+	bdevperf_construct_targets();
+	if (g_target_count == 0) {
+		g_request = NULL;
+		fprintf(stderr, "No valid bdevs found.\n");
+		spdk_jsonrpc_send_error_response(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						 spdk_strerror(-ENODEV));
+		return;
+	}
+
+	rc = bdevperf_test();
+	if (rc) {
+		g_request = NULL;
+		bdevperf_free_targets();
+		spdk_jsonrpc_send_error_response_fmt(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						     "Could not perform tests due to error: %s", spdk_strerror(-rc));
+		return;
+	}
+}
+SPDK_RPC_REGISTER("perform_tests", rpc_perform_tests, SPDK_RPC_RUNTIME)
+
 int
 main(int argc, char **argv)
 {
@@ -1306,7 +1391,7 @@ main(int argc, char **argv)
 	g_time_in_sec = 0;
 	g_mix_specified = false;
 
-	if ((rc = spdk_app_parse_args(argc, argv, &opts, "q:o:t:w:M:P:S:T:", NULL,
+	if ((rc = spdk_app_parse_args(argc, argv, &opts, "zq:o:t:w:M:P:S:T:", NULL,
 				      bdevperf_parse_arg, bdevperf_usage)) !=
 	    SPDK_APP_PARSE_ARGS_SUCCESS) {
 		return rc;

test/bdev/bdevperf/bdevperf.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+
+import logging
+import argparse
+import sys
+import shlex
+
+try:
+    from rpc.client import print_dict, JSONRPCException
+    import rpc
+except ImportError:
+    print("SPDK RPC library missing. Please add spdk/scripts/ directory to PYTHONPATH:")
+    print("'export PYTHONPATH=$PYTHONPATH:./spdk/scripts/'")
+    exit(1)
+
+try:
+    from shlex import quote
+except ImportError:
+    from pipes import quote
+
+
+def print_array(a):
+    print(" ".join((quote(v) for v in a)))
+
+
+def perform_tests_func(client):
+    """Perform bdevperf tests according to command line arguments when application was started.
+
+    Args:
+        none
+
+    Returns:
+        On success, 0 is returned. On error, -1 is returned.
+    """
+    params = {}
+    return client.call('perform_tests', params)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description='SPDK RPC command line interface. NOTE: spdk/scripts/ is expected in PYTHONPATH')
+    parser.add_argument('-s', dest='server_addr',
+                        help='RPC domain socket path or IP address', default='/var/tmp/spdk.sock')
+    parser.add_argument('-p', dest='port',
+                        help='RPC port number (if server_addr is IP address)',
+                        default=5260, type=int)
+    parser.add_argument('-t', dest='timeout',
+                        help='Timeout as a floating point number expressed in seconds waiting for response. Default: 60.0',
+                        default=60.0, type=float)
+    parser.add_argument('-v', dest='verbose', action='store_const', const="INFO",
+                        help='Set verbose mode to INFO', default="ERROR")
+    parser.add_argument('--verbose', dest='verbose', choices=['DEBUG', 'INFO', 'ERROR'],
+                        help="""Set verbose level. """)
+    subparsers = parser.add_subparsers(help='RPC methods')
+
+    def perform_tests(args):
+        print_dict(perform_tests_func(args.client))
+
+    p = subparsers.add_parser('perform_tests', help='Perform bdevperf tests')
+    p.set_defaults(func=perform_tests)
+
+    def call_rpc_func(args):
+        try:
+            args.func(args)
+        except JSONRPCException as ex:
+            print(ex.message)
+            exit(1)
+
+    def execute_script(parser, client, fd):
+        for rpc_call in map(str.rstrip, fd):
+            if not rpc_call.strip():
+                continue
+            args = parser.parse_args(shlex.split(rpc_call))
+            args.client = client
+            call_rpc_func(args)
+
+    args = parser.parse_args()
+    args.client = rpc.client.JSONRPCClient(args.server_addr, args.port, args.timeout, log_level=getattr(logging, args.verbose.upper()))
+    if hasattr(args, 'func'):
+        call_rpc_func(args)
+    elif sys.stdin.isatty():
+        # No arguments and no data piped through stdin
+        parser.print_help()
+        exit(1)
+    else:
+        execute_script(parser, args.client, sys.stdin)