acsThreadBase.h

Go to the documentation of this file.

#ifndef acsThreadBase_h
#define acsThreadBase_h

/*******************************************************************************
 * ALMA - Atacama Large Millimiter Array
 * (c) European Southern Observatory, 2004
 *
 *This library is free software; you can redistribute it and/or
 *modify it under the terms of the GNU Lesser General Public
 *License as published by the Free Software Foundation; either
 *version 2.1 of the License, or (at your option) any later version.
 *
 *This library 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
 *Lesser General Public License for more details.
 *
 *You should have received a copy of the GNU Lesser General Public
 *License along with this library; if not, write to the Free Software
 *Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
 *
 * "@(#) $Id: acsThreadBase.h,v 1.34 2009/10/07 14:46:39 bjeram Exp $"
 *
 * who       when        what
 * --------  ----------  ----------------------------------------------
 * almamgr   2000-12-03  created
 */

#include <acsutil.h>
#include <acsutilThreadInit.h>

#include <map>
#include <vector>

#include <ace/Thread.h>
#include <ace/Thread_Manager.h>
#include <ace/Synch.h>

#include <acscommonC.h>
#include <ace/SString.h>
#include <logging.h>
#include <loggingLoggable.h>

#include "acsthreadErrType.h"

namespace ACS {

    TimeInterval getTime();

    class ThreadBase : public Logging::Loggable
    {
      public:

        static TimeInterval defaultResponseTime;

        static TimeInterval defaultSleepTime;

        static ThreadBase * NullThreadBase;

        static InitThreadFunc InitThread;

        static DoneThreadFunc DoneThread;

      public:

        enum SleepReturn {
            SLEEP_OK=0,
            SLEEP_SUSPEND,
            SLEEP_INTERRUPTED,
            SLEEP_ERROR
        };

        ThreadBase(const ACE_CString& _name, ACE_Thread_Manager* _threadManager,
                   void* _threadProcedure, void* _parameter,
                   const TimeInterval& _responseTime=ThreadBase::defaultResponseTime,
                   const TimeInterval& _sleepTime=ThreadBase::defaultSleepTime,
                   const bool _create=true,
                   const long _thrFlags= THR_NEW_LWP | THR_DETACHED,
                   const size_t _stackSize=ACE_DEFAULT_THREAD_STACKSIZE);

        virtual ~ThreadBase();

        static void setInitializers(InitThreadFunc InitThread_, DoneThreadFunc DoneThread_) {
            InitThread=InitThread_;
            DoneThread=DoneThread_;
        }

        ACE_CString getName() const { return name_m; }

        void* getThreadProcedure() const { return threadProcedure_mp; }

        TimeInterval getResponseTime() const { return responseTime_m; }

        void setResponseTime(const TimeInterval& _responseTime) { responseTime_m=_responseTime; };

        TimeInterval getSleepTime() const { return sleepTime_m; }

        void setSleepTime(const TimeInterval& _sleepTime) {
            sleepTime_m=_sleepTime;
        };

        void setPriority(int _priority);

        int getPriority();

        bool suspend();

        virtual bool resume();

        bool isSuspended() const { return suspendStatus_m != 0; }

        bool isStopped() const { return stopped_m; }

        virtual void exit() { exitRequest_m=true; }

        bool exitRequested() const { return exitRequest_m; }

        void setStopped() { stopped_m=true; }
        // !!! this can be done by ThreadBase function automatically - using wrapper method

        bool stop( bool terminating = false );


        bool cancel();

        bool terminate();

        bool restart();

        void makeTimeInterval();

        bool isResponding() const;

        bool isAlive() const { return !stopped_m; }

        bool check();

        SleepReturn sleep(TimeInterval timeIn100ns = 0) const;


        ACE_thread_t getThreadID() { return  threadID_m; }

      protected:
        bool create(const long _thrFlags= THR_NEW_LWP | THR_DETACHED);

        virtual void yield();

      private:

        void* threadProcedure_mp;

        const void* parameter_mp;

        TimeInterval responseTime_m;

        TimeInterval sleepTime_m;

        TimeInterval timeStamp_m;

        volatile int suspendStatus_m;

        volatile bool exitRequest_m;

        volatile bool stopped_m;

        ACE_CString name_m;

        // ACE speciic
        ACE_thread_t threadID_m;

        long thrFlags_m;

        size_t stackSize_m;

        ACE_Thread_Manager * threadManager_mp;

        mutable ACE_Thread_Semaphore m_suspendSemaphore;
        mutable ACE_Semaphore m_sleepSemaphore;
    };



    typedef std::map<ACE_CString, ThreadBase*> ThreadMap;

// fwd
//    class Thread;


    class ThreadManagerBase
    {
//    friend class ACS::Thread;

      public:

        ThreadManagerBase()
            {
                threadManager_mp = ACE_Thread_Manager::instance();
            }

        ~ThreadManagerBase();

        int getThreadCount() const {return threads_m.size();}

        ACE_CString getThreadName(const int pos) const { return threads_m[pos]->getName(); }


        ThreadBase* getThreadAt(const int pos) const { return static_cast<ThreadBase*>(threads_m[pos]); }


        ThreadBase* getThreadByName(const ACE_CString& name) {
            ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_addRemoveMutex);

            ThreadMap::iterator i = threadMap_m.find(name);
            if (i!=threadMap_m.end())
                return static_cast<ThreadBase*>((*i).second);
            else
                return NULL;
        }

        ThreadBase* getThreadByID(ACE_thread_t id)
            {

                for(unsigned int i=0UL; i < threads_m.size(); i++)
                    if (threads_m[i]->getThreadID() == id)
                        return threads_m[i];
                return NULL;
            }


        ThreadBase * create(const ACE_CString& name, void * threadProc, void * parameter,
                            const TimeInterval& responseTime=ThreadBase::defaultResponseTime,
                            const TimeInterval& sleepTime=ThreadBase::defaultSleepTime,
                            const long _thrFlags= THR_NEW_LWP | THR_DETACHED,
                            const size_t _stackSize=ACE_DEFAULT_THREAD_STACKSIZE);

        bool add(const ACE_CString& name, ThreadBase * acsBaseThread);

        bool stop(const ACE_CString& name);

        bool stopAll();

        void exit(const ACE_CString& name);

        void exitAll();

        bool cancel(const ACE_CString& name);

        bool cancelAll();

        bool terminate(const ACE_CString& name);

        bool terminateAll();

        bool restart(const ACE_CString& name);

        bool restartAll();

        bool restartDead();

        bool suspend(const ACE_CString& name);

        bool suspendAll();

        bool resume(const ACE_CString& name);

        bool resumeAll();

        bool isAlive(const ACE_CString& name);

        bool areAllAlive();

        int join(const ACE_thread_t& tid);

        ACE_Thread_Manager* getACEThreadManager() { return threadManager_mp; }

      protected:

        void add2map(const ACE_CString& name, ThreadBase* thread)
            {

                threadMap_m[name]=thread;
                threads_m.push_back(thread);
            }

        void removeFromMap(const ACE_CString& name)
            {
                ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_addRemoveMutex);

                ThreadMap::iterator i = threadMap_m.find(name);
                if (i!=threadMap_m.end())
                    {
                    for (std::vector<ThreadBase*>::iterator thr = threads_m.begin(); thr!=threads_m.end(); ++thr)
                        {
                        if (static_cast<ThreadBase*>((*i).second) == *thr)
                            {
                            threads_m.erase(thr);
                            break;
                            }//if
                        }//for
                    }//if
                threadMap_m.erase(name);
            }//removeFromMap

      protected:
        ACE_Recursive_Thread_Mutex m_addRemoveMutex;

      private:

        ACE_Thread_Manager * threadManager_mp;

        ThreadMap threadMap_m;

        std::vector<ThreadBase*> threads_m;

        void operator=(const ThreadManagerBase&);

        ThreadManagerBase(const ThreadManagerBase&);

    };

    class ThreadBaseParameter {

      public:

        ThreadBaseParameter(ThreadBase * thread,
                            const void * parameter = 0) :
            thread_mp(thread),  parameter_mp(parameter) {}

        const void * getParameter() const { return parameter_mp; }

        ThreadBase * getThreadBase() const { return thread_mp; }

//    ThreadBase * getBACIThread() const { return getThreadBase(); }
      private:

        ThreadBase * thread_mp;

        const void * parameter_mp;
    };


    class ThreadSyncGuard
    {
      public:

        ThreadSyncGuard(ACE_Recursive_Thread_Mutex * mutex, bool block=true);

        ~ThreadSyncGuard();

        void acquire();

        void release();

        ACE_Recursive_Thread_Mutex * mutex_mp;

        bool acquired_m;
    };

}; //namespace ACS

#endif