rpc: Add thread_set_cpumask RPC

Add an new JSON RPC thread_set_cpumask to control CPU affinity of
the thread dynamically from outside of SPDK application.

Signed-off-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Change-Id: I6b6b794405f612200a2c3cb27dc4fc1ad2e88bec
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/501
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Aleksey Marchuk <alexeymar@mellanox.com>

doc/jsonrpc.md

@@ -574,6 +574,48 @@ Example response:
 }
 ~~~
 
+## thread_set_cpumask {#rpc_thread_set_cpumask}
+
+Set the cpumask of the thread to the specified value. The thread may be migrated
+to one of the specified CPUs.
+
+### Parameters
+
+Name                    | Optional | Type        | Description
+----------------------- | -------- | ----------- | -----------
+id                      | Required | string      | Thread ID
+cpumask                 | Required | string      | Cpumask for this thread
+
+### Response
+
+Completion status of the operation is returned as a boolean.
+
+### Example
+
+Example request:
+
+~~~
+{
+  "jsonrpc": "2.0",
+  "method": "thread_set_cpumask",
+  "id": 1,
+  "params": {
+    "id": "1",
+    "cpumask": "1"
+  }
+}
+~~~
+
+Example response:
+
+~~~
+{
+  "jsonrpc": "2.0",
+  "id": 1,
+  "result": true
+}
+~~~
+
 # Block Device Abstraction Layer {#jsonrpc_components_bdev}
 
 ## bdev_set_options {#rpc_bdev_set_options}

module/event/rpc/app_rpc.c

@@ -292,3 +292,111 @@ spdk_rpc_framework_get_reactors(struct spdk_jsonrpc_request *request,
 }
 
 SPDK_RPC_REGISTER("framework_get_reactors", spdk_rpc_framework_get_reactors, SPDK_RPC_RUNTIME)
+
+struct rpc_thread_set_cpumask_ctx {
+	struct spdk_jsonrpc_request *request;
+	struct spdk_cpuset cpumask;
+	int status;
+	struct spdk_thread *orig_thread;
+};
+
+static void
+rpc_thread_set_cpumask_done(void *_ctx)
+{
+	struct rpc_thread_set_cpumask_ctx *ctx = _ctx;
+	struct spdk_json_write_ctx *w;
+
+	if (ctx->status == 0) {
+		w = spdk_jsonrpc_begin_result(ctx->request);
+		spdk_json_write_bool(w, true);
+		spdk_jsonrpc_end_result(ctx->request, w);
+	} else {
+		spdk_jsonrpc_send_error_response(ctx->request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						 spdk_strerror(-ctx->status));
+	}
+
+	free(ctx);
+}
+
+static void
+rpc_thread_set_cpumask(void *_ctx)
+{
+	struct rpc_thread_set_cpumask_ctx *ctx = _ctx;
+
+	ctx->status = spdk_thread_set_cpumask(&ctx->cpumask);
+
+	spdk_thread_send_msg(ctx->orig_thread, rpc_thread_set_cpumask_done, ctx);
+}
+
+struct rpc_thread_set_cpumask {
+	uint64_t id;
+	char *cpumask;
+};
+
+static const struct spdk_json_object_decoder rpc_thread_set_cpumask_decoders[] = {
+	{"id", offsetof(struct rpc_thread_set_cpumask, id), spdk_json_decode_uint64},
+	{"cpumask", offsetof(struct rpc_thread_set_cpumask, cpumask), spdk_json_decode_string},
+};
+
+static void
+spdk_rpc_thread_set_cpumask(struct spdk_jsonrpc_request *request,
+			    const struct spdk_json_val *params)
+{
+	struct rpc_thread_set_cpumask req = {};
+	struct rpc_thread_set_cpumask_ctx *ctx;
+	struct spdk_thread *thread;
+	int rc;
+
+	ctx = calloc(1, sizeof(*ctx));
+	if (ctx == NULL) {
+		SPDK_ERRLOG("Memory allocation failed\n");
+		spdk_jsonrpc_send_error_response(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						 "Memory allocation failed");
+		return;
+	}
+
+	if (spdk_json_decode_object(params, rpc_thread_set_cpumask_decoders,
+				    SPDK_COUNTOF(rpc_thread_set_cpumask_decoders),
+				    &req)) {
+		SPDK_ERRLOG("spdk_json_decode_object failed\n");
+		spdk_jsonrpc_send_error_response(request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
+						 "spdk_json_decode_object failed");
+		goto err;
+	}
+
+	thread = spdk_thread_get_by_id(req.id);
+	if (thread == NULL) {
+		SPDK_ERRLOG("Thread %" PRIu64 " does not exist\n", req.id);
+		spdk_jsonrpc_send_error_response_fmt(request, SPDK_JSONRPC_ERROR_INVALID_PARAMS,
+						     "Thread %" PRIu64 " does not exist", req.id);
+		goto err;
+	}
+
+	rc = spdk_app_parse_core_mask(req.cpumask, &ctx->cpumask);
+	if (rc != 0) {
+		SPDK_ERRLOG("Invalid cpumask %s\n", req.cpumask);
+		spdk_jsonrpc_send_error_response_fmt(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						     "Invalid cpumask %s", req.cpumask);
+		goto err;
+	}
+
+	if (spdk_cpuset_count(&ctx->cpumask) == 0) {
+		spdk_jsonrpc_send_error_response_fmt(request, SPDK_JSONRPC_ERROR_INTERNAL_ERROR,
+						     "No CPU is selected from reactor mask %s\n",
+						     spdk_cpuset_fmt(spdk_app_get_core_mask()));
+		goto err;
+	}
+
+	ctx->request = request;
+	ctx->orig_thread = spdk_get_thread();
+
+	spdk_thread_send_msg(thread, rpc_thread_set_cpumask, ctx);
+
+	free(req.cpumask);
+	return;
+
+err:
+	free(req.cpumask);
+	free(ctx);
+}
+SPDK_RPC_REGISTER("thread_set_cpumask", spdk_rpc_thread_set_cpumask, SPDK_RPC_RUNTIME)

scripts/rpc.py

@@ -2269,6 +2269,17 @@ Format: 'user:u1 secret:s1 muser:mu1 msecret:ms1,user:u2 secret:s2 muser:mu2 mse
         'thread_get_stats', help='Display current statistics of all the threads')
     p.set_defaults(func=thread_get_stats)
 
+    def thread_set_cpumask(args):
+        ret = rpc.app.thread_set_cpumask(args.client,
+                                         id=args.id,
+                                         cpumask=args.cpumask)
+    p = subparsers.add_parser('thread_set_cpumask',
+                              help="""set the cpumask of the thread whose ID matches to the
+    specified value. The thread may be migrated to one of the specified CPUs.""")
+    p.add_argument('-i', '--id', type=int, help='thread ID')
+    p.add_argument('-m', '--cpumask', help='cpumask for this thread')
+    p.set_defaults(func=thread_set_cpumask)
+
     def env_dpdk_get_mem_stats(args):
         print_dict(rpc.env_dpdk.env_dpdk_get_mem_stats(args.client))
 

scripts/rpc/app.py

@@ -44,3 +44,17 @@ def thread_get_stats(client):
         Current threads statistics.
     """
     return client.call('thread_get_stats')
+
+
+def thread_set_cpumask(client, id, cpumask):
+    """Set the cpumask of the thread whose ID matches to the specified value.
+
+    Args:
+        id: thread ID
+        cpumask: cpumask for this thread
+
+    Returns:
+        True or False
+    """
+    params = {'id': id, 'cpumask': cpumask}
+    return client.call('thread_set_cpumask', params)