src/server/util/thrpool.c

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2013 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.
 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "atomic.h"
#include "thrpool.h"

static threadpool_job kill_job;

threadpool_t* threadpool_new(int min, int max) {
    log_ereport(LOG_INFORM, "new threadpool (min: %d, max: %d)", min, max);
    threadpool_t *pool = malloc(sizeof(threadpool_t));
    pool->queue = NULL;
    pool->queue_len = 0;
    pool->num_idle = 0;
    pool->min_threads = min;
    pool->max_threads = max;
    pool->num_threads = 0;
    pool->last_job = 0;
    pool->last_thread = -1;
    
    pool->threads = calloc(max, sizeof(pthread_t));
    pool->thrstatus = calloc(max, sizeof(int));

    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(threadpool_t *pool) {
    /* create pool threads */
    for(int i=0;i<pool->min_threads;i++) {
        if (pthread_create(pool->threads + i, NULL, threadpool_func, pool) != 0) {
            log_ereport(LOG_FAILURE, "threadpool_start: pthread_create failed: %s", strerror(errno));
            return 1;
        }
    }
    return 0;
}

void* threadpool_func(void *data) {
    threadpool_t *pool = (threadpool_t*)data;
    
    pthread_t thr_self = pthread_self();
    int thr_index = -1;
    for(int i=0;i<pool->max_threads;i++) {
        if(pool->threads[i] == thr_self) {
            thr_index = i;
            break;
        }
    }
    
    if(thr_index == -1) {
        log_ereport(LOG_CATASTROPHE, "threadpool: cannot find thread index for thread %ull\n", (unsigned long long)thr_self);
        return NULL;
    }
    
    ws_atomic_inc32(&pool->num_threads);
    for(;;) {
        threadpool_job *job = threadpool_get_job(pool, thr_index);
        if(job == &kill_job) {
            break;
        }

        job->callback(job->data);

        free(job);
    }
    uint32_t nthreads = ws_atomic_dec32(&pool->num_threads);
    if(nthreads == 0) {
        log_ereport(LOG_INFORM, "threadpool closed"); // TODO: log threadpool name
    }
    return NULL;
}

threadpool_job* threadpool_get_job(threadpool_t *pool, int thread_index) {
    struct timespec timeout;
    clock_gettime(CLOCK_REALTIME, &timeout);
    timeout.tv_sec += 30;
    
    while(pthread_mutex_timedlock(&pool->queue_lock, &timeout)) {
        log_ereport(LOG_INFORM, "threadpool_get_job: mutex timeout");
        timeout.tv_sec += 30;
    }

    threadpool_job *job = NULL;
    pool->num_idle++;
    while(job == NULL) {
        if(pool->queue_len == 0) {
            timeout.tv_sec += 30;
            while(pthread_cond_timedwait(&pool->available, &pool->queue_lock, &timeout)) {
                log_ereport(LOG_DEBUG, "threadpool_get_job: cond timeout: thread: %d queue: %u", thread_index, (unsigned int)pool->queue_len);
                timeout.tv_sec += 60;
            }
            continue;
        } else {
            pool_queue_t *q = pool->queue;
            job = q->job;
            pool->queue = q->next;
            pool->queue_len--;
            free(q);
        }
    }
    pool->num_idle--;
    
    pool->last_thread = thread_index;
    pool->last_job = time(NULL);

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

void threadpool_run(threadpool_t *pool, job_callback_f func, void *data) {
    // TODO: handle errors
    
    if(pool->num_threads == 0) {
        threadpool_start(pool);
    }
    
    threadpool_job *job = malloc(sizeof(threadpool_job));
    job->callback = func;
    job->data = data;

    struct timespec timeout;
    clock_gettime(CLOCK_REALTIME, &timeout);
    timeout.tv_sec += 30;
    
    while(pthread_mutex_timedlock(&pool->queue_lock, &timeout)) {
        log_ereport(LOG_INFORM, "threadpool_run: mutex timeout");
        timeout.tv_sec += 30;
    }
    threadpool_enqueue_job(pool, job);

    int create_thread = 0;
    int destroy_thread = 0;
    int diff = pool->queue_len - pool->num_idle;
    if(diff > 0 && pool->num_threads < pool->max_threads) {
        create_thread = 1;
    } else if(diff < 0 && pool->num_threads > pool->min_threads) {
        destroy_thread = 1;
    }
    
    //if(pool->queue_len == 1) {
    pthread_cond_signal(&pool->available);
    //}
    
    if(create_thread) {
        pthread_t t;
        if (pthread_create(&t, NULL, threadpool_func, pool) != 0) {
            log_ereport(LOG_FAILURE, "threadpool_run: pthread_create failed: %s", strerror(errno));
        }
    }
    if(destroy_thread) {
        threadpool_enqueue_job(pool, &kill_job);
        pthread_cond_signal(&pool->available);
    }
    
    // some diagnostics:
    // if the queue has multiple elements, but the last job was started
    // over a minute ago, print some diagnostic message, because
    // this does look wrong
    if(pool->queue_len > 5 && pool->last_job != 0) {
        // reuse timeout sec value, because we don't need the most accurate
        // time value here
        time_t current = timeout.tv_sec;
        if(pool->last_job + 60 < current) {
            // looks like the threadpool is blocked
            struct tm lastjob;
            localtime_r(&pool->last_job, &lastjob);
            log_ereport(
                    LOG_WARN,
                    "high threadpool wait time: queue: %u lastjob: %02d:%02d:%02d",
                    (unsigned int)pool->queue_len,
                    lastjob.tm_hour, lastjob.tm_min, lastjob.tm_sec);
        }
    }
    
    pthread_mutex_unlock(&pool->queue_lock);
}

void threadpool_enqueue_job(threadpool_t *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++;
}

void threadpool_shutdown(threadpool_t *pool) {
    int nthreads = pool->max_threads;
    for(int i=0;i<nthreads;i++) {
        pthread_mutex_lock(&pool->queue_lock);
        threadpool_enqueue_job(pool, &kill_job);
        pthread_cond_signal(&pool->available);
        pthread_mutex_unlock(&pool->queue_lock);
    }
}