ibrcommon/ibrcommon/thread/Thread.cpp

Go to the documentation of this file.

/*
 * Thread.cpp
 *
 *  Created on: 30.07.2009
 *      Author: morgenro
 */

#include "ibrcommon/config.h"
#include "ibrcommon/thread/Thread.h"
#include "ibrcommon/thread/MutexLock.h"
#include "ibrcommon/Logger.h"
#include <stdexcept>
#include <pthread.h>
#include <sys/times.h>
#include <stdio.h>
#include <unistd.h>
#include <signal.h>
#include <cassert>


namespace ibrcommon
{
        ThreadFinalizer::ThreadFinalizer(Thread *t) : _thread(t) { };
        ThreadFinalizer::~ThreadFinalizer()
        {
                // call the finally method
                _thread->finally();

                // delete the thread-object is requested
                if (_thread->__delete_on_exit)
                {
                        delete _thread;
                }
        };

        void* Thread::exec_thread(void *obj)
        {
                assert(obj != NULL);

                Thread *th = static_cast<Thread *>(obj);

                // run threading routines in a try-catch environment
                try {
                        // this object destroys detached threads and call the finalize method
                        ThreadFinalizer finalizer(th);

                        // signal ready to start() routine
                        th->ready();

                        // run threads run routine
                        th->run();
                } catch (const std::exception &ex) {
                        // an exception occured in run() or ready(), but this is allowed
                        IBRCOMMON_LOGGER_DEBUG(40) << "Exception during thread execution: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
                }

                // Exit this thread. This is required with uClibc.
                Thread::exit();

                return NULL;
        }

        Thread::Thread(size_t size, bool delete_on_exit)
         : _cancel(false), tid(0), stack(size), priority(0), _ready(false), __delete_on_exit(delete_on_exit)
        {
                pthread_attr_init(&attr);
        }

        void Thread::yield(void)
        {
        #if defined(HAVE_PTHREAD_YIELD_NP)
                pthread_yield_np();
        #elif defined(HAVE_PTHREAD_YIELD)
                pthread_yield();
        #else
                sched_yield();
        #endif
        }

        void Thread::testcancel()
        {
                ::pthread_testcancel();
        }

        void Thread::concurrency(int level)
        {
        #if defined(HAVE_PTHREAD_SETCONCURRENCY)
                pthread_setconcurrency(level);
        #endif
        }

        void Thread::sleep(size_t timeout)
        {
                timespec ts;
                ts.tv_sec = timeout / 1000l;
                ts.tv_nsec = (timeout % 1000l) * 1000000l;
        #if defined(HAVE_PTHREAD_DELAY)
                pthread_delay(&ts);
        #elif defined(HAVE_PTHREAD_DELAY_NP)
                pthread_delay_np(&ts);
        #else
                usleep(timeout * 1000);
        #endif
        }

        Thread::~Thread()
        {
                pthread_attr_destroy(&attr);
        }

        void Thread::detach(void)
        {
                pthread_detach(tid);
        }

        void Thread::exit(void)
        {
                pthread_exit(NULL);
        }

        int Thread::kill(int sig)
        {
                if (tid == 0) return -1;
                return pthread_kill(tid, sig);
        }

        void Thread::cancel() throw (ThreadException)
        {
                ibrcommon::MutexLock l(_cancelLock);
                if (_cancel)
                        return;
                if (!__cancellation())
                {
                        // check if the thread is still alive
                        if (Thread::kill(0) == 0)
                        {
                                int ret = pthread_cancel(tid);
                                if ( 0 != ret) throw ThreadException(ret, "pthread_cancel");
                        }
                }
                _cancel=true;
        }

        int Thread::enableCancel(int &state)
        {
                return pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &state);
        }

        int Thread::disableCancel(int &state)
        {
                return pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &state);
        }

        int Thread::restoreCancel(const int &state)
        {
                return pthread_setcancelstate(state, NULL);
        }

        int Thread::setCancel(bool val)
        {
                int old = 0;

                if (val)
                {
                        return pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, &old);
                }
                else
                {
                        return pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, &old);
                }
        }

        bool Thread::equal(pthread_t t1, pthread_t t2)
        {
                return (pthread_equal(t1, t2) != 0);
        }

        void Thread::ready()
        {
                ibrcommon::MutexLock l(_readycond);
                _ready = true;
                _readycond.signal(true);
        }

        void Thread::waitready()
        {
                ibrcommon::MutexLock l(_readycond);
                while (_ready == false)
                {
                        _readycond.wait();
                }
        }

        Thread::CancelProtector::CancelProtector(bool enable) : _oldstate(0)
        {
                if (enable)
                {
                        Thread::enableCancel(_oldstate);
                }
                else
                {
                        Thread::disableCancel(_oldstate);
                }
        }

        Thread::CancelProtector::~CancelProtector()
        {
                Thread::restoreCancel(_oldstate);
        }

        JoinableThread::JoinableThread(size_t size)
         : Thread(size, false), running_mutex(ibrcommon::Mutex::MUTEX_ERRORCHECK), running(false), joined(false)
        {
        }

        JoinableThread::~JoinableThread()
        {
                join();
        }

        void JoinableThread::start(int adj) throw (ThreadException)
        {
                int result;

                ibrcommon::MutexLock l(running_mutex);
                if(running)
                        return;

                priority = adj;

                pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
                pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);

        // we typically use "stack 1" for min stack...
        #ifdef  PTHREAD_STACK_MIN
                if(stack && stack < PTHREAD_STACK_MIN)
                        stack = PTHREAD_STACK_MIN;
        #else
                if(stack && stack < 2)
                        stack = 0;
        #endif

                if(stack)
                        pthread_attr_setstacksize(&attr, stack);
                result = pthread_create(&tid, &attr, &Thread::exec_thread, this);
                
                switch (result)
                {
                case EAGAIN:
                        throw ThreadException(result, "The system lacked the necessary resources to create another thread, or the system-imposed limit on the total number of threads in a process PTHREAD_THREADS_MAX would be exceeded.");
                case EINVAL:
                        throw ThreadException(result, "The value specified by attr is invalid.");
                case EPERM:
                        throw ThreadException(result, "The caller does not have appropriate permission to set the required scheduling parameters or scheduling policy.");
                }

                // wait until the thread is running
                waitready();

                running = true;
        }

        void JoinableThread::stop() throw (ThreadException)
        {
                // Cancel throws the exception of the terminated thread
                // so we have to catch any possible exception here
                try {
                        ibrcommon::MutexLock l(running_mutex);
                        Thread::cancel();
                } catch (ThreadException) {
                        throw;
                } catch (std::exception) {
                }
        }

        void JoinableThread::join(void)
        {
                try {
                        pthread_t self = pthread_self();

                        {
                                ibrcommon::MutexLock l(running_mutex);
                                if(running && equal(tid, self))
                                        Thread::exit();

                                // we're not running, so we ignore...
                                if (!running) return;
                        }

                        // already joined, so we ignore...
                        ibrcommon::MutexLock l(join_mutex);
                        if (joined || (tid == 0)) return;

                        if(!pthread_join(tid, NULL))
                        {
                                ibrcommon::MutexLock l(running_mutex);
                                running = false;
                                joined = true;
                        }
                } catch (const std::exception &ex) {
                        IBRCOMMON_LOGGER(error) << "Exception on Thread[" << tid << "]::join(): " << ex.what() << IBRCOMMON_LOGGER_ENDL;
                }
        }

        DetachedThread::DetachedThread(size_t size) : Thread(size, true)
        {
                stack = size;
        }

        void DetachedThread::exit(void)
        {
                pthread_exit(NULL);
        }

        DetachedThread::~DetachedThread()
        {
        }

        void DetachedThread::start(int adj) throw (ThreadException)
        {
                priority = adj;
        #ifndef __PTH__
                pthread_attr_t attr;
                pthread_attr_init(&attr);
                pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
                pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);
        #endif
        // we typically use "stack 1" for min stack...
        #ifdef  PTHREAD_STACK_MIN
                if(stack && stack < PTHREAD_STACK_MIN)
                        stack = PTHREAD_STACK_MIN;
        #else
                if(stack && stack < 2)
                        stack = 0;
        #endif
        #ifdef  __PTH__
                tid = pth_spawn(PTH_ATTR_DEFAULT, &Thread::exec_thread, this);
        #else
                if(stack)
                        pthread_attr_setstacksize(&attr, stack);
                int ret = pthread_create(&tid, &attr, &Thread::exec_thread, this);
                pthread_attr_destroy(&attr);

                switch (ret)
                {
                case EAGAIN:
                        throw ThreadException(ret, "The system lacked the necessary resources to create another thread, or the system-imposed limit on the total number of threads in a process PTHREAD_THREADS_MAX would be exceeded.");
                case EINVAL:
                        throw ThreadException(ret, "The value specified by attr is invalid.");
                case EPERM:
                        throw ThreadException(ret, "The caller does not have appropriate permission to set the required scheduling parameters or scheduling policy.");
                }
        #endif
        }

        void DetachedThread::stop() throw (ThreadException)
        {
                // Cancel throws the exception of the terminated thread
                // so we have to catch any possible exception here
                try {
                        Thread::cancel();
                } catch (ThreadException) {
                        throw;
                } catch (std::exception) {
                }
        }
}