Thread.h

Go to the documentation of this file.

// $Id: Thread.h 3862 2018-05-11 10:06:18Z linev $
 
/************************************************************
 * The Data Acquisition Backbone Core (DABC)                *
 ************************************************************
 * Copyright (C) 2009 -                                     *
 * GSI Helmholtzzentrum fuer Schwerionenforschung GmbH      *
 * Planckstr. 1, 64291 Darmstadt, Germany                   *
 * Contact:  http://dabc.gsi.de                             *
 ************************************************************
 * This software can be used under the GPL license          *
 * agreements as stated in LICENSE.txt file                 *
 * which is part of the distribution.                       *
 ************************************************************/
 
#ifndef DABC_Thread
#define DABC_Thread
 
#ifndef DABC_Object
#include "dabc/Object.h"
#endif
 
#ifndef DABC_defines
#include "dabc/defines.h"
#endif
 
#ifndef DABC_threads
#include "dabc/threads.h"
#endif
 
#ifndef DABC_Queue
#include "dabc/Queue.h"
#endif
 
#ifndef DABC_Command
#include "dabc/Command.h"
#endif
 
#ifndef DABC_CommandsQueue
#include "dabc/CommandsQueue.h"
#endif
 
#ifndef DABC_timing
#include "dabc/timing.h"
#endif
 
#ifndef DABC_Exception
#include "dabc/Exception.h"
#endif
 
#ifndef DABC_logging
#include "dabc/logging.h"
#endif
 
#include <vector>
 
namespace dabc {
 
   class Worker;
   class WorkerAddon;
   class ThreadRef;
 
 
   struct EventId {
      uint64_t value;
 
      inline EventId() : value(0) {}
 
      inline EventId(const EventId& src) : value(src.value) {}
 
      inline EventId(uint16_t code) : value(code) {}
 
      inline EventId(uint16_t code, uint16_t item) : value((uint64_t) item << 48 | code) {}
 
      inline EventId(uint16_t code, uint16_t item, uint32_t arg) :
         value((((uint64_t) item) << 48) | (((uint64_t) arg) << 16) | code) {}
 
      inline EventId& operator=(const EventId& src) { value = src.value; return *this; }
 
      inline EventId& operator=(uint64_t _value) { value = _value; return *this; }
 
      inline bool isnull() const { return value==0; }
 
      inline bool notnull() const { return value!=0; }
 
      inline uint16_t GetCode() const { return value &0xffffLU; }
      inline uint16_t GetItem() const { return value >> 48; }
      inline uint32_t GetArg() const { return (value >> 16) & 0xffffffffLU; }
 
      std::string asstring() const;
 
      inline static EventId Null() { return EventId(); }
   };
 
   class Thread : public Object,
                  protected PosixThread,
                  protected Runnable {
      protected:
 
         friend class Worker;
         friend class ThreadRef;
 
         class RecursionGuard {
            private:
               Thread*  thrd;      
               unsigned workerid;  
            public:
               RecursionGuard(Thread* t, unsigned id) :
                  thrd(t),
                  workerid(id)
               {
                  if (thrd) thrd->ChangeRecursion(workerid, true);
               }
 
               ~RecursionGuard()
               {
                  if (thrd) thrd->ChangeRecursion(workerid, false);
               }
 
         };
 
         struct TimeoutRec {
            TimeStamp      tmout_mark;   
            double         tmout_interv; 
            bool           tmout_active; 
 
            TimeStamp      prev_fire;    
            TimeStamp      next_fire;    
 
            TimeoutRec() : tmout_mark(), tmout_interv(0.), tmout_active(false), prev_fire(), next_fire() {}
 
            TimeoutRec(const TimeoutRec& src) :
               tmout_mark(src.tmout_mark),
               tmout_interv(src.tmout_interv),
               tmout_active(src.tmout_active),
               prev_fire(src.prev_fire),
               next_fire(src.next_fire) {}
 
            bool Activate(double tmout)
            {
               bool dofire = !tmout_active;
               tmout_mark = dabc::Now();
               tmout_interv = tmout;
               tmout_active = true;
               return dofire;
            }
 
            bool CheckEvent(Mutex* thread_mutex)
            {
               TimeStamp mark;
               double interv(0.);
 
               {
                  LockGuard lock(thread_mutex);
                  if (!tmout_active) return false;
                  mark = tmout_mark;
                  interv = tmout_interv;
                  tmout_active = false;
               }
 
               if (interv<0) {
                  next_fire.Reset();
                  prev_fire.Reset();
               } else {
                  // if one activate timeout with positive interval, emulate
                  // that one already has previous call to ProcessTimeout
                  if (prev_fire.null() && (interv>0))
                     prev_fire = mark;
 
                  mark+=interv;
 
                  // set activation time only in the case if no other active timeout was used
                  // TODO: why such condition was here??
                  // every new activate call should set new marker for timeout processing
 
//                  if (next_fire.null() || (mark < next_fire)) {
                     next_fire = mark;
                     return true;
//                  }
               }
 
               return false;
            }
 
            bool CheckNextProcess(const TimeStamp& now, double& min_tmout, double& last_diff)
            {
               if (next_fire.null()) return false;
 
               double dist = next_fire - now;
 
               if (dist>=0.) {
                  if ((min_tmout<0.) || (dist<min_tmout))  min_tmout = dist;
                  return false;
               }
 
               last_diff = prev_fire.null() ? 0. : now - prev_fire;
 
               return true;
            }
 
            void SetNextFire(const TimeStamp& now, double dist, double& min_tmout)
            {
               if (dist>=0.) {
                  prev_fire = now;
                  next_fire = now + dist;
                  if ((min_tmout<0.) || (dist<min_tmout))  min_tmout = dist;
               } else {
                  prev_fire.Reset();
                  next_fire.Reset();
               }
            }
         };
 
 
 
         struct WorkerRec {
            Worker*        work;         
            WorkerAddon*   addon;        
            unsigned       doinghalt;    
            int            recursion;    
            unsigned       processed;    
            CommandsQueue  cmds;         
 
            TimeoutRec     tmout_worker; 
            TimeoutRec     tmout_addon;  
 
            WorkerRec(Worker* w, WorkerAddon* a) :
               work(w),
               addon(a),
               doinghalt(0),
               recursion(0),
               processed(0),
               cmds(CommandsQueue::kindSubmit),
               tmout_worker(),
               tmout_addon() {}
         };
 
         typedef std::vector<WorkerRec*> WorkersVector;
 
         class EventsQueue : public Queue<EventId, true> {
            public:
               int scaler;
 
               EventsQueue() :
                  Queue<EventId, true>(),
                  scaler(8)
                  {
                  }
         };
 
         class ExecWorker;
 
         friend class ExecWorker;
         friend class Object;
         friend class RecursionGuard;
 
         enum EHaltActions { actDestroy = 1, actHalt = 2 };
 
         enum EThreadState { stCreated, stRunning, stStopped, stError, stChanging };
 
         EThreadState         fState;         
 
         bool                 fThrdWorking;   
         bool                 fRealThrd;      
 
         Condition            fWorkCond;      
 
         EventsQueue         *fQueues;         
         int                  fNumQueues;      
 
         TimeStamp            fNextTimeout;    
         int                  fProcessingTimeouts; 
 
         WorkersVector        fWorkers;          
 
         unsigned             fExplicitLoop;     
 
         ExecWorker*          fExec;             
         bool                 fDidDecRefCnt;     
         bool                 fCheckThrdCleanup; 
 
         bool                 fProfiling;        
         TimeStamp            fLastProfileTime;  
         double               fThreadRunTime;    
 
         double               fThrdStopTimeout;  
 
         static unsigned      fThreadInstances;
 
 
         int CheckWorkerCanBeHalted(unsigned id, unsigned request = 0, Command cmd = nullptr);
 
         void IncWorkerFiredEvents(Worker* work);
 
         int RunCommandInTheThread(Worker* caller, Worker* dest, Command cmd);
 
 
      public:
 
         enum EEvents { evntCheckTmoutWorker = 1,  
                        evntCheckTmoutAddon, 
                        evntCleanupThrd,     
                        evntDoNothing,       
                        evntStopThrd,        
                        evntLastThrd,
                        evntUser = 10000 };
 
         enum EPriority { priorityHighest = 0,
                           priorityNormal = 1,
                           priorityLowest = -1 };
 
         static unsigned NumThreadInstances() { return fThreadInstances; }
 
         Thread(Reference parent, const std::string &name, Command cmd = nullptr);
 
         virtual ~Thread();
 
         virtual void* MainLoop();
 
         inline bool IsItself() const { return PosixThread::IsItself(); }
         void SetPriority(int prio = 0) { PosixThread::SetPriority(prio); }
 
         // set stop timeout if required in the thread destructir
         void SetStopTimeout(double timeout_sec) { fThrdStopTimeout = timeout_sec; }
         double GetStopTimeout() const { return fThrdStopTimeout; }
 
         // thread manipulation, thread safe routines
         bool Start(double timeout_sec = -1., bool real_thread = true);
         bool Stop(double timeout_sec = 10);
         bool Sync(double timeout_sec = -1);
 
         virtual const char* ClassName() const { return typeThread; }
 
         virtual bool CompatibleClass(const std::string &clname) const;
 
         void FireDoNothingEvent();
 
         bool Execute(Command cmd, double tmout = -1);
 
         bool SingleLoop(unsigned workerid = 0, double tmout = -1);
 
         void RunEventLoop(double tmout = 1.);
 
         virtual void Print(int lvl = 0);
 
         unsigned TotalNumberOfEvents();
 
      protected:
 
         inline Mutex* ThreadMutex() const { return ObjectMutex(); }
 
         bool IsTemporaryThread() const { return GetParent() == 0; }
 
         virtual int ExecuteThreadCommand(Command cmd);
 
         virtual bool WaitEvent(EventId&, double tmout);
 
         void ProcessEvent(const EventId&);
 
         virtual void ProcessExtraThreadEvent(const EventId&) {}
 
         void ProcessNoneEvent();
 
         bool _GetNextEvent(EventId&);
 
         virtual void RunnableCancelled();
 
        #ifdef DABC_EXTRA_CHECKS
          static unsigned maxlimit;
        #endif
 
         unsigned _TotalNumberOfEvents();
 
         inline void _PushEvent(const EventId& arg, int nq)
         {
            #ifdef DABC_EXTRA_CHECKS
            if ((fNumQueues==0) || (fQueues==0) || (nq>=fNumQueues)) {
               EOUT("False arguments fNumQueues:%d nq:%d", fNumQueues, nq);
               return;
            }
            #endif
 
            fQueues[nq<0 ? fNumQueues - 1 : nq].Push(arg);
 
            #ifdef DABC_EXTRA_CHECKS
            if (nq<0) nq = fNumQueues - 1;
            if (fQueues[nq].Size()>maxlimit) {
               EOUT("Thrd:%s Queue %d Event code:%u item:%u arg:%u exceed limit: %u", GetName(), nq,
                     arg.GetCode(), arg.GetItem(), arg.GetArg(), maxlimit);
               maxlimit *= 2;
            }
            #endif
         }
 
         virtual void _Fire(const EventId& arg, int nq);
 
         inline void Fire(const EventId& arg, int nq)
         {
            LockGuard lock(ThreadMutex());
            _Fire(arg, nq);
         }
 
         double CheckTimeouts(bool forcerecheck = false);
 
         bool AddWorker(Reference ref, bool sync = true);
 
         bool HaltWorker(Worker* proc);
 
         void WorkerAddonChanged(Worker* work, bool assign = true);
 
         bool SetExplicitLoop(Worker* work);
 
         void RunExplicitLoop();
 
         virtual void WorkersSetChanged() {}
 
         bool InvokeWorkerDestroy(Worker* work);
 
         void ChangeRecursion(unsigned id, bool inc);
 
         virtual void ObjectCleanup();
 
         virtual bool _DoDeleteItself();
 
         unsigned NumWorkers();
   };
 
   // __________________________________________________________________________
 
   class ThreadRef : public Reference {
 
      friend class Worker;
      friend class WorkerAddon;
 
      DABC_REFERENCE(ThreadRef, Reference, Thread)
 
      protected:
         bool _ActivateWorkerTimeout(unsigned workerid, int priority, double tmout);
         bool _ActivateAddonTimeout(unsigned workerid, int priority, double tmout);
 
         bool _AcquireThreadRef(ThreadRef& ref) { return AcquireRefWithoutMutex(ref); }
 
      public:
         bool IsItself() const { return GetObject() ? GetObject()->IsItself() : false; }
 
         bool Execute(Command cmd, double tmout = -1.);
 
         void RunEventLoop(double tmout = 1.);
 
         inline bool IsRealThrd() const { return GetObject() ? GetObject()->fRealThrd : false; }
 
         bool MakeWorkerFor(WorkerAddon* addon, const std::string &name = "");
   };
 
}
 
#endif