thread.hh

Go to the documentation of this file.

#ifndef _DRAMA2_THREAD_INC
#define _DRAMA2_THREAD_INC
 
/*
 * History:
     04-Mar-2014 - TJF - Original version
 
 * The above ID is for Doxygen, this one has the format ACMM is looking for.
 * "@(#) $Id$"
 */
 
#include "drama/messagehandler.hh"
#include <thread>
#include <queue>
#include <deque>
#include <map>
#include <chrono>
#include <future>
#include <condition_variable>
#include <csignal>
namespace drama { 
 
    class Path;  /* drama::Path */
 
    namespace thread {
 
         /* @internal
         * Information needed by threads waiting on conditions.
         *
         *  Note - for kick events, the arg will stay around until
         *   the next event on this thread.  This is because we can have
         *   multiple threads waiting for a kick event and we must ensure
         *   any argument is valid for all.  We use a shared_ptr type to
         *   manage this.
         */
        struct WaitEventData {
            sds::IdPtr arg;
            TransEvtInfo eventInfo;
        };
 
        /* @internal
         * A queue of WaitEventData.
         */
        typedef std::queue<WaitEventData> WaitEventDataQueue;
 
 
        /* @internal
         * The type of a pointer to a condition variable
         */
        typedef std::shared_ptr<std::condition_variable_any> 
            WaitEventCondPtrType;
 
        /* @internal
         * Information on events we are expecting related to a particular
         * transaction.
         */
        struct WaitEventDetails {
            std::thread::id      _thread;    // Thread which is waiting
            DitsTransIdType      _transId;   // Transaction we are waiting on, 0 for kick.
            WaitEventDataQueue   _dataQueue; // Queue of transaction data.
            WaitEventCondPtrType _condition; // Used to notify the thread
                                             // of events. 
            bool                 _waiting;   // Are we waiting.
 
            Path                *_pathObj;   // Path message is along. IF not kick
            DitsPathType         _ditsPath;  // Underlying DITS path.
            DitsMsgType          _type;      // Message type, if not kick.
            std::string          _messName;  // Message Name, if not kick
            
 
            explicit WaitEventDetails(
                DitsTransIdType transid,
                Path *thePath);
            bool DataAvail() const {
                return !_dataQueue.empty();
            }
            ~WaitEventDetails() {
#if 0
                fprintf(stderr,
                        "WaitEvenDetails destructor - transid %p, this %p\n",
                        static_cast<void *>(_transId),
                        static_cast<void *>(this));
#endif
            }
            
        };
 
        /*
         * A map of wait event details, indexed by thread id.
         */
        typedef std::map<std::thread::id, WaitEventDetails> WaitEventMapType;
        typedef WaitEventMapType::iterator WaitEventMapIter;
        typedef WaitEventMapType::const_iterator WaitEventMapIterConst;
 
        class TMessHandler;
 
 
        /*
         * Return the number of events outstanding across all threads.
         */
        unsigned WaitEventMapGetEvents(std::weak_ptr<Task> theTask, 
                                       const WaitEventMapType &waitEventMap);
        class ProcessInfo {
        private:
            std::shared_ptr<TMessHandler> _handler;
            std::shared_ptr<Path> _pathObj;
            const DitsMsgType   _type;      // Message type, if not kick.
            const std::string   _messName;  // Message Name, if not kick
        public:
            ProcessInfo(std::shared_ptr<TMessHandler> handler, 
                        const WaitEventDetails &details) :
                _handler(handler), 
                _pathObj(details._pathObj, drama::nodel()),
                _type(details._type), 
                _messName(details._messName) {
                
            }
            const TMessHandler &GetHandler() const {
                return *_handler;
            }
            const Path &GetPath() const {
                return *_pathObj;
            }
            DitsMsgType GetMessType() const {
                return _type;
            }
            const char * GetMessTypeCStr() const {
                return DitsGetMsgTypeStr(_type);
            }
            const std::string &GetMessName() const {
                return _messName;
            }
            
        }; // class ProcessInfo.
        
        /*
         * Abstract class for objects used to process transaction events.
         */
        class TransEvtProcessor {
        public:
            virtual bool Process(ProcessInfo          messInfo,
                                 const TransEvtInfo & eventInfo,
                                 const sds::IdPtr &arg) = 0;
 
            virtual void NewTransaction(DitsMsgType msgType, DitsTransIdType tid) = 0;
            virtual bool WaitTimeout(std::chrono::steady_clock::time_point *until);
 
            virtual ~TransEvtProcessor() {}
        };
 
        class TSdsListToUserObj : public sds::PrintObjectCR {
        private:
            TMessHandler *_handler;
        public:
            TSdsListToUserObj(TMessHandler *messageHandler);
            void Print(const std::string &line) const override;
        };
        class TMessHandler  {
            TSdsListToUserObj _sdsListToUserObj = this;
        public:
 
            TMessHandler& operator=(const TMessHandler &rhs) = delete;
 
            TMessHandler(const TMessHandler &source) = delete;
 
            TMessHandler() {}
 
 
            virtual Task::mutexType & Lock() const = 0;
            virtual void SetupWaitEvent(DitsTransIdType tid,
                                        drama::Path *pathObj ) = 0;
            virtual void MessageUser(const std::string &text) const = 0;
 
            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) const {
                            /*
                 * 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) const {
                MessageUser(fmt::vformat(fmt.get(), fmt::make_format_args(args...)));
            }
          
            virtual void WaitForTransaction(
                std::weak_ptr<Task> theTask,
                TransEvtInfo * const eventInfo,
                TransEvtProcessor * const eventProcessor,
                sds::IdPtr * const arg = nullptr) ;
 
 
            virtual bool WaitForTransactionUntil(
                std::weak_ptr<Task> theTask,
                TransEvtInfo * const eventInfo,
                TransEvtProcessor * const eventProcessor,
                const std::chrono::steady_clock::time_point &until,
                sds::IdPtr * const arg = nullptr);
 
 
 
            virtual Dits___CurActType GetMessageContext() const = 0;
 
            virtual std::shared_ptr<Task>  GetTask() const = 0;
            virtual const sds::PrintObjectCR &SdsListToUser() const;
 
            virtual ~TMessHandler() {}
 
 
        private:
            /* Features used by WaitForTransaction()
             */
 
            virtual void ClearWait(bool complete) = 0;
            /*
             * Return a pointer to the event details for the specified thread.
             */
            virtual WaitEventDetails *FindWaitEventDetails(std::thread::id) = 0;
 
            /*
             *
             * Initiate waiting for a transaction and return a pointer
             * to the item the wait event details.  Presumption is that
             * SetupWaitEvent() has been invoked.
             */
             WaitEventDetails * WaitForTransactionStart();
            /*
             * The wait for transaction operation has finished - tidy up.
             *
             * @param Event       details returned by WaitForTransactionStart()
             * @param evenInfo    Details on the event which caused the
             *                      wait to finish are returned here.
             * @param  arg        The output argument from the transaction, if any.  
             *                     An address of and sds::IdPtr is supplied.
             *                     If you supply a  null pointer, it is 
             *                      ignored.  Otherwise, any argument to the 
             *                      transaction event is copied into here. 
             *                     If there is no argument to the 
             *                      transaction it will refer to a null 
             *                      SDS item.  
             */
            void WaitForTransactionFinish(
                const WaitEventData &details,
                TransEvtInfo * const eventInfo,
                sds::IdPtr * const arg);
            
            
        }; // class TMessHandler
 
        class AccessDrama {
            Dits___CurActType _entryDetails;
            Task::guardType _lock;
        public:
            AccessDrama(const TMessHandler &messHandler);
            AccessDrama(Task::mutexType &taskLock);
            ~AccessDrama();
        };
 
        class SignalBlocker {
            sigset_t oldSet;
        public:
            SignalBlocker();
            ~SignalBlocker();
        };
        
        /*
         * Block all signals to the current thread.
         */
        extern void BlockSignals();
            
        
        /* @internal
         * 
         * Puts an entry in waitEventMap for this thread and the specified
         *  transaction id. If the transaction id is 0, then it is presumed
         *  the entry is for a kick.
         *
         * Thread/Lock Entry Info:
         *   May/may not be running in DRAMA message thread.  (Probably not)
         *   DRAMA Lock Taken.
         * 
         */
        void SetupWaitEvent(DitsTransIdType id, 
                            drama::Path *pathObj,
                            TMessHandler *h,  
                            WaitEventMapType *waitEventMap);
 
 
 
        void RunProgram(TMessHandler *messHandler,
                        const std::string &fileName,
                        const std::string &arguments="",
                        const std::string &node="",
                        int priority = 0,
                        bool absPriority = false,
                        bool setNames = true,
                        unsigned stackBytes = 0);
                        
        bool RunProgramWaitUntil(std::chrono::steady_clock::time_point until,
                                 TMessHandler *messHandler,
                                 const std::string &fileName,
                                 const std::string &arguments="",
                                 const std::string &node="",
                                 int priority = 0,
                                 bool absPriority = false,
                                 bool setNames = true,
                                 unsigned stackBytes = 0);
                        
 
                        
 
    } // namespace thread
} // namespace drama
 
#endif