update source to Ceph Pacific 16.2.2

[ceph.git] / ceph / src / spdk / module / bdev / delay / vbdev_delay.c

/*-
 *   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 "vbdev_delay.h"
#include "spdk/rpc.h"
#include "spdk/env.h"
#include "spdk/conf.h"
#include "spdk/endian.h"
#include "spdk/string.h"
#include "spdk/thread.h"
#include "spdk/util.h"

#include "spdk/bdev_module.h"
#include "spdk_internal/log.h"


static int vbdev_delay_init(void);
static int vbdev_delay_get_ctx_size(void);
static void vbdev_delay_examine(struct spdk_bdev *bdev);
static void vbdev_delay_finish(void);
static int vbdev_delay_config_json(struct spdk_json_write_ctx *w);

static struct spdk_bdev_module delay_if = {
        .name = "delay",
        .module_init = vbdev_delay_init,
        .config_text = NULL,
        .get_ctx_size = vbdev_delay_get_ctx_size,
        .examine_config = vbdev_delay_examine,
        .module_fini = vbdev_delay_finish,
        .config_json = vbdev_delay_config_json
};

SPDK_BDEV_MODULE_REGISTER(delay, &delay_if)

/* Associative list to be used in examine */
struct bdev_association {
        char                    *vbdev_name;
        char                    *bdev_name;
        uint64_t                avg_read_latency;
        uint64_t                p99_read_latency;
        uint64_t                avg_write_latency;
        uint64_t                p99_write_latency;
        TAILQ_ENTRY(bdev_association)   link;
};
static TAILQ_HEAD(, bdev_association) g_bdev_associations = TAILQ_HEAD_INITIALIZER(
                        g_bdev_associations);

/* List of virtual bdevs and associated info for each. */
struct vbdev_delay {
        struct spdk_bdev                *base_bdev; /* the thing we're attaching to */
        struct spdk_bdev_desc           *base_desc; /* its descriptor we get from open */
        struct spdk_bdev                delay_bdev;    /* the delay virtual bdev */
        uint64_t                        average_read_latency_ticks; /* the average read delay */
        uint64_t                        p99_read_latency_ticks; /* the p99 read delay */
        uint64_t                        average_write_latency_ticks; /* the average write delay */
        uint64_t                        p99_write_latency_ticks; /* the p99 write delay */
        TAILQ_ENTRY(vbdev_delay)        link;
        struct spdk_thread              *thread;    /* thread where base device is opened */
};
static TAILQ_HEAD(, vbdev_delay) g_delay_nodes = TAILQ_HEAD_INITIALIZER(g_delay_nodes);

struct delay_bdev_io {
        int status;

        uint64_t completion_tick;

        enum delay_io_type type;

        struct spdk_io_channel *ch;

        struct spdk_bdev_io_wait_entry bdev_io_wait;

        STAILQ_ENTRY(delay_bdev_io) link;
};

struct delay_io_channel {
        struct spdk_io_channel  *base_ch; /* IO channel of base device */
        STAILQ_HEAD(, delay_bdev_io) avg_read_io;
        STAILQ_HEAD(, delay_bdev_io) p99_read_io;
        STAILQ_HEAD(, delay_bdev_io) avg_write_io;
        STAILQ_HEAD(, delay_bdev_io) p99_write_io;
        struct spdk_poller *io_poller;
        unsigned int rand_seed;
};

static void
vbdev_delay_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io);


/* Callback for unregistering the IO device. */
static void
_device_unregister_cb(void *io_device)
{
        struct vbdev_delay *delay_node  = io_device;

        /* Done with this delay_node. */
        free(delay_node->delay_bdev.name);
        free(delay_node);
}

static void
_vbdev_delay_destruct(void *ctx)
{
        struct spdk_bdev_desc *desc = ctx;

        spdk_bdev_close(desc);
}

static int
vbdev_delay_destruct(void *ctx)
{
        struct vbdev_delay *delay_node = (struct vbdev_delay *)ctx;

        /* It is important to follow this exact sequence of steps for destroying
         * a vbdev...
         */

        TAILQ_REMOVE(&g_delay_nodes, delay_node, link);

        /* Unclaim the underlying bdev. */
        spdk_bdev_module_release_bdev(delay_node->base_bdev);

        /* Close the underlying bdev on its same opened thread. */
        if (delay_node->thread && delay_node->thread != spdk_get_thread()) {
                spdk_thread_send_msg(delay_node->thread, _vbdev_delay_destruct, delay_node->base_desc);
        } else {
                spdk_bdev_close(delay_node->base_desc);
        }

        /* Unregister the io_device. */
        spdk_io_device_unregister(delay_node, _device_unregister_cb);

        return 0;
}

static int
_process_io_stailq(void *arg, uint64_t ticks)
{
        STAILQ_HEAD(, delay_bdev_io) *head = arg;
        struct delay_bdev_io *io_ctx, *tmp;
        int completions = 0;

        STAILQ_FOREACH_SAFE(io_ctx, head, link, tmp) {
                if (io_ctx->completion_tick <= ticks) {
                        STAILQ_REMOVE(head, io_ctx, delay_bdev_io, link);
                        spdk_bdev_io_complete(spdk_bdev_io_from_ctx(io_ctx), io_ctx->status);
                        completions++;
                } else {
                        /* In the general case, I/O will become ready in an fifo order. When timeouts are dynamically
                         * changed, this is not necessarily the case. However, the normal behavior will be restored
                         * after the outstanding I/O at the time of the change have been completed.
                         * This essentially means that moving from a high to low latency creates a dam for the new I/O
                         * submitted after the latency change. This is considered desirable behavior for the use case where
                         * we are trying to trigger a pre-defined timeout on an initiator.
                         */
                        break;
                }
        }

        return completions;
}

static int
_delay_finish_io(void *arg)
{
        struct delay_io_channel *delay_ch = arg;
        uint64_t ticks = spdk_get_ticks();
        int completions = 0;

        completions += _process_io_stailq(&delay_ch->avg_read_io, ticks);
        completions += _process_io_stailq(&delay_ch->avg_write_io, ticks);
        completions += _process_io_stailq(&delay_ch->p99_read_io, ticks);
        completions += _process_io_stailq(&delay_ch->p99_write_io, ticks);

        return completions == 0 ? SPDK_POLLER_IDLE : SPDK_POLLER_BUSY;
}

/* Completion callback for IO that were issued from this bdev. The original bdev_io
 * is passed in as an arg so we'll complete that one with the appropriate status
 * and then free the one that this module issued.
 */
static void
_delay_complete_io(struct spdk_bdev_io *bdev_io, bool success, void *cb_arg)
{
        struct spdk_bdev_io *orig_io = cb_arg;
        struct vbdev_delay *delay_node = SPDK_CONTAINEROF(orig_io->bdev, struct vbdev_delay, delay_bdev);
        struct delay_bdev_io *io_ctx = (struct delay_bdev_io *)orig_io->driver_ctx;
        struct delay_io_channel *delay_ch = spdk_io_channel_get_ctx(io_ctx->ch);

        io_ctx->status = success ? SPDK_BDEV_IO_STATUS_SUCCESS : SPDK_BDEV_IO_STATUS_FAILED;
        spdk_bdev_free_io(bdev_io);

        /* Put the I/O into the proper list for processing by the channel poller. */
        switch (io_ctx->type) {
        case DELAY_AVG_READ:
                io_ctx->completion_tick = spdk_get_ticks() + delay_node->average_read_latency_ticks;
                STAILQ_INSERT_TAIL(&delay_ch->avg_read_io, io_ctx, link);
                break;
        case DELAY_AVG_WRITE:
                io_ctx->completion_tick = spdk_get_ticks() + delay_node->average_write_latency_ticks;
                STAILQ_INSERT_TAIL(&delay_ch->avg_write_io, io_ctx, link);
                break;
        case DELAY_P99_READ:
                io_ctx->completion_tick = spdk_get_ticks() + delay_node->p99_read_latency_ticks;
                STAILQ_INSERT_TAIL(&delay_ch->p99_read_io, io_ctx, link);
                break;
        case DELAY_P99_WRITE:
                io_ctx->completion_tick = spdk_get_ticks() + delay_node->p99_write_latency_ticks;
                STAILQ_INSERT_TAIL(&delay_ch->p99_write_io, io_ctx, link);
                break;
        case DELAY_NONE:
        default:
                spdk_bdev_io_complete(orig_io, io_ctx->status);
                break;
        }
}

static void
vbdev_delay_resubmit_io(void *arg)
{
        struct spdk_bdev_io *bdev_io = (struct spdk_bdev_io *)arg;
        struct delay_bdev_io *io_ctx = (struct delay_bdev_io *)bdev_io->driver_ctx;

        vbdev_delay_submit_request(io_ctx->ch, bdev_io);
}

static void
vbdev_delay_queue_io(struct spdk_bdev_io *bdev_io)
{
        struct delay_bdev_io *io_ctx = (struct delay_bdev_io *)bdev_io->driver_ctx;
        struct delay_io_channel *delay_ch = spdk_io_channel_get_ctx(io_ctx->ch);
        int rc;

        io_ctx->bdev_io_wait.bdev = bdev_io->bdev;
        io_ctx->bdev_io_wait.cb_fn = vbdev_delay_resubmit_io;
        io_ctx->bdev_io_wait.cb_arg = bdev_io;

        rc = spdk_bdev_queue_io_wait(bdev_io->bdev, delay_ch->base_ch, &io_ctx->bdev_io_wait);
        if (rc != 0) {
                SPDK_ERRLOG("Queue io failed in vbdev_delay_queue_io, rc=%d.\n", rc);
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        }
}

static void
delay_read_get_buf_cb(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io, bool success)
{
        struct vbdev_delay *delay_node = SPDK_CONTAINEROF(bdev_io->bdev, struct vbdev_delay,
                                         delay_bdev);
        struct delay_io_channel *delay_ch = spdk_io_channel_get_ctx(ch);
        int rc;

        if (!success) {
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
                return;
        }

        rc = spdk_bdev_readv_blocks(delay_node->base_desc, delay_ch->base_ch, bdev_io->u.bdev.iovs,
                                    bdev_io->u.bdev.iovcnt, bdev_io->u.bdev.offset_blocks,
                                    bdev_io->u.bdev.num_blocks, _delay_complete_io,
                                    bdev_io);

        if (rc == -ENOMEM) {
                SPDK_ERRLOG("No memory, start to queue io for delay.\n");
                vbdev_delay_queue_io(bdev_io);
        } else if (rc != 0) {
                SPDK_ERRLOG("ERROR on bdev_io submission!\n");
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        }
}

static void
vbdev_delay_reset_dev(struct spdk_io_channel_iter *i, int status)
{
        struct spdk_bdev_io *bdev_io = spdk_io_channel_iter_get_ctx(i);
        struct spdk_io_channel *ch = spdk_io_channel_iter_get_channel(i);
        struct delay_io_channel *delay_ch = spdk_io_channel_get_ctx(ch);
        struct vbdev_delay *delay_node = spdk_io_channel_iter_get_io_device(i);
        int rc;

        rc = spdk_bdev_reset(delay_node->base_desc, delay_ch->base_ch,
                             _delay_complete_io, bdev_io);

        if (rc == -ENOMEM) {
                SPDK_ERRLOG("No memory, start to queue io for delay.\n");
                vbdev_delay_queue_io(bdev_io);
        } else if (rc != 0) {
                SPDK_ERRLOG("ERROR on bdev_io submission!\n");
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        }
}

static void
_abort_all_delayed_io(void *arg)
{
        STAILQ_HEAD(, delay_bdev_io) *head = arg;
        struct delay_bdev_io *io_ctx, *tmp;

        STAILQ_FOREACH_SAFE(io_ctx, head, link, tmp) {
                STAILQ_REMOVE(head, io_ctx, delay_bdev_io, link);
                spdk_bdev_io_complete(spdk_bdev_io_from_ctx(io_ctx), SPDK_BDEV_IO_STATUS_ABORTED);
        }
}

static void
vbdev_delay_reset_channel(struct spdk_io_channel_iter *i)
{
        struct spdk_io_channel *ch = spdk_io_channel_iter_get_channel(i);
        struct delay_io_channel *delay_ch = spdk_io_channel_get_ctx(ch);

        _abort_all_delayed_io(&delay_ch->avg_read_io);
        _abort_all_delayed_io(&delay_ch->avg_write_io);
        _abort_all_delayed_io(&delay_ch->p99_read_io);
        _abort_all_delayed_io(&delay_ch->p99_write_io);

        spdk_for_each_channel_continue(i, 0);
}

static bool
abort_delayed_io(void *_head, struct spdk_bdev_io *bio_to_abort)
{
        STAILQ_HEAD(, delay_bdev_io) *head = _head;
        struct delay_bdev_io *io_ctx_to_abort = (struct delay_bdev_io *)bio_to_abort->driver_ctx;
        struct delay_bdev_io *io_ctx;

        STAILQ_FOREACH(io_ctx, head, link) {
                if (io_ctx == io_ctx_to_abort) {
                        STAILQ_REMOVE(head, io_ctx_to_abort, delay_bdev_io, link);
                        spdk_bdev_io_complete(bio_to_abort, SPDK_BDEV_IO_STATUS_ABORTED);
                        return true;
                }
        }

        return false;
}

static int
vbdev_delay_abort(struct vbdev_delay *delay_node, struct delay_io_channel *delay_ch,
                  struct spdk_bdev_io *bdev_io)
{
        struct spdk_bdev_io *bio_to_abort = bdev_io->u.abort.bio_to_abort;

        if (abort_delayed_io(&delay_ch->avg_read_io, bio_to_abort) ||
            abort_delayed_io(&delay_ch->avg_write_io, bio_to_abort) ||
            abort_delayed_io(&delay_ch->p99_read_io, bio_to_abort) ||
            abort_delayed_io(&delay_ch->p99_write_io, bio_to_abort)) {
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_SUCCESS);
                return 0;
        }

        return spdk_bdev_abort(delay_node->base_desc, delay_ch->base_ch, bio_to_abort,
                               _delay_complete_io, bdev_io);
}

static void
vbdev_delay_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io)
{
        struct vbdev_delay *delay_node = SPDK_CONTAINEROF(bdev_io->bdev, struct vbdev_delay, delay_bdev);
        struct delay_io_channel *delay_ch = spdk_io_channel_get_ctx(ch);
        struct delay_bdev_io *io_ctx = (struct delay_bdev_io *)bdev_io->driver_ctx;
        int rc = 0;
        bool is_p99;

        is_p99 = rand_r(&delay_ch->rand_seed) % 100 == 0 ? true : false;

        io_ctx->ch = ch;
        io_ctx->type = DELAY_NONE;

        switch (bdev_io->type) {
        case SPDK_BDEV_IO_TYPE_READ:
                io_ctx->type = is_p99 ? DELAY_P99_READ : DELAY_AVG_READ;
                spdk_bdev_io_get_buf(bdev_io, delay_read_get_buf_cb,
                                     bdev_io->u.bdev.num_blocks * bdev_io->bdev->blocklen);
                break;
        case SPDK_BDEV_IO_TYPE_WRITE:
                io_ctx->type = is_p99 ? DELAY_P99_WRITE : DELAY_AVG_WRITE;
                rc = spdk_bdev_writev_blocks(delay_node->base_desc, delay_ch->base_ch, bdev_io->u.bdev.iovs,
                                             bdev_io->u.bdev.iovcnt, bdev_io->u.bdev.offset_blocks,
                                             bdev_io->u.bdev.num_blocks, _delay_complete_io,
                                             bdev_io);
                break;
        case SPDK_BDEV_IO_TYPE_WRITE_ZEROES:
                rc = spdk_bdev_write_zeroes_blocks(delay_node->base_desc, delay_ch->base_ch,
                                                   bdev_io->u.bdev.offset_blocks,
                                                   bdev_io->u.bdev.num_blocks,
                                                   _delay_complete_io, bdev_io);
                break;
        case SPDK_BDEV_IO_TYPE_UNMAP:
                rc = spdk_bdev_unmap_blocks(delay_node->base_desc, delay_ch->base_ch,
                                            bdev_io->u.bdev.offset_blocks,
                                            bdev_io->u.bdev.num_blocks,
                                            _delay_complete_io, bdev_io);
                break;
        case SPDK_BDEV_IO_TYPE_FLUSH:
                rc = spdk_bdev_flush_blocks(delay_node->base_desc, delay_ch->base_ch,
                                            bdev_io->u.bdev.offset_blocks,
                                            bdev_io->u.bdev.num_blocks,
                                            _delay_complete_io, bdev_io);
                break;
        case SPDK_BDEV_IO_TYPE_RESET:
                /* During reset, the generic bdev layer aborts all new I/Os and queues all new resets.
                 * Hence we can simply abort all I/Os delayed to complete.
                 */
                spdk_for_each_channel(delay_node, vbdev_delay_reset_channel, bdev_io,
                                      vbdev_delay_reset_dev);
                break;
        case SPDK_BDEV_IO_TYPE_ABORT:
                rc = vbdev_delay_abort(delay_node, delay_ch, bdev_io);
                break;
        default:
                SPDK_ERRLOG("delay: unknown I/O type %d\n", bdev_io->type);
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
                return;
        }

        if (rc == -ENOMEM) {
                SPDK_ERRLOG("No memory, start to queue io for delay.\n");
                vbdev_delay_queue_io(bdev_io);
        } else if (rc != 0) {
                SPDK_ERRLOG("ERROR on bdev_io submission!\n");
                spdk_bdev_io_complete(bdev_io, SPDK_BDEV_IO_STATUS_FAILED);
        }
}

static bool
vbdev_delay_io_type_supported(void *ctx, enum spdk_bdev_io_type io_type)
{
        struct vbdev_delay *delay_node = (struct vbdev_delay *)ctx;

        if (io_type == SPDK_BDEV_IO_TYPE_ZCOPY) {
                return false;
        } else {
                return spdk_bdev_io_type_supported(delay_node->base_bdev, io_type);
        }
}

static struct spdk_io_channel *
vbdev_delay_get_io_channel(void *ctx)
{
        struct vbdev_delay *delay_node = (struct vbdev_delay *)ctx;
        struct spdk_io_channel *delay_ch = NULL;

        delay_ch = spdk_get_io_channel(delay_node);

        return delay_ch;
}

static void
_delay_write_conf_values(struct vbdev_delay *delay_node, struct spdk_json_write_ctx *w)
{
        spdk_json_write_named_string(w, "name", spdk_bdev_get_name(&delay_node->delay_bdev));
        spdk_json_write_named_string(w, "base_bdev_name", spdk_bdev_get_name(delay_node->base_bdev));
        spdk_json_write_named_int64(w, "avg_read_latency",
                                    delay_node->average_read_latency_ticks * SPDK_SEC_TO_USEC / spdk_get_ticks_hz());
        spdk_json_write_named_int64(w, "p99_read_latency",
                                    delay_node->p99_read_latency_ticks * SPDK_SEC_TO_USEC / spdk_get_ticks_hz());
        spdk_json_write_named_int64(w, "avg_write_latency",
                                    delay_node->average_write_latency_ticks * SPDK_SEC_TO_USEC / spdk_get_ticks_hz());
        spdk_json_write_named_int64(w, "p99_write_latency",
                                    delay_node->p99_write_latency_ticks * SPDK_SEC_TO_USEC / spdk_get_ticks_hz());
}

static int
vbdev_delay_dump_info_json(void *ctx, struct spdk_json_write_ctx *w)
{
        struct vbdev_delay *delay_node = (struct vbdev_delay *)ctx;

        spdk_json_write_name(w, "delay");
        spdk_json_write_object_begin(w);
        _delay_write_conf_values(delay_node, w);
        spdk_json_write_object_end(w);

        return 0;
}

/* This is used to generate JSON that can configure this module to its current state. */
static int
vbdev_delay_config_json(struct spdk_json_write_ctx *w)
{
        struct vbdev_delay *delay_node;

        TAILQ_FOREACH(delay_node, &g_delay_nodes, link) {
                spdk_json_write_object_begin(w);
                spdk_json_write_named_string(w, "method", "bdev_delay_create");
                spdk_json_write_named_object_begin(w, "params");
                _delay_write_conf_values(delay_node, w);
                spdk_json_write_object_end(w);
        }
        return 0;
}

/* We provide this callback for the SPDK channel code to create a channel using
 * the channel struct we provided in our module get_io_channel() entry point. Here
 * we get and save off an underlying base channel of the device below us so that
 * we can communicate with the base bdev on a per channel basis.  If we needed
 * our own poller for this vbdev, we'd register it here.
 */
static int
delay_bdev_ch_create_cb(void *io_device, void *ctx_buf)
{
        struct delay_io_channel *delay_ch = ctx_buf;
        struct vbdev_delay *delay_node = io_device;

        STAILQ_INIT(&delay_ch->avg_read_io);
        STAILQ_INIT(&delay_ch->p99_read_io);
        STAILQ_INIT(&delay_ch->avg_write_io);
        STAILQ_INIT(&delay_ch->p99_write_io);

        delay_ch->io_poller = SPDK_POLLER_REGISTER(_delay_finish_io, delay_ch, 0);
        delay_ch->base_ch = spdk_bdev_get_io_channel(delay_node->base_desc);
        delay_ch->rand_seed = time(NULL);

        return 0;
}

/* We provide this callback for the SPDK channel code to destroy a channel
 * created with our create callback. We just need to undo anything we did
 * when we created. If this bdev used its own poller, we'd unregsiter it here.
 */
static void
delay_bdev_ch_destroy_cb(void *io_device, void *ctx_buf)
{
        struct delay_io_channel *delay_ch = ctx_buf;

        spdk_poller_unregister(&delay_ch->io_poller);
        spdk_put_io_channel(delay_ch->base_ch);
}

/* Create the delay association from the bdev and vbdev name and insert
 * on the global list. */
static int
vbdev_delay_insert_association(const char *bdev_name, const char *vbdev_name,
                               uint64_t avg_read_latency, uint64_t p99_read_latency,
                               uint64_t avg_write_latency, uint64_t p99_write_latency)
{
        struct bdev_association *assoc;

        TAILQ_FOREACH(assoc, &g_bdev_associations, link) {
                if (strcmp(vbdev_name, assoc->vbdev_name) == 0) {
                        SPDK_ERRLOG("delay bdev %s already exists\n", vbdev_name);
                        return -EEXIST;
                }
        }

        assoc = calloc(1, sizeof(struct bdev_association));
        if (!assoc) {
                SPDK_ERRLOG("could not allocate bdev_association\n");
                return -ENOMEM;
        }

        assoc->bdev_name = strdup(bdev_name);
        if (!assoc->bdev_name) {
                SPDK_ERRLOG("could not allocate assoc->bdev_name\n");
                free(assoc);
                return -ENOMEM;
        }

        assoc->vbdev_name = strdup(vbdev_name);
        if (!assoc->vbdev_name) {
                SPDK_ERRLOG("could not allocate assoc->vbdev_name\n");
                free(assoc->bdev_name);
                free(assoc);
                return -ENOMEM;
        }

        assoc->avg_read_latency = avg_read_latency;
        assoc->p99_read_latency = p99_read_latency;
        assoc->avg_write_latency = avg_write_latency;
        assoc->p99_write_latency = p99_write_latency;

        TAILQ_INSERT_TAIL(&g_bdev_associations, assoc, link);

        return 0;
}

int
vbdev_delay_update_latency_value(char *delay_name, uint64_t latency_us, enum delay_io_type type)
{
        struct spdk_bdev *delay_bdev;
        struct vbdev_delay *delay_node;
        uint64_t ticks_mhz = spdk_get_ticks_hz() / SPDK_SEC_TO_USEC;

        delay_bdev = spdk_bdev_get_by_name(delay_name);
        if (delay_bdev == NULL) {
                return -ENODEV;
        } else if (delay_bdev->module != &delay_if) {
                return -EINVAL;
        }

        delay_node = SPDK_CONTAINEROF(delay_bdev, struct vbdev_delay, delay_bdev);

        switch (type) {
        case DELAY_AVG_READ:
                delay_node->average_read_latency_ticks = ticks_mhz * latency_us;
                break;
        case DELAY_AVG_WRITE:
                delay_node->average_write_latency_ticks = ticks_mhz * latency_us;
                break;
        case DELAY_P99_READ:
                delay_node->p99_read_latency_ticks = ticks_mhz * latency_us;
                break;
        case DELAY_P99_WRITE:
                delay_node->p99_write_latency_ticks = ticks_mhz * latency_us;
                break;
        default:
                return -EINVAL;
        }

        return 0;
}

static int
vbdev_delay_init(void)
{
        /* Not allowing for .ini style configuration. */
        return 0;
}

static void
vbdev_delay_finish(void)
{
        struct bdev_association *assoc;

        while ((assoc = TAILQ_FIRST(&g_bdev_associations))) {
                TAILQ_REMOVE(&g_bdev_associations, assoc, link);
                free(assoc->bdev_name);
                free(assoc->vbdev_name);
                free(assoc);
        }
}

static int
vbdev_delay_get_ctx_size(void)
{
        return sizeof(struct delay_bdev_io);
}

static void
vbdev_delay_write_config_json(struct spdk_bdev *bdev, struct spdk_json_write_ctx *w)
{
        /* No config per bdev needed */
}

/* When we register our bdev this is how we specify our entry points. */
static const struct spdk_bdev_fn_table vbdev_delay_fn_table = {
        .destruct               = vbdev_delay_destruct,
        .submit_request         = vbdev_delay_submit_request,
        .io_type_supported      = vbdev_delay_io_type_supported,
        .get_io_channel         = vbdev_delay_get_io_channel,
        .dump_info_json         = vbdev_delay_dump_info_json,
        .write_config_json      = vbdev_delay_write_config_json,
};

/* Called when the underlying base bdev goes away. */
static void
vbdev_delay_base_bdev_hotremove_cb(void *ctx)
{
        struct vbdev_delay *delay_node, *tmp;
        struct spdk_bdev *bdev_find = ctx;

        TAILQ_FOREACH_SAFE(delay_node, &g_delay_nodes, link, tmp) {
                if (bdev_find == delay_node->base_bdev) {
                        spdk_bdev_unregister(&delay_node->delay_bdev, NULL, NULL);
                }
        }
}

/* Create and register the delay vbdev if we find it in our list of bdev names.
 * This can be called either by the examine path or RPC method.
 */
static int
vbdev_delay_register(struct spdk_bdev *bdev)
{
        struct bdev_association *assoc;
        struct vbdev_delay *delay_node;
        uint64_t ticks_mhz = spdk_get_ticks_hz() / SPDK_SEC_TO_USEC;
        int rc = 0;

        /* Check our list of names from config versus this bdev and if
         * there's a match, create the delay_node & bdev accordingly.
         */
        TAILQ_FOREACH(assoc, &g_bdev_associations, link) {
                if (strcmp(assoc->bdev_name, bdev->name) != 0) {
                        continue;
                }

                delay_node = calloc(1, sizeof(struct vbdev_delay));
                if (!delay_node) {
                        rc = -ENOMEM;
                        SPDK_ERRLOG("could not allocate delay_node\n");
                        break;
                }

                /* The base bdev that we're attaching to. */
                delay_node->base_bdev = bdev;
                delay_node->delay_bdev.name = strdup(assoc->vbdev_name);
                if (!delay_node->delay_bdev.name) {
                        rc = -ENOMEM;
                        SPDK_ERRLOG("could not allocate delay_bdev name\n");
                        free(delay_node);
                        break;
                }
                delay_node->delay_bdev.product_name = "delay";

                delay_node->delay_bdev.write_cache = bdev->write_cache;
                delay_node->delay_bdev.required_alignment = bdev->required_alignment;
                delay_node->delay_bdev.optimal_io_boundary = bdev->optimal_io_boundary;
                delay_node->delay_bdev.blocklen = bdev->blocklen;
                delay_node->delay_bdev.blockcnt = bdev->blockcnt;

                delay_node->delay_bdev.ctxt = delay_node;
                delay_node->delay_bdev.fn_table = &vbdev_delay_fn_table;
                delay_node->delay_bdev.module = &delay_if;

                /* Store the number of ticks you need to add to get the I/O expiration time. */
                delay_node->average_read_latency_ticks = ticks_mhz * assoc->avg_read_latency;
                delay_node->p99_read_latency_ticks = ticks_mhz * assoc->p99_read_latency;
                delay_node->average_write_latency_ticks = ticks_mhz * assoc->avg_write_latency;
                delay_node->p99_write_latency_ticks = ticks_mhz * assoc->p99_write_latency;

                spdk_io_device_register(delay_node, delay_bdev_ch_create_cb, delay_bdev_ch_destroy_cb,
                                        sizeof(struct delay_io_channel),
                                        assoc->vbdev_name);

                rc = spdk_bdev_open(bdev, true, vbdev_delay_base_bdev_hotremove_cb,
                                    bdev, &delay_node->base_desc);
                if (rc) {
                        SPDK_ERRLOG("could not open bdev %s\n", spdk_bdev_get_name(bdev));
                        goto error_unregister;
                }

                /* Save the thread where the base device is opened */
                delay_node->thread = spdk_get_thread();

                rc = spdk_bdev_module_claim_bdev(bdev, delay_node->base_desc, delay_node->delay_bdev.module);
                if (rc) {
                        SPDK_ERRLOG("could not claim bdev %s\n", spdk_bdev_get_name(bdev));
                        goto error_close;
                }

                rc = spdk_bdev_register(&delay_node->delay_bdev);
                if (rc) {
                        SPDK_ERRLOG("could not register delay_bdev\n");
                        spdk_bdev_module_release_bdev(delay_node->base_bdev);
                        goto error_close;
                }

                TAILQ_INSERT_TAIL(&g_delay_nodes, delay_node, link);
        }

        return rc;

error_close:
        spdk_bdev_close(delay_node->base_desc);
error_unregister:
        spdk_io_device_unregister(delay_node, NULL);
        free(delay_node->delay_bdev.name);
        free(delay_node);
        return rc;
}

int
create_delay_disk(const char *bdev_name, const char *vbdev_name, uint64_t avg_read_latency,
                  uint64_t p99_read_latency, uint64_t avg_write_latency, uint64_t p99_write_latency)
{
        struct spdk_bdev *bdev = NULL;
        int rc = 0;

        if (p99_read_latency < avg_read_latency || p99_write_latency < avg_write_latency) {
                SPDK_ERRLOG("Unable to create a delay bdev where p99 latency is less than average latency.\n");
                return -EINVAL;
        }

        rc = vbdev_delay_insert_association(bdev_name, vbdev_name, avg_read_latency, p99_read_latency,
                                            avg_write_latency, p99_write_latency);
        if (rc) {
                return rc;
        }

        bdev = spdk_bdev_get_by_name(bdev_name);
        if (!bdev) {
                return 0;
        }

        return vbdev_delay_register(bdev);
}

void
delete_delay_disk(struct spdk_bdev *bdev, spdk_bdev_unregister_cb cb_fn, void *cb_arg)
{
        struct bdev_association *assoc;

        if (!bdev || bdev->module != &delay_if) {
                cb_fn(cb_arg, -ENODEV);
                return;
        }

        TAILQ_FOREACH(assoc, &g_bdev_associations, link) {
                if (strcmp(assoc->vbdev_name, bdev->name) == 0) {
                        TAILQ_REMOVE(&g_bdev_associations, assoc, link);
                        free(assoc->bdev_name);
                        free(assoc->vbdev_name);
                        free(assoc);
                        break;
                }
        }

        spdk_bdev_unregister(bdev, cb_fn, cb_arg);
}

static void
vbdev_delay_examine(struct spdk_bdev *bdev)
{
        vbdev_delay_register(bdev);

        spdk_bdev_module_examine_done(&delay_if);
}

SPDK_LOG_REGISTER_COMPONENT("vbdev_delay", SPDK_LOG_VBDEV_DELAY)