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

Code before:

static void *IOThreadMain(void *myid) {
    /* The ID is the thread ID number (from 1 to server.io_threads_num-1). ID 0 is the main thread. */
    long id = (long)myid;
    char thdname[32];

    snprintf(thdname, sizeof(thdname), "io_thd_%ld", id);
    valkey_set_thread_title(thdname);
    serverSetCpuAffinity(server.server_cpulist);
    initSharedQueryBuf();
    pthread_cleanup_push(freeSharedQueryBuf, NULL);

    thread_id = (int)id;
    size_t jobs_to_process = 0;
    IOJobQueue *jq = &io_jobs[id];
    monotime work_start_time = 0;
    monotime prev_work_start_time = 0;
    while (1) {
        /* Cancellation point so that pthread_cancel() from main thread is honored. */
        pthread_testcancel();

        prev_work_start_time = work_start_time;
        work_start_time = getMonotonicUs();
        if (prev_work_start_time != 0) {
            atomic_fetch_add_explicit(&used_active_time_io_thread[id], (work_start_time - prev_work_start_time), memory_order_relaxed);
        }

        jobs_to_process = IOJobQueue_availableJobs(jq);
        if (jobs_to_process == 0) {
            /* Wait for jobs */
            for (int j = 0; j < 1000000; j++) {
                jobs_to_process = IOJobQueue_availableJobs(jq);
                if (jobs_to_process) break;
            }
            work_start_time = getMonotonicUs();
        }

        /* Give the main thread a chance to stop this thread. */
        if (jobs_to_process == 0) {
            pthread_mutex_lock(&io_threads_mutex[id]);
            pthread_mutex_unlock(&io_threads_mutex[id]);
            continue;
        }

        for (size_t j = 0; j < jobs_to_process; j++) {
            job_handler handler;
            void *data;
            /* We keep the job in the queue until it's processed. This ensures that if the main thread checks
             * and finds the queue empty, it can be certain that the IO thread is not currently handling any job. */
            IOJobQueue_peek(jq, &handler, &data);
            handler(data);

Review comment (suggestion):
(Updated) suggestion about how to exclude time the thread is inactivated and hangs waiting for mutex.

* No jobs to process in the last cycle means no active work was done.
* prev_work_start_time can be a local variable inside the loop. It's only used very temporarily.
* Line length.

```suggestion
        monotime prev_work_start_time = work_start_time;
        work_start_time = getMonotonicUs();
        if (jobs_to_process != 0) {
            atomic_fetch_add_explicit(&used_active_time_io_thread[id],
                                      work_start_time - prev_work_start_time,
                                      memory_order_relaxed);
        }
```

Code after:

static void *IOThreadMain(void *myid) {
    /* The ID is the thread ID number (from 1 to server.io_threads_num-1). ID 0 is the main thread. */
    long id = (long)myid;
    char thdname[32];

    snprintf(thdname, sizeof(thdname), "io_thd_%ld", id);
    valkey_set_thread_title(thdname);
    serverSetCpuAffinity(server.server_cpulist);
    initSharedQueryBuf();
    pthread_cleanup_push(freeSharedQueryBuf, NULL);

    thread_id = (int)id;
    size_t jobs_to_process = 0;
    IOJobQueue *jq = &io_jobs[id];
    monotime work_start_time = 0;
    while (1) {
        /* Cancellation point so that pthread_cancel() from main thread is honored. */
        pthread_testcancel();
        monotime prev_work_start_time = work_start_time;
        work_start_time = getMonotonicUs();
        if (jobs_to_process != 0) {
            atomic_fetch_add_explicit(&used_active_time_io_thread[id],
                                      work_start_time - prev_work_start_time,
                                      memory_order_relaxed);
        }

        jobs_to_process = IOJobQueue_availableJobs(jq);
        if (jobs_to_process == 0) {
            /* Wait for jobs */
            for (int j = 0; j < 1000000; j++) {
                jobs_to_process = IOJobQueue_availableJobs(jq);
                if (jobs_to_process) break;
            }
            work_start_time = getMonotonicUs();
        }

        /* Give the main thread a chance to stop this thread. */
        if (jobs_to_process == 0) {
            pthread_mutex_lock(&io_threads_mutex[id]);
            pthread_mutex_unlock(&io_threads_mutex[id]);
            continue;
        }

        for (size_t j = 0; j < jobs_to_process; j++) {
            job_handler handler;
            void *data;
            /* We keep the job in the queue until it's processed. This ensures that if the main thread checks
             * and finds the queue empty, it can be certain that the IO thread is not currently handling any job. */
            IOJobQueue_peek(jq, &handler, &data);
            handler(data);