From 1554a344c92957fa2b08dd22c11479f9808916c7 Mon Sep 17 00:00:00 2001 From: Shuhei Matsumoto Date: Tue, 4 Jun 2019 13:22:00 +0900 Subject: [PATCH] env: Add free_space parameter to spdk_ring_enqueue DPDK rte_ring_enqueue_bulk() has free_space parameter to return the amount of space in the ring after enqueue operation has finished. This parameter can be used to wait when the ring is almost full and wake up when there is enough space available in the ring. Hence we add free_space to spdk_ring_enqueue() and spdk_ring_enqueue() passes it to rte_ring_enqueue_bulk() simply. Signed-off-by: Shuhei Matsumoto Change-Id: I9b9d6a5a097cf6dc4b97dfda7442f2c4b0aed4d3 Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/456734 Tested-by: SPDK CI Jenkins Reviewed-by: Jim Harris Reviewed-by: Darek Stojaczyk --- CHANGELOG.md | 6 ++++++ include/spdk/env.h | 4 +++- lib/bdev/nvme/bdev_ftl.c | 2 +- lib/bdev/virtio/bdev_virtio_scsi.c | 2 +- lib/env_dpdk/env.c | 6 ++++-- lib/event/reactor.c | 2 +- lib/ftl/ftl_reloc.c | 8 ++++---- lib/ftl/ftl_rwb.c | 4 ++-- lib/thread/thread.c | 2 +- test/common/lib/test_env.c | 3 ++- 10 files changed, 25 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 36a181c3c..a9c24d253 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,12 @@ taken into account if trying to achieve an artificial latency on top of an nvme Added spdk_nvme_ctrlr_get_transport_id() to get the transport ID from a previously attached controller. +### env + +The parameter `free_space` has been added to spdk_ring_enqueue() to wait when +the ring is almost full and resume when there is enough space available in +the ring. + ## v19.04: ### nvme diff --git a/include/spdk/env.h b/include/spdk/env.h index 0203c8d53..25f1542a9 100644 --- a/include/spdk/env.h +++ b/include/spdk/env.h @@ -582,10 +582,12 @@ size_t spdk_ring_count(struct spdk_ring *ring); * \param ring A pointer to the ring. * \param objs A pointer to the array to be queued. * \param count Length count of the array of objects. + * \param free_space If non-NULL, amount of free space after the enqueue has finished. * * \return the number of objects enqueued. */ -size_t spdk_ring_enqueue(struct spdk_ring *ring, void **objs, size_t count); +size_t spdk_ring_enqueue(struct spdk_ring *ring, void **objs, size_t count, + size_t *free_space); /** * Dequeue count objects from the ring into the array objs. diff --git a/lib/bdev/nvme/bdev_ftl.c b/lib/bdev/nvme/bdev_ftl.c index 4acec500c..32657320d 100644 --- a/lib/bdev/nvme/bdev_ftl.c +++ b/lib/bdev/nvme/bdev_ftl.c @@ -247,7 +247,7 @@ bdev_ftl_cb(void *arg, int status) io->status = status; - cnt = spdk_ring_enqueue(io->ring, (void **)&io, 1); + cnt = spdk_ring_enqueue(io->ring, (void **)&io, 1, NULL); assert(cnt == 1); } diff --git a/lib/bdev/virtio/bdev_virtio_scsi.c b/lib/bdev/virtio/bdev_virtio_scsi.c index cf555f03f..b770a19b6 100644 --- a/lib/bdev/virtio/bdev_virtio_scsi.c +++ b/lib/bdev/virtio/bdev_virtio_scsi.c @@ -570,7 +570,7 @@ bdev_virtio_reset(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io) tmf_req->type = VIRTIO_SCSI_T_TMF; tmf_req->subtype = VIRTIO_SCSI_T_TMF_LOGICAL_UNIT_RESET; - enqueued_count = spdk_ring_enqueue(svdev->ctrlq_ring, (void **)&bdev_io, 1); + enqueued_count = spdk_ring_enqueue(svdev->ctrlq_ring, (void **)&bdev_io, 1, NULL); if (spdk_likely(enqueued_count == 1)) { return; } else { diff --git a/lib/env_dpdk/env.c b/lib/env_dpdk/env.c index 073881583..0421ade73 100644 --- a/lib/env_dpdk/env.c +++ b/lib/env_dpdk/env.c @@ -393,9 +393,11 @@ spdk_ring_count(struct spdk_ring *ring) } size_t -spdk_ring_enqueue(struct spdk_ring *ring, void **objs, size_t count) +spdk_ring_enqueue(struct spdk_ring *ring, void **objs, size_t count, + size_t *free_space) { - return rte_ring_enqueue_bulk((struct rte_ring *)ring, objs, count, NULL); + return rte_ring_enqueue_bulk((struct rte_ring *)ring, objs, count, + (unsigned int *)free_space); } size_t diff --git a/lib/event/reactor.c b/lib/event/reactor.c index eae3d6d42..e93b9c904 100644 --- a/lib/event/reactor.c +++ b/lib/event/reactor.c @@ -135,7 +135,7 @@ spdk_event_call(struct spdk_event *event) reactor = spdk_reactor_get(event->lcore); assert(reactor->events != NULL); - rc = spdk_ring_enqueue(reactor->events, (void **)&event, 1); + rc = spdk_ring_enqueue(reactor->events, (void **)&event, 1, NULL); if (rc != 1) { assert(false); } diff --git a/lib/ftl/ftl_reloc.c b/lib/ftl/ftl_reloc.c index d00e8b5df..8395cce39 100644 --- a/lib/ftl/ftl_reloc.c +++ b/lib/ftl/ftl_reloc.c @@ -158,7 +158,7 @@ _ftl_reloc_prep(struct ftl_band_reloc *breloc) for (i = 0; i < reloc->max_qdepth; ++i) { io = ftl_io_alloc(dev->ioch); - spdk_ring_enqueue(breloc->free_queue, (void **)&io, 1); + spdk_ring_enqueue(breloc->free_queue, (void **)&io, 1, NULL); } } @@ -215,7 +215,7 @@ ftl_reloc_free_io(struct ftl_band_reloc *breloc, struct ftl_io *io) { spdk_dma_free(io->iov[0].iov_base); free(io->lba.vector); - spdk_ring_enqueue(breloc->free_queue, (void **)&io, 1); + spdk_ring_enqueue(breloc->free_queue, (void **)&io, 1, NULL); } static void @@ -257,7 +257,7 @@ ftl_reloc_read_cb(void *arg, int status) } io->flags &= ~FTL_IO_INITIALIZED; - spdk_ring_enqueue(breloc->write_queue, (void **)&io, 1); + spdk_ring_enqueue(breloc->write_queue, (void **)&io, 1, NULL); } static void @@ -465,7 +465,7 @@ ftl_reloc_read(struct ftl_band_reloc *breloc, struct ftl_io *io) num_lbks = ftl_reloc_next_lbks(breloc, &ppa); if (!num_lbks) { - spdk_ring_enqueue(breloc->free_queue, (void **)&io, 1); + spdk_ring_enqueue(breloc->free_queue, (void **)&io, 1, NULL); return 0; } diff --git a/lib/ftl/ftl_rwb.c b/lib/ftl/ftl_rwb.c index 58e1d4392..0af06525f 100644 --- a/lib/ftl/ftl_rwb.c +++ b/lib/ftl/ftl_rwb.c @@ -368,7 +368,7 @@ 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) != 1) { + if (spdk_ring_enqueue(rwb->prio_queue, (void **)&batch, 1, NULL) != 1) { assert(0 && "Should never happen"); } } @@ -385,7 +385,7 @@ ftl_rwb_push(struct ftl_rwb_entry *entry) /* 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) != 1) { + if (spdk_ring_enqueue(rwb->submit_queue, (void **)&batch, 1, NULL) != 1) { assert(0 && "Should never happen"); } } diff --git a/lib/thread/thread.c b/lib/thread/thread.c index a10a773cf..471417ce1 100644 --- a/lib/thread/thread.c +++ b/lib/thread/thread.c @@ -692,7 +692,7 @@ spdk_thread_send_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void *ctx msg->fn = fn; msg->arg = ctx; - rc = spdk_ring_enqueue(thread->messages, (void **)&msg, 1); + rc = spdk_ring_enqueue(thread->messages, (void **)&msg, 1, NULL); if (rc != 1) { assert(false); spdk_mempool_put(g_spdk_msg_mempool, msg); diff --git a/test/common/lib/test_env.c b/test/common/lib/test_env.c index 44a3cf16a..9314f4948 100644 --- a/test/common/lib/test_env.c +++ b/test/common/lib/test_env.c @@ -358,7 +358,8 @@ spdk_ring_free(struct spdk_ring *ring) DEFINE_RETURN_MOCK(spdk_ring_enqueue, size_t); size_t -spdk_ring_enqueue(struct spdk_ring *ring, void **objs, size_t count) +spdk_ring_enqueue(struct spdk_ring *ring, void **objs, size_t count, + size_t *free_space) { struct spdk_ring_ele *ele; size_t i;