threadpool_throughput.c

#include <stdlib.h>
#include <assert.h>
#include <uv.h>

int num_jobs;
#define NUM_ELEMS 1001

int g_terminate;
int g_jobs_done;
uv_work_t *jobs;
uint32_t buffer[NUM_ELEMS];

void cancel_jobs()
{
    for (int idx=0; idx < num_jobs; idx++) {
        uv_cancel((uv_req_t*)&jobs[idx]);
    }
}

void timer_callback(uv_timer_t *handle)
{
    g_terminate = 1;
    cancel_jobs();
    uv_close((uv_handle_t*)handle, NULL);
}

void work_callback(uv_work_t *req)
{
    uint64_t t_start = uv_hrtime();
    uint32_t elapsed_us;
    uint32_t limit_us = 521;
    uint64_t sum = 0;

    do {
        for (int idx=0; idx < NUM_ELEMS; idx++) {
            sum += buffer[idx];
        }

        elapsed_us = (uv_hrtime() - t_start) / 1000;
    } while (elapsed_us < limit_us);

    req->data = (void*)sum;
}

void after_work_callback(uv_work_t *req, int status)
{
    if (status==UV_ECANCELED)
        return;

    g_jobs_done++;

    if (!g_terminate)
        uv_queue_work(req->loop, req, work_callback, after_work_callback);
}

int main()
{
    for (int idx=0; idx < NUM_ELEMS; idx++) {
        buffer[idx] = rand();
    }

    char envbuf[16];
    size_t envsz = sizeof(envbuf);
    int rc = uv_os_getenv("UV_THREADPOOL_SIZE", envbuf, &envsz);
    int nthreads = 4;
    if (rc!=UV_ENOENT) nthreads = atoi(envbuf);
    num_jobs = nthreads * 50;
    jobs = malloc(sizeof(*jobs)*num_jobs);

    uv_loop_t *loop = uv_default_loop();

    uv_timer_t timer_handle;
    uv_timer_init(loop, &timer_handle);
    uv_timer_start(&timer_handle, timer_callback, 1000, 0);

    for (int idx=0; idx < num_jobs; idx++) {
        uv_queue_work(loop, &jobs[idx], work_callback, after_work_callback);
    }

    uv_run(loop, 0);
    rc = uv_loop_close(loop);
    assert(rc != UV_EBUSY);

    free(jobs);

    printf("jobs done: %d\n", g_jobs_done);

    return 0;
}