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

Code before:
        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) {
    server.active_io_threads_num = 1; /* We start with threads not active. */
    server.io_poll_state = AE_IO_STATE_NONE;
    server.io_ae_fired_events = 0;

    /* Don't spawn any thread if the user selected a single thread:
     * we'll handle I/O directly from the main thread. */
    if (server.io_threads_num == 1) return;

    serverAssert(server.io_threads_num <= IO_THREADS_MAX_NUM);

    prefetchCommandsBatchInit();

Review comment (performance):
I think it would be safer to first drain all the queues, regardless of whether we increase or decrease the number of threads. Since we don't expect this to happen very often, performance is not an issue.

Code after:
}

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;
}

/* Initialize the data structures needed for I/O threads. */
void initIOThreads(void) {
    server.active_io_threads_num = 1; /* We start with threads not active. */
    server.io_poll_state = AE_IO_STATE_NONE;
    server.io_ae_fired_events = 0;

    /* Don't spawn any thread if the user selected a single thread:
     * we'll handle I/O directly from the main thread. */
    if (server.io_threads_num == 1) return;

    serverAssert(server.io_threads_num <= IO_THREADS_MAX_NUM);