env: add wrapper to call function without thread affinity (#60)

Make a wrapper that spdk can call a function without thread affinity, and
call this wrapper to open rbd image.

Change-Id: Iadc87a948f43632abf497f88165483a0e269ba54

include/spdk/env.h

@@ -231,6 +231,19 @@ int spdk_pci_addr_compare(const struct spdk_pci_addr *a1, const struct spdk_pci_
  */
 int spdk_pci_addr_parse(struct spdk_pci_addr *addr, const char *bdf);
 
+/**
+ * Call a function with CPU affinity unset.
+ *
+ * This can be used to run a function that creates other threads without inheriting the calling
+ * thread's CPU affinity.
+ *
+ * \param cb function to call
+ * \param arg parameter to cb function
+ *
+ * \return the return value of cb()
+ */
+void *spdk_call_unaffinitized(void *cb(void *arg), void *arg);
+
 #ifdef __cplusplus
 }
 #endif

lib/bdev/rbd/blockdev_rbd.c

@@ -47,6 +47,7 @@
 #include <rados/librados.h>
 
 #include "spdk/conf.h"
+#include "spdk/env.h"
 #include "spdk/log.h"
 #include "spdk/bdev.h"
 #include "spdk/io_channel.h"
@@ -429,6 +430,20 @@ blockdev_rbd_free_channel(struct blockdev_rbd_io_channel *ch)
 	}
 }
 
+static void *
+blockdev_rbd_handle(void *arg)
+{
+	struct blockdev_rbd_io_channel *ch = arg;
+	void *ret = arg;
+
+	if (rbd_open(ch->io_ctx, ch->disk->rbd_name, &ch->image, NULL) < 0) {
+		SPDK_ERRLOG("Failed to open specified rbd device\n");
+		ret = NULL;
+	}
+
+	return ret;
+}
+
 static int
 blockdev_rbd_create_cb(void *io_device, uint32_t priority,
 		       void *ctx_buf, void *unique_ctx)
@@ -450,9 +465,7 @@ blockdev_rbd_create_cb(void *io_device, uint32_t priority,
 		goto err;
 	}
 
-	ret = rbd_open(ch->io_ctx, ch->disk->rbd_name, &ch->image, NULL);
-	if (ret < 0) {
-		SPDK_ERRLOG("Failed to open specified rbd device\n");
+	if (spdk_call_unaffinitized(blockdev_rbd_handle, ch) == NULL) {
 		goto err;
 	}
 

lib/env/env.c

@@ -34,6 +34,7 @@
 #include "spdk/env.h"
 
 #include <string.h>
+#include <unistd.h>
 
 #include <rte_config.h>
 #include <rte_cycles.h>
@@ -184,3 +185,34 @@ void spdk_delay_us(unsigned int us)
 {
 	return rte_delay_us(us);
 }
+
+void *
+spdk_call_unaffinitized(void *cb(void *arg), void *arg)
+{
+	rte_cpuset_t orig_cpuset, new_cpuset;
+	void *ret;
+	long num_cores, i;
+
+	if (cb == NULL) {
+		return NULL;
+	}
+
+	rte_thread_get_affinity(&orig_cpuset);
+
+	CPU_ZERO(&new_cpuset);
+
+	num_cores = sysconf(_SC_NPROCESSORS_CONF);
+
+	/* Create a mask containing all CPUs */
+	for (i = 0; i < num_cores; i++) {
+		CPU_SET(i, &new_cpuset);
+	}
+
+	rte_thread_set_affinity(&new_cpuset);
+
+	ret = cb(arg);
+
+	rte_thread_set_affinity(&orig_cpuset);
+
+	return ret;
+}