Worker.h

Go to the documentation of this file.

// $Id: Worker.h 4475 2020-04-15 13:55:30Z 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_Worker
#define DABC_Worker
 
#ifndef DABC_Object
#include "dabc/Object.h"
#endif
 
#ifndef DABC_Thread
#include "dabc/Thread.h"
#endif
 
#ifndef DABC_Parameter
#include "dabc/Parameter.h"
#endif
 
#ifndef DABC_Hierarchy
#include "dabc/Hierarchy.h"
#endif
 
namespace dabc {
 
   class ParameterEvent;
 
   class WorkerRef;
   class Worker;
   class WorkerAddonRef;
 
   class WorkerAddon : public Object {
 
      friend class Thread;
      friend class Worker;
      friend class WorkerAddonRef;
 
      protected:
         Reference  fWorker;
 
         virtual void ObjectCleanup();
 
         virtual void ProcessEvent(const EventId&) {}
 
         void DeleteWorker();
 
         void DeleteAddonItself();
 
         void SubmitWorkerCmd(Command cmd);
 
         void FireWorkerEvent(unsigned evid);
 
         // method will be called by worker when thread was assigned
         virtual void OnThreadAssigned() {}
 
         // analog to Worker::ActiavteTimeout
         bool ActivateTimeout(double tmout_sec);
 
         // analog to Worker::ProcessTimeout
         virtual double ProcessTimeout(double last_diff) { return -1.; }
 
         virtual long Notify(const std::string&, int) { return 0; }
 
      public:
 
         virtual const char* ClassName() const { return "WorkerAddon"; }
 
         virtual std::string RequiredThrdClass() const { return std::string(); }
 
         WorkerAddon(const std::string &name);
         virtual ~WorkerAddon();
   };
 
   // ________________________________________________________________
 
   class WorkerAddonRef : public Reference {
      DABC_REFERENCE(WorkerAddonRef, Reference, WorkerAddon)
 
      long Notify(const std::string &cmd, int arg = 0)
         { return GetObject() ? GetObject()->Notify(cmd,arg) : 0; }
   };
 
   // _________________________________________________________________
 
   class Worker : public Object {
 
      enum EParsMasks {
            parsVisibilityMask = 0xFF,
            parsFixedMask      = 0x100,
            parsChangableMask  = 0x200,
            parsValidMask      = 0x400
      };
 
      friend class Thread;
      friend class Parameter;
      friend class ParameterContainer;
      friend class Command;
      friend class Object;
      friend class WorkerRef;
      friend class WorkerAddon;
 
      private:
 
         void ClearThreadRef();
 
         void InformThreadAssigned()
         {
            OnThreadAssigned();
            if (!fAddon.null()) fAddon()->OnThreadAssigned();
         }
 
         double ProcessAddonTimeout(double last_diff)
         {
            return fAddon.null() ? -1 : fAddon()->ProcessTimeout(last_diff);
         }
 
      protected:
         ThreadRef        fThread;                     
         WorkerAddonRef   fAddon;                      
         Reference        fPublisher;                  
 
         uint32_t         fWorkerId;                   
         int              fWorkerPriority;             
 
         Mutex*           fThreadMutex;                
 
         // FIXME: most workers should analyze FireEvent to recognize moment when worker is going into halt mode
         bool             fWorkerActive;               
         unsigned         fWorkerFiredEvents;          
 
         CommandsQueue    fWorkerCommands;             
 
         int              fWorkerCommandsLevel;        
 
         Hierarchy        fWorkerHierarchy;            
 
         int              fWorkerCfgId;                
 
 
         virtual bool DestroyByOwnThread();
 
         virtual void ObjectCleanup();
 
         virtual void ObjectDestroyed(Object*) {}
 
         inline bool IsWorkerActive() const { return fWorkerActive; }
 
         void SetWorkerCfgId(int id = -1) { fWorkerCfgId = id; }
 
         Worker(const ConstructorPair& pair);
 
      public:
 
         static int cmd_bool(bool v) { return v ? cmd_true : cmd_false; }
 
         enum EWorkerEventsCodes {
            evntFirstCore   = 1,    // events   1  .. 99 used only by Worker itself
            evntFirstAddOn  = 100,  // events 100 .. 199 are dedicated for specific add-ons
            evntFirstSystem = 200,  // events 200 .. 999 used only inside DABC class
            evntFirstUser =   1000  // event from 1000 are available for users
         };
 
         enum EPriotiryLevels {
              priorityMaximum = 0,   // event queue with number 0 always has highest priority
              priorityMinimum = -1,  // this event will be submitted to queue with maximum number
              priorityDefault = -7,  // this code will be replaced with default priority for specified operation
              priorityMagic = -77    // this priority allows to submit commands even when processor is stopped
         };
 
         Worker(Reference parent, const std::string &name);
         virtual ~Worker();
 
         virtual const char* ClassName() const { return "Worker"; }
 
         virtual std::string RequiredThrdClass() const
         { return fAddon.null() ? std::string() : fAddon()->RequiredThrdClass(); }
 
         bool HasThread() const;
 
         bool IsOwnThread() const;
 
         bool AssignToThread(ThreadRef thrd, bool sync = true);
 
         bool MakeThreadForWorker(const std::string &thrdname = "");
 
         bool DettachFromThread();
 
         ThreadRef thread();
 
         std::string ThreadName() const;
 
         bool ActivateTimeout(double tmout_sec);
 
         void SetWorkerPriority(int nq) { fWorkerPriority = nq; }
         inline int WorkerPriority() const { return fWorkerPriority; }
 
         uint32_t WorkerId() const { return fWorkerId; }
 
         // this all about parameters list, which can be managed for any working processor
 
         Parameter Par(const std::string &name) const;
 
         RecordField Cfg(const std::string &name, Command cmd = nullptr) const;
 
         Command Assign(Command cmd);
 
         bool CanSubmitCommand() const;
 
         bool Submit(Command cmd);
 
         bool Execute(Command cmd, double tmout = -1.);
 
         bool Execute(const std::string &cmd, double tmout = -1.) { return Execute(Command(cmd), tmout); }
 
         std::string WorkerAddress(bool full = true);
 
         void AssignAddon(WorkerAddon* addon);
 
      protected:
 
         virtual int PreviewCommand(Command cmd);
 
         virtual int ExecuteCommand(Command cmd);
 
         virtual bool ReplyCommand(Command cmd);
 
         // Method is called when requested time point is reached
         // Rewrite method in derived class to react on this event
         // return value specifies time interval to the next timeout
         // Argument last_diff identifies time distance to previous timeout
         // Return value: <0 - no new timeout is required
         //               =0 - provide timeout as soon as possible
         //               >0 - activate timeout after this interval
         virtual double ProcessTimeout(double last_diff) { return -1.; }
 
         inline bool _IsFireEvent() const
         {
            return fThread() && fWorkerActive;
         }
 
         inline bool _FireEvent(uint16_t evid)
         {
            if (!_IsFireEvent()) return false;
            fThread()->_Fire(EventId(evid, fWorkerId), fWorkerPriority);
            fWorkerFiredEvents++;
            return true;
         }
 
         inline bool FireEvent(uint16_t evid)
         {
            LockGuard lock(fThreadMutex);
            return _FireEvent(evid);
         }
 
         inline bool _FireEvent(uint16_t evid, uint32_t arg, int pri = -1)
         {
            if (!_IsFireEvent()) return false;
            fThread()->_Fire(EventId(evid, fWorkerId, arg), pri < 0 ? fWorkerPriority : pri);
            fWorkerFiredEvents++;
            return true;
         }
 
         inline bool FireEvent(uint16_t evid, uint32_t arg, int pri = -1)
         {
            LockGuard lock(fThreadMutex);
            return _FireEvent(evid, arg, pri);
         }
 
 
         inline bool _FireDoNothingEvent()
         {
            if (!_IsFireEvent()) return false;
            fThread()->_Fire(EventId(Thread::evntDoNothing), -1);
            return true;
         }
 
         virtual void ProcessEvent(const EventId&);
 
         bool ActivateMainLoop();
 
         bool SingleLoop(double tmout) { return fThread()->SingleLoop(fWorkerId, tmout); }
 
         void WorkerSleep(double tmout);
 
         bool ExecuteIn(Worker* dest, Command cmd);
 
         void CancelCommands();
 
         virtual void DoWorkerMainLoop() {}
 
         virtual void DoWorkerAfterMainLoop() {}
 
         // method called immediately after processor was assigned to thread
         // called comes from the thread context
         virtual void OnThreadAssigned() {}
 
         virtual bool Find(ConfigIO &cfg);
 
         virtual Parameter CreatePar(const std::string &name, const std::string &kind = "");
 
         void SetParValue(const std::string &name, const RecordField &v);
 
         CommandDefinition CreateCmdDef(const std::string &name);
 
         bool DestroyPar(const std::string &name);
 
         bool RegisterForParameterEvent(const std::string &mask, bool onlychangeevent = true);
 
         bool UnregisterForParameterEvent(const std::string &mask);
 
         virtual void ProcessParameterEvent(const ParameterEvent &evnt) {}
 
         Reference GetPublisher();
 
         virtual bool Publish(const Hierarchy &h, const std::string &path);
 
         virtual bool PublishPars(const std::string &path);
 
         virtual bool Unpublish(const Hierarchy& h, const std::string &path);
 
         virtual bool Subscribe(const std::string &path);
 
         virtual bool Unsubscribe(const std::string &path);
 
         void CleanupPublisher(bool sync = true);
 
         virtual void BeforeHierarchyScan(Hierarchy& h) {}
 
      private:
 
         enum { evntSubmitCommand = evntFirstCore,
                evntReplyCommand };
 
         void ProcessCoreEvent(EventId);
 
         int ProcessCommand(dabc::Command cmd);
         bool GetCommandReply(dabc::Command& cmd, bool* exe_ready);
 
         void WorkerParameterChanged(bool forcecall, ParameterContainer *par, const std::string &value);
 
         void ProcessParameterRecording(ParameterContainer *par);
   };
 
   // __________________________________________________________________________
 
   class WorkerRef : public Reference {
 
      DABC_REFERENCE(WorkerRef, Reference, Worker)
 
      public:
 
         bool Submit(Command cmd);
 
         bool Execute(Command cmd, double tmout = -1.);
 
         bool Execute(const std::string &cmd, double tmout = -1.);
 
         Parameter Par(const std::string &name) const;
 
         RecordField Cfg(const std::string &name, Command cmd = nullptr) const
         { return GetObject() ? GetObject()->Cfg(name, cmd) : cmd.GetField(name); }
 
         bool HasThread() const;
 
         std::string ThreadName() const
           { return GetObject() ? GetObject()->ThreadName() : std::string(); }
 
         ThreadRef thread()
            { return GetObject() ? GetObject()->thread() : ThreadRef(); }
 
         bool IsSameThread(const WorkerRef& ref);
 
         bool MakeThreadForWorker(const std::string &thrdname = "")
           { return GetObject() ? GetObject()->MakeThreadForWorker(thrdname) : false; }
 
         bool CanSubmitCommand() const;
 
         bool SyncWorker(double tmout = -1.);
 
         bool FireEvent(uint16_t evid, uint32_t arg) { return GetObject() ? GetObject()->FireEvent(evid, arg) : false; }
 
         bool FireEvent(const EventId& ev) { return GetObject() ? GetObject()->FireEvent(ev.GetCode(), ev.GetArg()) : false; }
   };
 
   // _____________________________________________________________________________
 
 
   class CmdParameterEvent : public Command {
 
      DABC_COMMAND(CmdParameterEvent, "CmdParameterEvent")
 
      CmdParameterEvent(const std::string &parname, const std::string &parvalue, int evid, bool attrmodified = false) throw() :
         dabc::Command(CmdName())
      {
         SetStr("ParName", parname);
         if (!parvalue.empty()) SetStr("ParValue", parvalue);
         if (evid!=parModified) SetInt("Event", evid);
         if (attrmodified) SetBool("AttrMod", true);
      }
   };
 
   // _____________________________________________________________________
 
   class ParameterEvent : protected CmdParameterEvent {
      friend class Worker;
 
      protected:
         ParameterEvent() : CmdParameterEvent() {}
 
         ParameterEvent(const Command& cmd) : CmdParameterEvent(cmd) {}
 
      public:
 
         std::string ParName() const { return GetStr("ParName"); }
         std::string ParValue() const { return GetStr("ParValue"); }
         int EventId() const { return GetInt("Event", parModified); }
         bool AttrModified() const { return GetBool("AttrMod", false); }
   };
 
 
   // --------------------------------------------------------------------------
 
   class LocalWorkerRef;
 
   class LocalWorker : public Object {
      protected:
 
         friend class LocalWorkerRef;
 
         virtual int ExecuteCommand(Command) { return cmd_false; }
 
      public:
 
         LocalWorker(const std::string &name) : Object(name, flAutoDestroy) {}
         virtual ~LocalWorker() {}
 
   };
 
   class LocalWorkerRef : public Reference {
 
      DABC_REFERENCE(LocalWorkerRef, Reference, LocalWorker)
 
      public:
 
         bool Execute(Command cmd)
         {
            if (GetObject())
               return GetObject()->ExecuteCommand(cmd) == cmd_true;
            return false;
         }
 
         bool Execute(const std::string &cmd)
         {
            return Execute(Command(cmd));
         }
   };
 
}
 
#endif