[SPDK] Re: How to fix the error generated while generating SPDK thread

Hi Tomek.
Thanks for your response.
With your guidance the previous error in thread creation was solved but a new one that leads to an exception error in the program has been observed.
I changed the code as follows:

#include <stdio.h>
#include <spdk/bdev.h>
#include <spdk/thread.h>
#include <spdk/queue.h>
#include "spdk/env.h"

static struct bdev_reactor *g_master_reactor = NULL;
static struct bdev_reactor *g_next_reactor = NULL;
static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
static bool g_reactors_exit = false;

struct bdev_lw_thread {
    TAILQ_ENTRY(bdev_lw_thread) link;
};

struct bdev_reactor {
    uint32_t core;
    pthread_mutex_t mutex;

    TAILQ_HEAD(, bdev_lw_thread)	threads;
    TAILQ_ENTRY(bdev_reactor)	link;
};

TAILQ_HEAD(, bdev_reactor) g_reactors = TAILQ_HEAD_INITIALIZER(g_reactors);

static int
bdev_reactor_run(void *arg)
{
    struct bdev_reactor *bdev_reactor = arg;
    struct bdev_lw_thread *lw_thread, *tmp;
    struct spdk_thread *thread;

    /* foreach all the lightweight threads in this bdev_reactor */
    do {
        pthread_mutex_lock(&bdev_reactor->mutex);
        TAILQ_FOREACH_SAFE(lw_thread, &bdev_reactor->threads, link, tmp) {
            thread = spdk_thread_get_from_ctx(lw_thread);

            spdk_thread_poll(thread, 0, 0);
        }
        pthread_mutex_unlock(&bdev_reactor->mutex);
    } while (!g_reactors_exit);

    /* free all the lightweight threads */
    pthread_mutex_lock(&bdev_reactor->mutex);
    TAILQ_FOREACH_SAFE(lw_thread, &bdev_reactor->threads, link, tmp) {
        thread = spdk_thread_get_from_ctx(lw_thread);
        TAILQ_REMOVE(&bdev_reactor->threads, lw_thread, link);
        spdk_set_thread(thread);
        spdk_thread_exit(thread);
        spdk_thread_destroy(thread);
    }
    pthread_mutex_unlock(&bdev_reactor->mutex);

    return 0;
}

static int
nvmf_schedule_spdk_thread(struct spdk_thread *thread)
{
    struct bdev_reactor *bdev_reactor;
    struct bdev_lw_thread *lw_thread;
    struct spdk_cpuset *cpumask;
    uint32_t i;

    /* Lightweight threads may have a requested cpumask.
     * This is a request only - the scheduler does not have to honor it.
     * For this scheduler implementation, each reactor is pinned to
     * a particular core so honoring the request is reasonably easy.
     */
    cpumask = spdk_thread_get_cpumask(thread);

    lw_thread = spdk_thread_get_ctx(thread);
    assert(lw_thread != NULL);
    memset(lw_thread, 0, sizeof(*lw_thread));

    /* assign lightweight threads to bdev reactor(core)
     * Here we use the mutex.The way the actual SPDK event framework
     * solves this is by using internal rings for messages between reactors
     */
    for (i = 0; i < spdk_env_get_core_count(); i++) {
        pthread_mutex_lock(&g_mutex);
        if (g_next_reactor == NULL) {
            g_next_reactor = TAILQ_FIRST(&g_reactors);
        }
        bdev_reactor = g_next_reactor;
        g_next_reactor = TAILQ_NEXT(g_next_reactor, link);
        pthread_mutex_unlock(&g_mutex);

        /* each spdk_thread has the core affinity */
        if (spdk_cpuset_get_cpu(cpumask, bdev_reactor->core)) {
            pthread_mutex_lock(&bdev_reactor->mutex);
            TAILQ_INSERT_TAIL(&bdev_reactor->threads, lw_thread, link);
            pthread_mutex_unlock(&bdev_reactor->mutex);
            break;
        }
    }

    if (i == spdk_env_get_core_count()) {
        fprintf(stderr, "failed to schedule spdk thread\n");
        return -1;
    }
    return 0;
}

static int
nvmf_init_threads(void)
{
    int rc;
    uint32_t i;
    char thread_name[32];
    struct bdev_reactor *bdev_reactor;
    struct spdk_thread *thread;
    struct spdk_cpuset cpumask;
    uint32_t master_core = spdk_env_get_current_core();

    /* Whenever SPDK creates a new lightweight thread it will call
     * nvmf_schedule_spdk_thread asking for the application to begin
     * polling it via spdk_thread_poll(). Each lightweight thread in
     * SPDK optionally allocates extra memory to be used by the application
     * framework. The size of the extra memory allocated is the second parameter.
     */
    spdk_thread_lib_init(nvmf_schedule_spdk_thread, sizeof(struct bdev_lw_thread));

    /* Spawn one system thread per CPU core. The system thread is called a reactor.
     * SPDK will spawn lightweight threads that must be mapped to reactors in
     * nvmf_schedule_spdk_thread. Using a single system thread per CPU core is a
     * choice unique to this application. SPDK itself does not require this specific
     * threading model. For example, another viable threading model would be
     * dynamically scheduling the lightweight threads onto a thread pool using a
     * work queue.
     */
    SPDK_ENV_FOREACH_CORE(i) {
        bdev_reactor = calloc(1, sizeof(struct bdev_reactor));
        if (!bdev_reactor) {
            fprintf(stderr, "failed to alloc bdev reactor\n");
            rc = -ENOMEM;
            goto err_exit;
        }

        bdev_reactor->core = i;
        pthread_mutex_init(&bdev_reactor->mutex, NULL);
        TAILQ_INIT(&bdev_reactor->threads);
        TAILQ_INSERT_TAIL(&g_reactors, bdev_reactor, link);

        if (i == master_core) {
            g_master_reactor = bdev_reactor;
            g_next_reactor = g_master_reactor;
        } else {
            rc = spdk_env_thread_launch_pinned(i,
                               bdev_reactor_run,
                               bdev_reactor);
            if (rc) {
                fprintf(stderr, "failed to pin reactor launch\n");
                goto err_exit;
            }
        }
    }

    /* Some SPDK libraries assume that there is at least some number of lightweight
     * threads that exist from the beginning of time. That assumption is currently
     * being removed from the SPDK libraries, but until that work is completed spawn
     * one lightweight thread per reactor here.
     */
    SPDK_ENV_FOREACH_CORE(i) {
        spdk_cpuset_zero(&cpumask);
        spdk_cpuset_set_cpu(&cpumask, i, true);
        snprintf(thread_name, sizeof(thread_name), "spdk_thread_%u", i);
        thread = spdk_thread_create(thread_name, &cpumask);
        if (!thread) {
            fprintf(stderr, "failed to create spdk thread\n");
            return -1;
        }
    }

    fprintf(stdout, "nvmf threads initlize successfully\n");
    return 0;

err_exit:
    return rc;
}

int main()
{
    int rc;
    struct spdk_env_opts opts;

    spdk_env_opts_init(&opts);
    opts.name = "multithread-example";
    if (spdk_env_init(&opts) < 0) {
        fprintf(stderr, "unable to initialize SPDK env\n");
        return -EINVAL;
    }

    rc = nvmf_init_threads();
    assert(rc == 0);

    struct spdk_thread* first_reader_thread =
            spdk_thread_create("first_reader_thread", NULL);

    if (first_reader_thread == NULL)
    {
        printf("First thread creation failed...\n");
        return 0 ;
    }

    struct spdk_thread* second_reader_thread =
            spdk_thread_create("second_reader_thread", NULL);
    if (second_reader_thread == NULL)
    {
        printf("Second thread creation failed...\n");
        return 0;
    }

    printf("first reader thread id is: %"PRIu64"\n",
           spdk_thread_get_id(first_reader_thread));
    printf("second reader thread id is: %"PRIu64"\n",
           spdk_thread_get_id(second_reader_thread));
    printf("Hello World!\n");
    return 0;
}

and the gdb back trace output is as follow:
#0 rte_mempool_get_bulk (n=1024, obj_table=0x7fffffffc0c0, mp=0x0) at /mnt/spdk/dpdk/build/include/rte_mempool.h:1538
#1  spdk_mempool_get_bulk (mp=0x0, ele_arr=0x7fffffffc0c0, count=1024) at env.c:279
#2  0x0000555555559af3 in spdk_thread_create (name=0x7fffffffe2a0 "spdk_thread_0", cpumask=<optimized out>) at thread.c:266
#3  0x00005555555592da in nvmf_init_threads () at main.c:169
#4  0x00005555555593e2 in main () at main.c:195