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

Code before:
    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();

    /* Spawn and initialize the I/O threads. */
    for (int i = 1; i < server.io_threads_num; i++) {
        createIOThread(i);
    }
}

int trySendReadToIOThreads(client *c) {

Review comment (suggestion):
It would be safer to just reset it to 1 (server.active_io_threads_num = 1), and later we can update it according to the workload.

Code after:
            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);

    prefetchCommandsBatchInit();

    /* Spawn and initialize the I/O threads. */
    for (int i = 1; i < server.io_threads_num; i++) {
        createIOThread(i);
    }
}