progettoso/lib/threadpool/threadpool.c

/**
 * @file threadpool.c
 * @brief File di implementazione dell'interfaccia Threadpool
 */

#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <errno.h>

#include "util.h"
#include "threadpool.h"

/**
 * @function void *threadpool_thread(void *threadpool)
 * @brief funzione eseguita dal thread worker che appartiene al pool
 */
static void *workerpool_thread(void *threadpool) {    
    threadpool_t *pool = (threadpool_t *)threadpool; // cast
    taskfun_t task;  // generic task
    pthread_t self = pthread_self();
    int myid = -1;

    // non efficiente, si puo' fare meglio.....
    do {
	for (int i=0;i<pool->numthreads;++i)
	    if (pthread_equal(pool->threads[i], self)) {
		myid = i;
		break;
	    }
    } while (myid < 0);

    LOCK_RETURN(&(pool->lock), NULL);
    for (;;) {
        // in attesa di un messaggio, controllo spurious wakeups.
        while((pool->count == 0) && (!pool->exiting)) {
            pthread_cond_wait(&(pool->cond), &(pool->lock));
	}

        if (pool->exiting > 1) break; // exit forzato, esco immediatamente
	// devo uscire ma ci sono messaggi pendenti 
	if (pool->exiting == 1 && !pool->count) break;  

	// nuovo task
        task.fun = pool->pending_queue[pool->head].fun;
        task.arg = pool->pending_queue[pool->head].arg;

        pool->head++; pool->count--;
        pool->head = (pool->head == abs(pool->queue_size)) ? 0 : pool->head;

	pool->taskonthefly++;
        UNLOCK_RETURN(&(pool->lock), NULL);

        // eseguo la funzione 
        (*(task.fun))(task.arg);
	
	LOCK_RETURN(&(pool->lock), NULL);
	pool->taskonthefly--;
    }
    UNLOCK_RETURN(&(pool->lock), NULL);

    fprintf(stderr, "thread %d exiting\n", myid);
    return NULL;
}



static int freePoolResources(threadpool_t *pool) {
    if(pool->threads) {
        free(pool->threads);
        free(pool->pending_queue);
	
        pthread_mutex_destroy(&(pool->lock));
        pthread_cond_destroy(&(pool->cond));
    }
    free(pool);    
    return 0;
}

threadpool_t *createThreadPool(int numthreads, int pending_size) {
    if(numthreads <= 0 || pending_size < 0) {
	errno = EINVAL;
        return NULL;
    }
    
    threadpool_t *pool = (threadpool_t *)malloc(sizeof(threadpool_t));
    if (pool == NULL) return NULL;

    // condizioni iniziali
    pool->numthreads   = 0;
    pool->taskonthefly = 0;
    pool->queue_size = (pending_size == 0 ? -1 : pending_size);
    pool->head = pool->tail = pool->count = 0;
    pool->exiting = 0;

    /* Allocate thread and task queue */
    pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * numthreads);
    if (pool->threads == NULL) {
	free(pool);
	return NULL;
    }
    pool->pending_queue = (taskfun_t *)malloc(sizeof(taskfun_t) * abs(pool->queue_size));
    if (pool->pending_queue == NULL) {
	free(pool->threads);
	free(pool);
	return NULL;
    }
    if ((pthread_mutex_init(&(pool->lock), NULL) != 0) ||
	(pthread_cond_init(&(pool->cond), NULL) != 0))  {
	free(pool->threads);
	free(pool->pending_queue);
	free(pool);
	return NULL;
    }
    for(int i = 0; i < numthreads; i++) {
        if(pthread_create(&(pool->threads[i]), NULL,
                          workerpool_thread, (void*)pool) != 0) {
	    /* errore fatale, libero tutto forzando l'uscita dei threads */
            destroyThreadPool(pool, 1);
	    errno = EFAULT;
            return NULL;
        }
        pool->numthreads++;
    }
    return pool;
}


int destroyThreadPool(threadpool_t *pool, int force) {    
    if(pool == NULL || force < 0) {
	errno = EINVAL;
	return -1;
    }

    LOCK_RETURN(&(pool->lock), -1);

    pool->exiting = 1 + force;

    if (pthread_cond_broadcast(&(pool->cond)) != 0) {
      UNLOCK_RETURN(&(pool->lock),-1);
      errno = EFAULT;
      return -1;
    }
    UNLOCK_RETURN(&(pool->lock), -1);

    for(int i = 0; i < pool->numthreads; i++) {
	if (pthread_join(pool->threads[i], NULL) != 0) {
	    errno = EFAULT;
	    UNLOCK_RETURN(&(pool->lock),-1);
	    return -1;
	}
    }
    freePoolResources(pool);
    return 0;
}

int addToThreadPool(threadpool_t *pool, void (*f)(void *), void *arg) {
    if(pool == NULL || f == NULL) {
	errno = EINVAL;
	return -1;
    }

    LOCK_RETURN(&(pool->lock), -1);
    int queue_size = abs(pool->queue_size);
    int nopending  = (pool->queue_size == -1); // non dobbiamo gestire messaggi pendenti

    // coda piena o in fase di uscita
    if (pool->count >= queue_size || pool->exiting)  {
      UNLOCK_RETURN(&(pool->lock),-1);
      return 1; // esco con valore "coda piena"
    }
    
    if (pool->taskonthefly >= pool->numthreads) {
	if (nopending) { 
	    // tutti i thread sono occupati e non si gestiscono task pendenti
	    assert(pool->count == 0);
	    
	    UNLOCK_RETURN(&(pool->lock),-1);
	    return 1;  // esco con valore "coda piena"
	}
    } 
    
    pool->pending_queue[pool->tail].fun = f;
    pool->pending_queue[pool->tail].arg = arg;
    pool->count++;    
    pool->tail++;
    if (pool->tail >= queue_size) pool->tail = 0;
    
    int r;
    if((r=pthread_cond_signal(&(pool->cond))) != 0) {
      UNLOCK_RETURN(&(pool->lock),-1);
      errno = r;
      return -1;
    }
    
    UNLOCK_RETURN(&(pool->lock),-1);
    return 0;
}


/**
 * @function void *thread_proxy(void *argl)
 * @brief funzione eseguita dal thread worker che non appartiene al pool
 */
static void *proxy_thread(void *arg) {    
    taskfun_t *task = (taskfun_t*)arg;
    // eseguo la funzione 
    (*(task->fun))(task->arg);
    
    free(task);
    return NULL;
}

// fa lo spawn di un thread in modalità detached
int spawnThread(void (*f)(void*), void* arg) {
    if (f == NULL) {
	errno = EINVAL;
	return -1;
    }

    taskfun_t *task = malloc(sizeof(taskfun_t));   // la memoria verra' liberata dal proxy 
    if (!task) return -1;
    task->fun = f;
    task->arg = arg;

    pthread_t thread;
    pthread_attr_t attr;
    if (pthread_attr_init(&attr) != 0) return -1;
    if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) != 0) return -1;
    if (pthread_create(&thread, &attr,
		      proxy_thread, (void*)task) != 0) {
	free(task);
	errno = EFAULT;
	return -1;
    }
    return 0;
}