Repository: valkey-io/valkey
File: src/io_threads.c (C)

Code before:
    if ((err = pthread_join(tid, NULL)) != 0) {
        serverLog(LL_WARNING, "IO thread(tid:%lu) can not be joined: %s", (unsigned long)tid, strerror(err));
    } else {
        serverLog(LL_NOTICE, "IO thread(tid:%lu) terminated", (unsigned long)tid);
    }

    IOJobQueue_cleanup(&io_jobs[id]);
}

void killIOThreads(void) {
    for (int j = 1; j < server.io_threads_num; j++) { /* We don't kill thread 0, which is the main thread. */
        shutdownIOThread(j);
    }
}

int updateIOThreads(const char **err) {
    serverAssert(inMainThread());
    UNUSED(err);
    int prev_threads_num = 1;
    for (int i = IO_THREADS_MAX_NUM - 1; i > 0; i--) {
        if (io_threads[i]) {
            prev_threads_num = i + 1;
            break;
        }
    }
    if (prev_threads_num == server.io_threads_num) goto done;
    serverLog(LL_DEBUG, "Changing number of IO threads from %d to %d.", prev_threads_num, server.io_threads_num);
    // Create new threads.
    if (server.io_threads_num > prev_threads_num) {
        for (int i = prev_threads_num; i < server.io_threads_num; i++) {
            createIOThread(i);
        }
    }
    // Decrease the number of threads.
    else {
        for (int i = prev_threads_num - 1; i >= server.io_threads_num; i--) {
            drainIOThreadsQueueInternal(i);
            // Unblock inactive thread.
            if (i >= server.active_io_threads_num)
                pthread_mutex_unlock(&io_threads_mutex[i]);
            shutdownIOThread(i);
            io_threads[i] = 0;
        }
    }
    server.active_io_threads_num = min(server.active_io_threads_num, server.io_threads_num);
done:
    return 1;
}

/* Initialize the data structures needed for I/O threads. */
void initIOThreads(void) {

Review comment (suggestion):
```suggestion
    if (prev_threads_num == server.io_threads_num) return 1;
```
There is no other logic in done, so we can just return here.

Code after:
    } else {
        serverLog(LL_NOTICE, "IO thread(tid:%lu) terminated", (unsigned long)tid);
    }
    pthread_mutex_destroy(&io_threads_mutex[id]);
    IOJobQueue_cleanup(&io_jobs[id]);
}

void killIOThreads(void) {
    for (int j = 1; j < server.io_threads_num; j++) { /* We don't kill thread 0, which is the main thread. */
        shutdownIOThread(j);
    }
}

int updateIOThreads(const char **err) {
    serverAssert(inMainThread());
    UNUSED(err);
    int prev_threads_num = 1;
    for (int i = IO_THREADS_MAX_NUM - 1; i > 0; i--) {
        if (io_threads[i]) {
            prev_threads_num = i + 1;
            break;
        }
    }
    if (prev_threads_num == server.io_threads_num) return 1;

    serverLog(LL_NOTICE, "Changing number of IO threads from %d to %d.", prev_threads_num, server.io_threads_num);
    drainIOThreadsQueue();
    /* Set active threads to 1, will be adjusted based on workload later. */
    for (int i = 1; i < server.active_io_threads_num; i++) {
        pthread_mutex_lock(&io_threads_mutex[i]);
    }
    server.active_io_threads_num = 1;

    // Create new threads.
    if (server.io_threads_num > prev_threads_num) {
        for (int i = prev_threads_num; i < server.io_threads_num; i++) {
            createIOThread(i);
        }
    }
    // Decrease the number of threads.
    else {
        for (int i = prev_threads_num - 1; i >= server.io_threads_num; i--) {
            // Unblock inactive thread.
            pthread_mutex_unlock(&io_threads_mutex[i]);
            shutdownIOThread(i);
            io_threads[i] = 0;
        }
    }
    return 1;
}