Commit e9aec674 authored by Shuhei Matsumoto's avatar Shuhei Matsumoto Committed by Tomasz Zawadzki
Browse files

lib/thread: thread_poll() polls until the exiting thread is exited 



Extract _spdk_thread_exit() from spdk_thread_exit() and
_spdk_thread_poll() calls _spdk_thread_exit() if the thread is in
the exiting state. spdk_thread_exit() changes to move the state to
the exiting state. The spdk_thread_poll() loop will end after the
thread moves to the exited state because the caller of
spdk_thread_poll() will check if the thread is in the exited state,
and break the loop if true.

If the user does not call spdk_thread_exit() explicitly, the reactor
has to terminate all existing threads at its shutdown. In this case,
multiple threads may have some dependency to release I/O channels or
unregister pollers. So the reactor has the large two loops, the first
loop calls spdk_thread_exit() on all threads, the second loop calls
spdk_thread_destroy() if exited or spdk_thread_poll() otherwise for
each thread until all threads are destroyed.

Besides, change the return value of spdk_thread_exit() to return
always 0. Keep it for ABI compatibility. Change ERRLOG to INFOLOG
for _spdk_thread_exit() because it is called repeatedly now. Remove
the check of I/O reference count from _spdk_thread_exit() because
_free_thread() cannot free I/O channel. Refine the unit test
accordingly.

Fixes issue #1288.

Signed-off-by: default avatarShuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Change-Id: Iee5fb984a96bfac53110fe991dd994ded31dffa4
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/1423


Tested-by: default avatarSPDK CI Jenkins <sys_sgci@intel.com>
Reviewed-by: default avatarAleksey Marchuk <alexeymar@mellanox.com>
Reviewed-by: default avatarJim Harris <james.r.harris@intel.com>
parent 282b1505

include/spdk/thread.h

+6 −4
Original line number Diff line number Diff line
@@ -267,13 +267,15 @@ void spdk_set_thread(struct spdk_thread *thread); 
 * spdk_poller_register(), and spdk_get_io_channel() calls. May only be called

 * within an spdk poller or message.

 *

 * All I/O channel references associated with the thread must be released using

 * spdk_put_io_channel(), and all active pollers associated with the thread must

 * be unregistered using spdk_poller_unregister(), prior to calling this function.

 * All I/O channel references associated with the thread must be released

 * using spdk_put_io_channel(), and all active pollers associated with the thread

 * should be unregistered using spdk_poller_unregister(), prior to calling

 * this function. This function will complete these processing. The completion can

 * be queried by spdk_thread_is_exited().

 *

 * \param thread The thread to destroy.

 *

 * \return 0 on success, negated errno on failure.

 * \return always 0. (return value was deprecated but keep it for ABI compatibility.)

 */

int spdk_thread_exit(struct spdk_thread *thread);

lib/event/reactor.c

+16 −10
Original line number Diff line number Diff line
@@ -366,7 +366,6 @@ _spdk_reactor_run(void *arg) 
	struct spdk_thread	*thread;

	struct spdk_lw_thread	*lw_thread, *tmp;

	char			thread_name[32];

	int			rc __attribute__((unused));



	SPDK_NOTICELOG("Reactor started on core %u\n", reactor->lcore);
@@ -386,18 +385,25 @@ _spdk_reactor_run(void *arg) 
		}

	}



	TAILQ_FOREACH(lw_thread, &reactor->threads, link) {

		thread = spdk_thread_get_from_ctx(lw_thread);

		spdk_set_thread(thread);

		spdk_thread_exit(thread);

	}



	while (!TAILQ_EMPTY(&reactor->threads)) {

		TAILQ_FOREACH_SAFE(lw_thread, &reactor->threads, link, tmp) {

			thread = spdk_thread_get_from_ctx(lw_thread);

			spdk_set_thread(thread);

			if (spdk_thread_is_exited(thread)) {

				TAILQ_REMOVE(&reactor->threads, lw_thread, link);

				assert(reactor->thread_count > 0);

				reactor->thread_count--;

		spdk_set_thread(thread);

		rc = spdk_thread_exit(thread);

		assert(rc == 0);

		while (!spdk_thread_is_exited(thread)) {

				spdk_thread_destroy(thread);

			} else {

				spdk_thread_poll(thread, 0, 0);

			}

		spdk_thread_destroy(thread);

		}

	}



	return 0;

lib/thread/thread.c

+23 −17
Original line number Diff line number Diff line
@@ -316,7 +316,7 @@ spdk_set_thread(struct spdk_thread *thread) 
	tls_thread = thread;

}



static int

static void

_spdk_thread_exit(struct spdk_thread *thread)

{

	struct spdk_poller *poller;
@@ -324,36 +324,37 @@ _spdk_thread_exit(struct spdk_thread *thread) 


	TAILQ_FOREACH(poller, &thread->active_pollers, tailq) {

		if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) {

			SPDK_ERRLOG("thread %s still has active poller %s\n",

			SPDK_INFOLOG(SPDK_LOG_THREAD,

				     "thread %s still has active poller %s\n",

				     thread->name, poller->name);

			return -EBUSY;

			return;

		}

	}



	TAILQ_FOREACH(poller, &thread->timed_pollers, tailq) {

		if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) {

			SPDK_ERRLOG("thread %s still has active timed poller %s\n",

			SPDK_INFOLOG(SPDK_LOG_THREAD,

				     "thread %s still has active timed poller %s\n",

				     thread->name, poller->name);

			return -EBUSY;

			return;

		}

	}



	TAILQ_FOREACH(poller, &thread->paused_pollers, tailq) {

		SPDK_ERRLOG("thread %s still has paused poller %s\n",

		SPDK_INFOLOG(SPDK_LOG_THREAD,

			     "thread %s still has paused poller %s\n",

			     thread->name, poller->name);

		return -EBUSY;

		return;

	}



	TAILQ_FOREACH(ch, &thread->io_channels, tailq) {

		if (ch->ref != 0) {

			SPDK_ERRLOG("thread %s still has active channel for io_device %s\n",

		SPDK_INFOLOG(SPDK_LOG_THREAD,

			     "thread %s still has channel for io_device %s\n",

			     thread->name, ch->dev->name);

			return -EBUSY;

		}

		return;

	}



	thread->state = SPDK_THREAD_STATE_EXITED;

	return 0;

}



int
@@ -370,7 +371,8 @@ spdk_thread_exit(struct spdk_thread *thread) 
		return 0;

	}



	return _spdk_thread_exit(thread);

	thread->state = SPDK_THREAD_STATE_EXITING;

	return 0;

}



bool
@@ -661,6 +663,10 @@ spdk_thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now) 


	rc = _spdk_thread_poll(thread, max_msgs, now);



	if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITING)) {

		_spdk_thread_exit(thread);

	}



	_spdk_thread_update_stats(thread, spdk_get_ticks(), now, rc);



	tls_thread = orig_thread;

lib/vhost/vhost.c

+1 −4
Original line number Diff line number Diff line
@@ -590,10 +590,7 @@ vhost_parse_core_mask(const char *mask, struct spdk_cpuset *cpumask) 
static void

vhost_dev_thread_exit(void *arg1)

{

	int rc __attribute__((unused));



	rc = spdk_thread_exit(spdk_get_thread());

	assert(rc == 0);

	spdk_thread_exit(spdk_get_thread());

}



int

test/common/lib/ut_multithread.c

+21 −8
Original line number Diff line number Diff line
@@ -102,20 +102,33 @@ allocate_threads(int num_threads) 
void

free_threads(void)

{

	uint32_t i;

	uint32_t i, num_threads;

	struct spdk_thread *thread;

	int rc __attribute__((unused));



	for (i = 0; i < g_ut_num_threads; i++) {

		set_thread(i);

		thread = g_ut_threads[i].thread;

		rc = spdk_thread_exit(thread);

		assert(rc == 0);

		while (!spdk_thread_is_exited(thread)) {

			spdk_thread_poll(thread, 0, 0);

		spdk_thread_exit(thread);

	}

		spdk_thread_destroy(thread);



	num_threads = g_ut_num_threads;



	while (num_threads != 0) {

		for (i = 0; i < g_ut_num_threads; i++) {

			set_thread(i);

			thread = g_ut_threads[i].thread;

			if (thread == NULL) {

				continue;

			}



			if (spdk_thread_is_exited(thread)) {

				g_ut_threads[i].thread = NULL;

				num_threads--;

				spdk_thread_destroy(thread);

			} else {

				spdk_thread_poll(thread, 0, 0);

			}

		}

	}



	g_ut_num_threads = 0;