Worker.cxx

Go to the documentation of this file.

// $Id: Worker.cxx 4764 2021-04-23 07:01:04Z 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.                       *
 ************************************************************/
 
#include "dabc/Worker.h"
 
#include <cstdlib>
#include <cmath>
 
#include "dabc/Manager.h"
#include "dabc/Publisher.h"
#include "dabc/HierarchyStore.h"
#include "dabc/Url.h"
#include "dabc/defines.h"
 
dabc::WorkerAddon::WorkerAddon(const std::string &name) :
   Object(0, name),
   fWorker()
{
   SetFlag(flAutoDestroy, true);
}
 
dabc::WorkerAddon::~WorkerAddon()
{
}
 
 
void dabc::WorkerAddon::ObjectCleanup()
{
   fWorker.Release();
}
 
void dabc::WorkerAddon::DeleteWorker()
{
   if (fWorker.null()) DeleteThis();
                 else fWorker.Destroy();
}
 
void dabc::WorkerAddon::DeleteAddonItself()
{
   if (fWorker.null()) DeleteThis();
                else SubmitWorkerCmd(dabc::Command("DeleteAddon"));
}
 
void dabc::WorkerAddon::FireWorkerEvent(unsigned evid)
{
   dabc::Worker* wrk = (dabc::Worker*) fWorker();
   if (wrk) wrk->FireEvent(evid);
}
 
bool dabc::WorkerAddon::ActivateTimeout(double tmout_sec)
{
   dabc::Worker* wrk = (dabc::Worker*) fWorker();
 
   if (wrk==0) return false;
 
   LockGuard lock(wrk->fThreadMutex);
 
   return wrk->fThread._ActivateAddonTimeout(wrk->fWorkerId, wrk->WorkerPriority(), tmout_sec);
}
 
void dabc::WorkerAddon::SubmitWorkerCmd(Command cmd)
{
   dabc::Worker* wrk = (dabc::Worker*) fWorker();
 
   if (wrk) wrk->Submit(cmd);
}
 
// ================================================================================
 
 
dabc::Worker::Worker(Reference parent, const std::string &name) :
   Object(parent, name),
   fThread(),
   fAddon(),
   fPublisher(),
   fWorkerId(0),
   fWorkerPriority(-1), // minimum priority per default
 
   fThreadMutex(0),
   fWorkerActive(false),
   fWorkerFiredEvents(0),
 
   fWorkerCommands(CommandsQueue::kindNone),
 
   fWorkerCommandsLevel(0),
   fWorkerHierarchy(),
   fWorkerCfgId(-1)
{
   SetFlag(flHasThread, false);
 
#ifdef DABC_EXTRA_CHECKS
   DebugObject("Worker", this, 10);
#endif
}
 
 
dabc::Worker::Worker(const ConstructorPair& pair) :
   Object(pair),
   fThread(),
   fAddon(),
   fWorkerId(0),
   fWorkerPriority(-1), // minimum priority per default
 
   fThreadMutex(0),
   fWorkerActive(false),
   fWorkerFiredEvents(0),
 
   fWorkerCommands(CommandsQueue::kindNone),
 
   fWorkerCommandsLevel(0),
   fWorkerHierarchy(),
   fWorkerCfgId(-1)
{
   SetFlag(flHasThread, false);
 
#ifdef DABC_EXTRA_CHECKS
   DebugObject("Worker", this, 10);
#endif
}
 
dabc::Worker::~Worker()
{
   DOUT3("~Worker %p %d thrd:%p 000 ", this, fWorkerId, fThread());
 
   CancelCommands();
 
   DOUT3("~Worker %p %d thrd:%p 111", this, fWorkerId, fThread());
 
   // just to ensure that reference is gone
   ClearThreadRef();
 
   DOUT3("~Worker %p %d thrd:%p done", this, fWorkerId, fThread());
 
   #ifdef DABC_EXTRA_CHECKS
   DebugObject("Worker", this, -10);
   #endif
}
 
bool dabc::Worker::HasThread() const
{
   LockGuard lock(fThreadMutex);
 
   return !fThread.null();
}
 
dabc::ThreadRef dabc::Worker::thread()
{
 
   dabc::ThreadRef res;
 
   {
      // this is lock over thread main mutex,
      // if we can get it, we also can increment ref counter directly by the thread
      LockGuard lock(fThreadMutex);
 
      // we can acquire new reference without additional lock of the mutex
      fThread._AcquireThreadRef(res);
   }
 
   return res;
}
 
 
bool dabc::Worker::IsOwnThread() const
{
   LockGuard lock(fThreadMutex);
 
   if (fThread.null()) return false;
 
   return fThread.IsItself();
}
 
 
std::string dabc::Worker::ThreadName() const
{
   LockGuard lock(fThreadMutex);
 
   return std::string(fThread() ? fThread()->GetName() : "");
}
 
 
bool dabc::Worker::DestroyByOwnThread()
{
   // method can be called from any thread therefore we should first ensure that
   // correct reference will exists on the thread
   // Once true is returned thread guarantees that DestroyCalledFromOwnThread() method
   // will be called from thread context
   // One should also remember that reference counter is granted to thread
 
   ThreadRef thrd = thread();
 
   if (thrd()==0) return false;
 
   if (IsLogging())
      DOUT0("Worker %p %s ask thread to destroy it ", this, GetName());
 
   return thrd()->InvokeWorkerDestroy(this);
}
 
 
void dabc::Worker::ClearThreadRef()
{
   ThreadRef ref;
 
   {
      LockGuard lock(fThreadMutex);
      ref << fThread;
 
      fThreadMutex = 0;
      fWorkerId = 0;
      fWorkerActive = false;
//      fWorkerFiredEvents = 0;
   }
 
   if (IsLogging())
      DOUT0("Worker %s releases thread reference %p", GetName(), ref());
 
   ref.Release();
 
   dabc::LockGuard lock(fObjectMutex);
 
   SetFlag(Object::flHasThread, false);
 
}
 
void dabc::Worker::ObjectCleanup()
{
   // TODO: that is correct sequence - first delete child, than clean ourself  (current) or vice-versa
 
   DOUT4("START worker %s class %s cleanup refcnt = %d thrd %s publ %p publthrd %s", GetName(), ClassName(), fObjectRefCnt, thread().GetName(), fPublisher(), WorkerRef(fPublisher).thread().GetName());
 
   CleanupPublisher(false);
 
   // we do standard object cleanup - remove all our childs and remove ourself from parent
   dabc::Object::ObjectCleanup();
 
   // from this moment on no new commands/events
   DettachFromThread();
 
   // now process old commands
   if (fWorkerCommandsLevel>0) {
      EOUT("We are in real trouble - when Worker %s %p executes command, cleanup is called", GetName(), this);
      exit(076);
   }
 
   // process all postponed and submitted commands
   do {
      Command cmd = fWorkerCommands.PopWithKind(CommandsQueue::kindPostponed);
      if (cmd.null()) cmd = fWorkerCommands.PopWithKind(CommandsQueue::kindSubmit);
      if (cmd.null()) break;
      ProcessCommand(cmd);
   } while (true);
 
   // here we cancel all commands we submit or get as replied
   CancelCommands();
 
   // here addon must be destroyed or at least cross-reference removed
   // DOUT0("Worker:%s Destroy addon:%p in ObjectCleanup", GetName(), fAddon());
   fAddon.Release();
 
   DOUT4("DID worker %s class %s cleanup refcnt = %d", GetName(), ClassName(), fObjectRefCnt);
}
 
 
void dabc::Worker::AssignAddon(WorkerAddon* addon)
{
   if (!fAddon.null()) {
      // first remove addon from thread itself
 
      if (!fThread.null() && (fWorkerId>0))
         fThread()->WorkerAddonChanged(this, false);
 
      // clean worker pointer
      fAddon()->fWorker.Release();
 
      // release addon - it should be destroyed as soon as possible
      fAddon.Release();
   }
 
   fAddon = addon;
   if (addon) addon->fWorker = this;
 
   if (!fThread.null() && (fWorkerId>0) && addon) {
      fThread()->WorkerAddonChanged(this, true);
      addon->OnThreadAssigned();
   }
}
 
 
bool dabc::Worker::MakeThreadForWorker(const std::string &thrdname)
{
   std::string newname = thrdname;
 
   if (newname.empty()) {
      DOUT2("Thread name not specified - generate default, for a moment - processor name");
      if (GetName() != 0) newname = GetName();
   }
 
   if (newname.empty()) {
      EOUT("Still no thread name - used name Thread");
      newname = "Thread";
   }
 
   ThreadRef thrd = dabc::mgr.CreateThread(newname, RequiredThrdClass());
 
   return thrd.null() ? false : AssignToThread(thrd);
}
 
 
 
// FIXME: old code, need revising
 
 
bool dabc::Worker::AssignToThread(ThreadRef thrd, bool sync)
{
   if (HasThread()) {
      EOUT("Worker %s already has assigned thred %s", GetName(), ThreadName().c_str());
      return false;
   }
 
   if (thrd.null()) {
      EOUT("Thread is not specified");
      return false;
   }
 
   std::string thrdcl = RequiredThrdClass();
 
   if (thrdcl.length()>0)
     if (thrdcl.compare(thrd.ClassName())!=0) {
        EOUT("Processor requires class %s than thread %s of class %s" , thrdcl.c_str(), thrd.GetName(), thrd.ClassName());
        return false;
     }
 
   Reference ref(this);
   if (ref()==0) {
      EOUT("Cannot obtain reference on itself");
      return false;
   }
 
   // this indicates that we are not in the thread, any events/commands will be rejected
   fWorkerId = 0;
   fWorkerActive = false;
 
   fWorkerFiredEvents = 0;
 
   fThread = thrd; // we copy reference, no extra locks required
 
   fThreadMutex = fThread()->ThreadMutex();
 
   {
      dabc::LockGuard lock(fObjectMutex);
      SetFlag(Object::flHasThread, true);
   }
 
   DOUT2("Assign worker %s to thread sync = %s", GetName(), DBOOL(sync));
 
   return fThread()->AddWorker(ref, sync);
}
 
bool dabc::Worker::DettachFromThread()
{
   ThreadRef thrd = thread();
 
   if (thrd.null()) return true;
 
   bool res = thrd()->HaltWorker(this);
 
   if (res) ClearThreadRef();
 
   return res;
}
 
bool dabc::Worker::ActivateTimeout(double tmout_sec)
{
   LockGuard lock(fThreadMutex);
 
   return fThread._ActivateWorkerTimeout(fWorkerId, WorkerPriority(), tmout_sec);
}
 
void dabc::Worker::ProcessCoreEvent(EventId evnt)
{
   DOUT3("Processor %p %u thrd %p CoreEvent %u", this, fWorkerId, fThread(), evnt.GetCode());
 
   switch (evnt.GetCode()) {
      case evntSubmitCommand: {
 
         DOUT4("Process evntSubmitCommand proc %p %u thrd %p arg %u", this, fWorkerId, fThread(), evnt.GetArg());
 
         Command cmd;
         {
            LockGuard lock(fThreadMutex);
            cmd = fWorkerCommands.PopWithId(evnt.GetArg());
         }
 
         DOUT4("Thread:%p Worker: %s Command process %s lvl %d", fThread(), GetName(), cmd.GetName(), fWorkerCommandsLevel);
 
         ProcessCommand(cmd);
 
         // check if we have postponed commands
         while(fWorkerCommandsLevel==0) {
            {
               LockGuard lock(fThreadMutex);
               cmd = fWorkerCommands.PopWithKind(CommandsQueue::kindPostponed);
            }
            if (cmd.null()) break;
            ProcessCommand(cmd);
         }
 
         DOUT4("Process evntSubmitCommand done proc %p", this);
 
         break;
      }
      case evntReplyCommand: {
 
//         DOUT0("Process evntReplyCommand arg %u", evnt.GetArg());
 
         dabc::Command cmd;
         {
            LockGuard lock(fThreadMutex);
            cmd = fWorkerCommands.PopWithId(evnt.GetArg());
         }
 
         if (cmd.null())
            EOUT("evntReplyCommand: no command with specified id %u", evnt.GetArg());
         else {
            if (ReplyCommand(cmd)) cmd.Reply();
         }
 
         break;
      }
 
      default:
         EOUT("Core event %u arg:%u not processed", evnt.GetCode(), evnt.GetArg());
   }
}
 
int dabc::Worker::ProcessCommand(Command cmd)
{
   if (cmd.null() || !IsNormalState()) return cmd_false;
 
   DOUT3("ProcessCommand cmd %s lvl %d isync %s", cmd.GetName(), fWorkerCommandsLevel, DBOOL(cmd.IsLastCallerSync()));
 
   if (cmd.IsCanceled()) {
      cmd.Reply(cmd_canceled);
      return cmd_canceled;
   }
 
   // when other command is in processing state,
   // and this cmd not need to be synchronous,
   // one can process such command later
   // synchronous command must be processed immediately
 
   if ((fWorkerCommandsLevel>0) && !cmd.IsLastCallerSync()) {
      LockGuard lock(fThreadMutex);
      fWorkerCommands.Push(cmd, CommandsQueue::kindPostponed);
      return cmd_postponed;
   }
 
   IntGuard guard(fWorkerCommandsLevel);
 
   if (IsLogging())
     DOUT0("Worker %p %s process command %s", this, GetName(), cmd.GetName());
 
   int cmd_res = PreviewCommand(cmd);
 
   if (IsLogging())
     DOUT0("Worker %p %s did preview command %s ignored %s", this, GetName(), cmd.GetName(), DBOOL(cmd_res == cmd_ignore));
 
   if (cmd_res == cmd_ignore)
      cmd_res = ExecuteCommand(cmd);
 
   if (cmd_res == cmd_ignore) {
      EOUT("Command ignored %s", cmd.GetName());
      cmd_res = cmd_false;
   }
 
   // FIXME: is it only postponed command is not completed ???
   bool completed = cmd_res!=cmd_postponed;
 
   DOUT3("Thrd: %p Worker: %s ProcessCommand cmd %s lvl %d done", fThread(), GetName(), cmd.GetName(), fWorkerCommandsLevel);
 
   if (completed) cmd.Reply(cmd_res);
 
   return cmd_res;
}
 
void dabc::Worker::ProcessEvent(const EventId& evnt)
{
   EOUT("Event %u arg:%u not processed req:%s", evnt.GetCode(), evnt.GetArg(),
         RequiredThrdClass().c_str());
}
 
bool dabc::Worker::ActivateMainLoop()
{
   // call should be performed from the thread context
   if (fThread())
      return fThread()->SetExplicitLoop(this);
   else
      return false;
}
 
// all about parameters handling
 
dabc::Parameter dabc::Worker::Par(const std::string &name) const
{
   return FindChildRef(name.c_str());
}
 
dabc::RecordField dabc::Worker::Cfg(const std::string &name, Command cmd) const
{
 
   DOUT2("Worker %s Cfg %s", ItemName().c_str(), name.c_str());
 
   // first check in the command
   if (cmd.HasField(name)) return cmd.GetField(name);
 
   DOUT3("Check Cfg %s in own parameters", name.c_str());
 
   // second - as parameter
   dabc::RecordField res = Par(name).Value();
   if (!res.null()) return res;
 
   DOUT3("Check Cfg %s in xml file", name.c_str());
 
   // third - in xml file
   ConfigIO io(dabc::mgr()->cfg(), fWorkerCfgId);
   if (io.ReadRecordField((Object*) this, name, &res, 0)) {
      DOUT2("Worker %s Cfg %s xml %s", ItemName().c_str(), name.c_str(), res.AsStr().c_str());
      return res;
   }
 
   DOUT3("Check Cfg %s in parent parameters", name.c_str());
 
   // forth - in parameters of all parents
   Object* prnt = GetParent();
   while (prnt!=0) {
      res = WorkerRef(prnt).Par(name).Value();
      if (!res.null()) return res;
 
      prnt = prnt->GetParent();
   }
 
   return res;
}
 
dabc::Parameter dabc::Worker::CreatePar(const std::string &name, const std::string &kind)
{
   Parameter par = Par(name);
   if (par.null()) {
 
      bool hidden = (kind == "cmddef");
 
      ParameterContainer* cont = new ParameterContainer(this, name, kind, hidden);
 
      ConfigIO io(dabc::mgr()->cfg());
 
      io.ReadRecordField(this, name, 0, &(cont->Fields()));
 
      par = cont;
 
      cont->FireParEvent(parCreated);
   }
   return par;
}
 
void dabc::Worker::SetParValue(const std::string &name, const RecordField &v)
{
   Parameter par = Par(name);
   if (par.null()) return;
 
   par.SetValue(v);
 
   Hierarchy chld = fWorkerHierarchy.FindChild(name.c_str());
   if (!chld.null()) {
      par.ScanParamFields(&chld()->Fields());
      fWorkerHierarchy.MarkChangedItems();
   }
}
 
bool dabc::Worker::DestroyPar(const std::string &name)
{
   Parameter par = Par(name);
 
   if (par.null()) return false;
 
   par.Destroy();
 
   return true;
}
 
dabc::CommandDefinition dabc::Worker::CreateCmdDef(const std::string &name)
{
   return CreatePar(name, "cmddef");
}
 
bool dabc::Worker::Find(ConfigIO &cfg)
{
   DOUT3("Worker::Find %p name = %s parent %p", this, GetName(), GetParent());
 
   if (GetParent()==0) return false;
 
   while (cfg.FindItem(ClassName()))
      if (cfg.CheckAttr(xmlNameAttr, GetName())) return true;
 
   return false;
}
 
void dabc::Worker::WorkerParameterChanged(bool force_call, ParameterContainer *par, const std::string &value)
{
   if (force_call || IsOwnThread()) {
 
      unsigned mask = par->ConfirmFromWorker();
 
      if (mask & 1)
         ProcessParameterRecording(par);
 
      if (mask & 2)
         ProcessParameterEvent(CmdParameterEvent(par->GetName(), value, parModified));
   } else {
      CmdParameterEvent evnt(par->GetName(), value, parModified);
      evnt.SetBool("local", true);
      Submit(evnt);
   }
}
 
void dabc::Worker::ProcessParameterRecording(ParameterContainer *par)
{
   if (fWorkerHierarchy.null()) return;
 
   LockGuard guard(fWorkerHierarchy.GetHMutex());
 
   Hierarchy chld = fWorkerHierarchy.FindChild(par->GetName());
 
   if (!chld.null()) par->BuildFieldsMap(&chld()->Fields());
   fWorkerHierarchy.MarkChangedItems();
}
 
int dabc::Worker::PreviewCommand(Command cmd)
{
   int cmd_res = cmd_ignore;
 
   if (cmd.IsName(CmdSetParameter::CmdName())) {
 
      Parameter par = Par(cmd.GetStr(CmdSetParameter::ParName()));
 
      cmd_res = par.ExecuteChange(cmd);
 
   } else
 
   if (cmd.IsName(CmdParameterEvent::CmdName())) {
 
      ParameterEvent evnt(cmd);
 
      if (cmd.GetBool("local")) {
         Parameter par = Par(evnt.ParName());
         if (par.null()) {
            EOUT("FAIL to find local parameter %s", evnt.ParName().c_str());
            cmd_res = cmd_false;
         } else {
            WorkerParameterChanged(true, par(), evnt.ParValue());
            cmd_res = cmd_true;
         }
      } else {
         ProcessParameterEvent(evnt);
         cmd_res = cmd_true;
      }
 
   } else
 
   if (cmd.IsName("ObjectDestroyed")) {
      // process cleanup in worker thread
      ObjectDestroyed((Object*)cmd.GetPtr("#Object"));
      cmd_res = cmd_true;
   } else
 
   if (cmd.IsName("DestroyParameter")) {
      cmd_res = cmd_bool(DestroyPar(cmd.GetStr("ParName")));
   } else
 
   if (cmd.IsName("SyncWorker")) {
      // this is just dummy command, which is submitted with minimum priority
      cmd_res = cmd_true;
   } else
 
   if (cmd.IsName("DeleteAddon")) {
     // this is way to delete addon
     AssignAddon(nullptr);
     cmd_res = cmd_true;
   } else
 
   if (cmd.IsName(CmdPublisher::CmdName())) {
      dabc::Hierarchy h = (HierarchyContainer*) cmd.GetPtr("hierarchy");
      HierarchyStore* store = (HierarchyStore*) cmd.GetPtr("store");
      unsigned version = cmd.GetUInt("version");
 
      if (!h.null()) {
 
         BeforeHierarchyScan(h);
 
         LockGuard lock(h.GetHMutex());
 
         Buffer diff = h.SaveToBuffer(dabc::stream_NamesList, version);
         cmd.SetRawData(diff);
 
         cmd.SetUInt("version", h.GetVersion());
 
         if (store) store->ExtractData(h);
 
         cmd_res = cmd_true;
      }
 
      if (store) store->WriteExtractedData();
 
   } else
   if (cmd.IsName(dabc::CmdGetBinary::CmdName())) {
 
      dabc::Hierarchy h = (HierarchyContainer*) cmd.GetPtr("hierarchy");
      if (h.null()) return cmd_ignore;
 
      std::string item = cmd.GetStr("subitem");
      std::string fullname = cmd.GetStr("Item");
      std::string binkind = cmd.GetStr("Kind");
      std::string query = cmd.GetStr("Query");
 
      DOUT4("Worker %s Process CmdGetBinary item:%s full:%s kind %s", GetName(), item.c_str(), fullname.c_str(), binkind.c_str());
 
      std::string surl = "getitem";
      if (query.length()>0) { surl.append("?"); surl.append(query); }
 
      dabc::Url url(surl);
      if (!url.IsValid()) {
         EOUT("Cannot decode query url %s", query.c_str());
         return cmd_ignore;
      }
 
      unsigned hlimit(0);
      uint64_t version(0);
      int compact(0);
      bool with_childs(false);
 
      if (url.HasOption("history")) {
         int hist = url.GetOptionInt("history", 0);
         if (hist>0) hlimit = (unsigned) hist;
      }
      if (url.HasOption("version")) {
         int v = url.GetOptionInt("version", 0);
         if (v>0) version = (unsigned) v;
      }
      if (url.HasOption("compact"))
         compact = url.GetOptionInt("compact", 3);
      if (url.HasOption("childs"))
         with_childs = true;
 
      LockGuard lock(h.GetHMutex());
 
      dabc::Hierarchy sub = h.GetFolder(item);
 
      if (binkind=="hierarchy") {
         if (sub.null()) return cmd_ignore;
         // we record only fields, everything else is ignored - even name of entry is not stored
         Buffer raw = sub.SaveToBuffer(with_childs ? dabc::stream_Full : dabc::stream_Value, version, hlimit);
         if (raw.null()) return cmd_ignore;
 
         cmd.SetRawData(raw);
         cmd.SetUInt("version", sub.GetVersion());
         cmd_res = cmd_true;
      } else
      if (binkind=="cmd.json") {
         std::string cmdname = url.GetOptionStr("command");
         // make protection - append prefix to exclude misuse of interface
         if (!cmdname.empty()) cmdname = std::string("HCMD_") + cmdname; else
         if (sub.HasField("_numargs")) cmdname = "ROOTCMD";
         if (cmdname.empty()) return cmd_ignore;
         dabc::Command subcmd(cmdname);
         subcmd.SetRef("item", sub);
         subcmd.SetStr("query", query);
 
         // we misuse command interface for executing actions together with subitem
         // one could have problems if any
         if (ExecuteCommand(subcmd)!=cmd_true) return cmd_ignore;
 
         Buffer raw = subcmd.GetRawData();
         cmd.SetRawData(raw);
 
         cmd_res = cmd_true;
      } else
      if ((binkind=="get.json") || (binkind=="value.json") || (binkind=="item.json") || (binkind=="get.xml") || (binkind=="dabc.json") || (binkind=="dabc.xml")) {
         std::string field = "";
         if (binkind=="value.json") field = "value";
                               else field = url.GetOptionStr("field", "");
 
         if (sub.null() && field.empty()) {
            size_t separ = item.find_last_of('/');
            if ((separ != std::string::npos) && (separ>0) && (separ < item.length()-1)) {
               field = item.substr(separ+1);
               item.resize(separ);
               sub = h.GetFolder(item);
            }
         }
 
         if (sub.null()) return cmd_ignore;
 
         // DOUT0("Request JSON for item %s field %s compact %d", item.c_str(), field.c_str(), compact);
 
         bool isxml = binkind.find(".xml")!=std::string::npos;
 
         std::string replybuf;
 
         dabc::NumericLocale loc; // ensure correct locale for conversion
 
         if (field.empty()) {
            if (compact<0) compact = 0; else
            if (compact>storemask_Compact) compact = storemask_Compact;
            unsigned mask = compact;
            if (hlimit>0) mask |= storemask_NoChilds | dabc::storemask_History | storemask_TopVersion;
            if (isxml) mask |= dabc::storemask_AsXML;
 
            dabc::HStore store(mask);
            store.SetLimits(version, hlimit);
            if (sub.SaveTo(store))
               replybuf = store.GetResult();
 
         } else {
            if (!sub.HasField(field)) return cmd_ignore;
 
            replybuf = sub.GetField(field).AsJson();
         }
 
         cmd.SetStrRawData(replybuf);
 
         cmd_res = cmd_true;
      }
   }
 
   return cmd_res;
}
 
 
int dabc::Worker::ExecuteCommand(Command cmd)
{
   return cmd_false;
}
 
bool dabc::Worker::ReplyCommand(Command cmd)
{
   return true;
}
 
 
bool dabc::Worker::ExecuteIn(dabc::Worker* dest, dabc::Command cmd)
{
   // this is pointer of thread from which command is called
   ThreadRef thrd = thread();
 
   // we must be sure that call is done from thread itself - otherwise it is wrong
   if (!thrd.null()) return false;
 
   return thrd()->RunCommandInTheThread(this, dest, cmd) > 0;
}
 
bool dabc::Worker::Execute(Command cmd, double tmout)
{
   if (cmd.null()) return false;
 
   if (tmout>0) cmd.SetTimeout(tmout);
 
   ThreadRef thrd;
   bool exe_direct(false);
 
   if (IsLogging())
      DOUT0("Worker %p %s Executes command %s", this, GetName(), cmd.GetName());
 
   {
      LockGuard lock(fThreadMutex);
 
      if (fThread.null() || (fWorkerId==0)) {
 
         // command execution possible without thread,
         // but only manager allows to do it without warnings
         if (this != dabc::mgr())
            EOUT("warning: Cannot execute command %s without working thread, execute directly", cmd.GetName());
         exe_direct = true;
      } else
      if (fThread.IsItself()) {
         // DOUT0("Mutex = %p thrdmutex %p locked %s", fThreadMutex, fThread()->ThreadMutex(), DBOOL(fThreadMutex->IsLocked()));
         fThread._AcquireThreadRef(thrd);
      }
   }
 
   if (exe_direct) return ProcessCommand(cmd) > 0;
 
   // if command was called not from the same thread,
   // find handle of caller thread to let run its event loop
   if (thrd.null())
      thrd = dabc::mgr.CurrentThread();
 
   if (thrd()) {
      return thrd()->RunCommandInTheThread(0, this, cmd) > 0;
      //if (res!=cmd_ignore) return res>0;
      //return ProcessCommand(cmd) > 0;
   }
 
   // if there is no Thread with such id (most probably, some user-managed thrd)
   // than we create fake object only to handle commands and events,
   // but not really take over thread mainloop
   // This object is not seen from the manager, therefore many such instances may exist in parallel
 
   DOUT3("**** We really creating dummy thread for cmd %s worker:%p %s", cmd.GetName(), this, GetName());
 
   Thread curr(Reference(), "Current");
 
   curr.Start(0, false);
 
   return curr.RunCommandInTheThread(0, this, cmd) > 0;
}
 
dabc::Command dabc::Worker::Assign(dabc::Command cmd)
{
   if (cmd.null()) return cmd;
 
   {
      LockGuard lock(fThreadMutex);
 
      if (fThread()==0) {
         EOUT("Worker: %s - command %s cannot be assigned without thread", GetName(), cmd.GetName());
         return Command();
      }
 
      fWorkerCommands.Push(cmd, CommandsQueue::kindAssign);
   }
 
   cmd.AddCaller(this, 0);
 
   return cmd;
}
 
bool dabc::Worker::CanSubmitCommand() const
{
   LockGuard lock(fThreadMutex);
 
   return !fThread.null() && fWorkerActive;
}
 
bool dabc::Worker::Submit(dabc::Command cmd)
{
   if (cmd.null()) return false;
 
   int priority = cmd.GetPriority();
 
   {
      LockGuard lock(fThreadMutex);
 
      // we can submit commands when worker is active (does not tries to halt)
      if (!fThread.null())
         if (fWorkerActive || (priority == priorityMagic)) {
 
            if (priority == priorityMagic) priority = fWorkerPriority; else
            if (priority == priorityDefault) priority = 0; else
            if (priority == priorityMinimum) priority = -1;
 
            uint32_t arg = fWorkerCommands.Push(cmd, CommandsQueue::kindSubmit);
 
            DOUT5("Fire event for worker %d with priority %d", fWorkerId, priority);
 
            fThread()->_Fire(EventId(evntSubmitCommand, fWorkerId, arg), priority);
 
            fWorkerFiredEvents++;
 
            return true;
      }
   }
 
   DOUT0("Worker:%s Command %s cannot be submitted - thread is not assigned", GetName(), cmd.GetName());
 
   cmd.ReplyFalse();
 
   return false;
}
 
bool dabc::Worker::GetCommandReply(dabc::Command& cmd, bool* exe_ready)
{
   if (cmd.null()) return false;
 
   LockGuard lock(fThreadMutex);
 
   DOUT3("GetCommandReply %s exe_ready %p Thread %p %s", cmd.GetName(), exe_ready, fThread(), fThread()->GetName());
 
   if (exe_ready) {
      if (_FireDoNothingEvent()) {
         // we must be sure that event is accepted, it can be that worker is just break all its activity and pointer is no longer valid
         *exe_ready = true;
         return true;
      }
 
      return false;
   }
 
   // here command will be overtaken from the reference
   uint32_t id = fWorkerCommands.ChangeKind(cmd, CommandsQueue::kindReply);
 
   if (!_FireEvent(evntReplyCommand, id, 0)) {
      fWorkerCommands.PopWithId(id);
      if (!cmd.HasField("#no_warnings")) EOUT("Worker %s don't want get command for reply", GetName());
      return false;
   }
 
   return true;
}
 
void dabc::Worker::CancelCommands()
{
   fWorkerCommands.Cleanup(fThreadMutex, this, cmd_false);
}
 
void dabc::Worker::WorkerSleep(double tmout)
{
   ThreadRef thrd = thread();
 
   if (!thrd.null() && thrd.IsItself()) {
      thrd.RunEventLoop(tmout);
      return;
   }
 
   thrd = dabc::mgr.CurrentThread();
   if (!thrd.null()) {
      thrd.RunEventLoop(tmout);
      return;
   }
 
   dabc::Sleep(tmout);
}
 
 
bool dabc::Worker::RegisterForParameterEvent(const std::string &mask, bool onlychangeevent)
{
   return dabc::mgr.ParameterEventSubscription(this, true, mask, onlychangeevent);
}
 
 
bool dabc::Worker::UnregisterForParameterEvent(const std::string &mask)
{
   return dabc::mgr.ParameterEventSubscription(this, false, mask);
}
 
 
std::string dabc::Worker::WorkerAddress(bool full)
{
   return full ? dabc::mgr.ComposeAddress("",ItemName()) : ItemName();
}
 
 
dabc::Reference dabc::Worker::GetPublisher()
{
   if (fPublisher.null()) fPublisher = dabc::mgr.FindItem(dabc::Publisher::DfltName());
   return fPublisher;
}
 
 
bool dabc::Worker::Publish(const Hierarchy& h, const std::string &path)
{
   return PublisherRef(GetPublisher()).Register(path, ItemName(), h());
}
 
 
bool dabc::Worker::PublishPars(const std::string &path)
{
   // no need to publish if publisher not exists
   if (GetPublisher().null()) return true;
 
   if (fWorkerHierarchy.null())
      fWorkerHierarchy.Create("Worker");
 
   for (unsigned n=0;n<NumChilds();n++) {
      Parameter par = dynamic_cast<ParameterContainer*> (GetChild(n));
      if (par.null()) continue;
 
      Hierarchy chld = fWorkerHierarchy.FindChild(par.GetName());
      if (!chld.null()) continue;
 
      chld = fWorkerHierarchy.CreateHChild(par.GetName());
 
      if (par.IsRatemeter() || (par.Kind() == "info") || par.GetField("#record").AsBool()) {
         chld.EnableHistory(100);
         par()->fRecorded = true;
      }
 
      par.ScanParamFields(&chld()->Fields());
   }
   fWorkerHierarchy.MarkChangedItems();
 
   return Publish(fWorkerHierarchy, path);
}
 
 
bool dabc::Worker::Unpublish(const Hierarchy& h, const std::string &path)
{
   return PublisherRef(GetPublisher()).Unregister(path, ItemName(), h());
}
 
bool dabc::Worker::Subscribe(const std::string &path)
{
   return PublisherRef(GetPublisher()).Subscribe(path, ItemName());
}
 
bool dabc::Worker::Unsubscribe(const std::string &path)
{
   return PublisherRef(GetPublisher()).Unsubscribe(path, ItemName());
}
 
void dabc::Worker::CleanupPublisher(bool sync)
{
   if (fPublisher.null()) return;
 
   PublisherRef ref;
 
   ref << fPublisher;
 
   if (!ref.thread().null()) ref.RemoveWorker(ItemName(), sync);
}
 
// ===========================================================================================
 
bool dabc::WorkerRef::Submit(dabc::Command cmd)
{
   if (GetObject()) return GetObject()->Submit(cmd);
 
   cmd.ReplyFalse();
   return false;
}
 
bool dabc::WorkerRef::Execute(Command cmd, double tmout)
{
   if (GetObject()==0) return false;
   return GetObject()->Execute(cmd, tmout);
}
 
bool dabc::WorkerRef::Execute(const std::string &cmd, double tmout)
{
   if (GetObject()==0) return false;
   return GetObject()->Execute(cmd, tmout);
}
 
dabc::Parameter dabc::WorkerRef::Par(const std::string &name) const
{
   if (GetObject()==0) return dabc::Parameter();
   return GetObject()->Par(name);
}
 
bool dabc::WorkerRef::HasThread() const
{
   return GetObject() ? GetObject()->HasThread() : false;
}
 
bool dabc::WorkerRef::IsSameThread(const WorkerRef& ref)
{
   if (GetObject() && ref.GetObject())
      return GetObject()->thread()() == ref.GetObject()->thread()();
 
   return false;
}
 
 
bool dabc::WorkerRef::CanSubmitCommand() const
{
   return GetObject() ? GetObject()->CanSubmitCommand() : false;
}
 
bool dabc::WorkerRef::SyncWorker(double tmout)
{
   if (GetObject()==0) return true;
 
   dabc::Command cmd("SyncWorker");
   cmd.SetPriority(dabc::Worker::priorityMinimum);
   if (tmout>=0) cmd.SetTimeout(tmout);
   return Execute(cmd) == cmd_true;
}