threadaction.hh

Go to the documentation of this file.

#ifndef _DRAMA2_THREADACTION_INC
#define _DRAMA2_THREADACTION_INC
/*
 * History:
     07-Jan-2014 - TJF - Original version
     30-Sep-2016 - TJF - Start of history after development.
                         SignalWaitingThreads() returns number of threads
                          signaled.
 
 * The above ID is for Doxygen, this one has the format ACMM is looking for.
 * "@(#) $Id$"
 */
#include "drama/messagehandler.hh"
#include "drama/thread.hh"
#include <thread>
#include <queue>
#include <deque>
#include <map>
#include <chrono>
#include <future>
#include <condition_variable>
#include <functional>
namespace drama { 
 
    namespace thread {
 
        /*
         * This enum is used for indicating what type of event a signal
         * from the thread to DRAMA is about.
         */
        enum class ThreadSignalType { 
            Complete,     // Main thread has completed.
            WaitSignal,// The thread is waiting for a signal.
            WaitKick  // The thread is interested in kicks
        };
 
        /*
         * Details on a signal from a thread.  
         */
        struct DramaSignalDetails {
            std::thread::id  _fromThread;
            ThreadSignalType _sigType;
 
            DramaSignalDetails(ThreadSignalType sigType) :
                _fromThread(std::this_thread::get_id()),
                _sigType(sigType) {}
 
 
        };
        /*
         * A queue of signal details.  We have a queue of these as
         * we support multiple threads as part of implementing the
         * action, so we could have multiple signals outstanding.
         *
         * A std::queue would be a better representation of this
         * then "std::deque", but we do want to clear the queue and
         * std::queue does not support that.
         */
        typedef std::deque<DramaSignalDetails> SignalDetailsQueue;
 
 
 
        class TAction;
 
        /* Used internally by TAction, this class is used as the
         * reschedule handler for DRAMA on Obey reschedule messages to
         * threaded actions.
         *
         *   TAction::_obeyRescheduleObj member is of this type.
         *
         */
        class _ThreadMessage : public MessageHandler {
            TAction *_threadHandlerObj;
        public:
            _ThreadMessage(TAction *threadHandler) : 
                _threadHandlerObj(threadHandler) {}
        
        private:
            /*
             * Made private, parent class has it as protected. 
             * But I don't want users to invoke it 
             */
            Request MessageReceived() override final;
        };
        /* Used internally by TAction,  this class is used as the
         * reschedule handler for DRAMA on Kick messages to
         * threaded actions.
         *
         *   TAction::_kickRescheduleObj member is of this type.
         */
        class _ThreadKick : public MessageHandler {
            TAction *_threadHandlerObj;
        public:
            _ThreadKick(TAction *threadHandler) : 
                _threadHandlerObj(threadHandler) {}
        
        private:
            /*
             * Made private, parent class has it as protected. 
             * But I don't want users to invoke it 
             */
            Request MessageReceived() override final;
        };
 
 
        using ThreadActionFunction = std::function<void (TAction *, const sds::Id &)> ;
 
        // TMessHandler and RunDramaExitNotifier are abstract only.
        class TAction : public MessageHandler, 
                        public TMessHandler, public RunDramaExitNotifier  {
 
            // We need a function to implement the thread that
            // is not part of TAction, but which can call
            //   ActionThread.  This must be in the drama namespace,
            //   since that is how we are declaring it now.
 
            //friend void MyThread(TAction *handler, const sds::Id &obeyArg);
 
            friend class  _ThreadMessage;
            friend class  _ThreadKick;
 
        private:
            std::future<void> _actionFuture; // Future of first thread in action.
 
            /*
             * Pointer to the DRAMA task we are part of.
             * The pointer is const - we never change it after the constructor.
             *  The task pointed to is NOT const - since we can operate on it.
             */
            std::weak_ptr<Task> _theTask;
            double      _timeout;           // Thread wait timeout, seconds.
 
 
            void RunActionThread(const sds::IdPtr obeyArg);
 
 
        public:
 
            TAction(std::weak_ptr<Task> dramaTask, double timeout=0) : 
                _theTask(dramaTask), _timeout(timeout),
                _obeyRescheduleObj(this),_kickMessageObj(this)  {
 
  
            }
 
            virtual ~TAction();
 
 
            TAction& operator=(const TAction &rhs) = delete;
            TAction(const TAction &source) = delete;
 
        private:
            /* 
             * These are make private, as they are not meant to be 
             * called outside the object, just implemented outside
             */
 
            /*
             * The method invoked to handle the obey message.   This
             * is an implementation of MessageHandler::MessageReceived().
             */
            virtual Request MessageReceived() final;
 
 
            virtual void ActionThread(const drama::sds::Id &obeyArg) = 0;
 
 
            virtual int RunDramaHasExited() override;
            virtual bool JoinThreads(std::chrono::steady_clock::time_point until) override;
 
 
            virtual void KickedWhenNotWaiting(sds::IdPtr Arg);
 
            virtual void SignaledWhenNotWaiting(sds::IdPtr Arg);
 
 
        public:
            std::shared_ptr<Task>  GetTask() const override {
                return std::shared_ptr<Task>(_theTask);
            }
            bool HasTaskExpired() const noexcept {
                return _theTask.expired();
            }
 
 
            void MessageUser(const std::string &text) const override;
 
 
            template<typename... Types>
#ifndef RUNNING_DOXYGEN
            D2_FMT_DEPRECATED("Replace MessageUser() by MessageUserF().  See MessageUserF() page for details.")
#endif
            void MessageUser(const char *format, Types... args) {
                
                /*
                 * Our approach is to write the output to a string, via
                 * SafePrintf(), then output that in one operation.  Since
                 * std::ostream devices will then do the output in one operation.
                 */
                std::stringstream sstrm;
                drama::SafePrintf(sstrm, format, args...);
                MessageUser(sstrm.str());
 
 
            }
            template<typename... Args>
            void MessageUserF(const fmt::format_string<Args...> fmt, Args&&... args) {
                
                MessageUser(fmt::vformat(fmt.get(), fmt::make_format_args(args...)));
                
            }
 
            /* SdsListToUser() is in both TMessHandler and MessageHandler, 
               two of our base classes.  So we must sort out the 
               confusion here
            */
            virtual const sds::PrintObjectCR &SdsListToUser() const override {
                return TMessHandler::SdsListToUser();
            }
 
 
            void SendTrigger(const sds::Id &arg) const;
 
            double GetTimeout() const  {
                return _timeout;
            }
            void SetTimeout(double newTimeout)  {
                _timeout = newTimeout;
            }
 
            void WaitForEvent(EntryCode *event, sds::IdPtr * const arg=nullptr) {
                /*
                 * Signal DRAMA that we are waiting for an event.  
                 *  This will put an entry in _waitEventMap for this
                 *  thread.
                 */
                SignalAmWaitingForKick();
 
                /*
                 * Access the entry in _waitEventMap.
                 *
                 *  Note - this first call will throw if the weak pointer _theTask 
                 *           does not have a valid object.  
                 *          
                 */
                Task::uniqueLockType DramaLock(
                    std::shared_ptr<Task>(_theTask)->Lock());
 
                std::thread::id  threadId(std::this_thread::get_id());
                auto mapItem = _waitEventMap.find(threadId);
                assert (mapItem != _waitEventMap.end());
                WaitEventDetails *waitEvent =  &(mapItem->second);
                
 
                /*
                 * Wait for the condition 
                 *
                 * We use a Lambda function to access the DataAvail() method
                 * in waitEvent.
                 *
                 */
                waitEvent->_condition->wait(
                    DramaLock, 
                    [waitEvent]{return waitEvent->DataAvail();});
 
 
                WaitEventData details = waitEvent->_dataQueue.front();
 
                *event = details.eventInfo.entryReason;
                /*
                 * If the user wants the argument, return it.
                 */
                if (arg)
                {
                    *arg = details.arg;
                }
                std::shared_ptr<Task>(_theTask)->Logger().LogNF(D2LOG_DRAMA2, 
                                              "d2::TACT::WaitForEvent",
                                              "Event received");
 
 
                waitEvent->_dataQueue.pop();
                // Should we complain here (throw execption?) if queue not empty?
                ClearWait(false);
            }
            template <class Rep, class Period>
            bool WaitForEventTimeoutIn(
                EntryCode *event,
                const std::chrono::duration<Rep,Period>& rel_time,
                sds::IdPtr  * const arg=0) {
                /*
                 * Signal DRAMA that we are waiting for a kick.  
                 *  This will put an entry in _waitEventMap for this
                 *  thread.
                 */
                SignalAmWaitingForKick();
                /*
                 * Access the entry in _waitEventMap.
                 */
                Task::uniqueLockType DramaLock(
                    std::shared_ptr<Task>(_theTask)->Lock());
                std::thread::id  threadId(std::this_thread::get_id());
                auto mapItem = _waitEventMap.find(threadId);
                assert (mapItem != _waitEventMap.end());
                WaitEventDetails *waitEvent =  &(mapItem->second);
                /*
                 * Wait for the condition 
                 *
                 * We use a Lambda function to access the DataAvail() method
                 * in waitEvent.
                 *
                 */
                waitEvent->_condition->wait_for(
                    DramaLock, rel_time, 
                    [waitEvent]{return waitEvent->DataAvail();});
                if (!waitEvent->DataAvail())
                {
                    std::shared_ptr<Task>(_theTask)->Logger().LogNF(
                        D2LOG_DRAMA2, 
                        "d2::TACT::WaitEventTOIn",
                        "Timeout waiting for signal/kick");
 
 
                    // Timeout.
                    //fprintf(stderr,"TAction::WaitKickFor() - timeout\n");
                    ClearWait(false);
                    *event = EntryCode::DramaAbortWaits;
                    return false;
                }
                //fprintf(stderr,"TAction::WaitEventFor() - event/signal\n");
 
                WaitEventData details = waitEvent->_dataQueue.front();
                *event = details.eventInfo.entryReason;
                /*
                 * If the user wants the argument, return it.
                 */
                if (arg)
                {
                    *arg = details.arg;
                }
                std::shared_ptr<Task>(_theTask)->Logger().LogNF(D2LOG_DRAMA2, 
                                              "d2::TACT::WaitEventTOIn",
                                              "kick/signal received");
 
 
                waitEvent->_dataQueue.pop();
                // Should complain here (throw execption?) if queue not empty
                ClearWait(false);
 
                return true;
 
            }
            bool WaitForEventTimeoutIn(
                EntryCode *event,
                unsigned seconds,
                sds::IdPtr  * const arg=0) {
 
                return WaitForEventTimeoutIn(event, std::chrono::seconds(seconds), arg);
 
            }
 
 
 
 
 
            template <class Clock, class Duration>
            bool WaitEventTimeoutAt(
                EntryCode *event,
                const std::chrono::time_point<Clock,Duration>& abs_time,
                sds::IdPtr  * const arg=0 ) {
 
                /*
                 * Signal DRAMA that we are waiting for a kick.  
                 *  This will put an entry in _waitEventMap for this
                 *  thread.
                 */
                SignalAmWaitingForKick();
                /*
                 * Access the entry in _waitEventMap.
                 */
                Task::uniqueLockType DramaLock(
                    std::shared_ptr<Task>(_theTask)->Lock());
                std::thread::id  threadId(std::this_thread::get_id());
                auto mapItem = _waitEventMap.find(threadId);
                assert (mapItem != _waitEventMap.end());
                WaitEventDetails *waitEvent =  &(mapItem->second);
                /*
                 * Wait for the condition 
                 *
                 * We use a Lambda function to access the DataAvail() method
                 * in waitEvent.
                 */
                waitEvent->_condition->wait_until(
                    DramaLock, abs_time, 
                    [waitEvent]{return waitEvent->DataAvail();});
                if (!waitEvent->DataAvail())
                {
                    // Timeout.
                    std::shared_ptr<Task>(_theTask)->Logger().LogNF(D2LOG_DRAMA2, 
                                                  "d2::TACT::WaitEventTOAT",
                                                  "Timeout waiting for signal/kick");
 
                    ClearWait(false);
                    *event = EntryCode::DramaAbortWaits;
                    return false;
                }
                //fprintf(stderr,"TAction::WaitKickFor() - kicked\n");
                WaitEventData details = waitEvent->_dataQueue.front();
                *event = details.eventInfo.entryReason;
                /*
                 * If the user wants the argument, return it.
                 */
                if (arg)
                {
                    *arg = details.arg;
                }
                waitEvent->_dataQueue.pop();
                std::shared_ptr<Task>(_theTask)->Logger().LogNF(
                    D2LOG_DRAMA2, 
                    "d2::TACT::WaitEventTOAt",
                    "Signal/Kick received");
 
 
 
                // Should complain here (throw execption?) if queue not empty
                ClearWait(false);
 
                return true;
 
            }
            void WaitForKick(sds::IdPtr * const arg=nullptr) {
 
                // Can be implemented by WaitForEvent(), with a check on the event.
                EntryCode event;
                WaitForEvent(&event, arg);
                if (event != EntryCode::Kick)
                {
                    DramaTHROW(DRAMA2__SIGKICK,
                               "TAction::WaitForKick - signal received instead of kick, application programming error. You may want WaitForEvent()");
                }
            }
            template <class Rep, class Period>
            bool WaitForKickTimeoutIn(
                const std::chrono::duration<Rep,Period>& rel_time,
                sds::IdPtr  * const arg=0) {
 
                // Can be implemented by WaitForEventTimeoutIn(), with a check on the event.
                EntryCode event;
                bool kicked = WaitForEventTimeoutIn(&event, rel_time, arg);
                if (!kicked)
                    return false;
                if (event != EntryCode::Kick)
                {
                    DramaTHROW(DRAMA2__SIGKICK,
                               "TAction::WaitForKickTimeoutIn - signal received instead of kick, application programming error. You may want WaitForEventTimeoutIn()");
                }
                return true;
 
 
            }
 
            bool WaitForKickTimeoutIn(
                unsigned seconds,
                sds::IdPtr  * const arg=0) {
 
                return WaitForKickTimeoutIn(std::chrono::seconds(seconds), arg);
 
            }
 
            template <class Clock, class Duration>
            bool WaitKickForTimeoutAt(
                const std::chrono::time_point<Clock,Duration>& abs_time,
                sds::IdPtr  * const arg=0 ) {
 
 
                // Can be implemented by WaitForEventTimeoutAt(), with a check on the event.
                EntryCode event;
                bool kicked = WaitForEventTimeoutAt(&event, abs_time, arg);
                if (!kicked)
                    return false;
                if (event != EntryCode::Kick)
                {
                    DramaTHROW(DRAMA2__SIGKICK,
                               "TAction::WaitForKickTimeoutAt - signal received instead of kick, application programming error. You may want WaitForEventTimeoutAt()");
                }
                return true;
 
 
 
            }
 
            void SetupWaitEvent(DitsTransIdType tid, drama::Path *pathObj) override;
 
            void SetReturnArg(const sds::Id &arg, bool copy=true) {
 
                if (copy)
                {
                    _outArg = arg.Copy();
                    _outArgDelete = true;
                }
                else
                {
                    _outArg.ShallowCopy(arg);
                    _outArgDelete = false;
                }
 
                _outArgSet = true;
            }
            void SetExitOnCompletion() {
                _exitTask = true;
            }
            void ClearExitOnCompletion() {
                _exitTask = false;
            }
            void SetReturnArg(sds::Id *arg) {
 
                _outArg.ShallowCopy(arg, true);
                _outArgDelete = true;
                _outArgSet = true;
 
            }
 
            void PutObeyHandler(MessageHandlerPtr obj) override final;
 
            void PutKickHandler(MessageHandlerPtr obj) override final;
 
            
            Task::mutexType & Lock() const override;
        
            Dits___CurActType GetMessageContext() const override;
 
 
 
 
 
            WaitEventDetails *SetupWaitForKick() {
 
                /*
                 * Signal DRAMA that we are waiting for a kick.  
                 *  This will put an entry in _waitEventMap for this
                 *  thread.
                 */
                SignalAmWaitingForKick();
 
                /*
                 * Access the entry in _waitEventMap.
                 */
                Task::uniqueLockType DramaLock(
                    std::shared_ptr<Task>(_theTask)->Lock());
                std::thread::id  threadId(std::this_thread::get_id());
                auto mapItem = _waitEventMap.find(threadId);
                assert (mapItem != _waitEventMap.end());
                return  &(mapItem->second);
                
 
            }
            bool DoWaitForKick(WaitEventDetails *waitEvent,
                               sds::IdPtr * const arg=nullptr) {
 
                assert(waitEvent);
                /*
                 * Wait for the condition 
                 *
                 * We use a Lambda function to access the DataAvail() method
                 * in waitEvent.
                 *
                 */
                Task::uniqueLockType DramaLock(
                    std::shared_ptr<Task>(_theTask)->Lock());
                waitEvent->_condition->wait(
                    DramaLock, 
                    [waitEvent]{return waitEvent->DataAvail();});
 
 
                WaitEventData details = waitEvent->_dataQueue.front();
                /*
                 * If the user wants the argument, return it.
                 */
                if (arg)
                {
                    *arg = details.arg;
                }
                std::shared_ptr<Task>(_theTask)->Logger().LogNF(
                    D2LOG_DRAMA2, 
                    "d2::TACT::DoWaitKick",
                    "Kick received");
 
 
                waitEvent->_dataQueue.pop();
                // Should we complain here (throw execption?) if queue not empty?
                ClearWait(false);
 
                if (details.eventInfo.entryReason == 
                    EntryCode::DramaAbortWaits)
                    return false;
                return true;
            }
 
        private:
            /*
             * Objects used as part of handling obey reschedule and 
             * kick messages.
             *
             * They simply result in the ObeyReschedule() and KickMessage()
             * methods below being invoked.
             */
            _ThreadMessage _obeyRescheduleObj;
            _ThreadKick     _kickMessageObj;
 
            /*
             * The details on the DITS action are recorded here before
             * we start the thread.
             */
            int _actionPtr;
            Dits___CurActType _actionDetails;
 
            std::string _actionName;
 
 
            /*
             * Count of kick messages received.
             */
            unsigned _numKicks;
            /*
             * Argument to the last kick, if any.
             */
            //sds::IdPtr _lastKickArg;
 
            /*
             * Argument to the obey, if any.
             */
            sds::IdPtr _obeyArg;
 
            /*
             * Will task exit on action completion.
             */
            bool _exitTask = false;
            /*
             * Output argument details.
             */
            sds::Id _outArg;            // Actual argument.
            bool _outArgDelete = false; // Should it be deleted by DRAMA?
            bool _outArgSet = false;    // Has it been set.
 
            /*
             * A double ended queue of information about signals sent from
             * the threads to DRAMA. 
             *
             * We add signals to the front, remove them from the back.
             */
            SignalDetailsQueue _signalQueue;
 
            /*
             * A map which contains all the events we are waiting on.
             *  (one entry in the map per subsidiary thread).
             */
            WaitEventMapType _waitEventMap;
 
 
            /*
             * Method invoked to handle obey reschedule events.
             */
            Request ObeyReschedule();
            /*
             * Method invoked to handle kick messages
             */
            Request KickMessage();
 
            /*
             * Method used to signal the action code from threads.
             */
            void SignalDrama(ThreadSignalType why);
 
            /*
             * Signal that we are waiting for a kick.
             */
            void SignalAmWaitingForKick();
            /* Process a DRAMA Signal event message.
             */
            Request ProcessSignal();
            /* Process a DRAMA 2 Signal event message (used for communications
             * between threads and the Main DRAMA code)
             */
            Request ProcessDrama2Signal();
            /*
             * Process a Subsidiary  action message.
             */
            Request ProcessSubsidiaryMessage();
 
            /*
             * Clear the thread' waiting flag in _waitEventMap.
             *
             * @param  complete If true, then it is known the transaction is
             *    actually complete.  Set false if know to be not
             *    complete or unknown.  Determines if we orphan
             *    the transaction.
             */
            void ClearWait(bool complete) override;
 
           /*
             * Return a pointer to the event details for the specified thread.
             */
             WaitEventDetails *FindWaitEventDetails(std::thread::id) 
                override;
 
 
            /*
             * Wake up threads waiting on messages.  Used to wake up waiting
             * action threads after a kick or signal is received.
             *
             * @return The count of threads found and signaled.
             */
            unsigned WakeUpWaitingThreads(sds::IdPtr arg);
 
            /*
             * Indicate and action thread is complete.  Called at the end
             * of the thread.
             */
            void ActionThreadComplete();
 
            /*
             * Invoked to signal any threads waiting for messages to 
             * cancel their wait events. Returns the number of signals
             * sent.
             */
            unsigned SignalWaitingThreads(EntryCode, StatusType);
 
            /*
             * Invoke when the action is ending - will make any outstanding
             * events into orphans.
             */
            void OrphanOutstandingEvents();
 
        public:
            void SignalWaitingEvent(WaitEventDetails *, EntryCode, StatusType);
            unsigned GetKickCount() const {
                return _numKicks;
            }
#if 0
            sds::IdPtr GetLastKickArg() const {
                return _lastKickArg;
            }
#endif
            void ResetKickCount();
 
            void AbortMessageWaits(StatusType status);
 
 
            std::string GetActionName() const {
                return _actionName;
            }
 
        }; // class TAction
 
 
        class TActionViaFunctor : public TAction {
        private:
            const ThreadActionFunction _func; // The function.
        public:
            TActionViaFunctor(std::weak_ptr<Task> dramaTask,
                              const ThreadActionFunction func) : 
                TAction(dramaTask, 0), _func(func) { }
 
            void ActionThread(const drama::sds::Id &obeyArg) override {
                _func(this, obeyArg);
            }
        };
 
 
        class KickNotifier : public RunDramaExitNotifier {
        private:
            bool _wasKicked = false;  // Has action been kicked.
            bool _waiting   = false;  // Is thread waiting for kick.
            bool _threadThrew = false; // Has the thread died by throwing?
            TAction *_theAction = nullptr; // Action we are part of.
 
 
            std::future<void> _threadFuture;  // Future used for WaitThread
            // The condition variable is used to ensure the constructor
            //  does not return until the child thread is running and
            //  has initialised itself, indicated by _waitEvent being
            //  valid (not a nullptr).
            std::condition_variable_any _threadReadyCond;
            WaitEventDetails *_waitEvent = nullptr;
 
            void WaitThread();  // Entry point for child thread.
        public:
            KickNotifier(TAction *action);
 
            virtual bool Kicked(const sds::Id & arg);
 
 
            bool WasKicked() { 
                Task::guardType DramaLock(std::shared_ptr<Task>(_theAction->GetTask())->Lock());
                return _wasKicked;
            }
 
            bool ThreadThrewException() const {
                return _threadThrew;
            }
 
            virtual ~KickNotifier();
 
            KickNotifier& operator=(const KickNotifier &rhs) = delete;
            KickNotifier(const KickNotifier &source) = delete;
 
            virtual int RunDramaHasExited() override;
 
            bool JoinThreads(std::chrono::steady_clock::time_point until) override;
 
 
        }; // class KickNotifier
        
    } // namespace thread
} // namespace drama
 
#endif