thread: Add cpumask parameter to spdk_thread_create

Allow users to optionally specify an affinity mask when
creating a thread. This isn't currently used, but we want
the API to be in its final form for the next release.

Change-Id: I7bd05e921ece6d8d5f61775bd14286f6a58f267f
Signed-off-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/451683
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>

examples/bdev/fio_plugin/fio_plugin.c

@@ -104,7 +104,7 @@ spdk_fio_init_thread(struct thread_data *td)
 	fio_thread->td = td;
 	td->io_ops_data = fio_thread;
 
-	fio_thread->thread = spdk_thread_create("fio_thread");
+	fio_thread->thread = spdk_thread_create("fio_thread", NULL);
 	if (!fio_thread->thread) {
 		free(fio_thread);
 		SPDK_ERRLOG("failed to allocate thread\n");

include/spdk/thread.h

@@ -40,6 +40,7 @@
 
 #include "spdk/stdinc.h"
 
+#include "spdk/cpuset.h"
 #include "spdk/queue.h"
 
 #ifdef __cplusplus
@@ -196,10 +197,13 @@ void spdk_thread_lib_fini(void);
  * \param name Human-readable name for the thread; can be retrieved with spdk_thread_get_name().
  * The string is copied, so the pointed-to data only needs to be valid during the
  * spdk_thread_create() call. May be NULL to specify no name.
+ * \param cpumask Optional mask of CPU cores on which to schedule this thread. This is only
+ * a suggestion to the scheduler. The value is copied, so cpumask may be released when
+ * this function returns. May be NULL if no mask is required.
  *
  * \return a pointer to the allocated thread on success or NULL on failure..
  */
-struct spdk_thread *spdk_thread_create(const char *name);
+struct spdk_thread *spdk_thread_create(const char *name, struct spdk_cpuset *cpumask);
 
 /**
  * Release any resources related to the given thread and destroy it. Execution

lib/event/app.c

@@ -638,7 +638,7 @@ spdk_app_start(struct spdk_app_opts *opts, spdk_msg_fn start_fn,
 
 	/* Now that the reactors have been initialized, we can create an
 	 * initialization thread. */
-	g_app_thread = spdk_thread_create("app_thread");
+	g_app_thread = spdk_thread_create("app_thread", NULL);
 	if (!g_app_thread) {
 		SPDK_ERRLOG("Unable to create an spdk_thread for initialization\n");
 		goto app_start_log_close_err;

lib/event/reactor.c

@@ -323,7 +323,7 @@ spdk_reactors_start(void)
 
 			/* For now, for each reactor spawn one thread. */
 			snprintf(thread_name, sizeof(thread_name), "reactor_%u", reactor->lcore);
-			spdk_thread_create(thread_name);
+			spdk_thread_create(thread_name, NULL);
 		}
 		spdk_cpuset_set_cpu(g_spdk_app_core_mask, i, true);
 	}

lib/rocksdb/env_spdk.cc

@@ -584,7 +584,7 @@ void SpdkInitializeThread(void)
 	struct spdk_thread *thread;
 
 	if (g_fs != NULL) {
-		thread = spdk_thread_create("spdk_rocksdb");
+		thread = spdk_thread_create("spdk_rocksdb", NULL);
 		spdk_set_thread(thread);
 		g_sync_args.channel = spdk_fs_alloc_thread_ctx(g_fs);
 	}

lib/thread/thread.c

@@ -113,6 +113,8 @@ struct spdk_thread {
 	TAILQ_ENTRY(spdk_thread)	tailq;
 	char				*name;
 
+	struct spdk_cpuset		*cpumask;
+
 	uint64_t			tsc_last;
 	struct spdk_thread_stats	stats;
 
@@ -200,7 +202,7 @@ spdk_thread_lib_fini(void)
 }
 
 struct spdk_thread *
-spdk_thread_create(const char *name)
+spdk_thread_create(const char *name, struct spdk_cpuset *cpumask)
 {
 	struct spdk_thread *thread;
 	struct spdk_msg *msgs[SPDK_MSG_MEMPOOL_CACHE_SIZE];
@@ -212,6 +214,19 @@ spdk_thread_create(const char *name)
 		return NULL;
 	}
 
+	thread->cpumask = spdk_cpuset_alloc();
+	if (!thread->cpumask) {
+		free(thread);
+		SPDK_ERRLOG("Unable to allocate memory for CPU mask\n");
+		return NULL;
+	}
+
+	if (cpumask) {
+		spdk_cpuset_copy(thread->cpumask, cpumask);
+	} else {
+		spdk_cpuset_negate(thread->cpumask);
+	}
+
 	TAILQ_INIT(&thread->io_channels);
 	TAILQ_INIT(&thread->active_pollers);
 	TAILQ_INIT(&thread->timer_pollers);
@@ -304,6 +319,8 @@ spdk_thread_exit(struct spdk_thread *thread)
 		free(poller);
 	}
 
+	spdk_cpuset_free(thread->cpumask);
+
 	pthread_mutex_lock(&g_devlist_mutex);
 	assert(g_thread_count > 0);
 	g_thread_count--;

test/common/lib/ut_multithread.c

@@ -87,7 +87,7 @@ allocate_threads(int num_threads)
 
 	for (i = 0; i < g_ut_num_threads; i++) {
 		set_thread(i);
-		thread = spdk_thread_create(NULL);
+		thread = spdk_thread_create(NULL, NULL);
 		assert(thread != NULL);
 		g_ut_threads[i].thread = thread;
 	}

test/unit/lib/blobfs/blobfs_sync_ut/blobfs_sync_ut.c

@@ -380,10 +380,10 @@ int main(int argc, char **argv)
 
 	spdk_thread_lib_init(NULL, 0);
 
-	thread = spdk_thread_create("test_thread");
+	thread = spdk_thread_create("test_thread", NULL);
 	spdk_set_thread(thread);
 
-	g_dispatch_thread = spdk_thread_create("dispatch_thread");
+	g_dispatch_thread = spdk_thread_create("dispatch_thread", NULL);
 	pthread_create(&spdk_tid, NULL, spdk_thread, g_dispatch_thread);
 
 	g_dev_buffer = calloc(1, DEV_BUFFER_SIZE);

test/unit/lib/ftl/common/utils.c

@@ -56,7 +56,7 @@ test_init_ftl_dev(const struct spdk_ocssd_geometry_data *geo,
 	dev->xfer_size = geo->ws_opt;
 	dev->geo = *geo;
 	dev->range = *range;
-	dev->core_thread.thread = spdk_thread_create("unit_test_thread");
+	dev->core_thread.thread = spdk_thread_create("unit_test_thread", NULL);
 	spdk_set_thread(dev->core_thread.thread);
 
 	dev->bands = calloc(geo->num_chk, sizeof(*dev->bands));

test/unit/lib/nvmf/tcp.c/tcp_ut.c

@@ -239,7 +239,7 @@ test_nvmf_tcp_create(void)
 	struct spdk_nvmf_tcp_transport *ttransport;
 	struct spdk_nvmf_transport_opts opts;
 
-	thread = spdk_thread_create(NULL);
+	thread = spdk_thread_create(NULL, NULL);
 	SPDK_CU_ASSERT_FATAL(thread != NULL);
 	spdk_set_thread(thread);
 
@@ -311,7 +311,7 @@ test_nvmf_tcp_destroy(void)
 	struct spdk_nvmf_transport *transport;
 	struct spdk_nvmf_transport_opts opts;
 
-	thread = spdk_thread_create(NULL);
+	thread = spdk_thread_create(NULL, NULL);
 	SPDK_CU_ASSERT_FATAL(thread != NULL);
 	spdk_set_thread(thread);
 
@@ -341,7 +341,7 @@ test_nvmf_tcp_poll_group_create(void)
 	struct spdk_thread *thread;
 	struct spdk_nvmf_transport_opts opts;
 
-	thread = spdk_thread_create(NULL);
+	thread = spdk_thread_create(NULL, NULL);
 	SPDK_CU_ASSERT_FATAL(thread != NULL);
 	spdk_set_thread(thread);
 

test/unit/lib/thread/thread.c/thread_ut.c

@@ -345,7 +345,7 @@ thread_name(void)
 	const char *name;
 
 	/* Create thread with no name, which automatically generates one */
-	thread = spdk_thread_create(NULL);
+	thread = spdk_thread_create(NULL, NULL);
 	spdk_set_thread(thread);
 	thread = spdk_get_thread();
 	SPDK_CU_ASSERT_FATAL(thread != NULL);
@@ -354,7 +354,7 @@ thread_name(void)
 	spdk_thread_exit(thread);
 
 	/* Create thread named "test_thread" */
-	thread = spdk_thread_create("test_thread");
+	thread = spdk_thread_create("test_thread", NULL);
 	spdk_set_thread(thread);
 	thread = spdk_get_thread();
 	SPDK_CU_ASSERT_FATAL(thread != NULL);