lib/nvme: implement external IO processing (72e079a8) · Commits · Public Repositories / spdk

include/spdk/nvme.h

+20 −0

Original line number	Diff line number	Diff line
		@@ -1244,6 +1244,26 @@ int spdk_nvme_ctrlr_cmd_io_raw_with_md(struct spdk_nvme_ctrlr *ctrlr,
		int32_t spdk_nvme_qpair_process_completions(struct spdk_nvme_qpair *qpair,
		uint32_t max_completions);

		/**
		* Process IO message sent to controller from external module.
		*
		* This call process requests from the ring, send IO to an allocated qpair or
		* admin commands in its context. This call is non-blocking and intended to be
		* polled by SPDK thread to provide safe environment for NVMe request
		* completition sent by external module to controller.
		*
		* The caller must ensure that each controller is polled by only one thread at
		* a time.
		*
		* This function may be called at any point while the controller is attached to
		* the SPDK NVMe driver.
		*
		* \param ctrlr Opaque handle to NVMe controller.
		*
		* \return number of processed external IO messages.
		*/
		int spdk_nvme_io_msg_process(struct spdk_nvme_ctrlr *ctrlr);

		/**
		* Send the given admin command to the NVMe controller.
		*

lib/nvme/Makefile

+1 −1

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

		C_SRCS = nvme_ctrlr_cmd.c nvme_ctrlr.c nvme_fabric.c nvme_ns_cmd.c nvme_ns.c nvme_pcie.c nvme_qpair.c nvme.c nvme_quirks.c nvme_transport.c nvme_uevent.c nvme_ctrlr_ocssd_cmd.c \
		nvme_ns_ocssd_cmd.c nvme_tcp.c nvme_opal.c
		nvme_ns_ocssd_cmd.c nvme_tcp.c nvme_opal.c nvme_io_msg.c
		C_SRCS-$(CONFIG_RDMA) += nvme_rdma.c
		LIBNAME = nvme
		LOCAL_SYS_LIBS = -luuid

lib/nvme/nvme.c

+4 −0

Original line number	Diff line number	Diff line
		@@ -33,6 +33,7 @@

		#include "spdk/nvmf_spec.h"
		#include "nvme_internal.h"
		#include "nvme_io_msg.h"

		#define SPDK_NVME_DRIVER_NAME "spdk_nvme_driver"

		@@ -68,6 +69,7 @@ spdk_nvme_detach(struct spdk_nvme_ctrlr *ctrlr)
		nvme_ctrlr_proc_put_ref(ctrlr);

		if (nvme_ctrlr_get_ref_count(ctrlr) == 0) {
		nvme_io_msg_ctrlr_stop(ctrlr, NULL, true);
		if (nvme_ctrlr_shared(ctrlr)) {
		TAILQ_REMOVE(&g_spdk_nvme_driver->shared_attached_ctrlrs, ctrlr, tailq);
		} else {
		@@ -458,6 +460,8 @@ nvme_ctrlr_poll_internal(struct spdk_nvme_ctrlr *ctrlr,
		return 0;
		}

		STAILQ_INIT(&ctrlr->io_producers);

		/*
		* Controller has been initialized.
		* Move it to the attached_ctrlrs list.

lib/nvme/nvme_internal.h

+6 −0

Original line number	Diff line number	Diff line
		@@ -714,6 +714,12 @@ struct spdk_nvme_ctrlr {
		/* CB to notify the user when the ctrlr is removed/failed. */
		spdk_nvme_remove_cb remove_cb;
		void *cb_ctx;

		struct spdk_nvme_qpair *external_io_msgs_qpair;
		pthread_mutex_t external_io_msgs_lock;
		struct spdk_ring *external_io_msgs;

		STAILQ_HEAD(, nvme_io_msg_producer) io_producers;
		};

		struct spdk_nvme_probe_ctx {

lib/nvme/nvme_io_msg.c

0 → 100644

+172 −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 "nvme_internal.h"
		#include "nvme_io_msg.h"

		#define SPDK_NVME_MSG_IO_PROCESS_SIZE 8

		/**
		* Send message to IO queue.
		*/
		int
		nvme_io_msg_send(struct spdk_nvme_ctrlr *ctrlr, uint32_t nsid, spdk_nvme_io_msg_fn fn,
		void *arg)
		{
		int rc;
		struct spdk_nvme_io_msg *io;

		/* Protect requests ring against preemptive producers */
		pthread_mutex_lock(&ctrlr->external_io_msgs_lock);

		io = (struct spdk_nvme_io_msg *)calloc(1, sizeof(struct spdk_nvme_io_msg));
		if (!io) {
		SPDK_ERRLOG("IO msg allocation failed.");
		pthread_mutex_unlock(&ctrlr->external_io_msgs_lock);
		return -ENOMEM;
		}

		io->ctrlr = ctrlr;
		io->nsid = nsid;
		io->fn = fn;
		io->arg = arg;

		rc = spdk_ring_enqueue(ctrlr->external_io_msgs, (void **)&io, 1, NULL);
		if (rc != 1) {
		assert(false);
		pthread_mutex_unlock(&ctrlr->external_io_msgs_lock);
		return -ENOMEM;
		}

		pthread_mutex_unlock(&ctrlr->external_io_msgs_lock);

		return 0;
		}

		int
		spdk_nvme_io_msg_process(struct spdk_nvme_ctrlr *ctrlr)
		{
		int i;
		int count;
		struct spdk_nvme_io_msg *io;
		void *requests[SPDK_NVME_MSG_IO_PROCESS_SIZE];

		if (!ctrlr->external_io_msgs \|\| !ctrlr->external_io_msgs_qpair) {
		/* Not ready or pending reset */
		return 0;
		}

		spdk_nvme_qpair_process_completions(ctrlr->external_io_msgs_qpair, 0);

		count = spdk_ring_dequeue(ctrlr->external_io_msgs, requests,
		SPDK_NVME_MSG_IO_PROCESS_SIZE);
		if (count == 0) {
		return 0;
		}

		for (i = 0; i < count; i++) {
		io = requests[i];

		assert(io != NULL);

		io->fn(io->ctrlr, io->nsid, io->arg);
		free(io);
		}

		return count;
		}

		int
		nvme_io_msg_ctrlr_start(struct spdk_nvme_ctrlr ctrlr, struct nvme_io_msg_producer io_msg_producer)
		{
		if (io_msg_producer == NULL) {
		SPDK_ERRLOG("io_msg_producer cannot be NULL\n");
		return -EINVAL;
		}

		if (!STAILQ_EMPTY(&ctrlr->io_producers) \|\| ctrlr->is_resetting) {
		/* There are registered producers - IO messaging already started */
		STAILQ_INSERT_TAIL(&ctrlr->io_producers, io_msg_producer, link);
		return 0;
		}

		pthread_mutex_init(&ctrlr->external_io_msgs_lock, NULL);

		/**
		* Initialize ring and qpair for controller
		*/
		ctrlr->external_io_msgs = spdk_ring_create(SPDK_RING_TYPE_MP_SC, 65536, SPDK_ENV_SOCKET_ID_ANY);
		if (!ctrlr->external_io_msgs) {
		SPDK_ERRLOG("Unable to allocate memory for message ring\n");
		return -ENOMEM;
		}

		ctrlr->external_io_msgs_qpair = spdk_nvme_ctrlr_alloc_io_qpair(ctrlr, NULL, 0);
		if (ctrlr->external_io_msgs_qpair == NULL) {
		SPDK_ERRLOG("spdk_nvme_ctrlr_alloc_io_qpair() failed\n");
		spdk_ring_free(ctrlr->external_io_msgs);
		return -1;
		}

		STAILQ_INSERT_TAIL(&ctrlr->io_producers, io_msg_producer, link);

		return 0;
		}

		void
		nvme_io_msg_ctrlr_stop(struct spdk_nvme_ctrlr ctrlr, struct nvme_io_msg_producer io_msg_producer,
		bool shutdown)
		{
		if (STAILQ_EMPTY(&ctrlr->io_producers) && shutdown) {
		/* Shutdown path with no producers registered = io msg ctrlr not started */
		return;
		}

		if (io_msg_producer != NULL) {
		STAILQ_REMOVE(&ctrlr->io_producers, io_msg_producer, nvme_io_msg_producer, link);
		}

		if (!STAILQ_EMPTY(&ctrlr->io_producers) && !shutdown) {
		/* There are still some registered producers */
		return;
		}

		assert(ctrlr->external_io_msgs);
		spdk_ring_free(ctrlr->external_io_msgs);

		if (ctrlr->external_io_msgs_qpair) {
		spdk_nvme_ctrlr_free_io_qpair(ctrlr->external_io_msgs_qpair);
		}

		pthread_mutex_destroy(&ctrlr->external_io_msgs_lock);
		}