Spdk/examples/nvmf/nvmf/nvmf.c

/*   SPDX-License-Identifier: BSD-3-Clause
 *   Copyright (C) 2019 Intel Corporation.
 *   All rights reserved.
 */

#include "spdk/stdinc.h"

#include "spdk/env.h"
#include "spdk/event.h"
#include "spdk/init.h"
#include "spdk/string.h"
#include "spdk/thread.h"
#include "spdk/bdev.h"
#include "spdk/rpc.h"
#include "spdk/nvmf.h"
#include "spdk/likely.h"

#include "spdk_internal/event.h"

#define NVMF_DEFAULT_SUBSYSTEMS		32

static const char *g_rpc_addr = SPDK_DEFAULT_RPC_ADDR;

enum nvmf_target_state {
	NVMF_INIT_SUBSYSTEM = 0,
	NVMF_INIT_TARGET,
	NVMF_INIT_POLL_GROUPS,
	NVMF_INIT_START_SUBSYSTEMS,
	NVMF_RUNNING,
	NVMF_FINI_STOP_SUBSYSTEMS,
	NVMF_FINI_POLL_GROUPS,
	NVMF_FINI_TARGET,
	NVMF_FINI_SUBSYSTEM,
};

struct nvmf_lw_thread {
	TAILQ_ENTRY(nvmf_lw_thread) link;
	bool resched;
};

struct nvmf_reactor {
	uint32_t core;

	struct spdk_ring		*threads;
	TAILQ_ENTRY(nvmf_reactor)	link;
};

struct nvmf_target_poll_group {
	struct spdk_nvmf_poll_group		*group;
	struct spdk_thread			*thread;

	TAILQ_ENTRY(nvmf_target_poll_group)	link;
};

struct nvmf_target {
	struct spdk_nvmf_tgt	*tgt;

	int max_subsystems;
};

TAILQ_HEAD(, nvmf_reactor) g_reactors = TAILQ_HEAD_INITIALIZER(g_reactors);
TAILQ_HEAD(, nvmf_target_poll_group) g_poll_groups = TAILQ_HEAD_INITIALIZER(g_poll_groups);
static uint32_t g_num_poll_groups = 0;

static struct nvmf_reactor *g_main_reactor = NULL;
static struct nvmf_reactor *g_next_reactor = NULL;
static struct spdk_thread *g_init_thread = NULL;
static struct spdk_thread *g_fini_thread = NULL;
static struct nvmf_target g_nvmf_tgt = {
	.max_subsystems = NVMF_DEFAULT_SUBSYSTEMS,
};

static struct nvmf_target_poll_group *g_next_pg = NULL;
static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
static bool g_reactors_exit = false;
static enum nvmf_target_state g_target_state;
static bool g_intr_received = false;

static uint32_t g_migrate_pg_period_us = 0;
static struct spdk_poller *g_migrate_pg_poller = NULL;

static void nvmf_target_advance_state(void);
static int nvmf_schedule_spdk_thread(struct spdk_thread *thread);

static void
usage(char *program_name)
{
	printf("%s options", program_name);
	printf("\n");
	printf("\t[-g period of round robin poll group migration (us) (default: 0 (disabled))]\n");
	printf("\t[-h show this usage]\n");
	printf("\t[-i shared memory ID (optional)]\n");
	printf("\t[-m core mask for DPDK]\n");
	printf("\t[-n max subsystems for target(default: 32)]\n");
	printf("\t[-r RPC listen address (default /var/tmp/spdk.sock)]\n");
	printf("\t[-s memory size in MB for DPDK (default: 0MB)]\n");
	printf("\t[-u disable PCI access]\n");
}

static int
parse_args(int argc, char **argv, struct spdk_env_opts *opts)
{
	int op;
	long int value;

	while ((op = getopt(argc, argv, "g:i:m:n:p:r:s:u:h")) != -1) {
		switch (op) {
		case 'g':
			value = spdk_strtol(optarg, 10);
			if (value < 0) {
				fprintf(stderr, "converting a string to integer failed\n");
				return -EINVAL;
			}
			g_migrate_pg_period_us = value;
			break;
		case 'i':
			value = spdk_strtol(optarg, 10);
			if (value < 0) {
				fprintf(stderr, "converting a string to integer failed\n");
				return -EINVAL;
			}
			opts->shm_id = value;
			break;
		case 'm':
			opts->core_mask = optarg;
			break;
		case 'n':
			g_nvmf_tgt.max_subsystems = spdk_strtol(optarg, 10);
			if (g_nvmf_tgt.max_subsystems < 0) {
				fprintf(stderr, "converting a string to integer failed\n");
				return -EINVAL;
			}
			break;
		case 'r':
			g_rpc_addr = optarg;
			break;
		case 's':
			value = spdk_strtol(optarg, 10);
			if (value < 0) {
				fprintf(stderr, "converting a string to integer failed\n");
				return -EINVAL;
			}
			opts->mem_size = value;
			break;
		case 'u':
			opts->no_pci = true;
			break;
		case 'h':
			usage(argv[0]);
			exit(EXIT_SUCCESS);
		default:
			usage(argv[0]);
			return 1;
		}
	}

	return 0;
}

static int
nvmf_reactor_run(void *arg)
{
	struct nvmf_reactor *nvmf_reactor = arg;
	struct nvmf_lw_thread *lw_thread;
	struct spdk_thread *thread;

	/* run all the lightweight threads in this nvmf_reactor by FIFO. */
	do {
		if (spdk_ring_dequeue(nvmf_reactor->threads, (void **)&lw_thread, 1)) {
			thread = spdk_thread_get_from_ctx(lw_thread);

			spdk_thread_poll(thread, 0, 0);

			if (spdk_unlikely(spdk_thread_is_exited(thread) &&
					  spdk_thread_is_idle(thread))) {
				spdk_thread_destroy(thread);
			} else if (spdk_unlikely(lw_thread->resched)) {
				lw_thread->resched = false;
				nvmf_schedule_spdk_thread(thread);
			} else {
				spdk_ring_enqueue(nvmf_reactor->threads, (void **)&lw_thread, 1, NULL);
			}
		}
	} while (!g_reactors_exit);

	/* free all the lightweight threads */
	while (spdk_ring_dequeue(nvmf_reactor->threads, (void **)&lw_thread, 1)) {
		thread = spdk_thread_get_from_ctx(lw_thread);
		spdk_set_thread(thread);

		if (spdk_thread_is_exited(thread)) {
			spdk_thread_destroy(thread);
		} else {
			/* This thread is not exited yet, and may need to communicate with other threads
			 * to be exited. So mark it as exiting, and check again after traversing other threads.
			 */
			spdk_thread_exit(thread);
			spdk_thread_poll(thread, 0, 0);
			spdk_ring_enqueue(nvmf_reactor->threads, (void **)&lw_thread, 1, NULL);
		}
	}

	return 0;
}

static int
nvmf_schedule_spdk_thread(struct spdk_thread *thread)
{
	struct nvmf_reactor *nvmf_reactor;
	struct nvmf_lw_thread *lw_thread;
	struct spdk_cpuset *cpumask;
	uint32_t i;

	/* Lightweight threads may have a requested cpumask.
	 * This is a request only - the scheduler does not have to honor it.
	 * For this scheduler implementation, each reactor is pinned to
	 * a particular core so honoring the request is reasonably easy.
	 */
	cpumask = spdk_thread_get_cpumask(thread);

	lw_thread = spdk_thread_get_ctx(thread);
	assert(lw_thread != NULL);
	memset(lw_thread, 0, sizeof(*lw_thread));

	/* assign lightweight threads to nvmf reactor(core)
	 * Here we use the mutex.The way the actual SPDK event framework
	 * solves this is by using internal rings for messages between reactors
	 */
	pthread_mutex_lock(&g_mutex);
	for (i = 0; i < spdk_env_get_core_count(); i++) {
		if (g_next_reactor == NULL) {
			g_next_reactor = TAILQ_FIRST(&g_reactors);
		}
		nvmf_reactor = g_next_reactor;
		g_next_reactor = TAILQ_NEXT(g_next_reactor, link);

		/* each spdk_thread has the core affinity */
		if (spdk_cpuset_get_cpu(cpumask, nvmf_reactor->core)) {
			spdk_ring_enqueue(nvmf_reactor->threads, (void **)&lw_thread, 1, NULL);
			break;
		}
	}
	pthread_mutex_unlock(&g_mutex);

	if (i == spdk_env_get_core_count()) {
		fprintf(stderr, "failed to schedule spdk thread\n");
		return -1;
	}
	return 0;
}

static void
nvmf_request_spdk_thread_reschedule(struct spdk_thread *thread)
{
	struct nvmf_lw_thread *lw_thread;

	assert(thread == spdk_get_thread());

	lw_thread = spdk_thread_get_ctx(thread);

	assert(lw_thread != NULL);

	lw_thread->resched = true;
}

static int
nvmf_reactor_thread_op(struct spdk_thread *thread, enum spdk_thread_op op)
{
	switch (op) {
	case SPDK_THREAD_OP_NEW:
		return nvmf_schedule_spdk_thread(thread);
	case SPDK_THREAD_OP_RESCHED:
		nvmf_request_spdk_thread_reschedule(thread);
		return 0;
	default:
		return -ENOTSUP;
	}
}

static bool
nvmf_reactor_thread_op_supported(enum spdk_thread_op op)
{
	switch (op) {
	case SPDK_THREAD_OP_NEW:
	case SPDK_THREAD_OP_RESCHED:
		return true;
	default:
		return false;
	}
}

static int
nvmf_init_threads(void)
{
	int rc;
	uint32_t i;
	char thread_name[32];
	struct nvmf_reactor *nvmf_reactor;
	struct spdk_cpuset cpumask;
	uint32_t main_core = spdk_env_get_current_core();

	/* Whenever SPDK creates a new lightweight thread it will call
	 * nvmf_schedule_spdk_thread asking for the application to begin
	 * polling it via spdk_thread_poll(). Each lightweight thread in
	 * SPDK optionally allocates extra memory to be used by the application
	 * framework. The size of the extra memory allocated is the second parameter.
	 */
	spdk_thread_lib_init_ext(nvmf_reactor_thread_op, nvmf_reactor_thread_op_supported,
				 sizeof(struct nvmf_lw_thread), SPDK_DEFAULT_MSG_MEMPOOL_SIZE);

	/* Spawn one system thread per CPU core. The system thread is called a reactor.
	 * SPDK will spawn lightweight threads that must be mapped to reactors in
	 * nvmf_schedule_spdk_thread. Using a single system thread per CPU core is a
	 * choice unique to this application. SPDK itself does not require this specific
	 * threading model. For example, another viable threading model would be
	 * dynamically scheduling the lightweight threads onto a thread pool using a
	 * work queue.
	 */
	SPDK_ENV_FOREACH_CORE(i) {
		nvmf_reactor = calloc(1, sizeof(struct nvmf_reactor));
		if (!nvmf_reactor) {
			fprintf(stderr, "failed to alloc nvmf reactor\n");
			rc = -ENOMEM;
			goto err_exit;
		}

		nvmf_reactor->core = i;

		nvmf_reactor->threads = spdk_ring_create(SPDK_RING_TYPE_MP_SC, 1024, SPDK_ENV_SOCKET_ID_ANY);
		if (!nvmf_reactor->threads) {
			fprintf(stderr, "failed to alloc ring\n");
			free(nvmf_reactor);
			rc = -ENOMEM;
			goto err_exit;
		}

		TAILQ_INSERT_TAIL(&g_reactors, nvmf_reactor, link);

		if (i == main_core) {
			g_main_reactor = nvmf_reactor;
			g_next_reactor = g_main_reactor;
		} else {
			rc = spdk_env_thread_launch_pinned(i,
							   nvmf_reactor_run,
							   nvmf_reactor);
			if (rc) {
				fprintf(stderr, "failed to pin reactor launch\n");
				goto err_exit;
			}
		}
	}

	/* Spawn a lightweight thread only on the current core to manage this application. */
	spdk_cpuset_zero(&cpumask);
	spdk_cpuset_set_cpu(&cpumask, main_core, true);
	snprintf(thread_name, sizeof(thread_name), "nvmf_main_thread");
	g_init_thread = spdk_thread_create(thread_name, &cpumask);
	if (!g_init_thread) {
		fprintf(stderr, "failed to create spdk thread\n");
		return -1;
	}

	fprintf(stdout, "nvmf threads initialize successfully\n");
	return 0;

err_exit:
	return rc;
}

static void
nvmf_destroy_threads(void)
{
	struct nvmf_reactor *nvmf_reactor, *tmp;

	TAILQ_FOREACH_SAFE(nvmf_reactor, &g_reactors, link, tmp) {
		spdk_ring_free(nvmf_reactor->threads);
		free(nvmf_reactor);
	}

	pthread_mutex_destroy(&g_mutex);
	spdk_thread_lib_fini();
	fprintf(stdout, "nvmf threads destroy successfully\n");
}

static void
nvmf_tgt_destroy_done(void *ctx, int status)
{
	fprintf(stdout, "destroyed the nvmf target service\n");

	g_target_state = NVMF_FINI_SUBSYSTEM;
	nvmf_target_advance_state();
}

static void
nvmf_destroy_nvmf_tgt(void)
{
	if (g_nvmf_tgt.tgt) {
		spdk_nvmf_tgt_destroy(g_nvmf_tgt.tgt, nvmf_tgt_destroy_done, NULL);
	} else {
		g_target_state = NVMF_FINI_SUBSYSTEM;
	}
}

static void
nvmf_create_nvmf_tgt(void)
{
	struct spdk_nvmf_subsystem *subsystem;
	struct spdk_nvmf_target_opts tgt_opts = {};

	tgt_opts.max_subsystems = g_nvmf_tgt.max_subsystems;
	snprintf(tgt_opts.name, sizeof(tgt_opts.name), "%s", "nvmf_example");
	/* Construct the default NVMe-oF target
	 * An NVMe-oF target is a collection of subsystems, namespace, and poll
	 * groups, and defines the scope of the NVMe-oF discovery service.
	 */
	g_nvmf_tgt.tgt = spdk_nvmf_tgt_create(&tgt_opts);
	if (g_nvmf_tgt.tgt == NULL) {
		fprintf(stderr, "spdk_nvmf_tgt_create() failed\n");
		goto error;
	}

	/* Create and add discovery subsystem to the NVMe-oF target.
	 * NVMe-oF defines a discovery mechanism that a host uses to determine
	 * the NVM subsystems that expose namespaces that the host may access.
	 * It provides a host with following capabilities:
	 *	1,The ability to discover a list of NVM subsystems with namespaces
	 *	  that are accessible to the host.
	 *	2,The ability to discover multiple paths to an NVM subsystem.
	 *	3,The ability to discover controllers that are statically configured.
	 */
	subsystem = spdk_nvmf_subsystem_create(g_nvmf_tgt.tgt, SPDK_NVMF_DISCOVERY_NQN,
					       SPDK_NVMF_SUBTYPE_DISCOVERY, 0);
	if (subsystem == NULL) {
		fprintf(stderr, "failed to create discovery nvmf library subsystem\n");
		goto error;
	}

	/* Allow any host to access the discovery subsystem */
	spdk_nvmf_subsystem_set_allow_any_host(subsystem, true);

	fprintf(stdout, "created a nvmf target service\n");

	g_target_state = NVMF_INIT_POLL_GROUPS;
	return;

error:
	g_target_state = NVMF_FINI_TARGET;
}

static void
nvmf_tgt_subsystem_stop_next(struct spdk_nvmf_subsystem *subsystem,
			     void *cb_arg, int status)
{
	int rc;

	subsystem = spdk_nvmf_subsystem_get_next(subsystem);
	if (subsystem) {
		rc = spdk_nvmf_subsystem_stop(subsystem,
					      nvmf_tgt_subsystem_stop_next,
					      cb_arg);
		if (rc) {
			nvmf_tgt_subsystem_stop_next(subsystem, cb_arg, 0);
			fprintf(stderr, "Unable to stop NVMe-oF subsystem. Trying others.\n");
		}
		return;
	}

	fprintf(stdout, "all subsystems of target stopped\n");

	g_target_state = NVMF_FINI_POLL_GROUPS;
	nvmf_target_advance_state();
}

static void
nvmf_tgt_stop_subsystems(struct nvmf_target *nvmf_tgt)
{
	struct spdk_nvmf_subsystem *subsystem;
	int rc;

	subsystem = spdk_nvmf_subsystem_get_first(nvmf_tgt->tgt);
	if (spdk_likely(subsystem)) {
		rc = spdk_nvmf_subsystem_stop(subsystem,
					      nvmf_tgt_subsystem_stop_next,
					      NULL);
		if (rc) {
			nvmf_tgt_subsystem_stop_next(subsystem, NULL, 0);
			fprintf(stderr, "Unable to stop NVMe-oF subsystem. Trying others.\n");
		}
	} else {
		g_target_state = NVMF_FINI_POLL_GROUPS;
	}
}

static void
nvmf_tgt_subsystem_start_next(struct spdk_nvmf_subsystem *subsystem,
			      void *cb_arg, int status)
{
	int rc;

	subsystem = spdk_nvmf_subsystem_get_next(subsystem);
	if (subsystem) {
		rc = spdk_nvmf_subsystem_start(subsystem, nvmf_tgt_subsystem_start_next,
					       cb_arg);
		if (rc) {
			g_target_state = NVMF_FINI_STOP_SUBSYSTEMS;
			fprintf(stderr, "Unable to start NVMe-oF subsystem. shutting down app.\n");
			nvmf_target_advance_state();
		}
		return;
	}

	fprintf(stdout, "all subsystems of target started\n");

	g_target_state = NVMF_RUNNING;
	nvmf_target_advance_state();
}

static void
nvmf_tgt_start_subsystems(struct nvmf_target *nvmf_tgt)
{
	struct spdk_nvmf_subsystem *subsystem;
	int rc;

	/* Subsystem is the NVM subsystem which is a combine of namespaces
	 * except the discovery subsystem which is used for discovery service.
	 * It also controls the hosts that means the subsystem determines whether
	 * the host can access this subsystem.
	 */
	subsystem = spdk_nvmf_subsystem_get_first(nvmf_tgt->tgt);
	if (spdk_likely(subsystem)) {
		/* In SPDK there are three states in subsystem: Inactive, Active, Paused.
		 * Start subsystem means make it from inactive to active that means
		 * subsystem start to work or it can be accessed.
		 */
		rc = spdk_nvmf_subsystem_start(subsystem,
					       nvmf_tgt_subsystem_start_next,
					       NULL);
		if (rc) {
			fprintf(stderr, "Unable to start NVMe-oF subsystem. shutting down app.\n");
			g_target_state = NVMF_FINI_STOP_SUBSYSTEMS;
		}
	} else {
		g_target_state = NVMF_RUNNING;
	}
}

static void
nvmf_tgt_create_poll_groups_done(void *ctx)
{
	struct nvmf_target_poll_group *pg = ctx;

	if (!g_next_pg) {
		g_next_pg = pg;
	}

	TAILQ_INSERT_TAIL(&g_poll_groups, pg, link);

	assert(g_num_poll_groups < spdk_env_get_core_count());

	if (++g_num_poll_groups == spdk_env_get_core_count()) {
		fprintf(stdout, "create targets's poll groups done\n");

		g_target_state = NVMF_INIT_START_SUBSYSTEMS;
		nvmf_target_advance_state();
	}
}

static void
nvmf_tgt_create_poll_group(void *ctx)
{
	struct nvmf_target_poll_group *pg;

	pg = calloc(1, sizeof(struct nvmf_target_poll_group));
	if (!pg) {
		fprintf(stderr, "failed to allocate poll group\n");
		assert(false);
		return;
	}

	pg->thread = spdk_get_thread();
	pg->group = spdk_nvmf_poll_group_create(g_nvmf_tgt.tgt);
	if (!pg->group) {
		fprintf(stderr, "failed to create poll group of the target\n");
		free(pg);
		assert(false);
		return;
	}

	spdk_thread_send_msg(g_init_thread, nvmf_tgt_create_poll_groups_done, pg);
}

/* Create a lightweight thread per poll group instead of assuming a pool of lightweight
 * threads already exist at start up time. A poll group is a collection of unrelated NVMe-oF
 * connections. Each poll group is only accessed from the associated lightweight thread.
 */
static void
nvmf_poll_groups_create(void)
{
	struct spdk_cpuset tmp_cpumask = {};
	uint32_t i;
	char thread_name[32];
	struct spdk_thread *thread;

	assert(g_init_thread != NULL);

	SPDK_ENV_FOREACH_CORE(i) {
		spdk_cpuset_zero(&tmp_cpumask);
		spdk_cpuset_set_cpu(&tmp_cpumask, i, true);
		snprintf(thread_name, sizeof(thread_name), "nvmf_tgt_poll_group_%u", i);

		thread = spdk_thread_create(thread_name, &tmp_cpumask);
		assert(thread != NULL);

		spdk_thread_send_msg(thread, nvmf_tgt_create_poll_group, NULL);
	}
}

static void
_nvmf_tgt_destroy_poll_groups_done(void *ctx)
{
	assert(g_num_poll_groups > 0);

	if (--g_num_poll_groups == 0) {
		fprintf(stdout, "destroy targets's poll groups done\n");

		g_target_state = NVMF_FINI_TARGET;
		nvmf_target_advance_state();
	}
}

static void
nvmf_tgt_destroy_poll_groups_done(void *cb_arg, int status)
{
	struct nvmf_target_poll_group *pg = cb_arg;

	free(pg);

	spdk_thread_send_msg(g_fini_thread, _nvmf_tgt_destroy_poll_groups_done, NULL);

	spdk_thread_exit(spdk_get_thread());
}

static void
nvmf_tgt_destroy_poll_group(void *ctx)
{
	struct nvmf_target_poll_group *pg = ctx;

	spdk_nvmf_poll_group_destroy(pg->group, nvmf_tgt_destroy_poll_groups_done, pg);
}

static void
nvmf_poll_groups_destroy(void)
{
	struct nvmf_target_poll_group *pg, *tmp;

	g_fini_thread = spdk_get_thread();
	assert(g_fini_thread != NULL);

	TAILQ_FOREACH_SAFE(pg, &g_poll_groups, link, tmp) {
		TAILQ_REMOVE(&g_poll_groups, pg, link);
		spdk_thread_send_msg(pg->thread, nvmf_tgt_destroy_poll_group, pg);
	}
}

static void
nvmf_subsystem_fini_done(void *cb_arg)
{
	fprintf(stdout, "bdev subsystem finish successfully\n");
	spdk_rpc_finish();
	g_reactors_exit = true;
}

static void
nvmf_subsystem_init_done(int rc, void *cb_arg)
{
	fprintf(stdout, "bdev subsystem init successfully\n");

	rc = spdk_rpc_initialize(g_rpc_addr);
	if (rc) {
		spdk_app_stop(rc);
		return;
	}

	spdk_rpc_set_state(SPDK_RPC_RUNTIME);

	g_target_state = NVMF_INIT_TARGET;
	nvmf_target_advance_state();
}

static void
migrate_poll_group_by_rr(void *ctx)
{
	uint32_t current_core, next_core;
	struct spdk_cpuset cpumask = {};

	current_core = spdk_env_get_current_core();
	next_core = spdk_env_get_next_core(current_core);
	if (next_core == UINT32_MAX) {
		next_core = spdk_env_get_first_core();
	}

	spdk_cpuset_set_cpu(&cpumask, next_core, true);

	spdk_thread_set_cpumask(&cpumask);
}

static int
migrate_poll_groups_by_rr(void *ctx)
{
	struct nvmf_target_poll_group *pg;

	TAILQ_FOREACH(pg, &g_poll_groups, link) {
		spdk_thread_send_msg(pg->thread, migrate_poll_group_by_rr, NULL);
	}

	return SPDK_POLLER_BUSY;
}

static void
nvmf_target_advance_state(void)
{
	enum nvmf_target_state prev_state;

	do {
		prev_state = g_target_state;

		switch (g_target_state) {
		case NVMF_INIT_SUBSYSTEM:
			/* initialize the bdev layer */
			spdk_subsystem_init(nvmf_subsystem_init_done, NULL);
			return;
		case NVMF_INIT_TARGET:
			nvmf_create_nvmf_tgt();
			break;
		case NVMF_INIT_POLL_GROUPS:
			nvmf_poll_groups_create();
			break;
		case NVMF_INIT_START_SUBSYSTEMS:
			nvmf_tgt_start_subsystems(&g_nvmf_tgt);
			break;
		case NVMF_RUNNING:
			fprintf(stdout, "nvmf target is running\n");
			if (g_migrate_pg_period_us != 0) {
				g_migrate_pg_poller = SPDK_POLLER_REGISTER(migrate_poll_groups_by_rr, NULL,
						      g_migrate_pg_period_us);
			}
			break;
		case NVMF_FINI_STOP_SUBSYSTEMS:
			spdk_poller_unregister(&g_migrate_pg_poller);
			nvmf_tgt_stop_subsystems(&g_nvmf_tgt);
			break;
		case NVMF_FINI_POLL_GROUPS:
			nvmf_poll_groups_destroy();
			break;
		case NVMF_FINI_TARGET:
			nvmf_destroy_nvmf_tgt();
			break;
		case NVMF_FINI_SUBSYSTEM:
			spdk_subsystem_fini(nvmf_subsystem_fini_done, NULL);
			break;
		}
	} while (g_target_state != prev_state);
}

static void
nvmf_target_app_start(void *arg)
{
	g_target_state = NVMF_INIT_SUBSYSTEM;
	nvmf_target_advance_state();
}

static void
_nvmf_shutdown_cb(void *ctx)
{
	/* Still in initialization state, defer shutdown operation */
	if (g_target_state < NVMF_RUNNING) {
		spdk_thread_send_msg(spdk_get_thread(), _nvmf_shutdown_cb, NULL);
		return;
	} else if (g_target_state > NVMF_RUNNING) {
		/* Already in Shutdown status, ignore the signal */
		return;
	}

	g_target_state = NVMF_FINI_STOP_SUBSYSTEMS;
	nvmf_target_advance_state();
}

static void
nvmf_shutdown_cb(int signo)
{
	if (!g_intr_received) {
		g_intr_received = true;
		spdk_thread_send_msg(g_init_thread, _nvmf_shutdown_cb, NULL);
	}
}

static int
nvmf_setup_signal_handlers(void)
{
	struct sigaction	sigact;
	sigset_t		sigmask;
	int			signals[] = {SIGINT, SIGTERM};
	int			num_signals = sizeof(signals) / sizeof(int);
	int			rc, i;

	rc = sigemptyset(&sigmask);
	if (rc) {
		fprintf(stderr, "errno:%d--failed to empty signal set\n", errno);
		return rc;
	}
	memset(&sigact, 0, sizeof(sigact));
	rc = sigemptyset(&sigact.sa_mask);
	if (rc) {
		fprintf(stderr, "errno:%d--failed to empty signal set\n", errno);
		return rc;
	}

	/* Install the same handler for SIGINT and SIGTERM */
	sigact.sa_handler = nvmf_shutdown_cb;

	for (i = 0; i < num_signals; i++) {
		rc = sigaction(signals[i], &sigact, NULL);
		if (rc < 0) {
			fprintf(stderr, "errno:%d--sigaction() failed\n", errno);
			return rc;
		}
		rc = sigaddset(&sigmask, signals[i]);
		if (rc) {
			fprintf(stderr, "errno:%d--failed to add set\n", errno);
			return rc;
		}
	}

	pthread_sigmask(SIG_UNBLOCK, &sigmask, NULL);

	return 0;
}

int
main(int argc, char **argv)
{
	int rc;
	struct spdk_env_opts opts;

	spdk_env_opts_init(&opts);
	opts.name = "nvmf-example";

	rc = parse_args(argc, argv, &opts);
	if (rc != 0) {
		return rc;
	}

	if (spdk_env_init(&opts) < 0) {
		fprintf(stderr, "unable to initialize SPDK env\n");
		return -EINVAL;
	}

	/* Initialize the threads */
	rc = nvmf_init_threads();
	assert(rc == 0);

	/* Send a message to the thread assigned to the main reactor
	 * that continues initialization. This is how we bootstrap the
	 * program so that all code from here on is running on an SPDK thread.
	 */
	assert(g_init_thread != NULL);

	rc = nvmf_setup_signal_handlers();
	assert(rc == 0);

	spdk_thread_send_msg(g_init_thread, nvmf_target_app_start, NULL);

	nvmf_reactor_run(g_main_reactor);

	spdk_env_thread_wait_all();
	nvmf_destroy_threads();

	spdk_env_fini();

	return rc;
}