Add event-driven application framework (eeeac667) · Commits · Public Repositories / spdk

autotest.sh

+1 −0

Original line number	Diff line number	Diff line
		@@ -44,6 +44,7 @@ timing_exit afterboot

		timing_enter lib

		time test/lib/event/event.sh
		time test/lib/nvme/nvme.sh
		time test/lib/memory/memory.sh
		time test/lib/ioat/ioat.sh

include/spdk/event.h

0 → 100644

+290 −0

Original line number	Diff line number	Diff line
		/*-
		* BSD LICENSE
		*
		* Copyright (c) Intel Corporation.
		* All rights reserved.
		*
		* Redistribution and use in source and binary forms, with or without
		* modification, are permitted provided that the following conditions
		* are met:
		*
		* * Redistributions of source code must retain the above copyright
		* notice, this list of conditions and the following disclaimer.
		* * Redistributions in binary form must reproduce the above copyright
		* notice, this list of conditions and the following disclaimer in
		* the documentation and/or other materials provided with the
		* distribution.
		* * Neither the name of Intel Corporation nor the names of its
		* contributors may be used to endorse or promote products derived
		* from this software without specific prior written permission.
		*
		* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
		* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
		* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
		* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
		* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
		* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
		* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
		* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
		* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
		* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
		* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		*/

		/** \file
		* Event framework public API.
		*
		* This is a framework for writing asynchronous, polled-mode, shared-nothing
		* server applications. The framework relies on DPDK for much of its underlying
		* architecture. The framework defines several concepts - reactors, events, pollers,
		* and subsystems - that are described in the following sections.
		*
		* The framework runs one thread per core (the user provides a core mask), where
		* each thread is a tight loop. The threads never block for any reason. These threads
		* are called reactors and their main responsibility is to process incoming events
		* from a queue.
		*
		* An event, defined by \ref spdk_event is a bundled function pointer and arguments that
		* can be sent to a different core and executed. The function pointer is executed only once,
		* and then the entire event is freed. These functions should never block and preferably
		* should execute very quickly. Events also have a pointer to a 'next' event that will be
		* executed upon completion of the given event, which allows chaining. This is
		* very much a simplified version of futures, promises, and continuations designed within
		* the constraints of the C programming language.
		*
		* The framework also defines another type of function called a poller. Pollers are also
		* functions with arguments that can be bundled and sent to a different core to be executed,
		* but they are instead executed repeatedly on that core until unregistered. The reactor
		* will handle interspersing calls to the pollers with other event processing automatically.
		* Pollers are intended to poll hardware as a replacement for interrupts and they should not
		* generally be used for any other purpose.
		*
		* The framework also defines an interface for subsystems, which are libraries of code that
		* depend on this framework. A library can register itself as a subsystem and provide
		* pointers to initialize and destroy itself which will be called at the appropriate time.
		* This is purely for sequencing initialization code in a convenient manner within the
		* framework.
		*
		* The framework itself is bundled into a higher level abstraction called an "app". Once
		* \ref spdk_app_start is called it will block the current thread until the application
		* terminates (by calling \ref spdk_app_stop).
		*/

		#ifndef SPDK_EVENT_H
		#define SPDK_EVENT_H

		#include <stdbool.h>
		#include <stdint.h>
		#include <stdio.h>
		#include <string.h>

		#include "spdk/queue.h"

		#define SPDK_APP_DEFAULT_LOG_FACILITY "local7"
		#define SPDK_APP_DEFAULT_LOG_PRIORITY "info"

		typedef struct spdk_event *spdk_event_t;
		typedef void (*spdk_event_fn)(spdk_event_t);

		/**
		* \brief An event is a function that is passed to and called on an lcore.
		*/
		struct spdk_event {
		uint32_t lcore;
		spdk_event_fn fn;
		void *arg1;
		void *arg2;
		struct spdk_event *next;
		};

		typedef void (spdk_poller_fn)(void arg);

		/**
		* \brief A poller is a function that is repeatedly called on an lcore.
		*/
		struct spdk_poller {
		uint32_t lcore;
		spdk_poller_fn fn;
		void *arg;
		};

		#define SPDK_POLLER_RING_SIZE 4096

		/*
		* -1 accounts for the empty slot needed to differentiate between ring empty
		* and ring full.
		*/
		#define SPDK_MAX_POLLERS_PER_CORE (SPDK_POLLER_RING_SIZE - 1)

		typedef void (*spdk_app_shutdown_cb)(void);
		typedef void (*spdk_sighandler_t)(int);

		#define SPDK_APP_DPDK_DEFAULT_MEM_SIZE 2048
		#define SPDK_APP_DPDK_DEFAULT_MASTER_CORE 0
		#define SPDK_APP_DPDK_DEFAULT_MEM_CHANNEL 4
		#define SPDK_APP_DPDK_DEFAULT_CORE_MASK "0x1"

		/**
		* \brief Event framework initialization options
		*/
		struct spdk_app_opts {
		const char *name;
		const char *config_file;
		const char *reactor_mask;
		const char *log_facility;
		const char *tpoint_group_mask;

		int instance_id;

		spdk_app_shutdown_cb shutdown_cb;
		spdk_event_fn start_fn;
		spdk_sighandler_t usr1_handler;

		bool enable_coredump;
		uint32_t dpdk_mem_channel;
		uint32_t dpdk_master_core;
		int dpdk_mem_size;
		};

		/**
		* \brief Initialize the default value of opts
		*/
		void spdk_app_opts_init(struct spdk_app_opts *opts);

		/**
		* \brief Initialize DPDK via opts.
		*/
		void spdk_init_dpdk(struct spdk_app_opts *opts);

		/**
		* \brief Initialize an application to use the event framework. This must be called prior to using
		* any other functions in this library.
		*/
		void spdk_app_init(struct spdk_app_opts *opts);

		/**
		* \brief Perform final shutdown operations on an application using the event framework.
		*/
		void spdk_app_fini(void);

		/**
		* \brief Start the framework. Once started, the framework will call start_fn on the master
		* core with the arguments provided. This call will block until \ref spdk_app_stop is called.
		*/
		int spdk_app_start(spdk_event_fn start_fn, void arg1, void arg2);

		/**
		* \brief Stop the framework. This does not wait for all threads to exit. Instead, it kicks off
		* the shutdown process and returns. Once the shutdown process is complete, \ref spdk_app_start will return.
		*/
		void spdk_app_stop(int rc);

		/**
		* \brief Generate a configuration file that corresponds to the current running state.
		*/
		int spdk_app_get_running_config(char *config_str, char name);

		/**
		* \brief Return the instance id for this application.
		*/
		int spdk_app_get_instance_id(void);

		/**
		* \brief Convert a string containing a CPU core mask into a bitmask
		*/
		int spdk_app_parse_core_mask(const char mask, uint64_t cpumask);

		/**
		* \brief Return a mask of the CPU cores active for this application
		*/
		uint64_t spdk_app_get_core_mask(void);

		/**
		* \brief Return the number of CPU cores utilized by this application
		*/
		int spdk_app_get_core_count(void);

		/**
		* \brief Allocate an event to be passed to \ref spdk_event_call
		*/
		spdk_event_t spdk_event_allocate(uint32_t lcore, spdk_event_fn fn,
		void arg1, void arg2,
		spdk_event_t next);

		/**
		* \brief Pass the given event to the associated lcore and call the function.
		*/
		void spdk_event_call(spdk_event_t event);

		#define spdk_event_get_next(event) (event)->next
		#define spdk_event_get_arg1(event) (event)->arg1
		#define spdk_event_get_arg2(event) (event)->arg2

		/* TODO: This is only used by tests and should be made private */
		void spdk_event_queue_run_all(uint32_t lcore);

		/**
		* \brief Register a poller on the given lcore.
		*/
		void spdk_poller_register(struct spdk_poller *poller,
		uint32_t lcore,
		struct spdk_event *complete);

		/**
		* \brief Unregister a poller on the given lcore.
		*/
		void spdk_poller_unregister(struct spdk_poller *poller,
		struct spdk_event *complete);

		/**
		* \brief Move a poller from its current lcore to a new lcore.
		*/
		void spdk_poller_migrate(struct spdk_poller *poller, int new_lcore,
		struct spdk_event *complete);

		struct spdk_subsystem {
		const char *name;
		int (*init)(void);
		int (*fini)(void);
		void (config)(FILE fp);
		TAILQ_ENTRY(spdk_subsystem) tailq;
		};

		struct spdk_subsystem_depend {
		const char *name;
		const char *depends_on;
		TAILQ_ENTRY(spdk_subsystem_depend) tailq;
		};

		void spdk_add_subsystem(struct spdk_subsystem *subsystem);
		void spdk_add_subsystem_depend(struct spdk_subsystem_depend *depend);

		/**
		* \brief Register a new subsystem
		*/
		#define SPDK_SUBSYSTEM_REGISTER(_name, _init, _fini, _config) \
		static struct spdk_subsystem __subsystem_ ## _name = { \
		.name = #_name, \
		.init = _init, \
		.fini = _fini, \
		.config = _config, \
		}; \
		__attribute__((constructor)) static void _name ## _register(void) \
		{ \
		spdk_add_subsystem(&__subsystem_ ## _name); \
		}

		/**
		* \brief Declare that a subsystem depends on another subsystem.
		*/
		#define SPDK_SUBSYSTEM_DEPEND(_name, _depends_on) \
		static struct spdk_subsystem_depend __subsystem_ ## _name ## _depend_on ## _depends_on = { \
		.name = #_name, \
		.depends_on = #_depends_on, \
		}; \
		__attribute__((constructor)) static void _name ## _depend_on ## _depends_on(void) \
		{ \
		spdk_add_subsystem_depend(&__subsystem_ ## _name ## _depend_on ## _depends_on); \
		}

		#endif

lib/Makefile

+1 −1

Original line number	Diff line number	Diff line
		@@ -34,7 +34,7 @@
		SPDK_ROOT_DIR := $(abspath $(CURDIR)/..)
		include $(SPDK_ROOT_DIR)/mk/spdk.common.mk

		DIRS-y += conf cunit json jsonrpc log memory trace util nvme ioat
		DIRS-y += conf cunit event json jsonrpc log memory trace util nvme ioat

		.PHONY: all clean $(DIRS-y)

lib/event/Makefile

0 → 100644

+40 −0

Original line number	Diff line number	Diff line
		#
		# BSD LICENSE
		#
		# Copyright (c) Intel Corporation.
		# All rights reserved.
		#
		# Redistribution and use in source and binary forms, with or without
		# modification, are permitted provided that the following conditions
		# are met:
		#
		# * Redistributions of source code must retain the above copyright
		# notice, this list of conditions and the following disclaimer.
		# * Redistributions in binary form must reproduce the above copyright
		# notice, this list of conditions and the following disclaimer in
		# the documentation and/or other materials provided with the
		# distribution.
		# * Neither the name of Intel Corporation nor the names of its
		# contributors may be used to endorse or promote products derived
		# from this software without specific prior written permission.
		#
		# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
		# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
		# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
		# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
		# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
		# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
		# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
		# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
		# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
		# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
		# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		#

		SPDK_ROOT_DIR := $(abspath $(CURDIR)/../..)

		CFLAGS += $(DPDK_INC)
		LIBNAME = event
		C_SRCS = app.c dpdk_init.c reactor.c subsystem.c

		include $(SPDK_ROOT_DIR)/mk/spdk.lib.mk

lib/event/app.c

0 → 100644

+478 −0

Original line number	Diff line number	Diff line
		/*-
		* BSD LICENSE
		*
		* Copyright (c) Intel Corporation.
		* All rights reserved.
		*
		* Redistribution and use in source and binary forms, with or without
		* modification, are permitted provided that the following conditions
		* are met:
		*
		* * Redistributions of source code must retain the above copyright
		* notice, this list of conditions and the following disclaimer.
		* * Redistributions in binary form must reproduce the above copyright
		* notice, this list of conditions and the following disclaimer in
		* the documentation and/or other materials provided with the
		* distribution.
		* * Neither the name of Intel Corporation nor the names of its
		* contributors may be used to endorse or promote products derived
		* from this software without specific prior written permission.
		*
		* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
		* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
		* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
		* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
		* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
		* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
		* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
		* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
		* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
		* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
		* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		*/

		#include "spdk/event.h"

		#include <stdio.h>
		#include <stdlib.h>
		#include <sys/mman.h>
		#include <sys/resource.h>
		#include <sys/stat.h>
		#include <sys/types.h>
		#include <unistd.h>
		#include <errno.h>
		#include <string.h>
		#include <fcntl.h>
		#include <signal.h>

		#include <rte_debug.h>

		#include "spdk/log.h"
		#include "spdk/conf.h"
		#include "spdk/trace.h"

		#include "reactor.h"
		#include "subsystem.h"

		/* Add enough here to append ".pid" plus 2 digit instance ID */
		#define SPDK_APP_PIDFILE_MAX_LENGTH 40
		#define SPDK_APP_PIDFILE_PREFIX "/var/run"

		struct spdk_app {
		struct spdk_conf *config;
		char pidfile[SPDK_APP_PIDFILE_MAX_LENGTH];
		int instance_id;
		spdk_app_shutdown_cb shutdown_cb;
		int rc;
		};

		static struct spdk_app g_spdk_app;
		static spdk_event_t g_shutdown_event = NULL;

		static int spdk_app_write_pidfile(void);
		static void spdk_app_remove_pidfile(void);

		int
		spdk_app_get_instance_id(void)
		{
		return g_spdk_app.instance_id;
		}

		/* Global section */
		#define GLOBAL_CONFIG_TMPL \
		"# Configuration file\n" \
		"#\n" \
		"# Please write all parameters using ASCII.\n" \
		"# The parameter must be quoted if it includes whitespace.\n" \
		"#\n" \
		"# Configuration syntax:\n" \
		"# Spaces at head of line are deleted, other spaces are as separator\n" \
		"# Lines starting with '#' are comments and not evaluated.\n" \
		"# Lines ending with '\\' are concatenated with the next line.\n" \
		"# Bracketed keys are section keys grouping the following value keys.\n" \
		"# Number of section key is used as a tag number.\n" \
		"# Ex. [TargetNode1] = TargetNode section key with tag number 1\n" \
		"[Global]\n" \
		" Comment \"Global section\"\n" \
		"\n" \
		" # Users can restrict work items to only run on certain cores by\n" \
		" # specifying a ReactorMask. Default is to allow work items to run\n" \
		" # on all cores. Core 0 must be set in the mask if one is specified.\n" \
		" # Default: 0xFFFF (cores 0-15)\n" \
		" ReactorMask \"0x%" PRIX64 "\"\n" \
		"\n" \
		" # Tracepoint group mask for spdk trace buffers\n" \
		" # Default: 0x0 (all tracepoint groups disabled)\n" \
		" # Set to 0xFFFFFFFFFFFFFFFF to enable all tracepoint groups.\n" \
		" TpointGroupMask \"0x%" PRIX64 "\"\n" \
		"\n" \
		" # syslog facility\n" \
		" LogFacility \"%s\"\n" \
		"\n"

		static void
		spdk_app_config_dump_global_section(FILE *fp)
		{
		if (NULL == fp)
		return;

		/* FIXME - lookup log facility and put it in place of "local7" below */
		fprintf(fp, GLOBAL_CONFIG_TMPL,
		spdk_app_get_core_mask(), spdk_trace_get_tpoint_group_mask(),
		"local7");
		}

		int
		spdk_app_get_running_config(char *config_str, char name)
		{
		FILE *fp = NULL;
		int fd = -1;
		long length = 0, ret = 0;
		char vbuf[BUFSIZ];
		char config_template[64];

		snprintf(config_template, sizeof(config_template), "/tmp/%s.XXXXXX", name);
		/* Create temporary file to hold config */
		fd = mkstemp(config_template);
		if (fd == -1) {
		fprintf(stderr, "mkstemp failed\n");
		return -1;
		}
		fp = fdopen(fd, "wb+");
		if (NULL == fp) {
		fprintf(stderr, "error opening tmpfile fd = %d\n", fd);
		return -1;
		}

		/* Buffered IO */
		setvbuf(fp, vbuf, _IOFBF, BUFSIZ);

		spdk_app_config_dump_global_section(fp);
		spdk_subsystem_config(fp);

		length = ftell(fp);

		*config_str = malloc(length + 1);
		if (!*config_str) {
		perror("config_str");
		fclose(fp);
		return -1;
		}
		fseek(fp, 0, SEEK_SET);
		ret = fread(*config_str, sizeof(char), length, fp);
		if (ret < length)
		fprintf(stderr, "%s: warning - short read\n", __func__);
		fclose(fp);
		(*config_str)[length] = '\0';

		return 0;
		}

		static const char *
		spdk_get_log_facility(struct spdk_conf *config)
		{
		struct spdk_conf_section *sp;
		const char *logfacility;

		sp = spdk_conf_find_section(config, "Global");
		if (sp == NULL) {
		return SPDK_APP_DEFAULT_LOG_FACILITY;
		}

		logfacility = spdk_conf_section_get_val(sp, "LogFacility");
		if (logfacility == NULL) {
		return SPDK_APP_DEFAULT_LOG_FACILITY;
		}

		return logfacility;
		}

		static void
		__shutdown_signal(int signo)
		{
		/*
		* Call pre-allocated shutdown event. Note that it is not
		* safe to allocate the event within the signal handlers
		* context, since that context is not a DPDK thread so
		* buffer allocation is not permitted.
		*/
		if (g_shutdown_event != NULL) {
		spdk_event_call(g_shutdown_event);
		}
		}

		static void
		__shutdown_event_cb(spdk_event_t event)
		{
		g_spdk_app.shutdown_cb();
		}

		void
		spdk_app_opts_init(struct spdk_app_opts *opts)
		{
		if (!opts)
		return;

		memset(opts, 0, sizeof(*opts));

		opts->enable_coredump = true;
		opts->instance_id = -1;
		opts->dpdk_mem_size = -1;
		opts->dpdk_master_core = SPDK_APP_DPDK_DEFAULT_MASTER_CORE;
		opts->dpdk_mem_channel = SPDK_APP_DPDK_DEFAULT_MEM_CHANNEL;
		opts->reactor_mask = SPDK_APP_DPDK_DEFAULT_CORE_MASK;
		}

		void
		spdk_app_init(struct spdk_app_opts *opts)
		{
		struct spdk_conf *config;
		struct spdk_conf_section *sp;
		struct sigaction sigact;
		sigset_t signew;
		char shm_name[64];
		int rc;
		uint64_t tpoint_group_mask;
		char *end;

		if (opts->enable_coredump) {
		struct rlimit core_limits;

		core_limits.rlim_cur = core_limits.rlim_max = RLIM_INFINITY;
		setrlimit(RLIMIT_CORE, &core_limits);
		}

		config = spdk_conf_allocate();
		RTE_VERIFY(config != NULL);
		if (opts->config_file) {
		rc = spdk_conf_read(config, opts->config_file);
		if (rc != 0) {
		fprintf(stderr, "Could not read config file %s\n", opts->config_file);
		exit(EXIT_FAILURE);
		}
		if (config->section == NULL) {
		fprintf(stderr, "Invalid config file %s\n", opts->config_file);
		exit(EXIT_FAILURE);
		}
		}
		spdk_conf_set_as_default(config);

		if (opts->instance_id == -1) {
		sp = spdk_conf_find_section(config, "Global");
		if (sp != NULL) {
		opts->instance_id = spdk_conf_section_get_intval(sp, "InstanceID");
		}
		}

		if (opts->instance_id < 0) {
		opts->instance_id = 0;
		}

		memset(&g_spdk_app, 0, sizeof(g_spdk_app));
		g_spdk_app.config = config;
		g_spdk_app.instance_id = opts->instance_id;
		g_spdk_app.shutdown_cb = opts->shutdown_cb;
		snprintf(g_spdk_app.pidfile, sizeof(g_spdk_app.pidfile), "%s/%s.pid.%d",
		SPDK_APP_PIDFILE_PREFIX, opts->name, opts->instance_id);
		spdk_app_write_pidfile();

		/* open log files */
		if (opts->log_facility == NULL) {
		opts->log_facility = spdk_get_log_facility(g_spdk_app.config);
		if (opts->log_facility == NULL) {
		fprintf(stderr, "NULL logfacility\n");
		spdk_conf_free(g_spdk_app.config);
		exit(EXIT_FAILURE);
		}
		}
		rc = spdk_set_log_facility(opts->log_facility);
		if (rc < 0) {
		fprintf(stderr, "log facility error\n");
		spdk_conf_free(g_spdk_app.config);
		exit(EXIT_FAILURE);
		}

		rc = spdk_set_log_priority(SPDK_APP_DEFAULT_LOG_PRIORITY);
		if (rc < 0) {
		fprintf(stderr, "log priority error\n");
		spdk_conf_free(g_spdk_app.config);
		exit(EXIT_FAILURE);
		}
		spdk_open_log();

		if (opts->reactor_mask == NULL) {
		sp = spdk_conf_find_section(g_spdk_app.config, "Global");
		if (sp != NULL) {
		if (spdk_conf_section_get_val(sp, "WorkerMask")) {
		fprintf(stderr, "WorkerMask not valid key name."
		" Use ReactorMask instead.\n");
		spdk_conf_free(g_spdk_app.config);
		exit(EXIT_FAILURE);
		}
		opts->reactor_mask = spdk_conf_section_get_val(sp, "ReactorMask");
		}
		}

		/*
		* If mask not specified on command line or in configuration file,
		* reactor_mask will be NULL which will enable all cores to run
		* reactors.
		*/
		if (spdk_reactor_subsystem_init(opts->reactor_mask)) {
		fprintf(stderr, "Invalid reactor mask.\n");
		exit(EXIT_FAILURE);
		}

		/* setup signal handler thread */
		pthread_sigmask(SIG_SETMASK, NULL, &signew);

		memset(&sigact, 0, sizeof(sigact));
		sigact.sa_handler = SIG_IGN;
		sigemptyset(&sigact.sa_mask);
		rc = sigaction(SIGPIPE, &sigact, NULL);
		if (rc < 0) {
		SPDK_ERRLOG("sigaction(SIGPIPE) failed\n");
		exit(EXIT_FAILURE);
		}

		if (opts->shutdown_cb != NULL) {
		g_shutdown_event = spdk_event_allocate(rte_lcore_id(), __shutdown_event_cb,
		NULL, NULL, NULL);

		sigact.sa_handler = __shutdown_signal;
		sigemptyset(&sigact.sa_mask);
		rc = sigaction(SIGINT, &sigact, NULL);
		if (rc < 0) {
		SPDK_ERRLOG("sigaction(SIGINT) failed\n");
		exit(EXIT_FAILURE);
		}
		sigaddset(&signew, SIGINT);

		sigact.sa_handler = __shutdown_signal;
		sigemptyset(&sigact.sa_mask);
		rc = sigaction(SIGTERM, &sigact, NULL);
		if (rc < 0) {
		SPDK_ERRLOG("sigaction(SIGTERM) failed\n");
		exit(EXIT_FAILURE);
		}
		sigaddset(&signew, SIGTERM);
		}

		if (opts->usr1_handler != NULL) {
		sigact.sa_handler = opts->usr1_handler;
		sigemptyset(&sigact.sa_mask);
		rc = sigaction(SIGUSR1, &sigact, NULL);
		if (rc < 0) {
		SPDK_ERRLOG("sigaction(SIGUSR1) failed\n");
		exit(EXIT_FAILURE);
		}
		sigaddset(&signew, SIGUSR1);
		}

		sigaddset(&signew, SIGQUIT);
		sigaddset(&signew, SIGHUP);
		pthread_sigmask(SIG_SETMASK, &signew, NULL);

		snprintf(shm_name, sizeof(shm_name), "/%s_trace.%d", opts->name, opts->instance_id);
		spdk_trace_init(shm_name);

		if (opts->tpoint_group_mask == NULL) {
		sp = spdk_conf_find_section(g_spdk_app.config, "Global");
		if (sp != NULL) {
		opts->tpoint_group_mask = spdk_conf_section_get_val(sp, "TpointGroupMask");
		}
		}

		if (opts->tpoint_group_mask != NULL) {
		errno = 0;
		tpoint_group_mask = strtoull(opts->tpoint_group_mask, &end, 16);
		if (*end != '\0' \|\| errno) {
		SPDK_ERRLOG("invalid tpoint mask %s\n", opts->tpoint_group_mask);
		} else {
		spdk_trace_set_tpoint_group_mask(tpoint_group_mask);
		}
		}

		rc = spdk_subsystem_init();
		if (rc < 0) {
		SPDK_ERRLOG("spdk_subsystem_init() failed\n");
		exit(EXIT_FAILURE);
		}
		}

		void
		spdk_app_fini(void)
		{
		spdk_subsystem_fini();
		spdk_trace_cleanup();
		spdk_app_remove_pidfile();
		spdk_conf_free(g_spdk_app.config);
		spdk_close_log();
		}

		int
		spdk_app_start(spdk_event_fn start_fn, void arg1, void arg2)
		{
		spdk_event_t event;

		g_spdk_app.rc = 0;

		event = spdk_event_allocate(rte_get_master_lcore(), start_fn,
		arg1, arg2, NULL);
		/* Queues up the event, but can't run it until the reactors start */
		spdk_event_call(event);

		/* This blocks until spdk_app_stop is called */
		spdk_reactor_subsystem_start();

		return g_spdk_app.rc;
		}

		void
		spdk_app_stop(int rc)
		{
		spdk_reactor_subsystem_stop();
		g_spdk_app.rc = rc;
		}

		static int
		spdk_app_write_pidfile(void)
		{
		FILE *fp;
		pid_t pid;
		struct flock lock = {
		.l_type = F_WRLCK,
		.l_whence = SEEK_SET,
		.l_start = 0,
		.l_len = 0,
		};

		fp = fopen(g_spdk_app.pidfile, "w");
		if (fp == NULL) {
		SPDK_ERRLOG("pidfile open error %d\n", errno);
		return -1;
		}

		if (fcntl(fileno(fp), F_SETLK, &lock) != 0) {
		fprintf(stderr, "Cannot create lock on file %s, probably you"
		" should use different instance id\n", g_spdk_app.pidfile);
		exit(EXIT_FAILURE);
		}

		pid = getpid();
		fprintf(fp, "%d\n", (int)pid);
		fclose(fp);
		return 0;
		}

		static void
		spdk_app_remove_pidfile(void)
		{
		int rc;

		rc = remove(g_spdk_app.pidfile);
		if (rc != 0) {
		SPDK_ERRLOG("pidfile remove error %d\n", errno);
		/* ignore error */
		}
		}