thread.c

Go to the documentation of this file.

/*
Copyright (C) 2012  Gregory Nietsky <gregory@distrotetch.co.za>
        http://www.distrotech.co.za

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include <pthread.h>
#include <signal.h>
#include <unistd.h>
#include <stdint.h>

#include "include/dtsapp.h"

enum threadopt {
    TL_THREAD_NONE      = 1 << 0,
    TL_THREAD_RUN       = 1 << 1,
    TL_THREAD_DONE      = 1 << 2,
    TL_THREAD_JOIN      = 1 << 3,
    TL_THREAD_STOP      = 1 << 4,

    TL_THREAD_CAN_CANCEL    = 1 << 16,
    TL_THREAD_JOINABLE  = 1 << 17,
    TL_THREAD_RETURN    = 1 << 18
};

struct thread_pvt {
    void            *data;
    pthread_t       thr;
    threadcleanup       cleanup;
    threadfunc      func;
    threadsighandler    sighandler;
    enum                    threadopt flags;
};

struct threadcontainer {
    struct bucket_list  *list;
    struct thread_pvt   *manager;
};

struct threadcontainer *threads = NULL;

int thread_can_start = 1;

static int32_t hash_thread(const void *data, int key) {
    const struct thread_pvt *thread = data;
    const pthread_t *th = (key) ? data : &thread->thr;
    return jenhash(th, sizeof(pthread_t), 0);
}

static void close_threads(void *data) {
    struct threadcontainer *tc = data;

    if (tc->list) {
        objunref(tc->list);
    }

    if (tc->manager) {
        objunref(tc->manager);
        tc->manager = NULL;
    }
    threads = NULL;
}

static void free_thread(void *data) {
    struct thread_pvt *thread = data;

    if (thread->data) {
        objunref(thread->data);
    }
}

static struct thread_pvt *get_thread_from_id() {
    struct thread_pvt *thr;
    struct threadcontainer *tc;
    pthread_t me;

    if (!(tc = (objref(threads)) ? threads : NULL)) {
        return NULL;
    }
    me = pthread_self();

    objlock(tc);
    thr = bucket_list_find_key(tc->list, &me);
    objunlock(tc);
    objunref(tc);
    return thr;
}


extern int framework_threadok() {
    struct thread_pvt *thr;
    int ret;

    thr = get_thread_from_id();
    ret =(thr) ? testflag(thr, TL_THREAD_RUN) : 0;
    objunref(thr);

    return ret;
}

/*
 * close all threads when we get SIGHUP
 */
static int manager_sig(int sig, void *data) {
#ifndef __WIN32
    struct thread_pvt *thread;

    if (!(thread = get_thread_from_id())) {
        return 0;
    }


    switch(sig) {
        case SIGHUP:
            clearflag(thread, TL_THREAD_RUN);
            break;
        case SIGINT:
        case SIGTERM:
            clearflag(thread, TL_THREAD_RUN);
            setflag(thread, TL_THREAD_STOP);
    }
    objunref(thread);
    return 1;
#else
    return 0;
#endif
}

/* if im here im the last thread*/
static void manage_clean(void *data) {

    /*make sure im still here when turning off*/
    objlock(threads);
    thread_can_start = 0;
    objunlock(threads);
    objunref(threads);
}

static void stop_threads(void *data, void *data2) {
    struct thread_pvt *thread = data;
    struct thread_pvt *man = data2;

    /*Dont footbullet*/
    if (!pthread_equal(man->thr, thread->thr)) {
        if (thread->sighandler) {
            pthread_kill(thread->thr, SIGTERM);
        }
        if (testflag(thread, TL_THREAD_CAN_CANCEL) && testflag(thread, TL_THREAD_RUN)) {
            pthread_cancel(thread->thr);
        }
        clearflag(thread, TL_THREAD_RUN);
    }
}

/*
 * loop through all threads till they stoped
 * setting stop will flag threads to stop
 */
static void *managethread(void *data) {
    struct thread_pvt *thread;
    int last = 0;

    if (!(thread = get_thread_from_id())) {
        return NULL;
    }

    for(;;) {
        /*if im the last one leave this is done locked to make sure no items are added/removed*/
        objlock(threads);
        if (!(bucket_list_cnt(threads->list) - last)) {
            if (threads->manager) {
                objunref(threads->manager);
                threads->manager = NULL;
            }
            objunlock(threads);
            objunref(thread);
            break;
        }
        objunlock(threads);

        /* Ive been joined so i can leave when im alone*/
        if (testflag(thread, TL_THREAD_JOIN)) {
            clearflag(thread, TL_THREAD_JOIN);
            last = 1;
        }

        /*Cancel all running threads*/
        if (testflag(thread, TL_THREAD_STOP)) {
            clearflag(thread, TL_THREAD_STOP);
            /* Stop any more threads*/
            objlock(threads);
            if (threads->manager) {
                objunref(threads->manager);
                threads->manager = NULL;
            }
            objunlock(threads);

            /* cancel all threads now that they stoped*/
            bucketlist_callback(threads->list, stop_threads, thread);
            last = 1;
        }
#ifdef __WIN32__
        Sleep(1000);
#else
        sleep(1);
#endif
    }
    return NULL;
}

extern int startthreads(void) {
    struct threadcontainer *tc;

    tc = (objref(threads)) ? threads : NULL;

    if (tc) {
        objunref(tc);
        return 1;
    }

    if (!(tc = objalloc(sizeof(*threads), close_threads))) {
        return 0;
    }

    if (!tc->list && !(tc->list = create_bucketlist(4, hash_thread))) {
        objunref(tc);
        return 0;
    }

    threads = tc;
    if (!(tc->manager = framework_mkthread(managethread, manage_clean, manager_sig, NULL, THREAD_OPTION_JOINABLE | THREAD_OPTION_RETURN))) {
        objunref(tc);
        return 0;
    }

    return 1;
}

extern void stopthreads(int join) {
    struct threadcontainer *tc;

    tc = (objref(threads)) ? threads : NULL;
    if (!tc) {
        return;
    }

    objlock(tc);
    if (tc->manager) {
        setflag(tc->manager, TL_THREAD_STOP);
        if (join) {
            setflag(tc->manager, TL_THREAD_JOIN);
            objunlock(tc);
            pthread_join(tc->manager->thr, NULL);
        } else {
            objunlock(tc);
        }
    } else {
        objunlock(tc);
    }
    objunlock(tc);
    objunref(tc);
}

static void thread_cleanup(void *data) {
    struct thread_pvt *thread = data;

    /*remove from thread list manager unrefs threads in cleanup run 1st*/
    remove_bucket_item(threads->list, thread);

    /*Run cleanup*/
    clearflag(thread, TL_THREAD_RUN);
    setflag(thread, TL_THREAD_DONE);
    if (thread->cleanup) {
        thread->cleanup(thread->data);
    }

    /*remove thread reference*/
    objunref(thread);
}

static void *threadwrap(void *data) {
    struct thread_pvt *thread = data;
    void *ret = NULL;
    int cnt;

    objref(thread);

    for(cnt = 0;!testflag(thread, TL_THREAD_RUN) && (cnt < 100); cnt++) {
        usleep(1000);
    }

    if (cnt == 100) {
        return NULL;
    }

    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
    if (!testflag(thread, TL_THREAD_CAN_CANCEL)) {
        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
    }

    if (!testflag(thread, TL_THREAD_JOINABLE)) {
        pthread_detach(thread->thr);
    }

    pthread_cleanup_push(thread_cleanup, thread);
    ret = thread->func(thread->data);
    pthread_cleanup_pop(1);

    return (ret);
}

extern struct thread_pvt *framework_mkthread(threadfunc func, threadcleanup cleanup, threadsighandler sig_handler, void *data, int flags) {
    struct thread_pvt *thread;
    struct threadcontainer *tc = NULL;

    /*Grab a reference for threads in this scope start up if we can*/
    if (!(tc = (objref(threads)) ? threads : NULL)) {
        if (!thread_can_start) {
            return NULL;
        } else if (!startthreads()) {
            return NULL;
        }
        if (!(tc = (objref(threads)) ? threads : NULL)) {
            return NULL;
        }
    }

    objlock(tc);
    /* dont allow threads if no manager or it not started*/
    if ((!tc->manager || !func) && (func != managethread)) {
        /*im shuting down*/
        objunlock(tc);
        objunref(tc);
        return NULL;
    } else if (!(thread = objalloc(sizeof(*thread), free_thread))) {
        /* could not create*/
        objunlock(tc);
        objunref(tc);
        return NULL;
    }

    thread->data = (objref(data)) ? data : NULL;
    thread->flags = flags << 16;
    thread->cleanup = cleanup;
    thread->sighandler = sig_handler;
    thread->func = func;
    objunlock(tc);

    /* start thread and check it*/
    if (pthread_create(&thread->thr, NULL, threadwrap, thread) || pthread_kill(thread->thr, 0)) {
        objunref(thread);
        objunref(tc);
        return NULL;
    }

    /*Activate the thread it needs to be flaged to run or it will die*/
    objlock(tc);
    addtobucket(tc->list, thread);
    setflag(thread, TL_THREAD_RUN);
    objunlock(tc);
    objunref(tc);

    if (testflag(thread, TL_THREAD_RETURN)) {
        return thread;
    } else {
        objunref(thread);
        return NULL;
    }
}

extern void jointhreads(void) {
    struct threadcontainer *tc;

    tc = (objref(threads)) ? threads : NULL;
    if (!tc) {
        return;
    }

    objlock(tc);
    if (tc->manager) {
        setflag(tc->manager, TL_THREAD_JOIN);
        objunlock(tc);
        pthread_join(tc->manager->thr, NULL);
    } else {
        objunlock(tc);
    }
    objunref(tc);
}


#ifndef __WIN32
static int handle_thread_signal(struct thread_pvt *thread, int sig) {
    int ret;

    if (thread->sighandler) {
        thread->sighandler(sig, thread->data);
        ret = 1;
    } else {
        ret = -1;
    }
    return ret;
}
#endif

extern int thread_signal(int sig) {
    int ret = 0;
#ifndef __WIN32
    struct thread_pvt *thread = NULL;

    if (!(thread = get_thread_from_id())) {
        return 0;
    }

    switch(sig) {
        case SIGUSR1:
        case SIGUSR2:
        case SIGHUP:
        case SIGALRM:
            ret = handle_thread_signal(thread, sig);
            break;
        case SIGINT:
        case SIGTERM:
            ret = handle_thread_signal(thread, sig);
    }
    objunref(thread);
#endif
    return ret;
}