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

Code before:
    thread_id = (int)id;
    size_t jobs_to_process = 0;
    IOJobQueue *jq = &io_jobs[id];
    while (1) {
        struct timespec thread_uptime_start, thread_uptime_end;
        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &thread_uptime_start);
        /* Cancellation point so that pthread_cancel() from main thread is honored. */
        pthread_testcancel();

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

        /* 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]);
            clock_gettime(CLOCK_THREAD_CPUTIME_ID, &thread_uptime_end);
            atomic_fetch_add_explicit(&io_threads_uptime_us[id], timespec_diff_us(thread_uptime_start, thread_uptime_end), memory_order_relaxed);
            continue;
        }

        struct timespec work_start_time, work_end_time;
        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &work_start_time);

        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);
            /* Remove the job after it was processed */
            IOJobQueue_removeJob(jq);
        }
        /* Memory barrier to make sure the main thread sees the updated tail index.
         * We do it once per loop and not per tail-update for optimization reasons.
         * As the main-thread main concern is to check if the queue is empty, it's enough to do it once at the end. */
        atomic_thread_fence(memory_order_release);
        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &work_end_time);
        atomic_fetch_add_explicit(&io_threads_useful_time_us[id], timespec_diff_us(work_start_time, work_end_time), memory_order_relaxed);
        atomic_fetch_add_explicit(&io_threads_uptime_us[id], timespec_diff_us(thread_uptime_start, work_end_time), memory_order_relaxed);
    }
    pthread_cleanup_pop(0);
    return NULL;
}

long long getIOThreadUsefulTimeMicroseconds(int id) {
    return atomic_load_explicit(&io_threads_useful_time_us[id], memory_order_relaxed);

Review comment (suggestion):
Here, we call clock_gettime before and after the active work in each cycle.

I believe it's possible to do it only once per cycle, if there is work to do and no busy-loop spinning happens. Idea using pseudo-C code (using the CPU clock time as an integer):

```C
static void *IOThreadMain(void *myid) {
    // ...

    work_start_time = 0;
    while (1) {
        /* Cancellation point so that pthread_cancel() from main thread is honored. */
        pthread_testcancel();

        /* Active CPU time, excluding busy-wait loop */
        prev_start_time = work_start_time;
        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &work_start_time);
        if (prev_start_time != 0) {
            atomic_fetch_add_explicit(&io_threads_active_us[id],
                                      work_start_time - prev_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;
            }
            clock_gettime(CLOCK_THREAD_CPUTIME_ID, &work_start_time);
        }        

        // ...        
    }
}
```

Code after:
    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);
            /* Remove the job after it was processed */
            IOJobQueue_removeJob(jq);
        }
        /* Memory barrier to make sure the main thread sees the updated tail index.
         * We do it once per loop and not per tail-update for optimization reasons.
         * As the main-thread main concern is to check if the queue is empty, it's enough to do it once at the end. */
        atomic_thread_fence(memory_order_release);
    }
    pthread_cleanup_pop(0);
    return NULL;
}

long long getIOThreadActiveTimeMicroseconds(int id) {