lib/event: Add spdk_for_each_reactor to process event for each reactor

This API will be used in the upcoming reactor_get_stats RPC first.
This API is not public but internal in SPDK. Add necessary unit test
together.

Signed-off-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Change-Id: I761ffe3c8d6d8da4594c0d728b479f9f40275fc9
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/478024
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Community-CI: SPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Alexey Marchuk <alexeymar@mellanox.com>

include/spdk_internal/event.h

@@ -57,6 +57,20 @@ void spdk_reactors_fini(void);
 void spdk_reactors_start(void);
 void spdk_reactors_stop(void *arg1);
 
+/**
+ * Allocate and pass an event to each reactor, serially.
+ *
+ * The allocated event is processed asynchronously - i.e. spdk_for_each_reactor
+ * will return prior to `fn` being called on each reactor.
+ *
+ * \param fn This is the function that will be called on each reactor.
+ * \param arg1 Argument will be passed to fn when called.
+ * \param arg2 Argument will be passed to fn when called.
+ * \param cpl This will be called on the originating reactor after `fn` has been
+ * called on each reactor.
+ */
+void spdk_for_each_reactor(spdk_event_fn fn, void *arg1, void *arg2, spdk_event_fn cpl);
+
 struct spdk_subsystem {
 	const char *name;
 	/* User must call spdk_subsystem_init_next() when they are done with their initialization. */

lib/event/reactor.c

@@ -534,4 +534,67 @@ spdk_reactor_schedule_thread(struct spdk_thread *thread)
 	return 0;
 }
 
+struct call_reactor {
+	uint32_t cur_core;
+	spdk_event_fn fn;
+	void *arg1;
+	void *arg2;
+
+	uint32_t orig_core;
+	spdk_event_fn cpl;
+};
+
+static void
+spdk_on_reactor(void *arg1, void *arg2)
+{
+	struct call_reactor *cr = arg1;
+	struct spdk_event *evt;
+
+	cr->fn(cr->arg1, cr->arg2);
+
+	cr->cur_core = spdk_env_get_next_core(cr->cur_core);
+
+	if (cr->cur_core > spdk_env_get_last_core()) {
+		SPDK_DEBUGLOG(SPDK_LOG_REACTOR, "Completed reactor iteration\n");
+
+		evt = spdk_event_allocate(cr->orig_core, cr->cpl, cr->arg1, cr->arg2);
+		free(cr);
+	} else {
+		SPDK_DEBUGLOG(SPDK_LOG_REACTOR, "Continuing reactor iteration to %d\n",
+			      cr->cur_core);
+
+		evt = spdk_event_allocate(cr->cur_core, spdk_on_reactor, arg1, NULL);
+	}
+	assert(evt != NULL);
+	spdk_event_call(evt);
+}
+
+void
+spdk_for_each_reactor(spdk_event_fn fn, void *arg1, void *arg2, spdk_event_fn cpl)
+{
+	struct call_reactor *cr;
+	struct spdk_event *evt;
+
+	cr = calloc(1, sizeof(*cr));
+	if (!cr) {
+		SPDK_ERRLOG("Unable to perform reactor iteration\n");
+		cpl(arg1, arg2);
+		return;
+	}
+
+	cr->fn = fn;
+	cr->arg1 = arg1;
+	cr->arg2 = arg2;
+	cr->cpl = cpl;
+	cr->orig_core = spdk_env_get_current_core();
+	cr->cur_core = spdk_env_get_first_core();
+
+	SPDK_DEBUGLOG(SPDK_LOG_REACTOR, "Starting reactor iteration from %d\n", cr->orig_core);
+
+	evt = spdk_event_allocate(cr->cur_core, spdk_on_reactor, cr, NULL);
+	assert(evt != NULL);
+
+	spdk_event_call(evt);
+}
+
 SPDK_LOG_REGISTER_COMPONENT("reactor", SPDK_LOG_REACTOR)

test/unit/lib/event/reactor.c/reactor_ut.c

@@ -155,6 +155,69 @@ test_schedule_thread(void)
 	free_cores();
 }
 
+static void
+for_each_reactor_done(void *arg1, void *arg2)
+{
+	uint32_t *count = arg1;
+	bool *done = arg2;
+
+	(*count)++;
+	*done = true;
+}
+
+static void
+for_each_reactor_cb(void *arg1, void *arg2)
+{
+	uint32_t *count = arg1;
+
+	(*count)++;
+}
+
+static void
+test_for_each_reactor(void)
+{
+	uint32_t count = 0, i;
+	bool done = false;
+	struct spdk_reactor *reactor;
+
+	allocate_cores(5);
+
+	CU_ASSERT(spdk_reactors_init() == 0);
+
+	MOCK_SET(spdk_env_get_current_core, 0);
+
+	spdk_for_each_reactor(for_each_reactor_cb, &count, &done, for_each_reactor_done);
+
+	MOCK_CLEAR(spdk_env_get_current_core);
+
+	/* We have not processed any event yet, so count and done should be 0 and false,
+	 *  respectively.
+	 */
+	CU_ASSERT(count == 0);
+
+	/* Poll each reactor to verify the event is passed to each */
+	for (i = 0; i < 5; i++) {
+		reactor = spdk_reactor_get(i);
+		CU_ASSERT(reactor != NULL);
+
+		_spdk_event_queue_run_batch(reactor);
+		CU_ASSERT(count == (i + 1));
+		CU_ASSERT(done == false);
+	}
+
+	/* After each reactor is called, the completion calls it one more time. */
+	reactor = spdk_reactor_get(0);
+	CU_ASSERT(reactor != NULL);
+
+	_spdk_event_queue_run_batch(reactor);
+	CU_ASSERT(count == 6);
+	CU_ASSERT(done == true);
+
+	spdk_reactors_fini();
+
+	free_cores();
+}
+
 int
 main(int argc, char **argv)
 {
@@ -175,7 +238,8 @@ main(int argc, char **argv)
 		CU_add_test(suite, "test_create_reactor", test_create_reactor) == NULL ||
 		CU_add_test(suite, "test_init_reactors", test_init_reactors) == NULL ||
 		CU_add_test(suite, "test_event_call", test_event_call) == NULL ||
-		CU_add_test(suite, "test_schedule_thread", test_schedule_thread) == NULL
+		CU_add_test(suite, "test_schedule_thread", test_schedule_thread) == NULL ||
+		CU_add_test(suite, "test_for_each_reactor", test_for_each_reactor) == NULL
 	) {
 		CU_cleanup_registry();
 		return CU_get_error();