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>

#ifdef __DEVELOPMENT_ASSERTIONS__
#include <cassert>
#endif

namespace ibrcommon
{
        void Thread::finalize_thread(void *obj)
        {
#ifdef __DEVELOPMENT_ASSERTIONS__
                // the given object should never be null
                assert(obj != NULL);
#endif
                // cast the thread object
                Thread *th = static_cast<Thread *>(obj);

                // lock the state a last time
                {
                        ibrcommon::MutexLock l(th->_state_lock);

                        // set the state to DONE
                        th->_state = THREAD_FINALIZED;
                        th->_state_lock.signal(true);
                }

                // call the finally method
                th->finally();

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

        void* Thread::exec_thread(void *obj)
        {
#ifdef __DEVELOPMENT_ASSERTIONS__
                // the given object should never be null
                assert(obj != NULL);
#endif
                // cast the thread object
                Thread *th = static_cast<Thread *>(obj);

                // set the state to RUNNING
                try {
                        ibrcommon::MutexLock l(th->_state_lock);
                        if (th->_state == THREAD_CANCELLED)
                        {
                                throw ibrcommon::Exception("Thread has been canceled.");
                        }

                        // call the setup method
                        th->setup();

                        // set the state to running
                        th->_state = THREAD_RUNNING;
                        th->_state_lock.signal(true);
                } catch (const std::exception &ex) {
                        IBRCOMMON_LOGGER_DEBUG(40) << "Thread setup aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
                        Thread::exit();
                }

                // add cleanup-handler to finalize threads
                pthread_cleanup_push(Thread::finalize_thread, (void *)th);

                try {
                        // 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) << "Thread execution aborted: " << ex.what() << IBRCOMMON_LOGGER_ENDL;
                        Thread::exit();
                }

                // remove cleanup-handler and call it
                pthread_cleanup_pop(1);

                // exit the thread
                Thread::exit();

                return NULL;
        }

        Thread::Thread(size_t size, bool delete_on_exit)
         : _state(THREAD_INITIALIZED), tid(0), stack(size), priority(0), __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::cancellation_point()
        {
                ::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::interrupt()
        {
                kill(SIGINT);
        }

        void Thread::enable_interruption()
        {
                sigemptyset(&mask);
                sigaddset(&mask, SIGINT);
                pthread_sigmask(SIG_BLOCK, &mask, &orig_mask);
        }

        void Thread::disable_interruption()
        {
                pthread_sigmask(SIG_SETMASK, &orig_mask, NULL);
        }

        void Thread::cancel() throw (ThreadException)
        {
                // block multiple cancel calls
                ibrcommon::MutexLock l(_state_lock);
                if ((_state != THREAD_PREPARE) && (_state != THREAD_RUNNING)) return;
                _state = THREAD_CANCELLED;
                _state_lock.signal(true);

                // run custom cancellation
                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");
                        }
                }
        }

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

        Thread::CancelProtector::CancelProtector(bool enable) : _oldstate(0)
        {
                if (enable)
                {
                        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &_oldstate);
                }
                else
                {
                        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &_oldstate);
                }
        }

        Thread::CancelProtector::~CancelProtector()
        {
                pthread_setcancelstate(_oldstate, NULL);
        }

        JoinableThread::JoinableThread(size_t size)
         : Thread(size, false)
        {
        }

        JoinableThread::~JoinableThread()
        {
#ifdef __DEVELOPMENT_ASSERTIONS__
                // every thread should be joined, when the destructor is called.
                ibrcommon::Mutex l(_state_lock);
                assert(_state == THREAD_JOINED);
#endif
                join();
        }

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

                ibrcommon::MutexLock l(_state_lock);
                if (_state != THREAD_INITIALIZED) return;

                // set the thread state to PREPARE
                _state = THREAD_PREPARE;
                _state_lock.signal(true);

                // set priority
                priority = adj;

#ifndef __PTH__
                // modify the threads attributes - set as joinable thread
                pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

                // ignore scheduling policy and use the same as the parent
                pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);
#endif
                // we typically use "stack 1" for min stack...
#ifdef  PTHREAD_STACK_MIN
                // if the given stack size is too small...
                if(stack && stack < PTHREAD_STACK_MIN)
                {
                        // set it to the min stack size
                        stack = PTHREAD_STACK_MIN;
                }
#else
                // if stack size if larger than zero and smaller than two...
                if (stack && stack < 2)
                {
                        // set it to zero (we will not set the stack size)
                        stack = 0;
                }
#endif

#ifdef  __PTH__
                // spawn a new thread
                tid = pth_spawn(PTH_ATTR_DEFAULT, &Thread::exec_thread, this);
#else
                // if the stack size is specified (> 0)
                if (stack)
                {
                        // set the stack size attribute
                        pthread_attr_setstacksize(&attr, stack);
                }

                // spawn a new thread
                ret = pthread_create(&tid, &attr, &Thread::exec_thread, this);

                
                switch (ret)
                {
                case EAGAIN:
                        _state = THREAD_ERROR;
                        _state_lock.signal(true);
                        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:
                        _state = THREAD_ERROR;
                        _state_lock.signal(true);
                        throw ThreadException(ret, "The value specified by attr is invalid.");
                case EPERM:
                        _state = THREAD_ERROR;
                        _state_lock.signal(true);
                        throw ThreadException(ret, "The caller does not have appropriate permission to set the required scheduling parameters or scheduling policy.");
                }
#endif
        }

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

        void JoinableThread::join(void)
        {
                // first to some state checking
                {
                        ibrcommon::MutexLock l(_state_lock);

                        // if the thread never has been started: exit and deny any further startup
                        if (_state == THREAD_INITIALIZED)
                        {
                                _state = THREAD_JOINED;
                                _state_lock.signal(true);
                                return;
                        }

                        // if the thread has been joined already: exit
                        if (_state == THREAD_JOINED)
                        {
                                return;
                        }

                        // wait until the finalized state is reached
                        while (_state != THREAD_FINALIZED)
                        {
                                _state_lock.wait();
                        }

                        // get the thread-id of the calling thread
                        pthread_t self = pthread_self();

                        // if we try to join our own thread, just call Thread::exit()
                        if (equal(tid, self))
                        {
                                _state = THREAD_JOINED;
                                _state_lock.signal(true);
                                Thread::exit();
                                return;
                        }
                }

                // if the thread has been started, do join
                try {
                        if (!pthread_join(tid, NULL))
                        {
                                ibrcommon::MutexLock l(_state_lock);
                                _state = THREAD_JOINED;
                                _state_lock.signal(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)
        {
        }

        DetachedThread::~DetachedThread()
        {
        }

        void DetachedThread::start(int adj) throw (ThreadException)
        {
                int ret = 0;

                ibrcommon::MutexLock l(_state_lock);
                if (_state != THREAD_INITIALIZED) return;

                // set the thread state to PREPARE
                _state = THREAD_PREPARE;
                _state_lock.signal(true);

                // set the priority
                priority = adj;

#ifndef __PTH__
                // modify the threads attributes - set as detached thread
                pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

                // ignore scheduling policy and use the same as the parent
                pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED);
#endif
                // we typically use "stack 1" for min stack...
#ifdef  PTHREAD_STACK_MIN
                // if the given stack size is too small...
                if(stack && stack < PTHREAD_STACK_MIN)
                {
                        // set it to the min stack size
                        stack = PTHREAD_STACK_MIN;
                }
#else
                // if stack size if larger than zero and smaller than two...
                if (stack && stack < 2)
                {
                        // set it to zero (we will not set the stack size)
                        stack = 0;
                }
#endif

#ifdef  __PTH__
                // spawn a new thread
                tid = pth_spawn(PTH_ATTR_DEFAULT, &Thread::exec_thread, this);
#else
                // if the stack size is specified (> 0)
                if (stack)
                {
                        // set the stack size attribute
                        pthread_attr_setstacksize(&attr, stack);
                }

                // spawn a new thread
                ret = pthread_create(&tid, &attr, &Thread::exec_thread, this);

                // check for errors
                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 (const ThreadException&) {
                        throw;
                } catch (const std::exception&) {
                }
        }
}