ui/common/threadpool.c

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2024 Olaf Wintermann. All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   1. Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *
 *   2. Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in the
 *      documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#ifndef _WIN32

#include "threadpool.h"
#include "context.h"

#include <pthread.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>

static threadpool_job kill_job;

UiThreadpool* threadpool_new(int min, int max) {
    UiThreadpool *pool = malloc(sizeof(UiThreadpool));
    pool->queue = NULL;
    pool->queue_len = 0;
    pool->num_idle = 0;
    pool->min_threads = min;
    pool->max_threads = max;

    pthread_mutex_init(&pool->queue_lock, NULL);
    pthread_mutex_init(&pool->avlbl_lock, NULL);
    pthread_cond_init(&pool->available, NULL);  

    return pool;
}

int threadpool_start(UiThreadpool *pool) {
    pool->nthreads = pool->min_threads;
    pool->threads = calloc(pool->max_threads, sizeof(pthread_t));
    /* create pool threads */
    for(int i=0;i<pool->nthreads;i++) {
        if (pthread_create(&pool->threads[i], NULL, threadpool_func, pool) != 0) {
            fprintf(stderr, "uic: threadpool_start: pthread_create failed: %s", strerror(errno));
            return 1;
        }
    }
    return 0;
}

int threadpool_join(UiThreadpool *pool) {
    int err = 0;
    for(int i=0;i<pool->nthreads;i++) {
        if(pthread_join(pool->threads[i], NULL)) {
            err = 1;
        }
    }
    return err;
}

void* threadpool_func(void *data) {
    UiThreadpool *pool = (UiThreadpool*)data;
    
    for(;;) {
        threadpool_job *job = threadpool_get_job(pool);
        if(job == &kill_job) {
            break;
        }
        
        job->callback(job->data);

        free(job);
    }
    return NULL;
}

threadpool_job* threadpool_get_job(UiThreadpool *pool) {
    pthread_mutex_lock(&pool->queue_lock);

    threadpool_job *job = NULL;
    pool->num_idle++;
    while(job == NULL) {
        if(pool->queue_len == 0) {
            pthread_cond_wait(&pool->available, &pool->queue_lock);
            continue;
        } else {
            pool_queue_t *q = pool->queue;
            job = q->job;
            pool->queue = q->next;
            pool->queue_len--;
            free(q);
        }
    }
    pool->num_idle--;

    pthread_mutex_unlock(&pool->queue_lock);
    return job;
}

void threadpool_run(UiThreadpool *pool, job_callback_f func, void *data) {
    // TODO: handle errors
    
    threadpool_job *job = malloc(sizeof(threadpool_job));
    job->callback = func;
    job->data = data;

    pthread_mutex_lock(&pool->queue_lock);
    threadpool_enqueue_job(pool, job);

    int create_thread = 0;
    int destroy_thread = 0;
    int diff = pool->queue_len - pool->num_idle;
    
    //if(pool->queue_len == 1) {
    pthread_cond_signal(&pool->available);
    //}
    
    pthread_mutex_unlock(&pool->queue_lock);
}

void threadpool_enqueue_job(UiThreadpool *pool, threadpool_job *job) {
    pool_queue_t *q = malloc(sizeof(pool_queue_t));
    q->job = job;
    q->next = NULL;
    
    if(pool->queue == NULL) {
        pool->queue = q;
    } else {
        pool_queue_t *last_elem = pool->queue;
        while(last_elem->next != NULL) {
            last_elem = last_elem->next;
        }
        last_elem->next = q;
    }
    pool->queue_len++;
}


UiThreadpool* ui_threadpool_create(int nthreads) {
    UiThreadpool *pool = threadpool_new(nthreads, nthreads);
    threadpool_start(pool); // TODO: check return value
    return pool;
}

void ui_threadpool_destroy(UiThreadpool* pool) {
    
}

static int ui_threadpool_job_finish(void *data) {
    UiJob *job = data;
    UiEvent event;
    event.obj = job->obj;
    event.window = job->obj->window;
    event.document = job->obj->ctx->document;
    event.intval = 0;
    event.eventdata = NULL;
    event.eventdatatype = 0;
    job->finish_callback(&event, job->finish_data);
    free(job);
    return 0;
}

static void* ui_threadpool_job_func(void *data) {
    UiJob *job = data;
    if (!job->job_func(job->job_data) && job->finish_callback) {
        ui_call_mainthread(ui_threadpool_job_finish, job);
    } else {
        free(job);
    }
    return NULL;
}

void ui_threadpool_job(UiThreadpool* pool, UiObject* obj, ui_threadfunc tf, void* td, ui_callback f, void* fd) {
    UiJob* job = malloc(sizeof(UiJob));
    job->obj = obj;
    job->job_func = tf;
    job->job_data = td;
    job->finish_callback = f;
    job->finish_data = fd;
    threadpool_run(pool, ui_threadpool_job_func, job);
}


#endif