task.hh

Go to the documentation of this file.

#ifndef _DRAMA2_TASK_INC
#define _DRAMA2_TASK_INC
 
/*
 * History:
     07-Jan-2014 - TJF - Original version
     23-Mar-2018 - TJF - Add CheckLockTaken static method;
     18-Jan-2018 - TJF - Add SeparateThreadRunsDrama() method and properly support
                           this by maintaining IDs for both the thread that created
                           DRAMA and that which is running the message loop.
                         Replace GetTaskThreadId() by GetTaskCreThreadId() and
                         GetTaskRunThreadId() as part of this.
 
 * The above ID is for Doxygen, this one has the format ACMM is looking for.
 * "@(#) $Id$"
 */
 
#include "DitsTypes.h"
#include <string>
#include <DitsSys.h>
 
#include "drama/messagehandler.hh"
#include "drama/spawnable.hh"
#include "drama/logger.hh"
#include "drama/fmt/format.h"
 
#include <thread>
 
/* These two lines allows us to format thread::id values with fmt::format.  See https://fmt.dev/11.0/api/#stdostream-support */
#include "drama/fmt/ostream.h"
template <> struct fmt::formatter<std::thread::id> : ostream_formatter {};
 
#include <mutex>
#include <set>
 
#include <memory>   // For shared_ptr.
#include <iostream>
#include <deque>
#include <type_traits>
#include <functional>
 
/*
 * This file  uses DOXYGEN comments - the following is placed into the
 * index page.  note that the @file block says to include the typedefs,
 * functions etc from this file into the documentation.
 */
 
namespace drama { 
 
    class Task;
 
    namespace thread {
        /*
         * These declarations are needed to allow  
         *   AddTA(const std::string &name, thread::ThreadActionFunction func) to
         * be defined.
         *
         * See threadaction.hh for full details of these.
         */
        class TAction;  // Class used for threaded action implementations
        /* Function used to implementation actions */
        using ThreadActionFunction = std::function<void (TAction *, const sds::Id &)> ;
        /* Function from threadaction.cpp used to wrap up using a function
         * to implement a threaded action
         */
        void AddActionFunction(Task *, 
                               const std::string &name, 
                               thread::ThreadActionFunction func,
                               const std::string &descr);
 
    } // namespace thread;
 
    struct nodel
    {
        void operator()(void const *) const
            {
            }
    };
 
 
 
    struct TransEvtInfo {
        EntryCode      entryReason;   
        StatusType     entryStatus;   
        std::string    entryName;     
        std::string    loadErrText;   
        bool           complete;      
        DitsBulkInfoType bulkInfo;    
    };
 
    class OrphanDetails : public TransEvtInfo { 
        
    public:
        sds::IdPtr   argument;  
        DitsPathType path;      
        DitsTransIdType transId;
        OrphanDetails(TransEvtInfo evtInfo,
                      sds::IdPtr   arg,
                      DitsPathType thePath,
                      DitsTransIdType tid) :
            TransEvtInfo(evtInfo),
            argument(arg),
            path (thePath),
            transId(tid) {
        }
            
                      
    };
    /* @internal
     * A queue of OrphanDetails.
     *
     * 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<OrphanDetails> OrphanDetailsQueue;
 
 
 
 
    typedef std::shared_ptr<MessageHandler> MessageHandlerPtr;
 
#ifndef RUNNING_DOXYGEN
 
    /* 
     * If defined, we use our own sub-class of a std::recursive_mutex
     * such that we can implement checks around use of locks.  This
     * is the normal default and DRAMA is not well tested without 
     * it being defined
     */
#define DRAMA2_LOCK_DEBUG 
 
    /*
     * If defined, then we create a task specific file which records 
     * lock calls. This is rather complicated, but might be useful in
     * tracking deadlock situations.  The files written are named
     *  DramaLock-<pid>.log.  
     */
//#define DRAMA2_LOCK_DEBUG2
 
#endif // defined(RUNNING_DOXYGEN)
 
#ifdef DRAMA2_LOCK_DEBUG2
 
    /*
     * If DEBUG2 is enabled, we need DEBUG.
     */
#ifndef  DRAMA2_LOCK_DEBUG 
#define DRAMA2_LOCK_DEBUG
#endif
    /* 
     *  The DRAMA2_TASK_CPP macro is set by the task.cpp file.  We use
     * this to trigger a warning if DEBUG2 debugging is enabled.
     */
#ifdef DRAMA2_TASK_CPP
#warning "Lock debugging level 2 enabled - will slow things down"
#endif
 
    /* This macro allows to do stuff only when this logging is enabled*/
#define DRAMA2_LOCK_DEBUG2_Only(_A_) _A_
#else // !defined(DRAMA2_LOCK_DEBUG2)
#define DRAMA2_LOCK_DEBUG2_Only(_A_) /*_A_*/
 
#endif // !defined(DRAMA2_LOCK_DEBUG2)
 
#ifdef DRAMA2_LOCK_DEBUG
    /* Wrapper for a recursive timed mutex which allows us to
     * work out if it has been taken and if so, by who
     *
     */
    //class LockWithInfo : public std::recursive_timed_mutex {
    class LockWithInfo : public std::recursive_mutex {
    private:
        bool _taken;             /* Has the mutex been taken */
        unsigned _count;         /* Number of times a thread has taken it*/
        std::thread::id _takenBy;/* Thread which has it */
#ifdef DRAMA2_LOCK_DEBUG2
        static std::ofstream _debugFile;  /* Lock log file, process specific */
        static unsigned _lockCount;       /* How many locks in the process */
        char _code = '_';        /* This lock's prefix */
        /* Write time and thread details for the debug file to a stream */
        void TimeNowAndThreadToStream(std::ostream &stream);
#endif
    public:
        LockWithInfo(const std::string &whichLock);
        /*
         * Destroy the lock.  If DEBUG2 is defined, and this
         * is the last lock, close the log file.
         */
        ~LockWithInfo();
 
        /* Overload parent class lock() method, do our
         * stuff after taking the lock.
         */
        void lock();
        /* Overload parent class try_lock() method, do our
         * stuff after taking the lock.
         */
        bool try_lock() noexcept;
 
        /* Overload parent class unlock() method, do our
         * stuff before releasing the lock.
         */
        void unlock();
        /*
         * Ensure the current thread has taken the lock.  Throw
         * an exception if not.
         *
         * func, file and lineNum are used in constructing the
         *   exception so you know the source of the call.
         *
         * @param func Name of the function where the call was made
         * @param file Name of the file in which the call was made
         * @param lineNum Line number where the call was made.
         */
        void CheckTaken(const std::string &func,
                        const std::string &file,
                        const int lineNum) const;
        /*
         * Ensure the current thread has taken the lock and that
         * it is the specified thread. Throw an exception if not.
         *
         * This is typically used to ensure the DRAMA thread is the
         * current thread and has the lock.
         *
         * func, file and lineNum are used in constructing the
         *   exception so you know the source of the call.
         *
         * @param t    The thread which should be the current thread.
         * @param func Name of the function where the call was made 
         * @param file Name of the file in which the call was made
         * @param lineNum Line number where the call was made
         */
        void CheckTakenBy(std::thread::id t,
                          const std::string &func,
                          const std::string &file,
                          const int lineNum) const;
        /*
         * Ensure the current thread has taken the lock and that
         * it is the DRAMA task thread. Throw an exception if not.
         *
         * This is typically used to ensure the DRAMA thread is the
         * current thread and has the lock.
         *
         * func, file and lineNum are used in constructing the
         *   exception so you know the source of the call.
         *
         * @param task  Reference to the drama task.
         * @param func Name of the function where the call was made.
         * @param file Name of the file in which the call was made.
         * @param lineNum Line number where the call was made.
        */
        void CheckTakenBy(std::weak_ptr<Task> task,
                          const std::string &func,
                          const std::string &file,
                          const int lineNum) const;
        /*
         * Ensure the current thread has taken the lock and that
         * it is the DRAMA task thread. Throw an exception if not.
         *
         * This is typically used to ensure the DRAMA thread is the
         * current thread and has the lock.
         *
         * func, file and lineNum are used in constructing the
         *   exception so you know the source of the call.
         *
         * @param task  Reference to the drama task.
         * @param func Name of the function where the call was made.
         * @param file Name of the file in which the call was made
         * @param lineNum Line number where the call was made.
         */
        void CheckTakenBy(Task * task,
                          const std::string &func,
                          const std::string &file,
                          const int lineNum) const;
 
        /*
         * Ensure the current thread has not taken the lock.   Throws
         * an exception if it has.
         *
         * func, file and lineNum are used in constructing the
         *   exception so you know the source of the call.
         *
          * @param func Name of the function where the call was made.
         * @param file Name of the file in which the call was made.
         * @param lineNum Line number where the call was made.
         */
        void CheckNotTakenBy(const std::string func,
                             const std::string &file,
                             const int lineNum) const;
        /*
         * Show the status of the lock - output to std::cerr.
         */
        void Show() const;
    }; // class LockWithInfo
#endif /* if defined DRAMA2_LOCK_DEBUG */
 
    class RunDramaExitNotifier {
    public:
        /*
         * Constructor 
         */
        RunDramaExitNotifier() {}
        virtual int RunDramaHasExited() = 0;
 
        virtual bool JoinThreads(std::chrono::steady_clock::time_point until) = 0;
 
        virtual ~RunDramaExitNotifier() {}
    };
 
    class Task  : public std::enable_shared_from_this<Task> {
    public:
#ifdef DRAMA2_LOCK_DEBUG
        typedef LockWithInfo mutexType;
#else
        typedef std::recursive_timed_mutex mutexType;
#endif
        //typedef std::mutex mutexType;
        typedef std::lock_guard<mutexType> guardType;
        typedef std::unique_lock<mutexType> uniqueLockType;
 
        typedef drama::Request (Task::*ActionMethod)(drama::MessageHandler *mh);
        typedef void (Task::*ThreadActionMethod)(
            drama::thread::TAction *taction, 
            const drama::sds::Id& id);
 
    private:
        std::string _name;
 
        std::thread::id _taskCreThreadId;  // Id of thread that created task.
        std::thread::id _taskRunThreadId;  // Id of thread running the message loop.
        /*
         * This mutex is the main lock around DRAMA operations when
         * running in threaded mode. 
         */
        mutexType _dramaLock;
 
        /*
         * A set of the RunDRamaExitNotifier objects.
         */
        typedef std::set<RunDramaExitNotifier *> RunDramaExitNotifierSetType;
 
 
        /*
         *  This mutex locks DRAMA message notifications.  Unclear if it is
         *  actually needed.  We don't take the _dramaLock when waiting
         *  for messages, since we want other threads to be able to
         *  execute DRAMA calls.
         */
        mutexType _dramaMsgNotifyLock;
 
        OrphanDetailsQueue _orphanQueue;
 
        static void HandleDitsOrphan(StatusType *status);
        /*
         * Unfortunately, DitsPutOrphanHandler() does not allow
         * a clientData item to be passed to the handler routine.
         * We must maintain the address of our task in a static
         * so we can invoke methods on the right task.  Fortunately,
         * we can only have one task running - the DRAMA registration
         * process ensures that.
         */
        static Task *orphanTask;
        
        /*
         * Handle messages for orphaned transactions.  Non-static
         * method invoked by HandleDitsOrphan.
         */
        void HandleOrphanMes(StatusType *status);
 
 
        /* Invoked when a DRAMA message is available, processes it.
         */
        void ProcessMessage(long *exitFlag, StatusType *status);
 
        /*
         * A vector of notifiers to be invoked when the RunDrama() exits.
         */
        RunDramaExitNotifierSetType _loopExitNotifiers;
 
        /*
         * The task logger.  Only used if opened.
         */
        std::unique_ptr<logging::Logger> _taskLogger;
 
        /*
         * Add an action to the task.  Unlike the pubic interfaces, takes
         * a spawnable argument and an ActionHandlerPtr.
         *
         * This is invoked by all the public interfaces to add an
         * action to the DRAMA2 task.
         */ 
        virtual void AddAction(const std::string &name, bool spawnable,
                               ActionHandlerPtr obj,
                               const std::string &descr);
 
        /*
         * Shared pointer to self. This for std::shared<Task> operator.
         */
        std::shared_ptr<Task> _sharedSelf = nullptr;
        /*
         * Handling of shared pointers, as returned by TaskPtr() has
         * been initialised.
         */
        bool _sharedInit = false;
        /*
         * Will be set true if it is found by TaskPtr() that our object
         * already has a shared pointer.
         */
        bool _amShared = false;
 
        /*
         * If this is set true, then we don't complain if a separate thread
         * is being used to run DRAMA.
         */
        bool _sepThreadToRunDramaOK = false;
 
 
 
        Task& operator=(const Task &rhs) = delete;
        Task(const Task &source) = delete;
 
        /* Method used to implement AddMth */
       void AddMethodAction(
            const std::string &actionName,
            ActionMethod method,
            const std::string &descr);
        /* Method used to implement AddMthThd */
        void AddMethodAction(
            const std::string &actionName,
            ThreadActionMethod method,
            const std::string &descr);
 
 
 
 
    protected:
        virtual void OrphanHandler(const OrphanDetails &details);
 
    public:
 
        static const int DefBufSize;     
        static const int DefSelfBufSize; 
        Task(const std::string &name, int buffer=DefBufSize, 
             int flags=0, int selfBytes=DefSelfBufSize);
 
        virtual void RunDrama();
 
        void SeparateThreadRunsDrama() {
            _sepThreadToRunDramaOK = true;
        }
 
        
        virtual void Add(const std::string &name, MessageHandlerPtr obj,
                         const std::string &descr ="") {
            AddAction(name, false, obj,descr);
        }
        
        virtual void Add(const std::string &name, MessageHandler *obj,
                         const std::string &descr ="") {
            Add(name, MessageHandlerPtr(obj), descr);
        }
 
        virtual void Add(const std::string &name, MessageReceiveFunction func,
                         const std::string &descr ="") {
 
            Add(name, std::make_shared<MessageHandlerViaFunctor>(func), descr);
        }
 
        /* For some reason, this is ambigous with the above in some cases, so we need a different name */
        virtual void AddTA(const std::string &name, thread::ThreadActionFunction func,
                           const std::string &descr ="") {
            thread::AddActionFunction(this, name, func, descr);
        }
 
 
        virtual void AddSpawnable(const std::string &name, SpawnablePtr obj, 
                                  const std::string &descr ="") {
            AddAction(name, true, obj, descr);
        }
 
        virtual void AddSpawnable(const std::string &name, Spawnable *obj, 
                                  const std::string &descr ="") {
            AddAction(name, true, SpawnablePtr(obj), descr);
        }
 
        template <typename T>
        void AddMth(
            const std::string &actionName,
            T method,
            const std::string &descr ="") {
 
            //The important thing here is that we must allow a sub-class
            // of drama::Task to invoke specify its own methods.  This
            // cast allows that to happen.  The cast will fail if
            // the method doesn't have the right format.
            AddMethodAction(actionName, 
                            static_cast<drama::Task::ActionMethod>(method), descr);
        }
        template <typename T>
        void AddMthThd(
            const std::string &actionName,
            T method, 
            const std::string &descr ="") {
 
            //The important thing here is that we must allow a sub-class
            // of drama::Task to invoke specify its own methods.  This
            // cast allows that to happen.  The cast will fail if
            // the method doesn't have the right format.
            AddMethodAction(actionName, 
                            static_cast<drama::Task::ThreadActionMethod>(method), descr);
        }
        mutexType & Lock() {
            return  _dramaLock;
        }
        std::thread::id GetTaskCreThreadId() const {
            return _taskCreThreadId;
        }
        std::thread::id GetTaskRunThreadId() const {
            return _taskRunThreadId;
        }
 
        logging::Logger & Logger() {
            return *_taskLogger;
        }
 
        std::string TaskName() const {
            return _name;
        }
 
        void NotifyOnRunDramaExit(RunDramaExitNotifier *notifier);
        void CancelNotifyOnRunDramaExit(RunDramaExitNotifier *notifier);
 
        void AddOrphanToQueue(const OrphanDetails &orphan) {
#ifdef DRAMA2_LOCK_DEBUG // If debugging, check we have the lock.
            Lock().CheckTaken(__func__, __FILE__,__LINE__);
#endif
            _orphanQueue.push_front(orphan);
        }
 
 
        std::weak_ptr<Task> TaskPtr() {
 
            guardType guard(_dramaLock);           
            /*
             * This function initialises internal items on the
             * first time through - can't be done in constructor
             * as  shared_from_this() will always fail there.
             *
             * Note - shared_from_this() comes from std::enable_shared_from_this,
             *  which we inherit.
             */
            //std::thread::id thisThread = std::this_thread::get_id();
 
            //std::cerr << thisThread << "::TaskPtr() invoked, ";
            if (_sharedInit)
            {
                //std::cerr << "_shareInit is true, ";
                if (_amShared)
                {
                    //std::cerr << "_amShared, " << std::endl;
                    return shared_from_this();
                }
                else
                {
                    //std::cerr << "using _sharedSelf, " << std::endl;
                    return _sharedSelf;
                }
            }
            // Only get here if not initialised
            try
            {
                //std::cerr << "trying for shared_from_this(), ";
 
                std::shared_ptr<Task> p = shared_from_this();
                // If we have not thrown, then we are a shared pointer object.
                _sharedInit = true;
                _amShared = true;
                //std::cerr << "got it, am shared" << std::endl;
                return p;
            }
           /*
            * According to http://en.cppreference.com's page on std::enable_shared_from_this,
            * this is actually undefined behaviour this is not already a shared pointer, until
            * c++17.   It does at least seem to work for gnu C++ and Clang++ at c++11.
            */
            catch (const std::bad_weak_ptr &e)
            {
 
                //std::cerr << "Am not created from a shared, will init _sharedSelf" << std::endl;
 
                _sharedSelf = std::shared_ptr<Task>(this, drama::nodel());
                _amShared = false;
                _sharedInit = true;
                return _sharedSelf;
            }
 
        }
 
        template <typename T>
        std::shared_ptr<T> TaskPtrAs() {
            return std::dynamic_pointer_cast<T>(std::shared_ptr<Task>(TaskPtr()));
        }
 
        virtual double GetJoinTimeout() const;
 
        virtual ~Task();
 
        void Signal(const std::string &name, sds::Id *arg=nullptr, void *data=nullptr);
 
        void Signal(long int index, sds::Id *arg=nullptr, void *data=nullptr);
 
 
        virtual void _MessageUser(const std::string &text) const;
 
 
 
        void SetDetails(const std::string &descr, int type = 0);
 
        std::string GetTaskDescription(const std::string &taskName) const;
 
        int GetTaskType(const std::string &taskName) const;
 
 
        /*
         * Ensure the current thread has taken the DRAMA task lock.  
         * Throw an exception if not.
         *
         * This won't do anything unless DRAMA was compiled with
         * the DRAMA2_LOCK_DEBUG macro defined.
         *
         * This method is static and uses an internal static variable
         * to find the drama::Task object. 
         *
         * func, file and lineNum are used in constructing the
         *   exception so you know the source of the call.
         *
         * @param func Name of the function where the call was made
         * @param file Name of the file in which the call was made
         * @param lineNum Line number where the call was made.
         */
        static void CheckLockTaken(const std::string &func,
                                   const std::string &file,
                                   const int lineNum) ;
 
 
 
    }; // Class Task.
 
    template <class TaskClass, typename... ParamTypes >
    void CreateRunDramaTask(ParamTypes... taskPars) {
 
        static_assert(std::is_base_of<Task, TaskClass>(), "TaskClass must be a sub-class of drama::task");
 
        try 
        {
            /* Create and run the task */
            TaskClass task(std::forward<ParamTypes>(taskPars)...);
            task.RunDrama();
        }
        catch (const drama::Exception &e) 
        {
            std::cerr << "CreateRunDramaTask():drama::Exception thrown"
                      << std::endl
                      << e.toString(true)   // Outputs all the exception details.
                      << std::endl;
            
            exit (e.statusAsSysExitCode());
        }
        catch (const std::exception &e)
        {
            std::cerr << "CreateRunDramaTask():std::exception thrown."
                      << std::endl
                      << e.what() 
                      << std::endl;
            exit(1);
        }
        catch (...)
        {
            std::cerr << "CreateRunDramaTask():Non-standard exception thrown."
                      << std::endl;
            throw;  // Rethrow to get any core dump etc
            
        }
    } // CreateRunDramaTask()
#if 1
    template <class TaskClass>
    void CreateRunDramaTask() {
 
        static_assert(std::is_base_of<Task, TaskClass>(), "TaskClass must be a sub-class of drama::task");
        try 
        {
            /* Create and run the task */
            TaskClass task;
            task.RunDrama();
        }
        catch (const drama::Exception &e) 
        {
            std::cerr << "CreateRunDramaTask():drama::Exception thrown"
                      << std::endl
                      << e.toString(true)   // Outputs all the exception details.
                      << std::endl;
            
            exit (e.statusAsSysExitCode());
        }
        catch (const std::exception &e)
        {
            std::cerr << "CreateRunDramaTask():std::exception thrown."
                      << std::endl
                      << e.what() 
                      << std::endl;
            exit(1);
        }
        catch (...)
        {
            std::cerr << "CreateRunDramaTask():Non-standard excepton thrown."
                      << std::endl;
            throw;  // Rethrow to get any core dump etc
            
        }
    } // CreateRunDramaTask()
#endif
} // namespace drama
 
#endif