diff options
Diffstat (limited to '')
| -rw-r--r-- | src/spdk/lib/bdev/bdev.c | 3950 |
1 files changed, 3950 insertions, 0 deletions
diff --git a/src/spdk/lib/bdev/bdev.c b/src/spdk/lib/bdev/bdev.c new file mode 100644 index 00000000..ab82fffd --- /dev/null +++ b/src/spdk/lib/bdev/bdev.c @@ -0,0 +1,3950 @@ +/*- + * 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/stdinc.h" + +#include "spdk/bdev.h" +#include "spdk/conf.h" + +#include "spdk/config.h" +#include "spdk/env.h" +#include "spdk/event.h" +#include "spdk/thread.h" +#include "spdk/likely.h" +#include "spdk/queue.h" +#include "spdk/nvme_spec.h" +#include "spdk/scsi_spec.h" +#include "spdk/util.h" +#include "spdk/trace.h" + +#include "spdk/bdev_module.h" +#include "spdk_internal/log.h" +#include "spdk/string.h" + +#ifdef SPDK_CONFIG_VTUNE +#include "ittnotify.h" +#include "ittnotify_types.h" +int __itt_init_ittlib(const char *, __itt_group_id); +#endif + +#define SPDK_BDEV_IO_POOL_SIZE (64 * 1024) +#define SPDK_BDEV_IO_CACHE_SIZE 256 +#define BUF_SMALL_POOL_SIZE 8192 +#define BUF_LARGE_POOL_SIZE 1024 +#define NOMEM_THRESHOLD_COUNT 8 +#define ZERO_BUFFER_SIZE 0x100000 + +#define OWNER_BDEV 0x2 + +#define OBJECT_BDEV_IO 0x2 + +#define TRACE_GROUP_BDEV 0x3 +#define TRACE_BDEV_IO_START SPDK_TPOINT_ID(TRACE_GROUP_BDEV, 0x0) +#define TRACE_BDEV_IO_DONE SPDK_TPOINT_ID(TRACE_GROUP_BDEV, 0x1) + +#define SPDK_BDEV_QOS_TIMESLICE_IN_USEC 1000 +#define SPDK_BDEV_QOS_MIN_IO_PER_TIMESLICE 1 +#define SPDK_BDEV_QOS_MIN_BYTE_PER_TIMESLICE 512 +#define SPDK_BDEV_QOS_MIN_IOS_PER_SEC 10000 +#define SPDK_BDEV_QOS_MIN_BYTES_PER_SEC (10 * 1024 * 1024) +#define SPDK_BDEV_QOS_LIMIT_NOT_DEFINED UINT64_MAX + +static const char *qos_conf_type[] = {"Limit_IOPS", "Limit_BPS"}; +static const char *qos_rpc_type[] = {"rw_ios_per_sec", "rw_mbytes_per_sec"}; + +TAILQ_HEAD(spdk_bdev_list, spdk_bdev); + +struct spdk_bdev_mgr { + struct spdk_mempool *bdev_io_pool; + + struct spdk_mempool *buf_small_pool; + struct spdk_mempool *buf_large_pool; + + void *zero_buffer; + + TAILQ_HEAD(bdev_module_list, spdk_bdev_module) bdev_modules; + + struct spdk_bdev_list bdevs; + + bool init_complete; + bool module_init_complete; + +#ifdef SPDK_CONFIG_VTUNE + __itt_domain *domain; +#endif +}; + +static struct spdk_bdev_mgr g_bdev_mgr = { + .bdev_modules = TAILQ_HEAD_INITIALIZER(g_bdev_mgr.bdev_modules), + .bdevs = TAILQ_HEAD_INITIALIZER(g_bdev_mgr.bdevs), + .init_complete = false, + .module_init_complete = false, +}; + +static struct spdk_bdev_opts g_bdev_opts = { + .bdev_io_pool_size = SPDK_BDEV_IO_POOL_SIZE, + .bdev_io_cache_size = SPDK_BDEV_IO_CACHE_SIZE, +}; + +static spdk_bdev_init_cb g_init_cb_fn = NULL; +static void *g_init_cb_arg = NULL; + +static spdk_bdev_fini_cb g_fini_cb_fn = NULL; +static void *g_fini_cb_arg = NULL; +static struct spdk_thread *g_fini_thread = NULL; + +struct spdk_bdev_qos_limit { + /** IOs or bytes allowed per second (i.e., 1s). */ + uint64_t limit; + + /** Remaining IOs or bytes allowed in current timeslice (e.g., 1ms). + * For remaining bytes, allowed to run negative if an I/O is submitted when + * some bytes are remaining, but the I/O is bigger than that amount. The + * excess will be deducted from the next timeslice. + */ + int64_t remaining_this_timeslice; + + /** Minimum allowed IOs or bytes to be issued in one timeslice (e.g., 1ms). */ + uint32_t min_per_timeslice; + + /** Maximum allowed IOs or bytes to be issued in one timeslice (e.g., 1ms). */ + uint32_t max_per_timeslice; +}; + +struct spdk_bdev_qos { + /** Types of structure of rate limits. */ + struct spdk_bdev_qos_limit rate_limits[SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES]; + + /** The channel that all I/O are funneled through. */ + struct spdk_bdev_channel *ch; + + /** The thread on which the poller is running. */ + struct spdk_thread *thread; + + /** Queue of I/O waiting to be issued. */ + bdev_io_tailq_t queued; + + /** Size of a timeslice in tsc ticks. */ + uint64_t timeslice_size; + + /** Timestamp of start of last timeslice. */ + uint64_t last_timeslice; + + /** Poller that processes queued I/O commands each time slice. */ + struct spdk_poller *poller; +}; + +struct spdk_bdev_mgmt_channel { + bdev_io_stailq_t need_buf_small; + bdev_io_stailq_t need_buf_large; + + /* + * Each thread keeps a cache of bdev_io - this allows + * bdev threads which are *not* DPDK threads to still + * benefit from a per-thread bdev_io cache. Without + * this, non-DPDK threads fetching from the mempool + * incur a cmpxchg on get and put. + */ + bdev_io_stailq_t per_thread_cache; + uint32_t per_thread_cache_count; + uint32_t bdev_io_cache_size; + + TAILQ_HEAD(, spdk_bdev_shared_resource) shared_resources; + TAILQ_HEAD(, spdk_bdev_io_wait_entry) io_wait_queue; +}; + +/* + * Per-module (or per-io_device) data. Multiple bdevs built on the same io_device + * will queue here their IO that awaits retry. It makes it possible to retry sending + * IO to one bdev after IO from other bdev completes. + */ +struct spdk_bdev_shared_resource { + /* The bdev management channel */ + struct spdk_bdev_mgmt_channel *mgmt_ch; + + /* + * Count of I/O submitted to bdev module and waiting for completion. + * Incremented before submit_request() is called on an spdk_bdev_io. + */ + uint64_t io_outstanding; + + /* + * Queue of IO awaiting retry because of a previous NOMEM status returned + * on this channel. + */ + bdev_io_tailq_t nomem_io; + + /* + * Threshold which io_outstanding must drop to before retrying nomem_io. + */ + uint64_t nomem_threshold; + + /* I/O channel allocated by a bdev module */ + struct spdk_io_channel *shared_ch; + + /* Refcount of bdev channels using this resource */ + uint32_t ref; + + TAILQ_ENTRY(spdk_bdev_shared_resource) link; +}; + +#define BDEV_CH_RESET_IN_PROGRESS (1 << 0) +#define BDEV_CH_QOS_ENABLED (1 << 1) + +struct spdk_bdev_channel { + struct spdk_bdev *bdev; + + /* The channel for the underlying device */ + struct spdk_io_channel *channel; + + /* Per io_device per thread data */ + struct spdk_bdev_shared_resource *shared_resource; + + struct spdk_bdev_io_stat stat; + + /* + * Count of I/O submitted through this channel and waiting for completion. + * Incremented before submit_request() is called on an spdk_bdev_io. + */ + uint64_t io_outstanding; + + bdev_io_tailq_t queued_resets; + + uint32_t flags; + +#ifdef SPDK_CONFIG_VTUNE + uint64_t start_tsc; + uint64_t interval_tsc; + __itt_string_handle *handle; + struct spdk_bdev_io_stat prev_stat; +#endif + +}; + +struct spdk_bdev_desc { + struct spdk_bdev *bdev; + struct spdk_thread *thread; + spdk_bdev_remove_cb_t remove_cb; + void *remove_ctx; + bool remove_scheduled; + bool closed; + bool write; + TAILQ_ENTRY(spdk_bdev_desc) link; +}; + +struct spdk_bdev_iostat_ctx { + struct spdk_bdev_io_stat *stat; + spdk_bdev_get_device_stat_cb cb; + void *cb_arg; +}; + +#define __bdev_to_io_dev(bdev) (((char *)bdev) + 1) +#define __bdev_from_io_dev(io_dev) ((struct spdk_bdev *)(((char *)io_dev) - 1)) + +static void _spdk_bdev_write_zero_buffer_done(struct spdk_bdev_io *bdev_io, bool success, + void *cb_arg); +static void _spdk_bdev_write_zero_buffer_next(void *_bdev_io); + +void +spdk_bdev_get_opts(struct spdk_bdev_opts *opts) +{ + *opts = g_bdev_opts; +} + +int +spdk_bdev_set_opts(struct spdk_bdev_opts *opts) +{ + uint32_t min_pool_size; + + /* + * Add 1 to the thread count to account for the extra mgmt_ch that gets created during subsystem + * initialization. A second mgmt_ch will be created on the same thread when the application starts + * but before the deferred put_io_channel event is executed for the first mgmt_ch. + */ + min_pool_size = opts->bdev_io_cache_size * (spdk_thread_get_count() + 1); + if (opts->bdev_io_pool_size < min_pool_size) { + SPDK_ERRLOG("bdev_io_pool_size %" PRIu32 " is not compatible with bdev_io_cache_size %" PRIu32 + " and %" PRIu32 " threads\n", opts->bdev_io_pool_size, opts->bdev_io_cache_size, + spdk_thread_get_count()); + SPDK_ERRLOG("bdev_io_pool_size must be at least %" PRIu32 "\n", min_pool_size); + return -1; + } + + g_bdev_opts = *opts; + return 0; +} + +struct spdk_bdev * +spdk_bdev_first(void) +{ + struct spdk_bdev *bdev; + + bdev = TAILQ_FIRST(&g_bdev_mgr.bdevs); + if (bdev) { + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Starting bdev iteration at %s\n", bdev->name); + } + + return bdev; +} + +struct spdk_bdev * +spdk_bdev_next(struct spdk_bdev *prev) +{ + struct spdk_bdev *bdev; + + bdev = TAILQ_NEXT(prev, internal.link); + if (bdev) { + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Continuing bdev iteration at %s\n", bdev->name); + } + + return bdev; +} + +static struct spdk_bdev * +_bdev_next_leaf(struct spdk_bdev *bdev) +{ + while (bdev != NULL) { + if (bdev->internal.claim_module == NULL) { + return bdev; + } else { + bdev = TAILQ_NEXT(bdev, internal.link); + } + } + + return bdev; +} + +struct spdk_bdev * +spdk_bdev_first_leaf(void) +{ + struct spdk_bdev *bdev; + + bdev = _bdev_next_leaf(TAILQ_FIRST(&g_bdev_mgr.bdevs)); + + if (bdev) { + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Starting bdev iteration at %s\n", bdev->name); + } + + return bdev; +} + +struct spdk_bdev * +spdk_bdev_next_leaf(struct spdk_bdev *prev) +{ + struct spdk_bdev *bdev; + + bdev = _bdev_next_leaf(TAILQ_NEXT(prev, internal.link)); + + if (bdev) { + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Continuing bdev iteration at %s\n", bdev->name); + } + + return bdev; +} + +struct spdk_bdev * +spdk_bdev_get_by_name(const char *bdev_name) +{ + struct spdk_bdev_alias *tmp; + struct spdk_bdev *bdev = spdk_bdev_first(); + + while (bdev != NULL) { + if (strcmp(bdev_name, bdev->name) == 0) { + return bdev; + } + + TAILQ_FOREACH(tmp, &bdev->aliases, tailq) { + if (strcmp(bdev_name, tmp->alias) == 0) { + return bdev; + } + } + + bdev = spdk_bdev_next(bdev); + } + + return NULL; +} + +void +spdk_bdev_io_set_buf(struct spdk_bdev_io *bdev_io, void *buf, size_t len) +{ + struct iovec *iovs; + + iovs = bdev_io->u.bdev.iovs; + + assert(iovs != NULL); + assert(bdev_io->u.bdev.iovcnt >= 1); + + iovs[0].iov_base = buf; + iovs[0].iov_len = len; +} + +static void +spdk_bdev_io_put_buf(struct spdk_bdev_io *bdev_io) +{ + struct spdk_mempool *pool; + struct spdk_bdev_io *tmp; + void *buf, *aligned_buf; + bdev_io_stailq_t *stailq; + struct spdk_bdev_mgmt_channel *ch; + + assert(bdev_io->u.bdev.iovcnt == 1); + + buf = bdev_io->internal.buf; + ch = bdev_io->internal.ch->shared_resource->mgmt_ch; + + bdev_io->internal.buf = NULL; + + if (bdev_io->internal.buf_len <= SPDK_BDEV_SMALL_BUF_MAX_SIZE) { + pool = g_bdev_mgr.buf_small_pool; + stailq = &ch->need_buf_small; + } else { + pool = g_bdev_mgr.buf_large_pool; + stailq = &ch->need_buf_large; + } + + if (STAILQ_EMPTY(stailq)) { + spdk_mempool_put(pool, buf); + } else { + tmp = STAILQ_FIRST(stailq); + + aligned_buf = (void *)(((uintptr_t)buf + 511) & ~511UL); + spdk_bdev_io_set_buf(tmp, aligned_buf, tmp->internal.buf_len); + + STAILQ_REMOVE_HEAD(stailq, internal.buf_link); + tmp->internal.buf = buf; + tmp->internal.get_buf_cb(tmp->internal.ch->channel, tmp); + } +} + +void +spdk_bdev_io_get_buf(struct spdk_bdev_io *bdev_io, spdk_bdev_io_get_buf_cb cb, uint64_t len) +{ + struct spdk_mempool *pool; + bdev_io_stailq_t *stailq; + void *buf, *aligned_buf; + struct spdk_bdev_mgmt_channel *mgmt_ch; + + assert(cb != NULL); + assert(bdev_io->u.bdev.iovs != NULL); + + if (spdk_unlikely(bdev_io->u.bdev.iovs[0].iov_base != NULL)) { + /* Buffer already present */ + cb(bdev_io->internal.ch->channel, bdev_io); + return; + } + + assert(len <= SPDK_BDEV_LARGE_BUF_MAX_SIZE); + mgmt_ch = bdev_io->internal.ch->shared_resource->mgmt_ch; + + bdev_io->internal.buf_len = len; + bdev_io->internal.get_buf_cb = cb; + if (len <= SPDK_BDEV_SMALL_BUF_MAX_SIZE) { + pool = g_bdev_mgr.buf_small_pool; + stailq = &mgmt_ch->need_buf_small; + } else { + pool = g_bdev_mgr.buf_large_pool; + stailq = &mgmt_ch->need_buf_large; + } + + buf = spdk_mempool_get(pool); + + if (!buf) { + STAILQ_INSERT_TAIL(stailq, bdev_io, internal.buf_link); + } else { + aligned_buf = (void *)(((uintptr_t)buf + 511) & ~511UL); + spdk_bdev_io_set_buf(bdev_io, aligned_buf, len); + + bdev_io->internal.buf = buf; + bdev_io->internal.get_buf_cb(bdev_io->internal.ch->channel, bdev_io); + } +} + +static int +spdk_bdev_module_get_max_ctx_size(void) +{ + struct spdk_bdev_module *bdev_module; + int max_bdev_module_size = 0; + + TAILQ_FOREACH(bdev_module, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (bdev_module->get_ctx_size && bdev_module->get_ctx_size() > max_bdev_module_size) { + max_bdev_module_size = bdev_module->get_ctx_size(); + } + } + + return max_bdev_module_size; +} + +void +spdk_bdev_config_text(FILE *fp) +{ + struct spdk_bdev_module *bdev_module; + + TAILQ_FOREACH(bdev_module, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (bdev_module->config_text) { + bdev_module->config_text(fp); + } + } +} + +static void +spdk_bdev_qos_config_json(struct spdk_bdev *bdev, struct spdk_json_write_ctx *w) +{ + int i; + struct spdk_bdev_qos *qos = bdev->internal.qos; + uint64_t limits[SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES]; + + if (!qos) { + return; + } + + spdk_bdev_get_qos_rate_limits(bdev, limits); + + spdk_json_write_object_begin(w); + spdk_json_write_named_string(w, "method", "set_bdev_qos_limit"); + spdk_json_write_name(w, "params"); + + spdk_json_write_object_begin(w); + spdk_json_write_named_string(w, "name", bdev->name); + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (limits[i] > 0) { + spdk_json_write_named_uint64(w, qos_rpc_type[i], limits[i]); + } + } + spdk_json_write_object_end(w); + + spdk_json_write_object_end(w); +} + +void +spdk_bdev_subsystem_config_json(struct spdk_json_write_ctx *w) +{ + struct spdk_bdev_module *bdev_module; + struct spdk_bdev *bdev; + + assert(w != NULL); + + spdk_json_write_array_begin(w); + + spdk_json_write_object_begin(w); + spdk_json_write_named_string(w, "method", "set_bdev_options"); + spdk_json_write_name(w, "params"); + spdk_json_write_object_begin(w); + spdk_json_write_named_uint32(w, "bdev_io_pool_size", g_bdev_opts.bdev_io_pool_size); + spdk_json_write_named_uint32(w, "bdev_io_cache_size", g_bdev_opts.bdev_io_cache_size); + spdk_json_write_object_end(w); + spdk_json_write_object_end(w); + + TAILQ_FOREACH(bdev_module, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (bdev_module->config_json) { + bdev_module->config_json(w); + } + } + + TAILQ_FOREACH(bdev, &g_bdev_mgr.bdevs, internal.link) { + spdk_bdev_qos_config_json(bdev, w); + + if (bdev->fn_table->write_config_json) { + bdev->fn_table->write_config_json(bdev, w); + } + } + + spdk_json_write_array_end(w); +} + +static int +spdk_bdev_mgmt_channel_create(void *io_device, void *ctx_buf) +{ + struct spdk_bdev_mgmt_channel *ch = ctx_buf; + struct spdk_bdev_io *bdev_io; + uint32_t i; + + STAILQ_INIT(&ch->need_buf_small); + STAILQ_INIT(&ch->need_buf_large); + + STAILQ_INIT(&ch->per_thread_cache); + ch->bdev_io_cache_size = g_bdev_opts.bdev_io_cache_size; + + /* Pre-populate bdev_io cache to ensure this thread cannot be starved. */ + ch->per_thread_cache_count = 0; + for (i = 0; i < ch->bdev_io_cache_size; i++) { + bdev_io = spdk_mempool_get(g_bdev_mgr.bdev_io_pool); + assert(bdev_io != NULL); + ch->per_thread_cache_count++; + STAILQ_INSERT_TAIL(&ch->per_thread_cache, bdev_io, internal.buf_link); + } + + TAILQ_INIT(&ch->shared_resources); + TAILQ_INIT(&ch->io_wait_queue); + + return 0; +} + +static void +spdk_bdev_mgmt_channel_destroy(void *io_device, void *ctx_buf) +{ + struct spdk_bdev_mgmt_channel *ch = ctx_buf; + struct spdk_bdev_io *bdev_io; + + if (!STAILQ_EMPTY(&ch->need_buf_small) || !STAILQ_EMPTY(&ch->need_buf_large)) { + SPDK_ERRLOG("Pending I/O list wasn't empty on mgmt channel free\n"); + } + + if (!TAILQ_EMPTY(&ch->shared_resources)) { + SPDK_ERRLOG("Module channel list wasn't empty on mgmt channel free\n"); + } + + while (!STAILQ_EMPTY(&ch->per_thread_cache)) { + bdev_io = STAILQ_FIRST(&ch->per_thread_cache); + STAILQ_REMOVE_HEAD(&ch->per_thread_cache, internal.buf_link); + ch->per_thread_cache_count--; + spdk_mempool_put(g_bdev_mgr.bdev_io_pool, (void *)bdev_io); + } + + assert(ch->per_thread_cache_count == 0); +} + +static void +spdk_bdev_init_complete(int rc) +{ + spdk_bdev_init_cb cb_fn = g_init_cb_fn; + void *cb_arg = g_init_cb_arg; + struct spdk_bdev_module *m; + + g_bdev_mgr.init_complete = true; + g_init_cb_fn = NULL; + g_init_cb_arg = NULL; + + /* + * For modules that need to know when subsystem init is complete, + * inform them now. + */ + if (rc == 0) { + TAILQ_FOREACH(m, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (m->init_complete) { + m->init_complete(); + } + } + } + + cb_fn(cb_arg, rc); +} + +static void +spdk_bdev_module_action_complete(void) +{ + struct spdk_bdev_module *m; + + /* + * Don't finish bdev subsystem initialization if + * module pre-initialization is still in progress, or + * the subsystem been already initialized. + */ + if (!g_bdev_mgr.module_init_complete || g_bdev_mgr.init_complete) { + return; + } + + /* + * Check all bdev modules for inits/examinations in progress. If any + * exist, return immediately since we cannot finish bdev subsystem + * initialization until all are completed. + */ + TAILQ_FOREACH(m, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (m->internal.action_in_progress > 0) { + return; + } + } + + /* + * Modules already finished initialization - now that all + * the bdev modules have finished their asynchronous I/O + * processing, the entire bdev layer can be marked as complete. + */ + spdk_bdev_init_complete(0); +} + +static void +spdk_bdev_module_action_done(struct spdk_bdev_module *module) +{ + assert(module->internal.action_in_progress > 0); + module->internal.action_in_progress--; + spdk_bdev_module_action_complete(); +} + +void +spdk_bdev_module_init_done(struct spdk_bdev_module *module) +{ + spdk_bdev_module_action_done(module); +} + +void +spdk_bdev_module_examine_done(struct spdk_bdev_module *module) +{ + spdk_bdev_module_action_done(module); +} + +/** The last initialized bdev module */ +static struct spdk_bdev_module *g_resume_bdev_module = NULL; + +static int +spdk_bdev_modules_init(void) +{ + struct spdk_bdev_module *module; + int rc = 0; + + TAILQ_FOREACH(module, &g_bdev_mgr.bdev_modules, internal.tailq) { + g_resume_bdev_module = module; + rc = module->module_init(); + if (rc != 0) { + return rc; + } + } + + g_resume_bdev_module = NULL; + return 0; +} + + +static void +spdk_bdev_init_failed_complete(void *cb_arg) +{ + spdk_bdev_init_complete(-1); +} + +static void +spdk_bdev_init_failed(void *cb_arg) +{ + spdk_bdev_finish(spdk_bdev_init_failed_complete, NULL); +} + +void +spdk_bdev_initialize(spdk_bdev_init_cb cb_fn, void *cb_arg) +{ + struct spdk_conf_section *sp; + struct spdk_bdev_opts bdev_opts; + int32_t bdev_io_pool_size, bdev_io_cache_size; + int cache_size; + int rc = 0; + char mempool_name[32]; + + assert(cb_fn != NULL); + + sp = spdk_conf_find_section(NULL, "Bdev"); + if (sp != NULL) { + spdk_bdev_get_opts(&bdev_opts); + + bdev_io_pool_size = spdk_conf_section_get_intval(sp, "BdevIoPoolSize"); + if (bdev_io_pool_size >= 0) { + bdev_opts.bdev_io_pool_size = bdev_io_pool_size; + } + + bdev_io_cache_size = spdk_conf_section_get_intval(sp, "BdevIoCacheSize"); + if (bdev_io_cache_size >= 0) { + bdev_opts.bdev_io_cache_size = bdev_io_cache_size; + } + + if (spdk_bdev_set_opts(&bdev_opts)) { + spdk_bdev_init_complete(-1); + return; + } + + assert(memcmp(&bdev_opts, &g_bdev_opts, sizeof(bdev_opts)) == 0); + } + + g_init_cb_fn = cb_fn; + g_init_cb_arg = cb_arg; + + snprintf(mempool_name, sizeof(mempool_name), "bdev_io_%d", getpid()); + + g_bdev_mgr.bdev_io_pool = spdk_mempool_create(mempool_name, + g_bdev_opts.bdev_io_pool_size, + sizeof(struct spdk_bdev_io) + + spdk_bdev_module_get_max_ctx_size(), + 0, + SPDK_ENV_SOCKET_ID_ANY); + + if (g_bdev_mgr.bdev_io_pool == NULL) { + SPDK_ERRLOG("could not allocate spdk_bdev_io pool\n"); + spdk_bdev_init_complete(-1); + return; + } + + /** + * Ensure no more than half of the total buffers end up local caches, by + * using spdk_thread_get_count() to determine how many local caches we need + * to account for. + */ + cache_size = BUF_SMALL_POOL_SIZE / (2 * spdk_thread_get_count()); + snprintf(mempool_name, sizeof(mempool_name), "buf_small_pool_%d", getpid()); + + g_bdev_mgr.buf_small_pool = spdk_mempool_create(mempool_name, + BUF_SMALL_POOL_SIZE, + SPDK_BDEV_SMALL_BUF_MAX_SIZE + 512, + cache_size, + SPDK_ENV_SOCKET_ID_ANY); + if (!g_bdev_mgr.buf_small_pool) { + SPDK_ERRLOG("create rbuf small pool failed\n"); + spdk_bdev_init_complete(-1); + return; + } + + cache_size = BUF_LARGE_POOL_SIZE / (2 * spdk_thread_get_count()); + snprintf(mempool_name, sizeof(mempool_name), "buf_large_pool_%d", getpid()); + + g_bdev_mgr.buf_large_pool = spdk_mempool_create(mempool_name, + BUF_LARGE_POOL_SIZE, + SPDK_BDEV_LARGE_BUF_MAX_SIZE + 512, + cache_size, + SPDK_ENV_SOCKET_ID_ANY); + if (!g_bdev_mgr.buf_large_pool) { + SPDK_ERRLOG("create rbuf large pool failed\n"); + spdk_bdev_init_complete(-1); + return; + } + + g_bdev_mgr.zero_buffer = spdk_dma_zmalloc(ZERO_BUFFER_SIZE, ZERO_BUFFER_SIZE, + NULL); + if (!g_bdev_mgr.zero_buffer) { + SPDK_ERRLOG("create bdev zero buffer failed\n"); + spdk_bdev_init_complete(-1); + return; + } + +#ifdef SPDK_CONFIG_VTUNE + g_bdev_mgr.domain = __itt_domain_create("spdk_bdev"); +#endif + + spdk_io_device_register(&g_bdev_mgr, spdk_bdev_mgmt_channel_create, + spdk_bdev_mgmt_channel_destroy, + sizeof(struct spdk_bdev_mgmt_channel), + "bdev_mgr"); + + rc = spdk_bdev_modules_init(); + g_bdev_mgr.module_init_complete = true; + if (rc != 0) { + SPDK_ERRLOG("bdev modules init failed\n"); + spdk_thread_send_msg(spdk_get_thread(), spdk_bdev_init_failed, NULL); + return; + } + + spdk_bdev_module_action_complete(); +} + +static void +spdk_bdev_mgr_unregister_cb(void *io_device) +{ + spdk_bdev_fini_cb cb_fn = g_fini_cb_fn; + + if (spdk_mempool_count(g_bdev_mgr.bdev_io_pool) != g_bdev_opts.bdev_io_pool_size) { + SPDK_ERRLOG("bdev IO pool count is %zu but should be %u\n", + spdk_mempool_count(g_bdev_mgr.bdev_io_pool), + g_bdev_opts.bdev_io_pool_size); + } + + if (spdk_mempool_count(g_bdev_mgr.buf_small_pool) != BUF_SMALL_POOL_SIZE) { + SPDK_ERRLOG("Small buffer pool count is %zu but should be %u\n", + spdk_mempool_count(g_bdev_mgr.buf_small_pool), + BUF_SMALL_POOL_SIZE); + assert(false); + } + + if (spdk_mempool_count(g_bdev_mgr.buf_large_pool) != BUF_LARGE_POOL_SIZE) { + SPDK_ERRLOG("Large buffer pool count is %zu but should be %u\n", + spdk_mempool_count(g_bdev_mgr.buf_large_pool), + BUF_LARGE_POOL_SIZE); + assert(false); + } + + spdk_mempool_free(g_bdev_mgr.bdev_io_pool); + spdk_mempool_free(g_bdev_mgr.buf_small_pool); + spdk_mempool_free(g_bdev_mgr.buf_large_pool); + spdk_dma_free(g_bdev_mgr.zero_buffer); + + cb_fn(g_fini_cb_arg); + g_fini_cb_fn = NULL; + g_fini_cb_arg = NULL; + g_bdev_mgr.init_complete = false; + g_bdev_mgr.module_init_complete = false; +} + +static void +spdk_bdev_module_finish_iter(void *arg) +{ + struct spdk_bdev_module *bdev_module; + + /* Start iterating from the last touched module */ + if (!g_resume_bdev_module) { + bdev_module = TAILQ_LAST(&g_bdev_mgr.bdev_modules, bdev_module_list); + } else { + bdev_module = TAILQ_PREV(g_resume_bdev_module, bdev_module_list, + internal.tailq); + } + + while (bdev_module) { + if (bdev_module->async_fini) { + /* Save our place so we can resume later. We must + * save the variable here, before calling module_fini() + * below, because in some cases the module may immediately + * call spdk_bdev_module_finish_done() and re-enter + * this function to continue iterating. */ + g_resume_bdev_module = bdev_module; + } + + if (bdev_module->module_fini) { + bdev_module->module_fini(); + } + + if (bdev_module->async_fini) { + return; + } + + bdev_module = TAILQ_PREV(bdev_module, bdev_module_list, + internal.tailq); + } + + g_resume_bdev_module = NULL; + spdk_io_device_unregister(&g_bdev_mgr, spdk_bdev_mgr_unregister_cb); +} + +void +spdk_bdev_module_finish_done(void) +{ + if (spdk_get_thread() != g_fini_thread) { + spdk_thread_send_msg(g_fini_thread, spdk_bdev_module_finish_iter, NULL); + } else { + spdk_bdev_module_finish_iter(NULL); + } +} + +static void +_spdk_bdev_finish_unregister_bdevs_iter(void *cb_arg, int bdeverrno) +{ + struct spdk_bdev *bdev = cb_arg; + + if (bdeverrno && bdev) { + SPDK_WARNLOG("Unable to unregister bdev '%s' during spdk_bdev_finish()\n", + bdev->name); + + /* + * Since the call to spdk_bdev_unregister() failed, we have no way to free this + * bdev; try to continue by manually removing this bdev from the list and continue + * with the next bdev in the list. + */ + TAILQ_REMOVE(&g_bdev_mgr.bdevs, bdev, internal.link); + } + + if (TAILQ_EMPTY(&g_bdev_mgr.bdevs)) { + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Done unregistering bdevs\n"); + /* + * Bdev module finish need to be deffered as we might be in the middle of some context + * (like bdev part free) that will use this bdev (or private bdev driver ctx data) + * after returning. + */ + spdk_thread_send_msg(spdk_get_thread(), spdk_bdev_module_finish_iter, NULL); + return; + } + + /* + * Unregister the last bdev in the list. The last bdev in the list should be a bdev + * that has no bdevs that depend on it. + */ + bdev = TAILQ_LAST(&g_bdev_mgr.bdevs, spdk_bdev_list); + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Unregistering bdev '%s'\n", bdev->name); + spdk_bdev_unregister(bdev, _spdk_bdev_finish_unregister_bdevs_iter, bdev); +} + +void +spdk_bdev_finish(spdk_bdev_fini_cb cb_fn, void *cb_arg) +{ + struct spdk_bdev_module *m; + + assert(cb_fn != NULL); + + g_fini_thread = spdk_get_thread(); + + g_fini_cb_fn = cb_fn; + g_fini_cb_arg = cb_arg; + + TAILQ_FOREACH(m, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (m->fini_start) { + m->fini_start(); + } + } + + _spdk_bdev_finish_unregister_bdevs_iter(NULL, 0); +} + +static struct spdk_bdev_io * +spdk_bdev_get_io(struct spdk_bdev_channel *channel) +{ + struct spdk_bdev_mgmt_channel *ch = channel->shared_resource->mgmt_ch; + struct spdk_bdev_io *bdev_io; + + if (ch->per_thread_cache_count > 0) { + bdev_io = STAILQ_FIRST(&ch->per_thread_cache); + STAILQ_REMOVE_HEAD(&ch->per_thread_cache, internal.buf_link); + ch->per_thread_cache_count--; + } else if (spdk_unlikely(!TAILQ_EMPTY(&ch->io_wait_queue))) { + /* + * Don't try to look for bdev_ios in the global pool if there are + * waiters on bdev_ios - we don't want this caller to jump the line. + */ + bdev_io = NULL; + } else { + bdev_io = spdk_mempool_get(g_bdev_mgr.bdev_io_pool); + } + + return bdev_io; +} + +void +spdk_bdev_free_io(struct spdk_bdev_io *bdev_io) +{ + struct spdk_bdev_mgmt_channel *ch = bdev_io->internal.ch->shared_resource->mgmt_ch; + + assert(bdev_io != NULL); + assert(bdev_io->internal.status != SPDK_BDEV_IO_STATUS_PENDING); + + if (bdev_io->internal.buf != NULL) { + spdk_bdev_io_put_buf(bdev_io); + } + + if (ch->per_thread_cache_count < ch->bdev_io_cache_size) { + ch->per_thread_cache_count++; + STAILQ_INSERT_TAIL(&ch->per_thread_cache, bdev_io, internal.buf_link); + while (ch->per_thread_cache_count > 0 && !TAILQ_EMPTY(&ch->io_wait_queue)) { + struct spdk_bdev_io_wait_entry *entry; + + entry = TAILQ_FIRST(&ch->io_wait_queue); + TAILQ_REMOVE(&ch->io_wait_queue, entry, link); + entry->cb_fn(entry->cb_arg); + } + } else { + /* We should never have a full cache with entries on the io wait queue. */ + assert(TAILQ_EMPTY(&ch->io_wait_queue)); + spdk_mempool_put(g_bdev_mgr.bdev_io_pool, (void *)bdev_io); + } +} + +static bool +_spdk_bdev_qos_is_iops_rate_limit(enum spdk_bdev_qos_rate_limit_type limit) +{ + assert(limit != SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES); + + switch (limit) { + case SPDK_BDEV_QOS_RW_IOPS_RATE_LIMIT: + return true; + case SPDK_BDEV_QOS_RW_BPS_RATE_LIMIT: + return false; + case SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES: + default: + return false; + } +} + +static bool +_spdk_bdev_qos_io_to_limit(struct spdk_bdev_io *bdev_io) +{ + switch (bdev_io->type) { + case SPDK_BDEV_IO_TYPE_NVME_IO: + case SPDK_BDEV_IO_TYPE_NVME_IO_MD: + case SPDK_BDEV_IO_TYPE_READ: + case SPDK_BDEV_IO_TYPE_WRITE: + case SPDK_BDEV_IO_TYPE_UNMAP: + case SPDK_BDEV_IO_TYPE_WRITE_ZEROES: + return true; + default: + return false; + } +} + +static uint64_t +_spdk_bdev_get_io_size_in_byte(struct spdk_bdev_io *bdev_io) +{ + struct spdk_bdev *bdev = bdev_io->bdev; + + switch (bdev_io->type) { + case SPDK_BDEV_IO_TYPE_NVME_IO: + case SPDK_BDEV_IO_TYPE_NVME_IO_MD: + return bdev_io->u.nvme_passthru.nbytes; + case SPDK_BDEV_IO_TYPE_READ: + case SPDK_BDEV_IO_TYPE_WRITE: + case SPDK_BDEV_IO_TYPE_UNMAP: + case SPDK_BDEV_IO_TYPE_WRITE_ZEROES: + return bdev_io->u.bdev.num_blocks * bdev->blocklen; + default: + return 0; + } +} + +static void +_spdk_bdev_qos_update_per_io(struct spdk_bdev_qos *qos, uint64_t io_size_in_byte) +{ + int i; + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (qos->rate_limits[i].limit == SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + continue; + } + + switch (i) { + case SPDK_BDEV_QOS_RW_IOPS_RATE_LIMIT: + qos->rate_limits[i].remaining_this_timeslice--; + break; + case SPDK_BDEV_QOS_RW_BPS_RATE_LIMIT: + qos->rate_limits[i].remaining_this_timeslice -= io_size_in_byte; + break; + case SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES: + default: + break; + } + } +} + +static void +_spdk_bdev_qos_io_submit(struct spdk_bdev_channel *ch, struct spdk_bdev_qos *qos) +{ + struct spdk_bdev_io *bdev_io = NULL; + struct spdk_bdev *bdev = ch->bdev; + struct spdk_bdev_shared_resource *shared_resource = ch->shared_resource; + int i; + bool to_limit_io; + uint64_t io_size_in_byte; + + while (!TAILQ_EMPTY(&qos->queued)) { + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (qos->rate_limits[i].max_per_timeslice > 0 && + (qos->rate_limits[i].remaining_this_timeslice <= 0)) { + return; + } + } + + bdev_io = TAILQ_FIRST(&qos->queued); + TAILQ_REMOVE(&qos->queued, bdev_io, internal.link); + ch->io_outstanding++; + shared_resource->io_outstanding++; + to_limit_io = _spdk_bdev_qos_io_to_limit(bdev_io); + if (to_limit_io == true) { + io_size_in_byte = _spdk_bdev_get_io_size_in_byte(bdev_io); + _spdk_bdev_qos_update_per_io(qos, io_size_in_byte); + } + bdev->fn_table->submit_request(ch->channel, bdev_io); + } +} + +static void +_spdk_bdev_queue_io_wait_with_cb(struct spdk_bdev_io *bdev_io, spdk_bdev_io_wait_cb cb_fn) +{ + int rc; + + bdev_io->internal.waitq_entry.bdev = bdev_io->bdev; + bdev_io->internal.waitq_entry.cb_fn = cb_fn; + bdev_io->internal.waitq_entry.cb_arg = bdev_io; + rc = spdk_bdev_queue_io_wait(bdev_io->bdev, spdk_io_channel_from_ctx(bdev_io->internal.ch), + &bdev_io->internal.waitq_entry); + if (rc != 0) { + SPDK_ERRLOG("Queue IO failed, rc=%d\n", rc); + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_FAILED; + bdev_io->internal.cb(bdev_io, false, bdev_io->internal.caller_ctx); + } +} + +static bool +_spdk_bdev_io_type_can_split(uint8_t type) +{ + assert(type != SPDK_BDEV_IO_TYPE_INVALID); + assert(type < SPDK_BDEV_NUM_IO_TYPES); + + /* Only split READ and WRITE I/O. Theoretically other types of I/O like + * UNMAP could be split, but these types of I/O are typically much larger + * in size (sometimes the size of the entire block device), and the bdev + * module can more efficiently split these types of I/O. Plus those types + * of I/O do not have a payload, which makes the splitting process simpler. + */ + if (type == SPDK_BDEV_IO_TYPE_READ || type == SPDK_BDEV_IO_TYPE_WRITE) { + return true; + } else { + return false; + } +} + +static bool +_spdk_bdev_io_should_split(struct spdk_bdev_io *bdev_io) +{ + uint64_t start_stripe, end_stripe; + uint32_t io_boundary = bdev_io->bdev->optimal_io_boundary; + + if (io_boundary == 0) { + return false; + } + + if (!_spdk_bdev_io_type_can_split(bdev_io->type)) { + return false; + } + + start_stripe = bdev_io->u.bdev.offset_blocks; + end_stripe = start_stripe + bdev_io->u.bdev.num_blocks - 1; + /* Avoid expensive div operations if possible. These spdk_u32 functions are very cheap. */ + if (spdk_likely(spdk_u32_is_pow2(io_boundary))) { + start_stripe >>= spdk_u32log2(io_boundary); + end_stripe >>= spdk_u32log2(io_boundary); + } else { + start_stripe /= io_boundary; + end_stripe /= io_boundary; + } + return (start_stripe != end_stripe); +} + +static uint32_t +_to_next_boundary(uint64_t offset, uint32_t boundary) +{ + return (boundary - (offset % boundary)); +} + +static void +_spdk_bdev_io_split_done(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg); + +static void +_spdk_bdev_io_split_with_payload(void *_bdev_io) +{ + struct spdk_bdev_io *bdev_io = _bdev_io; + uint64_t current_offset, remaining; + uint32_t blocklen, to_next_boundary, to_next_boundary_bytes; + struct iovec *parent_iov, *iov; + uint64_t parent_iov_offset, iov_len; + uint32_t parent_iovpos, parent_iovcnt, child_iovcnt, iovcnt; + int rc; + + remaining = bdev_io->u.bdev.split_remaining_num_blocks; + current_offset = bdev_io->u.bdev.split_current_offset_blocks; + blocklen = bdev_io->bdev->blocklen; + parent_iov_offset = (current_offset - bdev_io->u.bdev.offset_blocks) * blocklen; + parent_iovcnt = bdev_io->u.bdev.iovcnt; + + for (parent_iovpos = 0; parent_iovpos < parent_iovcnt; parent_iovpos++) { + parent_iov = &bdev_io->u.bdev.iovs[parent_iovpos]; + if (parent_iov_offset < parent_iov->iov_len) { + break; + } + parent_iov_offset -= parent_iov->iov_len; + } + + child_iovcnt = 0; + while (remaining > 0 && parent_iovpos < parent_iovcnt && child_iovcnt < BDEV_IO_NUM_CHILD_IOV) { + to_next_boundary = _to_next_boundary(current_offset, bdev_io->bdev->optimal_io_boundary); + to_next_boundary = spdk_min(remaining, to_next_boundary); + to_next_boundary_bytes = to_next_boundary * blocklen; + iov = &bdev_io->child_iov[child_iovcnt]; + iovcnt = 0; + while (to_next_boundary_bytes > 0 && parent_iovpos < parent_iovcnt && + child_iovcnt < BDEV_IO_NUM_CHILD_IOV) { + parent_iov = &bdev_io->u.bdev.iovs[parent_iovpos]; + iov_len = spdk_min(to_next_boundary_bytes, parent_iov->iov_len - parent_iov_offset); + to_next_boundary_bytes -= iov_len; + + bdev_io->child_iov[child_iovcnt].iov_base = parent_iov->iov_base + parent_iov_offset; + bdev_io->child_iov[child_iovcnt].iov_len = iov_len; + + if (iov_len < parent_iov->iov_len - parent_iov_offset) { + parent_iov_offset += iov_len; + } else { + parent_iovpos++; + parent_iov_offset = 0; + } + child_iovcnt++; + iovcnt++; + } + + if (to_next_boundary_bytes > 0) { + /* We had to stop this child I/O early because we ran out of + * child_iov space. Make sure the iovs collected are valid and + * then adjust to_next_boundary before starting the child I/O. + */ + if ((to_next_boundary_bytes % blocklen) != 0) { + SPDK_ERRLOG("Remaining %" PRIu32 " is not multiple of block size %" PRIu32 "\n", + to_next_boundary_bytes, blocklen); + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_FAILED; + if (bdev_io->u.bdev.split_outstanding == 0) { + bdev_io->internal.cb(bdev_io, false, bdev_io->internal.caller_ctx); + } + return; + } + to_next_boundary -= to_next_boundary_bytes / blocklen; + } + + bdev_io->u.bdev.split_outstanding++; + + if (bdev_io->type == SPDK_BDEV_IO_TYPE_READ) { + rc = spdk_bdev_readv_blocks(bdev_io->internal.desc, + spdk_io_channel_from_ctx(bdev_io->internal.ch), + iov, iovcnt, current_offset, to_next_boundary, + _spdk_bdev_io_split_done, bdev_io); + } else { + rc = spdk_bdev_writev_blocks(bdev_io->internal.desc, + spdk_io_channel_from_ctx(bdev_io->internal.ch), + iov, iovcnt, current_offset, to_next_boundary, + _spdk_bdev_io_split_done, bdev_io); + } + + if (rc == 0) { + current_offset += to_next_boundary; + remaining -= to_next_boundary; + bdev_io->u.bdev.split_current_offset_blocks = current_offset; + bdev_io->u.bdev.split_remaining_num_blocks = remaining; + } else { + bdev_io->u.bdev.split_outstanding--; + if (rc == -ENOMEM) { + if (bdev_io->u.bdev.split_outstanding == 0) { + /* No I/O is outstanding. Hence we should wait here. */ + _spdk_bdev_queue_io_wait_with_cb(bdev_io, + _spdk_bdev_io_split_with_payload); + } + } else { + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_FAILED; + if (bdev_io->u.bdev.split_outstanding == 0) { + bdev_io->internal.cb(bdev_io, false, bdev_io->internal.caller_ctx); + } + } + + return; + } + } +} + +static void +_spdk_bdev_io_split_done(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg) +{ + struct spdk_bdev_io *parent_io = cb_arg; + + spdk_bdev_free_io(bdev_io); + + if (!success) { + parent_io->internal.status = SPDK_BDEV_IO_STATUS_FAILED; + } + parent_io->u.bdev.split_outstanding--; + if (parent_io->u.bdev.split_outstanding != 0) { + return; + } + + /* + * Parent I/O finishes when all blocks are consumed or there is any failure of + * child I/O and no outstanding child I/O. + */ + if (parent_io->u.bdev.split_remaining_num_blocks == 0 || + parent_io->internal.status != SPDK_BDEV_IO_STATUS_SUCCESS) { + parent_io->internal.cb(parent_io, parent_io->internal.status == SPDK_BDEV_IO_STATUS_SUCCESS, + parent_io->internal.caller_ctx); + return; + } + + /* + * Continue with the splitting process. This function will complete the parent I/O if the + * splitting is done. + */ + _spdk_bdev_io_split_with_payload(parent_io); +} + +static void +_spdk_bdev_io_split(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io) +{ + assert(_spdk_bdev_io_type_can_split(bdev_io->type)); + + bdev_io->u.bdev.split_current_offset_blocks = bdev_io->u.bdev.offset_blocks; + bdev_io->u.bdev.split_remaining_num_blocks = bdev_io->u.bdev.num_blocks; + bdev_io->u.bdev.split_outstanding = 0; + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_SUCCESS; + + _spdk_bdev_io_split_with_payload(bdev_io); +} + +static void +_spdk_bdev_io_submit(void *ctx) +{ + struct spdk_bdev_io *bdev_io = ctx; + struct spdk_bdev *bdev = bdev_io->bdev; + struct spdk_bdev_channel *bdev_ch = bdev_io->internal.ch; + struct spdk_io_channel *ch = bdev_ch->channel; + struct spdk_bdev_shared_resource *shared_resource = bdev_ch->shared_resource; + uint64_t tsc; + + tsc = spdk_get_ticks(); + bdev_io->internal.submit_tsc = tsc; + spdk_trace_record_tsc(tsc, TRACE_BDEV_IO_START, 0, 0, (uintptr_t)bdev_io, bdev_io->type); + bdev_ch->io_outstanding++; + shared_resource->io_outstanding++; + bdev_io->internal.in_submit_request = true; + if (spdk_likely(bdev_ch->flags == 0)) { + if (spdk_likely(TAILQ_EMPTY(&shared_resource->nomem_io))) { + bdev->fn_table->submit_request(ch, bdev_io); + } else { + bdev_ch->io_outstanding--; + shared_resource->io_outstanding--; + TAILQ_INSERT_TAIL(&shared_resource->nomem_io, bdev_io, internal.link); + } + } else if (bdev_ch->flags & BDEV_CH_RESET_IN_PROGRESS) { + spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED); + } else if (bdev_ch->flags & BDEV_CH_QOS_ENABLED) { + bdev_ch->io_outstanding--; + shared_resource->io_outstanding--; + TAILQ_INSERT_TAIL(&bdev->internal.qos->queued, bdev_io, internal.link); + _spdk_bdev_qos_io_submit(bdev_ch, bdev->internal.qos); + } else { + SPDK_ERRLOG("unknown bdev_ch flag %x found\n", bdev_ch->flags); + spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED); + } + bdev_io->internal.in_submit_request = false; +} + +static void +spdk_bdev_io_submit(struct spdk_bdev_io *bdev_io) +{ + struct spdk_bdev *bdev = bdev_io->bdev; + struct spdk_thread *thread = spdk_io_channel_get_thread(bdev_io->internal.ch->channel); + + assert(thread != NULL); + assert(bdev_io->internal.status == SPDK_BDEV_IO_STATUS_PENDING); + + if (bdev->split_on_optimal_io_boundary && _spdk_bdev_io_should_split(bdev_io)) { + if (bdev_io->type == SPDK_BDEV_IO_TYPE_READ) { + spdk_bdev_io_get_buf(bdev_io, _spdk_bdev_io_split, + bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen); + } else { + _spdk_bdev_io_split(NULL, bdev_io); + } + return; + } + + if (bdev_io->internal.ch->flags & BDEV_CH_QOS_ENABLED) { + if ((thread == bdev->internal.qos->thread) || !bdev->internal.qos->thread) { + _spdk_bdev_io_submit(bdev_io); + } else { + bdev_io->internal.io_submit_ch = bdev_io->internal.ch; + bdev_io->internal.ch = bdev->internal.qos->ch; + spdk_thread_send_msg(bdev->internal.qos->thread, _spdk_bdev_io_submit, bdev_io); + } + } else { + _spdk_bdev_io_submit(bdev_io); + } +} + +static void +spdk_bdev_io_submit_reset(struct spdk_bdev_io *bdev_io) +{ + struct spdk_bdev *bdev = bdev_io->bdev; + struct spdk_bdev_channel *bdev_ch = bdev_io->internal.ch; + struct spdk_io_channel *ch = bdev_ch->channel; + + assert(bdev_io->internal.status == SPDK_BDEV_IO_STATUS_PENDING); + + bdev_io->internal.in_submit_request = true; + bdev->fn_table->submit_request(ch, bdev_io); + bdev_io->internal.in_submit_request = false; +} + +static void +spdk_bdev_io_init(struct spdk_bdev_io *bdev_io, + struct spdk_bdev *bdev, void *cb_arg, + spdk_bdev_io_completion_cb cb) +{ + bdev_io->bdev = bdev; + bdev_io->internal.caller_ctx = cb_arg; + bdev_io->internal.cb = cb; + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_PENDING; + bdev_io->internal.in_submit_request = false; + bdev_io->internal.buf = NULL; + bdev_io->internal.io_submit_ch = NULL; +} + +static bool +_spdk_bdev_io_type_supported(struct spdk_bdev *bdev, enum spdk_bdev_io_type io_type) +{ + return bdev->fn_table->io_type_supported(bdev->ctxt, io_type); +} + +bool +spdk_bdev_io_type_supported(struct spdk_bdev *bdev, enum spdk_bdev_io_type io_type) +{ + bool supported; + + supported = _spdk_bdev_io_type_supported(bdev, io_type); + + if (!supported) { + switch (io_type) { + case SPDK_BDEV_IO_TYPE_WRITE_ZEROES: + /* The bdev layer will emulate write zeroes as long as write is supported. */ + supported = _spdk_bdev_io_type_supported(bdev, SPDK_BDEV_IO_TYPE_WRITE); + break; + default: + break; + } + } + + return supported; +} + +int +spdk_bdev_dump_info_json(struct spdk_bdev *bdev, struct spdk_json_write_ctx *w) +{ + if (bdev->fn_table->dump_info_json) { + return bdev->fn_table->dump_info_json(bdev->ctxt, w); + } + + return 0; +} + +static void +spdk_bdev_qos_update_max_quota_per_timeslice(struct spdk_bdev_qos *qos) +{ + uint32_t max_per_timeslice = 0; + int i; + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (qos->rate_limits[i].limit == SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + qos->rate_limits[i].max_per_timeslice = 0; + continue; + } + + max_per_timeslice = qos->rate_limits[i].limit * + SPDK_BDEV_QOS_TIMESLICE_IN_USEC / SPDK_SEC_TO_USEC; + + qos->rate_limits[i].max_per_timeslice = spdk_max(max_per_timeslice, + qos->rate_limits[i].min_per_timeslice); + + qos->rate_limits[i].remaining_this_timeslice = qos->rate_limits[i].max_per_timeslice; + } +} + +static int +spdk_bdev_channel_poll_qos(void *arg) +{ + struct spdk_bdev_qos *qos = arg; + uint64_t now = spdk_get_ticks(); + int i; + + if (now < (qos->last_timeslice + qos->timeslice_size)) { + /* We received our callback earlier than expected - return + * immediately and wait to do accounting until at least one + * timeslice has actually expired. This should never happen + * with a well-behaved timer implementation. + */ + return 0; + } + + /* Reset for next round of rate limiting */ + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + /* We may have allowed the IOs or bytes to slightly overrun in the last + * timeslice. remaining_this_timeslice is signed, so if it's negative + * here, we'll account for the overrun so that the next timeslice will + * be appropriately reduced. + */ + if (qos->rate_limits[i].remaining_this_timeslice > 0) { + qos->rate_limits[i].remaining_this_timeslice = 0; + } + } + + while (now >= (qos->last_timeslice + qos->timeslice_size)) { + qos->last_timeslice += qos->timeslice_size; + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + qos->rate_limits[i].remaining_this_timeslice += + qos->rate_limits[i].max_per_timeslice; + } + } + + _spdk_bdev_qos_io_submit(qos->ch, qos); + + return -1; +} + +static void +_spdk_bdev_channel_destroy_resource(struct spdk_bdev_channel *ch) +{ + struct spdk_bdev_shared_resource *shared_resource; + + if (!ch) { + return; + } + + if (ch->channel) { + spdk_put_io_channel(ch->channel); + } + + assert(ch->io_outstanding == 0); + + shared_resource = ch->shared_resource; + if (shared_resource) { + assert(ch->io_outstanding == 0); + assert(shared_resource->ref > 0); + shared_resource->ref--; + if (shared_resource->ref == 0) { + assert(shared_resource->io_outstanding == 0); + TAILQ_REMOVE(&shared_resource->mgmt_ch->shared_resources, shared_resource, link); + spdk_put_io_channel(spdk_io_channel_from_ctx(shared_resource->mgmt_ch)); + free(shared_resource); + } + } +} + +/* Caller must hold bdev->internal.mutex. */ +static void +_spdk_bdev_enable_qos(struct spdk_bdev *bdev, struct spdk_bdev_channel *ch) +{ + struct spdk_bdev_qos *qos = bdev->internal.qos; + int i; + + /* Rate limiting on this bdev enabled */ + if (qos) { + if (qos->ch == NULL) { + struct spdk_io_channel *io_ch; + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Selecting channel %p as QoS channel for bdev %s on thread %p\n", ch, + bdev->name, spdk_get_thread()); + + /* No qos channel has been selected, so set one up */ + + /* Take another reference to ch */ + io_ch = spdk_get_io_channel(__bdev_to_io_dev(bdev)); + qos->ch = ch; + + qos->thread = spdk_io_channel_get_thread(io_ch); + + TAILQ_INIT(&qos->queued); + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (_spdk_bdev_qos_is_iops_rate_limit(i) == true) { + qos->rate_limits[i].min_per_timeslice = + SPDK_BDEV_QOS_MIN_IO_PER_TIMESLICE; + } else { + qos->rate_limits[i].min_per_timeslice = + SPDK_BDEV_QOS_MIN_BYTE_PER_TIMESLICE; + } + + if (qos->rate_limits[i].limit == 0) { + qos->rate_limits[i].limit = SPDK_BDEV_QOS_LIMIT_NOT_DEFINED; + } + } + spdk_bdev_qos_update_max_quota_per_timeslice(qos); + qos->timeslice_size = + SPDK_BDEV_QOS_TIMESLICE_IN_USEC * spdk_get_ticks_hz() / SPDK_SEC_TO_USEC; + qos->last_timeslice = spdk_get_ticks(); + qos->poller = spdk_poller_register(spdk_bdev_channel_poll_qos, + qos, + SPDK_BDEV_QOS_TIMESLICE_IN_USEC); + } + + ch->flags |= BDEV_CH_QOS_ENABLED; + } +} + +static int +spdk_bdev_channel_create(void *io_device, void *ctx_buf) +{ + struct spdk_bdev *bdev = __bdev_from_io_dev(io_device); + struct spdk_bdev_channel *ch = ctx_buf; + struct spdk_io_channel *mgmt_io_ch; + struct spdk_bdev_mgmt_channel *mgmt_ch; + struct spdk_bdev_shared_resource *shared_resource; + + ch->bdev = bdev; + ch->channel = bdev->fn_table->get_io_channel(bdev->ctxt); + if (!ch->channel) { + return -1; + } + + mgmt_io_ch = spdk_get_io_channel(&g_bdev_mgr); + if (!mgmt_io_ch) { + return -1; + } + + mgmt_ch = spdk_io_channel_get_ctx(mgmt_io_ch); + TAILQ_FOREACH(shared_resource, &mgmt_ch->shared_resources, link) { + if (shared_resource->shared_ch == ch->channel) { + spdk_put_io_channel(mgmt_io_ch); + shared_resource->ref++; + break; + } + } + + if (shared_resource == NULL) { + shared_resource = calloc(1, sizeof(*shared_resource)); + if (shared_resource == NULL) { + spdk_put_io_channel(mgmt_io_ch); + return -1; + } + + shared_resource->mgmt_ch = mgmt_ch; + shared_resource->io_outstanding = 0; + TAILQ_INIT(&shared_resource->nomem_io); + shared_resource->nomem_threshold = 0; + shared_resource->shared_ch = ch->channel; + shared_resource->ref = 1; + TAILQ_INSERT_TAIL(&mgmt_ch->shared_resources, shared_resource, link); + } + + memset(&ch->stat, 0, sizeof(ch->stat)); + ch->stat.ticks_rate = spdk_get_ticks_hz(); + ch->io_outstanding = 0; + TAILQ_INIT(&ch->queued_resets); + ch->flags = 0; + ch->shared_resource = shared_resource; + +#ifdef SPDK_CONFIG_VTUNE + { + char *name; + __itt_init_ittlib(NULL, 0); + name = spdk_sprintf_alloc("spdk_bdev_%s_%p", ch->bdev->name, ch); + if (!name) { + _spdk_bdev_channel_destroy_resource(ch); + return -1; + } + ch->handle = __itt_string_handle_create(name); + free(name); + ch->start_tsc = spdk_get_ticks(); + ch->interval_tsc = spdk_get_ticks_hz() / 100; + memset(&ch->prev_stat, 0, sizeof(ch->prev_stat)); + } +#endif + + pthread_mutex_lock(&bdev->internal.mutex); + _spdk_bdev_enable_qos(bdev, ch); + pthread_mutex_unlock(&bdev->internal.mutex); + + return 0; +} + +/* + * Abort I/O that are waiting on a data buffer. These types of I/O are + * linked using the spdk_bdev_io internal.buf_link TAILQ_ENTRY. + */ +static void +_spdk_bdev_abort_buf_io(bdev_io_stailq_t *queue, struct spdk_bdev_channel *ch) +{ + bdev_io_stailq_t tmp; + struct spdk_bdev_io *bdev_io; + + STAILQ_INIT(&tmp); + + while (!STAILQ_EMPTY(queue)) { + bdev_io = STAILQ_FIRST(queue); + STAILQ_REMOVE_HEAD(queue, internal.buf_link); + if (bdev_io->internal.ch == ch) { + spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED); + } else { + STAILQ_INSERT_TAIL(&tmp, bdev_io, internal.buf_link); + } + } + + STAILQ_SWAP(&tmp, queue, spdk_bdev_io); +} + +/* + * Abort I/O that are queued waiting for submission. These types of I/O are + * linked using the spdk_bdev_io link TAILQ_ENTRY. + */ +static void +_spdk_bdev_abort_queued_io(bdev_io_tailq_t *queue, struct spdk_bdev_channel *ch) +{ + struct spdk_bdev_io *bdev_io, *tmp; + + TAILQ_FOREACH_SAFE(bdev_io, queue, internal.link, tmp) { + if (bdev_io->internal.ch == ch) { + TAILQ_REMOVE(queue, bdev_io, internal.link); + /* + * spdk_bdev_io_complete() assumes that the completed I/O had + * been submitted to the bdev module. Since in this case it + * hadn't, bump io_outstanding to account for the decrement + * that spdk_bdev_io_complete() will do. + */ + if (bdev_io->type != SPDK_BDEV_IO_TYPE_RESET) { + ch->io_outstanding++; + ch->shared_resource->io_outstanding++; + } + spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED); + } + } +} + +static void +spdk_bdev_qos_channel_destroy(void *cb_arg) +{ + struct spdk_bdev_qos *qos = cb_arg; + + spdk_put_io_channel(spdk_io_channel_from_ctx(qos->ch)); + spdk_poller_unregister(&qos->poller); + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Free QoS %p.\n", qos); + + free(qos); +} + +static int +spdk_bdev_qos_destroy(struct spdk_bdev *bdev) +{ + int i; + + /* + * Cleanly shutting down the QoS poller is tricky, because + * during the asynchronous operation the user could open + * a new descriptor and create a new channel, spawning + * a new QoS poller. + * + * The strategy is to create a new QoS structure here and swap it + * in. The shutdown path then continues to refer to the old one + * until it completes and then releases it. + */ + struct spdk_bdev_qos *new_qos, *old_qos; + + old_qos = bdev->internal.qos; + + new_qos = calloc(1, sizeof(*new_qos)); + if (!new_qos) { + SPDK_ERRLOG("Unable to allocate memory to shut down QoS.\n"); + return -ENOMEM; + } + + /* Copy the old QoS data into the newly allocated structure */ + memcpy(new_qos, old_qos, sizeof(*new_qos)); + + /* Zero out the key parts of the QoS structure */ + new_qos->ch = NULL; + new_qos->thread = NULL; + new_qos->poller = NULL; + TAILQ_INIT(&new_qos->queued); + /* + * The limit member of spdk_bdev_qos_limit structure is not zeroed. + * It will be used later for the new QoS structure. + */ + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + new_qos->rate_limits[i].remaining_this_timeslice = 0; + new_qos->rate_limits[i].min_per_timeslice = 0; + new_qos->rate_limits[i].max_per_timeslice = 0; + } + + bdev->internal.qos = new_qos; + + if (old_qos->thread == NULL) { + free(old_qos); + } else { + spdk_thread_send_msg(old_qos->thread, spdk_bdev_qos_channel_destroy, + old_qos); + } + + /* It is safe to continue with destroying the bdev even though the QoS channel hasn't + * been destroyed yet. The destruction path will end up waiting for the final + * channel to be put before it releases resources. */ + + return 0; +} + +static void +_spdk_bdev_io_stat_add(struct spdk_bdev_io_stat *total, struct spdk_bdev_io_stat *add) +{ + total->bytes_read += add->bytes_read; + total->num_read_ops += add->num_read_ops; + total->bytes_written += add->bytes_written; + total->num_write_ops += add->num_write_ops; + total->read_latency_ticks += add->read_latency_ticks; + total->write_latency_ticks += add->write_latency_ticks; +} + +static void +spdk_bdev_channel_destroy(void *io_device, void *ctx_buf) +{ + struct spdk_bdev_channel *ch = ctx_buf; + struct spdk_bdev_mgmt_channel *mgmt_ch; + struct spdk_bdev_shared_resource *shared_resource = ch->shared_resource; + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Destroying channel %p for bdev %s on thread %p\n", ch, ch->bdev->name, + spdk_get_thread()); + + /* This channel is going away, so add its statistics into the bdev so that they don't get lost. */ + pthread_mutex_lock(&ch->bdev->internal.mutex); + _spdk_bdev_io_stat_add(&ch->bdev->internal.stat, &ch->stat); + pthread_mutex_unlock(&ch->bdev->internal.mutex); + + mgmt_ch = shared_resource->mgmt_ch; + + _spdk_bdev_abort_queued_io(&ch->queued_resets, ch); + _spdk_bdev_abort_queued_io(&shared_resource->nomem_io, ch); + _spdk_bdev_abort_buf_io(&mgmt_ch->need_buf_small, ch); + _spdk_bdev_abort_buf_io(&mgmt_ch->need_buf_large, ch); + + _spdk_bdev_channel_destroy_resource(ch); +} + +int +spdk_bdev_alias_add(struct spdk_bdev *bdev, const char *alias) +{ + struct spdk_bdev_alias *tmp; + + if (alias == NULL) { + SPDK_ERRLOG("Empty alias passed\n"); + return -EINVAL; + } + + if (spdk_bdev_get_by_name(alias)) { + SPDK_ERRLOG("Bdev name/alias: %s already exists\n", alias); + return -EEXIST; + } + + tmp = calloc(1, sizeof(*tmp)); + if (tmp == NULL) { + SPDK_ERRLOG("Unable to allocate alias\n"); + return -ENOMEM; + } + + tmp->alias = strdup(alias); + if (tmp->alias == NULL) { + free(tmp); + SPDK_ERRLOG("Unable to allocate alias\n"); + return -ENOMEM; + } + + TAILQ_INSERT_TAIL(&bdev->aliases, tmp, tailq); + + return 0; +} + +int +spdk_bdev_alias_del(struct spdk_bdev *bdev, const char *alias) +{ + struct spdk_bdev_alias *tmp; + + TAILQ_FOREACH(tmp, &bdev->aliases, tailq) { + if (strcmp(alias, tmp->alias) == 0) { + TAILQ_REMOVE(&bdev->aliases, tmp, tailq); + free(tmp->alias); + free(tmp); + return 0; + } + } + + SPDK_INFOLOG(SPDK_LOG_BDEV, "Alias %s does not exists\n", alias); + + return -ENOENT; +} + +void +spdk_bdev_alias_del_all(struct spdk_bdev *bdev) +{ + struct spdk_bdev_alias *p, *tmp; + + TAILQ_FOREACH_SAFE(p, &bdev->aliases, tailq, tmp) { + TAILQ_REMOVE(&bdev->aliases, p, tailq); + free(p->alias); + free(p); + } +} + +struct spdk_io_channel * +spdk_bdev_get_io_channel(struct spdk_bdev_desc *desc) +{ + return spdk_get_io_channel(__bdev_to_io_dev(desc->bdev)); +} + +const char * +spdk_bdev_get_name(const struct spdk_bdev *bdev) +{ + return bdev->name; +} + +const char * +spdk_bdev_get_product_name(const struct spdk_bdev *bdev) +{ + return bdev->product_name; +} + +const struct spdk_bdev_aliases_list * +spdk_bdev_get_aliases(const struct spdk_bdev *bdev) +{ + return &bdev->aliases; +} + +uint32_t +spdk_bdev_get_block_size(const struct spdk_bdev *bdev) +{ + return bdev->blocklen; +} + +uint64_t +spdk_bdev_get_num_blocks(const struct spdk_bdev *bdev) +{ + return bdev->blockcnt; +} + +const char * +spdk_bdev_get_qos_rpc_type(enum spdk_bdev_qos_rate_limit_type type) +{ + return qos_rpc_type[type]; +} + +void +spdk_bdev_get_qos_rate_limits(struct spdk_bdev *bdev, uint64_t *limits) +{ + int i; + + memset(limits, 0, sizeof(*limits) * SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES); + + pthread_mutex_lock(&bdev->internal.mutex); + if (bdev->internal.qos) { + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (bdev->internal.qos->rate_limits[i].limit != + SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + limits[i] = bdev->internal.qos->rate_limits[i].limit; + if (_spdk_bdev_qos_is_iops_rate_limit(i) == false) { + /* Change from Byte to Megabyte which is user visible. */ + limits[i] = limits[i] / 1024 / 1024; + } + } + } + } + pthread_mutex_unlock(&bdev->internal.mutex); +} + +size_t +spdk_bdev_get_buf_align(const struct spdk_bdev *bdev) +{ + /* TODO: push this logic down to the bdev modules */ + if (bdev->need_aligned_buffer) { + return bdev->blocklen; + } + + return 1; +} + +uint32_t +spdk_bdev_get_optimal_io_boundary(const struct spdk_bdev *bdev) +{ + return bdev->optimal_io_boundary; +} + +bool +spdk_bdev_has_write_cache(const struct spdk_bdev *bdev) +{ + return bdev->write_cache; +} + +const struct spdk_uuid * +spdk_bdev_get_uuid(const struct spdk_bdev *bdev) +{ + return &bdev->uuid; +} + +uint64_t +spdk_bdev_get_qd(const struct spdk_bdev *bdev) +{ + return bdev->internal.measured_queue_depth; +} + +uint64_t +spdk_bdev_get_qd_sampling_period(const struct spdk_bdev *bdev) +{ + return bdev->internal.period; +} + +uint64_t +spdk_bdev_get_weighted_io_time(const struct spdk_bdev *bdev) +{ + return bdev->internal.weighted_io_time; +} + +uint64_t +spdk_bdev_get_io_time(const struct spdk_bdev *bdev) +{ + return bdev->internal.io_time; +} + +static void +_calculate_measured_qd_cpl(struct spdk_io_channel_iter *i, int status) +{ + struct spdk_bdev *bdev = spdk_io_channel_iter_get_ctx(i); + + bdev->internal.measured_queue_depth = bdev->internal.temporary_queue_depth; + + if (bdev->internal.measured_queue_depth) { + bdev->internal.io_time += bdev->internal.period; + bdev->internal.weighted_io_time += bdev->internal.period * bdev->internal.measured_queue_depth; + } +} + +static void +_calculate_measured_qd(struct spdk_io_channel_iter *i) +{ + struct spdk_bdev *bdev = spdk_io_channel_iter_get_ctx(i); + struct spdk_io_channel *io_ch = spdk_io_channel_iter_get_channel(i); + struct spdk_bdev_channel *ch = spdk_io_channel_get_ctx(io_ch); + + bdev->internal.temporary_queue_depth += ch->io_outstanding; + spdk_for_each_channel_continue(i, 0); +} + +static int +spdk_bdev_calculate_measured_queue_depth(void *ctx) +{ + struct spdk_bdev *bdev = ctx; + bdev->internal.temporary_queue_depth = 0; + spdk_for_each_channel(__bdev_to_io_dev(bdev), _calculate_measured_qd, bdev, + _calculate_measured_qd_cpl); + return 0; +} + +void +spdk_bdev_set_qd_sampling_period(struct spdk_bdev *bdev, uint64_t period) +{ + bdev->internal.period = period; + + if (bdev->internal.qd_poller != NULL) { + spdk_poller_unregister(&bdev->internal.qd_poller); + bdev->internal.measured_queue_depth = UINT64_MAX; + } + + if (period != 0) { + bdev->internal.qd_poller = spdk_poller_register(spdk_bdev_calculate_measured_queue_depth, bdev, + period); + } +} + +int +spdk_bdev_notify_blockcnt_change(struct spdk_bdev *bdev, uint64_t size) +{ + int ret; + + pthread_mutex_lock(&bdev->internal.mutex); + + /* bdev has open descriptors */ + if (!TAILQ_EMPTY(&bdev->internal.open_descs) && + bdev->blockcnt > size) { + ret = -EBUSY; + } else { + bdev->blockcnt = size; + ret = 0; + } + + pthread_mutex_unlock(&bdev->internal.mutex); + + return ret; +} + +/* + * Convert I/O offset and length from bytes to blocks. + * + * Returns zero on success or non-zero if the byte parameters aren't divisible by the block size. + */ +static uint64_t +spdk_bdev_bytes_to_blocks(struct spdk_bdev *bdev, uint64_t offset_bytes, uint64_t *offset_blocks, + uint64_t num_bytes, uint64_t *num_blocks) +{ + uint32_t block_size = bdev->blocklen; + + *offset_blocks = offset_bytes / block_size; + *num_blocks = num_bytes / block_size; + + return (offset_bytes % block_size) | (num_bytes % block_size); +} + +static bool +spdk_bdev_io_valid_blocks(struct spdk_bdev *bdev, uint64_t offset_blocks, uint64_t num_blocks) +{ + /* Return failure if offset_blocks + num_blocks is less than offset_blocks; indicates there + * has been an overflow and hence the offset has been wrapped around */ + if (offset_blocks + num_blocks < offset_blocks) { + return false; + } + + /* Return failure if offset_blocks + num_blocks exceeds the size of the bdev */ + if (offset_blocks + num_blocks > bdev->blockcnt) { + return false; + } + + return true; +} + +int +spdk_bdev_read(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + void *buf, uint64_t offset, uint64_t nbytes, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, nbytes, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_read_blocks(desc, ch, buf, offset_blocks, num_blocks, cb, cb_arg); +} + +int +spdk_bdev_read_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + void *buf, uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_READ; + bdev_io->u.bdev.iovs = &bdev_io->iov; + bdev_io->u.bdev.iovs[0].iov_base = buf; + bdev_io->u.bdev.iovs[0].iov_len = num_blocks * bdev->blocklen; + bdev_io->u.bdev.iovcnt = 1; + bdev_io->u.bdev.num_blocks = num_blocks; + bdev_io->u.bdev.offset_blocks = offset_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_readv(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + struct iovec *iov, int iovcnt, + uint64_t offset, uint64_t nbytes, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, nbytes, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_readv_blocks(desc, ch, iov, iovcnt, offset_blocks, num_blocks, cb, cb_arg); +} + +int spdk_bdev_readv_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + struct iovec *iov, int iovcnt, + uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_READ; + bdev_io->u.bdev.iovs = iov; + bdev_io->u.bdev.iovcnt = iovcnt; + bdev_io->u.bdev.num_blocks = num_blocks; + bdev_io->u.bdev.offset_blocks = offset_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_write(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + void *buf, uint64_t offset, uint64_t nbytes, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, nbytes, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_write_blocks(desc, ch, buf, offset_blocks, num_blocks, cb, cb_arg); +} + +int +spdk_bdev_write_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + void *buf, uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + return -EBADF; + } + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_WRITE; + bdev_io->u.bdev.iovs = &bdev_io->iov; + bdev_io->u.bdev.iovs[0].iov_base = buf; + bdev_io->u.bdev.iovs[0].iov_len = num_blocks * bdev->blocklen; + bdev_io->u.bdev.iovcnt = 1; + bdev_io->u.bdev.num_blocks = num_blocks; + bdev_io->u.bdev.offset_blocks = offset_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_writev(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + struct iovec *iov, int iovcnt, + uint64_t offset, uint64_t len, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, len, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_writev_blocks(desc, ch, iov, iovcnt, offset_blocks, num_blocks, cb, cb_arg); +} + +int +spdk_bdev_writev_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + struct iovec *iov, int iovcnt, + uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + return -EBADF; + } + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_WRITE; + bdev_io->u.bdev.iovs = iov; + bdev_io->u.bdev.iovcnt = iovcnt; + bdev_io->u.bdev.num_blocks = num_blocks; + bdev_io->u.bdev.offset_blocks = offset_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_write_zeroes(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + uint64_t offset, uint64_t len, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, len, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_write_zeroes_blocks(desc, ch, offset_blocks, num_blocks, cb, cb_arg); +} + +int +spdk_bdev_write_zeroes_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + return -EBADF; + } + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->type = SPDK_BDEV_IO_TYPE_WRITE_ZEROES; + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->u.bdev.offset_blocks = offset_blocks; + bdev_io->u.bdev.num_blocks = num_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + if (_spdk_bdev_io_type_supported(bdev, SPDK_BDEV_IO_TYPE_WRITE_ZEROES)) { + spdk_bdev_io_submit(bdev_io); + return 0; + } else if (_spdk_bdev_io_type_supported(bdev, SPDK_BDEV_IO_TYPE_WRITE)) { + assert(spdk_bdev_get_block_size(bdev) <= ZERO_BUFFER_SIZE); + bdev_io->u.bdev.split_remaining_num_blocks = num_blocks; + bdev_io->u.bdev.split_current_offset_blocks = offset_blocks; + _spdk_bdev_write_zero_buffer_next(bdev_io); + return 0; + } else { + spdk_bdev_free_io(bdev_io); + return -ENOTSUP; + } +} + +int +spdk_bdev_unmap(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + uint64_t offset, uint64_t nbytes, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, nbytes, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_unmap_blocks(desc, ch, offset_blocks, num_blocks, cb, cb_arg); +} + +int +spdk_bdev_unmap_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + return -EBADF; + } + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + if (num_blocks == 0) { + SPDK_ERRLOG("Can't unmap 0 bytes\n"); + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_UNMAP; + + bdev_io->u.bdev.iovs = &bdev_io->iov; + bdev_io->u.bdev.iovs[0].iov_base = NULL; + bdev_io->u.bdev.iovs[0].iov_len = 0; + bdev_io->u.bdev.iovcnt = 1; + + bdev_io->u.bdev.offset_blocks = offset_blocks; + bdev_io->u.bdev.num_blocks = num_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_flush(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + uint64_t offset, uint64_t length, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + uint64_t offset_blocks, num_blocks; + + if (spdk_bdev_bytes_to_blocks(desc->bdev, offset, &offset_blocks, length, &num_blocks) != 0) { + return -EINVAL; + } + + return spdk_bdev_flush_blocks(desc, ch, offset_blocks, num_blocks, cb, cb_arg); +} + +int +spdk_bdev_flush_blocks(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + uint64_t offset_blocks, uint64_t num_blocks, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + return -EBADF; + } + + if (!spdk_bdev_io_valid_blocks(bdev, offset_blocks, num_blocks)) { + return -EINVAL; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_FLUSH; + bdev_io->u.bdev.iovs = NULL; + bdev_io->u.bdev.iovcnt = 0; + bdev_io->u.bdev.offset_blocks = offset_blocks; + bdev_io->u.bdev.num_blocks = num_blocks; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +static void +_spdk_bdev_reset_dev(struct spdk_io_channel_iter *i, int status) +{ + struct spdk_bdev_channel *ch = spdk_io_channel_iter_get_ctx(i); + struct spdk_bdev_io *bdev_io; + + bdev_io = TAILQ_FIRST(&ch->queued_resets); + TAILQ_REMOVE(&ch->queued_resets, bdev_io, internal.link); + spdk_bdev_io_submit_reset(bdev_io); +} + +static void +_spdk_bdev_reset_freeze_channel(struct spdk_io_channel_iter *i) +{ + struct spdk_io_channel *ch; + struct spdk_bdev_channel *channel; + struct spdk_bdev_mgmt_channel *mgmt_channel; + struct spdk_bdev_shared_resource *shared_resource; + bdev_io_tailq_t tmp_queued; + + TAILQ_INIT(&tmp_queued); + + ch = spdk_io_channel_iter_get_channel(i); + channel = spdk_io_channel_get_ctx(ch); + shared_resource = channel->shared_resource; + mgmt_channel = shared_resource->mgmt_ch; + + channel->flags |= BDEV_CH_RESET_IN_PROGRESS; + + if ((channel->flags & BDEV_CH_QOS_ENABLED) != 0) { + /* The QoS object is always valid and readable while + * the channel flag is set, so the lock here should not + * be necessary. We're not in the fast path though, so + * just take it anyway. */ + pthread_mutex_lock(&channel->bdev->internal.mutex); + if (channel->bdev->internal.qos->ch == channel) { + TAILQ_SWAP(&channel->bdev->internal.qos->queued, &tmp_queued, spdk_bdev_io, internal.link); + } + pthread_mutex_unlock(&channel->bdev->internal.mutex); + } + + _spdk_bdev_abort_queued_io(&shared_resource->nomem_io, channel); + _spdk_bdev_abort_buf_io(&mgmt_channel->need_buf_small, channel); + _spdk_bdev_abort_buf_io(&mgmt_channel->need_buf_large, channel); + _spdk_bdev_abort_queued_io(&tmp_queued, channel); + + spdk_for_each_channel_continue(i, 0); +} + +static void +_spdk_bdev_start_reset(void *ctx) +{ + struct spdk_bdev_channel *ch = ctx; + + spdk_for_each_channel(__bdev_to_io_dev(ch->bdev), _spdk_bdev_reset_freeze_channel, + ch, _spdk_bdev_reset_dev); +} + +static void +_spdk_bdev_channel_start_reset(struct spdk_bdev_channel *ch) +{ + struct spdk_bdev *bdev = ch->bdev; + + assert(!TAILQ_EMPTY(&ch->queued_resets)); + + pthread_mutex_lock(&bdev->internal.mutex); + if (bdev->internal.reset_in_progress == NULL) { + bdev->internal.reset_in_progress = TAILQ_FIRST(&ch->queued_resets); + /* + * Take a channel reference for the target bdev for the life of this + * reset. This guards against the channel getting destroyed while + * spdk_for_each_channel() calls related to this reset IO are in + * progress. We will release the reference when this reset is + * completed. + */ + bdev->internal.reset_in_progress->u.reset.ch_ref = spdk_get_io_channel(__bdev_to_io_dev(bdev)); + _spdk_bdev_start_reset(ch); + } + pthread_mutex_unlock(&bdev->internal.mutex); +} + +int +spdk_bdev_reset(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_RESET; + bdev_io->u.reset.ch_ref = NULL; + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + pthread_mutex_lock(&bdev->internal.mutex); + TAILQ_INSERT_TAIL(&channel->queued_resets, bdev_io, internal.link); + pthread_mutex_unlock(&bdev->internal.mutex); + + _spdk_bdev_channel_start_reset(channel); + + return 0; +} + +void +spdk_bdev_get_io_stat(struct spdk_bdev *bdev, struct spdk_io_channel *ch, + struct spdk_bdev_io_stat *stat) +{ + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + *stat = channel->stat; +} + +static void +_spdk_bdev_get_device_stat_done(struct spdk_io_channel_iter *i, int status) +{ + void *io_device = spdk_io_channel_iter_get_io_device(i); + struct spdk_bdev_iostat_ctx *bdev_iostat_ctx = spdk_io_channel_iter_get_ctx(i); + + bdev_iostat_ctx->cb(__bdev_from_io_dev(io_device), bdev_iostat_ctx->stat, + bdev_iostat_ctx->cb_arg, 0); + free(bdev_iostat_ctx); +} + +static void +_spdk_bdev_get_each_channel_stat(struct spdk_io_channel_iter *i) +{ + struct spdk_bdev_iostat_ctx *bdev_iostat_ctx = spdk_io_channel_iter_get_ctx(i); + struct spdk_io_channel *ch = spdk_io_channel_iter_get_channel(i); + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + _spdk_bdev_io_stat_add(bdev_iostat_ctx->stat, &channel->stat); + spdk_for_each_channel_continue(i, 0); +} + +void +spdk_bdev_get_device_stat(struct spdk_bdev *bdev, struct spdk_bdev_io_stat *stat, + spdk_bdev_get_device_stat_cb cb, void *cb_arg) +{ + struct spdk_bdev_iostat_ctx *bdev_iostat_ctx; + + assert(bdev != NULL); + assert(stat != NULL); + assert(cb != NULL); + + bdev_iostat_ctx = calloc(1, sizeof(struct spdk_bdev_iostat_ctx)); + if (bdev_iostat_ctx == NULL) { + SPDK_ERRLOG("Unable to allocate memory for spdk_bdev_iostat_ctx\n"); + cb(bdev, stat, cb_arg, -ENOMEM); + return; + } + + bdev_iostat_ctx->stat = stat; + bdev_iostat_ctx->cb = cb; + bdev_iostat_ctx->cb_arg = cb_arg; + + /* Start with the statistics from previously deleted channels. */ + pthread_mutex_lock(&bdev->internal.mutex); + _spdk_bdev_io_stat_add(bdev_iostat_ctx->stat, &bdev->internal.stat); + pthread_mutex_unlock(&bdev->internal.mutex); + + /* Then iterate and add the statistics from each existing channel. */ + spdk_for_each_channel(__bdev_to_io_dev(bdev), + _spdk_bdev_get_each_channel_stat, + bdev_iostat_ctx, + _spdk_bdev_get_device_stat_done); +} + +int +spdk_bdev_nvme_admin_passthru(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + const struct spdk_nvme_cmd *cmd, void *buf, size_t nbytes, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + return -EBADF; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_NVME_ADMIN; + bdev_io->u.nvme_passthru.cmd = *cmd; + bdev_io->u.nvme_passthru.buf = buf; + bdev_io->u.nvme_passthru.nbytes = nbytes; + bdev_io->u.nvme_passthru.md_buf = NULL; + bdev_io->u.nvme_passthru.md_len = 0; + + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_nvme_io_passthru(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + const struct spdk_nvme_cmd *cmd, void *buf, size_t nbytes, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + /* + * Do not try to parse the NVMe command - we could maybe use bits in the opcode + * to easily determine if the command is a read or write, but for now just + * do not allow io_passthru with a read-only descriptor. + */ + return -EBADF; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_NVME_IO; + bdev_io->u.nvme_passthru.cmd = *cmd; + bdev_io->u.nvme_passthru.buf = buf; + bdev_io->u.nvme_passthru.nbytes = nbytes; + bdev_io->u.nvme_passthru.md_buf = NULL; + bdev_io->u.nvme_passthru.md_len = 0; + + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_nvme_io_passthru_md(struct spdk_bdev_desc *desc, struct spdk_io_channel *ch, + const struct spdk_nvme_cmd *cmd, void *buf, size_t nbytes, void *md_buf, size_t md_len, + spdk_bdev_io_completion_cb cb, void *cb_arg) +{ + struct spdk_bdev *bdev = desc->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + + if (!desc->write) { + /* + * Do not try to parse the NVMe command - we could maybe use bits in the opcode + * to easily determine if the command is a read or write, but for now just + * do not allow io_passthru with a read-only descriptor. + */ + return -EBADF; + } + + bdev_io = spdk_bdev_get_io(channel); + if (!bdev_io) { + return -ENOMEM; + } + + bdev_io->internal.ch = channel; + bdev_io->internal.desc = desc; + bdev_io->type = SPDK_BDEV_IO_TYPE_NVME_IO_MD; + bdev_io->u.nvme_passthru.cmd = *cmd; + bdev_io->u.nvme_passthru.buf = buf; + bdev_io->u.nvme_passthru.nbytes = nbytes; + bdev_io->u.nvme_passthru.md_buf = md_buf; + bdev_io->u.nvme_passthru.md_len = md_len; + + spdk_bdev_io_init(bdev_io, bdev, cb_arg, cb); + + spdk_bdev_io_submit(bdev_io); + return 0; +} + +int +spdk_bdev_queue_io_wait(struct spdk_bdev *bdev, struct spdk_io_channel *ch, + struct spdk_bdev_io_wait_entry *entry) +{ + struct spdk_bdev_channel *channel = spdk_io_channel_get_ctx(ch); + struct spdk_bdev_mgmt_channel *mgmt_ch = channel->shared_resource->mgmt_ch; + + if (bdev != entry->bdev) { + SPDK_ERRLOG("bdevs do not match\n"); + return -EINVAL; + } + + if (mgmt_ch->per_thread_cache_count > 0) { + SPDK_ERRLOG("Cannot queue io_wait if spdk_bdev_io available in per-thread cache\n"); + return -EINVAL; + } + + TAILQ_INSERT_TAIL(&mgmt_ch->io_wait_queue, entry, link); + return 0; +} + +static void +_spdk_bdev_ch_retry_io(struct spdk_bdev_channel *bdev_ch) +{ + struct spdk_bdev *bdev = bdev_ch->bdev; + struct spdk_bdev_shared_resource *shared_resource = bdev_ch->shared_resource; + struct spdk_bdev_io *bdev_io; + + if (shared_resource->io_outstanding > shared_resource->nomem_threshold) { + /* + * Allow some more I/O to complete before retrying the nomem_io queue. + * Some drivers (such as nvme) cannot immediately take a new I/O in + * the context of a completion, because the resources for the I/O are + * not released until control returns to the bdev poller. Also, we + * may require several small I/O to complete before a larger I/O + * (that requires splitting) can be submitted. + */ + return; + } + + while (!TAILQ_EMPTY(&shared_resource->nomem_io)) { + bdev_io = TAILQ_FIRST(&shared_resource->nomem_io); + TAILQ_REMOVE(&shared_resource->nomem_io, bdev_io, internal.link); + bdev_io->internal.ch->io_outstanding++; + shared_resource->io_outstanding++; + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_PENDING; + bdev->fn_table->submit_request(bdev_io->internal.ch->channel, bdev_io); + if (bdev_io->internal.status == SPDK_BDEV_IO_STATUS_NOMEM) { + break; + } + } +} + +static inline void +_spdk_bdev_io_complete(void *ctx) +{ + struct spdk_bdev_io *bdev_io = ctx; + uint64_t tsc; + + if (spdk_unlikely(bdev_io->internal.in_submit_request || bdev_io->internal.io_submit_ch)) { + /* + * Send the completion to the thread that originally submitted the I/O, + * which may not be the current thread in the case of QoS. + */ + if (bdev_io->internal.io_submit_ch) { + bdev_io->internal.ch = bdev_io->internal.io_submit_ch; + bdev_io->internal.io_submit_ch = NULL; + } + + /* + * Defer completion to avoid potential infinite recursion if the + * user's completion callback issues a new I/O. + */ + spdk_thread_send_msg(spdk_io_channel_get_thread(bdev_io->internal.ch->channel), + _spdk_bdev_io_complete, bdev_io); + return; + } + + tsc = spdk_get_ticks(); + spdk_trace_record_tsc(tsc, TRACE_BDEV_IO_DONE, 0, 0, (uintptr_t)bdev_io, 0); + + if (bdev_io->internal.status == SPDK_BDEV_IO_STATUS_SUCCESS) { + switch (bdev_io->type) { + case SPDK_BDEV_IO_TYPE_READ: + bdev_io->internal.ch->stat.bytes_read += bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen; + bdev_io->internal.ch->stat.num_read_ops++; + bdev_io->internal.ch->stat.read_latency_ticks += (tsc - bdev_io->internal.submit_tsc); + break; + case SPDK_BDEV_IO_TYPE_WRITE: + bdev_io->internal.ch->stat.bytes_written += bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen; + bdev_io->internal.ch->stat.num_write_ops++; + bdev_io->internal.ch->stat.write_latency_ticks += (tsc - bdev_io->internal.submit_tsc); + break; + default: + break; + } + } + +#ifdef SPDK_CONFIG_VTUNE + uint64_t now_tsc = spdk_get_ticks(); + if (now_tsc > (bdev_io->internal.ch->start_tsc + bdev_io->internal.ch->interval_tsc)) { + uint64_t data[5]; + + data[0] = bdev_io->internal.ch->stat.num_read_ops - bdev_io->internal.ch->prev_stat.num_read_ops; + data[1] = bdev_io->internal.ch->stat.bytes_read - bdev_io->internal.ch->prev_stat.bytes_read; + data[2] = bdev_io->internal.ch->stat.num_write_ops - bdev_io->internal.ch->prev_stat.num_write_ops; + data[3] = bdev_io->internal.ch->stat.bytes_written - bdev_io->internal.ch->prev_stat.bytes_written; + data[4] = bdev_io->bdev->fn_table->get_spin_time ? + bdev_io->bdev->fn_table->get_spin_time(bdev_io->internal.ch->channel) : 0; + + __itt_metadata_add(g_bdev_mgr.domain, __itt_null, bdev_io->internal.ch->handle, + __itt_metadata_u64, 5, data); + + bdev_io->internal.ch->prev_stat = bdev_io->internal.ch->stat; + bdev_io->internal.ch->start_tsc = now_tsc; + } +#endif + + assert(bdev_io->internal.cb != NULL); + assert(spdk_get_thread() == spdk_io_channel_get_thread(bdev_io->internal.ch->channel)); + + bdev_io->internal.cb(bdev_io, bdev_io->internal.status == SPDK_BDEV_IO_STATUS_SUCCESS, + bdev_io->internal.caller_ctx); +} + +static void +_spdk_bdev_reset_complete(struct spdk_io_channel_iter *i, int status) +{ + struct spdk_bdev_io *bdev_io = spdk_io_channel_iter_get_ctx(i); + + if (bdev_io->u.reset.ch_ref != NULL) { + spdk_put_io_channel(bdev_io->u.reset.ch_ref); + bdev_io->u.reset.ch_ref = NULL; + } + + _spdk_bdev_io_complete(bdev_io); +} + +static void +_spdk_bdev_unfreeze_channel(struct spdk_io_channel_iter *i) +{ + struct spdk_io_channel *_ch = spdk_io_channel_iter_get_channel(i); + struct spdk_bdev_channel *ch = spdk_io_channel_get_ctx(_ch); + + ch->flags &= ~BDEV_CH_RESET_IN_PROGRESS; + if (!TAILQ_EMPTY(&ch->queued_resets)) { + _spdk_bdev_channel_start_reset(ch); + } + + spdk_for_each_channel_continue(i, 0); +} + +void +spdk_bdev_io_complete(struct spdk_bdev_io *bdev_io, enum spdk_bdev_io_status status) +{ + struct spdk_bdev *bdev = bdev_io->bdev; + struct spdk_bdev_channel *bdev_ch = bdev_io->internal.ch; + struct spdk_bdev_shared_resource *shared_resource = bdev_ch->shared_resource; + + bdev_io->internal.status = status; + + if (spdk_unlikely(bdev_io->type == SPDK_BDEV_IO_TYPE_RESET)) { + bool unlock_channels = false; + + if (status == SPDK_BDEV_IO_STATUS_NOMEM) { + SPDK_ERRLOG("NOMEM returned for reset\n"); + } + pthread_mutex_lock(&bdev->internal.mutex); + if (bdev_io == bdev->internal.reset_in_progress) { + bdev->internal.reset_in_progress = NULL; + unlock_channels = true; + } + pthread_mutex_unlock(&bdev->internal.mutex); + + if (unlock_channels) { + spdk_for_each_channel(__bdev_to_io_dev(bdev), _spdk_bdev_unfreeze_channel, + bdev_io, _spdk_bdev_reset_complete); + return; + } + } else { + assert(bdev_ch->io_outstanding > 0); + assert(shared_resource->io_outstanding > 0); + bdev_ch->io_outstanding--; + shared_resource->io_outstanding--; + + if (spdk_unlikely(status == SPDK_BDEV_IO_STATUS_NOMEM)) { + TAILQ_INSERT_HEAD(&shared_resource->nomem_io, bdev_io, internal.link); + /* + * Wait for some of the outstanding I/O to complete before we + * retry any of the nomem_io. Normally we will wait for + * NOMEM_THRESHOLD_COUNT I/O to complete but for low queue + * depth channels we will instead wait for half to complete. + */ + shared_resource->nomem_threshold = spdk_max((int64_t)shared_resource->io_outstanding / 2, + (int64_t)shared_resource->io_outstanding - NOMEM_THRESHOLD_COUNT); + return; + } + + if (spdk_unlikely(!TAILQ_EMPTY(&shared_resource->nomem_io))) { + _spdk_bdev_ch_retry_io(bdev_ch); + } + } + + _spdk_bdev_io_complete(bdev_io); +} + +void +spdk_bdev_io_complete_scsi_status(struct spdk_bdev_io *bdev_io, enum spdk_scsi_status sc, + enum spdk_scsi_sense sk, uint8_t asc, uint8_t ascq) +{ + if (sc == SPDK_SCSI_STATUS_GOOD) { + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_SUCCESS; + } else { + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_SCSI_ERROR; + bdev_io->internal.error.scsi.sc = sc; + bdev_io->internal.error.scsi.sk = sk; + bdev_io->internal.error.scsi.asc = asc; + bdev_io->internal.error.scsi.ascq = ascq; + } + + spdk_bdev_io_complete(bdev_io, bdev_io->internal.status); +} + +void +spdk_bdev_io_get_scsi_status(const struct spdk_bdev_io *bdev_io, + int *sc, int *sk, int *asc, int *ascq) +{ + assert(sc != NULL); + assert(sk != NULL); + assert(asc != NULL); + assert(ascq != NULL); + + switch (bdev_io->internal.status) { + case SPDK_BDEV_IO_STATUS_SUCCESS: + *sc = SPDK_SCSI_STATUS_GOOD; + *sk = SPDK_SCSI_SENSE_NO_SENSE; + *asc = SPDK_SCSI_ASC_NO_ADDITIONAL_SENSE; + *ascq = SPDK_SCSI_ASCQ_CAUSE_NOT_REPORTABLE; + break; + case SPDK_BDEV_IO_STATUS_NVME_ERROR: + spdk_scsi_nvme_translate(bdev_io, sc, sk, asc, ascq); + break; + case SPDK_BDEV_IO_STATUS_SCSI_ERROR: + *sc = bdev_io->internal.error.scsi.sc; + *sk = bdev_io->internal.error.scsi.sk; + *asc = bdev_io->internal.error.scsi.asc; + *ascq = bdev_io->internal.error.scsi.ascq; + break; + default: + *sc = SPDK_SCSI_STATUS_CHECK_CONDITION; + *sk = SPDK_SCSI_SENSE_ABORTED_COMMAND; + *asc = SPDK_SCSI_ASC_NO_ADDITIONAL_SENSE; + *ascq = SPDK_SCSI_ASCQ_CAUSE_NOT_REPORTABLE; + break; + } +} + +void +spdk_bdev_io_complete_nvme_status(struct spdk_bdev_io *bdev_io, int sct, int sc) +{ + if (sct == SPDK_NVME_SCT_GENERIC && sc == SPDK_NVME_SC_SUCCESS) { + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_SUCCESS; + } else { + bdev_io->internal.error.nvme.sct = sct; + bdev_io->internal.error.nvme.sc = sc; + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_NVME_ERROR; + } + + spdk_bdev_io_complete(bdev_io, bdev_io->internal.status); +} + +void +spdk_bdev_io_get_nvme_status(const struct spdk_bdev_io *bdev_io, int *sct, int *sc) +{ + assert(sct != NULL); + assert(sc != NULL); + + if (bdev_io->internal.status == SPDK_BDEV_IO_STATUS_NVME_ERROR) { + *sct = bdev_io->internal.error.nvme.sct; + *sc = bdev_io->internal.error.nvme.sc; + } else if (bdev_io->internal.status == SPDK_BDEV_IO_STATUS_SUCCESS) { + *sct = SPDK_NVME_SCT_GENERIC; + *sc = SPDK_NVME_SC_SUCCESS; + } else { + *sct = SPDK_NVME_SCT_GENERIC; + *sc = SPDK_NVME_SC_INTERNAL_DEVICE_ERROR; + } +} + +struct spdk_thread * +spdk_bdev_io_get_thread(struct spdk_bdev_io *bdev_io) +{ + return spdk_io_channel_get_thread(bdev_io->internal.ch->channel); +} + +static void +_spdk_bdev_qos_config_limit(struct spdk_bdev *bdev, uint64_t *limits) +{ + uint64_t min_qos_set; + int i; + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (limits[i] != SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + break; + } + } + + if (i == SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES) { + SPDK_ERRLOG("Invalid rate limits set.\n"); + return; + } + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (limits[i] == SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + continue; + } + + if (_spdk_bdev_qos_is_iops_rate_limit(i) == true) { + min_qos_set = SPDK_BDEV_QOS_MIN_IOS_PER_SEC; + } else { + min_qos_set = SPDK_BDEV_QOS_MIN_BYTES_PER_SEC; + } + + if (limits[i] == 0 || limits[i] % min_qos_set) { + SPDK_ERRLOG("Assigned limit %" PRIu64 " on bdev %s is not multiple of %" PRIu64 "\n", + limits[i], bdev->name, min_qos_set); + SPDK_ERRLOG("Failed to enable QoS on this bdev %s\n", bdev->name); + return; + } + } + + if (!bdev->internal.qos) { + bdev->internal.qos = calloc(1, sizeof(*bdev->internal.qos)); + if (!bdev->internal.qos) { + SPDK_ERRLOG("Unable to allocate memory for QoS tracking\n"); + return; + } + } + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + bdev->internal.qos->rate_limits[i].limit = limits[i]; + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Bdev:%s QoS type:%d set:%lu\n", + bdev->name, i, limits[i]); + } + + return; +} + +static void +_spdk_bdev_qos_config(struct spdk_bdev *bdev) +{ + struct spdk_conf_section *sp = NULL; + const char *val = NULL; + int i = 0, j = 0; + uint64_t limits[SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES] = {}; + bool config_qos = false; + + sp = spdk_conf_find_section(NULL, "QoS"); + if (!sp) { + return; + } + + while (j < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES) { + limits[j] = SPDK_BDEV_QOS_LIMIT_NOT_DEFINED; + + i = 0; + while (true) { + val = spdk_conf_section_get_nmval(sp, qos_conf_type[j], i, 0); + if (!val) { + break; + } + + if (strcmp(bdev->name, val) != 0) { + i++; + continue; + } + + val = spdk_conf_section_get_nmval(sp, qos_conf_type[j], i, 1); + if (val) { + if (_spdk_bdev_qos_is_iops_rate_limit(j) == true) { + limits[j] = strtoull(val, NULL, 10); + } else { + limits[j] = strtoull(val, NULL, 10) * 1024 * 1024; + } + config_qos = true; + } + + break; + } + + j++; + } + + if (config_qos == true) { + _spdk_bdev_qos_config_limit(bdev, limits); + } + + return; +} + +static int +spdk_bdev_init(struct spdk_bdev *bdev) +{ + char *bdev_name; + + assert(bdev->module != NULL); + + if (!bdev->name) { + SPDK_ERRLOG("Bdev name is NULL\n"); + return -EINVAL; + } + + if (spdk_bdev_get_by_name(bdev->name)) { + SPDK_ERRLOG("Bdev name:%s already exists\n", bdev->name); + return -EEXIST; + } + + /* Users often register their own I/O devices using the bdev name. In + * order to avoid conflicts, prepend bdev_. */ + bdev_name = spdk_sprintf_alloc("bdev_%s", bdev->name); + if (!bdev_name) { + SPDK_ERRLOG("Unable to allocate memory for internal bdev name.\n"); + return -ENOMEM; + } + + bdev->internal.status = SPDK_BDEV_STATUS_READY; + bdev->internal.measured_queue_depth = UINT64_MAX; + bdev->internal.claim_module = NULL; + bdev->internal.qd_poller = NULL; + bdev->internal.qos = NULL; + + TAILQ_INIT(&bdev->internal.open_descs); + + TAILQ_INIT(&bdev->aliases); + + bdev->internal.reset_in_progress = NULL; + + _spdk_bdev_qos_config(bdev); + + spdk_io_device_register(__bdev_to_io_dev(bdev), + spdk_bdev_channel_create, spdk_bdev_channel_destroy, + sizeof(struct spdk_bdev_channel), + bdev_name); + + free(bdev_name); + + pthread_mutex_init(&bdev->internal.mutex, NULL); + return 0; +} + +static void +spdk_bdev_destroy_cb(void *io_device) +{ + int rc; + struct spdk_bdev *bdev; + spdk_bdev_unregister_cb cb_fn; + void *cb_arg; + + bdev = __bdev_from_io_dev(io_device); + cb_fn = bdev->internal.unregister_cb; + cb_arg = bdev->internal.unregister_ctx; + + rc = bdev->fn_table->destruct(bdev->ctxt); + if (rc < 0) { + SPDK_ERRLOG("destruct failed\n"); + } + if (rc <= 0 && cb_fn != NULL) { + cb_fn(cb_arg, rc); + } +} + + +static void +spdk_bdev_fini(struct spdk_bdev *bdev) +{ + pthread_mutex_destroy(&bdev->internal.mutex); + + free(bdev->internal.qos); + + spdk_io_device_unregister(__bdev_to_io_dev(bdev), spdk_bdev_destroy_cb); +} + +static void +spdk_bdev_start(struct spdk_bdev *bdev) +{ + struct spdk_bdev_module *module; + uint32_t action; + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Inserting bdev %s into list\n", bdev->name); + TAILQ_INSERT_TAIL(&g_bdev_mgr.bdevs, bdev, internal.link); + + /* Examine configuration before initializing I/O */ + TAILQ_FOREACH(module, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (module->examine_config) { + action = module->internal.action_in_progress; + module->internal.action_in_progress++; + module->examine_config(bdev); + if (action != module->internal.action_in_progress) { + SPDK_ERRLOG("examine_config for module %s did not call spdk_bdev_module_examine_done()\n", + module->name); + } + } + } + + if (bdev->internal.claim_module) { + return; + } + + TAILQ_FOREACH(module, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (module->examine_disk) { + module->internal.action_in_progress++; + module->examine_disk(bdev); + } + } +} + +int +spdk_bdev_register(struct spdk_bdev *bdev) +{ + int rc = spdk_bdev_init(bdev); + + if (rc == 0) { + spdk_bdev_start(bdev); + } + + return rc; +} + +int +spdk_vbdev_register(struct spdk_bdev *vbdev, struct spdk_bdev **base_bdevs, int base_bdev_count) +{ + int rc; + + rc = spdk_bdev_init(vbdev); + if (rc) { + return rc; + } + + spdk_bdev_start(vbdev); + return 0; +} + +void +spdk_bdev_destruct_done(struct spdk_bdev *bdev, int bdeverrno) +{ + if (bdev->internal.unregister_cb != NULL) { + bdev->internal.unregister_cb(bdev->internal.unregister_ctx, bdeverrno); + } +} + +static void +_remove_notify(void *arg) +{ + struct spdk_bdev_desc *desc = arg; + + desc->remove_scheduled = false; + + if (desc->closed) { + free(desc); + } else { + desc->remove_cb(desc->remove_ctx); + } +} + +void +spdk_bdev_unregister(struct spdk_bdev *bdev, spdk_bdev_unregister_cb cb_fn, void *cb_arg) +{ + struct spdk_bdev_desc *desc, *tmp; + bool do_destruct = true; + struct spdk_thread *thread; + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Removing bdev %s from list\n", bdev->name); + + thread = spdk_get_thread(); + if (!thread) { + /* The user called this from a non-SPDK thread. */ + if (cb_fn != NULL) { + cb_fn(cb_arg, -ENOTSUP); + } + return; + } + + pthread_mutex_lock(&bdev->internal.mutex); + + bdev->internal.status = SPDK_BDEV_STATUS_REMOVING; + bdev->internal.unregister_cb = cb_fn; + bdev->internal.unregister_ctx = cb_arg; + + TAILQ_FOREACH_SAFE(desc, &bdev->internal.open_descs, link, tmp) { + if (desc->remove_cb) { + do_destruct = false; + /* + * Defer invocation of the remove_cb to a separate message that will + * run later on its thread. This ensures this context unwinds and + * we don't recursively unregister this bdev again if the remove_cb + * immediately closes its descriptor. + */ + if (!desc->remove_scheduled) { + /* Avoid scheduling removal of the same descriptor multiple times. */ + desc->remove_scheduled = true; + spdk_thread_send_msg(desc->thread, _remove_notify, desc); + } + } + } + + if (!do_destruct) { + pthread_mutex_unlock(&bdev->internal.mutex); + return; + } + + TAILQ_REMOVE(&g_bdev_mgr.bdevs, bdev, internal.link); + pthread_mutex_unlock(&bdev->internal.mutex); + + spdk_bdev_fini(bdev); +} + +int +spdk_bdev_open(struct spdk_bdev *bdev, bool write, spdk_bdev_remove_cb_t remove_cb, + void *remove_ctx, struct spdk_bdev_desc **_desc) +{ + struct spdk_bdev_desc *desc; + struct spdk_thread *thread; + + thread = spdk_get_thread(); + if (!thread) { + SPDK_ERRLOG("Cannot open bdev from non-SPDK thread.\n"); + return -ENOTSUP; + } + + desc = calloc(1, sizeof(*desc)); + if (desc == NULL) { + SPDK_ERRLOG("Failed to allocate memory for bdev descriptor\n"); + return -ENOMEM; + } + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Opening descriptor %p for bdev %s on thread %p\n", desc, bdev->name, + spdk_get_thread()); + + pthread_mutex_lock(&bdev->internal.mutex); + + if (write && bdev->internal.claim_module) { + SPDK_ERRLOG("Could not open %s - %s module already claimed it\n", + bdev->name, bdev->internal.claim_module->name); + free(desc); + pthread_mutex_unlock(&bdev->internal.mutex); + return -EPERM; + } + + TAILQ_INSERT_TAIL(&bdev->internal.open_descs, desc, link); + + desc->bdev = bdev; + desc->thread = thread; + desc->remove_cb = remove_cb; + desc->remove_ctx = remove_ctx; + desc->write = write; + *_desc = desc; + + pthread_mutex_unlock(&bdev->internal.mutex); + + return 0; +} + +void +spdk_bdev_close(struct spdk_bdev_desc *desc) +{ + struct spdk_bdev *bdev = desc->bdev; + bool do_unregister = false; + + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Closing descriptor %p for bdev %s on thread %p\n", desc, bdev->name, + spdk_get_thread()); + + assert(desc->thread == spdk_get_thread()); + + pthread_mutex_lock(&bdev->internal.mutex); + + TAILQ_REMOVE(&bdev->internal.open_descs, desc, link); + + desc->closed = true; + + if (!desc->remove_scheduled) { + free(desc); + } + + /* If no more descriptors, kill QoS channel */ + if (bdev->internal.qos && TAILQ_EMPTY(&bdev->internal.open_descs)) { + SPDK_DEBUGLOG(SPDK_LOG_BDEV, "Closed last descriptor for bdev %s on thread %p. Stopping QoS.\n", + bdev->name, spdk_get_thread()); + + if (spdk_bdev_qos_destroy(bdev)) { + /* There isn't anything we can do to recover here. Just let the + * old QoS poller keep running. The QoS handling won't change + * cores when the user allocates a new channel, but it won't break. */ + SPDK_ERRLOG("Unable to shut down QoS poller. It will continue running on the current thread.\n"); + } + } + + spdk_bdev_set_qd_sampling_period(bdev, 0); + + if (bdev->internal.status == SPDK_BDEV_STATUS_REMOVING && TAILQ_EMPTY(&bdev->internal.open_descs)) { + do_unregister = true; + } + pthread_mutex_unlock(&bdev->internal.mutex); + + if (do_unregister == true) { + spdk_bdev_unregister(bdev, bdev->internal.unregister_cb, bdev->internal.unregister_ctx); + } +} + +int +spdk_bdev_module_claim_bdev(struct spdk_bdev *bdev, struct spdk_bdev_desc *desc, + struct spdk_bdev_module *module) +{ + if (bdev->internal.claim_module != NULL) { + SPDK_ERRLOG("bdev %s already claimed by module %s\n", bdev->name, + bdev->internal.claim_module->name); + return -EPERM; + } + + if (desc && !desc->write) { + desc->write = true; + } + + bdev->internal.claim_module = module; + return 0; +} + +void +spdk_bdev_module_release_bdev(struct spdk_bdev *bdev) +{ + assert(bdev->internal.claim_module != NULL); + bdev->internal.claim_module = NULL; +} + +struct spdk_bdev * +spdk_bdev_desc_get_bdev(struct spdk_bdev_desc *desc) +{ + return desc->bdev; +} + +void +spdk_bdev_io_get_iovec(struct spdk_bdev_io *bdev_io, struct iovec **iovp, int *iovcntp) +{ + struct iovec *iovs; + int iovcnt; + + if (bdev_io == NULL) { + return; + } + + switch (bdev_io->type) { + case SPDK_BDEV_IO_TYPE_READ: + iovs = bdev_io->u.bdev.iovs; + iovcnt = bdev_io->u.bdev.iovcnt; + break; + case SPDK_BDEV_IO_TYPE_WRITE: + iovs = bdev_io->u.bdev.iovs; + iovcnt = bdev_io->u.bdev.iovcnt; + break; + default: + iovs = NULL; + iovcnt = 0; + break; + } + + if (iovp) { + *iovp = iovs; + } + if (iovcntp) { + *iovcntp = iovcnt; + } +} + +void +spdk_bdev_module_list_add(struct spdk_bdev_module *bdev_module) +{ + + if (spdk_bdev_module_list_find(bdev_module->name)) { + SPDK_ERRLOG("ERROR: module '%s' already registered.\n", bdev_module->name); + assert(false); + } + + if (bdev_module->async_init) { + bdev_module->internal.action_in_progress = 1; + } + + /* + * Modules with examine callbacks must be initialized first, so they are + * ready to handle examine callbacks from later modules that will + * register physical bdevs. + */ + if (bdev_module->examine_config != NULL || bdev_module->examine_disk != NULL) { + TAILQ_INSERT_HEAD(&g_bdev_mgr.bdev_modules, bdev_module, internal.tailq); + } else { + TAILQ_INSERT_TAIL(&g_bdev_mgr.bdev_modules, bdev_module, internal.tailq); + } +} + +struct spdk_bdev_module * +spdk_bdev_module_list_find(const char *name) +{ + struct spdk_bdev_module *bdev_module; + + TAILQ_FOREACH(bdev_module, &g_bdev_mgr.bdev_modules, internal.tailq) { + if (strcmp(name, bdev_module->name) == 0) { + break; + } + } + + return bdev_module; +} + +static void +_spdk_bdev_write_zero_buffer_next(void *_bdev_io) +{ + struct spdk_bdev_io *bdev_io = _bdev_io; + uint64_t num_bytes, num_blocks; + int rc; + + num_bytes = spdk_min(spdk_bdev_get_block_size(bdev_io->bdev) * + bdev_io->u.bdev.split_remaining_num_blocks, + ZERO_BUFFER_SIZE); + num_blocks = num_bytes / spdk_bdev_get_block_size(bdev_io->bdev); + + rc = spdk_bdev_write_blocks(bdev_io->internal.desc, + spdk_io_channel_from_ctx(bdev_io->internal.ch), + g_bdev_mgr.zero_buffer, + bdev_io->u.bdev.split_current_offset_blocks, num_blocks, + _spdk_bdev_write_zero_buffer_done, bdev_io); + if (rc == 0) { + bdev_io->u.bdev.split_remaining_num_blocks -= num_blocks; + bdev_io->u.bdev.split_current_offset_blocks += num_blocks; + } else if (rc == -ENOMEM) { + _spdk_bdev_queue_io_wait_with_cb(bdev_io, _spdk_bdev_write_zero_buffer_next); + } else { + bdev_io->internal.status = SPDK_BDEV_IO_STATUS_FAILED; + bdev_io->internal.cb(bdev_io, false, bdev_io->internal.caller_ctx); + } +} + +static void +_spdk_bdev_write_zero_buffer_done(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg) +{ + struct spdk_bdev_io *parent_io = cb_arg; + + spdk_bdev_free_io(bdev_io); + + if (!success) { + parent_io->internal.status = SPDK_BDEV_IO_STATUS_FAILED; + parent_io->internal.cb(parent_io, false, parent_io->internal.caller_ctx); + return; + } + + if (parent_io->u.bdev.split_remaining_num_blocks == 0) { + parent_io->internal.status = SPDK_BDEV_IO_STATUS_SUCCESS; + parent_io->internal.cb(parent_io, true, parent_io->internal.caller_ctx); + return; + } + + _spdk_bdev_write_zero_buffer_next(parent_io); +} + +struct set_qos_limit_ctx { + void (*cb_fn)(void *cb_arg, int status); + void *cb_arg; + struct spdk_bdev *bdev; +}; + +static void +_spdk_bdev_set_qos_limit_done(struct set_qos_limit_ctx *ctx, int status) +{ + pthread_mutex_lock(&ctx->bdev->internal.mutex); + ctx->bdev->internal.qos_mod_in_progress = false; + pthread_mutex_unlock(&ctx->bdev->internal.mutex); + + ctx->cb_fn(ctx->cb_arg, status); + free(ctx); +} + +static void +_spdk_bdev_disable_qos_done(void *cb_arg) +{ + struct set_qos_limit_ctx *ctx = cb_arg; + struct spdk_bdev *bdev = ctx->bdev; + struct spdk_bdev_io *bdev_io; + struct spdk_bdev_qos *qos; + + pthread_mutex_lock(&bdev->internal.mutex); + qos = bdev->internal.qos; + bdev->internal.qos = NULL; + pthread_mutex_unlock(&bdev->internal.mutex); + + while (!TAILQ_EMPTY(&qos->queued)) { + /* Send queued I/O back to their original thread for resubmission. */ + bdev_io = TAILQ_FIRST(&qos->queued); + TAILQ_REMOVE(&qos->queued, bdev_io, internal.link); + + if (bdev_io->internal.io_submit_ch) { + /* + * Channel was changed when sending it to the QoS thread - change it back + * before sending it back to the original thread. + */ + bdev_io->internal.ch = bdev_io->internal.io_submit_ch; + bdev_io->internal.io_submit_ch = NULL; + } + + spdk_thread_send_msg(spdk_io_channel_get_thread(bdev_io->internal.ch->channel), + _spdk_bdev_io_submit, bdev_io); + } + + spdk_put_io_channel(spdk_io_channel_from_ctx(qos->ch)); + spdk_poller_unregister(&qos->poller); + + free(qos); + + _spdk_bdev_set_qos_limit_done(ctx, 0); +} + +static void +_spdk_bdev_disable_qos_msg_done(struct spdk_io_channel_iter *i, int status) +{ + void *io_device = spdk_io_channel_iter_get_io_device(i); + struct spdk_bdev *bdev = __bdev_from_io_dev(io_device); + struct set_qos_limit_ctx *ctx = spdk_io_channel_iter_get_ctx(i); + struct spdk_thread *thread; + + pthread_mutex_lock(&bdev->internal.mutex); + thread = bdev->internal.qos->thread; + pthread_mutex_unlock(&bdev->internal.mutex); + + spdk_thread_send_msg(thread, _spdk_bdev_disable_qos_done, ctx); +} + +static void +_spdk_bdev_disable_qos_msg(struct spdk_io_channel_iter *i) +{ + struct spdk_io_channel *ch = spdk_io_channel_iter_get_channel(i); + struct spdk_bdev_channel *bdev_ch = spdk_io_channel_get_ctx(ch); + + bdev_ch->flags &= ~BDEV_CH_QOS_ENABLED; + + spdk_for_each_channel_continue(i, 0); +} + +static void +_spdk_bdev_update_qos_rate_limit_msg(void *cb_arg) +{ + struct set_qos_limit_ctx *ctx = cb_arg; + struct spdk_bdev *bdev = ctx->bdev; + + pthread_mutex_lock(&bdev->internal.mutex); + spdk_bdev_qos_update_max_quota_per_timeslice(bdev->internal.qos); + pthread_mutex_unlock(&bdev->internal.mutex); + + _spdk_bdev_set_qos_limit_done(ctx, 0); +} + +static void +_spdk_bdev_enable_qos_msg(struct spdk_io_channel_iter *i) +{ + void *io_device = spdk_io_channel_iter_get_io_device(i); + struct spdk_bdev *bdev = __bdev_from_io_dev(io_device); + struct spdk_io_channel *ch = spdk_io_channel_iter_get_channel(i); + struct spdk_bdev_channel *bdev_ch = spdk_io_channel_get_ctx(ch); + + pthread_mutex_lock(&bdev->internal.mutex); + _spdk_bdev_enable_qos(bdev, bdev_ch); + pthread_mutex_unlock(&bdev->internal.mutex); + spdk_for_each_channel_continue(i, 0); +} + +static void +_spdk_bdev_enable_qos_done(struct spdk_io_channel_iter *i, int status) +{ + struct set_qos_limit_ctx *ctx = spdk_io_channel_iter_get_ctx(i); + + _spdk_bdev_set_qos_limit_done(ctx, status); +} + +static void +_spdk_bdev_set_qos_rate_limits(struct spdk_bdev *bdev, uint64_t *limits) +{ + int i; + + assert(bdev->internal.qos != NULL); + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (limits[i] != SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + bdev->internal.qos->rate_limits[i].limit = limits[i]; + + if (limits[i] == 0) { + bdev->internal.qos->rate_limits[i].limit = + SPDK_BDEV_QOS_LIMIT_NOT_DEFINED; + } + } + } +} + +void +spdk_bdev_set_qos_rate_limits(struct spdk_bdev *bdev, uint64_t *limits, + void (*cb_fn)(void *cb_arg, int status), void *cb_arg) +{ + struct set_qos_limit_ctx *ctx; + uint32_t limit_set_complement; + uint64_t min_limit_per_sec; + int i; + bool disable_rate_limit = true; + + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (limits[i] == SPDK_BDEV_QOS_LIMIT_NOT_DEFINED) { + continue; + } + + if (limits[i] > 0) { + disable_rate_limit = false; + } + + if (_spdk_bdev_qos_is_iops_rate_limit(i) == true) { + min_limit_per_sec = SPDK_BDEV_QOS_MIN_IOS_PER_SEC; + } else { + /* Change from megabyte to byte rate limit */ + limits[i] = limits[i] * 1024 * 1024; + min_limit_per_sec = SPDK_BDEV_QOS_MIN_BYTES_PER_SEC; + } + + limit_set_complement = limits[i] % min_limit_per_sec; + if (limit_set_complement) { + SPDK_ERRLOG("Requested rate limit %" PRIu64 " is not a multiple of %" PRIu64 "\n", + limits[i], min_limit_per_sec); + limits[i] += min_limit_per_sec - limit_set_complement; + SPDK_ERRLOG("Round up the rate limit to %" PRIu64 "\n", limits[i]); + } + } + + ctx = calloc(1, sizeof(*ctx)); + if (ctx == NULL) { + cb_fn(cb_arg, -ENOMEM); + return; + } + + ctx->cb_fn = cb_fn; + ctx->cb_arg = cb_arg; + ctx->bdev = bdev; + + pthread_mutex_lock(&bdev->internal.mutex); + if (bdev->internal.qos_mod_in_progress) { + pthread_mutex_unlock(&bdev->internal.mutex); + free(ctx); + cb_fn(cb_arg, -EAGAIN); + return; + } + bdev->internal.qos_mod_in_progress = true; + + if (disable_rate_limit == true && bdev->internal.qos) { + for (i = 0; i < SPDK_BDEV_QOS_NUM_RATE_LIMIT_TYPES; i++) { + if (limits[i] == SPDK_BDEV_QOS_LIMIT_NOT_DEFINED && + (bdev->internal.qos->rate_limits[i].limit > 0 && + bdev->internal.qos->rate_limits[i].limit != + SPDK_BDEV_QOS_LIMIT_NOT_DEFINED)) { + disable_rate_limit = false; + break; + } + } + } + + if (disable_rate_limit == false) { + if (bdev->internal.qos == NULL) { + /* Enabling */ + bdev->internal.qos = calloc(1, sizeof(*bdev->internal.qos)); + if (!bdev->internal.qos) { + pthread_mutex_unlock(&bdev->internal.mutex); + SPDK_ERRLOG("Unable to allocate memory for QoS tracking\n"); + free(ctx); + cb_fn(cb_arg, -ENOMEM); + return; + } + + _spdk_bdev_set_qos_rate_limits(bdev, limits); + + spdk_for_each_channel(__bdev_to_io_dev(bdev), + _spdk_bdev_enable_qos_msg, ctx, + _spdk_bdev_enable_qos_done); + } else { + /* Updating */ + _spdk_bdev_set_qos_rate_limits(bdev, limits); + + spdk_thread_send_msg(bdev->internal.qos->thread, + _spdk_bdev_update_qos_rate_limit_msg, ctx); + } + } else { + if (bdev->internal.qos != NULL) { + _spdk_bdev_set_qos_rate_limits(bdev, limits); + + /* Disabling */ + spdk_for_each_channel(__bdev_to_io_dev(bdev), + _spdk_bdev_disable_qos_msg, ctx, + _spdk_bdev_disable_qos_msg_done); + } else { + pthread_mutex_unlock(&bdev->internal.mutex); + _spdk_bdev_set_qos_limit_done(ctx, 0); + return; + } + } + + pthread_mutex_unlock(&bdev->internal.mutex); +} + +SPDK_LOG_REGISTER_COMPONENT("bdev", SPDK_LOG_BDEV) + +SPDK_TRACE_REGISTER_FN(bdev_trace) +{ + spdk_trace_register_owner(OWNER_BDEV, 'b'); + spdk_trace_register_object(OBJECT_BDEV_IO, 'i'); + spdk_trace_register_description("BDEV_IO_START", "", TRACE_BDEV_IO_START, OWNER_BDEV, + OBJECT_BDEV_IO, 1, 0, "type: "); + spdk_trace_register_description("BDEV_IO_DONE", "", TRACE_BDEV_IO_DONE, OWNER_BDEV, + OBJECT_BDEV_IO, 0, 0, ""); +} |