path.hh

Go to the documentation of this file.

#ifndef _DRAMA2_PATH_INC
#define _DRAMA2_PATH_INC
/*  
 * History:
      06-Mar-2014 - TJF - Original version. Starting from dcpppath.h
      30-Sep-2016 - TJF - Start of history after development.
                          Add  MonForwardEventHandler class used as event 
                           handler for monitors, main point is the triggers 
                           don't cause exceptions.
                          Don't throw exceptions when messages return bad 
                           status, the event handlers can do that, otherwise 
                           event handlers cannot stop such errors occurring.
                          Add Two new overrides of MonitorForward().
      30-Sep-2016 - TJF - Add SpawnKickArg() and SpawnKickArgUpdate() static
                           methods.
                          
                          
 
 * The above ID is for Doxygen, this one has the format ACMM is looking for.
 * "@(#) $Id$"
 */
#include "drama.hh"
#include "DitsInteraction.h"
#include "DitsFix.h"
#include "DitsBulk.h"
#include <drama/thread.hh>
#include <set>
#include <atomic>
 
#include <initializer_list>
 
namespace drama {
 
    inline std::chrono::steady_clock::time_point CreateFutureTimepoint(double secs) 
    {
        // Get now with the steady clock.
        std::chrono::steady_clock::time_point until = 
            std::chrono::steady_clock::now();
 
        // Duration with seconds in double.
        std::chrono::duration<double> myDuration(secs);
 
        /*
         * When adding - we need the duration_cast as double has a higher
         * resolution then steady_clock, since the later uses an integral type.
         *
         * But we get the highest resolution result we can wait on.
         */
        until += std::chrono::duration_cast<std::chrono::steady_clock::duration>(myDuration);
 
        return until;
 
    }
    class Buffers {
    private:
        DitsPathInfoType _bufInfo;
    public:
        Buffers( long int MessageBytes = 800,
                 long int MaxMessages = 2,
                 long int ReplyBytes = 800,
                 long int MaxReplies = 10) {
            _bufInfo.MessageBytes = MessageBytes;
            _bufInfo.MaxMessages  = MaxMessages;
            _bufInfo.ReplyBytes   = ReplyBytes;
            _bufInfo.MaxReplies   = MaxReplies; 
        }
        Buffers ( const DitsPathInfoType & info) : _bufInfo(info) {};
        operator const DitsPathInfoType * () const { 
                return &_bufInfo; 
        }
        long int MessageBytes() const { return _bufInfo.MessageBytes; }
        long int MaxMessages()  const { return _bufInfo.MaxMessages;  }
        long int ReplyBytes()   const { return _bufInfo.ReplyBytes;   }
        long int MaxReplies()   const { return _bufInfo.MaxReplies;   };
    }; // class Buffers.
 
    typedef std::vector<ErsMessageType> ErsMessageVector;
 
 
 
    class MessageEventHandler : public thread::TransEvtProcessor {
    private:
        /*
         * Converts argument to a MsgOut() message into call to MessageUser().
         */
        void MessageUserRaw(thread::ProcessInfo messInfo, const sds::IdPtr &arg);
 
        /*
         * Coverts argument to a Ers() message and calls ErrorReport().
         */
        virtual void ErrorReportRaw(thread::ProcessInfo messInfo,
                                 const sds::IdPtr &arg);
 
 
    public:
        virtual bool Process(thread::ProcessInfo         messInfo,
                             const TransEvtInfo & eventInfo,
                             const sds::IdPtr &arg) override final;
 
        virtual void NewTransaction(DitsMsgType msgType, DitsTransIdType tid) override;
 
 
 
        virtual void ThreadWaitAbort(thread::ProcessInfo messInfo,StatusType status);
        virtual void TriggerReceived(thread::ProcessInfo messInfo,
                                StatusType status, const sds::IdPtr &arg);
 
 
        virtual void SignalReceived(thread::ProcessInfo messInfo,
                                    const sds::IdPtr &arg);
 
 
        virtual void MessageRejected(thread::ProcessInfo messInfo,StatusType status);
 
        virtual void MessageComplete(thread::ProcessInfo messInfo,
                                 StatusType status, const sds::IdPtr &arg);
 
        virtual void TaskDied(thread::ProcessInfo messInfo, StatusType status);
 
        virtual void UserMessage(thread::ProcessInfo messInfo, 
                                 const std::string &task,
                                 const std::string &message);
 
        virtual void ErrorReport(thread::ProcessInfo messInfo,
                                 const std::string &task,
                                 const ErsMessageVector &messages);
 
        virtual bool KickReceived(thread::ProcessInfo messInfo,
                                  const sds::IdPtr &arg);
 
 
        virtual void BulkTransferred(thread::ProcessInfo messInfo,
                                     DitsBulkInfoType bulkInfo);
 
        virtual void BulkDone(thread::ProcessInfo messInfo);
 
    };
    class MonForwardEventHandler : public MessageEventHandler {
    public:
        virtual void TriggerReceived(thread::ProcessInfo messInfo,
                                     StatusType status, 
                                     const sds::IdPtr &arg) override;
 
    };
 
    class GetPathEventHandler : public thread::TransEvtProcessor {
    public:
        virtual bool Process(thread::ProcessInfo messInfo,
                             const TransEvtInfo & eventInfo,
                             const sds::IdPtr &arg) override;        
 
        virtual void NewTransaction(DitsMsgType msgType, DitsTransIdType tid) override;
    };
 
    class IsRunningType : public drama::MessageEventHandler {
    public:
        enum class EventStatus { 
            NotStarted,     
                Running,        
                Failed,         
                Completed       
                };
        EventStatus Get() const { return _status; };
        
        std::string GetStr(EventStatus val) const;
 
        std::string GetStr() const { return GetStr(_status); }
 
 
        bool operator()() const {
            if (_status == EventStatus::Running)
            {
                //std::cerr << "EventStatus true" << std::endl;
                return true;
            }
            else
            {
                //std::cerr << "EventStatus false" << std::endl;
            return false;
            }
        }
        IsRunningType() {
            _status = EventStatus::NotStarted;
        }
    protected:
        void SetStatus(EventStatus newVal) {
            _status = newVal;
        }
 
 
    private:
        /* The variable we maintain. Use atomic since this intended for 
         * read/write from one thread, read from another
         */
        std::atomic<EventStatus> _status;
 
    protected:
        void NewTransaction(DitsMsgType msgType, DitsTransIdType tid) override;
 
        /* We won't override ThreadWaitAbort, what to do is unclear as the 
         * transaction remains running */
 
        void MessageRejected(drama::thread::ProcessInfo messInfo,
                             StatusType status) override;
        virtual void MessageComplete(drama::thread::ProcessInfo messInfo,
                                     StatusType status, 
                                     const drama::sds::IdPtr &arg) override;
        virtual void TaskDied(drama::thread::ProcessInfo messInfo, 
                              StatusType status) override;
    
    }; // Class IsRunningType.
 
 
    class Path  {
 
    private:
 
        /*
         * We need a static list of Path objects, so that we can handle
         * DITS Disconnected events and translate them to Path objects to
         * be notified.
         *
         * Note - I traditionally use a list here, but in fact, a std::set
         * makes more sense.  We define the type here, but use a
         * static function to construct and access the object to avoid
         * static initialization order issues.
         */
        typedef std::set<Path *> ObjectListType;
        static ObjectListType &ObjectList();
 
 
        /* Indicates if the disconnect handler has been enabled.*/
        static bool _DisconnectEnabled;
 
        /*
         * The disconnect handler details when our disconnect handler
         * was enabled.
         */
        static DitsDisConnectRoutineType _OrigDisconnect;
        static DVOIDP _OrigDisconnectData;
 
        /*
         * Basic message even handler.  
         */
        static MessageEventHandler _simpleEventProcessor;
        static GetPathEventHandler _getPathEventProcessor;
        static MonForwardEventHandler _simpleMonEventProcessor;
        /*
         * Pointer to the DRAMA task we are part of.
         * (Note, in other examples, we make the pointer const, but
         *  if we do that here, we can't use the move assignment operator)
         */
        std::weak_ptr<Task> _theTask;
       
 
        /* The path to the task, if state != INITIAL    */      
        DitsPathType _path = nullptr;
        /* The state of this task object (NOTIFY not yet supported */
        enum class PathState { INITIAL, SELFINIT, PATHWAIT, LOADING, 
                                   ACTIVE, DIED, NOTIFY } 
            _state = PathState::INITIAL;
        /* Buffer sizes                         */
        Buffers _buffers;
        /* Flags to the PathGet operation       */
        int _pathFlags = 0;
        /* Task priority                        */
        int _priority = 0;
        /* Flags to the load command            */
        int _loadFlags = 0;
        /* Set true to indicate the last GetPath loaded this task */
        bool _loaded = false;
        /* Do we log loads                      */
        bool _logLoad = true;
        /* The name of the task                 */
        std::string _taskName;
        /* The location of the task             */
        std::string _taskLocation;
        /* The file to load the task from       */
        std::string _taskFile;
        /* The requested process name           */
        std::string _process;
        /* Argument to the load command */
        std::string _argument;
        /* Decode string                */
        std::string _decode;
 
        unsigned _stackSize = 0;
 
        /*
         * Various internal routines.
         */
        /* Enable the disconnect handler, if not already enabled 
         *   Can only be called if DitsAppInit() has been run, hence
         *   not from normally a constructor.
         */
        void EnableDisconnect();
 
        /*
         * Invoked by constructors to do common stuff.
         */
        void ConstructorHelper();
 
 
    public:  /* These two can be public */
        /*
         * Return a string (char *) version of a state enum.  Used in
         * logging and error messages.
         */
        const char *StateString(PathState state) const;
        /*
         * Return a string (char *) version of the current state.  Used in
         * logging and error messages.
         */
        const char *StateString() const {
            return StateString(_state);
        }
    private:
 
        /*
         * Start the GetPath operation, returning the transaction id.
         */
        DitsTransIdType GetPathStart(thread::TMessHandler *action);
 
        /*
         * Load the task.
         */
        DitsTransIdType LoadTask(thread::TMessHandler *action);
        /*
         * A load (started by GetPathStart() has completed, process it
         * and set the GetPath operation
         */
        DitsTransIdType LoadCompleteGetPath(
            thread::TMessHandler *action,
            DitsTransIdType tid,
            const TransEvtInfo &eventInfo); // Transaction event details.
 
        /*
         * Complete the GetPath operation.
         */
        void GetPathComplete(
            const TransEvtInfo &eventInfo); // Transaction event details.
 
        /*
         * Underlying send routine used by the various methods which send
         * a DRAMA message. 
         *
         * @param action  A pointer to the threaded action which is
         *                executing this operation.
         * @param type    Message type.
         * @param name    Name of the action to start.
         * @param argIn   Argument to the action.
         * @param eventProcessor An object to use to process events.  
         *                All Send() does is to invoke the NewTransaction()
         *                 method when it has sent the message.
         *
         * @return The DRAMA transaction ID of the message.
         */
        DitsTransIdType Send(
            thread::TMessHandler *action, 
            DitsMsgType Type,
            const std::string & name, 
            int flags,
            const sds::Id &arg,
            MessageEventHandler * const eventProcessor);
 
 
        /*
         * Underlying send bulk routine used by the various methods which send
         * a DRAMA message. 
         *
         * @param action  A pointer to the threaded action which is
         *                executing this operation.
         * @param type    Message type.
         * @param name    Name of the action to start.
         * @param argIn   The Bulk Data segment to be the argument to the 
         *                 obey message.
         * @param isSds   Does arg contain an SDS item?  If true,
         *                  the target task can access the item as an
         *                  argument using using SDS in the normal way
         *                  If false, the shared memory contains data in an 
         *                  application private format and DRAMA makes no attempt
         *                  to interpret it.
         * @param notifyBytes If non-zero, it indicates your action should
         *                  be notified (using EntryCode::BulkTransferred
         *                  entries) every time the specified number
         *                  of bytes are transferred.   Note the this effect 
         *                  is somewhat dependent on the target task 
         *                  behavior and you may only receive the final 
         *                  EntryCode::BulkDone message.
         * @param eventProcessor An object to use to process events.  
         *                All Send() does is to invoke the NewTransaction()
         *                 method when it has sent the message.
         *
         * @return The DRAMA transaction ID of the message.
         */
        DitsTransIdType SendBulk(
            thread::TMessHandler *action, 
            DitsMsgType Type,
            const std::string & name, 
            int flags,
            BulkData *argIn,
            bool isSds,
            unsigned notifyBytes,
            MessageEventHandler * const eventProcessor);
 
 
        Path& operator=(const Path &rhs) = delete;
        Path(const Path &source) = delete;
 
    protected:
        virtual void Disconnected();
 
 
 
    public:
        Path& operator=(Path &&rhs) = default;
        Path(Path &&source) = default;
        /* Normal constructor
         *
         * Allows the the task name, host and file to be set
         *
         * @note DRAMA does not directly deal with long program names (value of
         *     the file argument).  The limit is about 63 characters.  The work
         *     around for this is to use environment variables known to IMP_Master 
         *     on the machine doing the loading (node).  The normal approach is to 
         *     use the ENVVAR:program approach to specifying the node name.  E.g.
         *     rather then specifying a filename like this:
         * <pre>
         *     /directory1/diretory2/directory3/directory4/program
         * </pre>
         *
         *     There would be an environment variable know to the IMP_Master
         *     program defined like this:
         * <pre>
         *     export ENVVAR=/directory1/diretory2/directory3/directory4/
         * </pre>
         *     And then the ENVVAR:program name will work.  Remember that
         *     IMP_Master is normally started using dits_netstart.
         *
         *
         * @param theTask   The DRAMA Task we are part of.  This is passed
         *                     by pointer, but Path is not taking ownership of
         *                     the subject object, which must continue to exist
         *                     until the Path object is destroyed.  
         * @param name The name the task will be known as (unless loaded, 
         *              when the name the task registers as will be used)
         * @param host The node name on which the task is running or
         *              is to be loaded.  Only used if not already running
         *              locally or known locally.
         * @param file  The file of the task, as required by 
         *                 DCF(DitsLoad).  Normally a file name, but may
         *                 be different depending on the location, e.g.
         *                 on a VxWorks machine.
         */
        Path(std::weak_ptr<Task> theTask,
             const std::string &name, 
             const std::string &host = "", 
             const std::string &file = "");
 
 
       /* Normal constructor
         *
         * Allows the the task name, host and file to be set
         *
         * @note DRAMA does not directly deal with long program names (value of
         *     the file argument).  The limit is about 63 characters.  The work
         *     around for this is to use environment variables known to IMP_Master 
         *     on the machine doing the loading (node).  The normal approach is to 
         *     use the ENVVAR:program approach to specifying the node name.  E.g.
         *     rather then specifying a filename like this:
         * <pre>
         *     /directory1/diretory2/directory3/directory4/program
         * </pre>
         *
         *     There would be an environment variable know to the IMP_Master
         *     program defined like this:
         * <pre>
         *     export ENVVAR=/directory1/diretory2/directory3/directory4/
         * </pre>
         *     And then the ENVVAR:program name will work.  Remember that
         *     IMP_Master is normally started using dits_netstart.
         *
         *
         * @param theTask   The DRAMA Task we are part of.  This is passed
         *                     by pointer, but Path is not taking ownership of
         *                     the subject object, which must continue to exist
         *                     until the Path object is destroyed.  
         * @param name The name the task will be known as (unless loaded, 
         *              when the name the task registers as will be used)
         * @param host The node name on which the task is running or
         *              is to be loaded.  Only used if not already running
         *              locally or known locally.
         * @param file  The file of the task, as required by 
         *                 DCF(DitsLoad).  Normally a file name, but may
         *                 be different depending on the location, e.g.
         *                 on a VxWorks machine.
         */
        Path(std::shared_ptr<Task> theTask,
             const std::string &name, 
             const std::string &host = "", 
             const std::string &file = ""); 
 
 
        explicit Path(std::weak_ptr<Task> theTask);
 
 
        Path(std::weak_ptr<Task> theTask, DitsPathType path);
 
        Path() { }
 
        virtual ~Path();  
        void Report(thread::TMessHandler *action) const;
        /*
         *      Inquiries
         */
        bool GetPathLoaded() const { return (_loaded); };
        const std::string GetTaskName()  const; 
        const std::string &GetHost()  const { return (_taskLocation); }
        const std::string &GetArgument() const { return (_argument); };
 
        /*
         * Routines to set various load and get path options.
         */
        virtual void SetName(const std::string &TaskName);
        virtual void SetHost(const std::string & Host) {
            _taskLocation = Host;
        }
        virtual void SetFile(const std::string &File) {
            _taskFile = File;
        }
        virtual void SetBuffers(const Buffers & buffs) {
            this->_buffers = buffs;
        }
        virtual void SetFlowControl() {
            _pathFlags |= DITS_M_FLOW_CONTROL;
        }
        virtual void SetProcess(const std::string &ProcessName);
 
        virtual void SetStackSize(unsigned bytes);
 
        virtual void SetArgument(const std::string &LoadArg, 
                                 bool const Append=false);
        virtual void SetPriority(const int Priority, 
                                 const bool Absolute = false);
 
        virtual void SetNames(const bool Flag = true) {
            if (Flag)
                _loadFlags |= DITS_M_NAMES;
            else
                _loadFlags &= ~DITS_M_NAMES;
        }
        virtual void SetSymbols(const bool Flag = true) {
            Task::guardType DramaLock(std::shared_ptr<Task>(_theTask)->Lock()); // Grab DRAMA Lock if we don't have it.
            if (Flag)
                _loadFlags |= DITS_M_SYMBOL;
            else
                _loadFlags &= ~DITS_M_SYMBOL;
        };
        virtual void SetProg(const bool Flag = true) {
            Task::guardType DramaLock(std::shared_ptr<Task>(_theTask)->Lock()); // Grab DRAMA Lock if we don't have it.
            if (Flag)
                _loadFlags |= DITS_M_PROG;
            else
                _loadFlags &= ~DITS_M_PROG;
        };
        virtual void LogLoad(const bool Flag = true) {
            _logLoad = Flag;
        }
 
        virtual void ClearState() {
            _state = PathState::INITIAL;
        };
        virtual bool Active() const {
            return ((_state == PathState::ACTIVE)||
                    (_state == PathState::SELFINIT));
        }
        virtual bool IsDead() const {
            return (_state == PathState::DIED);
        }
        virtual bool Initial() const {
            return ((_state == PathState::INITIAL)||
                    (_state == PathState::DIED));
        }
        virtual bool Busy() const {
            return ((_state == PathState::PATHWAIT) ||
                    (_state == PathState::LOADING)   ||
                    (_state == PathState::NOTIFY));
        }
 
        virtual std::shared_ptr<Task>  GetTask() const {
            return std::shared_ptr<Task>(_theTask);
        }
 
        void GetPath(thread::TMessHandler *action,
                     thread::TransEvtProcessor  * const eventProcessor =  &_getPathEventProcessor );
 
        bool GetPathWaitUntil(std::chrono::steady_clock::time_point until,
                              thread::TMessHandler *action,
                              thread::TransEvtProcessor  * const eventProcessor =  &_getPathEventProcessor );
 
        void Obey(thread::TMessHandler *action, 
                  const std::string &name,
                  const sds::Id &argIn = sds::Id::CreateNullItem(),
                  sds::IdPtr * const argOut = nullptr,
                  MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
#if 0
            fprintf(stderr,"!!!OBey - sending action %s\n", name.c_str());
#endif
 
            Send(action, DITS_MSG_OBEY, name, 0 /* Flags */, argIn, eventProcessor);
#if 0            
            fprintf(stderr,"!!!Obey sent, waiting for reply\n");
#endif
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask, 
                                       &eventInfo, eventProcessor, argOut);
#if 0
            fprintf(stderr,"!!!Obey sent, wait complete\n");
#endif
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Obey of message {0} to task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Obey of message {0} to task {1}, returned but not complete",
                    name, _taskName);
            }
            
        }
         
        static std::chrono::steady_clock::time_point DeltaToTimePoint(int secs) {
            return  std::chrono::steady_clock::now() + std::chrono::seconds(secs);
        }
 
        bool ObeyWaitUntil(std::chrono::steady_clock::time_point until,
                           thread::TMessHandler *action, 
                           const std::string &name,
                           const sds::Id &argIn = sds::Id::CreateNullItem(),
                           sds::IdPtr * const argOut = nullptr,
                           MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_OBEY, name, 0 /* Flags */, argIn, eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
 
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Obey of action {0} to task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Obey of action {0} to task {1}, returned but not complete",
                    name, _taskName);
            }
            return true;
            
        }
         
        void ObeyBulk(thread::TMessHandler *action, 
                      const std::string &name,
                      BulkData *argIn,
                      bool isSds,
                      unsigned notifyBytes=1024*1024,
                      sds::IdPtr * const argOut = nullptr,
                      MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
#if 0
            fprintf(stderr,"!!!OBeyBulk - sending action %s\n", name.c_str());
#endif
 
            SendBulk(action, DITS_MSG_OBEY, name, 0 /* Flags */, argIn, isSds,
                     notifyBytes, eventProcessor);
#if 0            
            fprintf(stderr,"!!!ObeyBulk sent, waiting for reply\n");
#endif
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask, 
                                       &eventInfo, eventProcessor, argOut);
#if 0
            fprintf(stderr,"!!!ObeyBulk sent, wait complete\n");
#endif
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "ObeyBulk of message {0} to task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "ObeyBulk of message {0} to task {1}, returned but not complete",
                    name, _taskName);
            }
            
        }
         
         
        void ObeyBulk(thread::TMessHandler *action, 
                      const std::string &name,
                      BulkDataSds *argIn,
                      unsigned notifyBytes=1024*1024,
                      sds::IdPtr * const argOut = nullptr,
                      MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
 
            ObeyBulk(action, name, argIn, true, notifyBytes, argOut, eventProcessor);
        }
 
 
 
        bool ObeyBulkWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::string &name,
                               BulkData *argIn,
                               bool isSds,
                               unsigned notifyBytes=1024*1024,
                               sds::IdPtr * const argOut = nullptr,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            SendBulk(action, DITS_MSG_OBEY, name, 0 /* Flags */, argIn, 
                     isSds, notifyBytes, eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
 
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Obey of action {0} to task {1} completed with error",
                    name, _taskName);
            }
            else*/ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Obey of action {0} to task {1}, returned but not complete",
                    name, _taskName);
            }
            return true;
            
        }
         
 
        bool ObeyBulkWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::string &name,
                               BulkDataSds *argIn,
                               unsigned notifyBytes=1024*1024,
                               sds::IdPtr * const argOut = nullptr,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
 
            return ObeyBulkWaitUntil(until, action, name, argIn, true, 
                                     notifyBytes, argOut, eventProcessor);
 
        }
        void Kick(thread::TMessHandler *action, 
                  const std::string &name,
                  const sds::Id &argIn = sds::Id::CreateNullItem(),
                  sds::IdPtr * const argOut = nullptr,
                  MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_KICK, name, 0 /* Flags */, argIn, eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, 
                                       argOut);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1}, returned but not complete",
                    name, _taskName);
            }
            
        }
 
        bool KickWaitUntil(std::chrono::steady_clock::time_point until,
                           thread::TMessHandler *action, 
                           const std::string &name,
                           const sds::Id &argIn = sds::Id::CreateNullItem(),
                           sds::IdPtr * const argOut = nullptr,
                           MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_KICK, name, 0 /* Flags */, argIn, eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
 
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1}, returned but not complete",
                    name, _taskName);
            }
            return true;
            
        }
         
        void KickBulk(thread::TMessHandler *action, 
                      const std::string &name,
                      BulkData *argIn,
                      bool isSds,
                      unsigned notifyBytes=1024*1024,
                      sds::IdPtr * const argOut = nullptr,
                      MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            SendBulk(action, DITS_MSG_KICK, name, 0 /* Flags */, argIn, isSds,
                     notifyBytes, eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, 
                                       argOut);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1}, returned but not complete",
                    name, _taskName);
            }
            
        }
 
        void KickBulk(thread::TMessHandler *action, 
                      const std::string &name,
                      BulkDataSds *argIn,
                      unsigned notifyBytes=1024*1024,
                      sds::IdPtr * const argOut = nullptr,
                      MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            KickBulk(action, name, argIn, true, notifyBytes, argOut, eventProcessor);
        }
 
 
 
        bool KickBulkWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::string &name,
                               BulkData *argIn,
                               bool isSds,
                               unsigned notifyBytes=1024*1024,
                               sds::IdPtr * const argOut = nullptr,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            SendBulk(action, DITS_MSG_KICK, name, 0 /* Flags */, argIn, isSds,
                     notifyBytes, eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
 
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Kick of action {0} in task {1}, returned but not complete",
                    name, _taskName);
            }
            return true;
            
        }
         
 
        bool KickBulkWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::string &name,
                               BulkDataSds *argIn,
                               unsigned notifyBytes=1024*1024,
                               sds::IdPtr * const argOut = nullptr,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
            return KickBulkWaitUntil(until, action, name, argIn, 
                                     true, notifyBytes, argOut, eventProcessor);
        }
 
 
        void SetParam(thread::TMessHandler *action, 
                  const std::string &name,
                  const sds::Id &argIn,
                  MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_SETPARAM, name, 0 /* Flags */, argIn, eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Set of parameter {0} in task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Set of parameter {0} in task {1}, returned but not complete",
                    name, _taskName);
            }
            
        }
 
        bool SetParamWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::string &name,
                               const sds::Id &argIn,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_SETPARAM, name, 0 /* Flags */, argIn, eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until);
 
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Set of parameter {0} in task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Set of parameter {0} in task {1}, returned but did not complete",
                    name, _taskName);
            }
            return true;
            
        }
 
 
        void GetParam(thread::TMessHandler *action, 
                      const std::string &name,
                      sds::IdPtr * const argOut,
                      MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
            Send(action, DITS_MSG_GETPARAM, name, 0 /* Flags */, 
                 sds::Id::CreateNullItem() /* arg */,
                eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, 
                                       argOut);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameter {0} from task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameter {0} from task {1}, returned but not complete",
                    name, _taskName);
            }
            
        }
        void GetParam(thread::TMessHandler *action, 
                      const std::initializer_list<std::string> names,
                      sds::IdPtr * const argOut,
                      MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
            sds::Id argIn = sds::Id::CreateArgCmdStruct(names);
 
            Send(action, DITS_MSG_MGETPARAM, "", 0 /* Flags */, argIn, 
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, argOut);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameters from task {0} completed with error",
                    _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameters from task {0}, returned but not complete",
                    _taskName);
            }
            
        }
 
 
 
 
        bool GetParamWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::string &name,
                               sds::IdPtr * const argOut,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
            Send(action, DITS_MSG_GETPARAM, name, 0 /* Flags */, 
                 sds::Id::CreateNullItem() /* arg */,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameter {0} from task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameter {0} from task {1}, returned but not complete",
                    name, _taskName);
            }
            return true;
            
        }
        bool GetParamWaitUntil(std::chrono::steady_clock::time_point until,
                               thread::TMessHandler *action, 
                               const std::initializer_list<std::string> names,
                               sds::IdPtr * const argOut,
                               MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
            sds::Id argIn = sds::Id::CreateArgCmdStruct(names);
 
            Send(action, DITS_MSG_MGETPARAM, "", 0 /* Flags */, argIn,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
            if (!ok) return false;
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameters from task {0} completed with error",
                    _taskName);
            }
            else */  
            if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Get of parameters from task {0}, returned but not complete",
                    _taskName);
            }
            return true;
            
        }
 
 
        void Control(thread::TMessHandler *action, 
                     const std::string &name,
                     const sds::Id &argIn = sds::Id::CreateNullItem(),
                     sds::IdPtr * const argOut = nullptr,
                     MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_CONTROL, name, 0 /* Flags */, argIn,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, 
                                       argOut);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Control message {0} to task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Control message {0} to task {1}, returned but did not complete",
                    name, _taskName);
            }
            
        }
 
        bool ControlWaitUntil(std::chrono::steady_clock::time_point until,
                              thread::TMessHandler *action, 
                              const std::string &name,
                              const sds::Id &argIn = sds::Id::CreateNullItem(),
                              sds::IdPtr * const argOut = nullptr,
                              MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            Send(action, DITS_MSG_CONTROL, name, 0 /* Flags */, argIn,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            bool ok = action->WaitForTransactionUntil(
                _theTask,
                &eventInfo, eventProcessor, 
                until, argOut);
 
            if (!ok) return false;
 
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Control message {0} to task {1} completed with error",
                    name, _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Control message {0} to task {1}, returned but not complete",
                    name, _taskName);
            }
            return true;
            
        }
         
 
        virtual void GetPathImmed();
 
 
 
        virtual void Died() {
            _path = 0;
            _state = PathState::DIED;
        }
        virtual void LosePath();
        virtual void Delete(int force);
 
 
 
 
        void MonitorStart(thread::TMessHandler *action, 
                          const sds::Id &argIn,
                          MessageEventHandler * const eventProcessor) {
 
            //fprintf(stderr,"MonitorStart - about to send monitor\n");
            Send(action, DITS_MSG_MONITOR, "START", DITS_M_SENDCUR, argIn,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            //fprintf(stderr,"MonitorStart - about to wait\n");
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, nullptr);
            //fprintf(stderr,"MonitorStart - wait complete.\n");
/* This should not be needed, event processor will throw if needed */
/*         if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Monitor Start message to task {0} completed with error",
                    _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Monitor START to task {0}, returned but did not complete",
                    _taskName);
            }
            
        }
        void MonitorForward(thread::TMessHandler *action, 
                            const sds::Id &argIn,
                            MessageEventHandler * const eventProcessor = &_simpleMonEventProcessor) {
 
            Send(action, DITS_MSG_MONITOR, "FORWARD", DITS_M_SENDCUR, argIn,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, nullptr);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Monitor FORWARD message to task {0} completed with error",
                    _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Monitor FORWARD to task {0}, returned but not complete",
                    _taskName);
            }
            
        }
 
 
 
        void MonitorForward(thread::TMessHandler *taction, 
                            const std::string &task,
                            const std::string action,
                            std::initializer_list<std::string> pars,
                            MessageEventHandler * const eventProcessor = &_simpleMonEventProcessor) {
 
            drama::sds::Id args = drama::sds::Id::CreateArgStruct();
            args.Put("TASK", task);
            args.Put("ACTION", action);
            args.AddToArgCmdStruct(pars, 3);
 
            MonitorForward(taction, args, eventProcessor);
            
        }
        template <typename ContainerType>
        void MonitorForward(thread::TMessHandler *taction, 
                            const std::string &task,
                            const std::string action,
                            const ContainerType &pars,
                            MessageEventHandler * const eventProcessor = &_simpleMonEventProcessor) {
 
            static_assert(
                std::is_convertible<typename ContainerType::value_type,std::string>::value,
                "The container value type must be convertible to a std::string");
 
            drama::sds::Id args = drama::sds::Id::CreateArgStruct();
            args.Put("TASK", task);
            args.Put("ACTION", action);
            args.AddToArgCmdStruct(pars, 3);
 
            MonitorForward(taction, args, eventProcessor);
            
        }
 
 
        void MonitorCancel(thread::TMessHandler *action, 
                           int monId,
                           MessageEventHandler * const eventProcessor = &_simpleEventProcessor) {
 
            sds::Id argIn = sds::Id::CreateArgCmdStruct(std::vector<INT32>{monId});
            Send(action, DITS_MSG_MONITOR, "CANCEL", DITS_M_SENDCUR, argIn,
                 eventProcessor);
            
            TransEvtInfo eventInfo;
            action->WaitForTransaction(_theTask,
                                       &eventInfo, eventProcessor, nullptr);
/* This should not be needed, event processor will throw if needed */
/*            if (eventInfo.entryStatus != STATUS__OK)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Monitor CANCEL message to task {0} completed with error",
                    _taskName);
            }
            else */ if (eventInfo.complete == 0)
            {
                DramaTHROW_F(
                    eventInfo.entryStatus,
                    "Monitor CANCEL to task {0}, returned but not complete",
                    _taskName);
            }
            
        }
 
                  
        virtual DitsPathType GetDitsPath(bool nothrow=false);
 
        static DVOID HandleDisconnected(const char *task,
                                        DitsPathType path,
                                        StatusType *status);
 
        static drama::sds::Id SpawnKickArg(DitsTransIdType tid);
 
        static void SpawnKickArgUpdate(drama::sds::Id *arg, DitsTransIdType tid);
 
 
    }; /* class Path */
 
 
} // namespace drama
 
#endif