import 15.2.0 Octopus source

[ceph.git] / ceph / src / spdk / lib / ftl / ftl_rwb.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 "spdk/env.h"
#include "spdk/util.h"

#include "ftl_rwb.h"
#include "ftl_core.h"

struct ftl_rwb_batch {
        /* Parent RWB */
        struct ftl_rwb                          *rwb;

        /* Position within RWB */
        unsigned int                            pos;

        /* Number of acquired entries */
        unsigned int                            num_acquired;

        /* Number of entries ready for submission */
        unsigned int                            num_ready;

        /* RWB entry list */
        LIST_HEAD(, ftl_rwb_entry)              entry_list;

        /* Entry buffer */
        struct ftl_rwb_entry                    *entries;

        /* Data buffer */
        void                                    *buffer;

        /* Metadata buffer */
        void                                    *md_buffer;

        /* Queue entry */
        STAILQ_ENTRY(ftl_rwb_batch)             stailq;
};

struct ftl_rwb {
        /* Number of batches */
        size_t                                  num_batches;

        /* Information for interleaving */
        size_t                                  interleave_offset;
        /* Maximum number of active batches */
        size_t                                  max_active_batches;

        /* Number of entries per batch */
        size_t                                  xfer_size;
        /* Metadata's size */
        size_t                                  md_size;

        /* Number of acquired entries */
        unsigned int                            num_acquired[FTL_RWB_TYPE_MAX];
        /* User/internal limits */
        size_t                                  limits[FTL_RWB_TYPE_MAX];

        /* Active batch queue */
        STAILQ_HEAD(, ftl_rwb_batch)            active_queue;
        /* Number of active batches */
        unsigned int                            num_active_batches;

        /* Free batch queue */
        STAILQ_HEAD(, ftl_rwb_batch)            free_queue;
        /* Number of active batches */
        unsigned int                            num_free_batches;

        /* Submission batch queue */
        struct spdk_ring                        *submit_queue;
        /* High-priority batch queue */
        struct spdk_ring                        *prio_queue;

        /* Batch buffer */
        struct ftl_rwb_batch                    *batches;

        /* RWB lock */
        pthread_spinlock_t                      lock;
};

static int
ftl_rwb_batch_full(const struct ftl_rwb_batch *batch, size_t batch_size)
{
        struct ftl_rwb *rwb = batch->rwb;
        assert(batch_size <= rwb->xfer_size);
        return batch_size == rwb->xfer_size;
}

static int
ftl_rwb_batch_init_entry(struct ftl_rwb_batch *batch, size_t pos)
{
        struct ftl_rwb *rwb = batch->rwb;
        struct ftl_rwb_entry *entry, *prev;
        size_t batch_offset = pos % rwb->xfer_size;

        entry = &batch->entries[batch_offset];
        entry->pos = pos;
        entry->data = ((char *)batch->buffer) + FTL_BLOCK_SIZE * batch_offset;
        entry->md = rwb->md_size ? ((char *)batch->md_buffer) + rwb->md_size * batch_offset : NULL;
        entry->batch = batch;
        entry->rwb = batch->rwb;

        if (pthread_spin_init(&entry->lock, PTHREAD_PROCESS_PRIVATE)) {
                SPDK_ERRLOG("Spinlock initialization failure\n");
                return -1;
        }

        if (batch_offset > 0) {
                prev = &batch->entries[batch_offset - 1];
                LIST_INSERT_AFTER(prev, entry, list_entry);
        } else {
                LIST_INSERT_HEAD(&batch->entry_list, entry, list_entry);
        }

        return 0;
}

static int
ftl_rwb_batch_init(struct ftl_rwb *rwb, struct ftl_rwb_batch *batch, unsigned int pos)
{
        size_t md_size, i;

        md_size = rwb->md_size * rwb->xfer_size;
        batch->rwb = rwb;
        batch->pos = pos;

        batch->entries = calloc(rwb->xfer_size, sizeof(*batch->entries));
        if (!batch->entries) {
                return -1;
        }

        LIST_INIT(&batch->entry_list);

        batch->buffer = spdk_dma_zmalloc(FTL_BLOCK_SIZE * rwb->xfer_size,
                                         FTL_BLOCK_SIZE, NULL);
        if (!batch->buffer) {
                return -1;
        }

        if (md_size > 0) {
                batch->md_buffer = spdk_dma_zmalloc(md_size, 0, NULL);
                if (!batch->md_buffer) {
                        return -1;
                }
        }

        for (i = 0; i < rwb->xfer_size; ++i) {
                if (ftl_rwb_batch_init_entry(batch, pos * rwb->xfer_size + i)) {
                        return -1;
                }
        }

        return 0;
}

struct ftl_rwb *
ftl_rwb_init(const struct spdk_ftl_conf *conf, size_t xfer_size, size_t md_size, size_t num_punits)
{
        struct ftl_rwb *rwb = NULL;
        struct ftl_rwb_batch *batch;
        size_t i;

        rwb = calloc(1, sizeof(*rwb));
        if (!rwb) {
                SPDK_ERRLOG("Memory allocation failure\n");
                return NULL;
        }

        if (pthread_spin_init(&rwb->lock, PTHREAD_PROCESS_PRIVATE)) {
                SPDK_ERRLOG("Spinlock initialization failure\n");
                free(rwb);
                return NULL;
        }

        assert(conf->rwb_size % xfer_size == 0);
        rwb->xfer_size = xfer_size;
        rwb->interleave_offset = xfer_size / conf->num_interleave_units;
        rwb->max_active_batches = conf->num_interleave_units == 1 ? 1 : num_punits;
        rwb->md_size = md_size;
        rwb->num_batches = conf->rwb_size / (FTL_BLOCK_SIZE * xfer_size) + rwb->max_active_batches;

        rwb->batches = calloc(rwb->num_batches, sizeof(*rwb->batches));
        if (!rwb->batches) {
                goto error;
        }

        rwb->submit_queue = spdk_ring_create(SPDK_RING_TYPE_MP_SC,
                                             spdk_align32pow2(rwb->num_batches + 1),
                                             SPDK_ENV_SOCKET_ID_ANY);
        if (!rwb->submit_queue) {
                SPDK_ERRLOG("Failed to create submission queue\n");
                goto error;
        }

        rwb->prio_queue = spdk_ring_create(SPDK_RING_TYPE_MP_SC,
                                           spdk_align32pow2(rwb->num_batches + 1),
                                           SPDK_ENV_SOCKET_ID_ANY);
        if (!rwb->prio_queue) {
                SPDK_ERRLOG("Failed to create high-prio submission queue\n");
                goto error;
        }

        STAILQ_INIT(&rwb->free_queue);
        STAILQ_INIT(&rwb->active_queue);

        for (i = 0; i < rwb->num_batches; ++i) {
                batch = &rwb->batches[i];

                if (ftl_rwb_batch_init(rwb, batch, i)) {
                        SPDK_ERRLOG("Failed to initialize RWB entry buffer\n");
                        goto error;
                }

                STAILQ_INSERT_TAIL(&rwb->free_queue, batch, stailq);
                rwb->num_free_batches++;
        }

        for (unsigned int i = 0; i < FTL_RWB_TYPE_MAX; ++i) {
                rwb->limits[i] = ftl_rwb_entry_cnt(rwb);
        }

        return rwb;
error:
        ftl_rwb_free(rwb);
        return NULL;
}

void
ftl_rwb_free(struct ftl_rwb *rwb)
{
        struct ftl_rwb_entry *entry;
        struct ftl_rwb_batch *batch;

        if (!rwb) {
                return;
        }

        if (rwb->batches) {
                for (size_t i = 0; i < rwb->num_batches; ++i) {
                        batch = &rwb->batches[i];

                        if (batch->entries) {
                                ftl_rwb_foreach(entry, batch) {
                                        pthread_spin_destroy(&entry->lock);
                                }

                                free(batch->entries);
                        }

                        spdk_dma_free(batch->buffer);
                        spdk_dma_free(batch->md_buffer);
                }
        }

        pthread_spin_destroy(&rwb->lock);
        spdk_ring_free(rwb->submit_queue);
        spdk_ring_free(rwb->prio_queue);
        free(rwb->batches);
        free(rwb);
}

void
ftl_rwb_batch_release(struct ftl_rwb_batch *batch)
{
        struct ftl_rwb *rwb = batch->rwb;
        struct ftl_rwb_entry *entry;
        unsigned int num_acquired __attribute__((unused));

        batch->num_ready = 0;
        batch->num_acquired = 0;

        ftl_rwb_foreach(entry, batch) {
                num_acquired = __atomic_fetch_sub(&rwb->num_acquired[ftl_rwb_entry_type(entry)], 1,
                                                  __ATOMIC_SEQ_CST);
                assert(num_acquired  > 0);
        }

        pthread_spin_lock(&rwb->lock);
        STAILQ_INSERT_TAIL(&rwb->free_queue, batch, stailq);
        rwb->num_free_batches++;
        pthread_spin_unlock(&rwb->lock);
}

size_t
ftl_rwb_entry_cnt(const struct ftl_rwb *rwb)
{
        return rwb->num_batches * rwb->xfer_size;
}

size_t
ftl_rwb_num_batches(const struct ftl_rwb *rwb)
{
        return rwb->num_batches;
}

size_t
ftl_rwb_size(const struct ftl_rwb *rwb)
{
        return rwb->num_batches * rwb->xfer_size;
}

size_t
ftl_rwb_batch_get_offset(const struct ftl_rwb_batch *batch)
{
        return batch->pos;
}

void
ftl_rwb_set_limits(struct ftl_rwb *rwb,
                   const size_t limit[FTL_RWB_TYPE_MAX])
{
        assert(limit[FTL_RWB_TYPE_USER] <= ftl_rwb_entry_cnt(rwb));
        assert(limit[FTL_RWB_TYPE_INTERNAL] <= ftl_rwb_entry_cnt(rwb));
        memcpy(rwb->limits, limit, sizeof(rwb->limits));
}

void
ftl_rwb_get_limits(struct ftl_rwb *rwb,
                   size_t limit[FTL_RWB_TYPE_MAX])
{
        memcpy(limit, rwb->limits, sizeof(rwb->limits));
}

size_t
ftl_rwb_num_acquired(struct ftl_rwb *rwb, enum ftl_rwb_entry_type type)
{
        return __atomic_load_n(&rwb->num_acquired[type], __ATOMIC_SEQ_CST);
}

size_t
ftl_rwb_get_active_batches(const struct ftl_rwb *rwb)
{
        return rwb->num_active_batches;
}

void
ftl_rwb_batch_revert(struct ftl_rwb_batch *batch)
{
        struct ftl_rwb *rwb = batch->rwb;

        if (spdk_ring_enqueue(rwb->prio_queue, (void **)&batch, 1, NULL) != 1) {
                assert(0 && "Should never happen");
        }
}

void
ftl_rwb_push(struct ftl_rwb_entry *entry)
{
        struct ftl_rwb_batch *batch = entry->batch;
        struct ftl_rwb *rwb = batch->rwb;
        size_t batch_size;

        batch_size = __atomic_fetch_add(&batch->num_ready, 1, __ATOMIC_SEQ_CST) + 1;

        /* Once all of the entries are put back, push the batch on the */
        /* submission queue */
        if (ftl_rwb_batch_full(batch, batch_size)) {
                if (spdk_ring_enqueue(rwb->submit_queue, (void **)&batch, 1, NULL) != 1) {
                        assert(0 && "Should never happen");
                }
        }
}

static int
ftl_rwb_check_limits(struct ftl_rwb *rwb, enum ftl_rwb_entry_type type)
{
        return ftl_rwb_num_acquired(rwb, type) >= rwb->limits[type];
}

static struct ftl_rwb_batch *
_ftl_rwb_acquire_batch(struct ftl_rwb *rwb)
{
        struct ftl_rwb_batch *batch;
        size_t i;

        if (rwb->num_free_batches < rwb->max_active_batches) {
                return NULL;
        }

        for (i = 0; i < rwb->max_active_batches; i++) {
                batch = STAILQ_FIRST(&rwb->free_queue);
                STAILQ_REMOVE(&rwb->free_queue, batch, ftl_rwb_batch, stailq);
                rwb->num_free_batches--;

                STAILQ_INSERT_TAIL(&rwb->active_queue, batch, stailq);
                rwb->num_active_batches++;
        }

        return STAILQ_FIRST(&rwb->active_queue);
}

struct ftl_rwb_entry *
ftl_rwb_acquire(struct ftl_rwb *rwb, enum ftl_rwb_entry_type type)
{
        struct ftl_rwb_entry *entry = NULL;
        struct ftl_rwb_batch *current;

        if (ftl_rwb_check_limits(rwb, type)) {
                return NULL;
        }

        pthread_spin_lock(&rwb->lock);

        current = STAILQ_FIRST(&rwb->active_queue);
        if (!current) {
                current = _ftl_rwb_acquire_batch(rwb);
                if (!current) {
                        goto error;
                }
        }

        entry = &current->entries[current->num_acquired++];

        if (current->num_acquired >= rwb->xfer_size) {
                /* If the whole batch is filled, */
                /* remove the current batch from active_queue */
                /* since it will need to move to submit_queue */
                STAILQ_REMOVE(&rwb->active_queue, current, ftl_rwb_batch, stailq);
                rwb->num_active_batches--;
        } else if (current->num_acquired % rwb->interleave_offset == 0) {
                /* If the current batch is filled by the interleaving offset, */
                /* move the current batch at the tail of active_queue */
                /* to place the next logical blocks into another batch. */
                STAILQ_REMOVE(&rwb->active_queue, current, ftl_rwb_batch, stailq);
                STAILQ_INSERT_TAIL(&rwb->active_queue, current, stailq);
        }

        pthread_spin_unlock(&rwb->lock);
        __atomic_fetch_add(&rwb->num_acquired[type], 1, __ATOMIC_SEQ_CST);
        return entry;
error:
        pthread_spin_unlock(&rwb->lock);
        return NULL;
}

void
ftl_rwb_disable_interleaving(struct ftl_rwb *rwb)
{
        struct ftl_rwb_batch *batch, *temp;

        pthread_spin_lock(&rwb->lock);
        rwb->max_active_batches = 1;
        rwb->interleave_offset = rwb->xfer_size;

        STAILQ_FOREACH_SAFE(batch, &rwb->active_queue, stailq, temp) {
                if (batch->num_acquired == 0) {
                        STAILQ_REMOVE(&rwb->active_queue, batch, ftl_rwb_batch, stailq);
                        rwb->num_active_batches--;

                        assert(batch->num_ready == 0);
                        assert(batch->num_acquired == 0);

                        STAILQ_INSERT_TAIL(&rwb->free_queue, batch, stailq);
                        rwb->num_free_batches++;
                }
        }
        pthread_spin_unlock(&rwb->lock);
}

struct ftl_rwb_batch *
ftl_rwb_pop(struct ftl_rwb *rwb)
{
        struct ftl_rwb_batch *batch = NULL;

        if (spdk_ring_dequeue(rwb->prio_queue, (void **)&batch, 1) == 1) {
                return batch;
        }

        if (spdk_ring_dequeue(rwb->submit_queue, (void **)&batch, 1) == 1) {
                return batch;
        }

        return NULL;
}

static struct ftl_rwb_batch *
_ftl_rwb_next_batch(struct ftl_rwb *rwb, size_t pos)
{
        if (pos >= rwb->num_batches) {
                return NULL;
        }

        return &rwb->batches[pos];
}

struct ftl_rwb_batch *
ftl_rwb_next_batch(struct ftl_rwb_batch *batch)
{
        return _ftl_rwb_next_batch(batch->rwb, batch->pos + 1);
}

struct ftl_rwb_batch *
ftl_rwb_first_batch(struct ftl_rwb *rwb)
{
        return _ftl_rwb_next_batch(rwb, 0);
}

int
ftl_rwb_batch_empty(struct ftl_rwb_batch *batch)
{
        return __atomic_load_n(&batch->num_ready, __ATOMIC_SEQ_CST) == 0;
}

void *
ftl_rwb_batch_get_data(struct ftl_rwb_batch *batch)
{
        return batch->buffer;
}

void *
ftl_rwb_batch_get_md(struct ftl_rwb_batch *batch)
{
        return batch->md_buffer;
}

struct ftl_rwb_entry *
ftl_rwb_entry_from_offset(struct ftl_rwb *rwb, size_t offset)
{
        unsigned int b_off, e_off;

        b_off = offset / rwb->xfer_size;
        e_off = offset % rwb->xfer_size;

        assert(b_off < rwb->num_batches);

        return &rwb->batches[b_off].entries[e_off];
}

struct ftl_rwb_entry *
ftl_rwb_batch_first_entry(struct ftl_rwb_batch *batch)
{
        return LIST_FIRST(&batch->entry_list);
}