diff options
Diffstat (limited to 'src/spdk/lib/thread')
| -rw-r--r-- | src/spdk/lib/thread/Makefile | 45 | ||||
| -rw-r--r-- | src/spdk/lib/thread/spdk_thread.map | 55 | ||||
| -rw-r--r-- | src/spdk/lib/thread/thread.c | 1636 |
3 files changed, 1736 insertions, 0 deletions
diff --git a/src/spdk/lib/thread/Makefile b/src/spdk/lib/thread/Makefile new file mode 100644 index 000000000..ceb7a394e --- /dev/null +++ b/src/spdk/lib/thread/Makefile @@ -0,0 +1,45 @@ +# +# 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. +# + +SPDK_ROOT_DIR := $(abspath $(CURDIR)/../..) +include $(SPDK_ROOT_DIR)/mk/spdk.common.mk + +SO_VER := 3 +SO_MINOR := 0 + +C_SRCS = thread.c +LIBNAME = thread + +SPDK_MAP_FILE = $(abspath $(CURDIR)/spdk_thread.map) + +include $(SPDK_ROOT_DIR)/mk/spdk.lib.mk diff --git a/src/spdk/lib/thread/spdk_thread.map b/src/spdk/lib/thread/spdk_thread.map new file mode 100644 index 000000000..b71fa06eb --- /dev/null +++ b/src/spdk/lib/thread/spdk_thread.map @@ -0,0 +1,55 @@ +{ + global: + + # public functions in spdk/thread.h + spdk_thread_lib_init; + spdk_thread_lib_init_ext; + spdk_thread_lib_fini; + spdk_thread_create; + spdk_set_thread; + spdk_thread_exit; + spdk_thread_is_exited; + spdk_thread_destroy; + spdk_thread_get_ctx; + spdk_thread_get_cpumask; + spdk_thread_set_cpumask; + spdk_thread_get_from_ctx; + spdk_thread_poll; + spdk_thread_next_poller_expiration; + spdk_thread_has_active_pollers; + spdk_thread_has_pollers; + spdk_thread_is_idle; + spdk_thread_get_count; + spdk_get_thread; + spdk_thread_get_name; + spdk_thread_get_id; + spdk_thread_get_by_id; + spdk_thread_get_stats; + spdk_thread_get_last_tsc; + spdk_thread_send_msg; + spdk_thread_send_critical_msg; + spdk_for_each_thread; + spdk_poller_register; + spdk_poller_register_named; + spdk_poller_unregister; + spdk_poller_pause; + spdk_poller_resume; + spdk_io_device_register; + spdk_io_device_unregister; + spdk_get_io_channel; + spdk_put_io_channel; + spdk_io_channel_get_ctx; + spdk_io_channel_from_ctx; + spdk_io_channel_get_thread; + spdk_for_each_channel; + spdk_io_channel_iter_get_io_device; + spdk_io_channel_iter_get_channel; + spdk_io_channel_iter_get_ctx; + spdk_for_each_channel_continue; + + # internal functions in spdk_internal/thread.h + spdk_poller_state_str; + spdk_io_device_get_name; + + local: *; +}; diff --git a/src/spdk/lib/thread/thread.c b/src/spdk/lib/thread/thread.c new file mode 100644 index 000000000..65d91ce35 --- /dev/null +++ b/src/spdk/lib/thread/thread.c @@ -0,0 +1,1636 @@ +/*- + * 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/likely.h" +#include "spdk/queue.h" +#include "spdk/string.h" +#include "spdk/thread.h" +#include "spdk/util.h" + +#include "spdk_internal/log.h" +#include "spdk_internal/thread.h" + +#define SPDK_MSG_BATCH_SIZE 8 +#define SPDK_MAX_DEVICE_NAME_LEN 256 +#define SPDK_THREAD_EXIT_TIMEOUT_SEC 5 + +static pthread_mutex_t g_devlist_mutex = PTHREAD_MUTEX_INITIALIZER; + +static spdk_new_thread_fn g_new_thread_fn = NULL; +static spdk_thread_op_fn g_thread_op_fn = NULL; +static spdk_thread_op_supported_fn g_thread_op_supported_fn; +static size_t g_ctx_sz = 0; +/* Monotonic increasing ID is set to each created thread beginning at 1. Once the + * ID exceeds UINT64_MAX, further thread creation is not allowed and restarting + * SPDK application is required. + */ +static uint64_t g_thread_id = 1; + +struct io_device { + void *io_device; + char name[SPDK_MAX_DEVICE_NAME_LEN + 1]; + spdk_io_channel_create_cb create_cb; + spdk_io_channel_destroy_cb destroy_cb; + spdk_io_device_unregister_cb unregister_cb; + struct spdk_thread *unregister_thread; + uint32_t ctx_size; + uint32_t for_each_count; + TAILQ_ENTRY(io_device) tailq; + + uint32_t refcnt; + + bool unregistered; +}; + +static TAILQ_HEAD(, io_device) g_io_devices = TAILQ_HEAD_INITIALIZER(g_io_devices); + +struct spdk_msg { + spdk_msg_fn fn; + void *arg; + + SLIST_ENTRY(spdk_msg) link; +}; + +#define SPDK_MSG_MEMPOOL_CACHE_SIZE 1024 +static struct spdk_mempool *g_spdk_msg_mempool = NULL; + +static TAILQ_HEAD(, spdk_thread) g_threads = TAILQ_HEAD_INITIALIZER(g_threads); +static uint32_t g_thread_count = 0; + +static __thread struct spdk_thread *tls_thread = NULL; + +static inline struct spdk_thread * +_get_thread(void) +{ + return tls_thread; +} + +static int +_thread_lib_init(size_t ctx_sz) +{ + char mempool_name[SPDK_MAX_MEMZONE_NAME_LEN]; + + g_ctx_sz = ctx_sz; + + snprintf(mempool_name, sizeof(mempool_name), "msgpool_%d", getpid()); + g_spdk_msg_mempool = spdk_mempool_create(mempool_name, + 262144 - 1, /* Power of 2 minus 1 is optimal for memory consumption */ + sizeof(struct spdk_msg), + 0, /* No cache. We do our own. */ + SPDK_ENV_SOCKET_ID_ANY); + + if (!g_spdk_msg_mempool) { + return -1; + } + + return 0; +} + +int +spdk_thread_lib_init(spdk_new_thread_fn new_thread_fn, size_t ctx_sz) +{ + assert(g_new_thread_fn == NULL); + assert(g_thread_op_fn == NULL); + + if (new_thread_fn == NULL) { + SPDK_INFOLOG(SPDK_LOG_THREAD, "new_thread_fn was not specified at spdk_thread_lib_init\n"); + } else { + g_new_thread_fn = new_thread_fn; + } + + return _thread_lib_init(ctx_sz); +} + +int +spdk_thread_lib_init_ext(spdk_thread_op_fn thread_op_fn, + spdk_thread_op_supported_fn thread_op_supported_fn, + size_t ctx_sz) +{ + assert(g_new_thread_fn == NULL); + assert(g_thread_op_fn == NULL); + assert(g_thread_op_supported_fn == NULL); + + if ((thread_op_fn != NULL) != (thread_op_supported_fn != NULL)) { + SPDK_ERRLOG("Both must be defined or undefined together.\n"); + return -EINVAL; + } + + if (thread_op_fn == NULL && thread_op_supported_fn == NULL) { + SPDK_INFOLOG(SPDK_LOG_THREAD, "thread_op_fn and thread_op_supported_fn were not specified\n"); + } else { + g_thread_op_fn = thread_op_fn; + g_thread_op_supported_fn = thread_op_supported_fn; + } + + return _thread_lib_init(ctx_sz); +} + +void +spdk_thread_lib_fini(void) +{ + struct io_device *dev; + + TAILQ_FOREACH(dev, &g_io_devices, tailq) { + SPDK_ERRLOG("io_device %s not unregistered\n", dev->name); + } + + if (g_spdk_msg_mempool) { + spdk_mempool_free(g_spdk_msg_mempool); + g_spdk_msg_mempool = NULL; + } + + g_new_thread_fn = NULL; + g_thread_op_fn = NULL; + g_thread_op_supported_fn = NULL; + g_ctx_sz = 0; +} + +static void +_free_thread(struct spdk_thread *thread) +{ + struct spdk_io_channel *ch; + struct spdk_msg *msg; + struct spdk_poller *poller, *ptmp; + + TAILQ_FOREACH(ch, &thread->io_channels, tailq) { + SPDK_ERRLOG("thread %s still has channel for io_device %s\n", + thread->name, ch->dev->name); + } + + TAILQ_FOREACH_SAFE(poller, &thread->active_pollers, tailq, ptmp) { + if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) { + SPDK_WARNLOG("poller %s still registered at thread exit\n", + poller->name); + } + TAILQ_REMOVE(&thread->active_pollers, poller, tailq); + free(poller); + } + + TAILQ_FOREACH_SAFE(poller, &thread->timed_pollers, tailq, ptmp) { + if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) { + SPDK_WARNLOG("poller %s still registered at thread exit\n", + poller->name); + } + TAILQ_REMOVE(&thread->timed_pollers, poller, tailq); + free(poller); + } + + TAILQ_FOREACH_SAFE(poller, &thread->paused_pollers, tailq, ptmp) { + SPDK_WARNLOG("poller %s still registered at thread exit\n", poller->name); + TAILQ_REMOVE(&thread->paused_pollers, poller, tailq); + free(poller); + } + + pthread_mutex_lock(&g_devlist_mutex); + assert(g_thread_count > 0); + g_thread_count--; + TAILQ_REMOVE(&g_threads, thread, tailq); + pthread_mutex_unlock(&g_devlist_mutex); + + msg = SLIST_FIRST(&thread->msg_cache); + while (msg != NULL) { + SLIST_REMOVE_HEAD(&thread->msg_cache, link); + + assert(thread->msg_cache_count > 0); + thread->msg_cache_count--; + spdk_mempool_put(g_spdk_msg_mempool, msg); + + msg = SLIST_FIRST(&thread->msg_cache); + } + + assert(thread->msg_cache_count == 0); + + spdk_ring_free(thread->messages); + free(thread); +} + +struct spdk_thread * +spdk_thread_create(const char *name, struct spdk_cpuset *cpumask) +{ + struct spdk_thread *thread; + struct spdk_msg *msgs[SPDK_MSG_MEMPOOL_CACHE_SIZE]; + int rc = 0, i; + + thread = calloc(1, sizeof(*thread) + g_ctx_sz); + if (!thread) { + SPDK_ERRLOG("Unable to allocate memory for thread\n"); + return NULL; + } + + if (cpumask) { + spdk_cpuset_copy(&thread->cpumask, cpumask); + } else { + spdk_cpuset_negate(&thread->cpumask); + } + + TAILQ_INIT(&thread->io_channels); + TAILQ_INIT(&thread->active_pollers); + TAILQ_INIT(&thread->timed_pollers); + TAILQ_INIT(&thread->paused_pollers); + SLIST_INIT(&thread->msg_cache); + thread->msg_cache_count = 0; + + thread->tsc_last = spdk_get_ticks(); + + thread->messages = spdk_ring_create(SPDK_RING_TYPE_MP_SC, 65536, SPDK_ENV_SOCKET_ID_ANY); + if (!thread->messages) { + SPDK_ERRLOG("Unable to allocate memory for message ring\n"); + free(thread); + return NULL; + } + + /* Fill the local message pool cache. */ + rc = spdk_mempool_get_bulk(g_spdk_msg_mempool, (void **)msgs, SPDK_MSG_MEMPOOL_CACHE_SIZE); + if (rc == 0) { + /* If we can't populate the cache it's ok. The cache will get filled + * up organically as messages are passed to the thread. */ + for (i = 0; i < SPDK_MSG_MEMPOOL_CACHE_SIZE; i++) { + SLIST_INSERT_HEAD(&thread->msg_cache, msgs[i], link); + thread->msg_cache_count++; + } + } + + if (name) { + snprintf(thread->name, sizeof(thread->name), "%s", name); + } else { + snprintf(thread->name, sizeof(thread->name), "%p", thread); + } + + pthread_mutex_lock(&g_devlist_mutex); + if (g_thread_id == 0) { + SPDK_ERRLOG("Thread ID rolled over. Further thread creation is not allowed.\n"); + pthread_mutex_unlock(&g_devlist_mutex); + _free_thread(thread); + return NULL; + } + thread->id = g_thread_id++; + TAILQ_INSERT_TAIL(&g_threads, thread, tailq); + g_thread_count++; + pthread_mutex_unlock(&g_devlist_mutex); + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Allocating new thread (%" PRIu64 ", %s)\n", + thread->id, thread->name); + + if (g_new_thread_fn) { + rc = g_new_thread_fn(thread); + } else if (g_thread_op_supported_fn && g_thread_op_supported_fn(SPDK_THREAD_OP_NEW)) { + rc = g_thread_op_fn(thread, SPDK_THREAD_OP_NEW); + } + + if (rc != 0) { + _free_thread(thread); + return NULL; + } + + thread->state = SPDK_THREAD_STATE_RUNNING; + + return thread; +} + +void +spdk_set_thread(struct spdk_thread *thread) +{ + tls_thread = thread; +} + +static void +thread_exit(struct spdk_thread *thread, uint64_t now) +{ + struct spdk_poller *poller; + struct spdk_io_channel *ch; + + if (now >= thread->exit_timeout_tsc) { + SPDK_ERRLOG("thread %s got timeout, and move it to the exited state forcefully\n", + thread->name); + goto exited; + } + + TAILQ_FOREACH(poller, &thread->active_pollers, tailq) { + if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) { + SPDK_INFOLOG(SPDK_LOG_THREAD, + "thread %s still has active poller %s\n", + thread->name, poller->name); + return; + } + } + + TAILQ_FOREACH(poller, &thread->timed_pollers, tailq) { + if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) { + SPDK_INFOLOG(SPDK_LOG_THREAD, + "thread %s still has active timed poller %s\n", + thread->name, poller->name); + return; + } + } + + TAILQ_FOREACH(poller, &thread->paused_pollers, tailq) { + SPDK_INFOLOG(SPDK_LOG_THREAD, + "thread %s still has paused poller %s\n", + thread->name, poller->name); + return; + } + + TAILQ_FOREACH(ch, &thread->io_channels, tailq) { + SPDK_INFOLOG(SPDK_LOG_THREAD, + "thread %s still has channel for io_device %s\n", + thread->name, ch->dev->name); + return; + } + +exited: + thread->state = SPDK_THREAD_STATE_EXITED; +} + +int +spdk_thread_exit(struct spdk_thread *thread) +{ + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Exit thread %s\n", thread->name); + + assert(tls_thread == thread); + + if (thread->state >= SPDK_THREAD_STATE_EXITING) { + SPDK_INFOLOG(SPDK_LOG_THREAD, + "thread %s is already exiting\n", + thread->name); + return 0; + } + + thread->exit_timeout_tsc = spdk_get_ticks() + (spdk_get_ticks_hz() * + SPDK_THREAD_EXIT_TIMEOUT_SEC); + thread->state = SPDK_THREAD_STATE_EXITING; + return 0; +} + +bool +spdk_thread_is_exited(struct spdk_thread *thread) +{ + return thread->state == SPDK_THREAD_STATE_EXITED; +} + +void +spdk_thread_destroy(struct spdk_thread *thread) +{ + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Destroy thread %s\n", thread->name); + + assert(thread->state == SPDK_THREAD_STATE_EXITED); + + if (tls_thread == thread) { + tls_thread = NULL; + } + + _free_thread(thread); +} + +void * +spdk_thread_get_ctx(struct spdk_thread *thread) +{ + if (g_ctx_sz > 0) { + return thread->ctx; + } + + return NULL; +} + +struct spdk_cpuset * +spdk_thread_get_cpumask(struct spdk_thread *thread) +{ + return &thread->cpumask; +} + +int +spdk_thread_set_cpumask(struct spdk_cpuset *cpumask) +{ + struct spdk_thread *thread; + + if (!g_thread_op_supported_fn || !g_thread_op_supported_fn(SPDK_THREAD_OP_RESCHED)) { + SPDK_ERRLOG("Framework does not support reschedule operation.\n"); + assert(false); + return -ENOTSUP; + } + + thread = spdk_get_thread(); + if (!thread) { + SPDK_ERRLOG("Called from non-SPDK thread\n"); + assert(false); + return -EINVAL; + } + + spdk_cpuset_copy(&thread->cpumask, cpumask); + + /* Invoke framework's reschedule operation. If this function is called multiple times + * in a single spdk_thread_poll() context, the last cpumask will be used in the + * reschedule operation. + */ + g_thread_op_fn(thread, SPDK_THREAD_OP_RESCHED); + + return 0; +} + +struct spdk_thread * +spdk_thread_get_from_ctx(void *ctx) +{ + if (ctx == NULL) { + assert(false); + return NULL; + } + + assert(g_ctx_sz > 0); + + return SPDK_CONTAINEROF(ctx, struct spdk_thread, ctx); +} + +static inline uint32_t +msg_queue_run_batch(struct spdk_thread *thread, uint32_t max_msgs) +{ + unsigned count, i; + void *messages[SPDK_MSG_BATCH_SIZE]; + +#ifdef DEBUG + /* + * spdk_ring_dequeue() fills messages and returns how many entries it wrote, + * so we will never actually read uninitialized data from events, but just to be sure + * (and to silence a static analyzer false positive), initialize the array to NULL pointers. + */ + memset(messages, 0, sizeof(messages)); +#endif + + if (max_msgs > 0) { + max_msgs = spdk_min(max_msgs, SPDK_MSG_BATCH_SIZE); + } else { + max_msgs = SPDK_MSG_BATCH_SIZE; + } + + count = spdk_ring_dequeue(thread->messages, messages, max_msgs); + if (count == 0) { + return 0; + } + + for (i = 0; i < count; i++) { + struct spdk_msg *msg = messages[i]; + + assert(msg != NULL); + msg->fn(msg->arg); + + if (thread->msg_cache_count < SPDK_MSG_MEMPOOL_CACHE_SIZE) { + /* Insert the messages at the head. We want to re-use the hot + * ones. */ + SLIST_INSERT_HEAD(&thread->msg_cache, msg, link); + thread->msg_cache_count++; + } else { + spdk_mempool_put(g_spdk_msg_mempool, msg); + } + } + + return count; +} + +static void +poller_insert_timer(struct spdk_thread *thread, struct spdk_poller *poller, uint64_t now) +{ + struct spdk_poller *iter; + + poller->next_run_tick = now + poller->period_ticks; + + /* + * Insert poller in the thread's timed_pollers list in sorted order by next scheduled + * run time. + */ + TAILQ_FOREACH_REVERSE(iter, &thread->timed_pollers, timed_pollers_head, tailq) { + if (iter->next_run_tick <= poller->next_run_tick) { + TAILQ_INSERT_AFTER(&thread->timed_pollers, iter, poller, tailq); + return; + } + } + + /* No earlier pollers were found, so this poller must be the new head */ + TAILQ_INSERT_HEAD(&thread->timed_pollers, poller, tailq); +} + +static void +thread_insert_poller(struct spdk_thread *thread, struct spdk_poller *poller) +{ + if (poller->period_ticks) { + poller_insert_timer(thread, poller, spdk_get_ticks()); + } else { + TAILQ_INSERT_TAIL(&thread->active_pollers, poller, tailq); + } +} + +static inline void +thread_update_stats(struct spdk_thread *thread, uint64_t end, + uint64_t start, int rc) +{ + if (rc == 0) { + /* Poller status idle */ + thread->stats.idle_tsc += end - start; + } else if (rc > 0) { + /* Poller status busy */ + thread->stats.busy_tsc += end - start; + } + /* Store end time to use it as start time of the next spdk_thread_poll(). */ + thread->tsc_last = end; +} + +static int +thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now) +{ + uint32_t msg_count; + struct spdk_poller *poller, *tmp; + spdk_msg_fn critical_msg; + int rc = 0; + + critical_msg = thread->critical_msg; + if (spdk_unlikely(critical_msg != NULL)) { + critical_msg(NULL); + thread->critical_msg = NULL; + } + + msg_count = msg_queue_run_batch(thread, max_msgs); + if (msg_count) { + rc = 1; + } + + TAILQ_FOREACH_REVERSE_SAFE(poller, &thread->active_pollers, + active_pollers_head, tailq, tmp) { + int poller_rc; + + if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) { + TAILQ_REMOVE(&thread->active_pollers, poller, tailq); + free(poller); + continue; + } else if (poller->state == SPDK_POLLER_STATE_PAUSING) { + TAILQ_REMOVE(&thread->active_pollers, poller, tailq); + TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq); + poller->state = SPDK_POLLER_STATE_PAUSED; + continue; + } + + poller->state = SPDK_POLLER_STATE_RUNNING; + poller_rc = poller->fn(poller->arg); + + poller->run_count++; + if (poller_rc > 0) { + poller->busy_count++; + } + +#ifdef DEBUG + if (poller_rc == -1) { + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Poller %s returned -1\n", poller->name); + } +#endif + + if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) { + TAILQ_REMOVE(&thread->active_pollers, poller, tailq); + free(poller); + } else if (poller->state != SPDK_POLLER_STATE_PAUSED) { + poller->state = SPDK_POLLER_STATE_WAITING; + } + + if (poller_rc > rc) { + rc = poller_rc; + } + } + + TAILQ_FOREACH_SAFE(poller, &thread->timed_pollers, tailq, tmp) { + int timer_rc = 0; + + if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) { + TAILQ_REMOVE(&thread->timed_pollers, poller, tailq); + free(poller); + continue; + } else if (poller->state == SPDK_POLLER_STATE_PAUSING) { + TAILQ_REMOVE(&thread->timed_pollers, poller, tailq); + TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq); + poller->state = SPDK_POLLER_STATE_PAUSED; + continue; + } + + if (now < poller->next_run_tick) { + break; + } + + poller->state = SPDK_POLLER_STATE_RUNNING; + timer_rc = poller->fn(poller->arg); + + poller->run_count++; + if (timer_rc > 0) { + poller->busy_count++; + } + +#ifdef DEBUG + if (timer_rc == -1) { + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Timed poller %s returned -1\n", poller->name); + } +#endif + + if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) { + TAILQ_REMOVE(&thread->timed_pollers, poller, tailq); + free(poller); + } else if (poller->state != SPDK_POLLER_STATE_PAUSED) { + poller->state = SPDK_POLLER_STATE_WAITING; + TAILQ_REMOVE(&thread->timed_pollers, poller, tailq); + poller_insert_timer(thread, poller, now); + } + + if (timer_rc > rc) { + rc = timer_rc; + } + } + + return rc; +} + +int +spdk_thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now) +{ + struct spdk_thread *orig_thread; + int rc; + + orig_thread = _get_thread(); + tls_thread = thread; + + if (now == 0) { + now = spdk_get_ticks(); + } + + rc = thread_poll(thread, max_msgs, now); + + if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITING)) { + thread_exit(thread, now); + } + + thread_update_stats(thread, spdk_get_ticks(), now, rc); + + tls_thread = orig_thread; + + return rc; +} + +uint64_t +spdk_thread_next_poller_expiration(struct spdk_thread *thread) +{ + struct spdk_poller *poller; + + poller = TAILQ_FIRST(&thread->timed_pollers); + if (poller) { + return poller->next_run_tick; + } + + return 0; +} + +int +spdk_thread_has_active_pollers(struct spdk_thread *thread) +{ + return !TAILQ_EMPTY(&thread->active_pollers); +} + +static bool +thread_has_unpaused_pollers(struct spdk_thread *thread) +{ + if (TAILQ_EMPTY(&thread->active_pollers) && + TAILQ_EMPTY(&thread->timed_pollers)) { + return false; + } + + return true; +} + +bool +spdk_thread_has_pollers(struct spdk_thread *thread) +{ + if (!thread_has_unpaused_pollers(thread) && + TAILQ_EMPTY(&thread->paused_pollers)) { + return false; + } + + return true; +} + +bool +spdk_thread_is_idle(struct spdk_thread *thread) +{ + if (spdk_ring_count(thread->messages) || + thread_has_unpaused_pollers(thread) || + thread->critical_msg != NULL) { + return false; + } + + return true; +} + +uint32_t +spdk_thread_get_count(void) +{ + /* + * Return cached value of the current thread count. We could acquire the + * lock and iterate through the TAILQ of threads to count them, but that + * count could still be invalidated after we release the lock. + */ + return g_thread_count; +} + +struct spdk_thread * +spdk_get_thread(void) +{ + return _get_thread(); +} + +const char * +spdk_thread_get_name(const struct spdk_thread *thread) +{ + return thread->name; +} + +uint64_t +spdk_thread_get_id(const struct spdk_thread *thread) +{ + return thread->id; +} + +struct spdk_thread * +spdk_thread_get_by_id(uint64_t id) +{ + struct spdk_thread *thread; + + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_FOREACH(thread, &g_threads, tailq) { + if (thread->id == id) { + pthread_mutex_unlock(&g_devlist_mutex); + + return thread; + } + } + pthread_mutex_unlock(&g_devlist_mutex); + + return NULL; +} + +int +spdk_thread_get_stats(struct spdk_thread_stats *stats) +{ + struct spdk_thread *thread; + + thread = _get_thread(); + if (!thread) { + SPDK_ERRLOG("No thread allocated\n"); + return -EINVAL; + } + + if (stats == NULL) { + return -EINVAL; + } + + *stats = thread->stats; + + return 0; +} + +uint64_t +spdk_thread_get_last_tsc(struct spdk_thread *thread) +{ + return thread->tsc_last; +} + +int +spdk_thread_send_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void *ctx) +{ + struct spdk_thread *local_thread; + struct spdk_msg *msg; + int rc; + + assert(thread != NULL); + + if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITED)) { + SPDK_ERRLOG("Thread %s is marked as exited.\n", thread->name); + return -EIO; + } + + local_thread = _get_thread(); + + msg = NULL; + if (local_thread != NULL) { + if (local_thread->msg_cache_count > 0) { + msg = SLIST_FIRST(&local_thread->msg_cache); + assert(msg != NULL); + SLIST_REMOVE_HEAD(&local_thread->msg_cache, link); + local_thread->msg_cache_count--; + } + } + + if (msg == NULL) { + msg = spdk_mempool_get(g_spdk_msg_mempool); + if (!msg) { + SPDK_ERRLOG("msg could not be allocated\n"); + return -ENOMEM; + } + } + + msg->fn = fn; + msg->arg = ctx; + + rc = spdk_ring_enqueue(thread->messages, (void **)&msg, 1, NULL); + if (rc != 1) { + SPDK_ERRLOG("msg could not be enqueued\n"); + spdk_mempool_put(g_spdk_msg_mempool, msg); + return -EIO; + } + + return 0; +} + +int +spdk_thread_send_critical_msg(struct spdk_thread *thread, spdk_msg_fn fn) +{ + spdk_msg_fn expected = NULL; + + if (__atomic_compare_exchange_n(&thread->critical_msg, &expected, fn, false, __ATOMIC_SEQ_CST, + __ATOMIC_SEQ_CST)) { + return 0; + } + + return -EIO; +} + +static struct spdk_poller * +poller_register(spdk_poller_fn fn, + void *arg, + uint64_t period_microseconds, + const char *name) +{ + struct spdk_thread *thread; + struct spdk_poller *poller; + uint64_t quotient, remainder, ticks; + + thread = spdk_get_thread(); + if (!thread) { + assert(false); + return NULL; + } + + if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITED)) { + SPDK_ERRLOG("thread %s is marked as exited\n", thread->name); + return NULL; + } + + poller = calloc(1, sizeof(*poller)); + if (poller == NULL) { + SPDK_ERRLOG("Poller memory allocation failed\n"); + return NULL; + } + + if (name) { + snprintf(poller->name, sizeof(poller->name), "%s", name); + } else { + snprintf(poller->name, sizeof(poller->name), "%p", fn); + } + + poller->state = SPDK_POLLER_STATE_WAITING; + poller->fn = fn; + poller->arg = arg; + poller->thread = thread; + + if (period_microseconds) { + quotient = period_microseconds / SPDK_SEC_TO_USEC; + remainder = period_microseconds % SPDK_SEC_TO_USEC; + ticks = spdk_get_ticks_hz(); + + poller->period_ticks = ticks * quotient + (ticks * remainder) / SPDK_SEC_TO_USEC; + } else { + poller->period_ticks = 0; + } + + thread_insert_poller(thread, poller); + + return poller; +} + +struct spdk_poller * +spdk_poller_register(spdk_poller_fn fn, + void *arg, + uint64_t period_microseconds) +{ + return poller_register(fn, arg, period_microseconds, NULL); +} + +struct spdk_poller * +spdk_poller_register_named(spdk_poller_fn fn, + void *arg, + uint64_t period_microseconds, + const char *name) +{ + return poller_register(fn, arg, period_microseconds, name); +} + +void +spdk_poller_unregister(struct spdk_poller **ppoller) +{ + struct spdk_thread *thread; + struct spdk_poller *poller; + + poller = *ppoller; + if (poller == NULL) { + return; + } + + *ppoller = NULL; + + thread = spdk_get_thread(); + if (!thread) { + assert(false); + return; + } + + if (poller->thread != thread) { + SPDK_ERRLOG("different from the thread that called spdk_poller_register()\n"); + assert(false); + return; + } + + /* If the poller was paused, put it on the active_pollers list so that + * its unregistration can be processed by spdk_thread_poll(). + */ + if (poller->state == SPDK_POLLER_STATE_PAUSED) { + TAILQ_REMOVE(&thread->paused_pollers, poller, tailq); + TAILQ_INSERT_TAIL(&thread->active_pollers, poller, tailq); + poller->period_ticks = 0; + } + + /* Simply set the state to unregistered. The poller will get cleaned up + * in a subsequent call to spdk_thread_poll(). + */ + poller->state = SPDK_POLLER_STATE_UNREGISTERED; +} + +void +spdk_poller_pause(struct spdk_poller *poller) +{ + struct spdk_thread *thread; + + if (poller->state == SPDK_POLLER_STATE_PAUSED || + poller->state == SPDK_POLLER_STATE_PAUSING) { + return; + } + + thread = spdk_get_thread(); + if (!thread) { + assert(false); + return; + } + + /* If a poller is paused from within itself, we can immediately move it + * on the paused_pollers list. Otherwise we just set its state to + * SPDK_POLLER_STATE_PAUSING and let spdk_thread_poll() move it. It + * allows a poller to be paused from another one's context without + * breaking the TAILQ_FOREACH_REVERSE_SAFE iteration. + */ + if (poller->state != SPDK_POLLER_STATE_RUNNING) { + poller->state = SPDK_POLLER_STATE_PAUSING; + } else { + if (poller->period_ticks > 0) { + TAILQ_REMOVE(&thread->timed_pollers, poller, tailq); + } else { + TAILQ_REMOVE(&thread->active_pollers, poller, tailq); + } + + TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq); + poller->state = SPDK_POLLER_STATE_PAUSED; + } +} + +void +spdk_poller_resume(struct spdk_poller *poller) +{ + struct spdk_thread *thread; + + if (poller->state != SPDK_POLLER_STATE_PAUSED && + poller->state != SPDK_POLLER_STATE_PAUSING) { + return; + } + + thread = spdk_get_thread(); + if (!thread) { + assert(false); + return; + } + + /* If a poller is paused it has to be removed from the paused pollers + * list and put on the active / timer list depending on its + * period_ticks. If a poller is still in the process of being paused, + * we just need to flip its state back to waiting, as it's already on + * the appropriate list. + */ + if (poller->state == SPDK_POLLER_STATE_PAUSED) { + TAILQ_REMOVE(&thread->paused_pollers, poller, tailq); + thread_insert_poller(thread, poller); + } + + poller->state = SPDK_POLLER_STATE_WAITING; +} + +const char * +spdk_poller_state_str(enum spdk_poller_state state) +{ + switch (state) { + case SPDK_POLLER_STATE_WAITING: + return "waiting"; + case SPDK_POLLER_STATE_RUNNING: + return "running"; + case SPDK_POLLER_STATE_UNREGISTERED: + return "unregistered"; + case SPDK_POLLER_STATE_PAUSING: + return "pausing"; + case SPDK_POLLER_STATE_PAUSED: + return "paused"; + default: + return NULL; + } +} + +struct call_thread { + struct spdk_thread *cur_thread; + spdk_msg_fn fn; + void *ctx; + + struct spdk_thread *orig_thread; + spdk_msg_fn cpl; +}; + +static void +_on_thread(void *ctx) +{ + struct call_thread *ct = ctx; + int rc __attribute__((unused)); + + ct->fn(ct->ctx); + + pthread_mutex_lock(&g_devlist_mutex); + ct->cur_thread = TAILQ_NEXT(ct->cur_thread, tailq); + pthread_mutex_unlock(&g_devlist_mutex); + + if (!ct->cur_thread) { + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Completed thread iteration\n"); + + rc = spdk_thread_send_msg(ct->orig_thread, ct->cpl, ct->ctx); + free(ctx); + } else { + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Continuing thread iteration to %s\n", + ct->cur_thread->name); + + rc = spdk_thread_send_msg(ct->cur_thread, _on_thread, ctx); + } + assert(rc == 0); +} + +void +spdk_for_each_thread(spdk_msg_fn fn, void *ctx, spdk_msg_fn cpl) +{ + struct call_thread *ct; + struct spdk_thread *thread; + int rc __attribute__((unused)); + + ct = calloc(1, sizeof(*ct)); + if (!ct) { + SPDK_ERRLOG("Unable to perform thread iteration\n"); + cpl(ctx); + return; + } + + ct->fn = fn; + ct->ctx = ctx; + ct->cpl = cpl; + + thread = _get_thread(); + if (!thread) { + SPDK_ERRLOG("No thread allocated\n"); + free(ct); + cpl(ctx); + return; + } + ct->orig_thread = thread; + + pthread_mutex_lock(&g_devlist_mutex); + ct->cur_thread = TAILQ_FIRST(&g_threads); + pthread_mutex_unlock(&g_devlist_mutex); + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Starting thread iteration from %s\n", + ct->orig_thread->name); + + rc = spdk_thread_send_msg(ct->cur_thread, _on_thread, ct); + assert(rc == 0); +} + +void +spdk_io_device_register(void *io_device, spdk_io_channel_create_cb create_cb, + spdk_io_channel_destroy_cb destroy_cb, uint32_t ctx_size, + const char *name) +{ + struct io_device *dev, *tmp; + struct spdk_thread *thread; + + assert(io_device != NULL); + assert(create_cb != NULL); + assert(destroy_cb != NULL); + + thread = spdk_get_thread(); + if (!thread) { + SPDK_ERRLOG("called from non-SPDK thread\n"); + assert(false); + return; + } + + dev = calloc(1, sizeof(struct io_device)); + if (dev == NULL) { + SPDK_ERRLOG("could not allocate io_device\n"); + return; + } + + dev->io_device = io_device; + if (name) { + snprintf(dev->name, sizeof(dev->name), "%s", name); + } else { + snprintf(dev->name, sizeof(dev->name), "%p", dev); + } + dev->create_cb = create_cb; + dev->destroy_cb = destroy_cb; + dev->unregister_cb = NULL; + dev->ctx_size = ctx_size; + dev->for_each_count = 0; + dev->unregistered = false; + dev->refcnt = 0; + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Registering io_device %s (%p) on thread %s\n", + dev->name, dev->io_device, thread->name); + + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_FOREACH(tmp, &g_io_devices, tailq) { + if (tmp->io_device == io_device) { + SPDK_ERRLOG("io_device %p already registered (old:%s new:%s)\n", + io_device, tmp->name, dev->name); + free(dev); + pthread_mutex_unlock(&g_devlist_mutex); + return; + } + } + TAILQ_INSERT_TAIL(&g_io_devices, dev, tailq); + pthread_mutex_unlock(&g_devlist_mutex); +} + +static void +_finish_unregister(void *arg) +{ + struct io_device *dev = arg; + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Finishing unregistration of io_device %s (%p) on thread %s\n", + dev->name, dev->io_device, dev->unregister_thread->name); + + dev->unregister_cb(dev->io_device); + free(dev); +} + +static void +io_device_free(struct io_device *dev) +{ + int rc __attribute__((unused)); + + if (dev->unregister_cb == NULL) { + free(dev); + } else { + assert(dev->unregister_thread != NULL); + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "io_device %s (%p) needs to unregister from thread %s\n", + dev->name, dev->io_device, dev->unregister_thread->name); + rc = spdk_thread_send_msg(dev->unregister_thread, _finish_unregister, dev); + assert(rc == 0); + } +} + +void +spdk_io_device_unregister(void *io_device, spdk_io_device_unregister_cb unregister_cb) +{ + struct io_device *dev; + uint32_t refcnt; + struct spdk_thread *thread; + + thread = spdk_get_thread(); + if (!thread) { + SPDK_ERRLOG("called from non-SPDK thread\n"); + assert(false); + return; + } + + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_FOREACH(dev, &g_io_devices, tailq) { + if (dev->io_device == io_device) { + break; + } + } + + if (!dev) { + SPDK_ERRLOG("io_device %p not found\n", io_device); + assert(false); + pthread_mutex_unlock(&g_devlist_mutex); + return; + } + + if (dev->for_each_count > 0) { + SPDK_ERRLOG("io_device %s (%p) has %u for_each calls outstanding\n", + dev->name, io_device, dev->for_each_count); + pthread_mutex_unlock(&g_devlist_mutex); + return; + } + + dev->unregister_cb = unregister_cb; + dev->unregistered = true; + TAILQ_REMOVE(&g_io_devices, dev, tailq); + refcnt = dev->refcnt; + dev->unregister_thread = thread; + pthread_mutex_unlock(&g_devlist_mutex); + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Unregistering io_device %s (%p) from thread %s\n", + dev->name, dev->io_device, thread->name); + + if (refcnt > 0) { + /* defer deletion */ + return; + } + + io_device_free(dev); +} + +const char * +spdk_io_device_get_name(struct io_device *dev) +{ + return dev->name; +} + +struct spdk_io_channel * +spdk_get_io_channel(void *io_device) +{ + struct spdk_io_channel *ch; + struct spdk_thread *thread; + struct io_device *dev; + int rc; + + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_FOREACH(dev, &g_io_devices, tailq) { + if (dev->io_device == io_device) { + break; + } + } + if (dev == NULL) { + SPDK_ERRLOG("could not find io_device %p\n", io_device); + pthread_mutex_unlock(&g_devlist_mutex); + return NULL; + } + + thread = _get_thread(); + if (!thread) { + SPDK_ERRLOG("No thread allocated\n"); + pthread_mutex_unlock(&g_devlist_mutex); + return NULL; + } + + if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITED)) { + SPDK_ERRLOG("Thread %s is marked as exited\n", thread->name); + pthread_mutex_unlock(&g_devlist_mutex); + return NULL; + } + + TAILQ_FOREACH(ch, &thread->io_channels, tailq) { + if (ch->dev == dev) { + ch->ref++; + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Get io_channel %p for io_device %s (%p) on thread %s refcnt %u\n", + ch, dev->name, dev->io_device, thread->name, ch->ref); + + /* + * An I/O channel already exists for this device on this + * thread, so return it. + */ + pthread_mutex_unlock(&g_devlist_mutex); + return ch; + } + } + + ch = calloc(1, sizeof(*ch) + dev->ctx_size); + if (ch == NULL) { + SPDK_ERRLOG("could not calloc spdk_io_channel\n"); + pthread_mutex_unlock(&g_devlist_mutex); + return NULL; + } + + ch->dev = dev; + ch->destroy_cb = dev->destroy_cb; + ch->thread = thread; + ch->ref = 1; + ch->destroy_ref = 0; + TAILQ_INSERT_TAIL(&thread->io_channels, ch, tailq); + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Get io_channel %p for io_device %s (%p) on thread %s refcnt %u\n", + ch, dev->name, dev->io_device, thread->name, ch->ref); + + dev->refcnt++; + + pthread_mutex_unlock(&g_devlist_mutex); + + rc = dev->create_cb(io_device, (uint8_t *)ch + sizeof(*ch)); + if (rc != 0) { + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_REMOVE(&ch->thread->io_channels, ch, tailq); + dev->refcnt--; + free(ch); + pthread_mutex_unlock(&g_devlist_mutex); + return NULL; + } + + return ch; +} + +static void +put_io_channel(void *arg) +{ + struct spdk_io_channel *ch = arg; + bool do_remove_dev = true; + struct spdk_thread *thread; + + thread = spdk_get_thread(); + if (!thread) { + SPDK_ERRLOG("called from non-SPDK thread\n"); + assert(false); + return; + } + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, + "Releasing io_channel %p for io_device %s (%p) on thread %s\n", + ch, ch->dev->name, ch->dev->io_device, thread->name); + + assert(ch->thread == thread); + + ch->destroy_ref--; + + if (ch->ref > 0 || ch->destroy_ref > 0) { + /* + * Another reference to the associated io_device was requested + * after this message was sent but before it had a chance to + * execute. + */ + return; + } + + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_REMOVE(&ch->thread->io_channels, ch, tailq); + pthread_mutex_unlock(&g_devlist_mutex); + + /* Don't hold the devlist mutex while the destroy_cb is called. */ + ch->destroy_cb(ch->dev->io_device, spdk_io_channel_get_ctx(ch)); + + pthread_mutex_lock(&g_devlist_mutex); + ch->dev->refcnt--; + + if (!ch->dev->unregistered) { + do_remove_dev = false; + } + + if (ch->dev->refcnt > 0) { + do_remove_dev = false; + } + + pthread_mutex_unlock(&g_devlist_mutex); + + if (do_remove_dev) { + io_device_free(ch->dev); + } + free(ch); +} + +void +spdk_put_io_channel(struct spdk_io_channel *ch) +{ + struct spdk_thread *thread; + int rc __attribute__((unused)); + + thread = spdk_get_thread(); + if (!thread) { + SPDK_ERRLOG("called from non-SPDK thread\n"); + assert(false); + return; + } + + if (ch->thread != thread) { + SPDK_ERRLOG("different from the thread that called get_io_channel()\n"); + assert(false); + return; + } + + SPDK_DEBUGLOG(SPDK_LOG_THREAD, + "Putting io_channel %p for io_device %s (%p) on thread %s refcnt %u\n", + ch, ch->dev->name, ch->dev->io_device, thread->name, ch->ref); + + ch->ref--; + + if (ch->ref == 0) { + ch->destroy_ref++; + rc = spdk_thread_send_msg(thread, put_io_channel, ch); + assert(rc == 0); + } +} + +struct spdk_io_channel * +spdk_io_channel_from_ctx(void *ctx) +{ + return (struct spdk_io_channel *)((uint8_t *)ctx - sizeof(struct spdk_io_channel)); +} + +struct spdk_thread * +spdk_io_channel_get_thread(struct spdk_io_channel *ch) +{ + return ch->thread; +} + +struct spdk_io_channel_iter { + void *io_device; + struct io_device *dev; + spdk_channel_msg fn; + int status; + void *ctx; + struct spdk_io_channel *ch; + + struct spdk_thread *cur_thread; + + struct spdk_thread *orig_thread; + spdk_channel_for_each_cpl cpl; +}; + +void * +spdk_io_channel_iter_get_io_device(struct spdk_io_channel_iter *i) +{ + return i->io_device; +} + +struct spdk_io_channel * +spdk_io_channel_iter_get_channel(struct spdk_io_channel_iter *i) +{ + return i->ch; +} + +void * +spdk_io_channel_iter_get_ctx(struct spdk_io_channel_iter *i) +{ + return i->ctx; +} + +static void +_call_completion(void *ctx) +{ + struct spdk_io_channel_iter *i = ctx; + + if (i->cpl != NULL) { + i->cpl(i, i->status); + } + free(i); +} + +static void +_call_channel(void *ctx) +{ + struct spdk_io_channel_iter *i = ctx; + struct spdk_io_channel *ch; + + /* + * It is possible that the channel was deleted before this + * message had a chance to execute. If so, skip calling + * the fn() on this thread. + */ + pthread_mutex_lock(&g_devlist_mutex); + TAILQ_FOREACH(ch, &i->cur_thread->io_channels, tailq) { + if (ch->dev->io_device == i->io_device) { + break; + } + } + pthread_mutex_unlock(&g_devlist_mutex); + + if (ch) { + i->fn(i); + } else { + spdk_for_each_channel_continue(i, 0); + } +} + +void +spdk_for_each_channel(void *io_device, spdk_channel_msg fn, void *ctx, + spdk_channel_for_each_cpl cpl) +{ + struct spdk_thread *thread; + struct spdk_io_channel *ch; + struct spdk_io_channel_iter *i; + int rc __attribute__((unused)); + + i = calloc(1, sizeof(*i)); + if (!i) { + SPDK_ERRLOG("Unable to allocate iterator\n"); + return; + } + + i->io_device = io_device; + i->fn = fn; + i->ctx = ctx; + i->cpl = cpl; + + pthread_mutex_lock(&g_devlist_mutex); + i->orig_thread = _get_thread(); + + TAILQ_FOREACH(thread, &g_threads, tailq) { + TAILQ_FOREACH(ch, &thread->io_channels, tailq) { + if (ch->dev->io_device == io_device) { + ch->dev->for_each_count++; + i->dev = ch->dev; + i->cur_thread = thread; + i->ch = ch; + pthread_mutex_unlock(&g_devlist_mutex); + rc = spdk_thread_send_msg(thread, _call_channel, i); + assert(rc == 0); + return; + } + } + } + + pthread_mutex_unlock(&g_devlist_mutex); + + rc = spdk_thread_send_msg(i->orig_thread, _call_completion, i); + assert(rc == 0); +} + +void +spdk_for_each_channel_continue(struct spdk_io_channel_iter *i, int status) +{ + struct spdk_thread *thread; + struct spdk_io_channel *ch; + int rc __attribute__((unused)); + + assert(i->cur_thread == spdk_get_thread()); + + i->status = status; + + pthread_mutex_lock(&g_devlist_mutex); + if (status) { + goto end; + } + thread = TAILQ_NEXT(i->cur_thread, tailq); + while (thread) { + TAILQ_FOREACH(ch, &thread->io_channels, tailq) { + if (ch->dev->io_device == i->io_device) { + i->cur_thread = thread; + i->ch = ch; + pthread_mutex_unlock(&g_devlist_mutex); + rc = spdk_thread_send_msg(thread, _call_channel, i); + assert(rc == 0); + return; + } + } + thread = TAILQ_NEXT(thread, tailq); + } + +end: + i->dev->for_each_count--; + i->ch = NULL; + pthread_mutex_unlock(&g_devlist_mutex); + + rc = spdk_thread_send_msg(i->orig_thread, _call_completion, i); + assert(rc == 0); +} + + +SPDK_LOG_REGISTER_COMPONENT("thread", SPDK_LOG_THREAD) |