root/threadman/trunk/threadman.cpp @ 688

/* $Id$ $URL$ */

/**\file threadman.cpp
 * Definition of ThreadMan
 */

#include "threadman.h"

#include <SDL_thread.h>
#include <SDL_timer.h>

#include <algorithm>
#include <functional>
#include <map>
#include <queue>
#include <assert.h>

#ifdef __unix__
#include <unistd.h>
#else
# ifdef __WIN32__
# include <windows.h>
# ifndef IGNORE_WIN32_EXCEPTIONS
#  include <winbase.h>
#  include <winnt.h>
#  include <excpt.h>
#  include <time.h>
# endif
# else
#  include <stdio.h>
# endif
#endif

#ifdef _MSC_VER
//this needs some work yet for Microsoft compilers
#define IGNORE_WIN32_EXCEPTIONS
#endif

#if defined __WIN32__ && !defined IGNORE_WIN32_EXCEPTIONS
#ifndef _MSC_VER
#warning "Using Win32 signal/exception handler for threads in threadman"
#warning "define IGNORE_WIN32_EXCEPTIONS if you don't want this"
#endif
extern "C" {

static const int KILLTHREAD = 1; ///< Call ExitThread when an exception is caught

static
EXCEPTION_DISPOSITION
win32_threadsignal_handler(struct _EXCEPTION_RECORD* er,
                     void* establisher_frame, /* establisher frame */
                     struct _CONTEXT* context_record, /* context record */
                           void* dispatcher_context) { /* dispatcher context */
    /*
    typedef struct _EXCEPTION_RECORD {
        DWORD ExceptionCode;
        DWORD ExceptionFlags;
        struct _EXCEPTION_RECORD *ExceptionRecord;
        PVOID ExceptionAddress;
        DWORD NumberParameters;
        DWORD ExceptionInformation[EXCEPTION_MAXIMUM_PARAMETERS];
    } EXCEPTION_RECORD,*PEXCEPTION_RECORD,*LPEXCEPTION_RECORD;
    */

    FILE* err = fopen("exception_log.txt", "a");
    time_t now = time(0);
    if (!err)
        err = stderr;
    fprintf(err, "===> %s\t Caught a \"signal\" exception in Win32 at @0x%p: Code = 0x%x, Flags = 0x%x\n",
            ctime(&now), er->ExceptionAddress, (unsigned)er->ExceptionCode, (unsigned)er->ExceptionFlags );
    fprintf(err, "\tSDL_ThreadID() = %lu\n", (unsigned long)SDL_ThreadID());

    if (er->ExceptionFlags & 1)
        fprintf(err, "\tEH_NONCONTINUABLE\n");
    if (er->ExceptionFlags & 2)
        fprintf(err, "\tEH_UNWINDING\n");
    if (er->ExceptionFlags & 4)
        fprintf(err, "\tEH_EXIT_UNWIND\n");
    if (er->ExceptionFlags & 8)
        fprintf(err, "\tEH_STACK_INVALID\n");
    if (er->ExceptionFlags & 0x10)
        fprintf(err, "\tEH_NESTED_CALL\n");

    const char* strcode = "Unknown";
    switch (er->ExceptionCode) {
    case EXCEPTION_ACCESS_VIOLATION      : strcode = "ACCESS_VIOLATION"; break;
    case EXCEPTION_DATATYPE_MISALIGNMENT : strcode = "DATATYPE_MISALIGNMENT"; break;
    case EXCEPTION_BREAKPOINT    : strcode = "BREAKPOINT"; break;
    case EXCEPTION_SINGLE_STEP   : strcode = "SINGLE_STEP"; break;
    case EXCEPTION_ARRAY_BOUNDS_EXCEEDED : strcode = "ARRAY_BOUNDS_EXCEEDED"; break;
    case EXCEPTION_FLT_DENORMAL_OPERAND  : strcode = "FLOAT_DENORMAL_OPERAND"; break;
    case EXCEPTION_FLT_DIVIDE_BY_ZERO    : strcode = "FLOAT_DIVIDE_BY_ZERO"; break;
    case EXCEPTION_FLT_INEXACT_RESULT    : strcode = "FLOAT_INEXACT_RESULT"; break;
    case EXCEPTION_FLT_INVALID_OPERATION : strcode = "FLOAT_INVALID_OPERATION"; break;
    case EXCEPTION_FLT_OVERFLOW  : strcode = "FLOAT_OVERFLOW"; break;
    case EXCEPTION_FLT_STACK_CHECK       : strcode = "FLOAT_STACK_CHECK"; break;
    case EXCEPTION_FLT_UNDERFLOW : strcode = "FLOAT_UNDERFLOW"; break;
    case EXCEPTION_INT_DIVIDE_BY_ZERO    : strcode = "INTEGER_DIVIDE_BY_ZERO"; break;
    case EXCEPTION_INT_OVERFLOW  : strcode = "INTEGER_OVERFLOW"; break;
    case EXCEPTION_PRIV_INSTRUCTION      : strcode = "PRIVILEGED_INSTRUCTION"; break;
    case EXCEPTION_IN_PAGE_ERROR : strcode = "IN_PAGE_ERROR"; break;
    case EXCEPTION_ILLEGAL_INSTRUCTION   : strcode = "ILLEGAL_INSTRUCTION"; break;
    case EXCEPTION_NONCONTINUABLE_EXCEPTION      : strcode = "NONCONTINUABLE_EXCEPTION"; break;
    case EXCEPTION_STACK_OVERFLOW        : strcode = "STACK_OVERFLOW"; break;
    case EXCEPTION_INVALID_DISPOSITION   : strcode = "INVALID_DISPOSITION"; break;
    case EXCEPTION_GUARD_PAGE    : strcode = "GUARD_PAGE_VIOLATION"; break;
    case EXCEPTION_INVALID_HANDLE        : strcode = "INVALID_HANDLE"; break;
    }

    fprintf(err, "\tCode type: %s\n", strcode);

    if (KILLTHREAD) {
        fprintf(err, "\tI'm going to try killing the thread... This might utterly break things\n");
    }
    fprintf(err, "\n");

    fflush(err);
    if (err != stderr)
        fclose(err);


    if (KILLTHREAD) {
        ExitThread(er->ExceptionCode);
        return ExceptionContinueExecution;
    }

    return ExceptionContinueSearch;
    /* Options are:
    ExceptionContinueExecution
    ExceptionContinueSearch
    ExceptionNestedException
    ExceptionCollidedUnwind
    */
}

}
#define BEGIN_HANDLE_WIN32_SIGNALS() __try1(win32_threadsignal_handler)
#define END_HANDLE_WIN32_SIGNALS() __except1
#else
#define BEGIN_HANDLE_WIN32_SIGNALS()
#define END_HANDLE_WIN32_SIGNALS()
#endif

namespace {
    /** Function data */
    struct FnData {
        ThreadMan::THREAD fn; ///< The *actual* thread entry point
        void * data;          ///< Data passed to the thread by \a run()
        SDL_Thread *thr;      ///< Handle on the thread
        SDL_sem *startup_sem; ///< Semaphore to indicate to thread that it can call thread_done
        ThreadMan *tman;      ///< Constructor of this data
        ThreadMan::PRIORITY pri;         ///< The priority to set the thread to, once started
        ThreadMan::Runner *runner; ///< If fn == NULL, use runner->run(data)

        /** Constructor creates/starts the thread and posts to the semaphore */
        FnData (ThreadMan::THREAD f, void* d, ThreadMan *tm, ThreadMan::PRIORITY pri);
        /** Constructor taking a runner */
        FnData (ThreadMan::Runner *r, void* d, ThreadMan *tm, ThreadMan::PRIORITY pri);
        /** Destructor just nukes the semaphore */
        ~FnData() {
            if (runner)
                delete runner;
            if (startup_sem)
                SDL_DestroySemaphore(startup_sem);
        }
        /** Tell the FnData that its safe to delete itself */
        void disown() {
            SDL_SemPost(startup_sem);
        }
    private:
        FnData(const FnData&);
        FnData& operator=(const FnData&);
    };

    /** Function passed to SDL_CreateThread */
    int run_detached (void *fndata) {
        int ignored = -1;
        //BEGIN_HANDLE_WIN32_SIGNALS();
        FnData * f = (FnData*)fndata;
        ThreadMan::setMyPriority(f->pri);
        if (f->fn) {
            ignored = f->fn(f->data);
        } else if (f->runner) {
            ignored = f->runner->run(f->data);
        }
        SDL_SemWait(f->startup_sem);
        f->tman->thread_done(f->thr);
        delete f;
        //END_HANDLE_WIN32_SIGNALS();
        return ignored;
    }
}

/** Synchronised threads manager */
class SyncThreads {
    /** Data for each synchronised thread */
    struct SyncData {
        typedef std::queue<void*> WAITING_DATA; ///< Type for queue of waiting params to \a fn
        ThreadMan::THREAD fn;   ///< The function to call on each post()
        WAITING_DATA waiting;   ///< Queue of data waiting to be passed to \a fn
        SDL_sem *wait_count;    ///< Semaphore, indicating the number of items in \a waiting
        SDL_mutex *queue_mutex; ///< Mutex for \a waiting
        SDL_Thread *thr;        ///< Thread that is calling \a fn
        volatile bool running;  ///< Whether we are currently running
        ThreadMan::PRIORITY pri; ///< Thread priority to set
        ThreadMan::Runner *runner; ///< If fn == NULL, use runner->run(data)

        /** Thread loop (for \a thr) */
        int run();
        /** Set running to false and post to \a waiting */
        void stop();
        /** Wait for \a run to finish */
        int wait();
        /** Kill, gracelessly */
        void kill();
        /** Calls run() */
        static int start(void* data);

        /** Constructor for synchronised thread \a f */
        SyncData(ThreadMan::THREAD f, ThreadMan::PRIORITY);
        /** Constructor for Runner \a r */
        SyncData(ThreadMan::Runner *r, ThreadMan::PRIORITY);
        /** Destructor -- cleans up IPC vars */
        ~SyncData();
    private:
        SyncData(const SyncData&) {}
        SyncData& operator=(const SyncData&) {return *this;}
    };

    typedef std::map<ThreadMan::THREAD, SyncData*> SYNCS; ///< Type to map thread pointers to SyncData
    typedef std::vector<SyncData*> RSYNCS; ///< Type to map thread pointers to SyncData
    SYNCS sync;     ///< Maps thread pointers to SyncData
    RSYNCS rsync;   ///< Maps runner pointers to SyncData
    SDL_mutex *mut; ///< Mutex for \a sync
    volatile bool bail;   ///< Whether to bail out of threads

public:
    /** Schedule a call to \a fn with \a data */
    bool post(ThreadMan::THREAD fn, void* data, ThreadMan::PRIORITY priority, SDL_Thread **thr);
    /** Schedule a call to \a r with \a data */
    bool post(ThreadMan::Runner *r, void* data, ThreadMan::PRIORITY priority, SDL_Thread **thr);

    /** Indicate that all synchronised threads should stop */
    void stopAll();
    /** Wait for all synchronised threads to finish */
    void waitAll();
    /** Kill all synchronised threads, gracelessly */
    void killAll();

    /** Constructor -- creates mutex */
    SyncThreads();
    /** Destructor -- destroys mutex. \note does <em>NOT</em> call stopAll(), waitAll() or killAll() */
    ~SyncThreads();
};

bool SyncThreads::post(ThreadMan::THREAD fn, void* data, ThreadMan::PRIORITY priority, SDL_Thread **thr) {
    if (bail)
        return false;
    SDL_mutexP(mut);
    if (bail) {
        SDL_mutexV(mut);
        return false;
    }
    SYNCS::iterator it = sync.find(fn);
    if (it == sync.end())
        it = sync.insert(SYNCS::value_type(fn, new SyncData(fn, priority))).first;
    SDL_mutexV(mut);
    if (!it->second->thr)
        return false;
    if (thr)
        *thr = it->second->thr;
    SDL_mutexP(it->second->queue_mutex);
    it->second->waiting.push(data);
    SDL_mutexV(it->second->queue_mutex);
    SDL_SemPost(it->second->wait_count);
    return true;
}

bool SyncThreads::post(ThreadMan::Runner *r, void* data, ThreadMan::PRIORITY priority, SDL_Thread **thr) {
    SDL_mutexP(mut);
    RSYNCS::iterator it = rsync.begin();
    const RSYNCS::iterator end = rsync.end();
    for (; it != end; ++it)
        if (*r == *(*it)->runner)
            break;
    if (it == end) {
        rsync.push_back(RSYNCS::value_type(new SyncData(r, priority)));
        it = rsync.end() -1;
    } else if (r->cleanup) {
        delete r;
    }
    SDL_mutexV(mut);
    if (!(*it)->thr)
        return false;
    if (thr)
        *thr = (*it)->thr;
    SDL_mutexP((*it)->queue_mutex);
    (*it)->waiting.push(data);
    SDL_mutexV((*it)->queue_mutex);
    SDL_SemPost((*it)->wait_count);
    return true;
}

SyncThreads::SyncData::SyncData(ThreadMan::THREAD f, ThreadMan::PRIORITY priority)
:
fn(f), wait_count(SDL_CreateSemaphore(0)), queue_mutex(SDL_CreateMutex()), thr(0), running(true), pri(priority), runner(0) {
    if (wait_count && queue_mutex)
        thr = SDL_CreateThread(start, this);
    running = thr;
}

SyncThreads::SyncData::SyncData(ThreadMan::Runner *r, ThreadMan::PRIORITY priority)
:
fn(0), wait_count(SDL_CreateSemaphore(0)), queue_mutex(SDL_CreateMutex()), thr(0), running(true), pri(priority), runner(r) {
    if (wait_count && queue_mutex)
        thr = SDL_CreateThread(start, this);
    running = thr;
}

SyncThreads::SyncData::~SyncData() {
    if (runner)
        delete runner;
    if (wait_count)
        SDL_DestroySemaphore(wait_count);
    if (queue_mutex)
        SDL_DestroyMutex(queue_mutex);
}

int SyncThreads::SyncData::start(void *data) {
    int ret = -1;
    //BEGIN_HANDLE_WIN32_SIGNALS();
    ret = ((SyncData*)data)->run();
    //END_HANDLE_WIN32_SIGNALS();
    return ret;
}

int SyncThreads::SyncData::run() {
    ThreadMan::setMyPriority(pri);
    int intr = SDL_SemWait(wait_count);
    int lastret = 0;
    while (running) {
        if (intr) {
            intr = SDL_SemWait(wait_count);
            continue; // interrupted
        }
        SDL_mutexP(queue_mutex);
        if (waiting.empty()) {
            SDL_mutexV(queue_mutex);
            intr = SDL_SemWait(wait_count);
            continue; //error!
        }
        void *data = waiting.front();
        waiting.pop();
        SDL_mutexV(queue_mutex);
        if (fn) {
            lastret = fn(data);     //run the thread once
        } else {
            lastret = runner->run(data);
        }
        intr = SDL_SemWait(wait_count);
    }
    return lastret;
}

void SyncThreads::SyncData::stop() {
    if (running) {
        running = false;
        SDL_SemPost(wait_count);
    }
}

int SyncThreads::SyncData::wait() {
    int ret = -1;
    if (thr) {
        SDL_WaitThread(thr, &ret);
        thr = 0;
    }
    return ret;
}

void SyncThreads::SyncData::kill() {
    if (running) {
        running = false;
        SDL_KillThread(thr);
        thr = 0;
    }
}

void SyncThreads::stopAll() {
    bail = true; //don't let new stuff get created
    SDL_mutexP(mut);
    //holding the mutex here is safe -- stop doesn't block
    const SYNCS::iterator e = sync.end();
    for (SYNCS::iterator it = sync.begin(); it != e; ++it)
        it->second->stop();
    const RSYNCS::iterator re = rsync.end();
    for (RSYNCS::iterator it = rsync.begin(); it != re; ++it)
        (*it)->stop();
    SDL_mutexV(mut);
}

void SyncThreads::waitAll() {
    stopAll();
    SDL_mutexP(mut);
    SYNCS local_sync;
    RSYNCS local_rsync;
    std::swap(local_sync, sync);
    std::swap(local_rsync, rsync);
    SDL_mutexV(mut);

    const SYNCS::iterator e = local_sync.end();
    for (SYNCS::iterator it = local_sync.begin(); it != e; ++it) {
        it->second->wait();
        delete it->second;
    }
    const RSYNCS::iterator re = local_rsync.end();
    for (RSYNCS::iterator it = local_rsync.begin(); it != re; ++it) {
        (*it)->wait();
        delete *it;
    }
}

void SyncThreads::killAll() {
    stopAll();
    SDL_mutexP(mut);
    const SYNCS::iterator e = sync.end();
    for (SYNCS::iterator it = sync.begin(); it != e; ++it) {
        it->second->kill();
        delete it->second;
    }
    sync.clear();
    SDL_mutexV(mut);
}

SyncThreads::SyncThreads() : mut(0), bail(false) {
    mut = SDL_CreateMutex();
}

SyncThreads::~SyncThreads() {
    if (mut)
        SDL_DestroyMutex(mut);
}

FnData::FnData(ThreadMan::THREAD f, void* d, ThreadMan *tm, ThreadMan::PRIORITY priority)
:
fn(f), data(d), thr(0), startup_sem(SDL_CreateSemaphore(0)), tman(tm), pri(priority), runner(0) {
    if ((thr = SDL_CreateThread(run_detached, this))) {
        tman->thread_add(thr);
    }
}

FnData::FnData(ThreadMan::Runner *r, void* d, ThreadMan *tm, ThreadMan::PRIORITY priority)
:
fn(0), data(d), thr(0), startup_sem(SDL_CreateSemaphore(0)), tman(tm), pri(priority), runner(r) {
    if ((thr = SDL_CreateThread(run_detached, this))) {
        tman->thread_add(thr);
    }
}

ThreadMan::ThreadMan() : thrmut(0), syncman(0) {
    thrmut = SDL_CreateMutex();
    syncman = new SyncThreads();
}

ThreadMan::~ThreadMan() {
    if (thrmut)
        SDL_DestroyMutex(thrmut);
    if (syncman)
        delete syncman;
}

int ThreadMan::ThreadKiller::wait() {
    int retval;
    SDL_WaitThread(thr, &retval);
    return retval;
}

bool ThreadMan::ThreadKiller::done() {
    int ignored;
    if (finished)
        SDL_WaitThread(thr, &ignored);
    return finished;
}

ThreadMan::ThreadKiller::ThreadKiller(SDL_Thread *t)
:
thr(t), finished(false)
{}

bool ThreadMan::ThreadKiller::operator==(SDL_Thread *t) {
    return thr == t;
}

void ThreadMan::thread_cleanup() {
    SDL_mutexP(thrmut);
    //holding a mutex while checking for done is fine
    threads.erase(std::remove_if(threads.begin(),
                                 threads.end(),
                                 std::mem_fun_ref(&ThreadKiller::done)),
                  threads.end());
    SDL_mutexV(thrmut);
}

void ThreadMan::stopAsync() {
    /* first, clean up ones we know are stopped */
    SDL_mutexP(thrmut);//rely on the recursive mutex
    thread_cleanup();
    //here we need to release the mutex (blocking on WaitThread while holding a mutex?? you crazy?)
    //but we can't let other threads call done() while we wait. But we are clearing up everything
    //so let's make a copy and clear threads straight away
    THREADS local;  //initially empty
    std::swap(local, threads);
    SDL_mutexV(thrmut);

    std::for_each(local.begin(),
                  local.end(),
                  std::mem_fun_ref(&ThreadKiller::wait));
}

void ThreadMan::thread_done(SDL_Thread *t) {
    SDL_mutexP(thrmut);
    THREADS::iterator it = std::find(threads.begin(), threads.end(), t);
    if (it != threads.end())
        it->finished = true;
    SDL_mutexV(thrmut);
}

void ThreadMan::thread_add(SDL_Thread *t) {
    SDL_mutexP(thrmut);
    threads.push_back(THREADS::value_type(t));
    SDL_mutexV(thrmut);
}

namespace {
    /** See if the thread in \a t started, if so, assign to \a thr if non-null */
    bool check_thread(FnData* t, SDL_Thread **thr) {
        if (!t->thr) {
            delete t;
            return false;
        }
        /* else we are running */
        if (thr)
            *thr = t->thr;
        t->disown();
        return true;
    }
}

bool ThreadMan::run(THREAD fn, void* data, bool sync, PRIORITY pri, SDL_Thread **thr) {
    if (sync && syncman)
        return syncman->post(fn, data, pri, thr);
    thread_cleanup();
    return check_thread(new FnData(fn, data, this, pri), thr);
}

bool ThreadMan::run(Runner *r, void* data, bool sync, PRIORITY pri, SDL_Thread **thr) {
    if (sync && syncman)
        return syncman->post(r, data, pri, thr);
    thread_cleanup();
    return check_thread(new FnData(r, data, this, pri), thr);
}

void ThreadMan::stopSync() {
    if (syncman)
        syncman->stopAll();
}

void ThreadMan::joinSync() {
    if (syncman)
        syncman->waitAll();
}

void ThreadMan::setMyPriority(ThreadMan::PRIORITY pri) {
#ifdef __WIN32__
    switch(pri) {
    case INHERIT: break;
    case NORMAL:       ::SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_NORMAL); break;
    case BELOW_NORMAL: ::SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_BELOW_NORMAL); break;
    case LOW:          ::SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_LOWEST); break;
    case IDLE:         ::SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_IDLE); break;
    }
#else
# ifdef __unix__
    switch(pri) {
    case INHERIT: break;
    case NORMAL:       ::nice(0); break;
    case BELOW_NORMAL: ::nice(5); break;
    case LOW:          ::nice(10); break;
    case IDLE:         ::nice(19); break;
    }
# else
    static bool thunk = 1;
    if (thunk) {
        ::fputs("== CANNOT CHANGE THREAD PRIORITY ON THIS PLATFORM ==", stderr);
        thunk = 0;
    }
# endif
#endif
}

void ThreadMan::sleep(unsigned  ms) {
    SDL_Delay(ms);
}

int ThreadMan::thread_id() {
    return SDL_ThreadID();
}

ThreadMan& ThreadMan::tman() {
    static ThreadMan *threadman = new ThreadMan();
    return *threadman;
}