Commit 59955a12 authored by Daniel Verkamp's avatar Daniel Verkamp
Browse files

event: allow unregistering a poller within its poller fn 



Modify the spdk_poller_unregister() function so that it works correctly
when unregistering a poller from its own callback function.

Change-Id: I57fa5ebd8a8bad522e34f597b406a4726f1b76ad
Signed-off-by: default avatarDaniel Verkamp <daniel.verkamp@intel.com>
parent 08a6cba2

lib/event/reactor.c

+87 −15
Original line number Diff line number Diff line
@@ -62,13 +62,30 @@ 


#define SPDK_REACTOR_SPIN_TIME_US	1



enum spdk_poller_state {

	/* The poller is registered with a reactor but not currently executing its fn. */

	SPDK_POLLER_STATE_WAITING,



	/* The poller is currently running its fn. */

	SPDK_POLLER_STATE_RUNNING,



	/* The poller was unregistered during the execution of its fn. */

	SPDK_POLLER_STATE_UNREGISTERED,

};



struct spdk_poller {

	TAILQ_ENTRY(spdk_poller)	tailq;

	uint32_t			lcore;



	/* Current state of the poller; should only be accessed from the poller's thread. */

	enum spdk_poller_state		state;



	uint64_t			period_ticks;

	uint64_t			next_run_tick;

	spdk_poller_fn			fn;

	void				*arg;



	struct spdk_event		*unregister_complete_event;

};



enum spdk_reactor_state {
@@ -272,6 +289,16 @@ spdk_poller_insert_timer(struct spdk_reactor *reactor, struct spdk_poller *polle 
	TAILQ_INSERT_HEAD(&reactor->timer_pollers, poller, tailq);

}



static void

_spdk_poller_unregister_complete(struct spdk_poller *poller)

{

	if (poller->unregister_complete_event) {

		spdk_event_call(poller->unregister_complete_event);

	}



	free(poller);

}



/**



\brief This is the main function of the reactor thread.
@@ -330,8 +357,14 @@ _spdk_reactor_run(void *arg) 
		poller = TAILQ_FIRST(&reactor->active_pollers);

		if (poller) {

			TAILQ_REMOVE(&reactor->active_pollers, poller, tailq);

			poller->state = SPDK_POLLER_STATE_RUNNING;

			poller->fn(poller->arg);

			if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) {

				_spdk_poller_unregister_complete(poller);

			} else {

				poller->state = SPDK_POLLER_STATE_WAITING;

				TAILQ_INSERT_TAIL(&reactor->active_pollers, poller, tailq);

			}

			last_action = rte_get_timer_cycles();

		}
@@ -341,10 +374,16 @@ _spdk_reactor_run(void *arg) 


			if (now >= poller->next_run_tick) {

				TAILQ_REMOVE(&reactor->timer_pollers, poller, tailq);

				poller->state = SPDK_POLLER_STATE_RUNNING;

				poller->fn(poller->arg);

				if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) {

					_spdk_poller_unregister_complete(poller);

				} else {

					poller->state = SPDK_POLLER_STATE_WAITING;

					spdk_poller_insert_timer(reactor, poller, now);

				}

			}

		}



		/* Determine if the thread can sleep */

		if (sleep_cycles > 0) {
@@ -657,6 +696,7 @@ spdk_poller_register(struct spdk_poller **ppoller, spdk_poller_fn fn, void *arg, 
	}



	poller->lcore = lcore;

	poller->state = SPDK_POLLER_STATE_WAITING;

	poller->fn = fn;

	poller->arg = arg;
@@ -679,30 +719,48 @@ spdk_poller_register(struct spdk_poller **ppoller, spdk_poller_fn fn, void *arg, 
}



static void

_spdk_event_remove_poller(spdk_event_t event)

_spdk_poller_unregister(struct spdk_reactor *reactor, struct spdk_poller *poller,

			struct spdk_event *next)

{

	struct spdk_poller *poller = spdk_event_get_arg1(event);

	struct spdk_reactor *reactor = spdk_reactor_get(poller->lcore);

	struct spdk_event *next = spdk_event_get_next(event);

	assert(poller->lcore == reactor->lcore);

	assert(poller->lcore == spdk_app_get_current_core());



	poller->unregister_complete_event = next;



	if (poller->state == SPDK_POLLER_STATE_RUNNING) {

		/*

		 * We are being called from the poller_fn, so set the state to unregistered

		 * and let the reactor loop free the poller.

		 */

		poller->state = SPDK_POLLER_STATE_UNREGISTERED;

	} else {

		/* Poller is not running currently, so just free it. */

		if (poller->period_ticks) {

			TAILQ_REMOVE(&reactor->timer_pollers, poller, tailq);

		} else {

			TAILQ_REMOVE(&reactor->active_pollers, poller, tailq);

		}



	free(poller);



	if (next) {

		spdk_event_call(next);

		_spdk_poller_unregister_complete(poller);

	}

}



static void

_spdk_event_remove_poller(spdk_event_t event)

{

	struct spdk_poller *poller = spdk_event_get_arg1(event);

	struct spdk_reactor *reactor = spdk_reactor_get(poller->lcore);

	struct spdk_event *next = spdk_event_get_next(event);



	_spdk_poller_unregister(reactor, poller, next);

}



void

spdk_poller_unregister(struct spdk_poller **ppoller,

		       struct spdk_event *complete)

{

	struct spdk_poller *poller;

	uint32_t lcore;



	poller = *ppoller;
@@ -715,6 +773,20 @@ spdk_poller_unregister(struct spdk_poller **ppoller, 
		return;

	}



	spdk_event_call(spdk_event_allocate(poller->lcore, _spdk_event_remove_poller, poller, NULL,

	lcore = poller->lcore;



	if (lcore == spdk_app_get_current_core()) {

		/*

		 * The poller is registered on the current core, so call the remove function

		 * directly.

		 */

		_spdk_poller_unregister(spdk_reactor_get(lcore), poller, complete);

	} else {

		/*

		 * The poller is registered on a different core.

		 * Schedule an event to run on the poller's core that will remove the poller.

		 */

		spdk_event_call(spdk_event_allocate(lcore, _spdk_event_remove_poller, poller, NULL,

						    complete));

	}

}

test/lib/event/reactor/reactor.c

+10 −0
Original line number Diff line number Diff line
@@ -43,6 +43,7 @@ static struct spdk_poller *test_end_poller; 
static struct spdk_poller *poller_100ms;

static struct spdk_poller *poller_250ms;

static struct spdk_poller *poller_500ms;

static struct spdk_poller *poller_oneshot;



static void

test_end(void *arg)
@@ -59,6 +60,13 @@ tick(void *arg) 
	printf("tick %" PRIu64 "\n", (uint64_t)period);

}



static void

oneshot(void *arg)

{

	printf("oneshot\n");

	spdk_poller_unregister(&poller_oneshot, NULL);

}



static void

test_start(spdk_event_t evt)

{
@@ -70,6 +78,7 @@ test_start(spdk_event_t evt) 
	spdk_poller_register(&poller_100ms, tick, (void *)100, 0, NULL, 100000);

	spdk_poller_register(&poller_250ms, tick, (void *)250, 0, NULL, 250000);

	spdk_poller_register(&poller_500ms, tick, (void *)500, 0, NULL, 500000);

	spdk_poller_register(&poller_oneshot, oneshot, NULL, 0, NULL, 0);

}



static void
@@ -81,6 +90,7 @@ test_cleanup(void) 
	spdk_poller_unregister(&poller_100ms, NULL);

	spdk_poller_unregister(&poller_250ms, NULL);

	spdk_poller_unregister(&poller_500ms, NULL);

	/* poller_oneshot unregisters itself */

}



static void