Manager.h

Go to the documentation of this file.

// $Id: Manager.h 4471 2020-04-15 13:19:26Z 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_Manager
#define DABC_Manager
 
#ifndef DABC_Worker
#include "dabc/Worker.h"
#endif
 
#ifndef DABC_Application
#include "dabc/Application.h"
#endif
 
#ifndef DABC_Module
#include "dabc/Module.h"
#endif
 
namespace dabc {
 
   class Mutex;
   class Thread;
   class Port;
   class MemoryPool;
   class Manager;
   class ManagerRef;
   class Device;
   class Factory;
   class FactoryPlugin;
   class DependPairList;
   class DataInput;
   class InputTransport;
   class DataOutput;
   class Configuration;
   class StdManagerFactory;
   class ReferencesVector;
   class Thread;
   class ParamEventReceiverList;
 
   class CmdDeletePool : public Command {
      public:
         static const char* CmdName() { return "DeletePool"; }
 
         CmdDeletePool(const std::string &name) : Command(CmdName())
            { SetStr("PoolName", name); }
   };
 
 
   class CmdModule : public Command {
      public:
         static const char* ModuleArg() { return "Module"; }
 
         CmdModule(const char* cmdname, const std::string &module) : Command(cmdname)
         {
            SetStr(ModuleArg(), module);
         }
   };
 
   class CmdCreateModule : public CmdModule {
      public:
         static const char* CmdName() { return "CreateModule"; }
 
         CmdCreateModule(const std::string &classname, const std::string &modulename, const std::string &thrdname = "") :
            CmdModule(CmdName(), modulename)
            {
               SetStr(xmlClassAttr, classname);
               if (!thrdname.empty()) SetStr(xmlThreadAttr, thrdname);
            }
   };
 
   class CmdStartModule : public CmdModule {
      public:
         static const char* CmdName() { return "StartModule"; }
 
         CmdStartModule(const std::string &modulename = "*") : CmdModule(CmdName(), modulename) {}
   };
 
   class CmdStopModule : public CmdModule {
      public:
         static const char* CmdName() { return "StopModule"; }
 
         CmdStopModule(const std::string &modulename = "*") : CmdModule(CmdName(), modulename) {}
   };
 
   class CmdDeleteModule : public CmdModule {
      public:
         static const char* CmdName() { return "DeleteModule"; }
 
         CmdDeleteModule(const std::string &modulename) : CmdModule(CmdName(), modulename) {}
   };
 
   class CmdCleanupApplication : public Command {
      DABC_COMMAND(CmdCleanupApplication, "CleanupManager");
   };
 
   class CmdCreateApplication : public Command {
 
      DABC_COMMAND(CmdCreateApplication, "CreateApplication");
 
      CmdCreateApplication(const std::string &appclass, const std::string &thrdname = "") :
         Command(CmdName())
      {
         SetStr("AppClass", appclass);
         if (!thrdname.empty()) SetStr(xmlThreadAttr, thrdname);
      }
   };
 
 
   class CmdCreateDevice : public Command {
      public:
         static const char* CmdName() { return "CreateDevice"; }
 
         CmdCreateDevice(const std::string &devclass, const std::string &devname) :
            Command(CmdName())
         {
            SetStr("DevClass", devclass);
            SetStr("DevName", devname);
         }
   };
 
   class CmdDestroyDevice : public Command {
      public:
         static const char* CmdName() { return "DestroyDevice"; }
 
         CmdDestroyDevice(const std::string &devname) :
            Command(CmdName())
         {
            SetStr("DevName", devname);
         }
   };
 
   class CmdCreateThread : public Command {
      public:
         static const char* CmdName() { return "CreateThread"; }
 
         static const char* ThrdNameArg() { return "ThrdName"; }
 
         CmdCreateThread(const std::string &thrdname, const std::string &thrdclass = "",  const std::string &devname = "") :
            Command(CmdName())
         {
            SetStr(ThrdNameArg(), thrdname);
            SetStr("ThrdClass", thrdclass);
            SetStr("ThrdDev", devname);
         }
 
         std::string GetThrdName() const { return GetStr(ThrdNameArg()); }
   };
 
   class CmdCreateObject : public Command {
 
      DABC_COMMAND(CmdCreateObject, "CreateObject");
 
      CmdCreateObject(const std::string &classname, const std::string &objname) :
         Command(CmdName())
      {
         SetStr("ClassName", classname);
         SetStr("ObjName", objname);
      }
   };
 
 
   class CmdCreateDataInput : public Command {
      DABC_COMMAND(CmdCreateDataInput, "CreateDataInput");
   };
 
   class CmdCreateTransport : public Command {
 
      DABC_COMMAND(CmdCreateTransport, "CreateTransport");
 
      static const char* PortArg() { return "PortName"; }
      static const char* KindArg() { return "TransportKind"; }
 
      CmdCreateTransport(const std::string &portname, const std::string &transportkind, const std::string &thrdname = "") :
         Command(CmdName())
      {
         SetStr(PortArg(), portname);
         SetStr(KindArg(), transportkind);
         if (!thrdname.empty()) SetStr(xmlThreadAttr, thrdname);
      }
 
      void SetPoolName(const std::string &name) { if (!name.empty()) SetStr(xmlPoolName, name); }
 
      std::string PortName() const { return GetStr(PortArg()); }
      std::string TransportKind() const { return GetStr(KindArg()); }
      std::string PoolName() const { return GetStr(xmlPoolName); }
   };
 
   class CmdDestroyTransport : public Command {
 
      DABC_COMMAND(CmdDestroyTransport, "DestroyTransport");
 
      CmdDestroyTransport(const std::string &portname) :
            Command(CmdName())
      {
         SetStr("PortName", portname);
      }
   };
 
   class CmdCreateAny : public Command {
      DABC_COMMAND(CmdCreateAny, "CreateAny");
   };
 
 
   class CmdSetParameter : public Command {
 
      DABC_COMMAND(CmdSetParameter, "SetParameter");
 
      static const char* ParName() { return "ParName"; }
      static const char* ParValue() { return "ParValue"; }
 
      CmdSetParameter(const std::string &parname) :
         Command(CmdName())
      {
         SetStr(ParName(), parname);
      }
 
      CmdSetParameter(const std::string &parname, const RecordField& v) :
         Command(CmdName())
      {
         SetStr(ParName(), parname);
         SetParValue(v);
      }
 
      void SetParValue(const RecordField& v) { SetField(ParValue(), v); }
 
      std::string GetParName() { return GetStr(ParName()); }
      RecordField GetParValue() { return GetField(ParValue()); }
   };
 
   class CmdGetNodesState : public Command {
 
      DABC_COMMAND(CmdGetNodesState, "CmdGetNodesState")
 
      static const char* States() { return "States"; }
      static const char* NodeId() { return "NodeId"; }
 
      static void SetState(std::string& sbuf, unsigned node, bool on)
      {
         while (sbuf.length() <= node) sbuf.append("x");
         sbuf[node] = on ? 'x' : 'o';
      }
 
      static bool GetState(const std::string &sbuf, unsigned node)
      {
         if (node>=sbuf.length()) return false;
         return sbuf[node] == 'x';
      }
   };
 
   template<class T>
   class CleanupEnvelope : public Object {
      protected:
         T* fObj;
      public:
         CleanupEnvelope(T* obj) : Object(0, "noname"), fObj(obj) {}
         virtual ~CleanupEnvelope() { delete fObj; }
   };
 
 
   enum ThreadsLayout {
      layoutMinimalistic,  // minimal number of threads
      layoutPerModule,    // each module should use single thread, including all transports
      layoutBalanced,     // per each module three threads - for module, all input and all output transports
      layoutMaximal       // each entity will gets its own thread
   };
 
 
 
   class Manager : public Worker {
 
      friend class Object;
      friend class Factory;
      friend class FactoryPlugin;
      friend class ParameterContainer;
      friend class StdManagerFactory;
      friend class ManagerRef;
 
      private:
 
         enum { MagicInstanceId = 7654321 };
 
         // this method is used from Factory to register factory when it created
         static void ProcessFactory(Factory* factory);
 
         static Manager* fInstance;  
         static int fInstanceId;     
         static Factory* fFirstFactories[10]; 
         static int fFirstFactoriesId;  
 
 
      protected:
 
         enum MgrEvents { evntDestroyObj = evntFirstSystem, evntManagerParam };
 
         struct ParamRec {
            Parameter par;  
            int event;
 
            ParamRec() : par(), event(0) {}
            ParamRec(Parameter p, int e) : par(p), event(e) {}
            ParamRec(const ParamRec& src) : par(src.par), event(src.event) {}
            ~ParamRec() { reset(); }
 
            ParamRec& operator=(const ParamRec& src)
            {
               par = src.par;
               event = src.event;
               return* this;
            }
 
            void reset()
            {
               par.Release();
               event = 0;
            }
         };
 
         TimeStamp            fMgrStoppedTime;  
         bool                 fAppFinished;     
 
         // FIXME: revise usage of manager mutex, it is not necessary everywhere
         Mutex                *fMgrMutex; // main mutex to protect manager queues
 
         ReferencesVector     *fDestroyQueue;
         // FIXME: use stl containers while RecordsQueue uses memcpy for struct with virtual table
         RecordsQueue<ParamRec> fParsQueue;
 
         // TODO: timed parameters should be seen in the special manager folder ?
         ReferencesVector     *fTimedPars;
 
         ParamEventReceiverList*  fParEventsReceivers;
 
         // TODO: dependency pairs should be seen in special manager folder
         DependPairList       *fDepend; // list of objects dependencies
 
         Configuration        *fCfg;
         std::string           fCfgHost;
 
         int                   fNodeId;
         int                   fNumNodes;
 
         std::string           fLocalAddress;
 
         ThreadsLayout         fThrLayout; 
 
         std::string fLastCreatedDevName;  
 
         Reference FindItem(const std::string &itemname, bool islocal = false);
 
         void FillItemName(const Object* ptr, std::string& itemname, bool compact = true);
 
         ThreadRef FindThread(const std::string &name, const std::string &required_class = "");
 
         ThreadRef CurrentThread();
 
         WorkerRef FindDevice(const std::string &name);
 
         Reference FindPool(const std::string &name);
 
         ModuleRef FindModule(const std::string &name);
 
         Reference FindPort(const std::string &name);
 
         ApplicationRef app();
 
         bool IsAnyModuleRunning();
 
         bool DoCreateMemoryPool(Command cmd);
 
         ThreadRef DoCreateThread(const std::string &thrdname, const std::string &classname = "", const std::string &devname = "", Command cmd = nullptr);
 
         WorkerRef DoCreateModule(const std::string &classname, const std::string &modulename, Command cmd);
 
         Reference DoCreateObject(const std::string &classname, const std::string &objname = "", Command cmd = nullptr);
 
         WorkerRef GetCommandChannel() { return FindItem(CmdChlName()); }
 
         Reference GetFactoriesFolder(bool force = false) { return GetFolder(FactoriesFolderName(), force); }
 
         Reference GetThreadsFolder(bool force = false) { return GetFolder(ThreadsFolderName(), force); }
 
         void Sleep(double tmout, const char* prefix = 0);
 
         std::string GetLastCreatedDevName();
 
         void RunManagerMainLoop(double runtime = 0.);
 
         void RunManagerCmdLoop(double runtime = 0., const std::string &remnode = "");
 
         bool DoCleanupApplication();
 
         // virtual methods from Worker
 
         virtual double ProcessTimeout(double last_diff);
 
         virtual int PreviewCommand(Command cmd);
         virtual int ExecuteCommand(Command cmd);
         virtual bool ReplyCommand(Command cmd);
 
         virtual void ProcessEvent(const EventId&);
 
      public:
 
         Manager(const std::string &managername, Configuration* cfg = 0);
         virtual ~Manager();
 
         virtual const char* ClassName() const { return "Manager"; }
 
          // candidates for protected
 
         void HaltManager();
 
         // -------------- generic folder structure of manager
 
         static const char* FactoriesFolderName() { return "Factories"; }
         static const char* ThreadsFolderName() { return "Threads"; }
 
         static const char* MgrThrdName()       { return "ManagerThrd"; }
         static const char* AppThrdName()       { return "AppThrd"; }
         static const char* ConnMgrName()       { return "/ConnMgr"; }
         static const char* CmdChlName()        { return "/CommandChl"; }
 
 
         int NodeId() const { return fNodeId; }
         int NumNodes() { return fNumNodes; }
 
         std::string GetLocalAddress();
 
         std::string GetNodeAddress(int nodeid);
         bool DecomposeAddress(const std::string &url, bool& islocal, std::string& server, std::string& itemtname);
         std::string ComposeAddress(const std::string &server, const std::string &itemtname = "");
 
         ThreadsLayout GetThreadsLayout() const { return fThrLayout; }
 
         // -------------------------- misc functions ---------------
 
         Configuration* cfg() const { return fCfg; }
 
         virtual void Print(int lvl = 0);
 
         virtual bool DestroyObject(Reference ref);
 
         template<class T> bool DeleteAnyObject(T* obj)
         {
             return DestroyObject(Reference(new CleanupEnvelope<T>(obj)));
         }
 
         bool RegisterDependency(Object* src, Object* tgt, bool bidirectional = false);
 
         bool UnregisterDependency(Object* src, Object* tgt, bool bidirectional = false);
 
         bool InstallSignalHandlers();
         void ProcessCtrlCSignal();
         void ProcessPipeSignal();
 
         virtual bool Find(ConfigIO &cfg);
 
         // ------------ access to factories method -------------
 
         static void SetAutoDestroy(bool on);
 
 
         bool ProcessDestroyQueue();
         bool ProcessParameterEvents();
         void ProduceParameterEvent(ParameterContainer* par, int evid);
   };
 
   class ManagerRef : public WorkerRef {
 
      DABC_REFERENCE(ManagerRef, WorkerRef, Manager)
 
      protected:
 
         friend class Manager;
         friend class Worker;
         friend class Port;   // need to activate connection manager
         friend class InputTransport;  // need to create input object
 
         bool ParameterEventSubscription(Worker* ptr, bool subscribe, const std::string &mask, bool onlychangeevent = true);
 
         bool IsLocalItem(const std::string &name);
 
         DataInput* CreateDataInput(const std::string &kind);
 
      public:
 
         ApplicationRef app();
 
         bool CleanupApplication();
 
         int NodeId() const;
 
         int NumNodes() const;
 
         std::string GetLocalAddress()
         {  return null() ? std::string() : GetObject()->GetLocalAddress(); }
 
         std::string GetNodeAddress(int nodeid)
            { return null() ? std::string() : GetObject()->GetNodeAddress(nodeid); }
 
         bool DecomposeAddress(const std::string &url, bool& islocal, std::string& server, std::string& itemtname)
           { return null() ? false : GetObject()->DecomposeAddress(url, islocal, server, itemtname); }
 
         std::string ComposeAddress(const std::string &server, const std::string &itemname = "")
           {  return null() ? std::string() : GetObject()->ComposeAddress(server, itemname); }
 
         bool CreateApplication(const std::string &classname = "", const std::string &appthrd = "");
 
         ThreadRef CreateThread(const std::string &thrdname, const std::string &classname = "", const std::string &devname = "");
 
         ThreadRef CurrentThread();
 
         bool CreateDevice(const std::string &classname, const std::string &devname);
         bool DeleteDevice(const std::string &devname);
         WorkerRef FindDevice(const std::string &name);
 
         Reference CreateObject(const std::string &classname, const std::string &objname);
 
         bool CreateMemoryPool(const std::string &poolname,
                               unsigned buffersize = 0,
                               unsigned numbuffers = 0);
 
         bool DeletePool(const std::string &name);
 
         ModuleRef CreateModule(const std::string &classname, const std::string &modulename, const std::string &thrdname = "");
         ModuleRef FindModule(const std::string &name);
         void StartModule(const std::string &modulename);
         void StopModule(const std::string &modulename);
         bool StartAllModules();
         bool StopAllModules();
         bool DeleteModule(const std::string &name);
 
         bool CreateTransport(const std::string &portname, const std::string &transportkind = "", const std::string &thrdname = "");
 
         void* CreateAny(const std::string &classname, const std::string &objname = "");
 
         bool ActivateConnections(double tmout, bool sync = true);
 
         Reference FindItem(const std::string &name);
 
         Reference FindPort(const std::string &port);
 
         Parameter FindPar(const std::string &parname);
 
         Reference FindPool(const std::string &name);
 
         ThreadRef FindThread(const std::string &name, const std::string &required_class = "")
            { return null() ? ThreadRef() : GetObject()->FindThread(name, required_class); }
 
         unsigned NumThreads() const
            { return null() ? 0 : GetObject()->GetThreadsFolder().NumChilds(); }
 
         ThreadsLayout GetThreadsLayout() const
           { return null() ? layoutBalanced : GetObject()->GetThreadsLayout(); }
 
         bool IsTerminated() const
         { return null() ? true : !GetObject()->fMgrStoppedTime.null(); }
 
         ConnectionRequest Connect(const std::string &port1, const std::string &port2);
 
         void StopApplication();
 
         void Sleep(double tmout, const char* prefix = 0);
 
         void RunMainLoop(double runtime = 0.)
            { if (GetObject()) GetObject()->RunManagerMainLoop(runtime); }
 
         void RunCmdLoop(double runtime = 0., const std::string &remnode = "")
            { if (GetObject()) GetObject()->RunManagerCmdLoop(runtime, remnode); }
 
         WorkerRef GetCommandChannel()
           { return GetObject() ? GetObject()->GetCommandChannel() : WorkerRef();  }
 
         bool CreateControl(bool withserver, int serv_port = 0, bool allow_clients = true);
 
         bool CreatePublisher();
 
   };
 
   extern ManagerRef mgr;
}
 
#endif