thread.hpp

Go to the documentation of this file.

/*
  THREAD.hpp  -  thin convenience wrapper for starting threads

   Copyright (C)
     2008, 2010,      Christian Thaeter <ct@pipapo.org>
     2008, 2010,      Hermann Vosseler <Ichthyostega@web.de>
     2023,            Hermann Vosseler <Ichthyostega@web.de>

  **Lumiera** is free software; you can redistribute it and/or modify it
  under the terms of the GNU General Public License as published by the
  Free Software Foundation; either version 2 of the License, or (at your
  option) any later version. See the file COPYING for further details.

*/


#ifndef LIB_THREAD_H
#define LIB_THREAD_H


#include "lib/error.hpp"
#include "lib/nocopy.hpp"
#include "include/logging.h"
#include "lib/meta/trait.hpp"
#include "lib/meta/function.hpp"
#include "lib/format-util.hpp"
#include "lib/result.hpp"

#include <utility>
#include <thread>
#include <string>
#include <tuple>


namespace util {
  std::string sanitise (std::string const&);
}
namespace lib {

  using std::string;
  using std::tuple;
  
  
  
  namespace thread {// Thread-wrapper base implementation...
    
    using lib::meta::lateBindInstance;
    using lib::meta::typeSymbol;
    using lib::meta::_Fun;
    using util::isnil;
    using std::function;
    using std::forward;
    using std::move;
    using std::decay_t;
    using std::invoke_result_t;
    using std::is_constructible;
    using std::make_from_tuple;
    using std::tuple_cat;
    using std::is_same;
    using std::__or_;
    
    struct ThreadWrapper
      : util::MoveOnly
      {
        const string threadID_;
        std::thread threadImpl_;
        
        bool isLive()  const { return threadImpl_.joinable(); }
        
        
        ThreadWrapper()
          : threadID_{util::BOTTOM_INDICATOR}
          , threadImpl_{}
          { }
        
        ThreadWrapper (string const& threadID)
          : threadID_{isnil(threadID)? "sub-thread" : util::sanitise (threadID)}
          , threadImpl_{} //Note: deliberately not starting the thread yet...
          { }
        
        template<class...INVO>
        void
        launchThread (tuple<INVO...>&& invocation)
          {
            ASSERT (not isLive(), "Thread already running");
            threadImpl_ = make_from_tuple<std::thread> (invocation);
          };
        
        
        bool invokedWithinThread()  const;
        
        void markThreadStart();
        void markThreadEnd  ();
        void setThreadName  ();
        void waitGracePeriod()  noexcept;
        
        /* empty implementation for some policy methods */
        void handle_begin_thread() { }   
        void handle_after_thread() { }   
        void handle_loose_thread() { }   
        
        static string decorate_with_global_count (string const&);
        
        void detach_thread_from_wrapper()
          {
            if (isLive())
              threadImpl_.detach();
          }
      };
    
    
    
    template<class BAS, typename=void>
    struct PolicyLaunchOnly
      : BAS
      {
        using BAS::BAS;
        
        template<class FUN, typename...ARGS>
        void
        perform_thread_function(FUN&& callable, ARGS&& ...args)
          {
            try {
                 //  execute the actual operation in this new thread
                std::invoke (forward<FUN> (callable), forward<ARGS> (args)...);
              }
            ERROR_LOG_AND_IGNORE (thread, "Thread function")
          }
        
        void
        handle_after_thread()
          {
            BAS::detach_thread_from_wrapper();
          }
        
        void
        handle_loose_thread()
          {
            BAS::waitGracePeriod();
          }
      };
    
    
    template<class BAS, class TAR>
    struct PolicyLifecycleHook
      : PolicyLaunchOnly<BAS>
      {
        using BasePol = PolicyLaunchOnly<BAS>;
        using BasePol::BasePol;
        
        using Self = PolicyLifecycleHook;
        using Hook = function<void(Self&)>;
        
        Hook hook_beginThread{};
        Hook hook_afterThread{};
        Hook hook_looseThread{};
        
        void
        handle_begin_thread()
          {
            if (hook_beginThread)
              hook_beginThread (*this);
            else
              BasePol::handle_begin_thread();
          }
        
        void
        handle_after_thread()
          {
            if (hook_afterThread)
              hook_afterThread (*this);
            // Note: ensure thread is detached at end
            BAS::detach_thread_from_wrapper();
          }
        
        void
        handle_loose_thread()
          {
            if (hook_looseThread)
              hook_looseThread (*this);
            else
              BasePol::handle_loose_thread();
          }
      };
    
    
    template<class BAS, typename RES>
    struct PolicyResultJoin
      : BAS
      {
        using BAS::BAS;
        
        lib::Result<RES> result_{error::Logic{"No result yet, thread still running; need to join() first."}};
        
        
        template<class FUN, typename...ARGS>
        void
        perform_thread_function(FUN&& callable, ARGS&& ...args)
          {
            static_assert (__or_<is_same<RES,void>
                                ,is_constructible<RES, invoke_result_t<FUN,ARGS...>>>());
            
            // perform the given operation (failsafe) within this thread and capture result...
            result_ = std::move (
                        lib::Result{forward<FUN>(callable)
                                   ,forward<ARGS>(args)...});
          }
        
        void
        handle_after_thread()
          {
            /* do nothing -- thread must be joined manually */;
          }
        
        void
        handle_loose_thread()
          {
            ALERT (thread, "Thread '%s' was not joined. Abort.", BAS::threadID_.c_str());
          }
      };
    
    
    
    template<template<class,class> class POL, typename RES =void>
    class ThreadLifecycle
      : protected POL<ThreadWrapper, RES>
      {
        using Policy = POL<ThreadWrapper,RES>;
        
        template<typename...ARGS>
        void
        invokeThreadFunction (ARGS&& ...args)
          {
            while (not Policy::isLive()) // wait for thread-ID to become visible
              std::this_thread::yield();//  (typically happens when debugging)
            Policy::handle_begin_thread();
            Policy::markThreadStart();
            Policy::perform_thread_function (forward<ARGS> (args)...);
            Policy::markThreadEnd();
            Policy::handle_after_thread();
          }
        
        
      protected:
       ~ThreadLifecycle()
          {
            if (Policy::isLive())
              Policy::handle_loose_thread();
          }
        
        ThreadLifecycle()
          : Policy{}
          { }
        
      public:
        template<class W, class...INVO>
        static auto
        buildInvocation (W& wrapper, tuple<INVO...>&& invocation)
        {                                           //the thread-main function
          return tuple_cat (tuple{&ThreadLifecycle::invokeThreadFunction<INVO...>
                                 , &wrapper}      //  passing the wrapper as instance-this
                           ,move (invocation));  //...invokeThreadFunction() in turn delegates
        }                                       //    to the user-provided thread-operation
        
        template<class...INVO>
        static auto
        buildLauncher (INVO&& ...args)
        {
          tuple<decay_t<INVO>...> argCopy{forward<INVO> (args)...};
          return [invocation = move(argCopy)]// Note: functor+args bound by-value into the λ
                 (ThreadLifecycle& wrapper)
                    {                        // special treatment for launchDetached
                      auto boundInvocation = lateBindInstance (wrapper, move (invocation));
                      wrapper.launchThread (buildInvocation (wrapper, move(boundInvocation)));
                    };
        }
        
        
        struct Launch
          : util::MoveOnly
          {
            using Act = function<void(ThreadLifecycle&)>;
            
            Act launch;
            string id;
            
            template<class FUN, typename...ARGS>
            Launch (FUN&& threadFunction, ARGS&& ...args)
              : launch{buildLauncher (forward<FUN>(threadFunction), forward<ARGS>(args)...)}
              { }
            
            template<class TAR, typename...ARGS>
            Launch (RES (TAR::*memFun) (ARGS...), ARGS ...args)  
              : Launch{move (memFun)
                      ,lib::meta::InstancePlaceholder<TAR>{}
                      ,forward<ARGS> (args)... }
              { }
            
            Launch&&
            threadID (string const& threadID)
              {
                id = threadID;
                return move(*this);
              }
            
            Launch&&
            decorateCounter()
              {
                id = Policy::decorate_with_global_count (id);
                return move(*this);
              }
            
            template<typename HOOK>
            Launch&&
            atStart (HOOK&& hook)
              {
                return addHook (&Policy::hook_beginThread, forward<HOOK> (hook));
              }
            
            template<typename HOOK>
            Launch&&
            atExit (HOOK&& hook)
              {
                return addHook (&Policy::hook_afterThread, forward<HOOK> (hook));
              }
            
            template<typename HOOK>
            Launch&&
            onOrphan (HOOK&& hook)
              {
                return addHook (&Policy::hook_looseThread, forward<HOOK> (hook));
              }
            
          private:
            template<typename HOOK, class FUN>
            auto
            adaptedHook (FUN Policy::*, HOOK&& hook)
              {
                static_assert(1 == _Fun<FUN>::ARITY);
                static_assert(1 >= _Fun<HOOK>::ARITY);
                // argument type expected by the hooks down in the policy class
                using Arg = typename _Fun<FUN>::Args::List::Head;
                // distinguish if user provided functor takes zero or one argument
                if constexpr (0 == _Fun<HOOK>::ARITY)
                  return [hook = forward<HOOK>(hook)](Arg){ hook(); };
                else
                  { // instance type expected by the user-provided hook
                    using Target = typename _Fun<HOOK>::Args::List::Head;
                    return [hook = forward<HOOK>(hook)]
                           (Arg& threadWrapper)
                              { // build a two-step cast path from the low-level wrapper to user type
                                ThreadLifecycle& base = static_cast<ThreadLifecycle&> (threadWrapper);
                                Target&        target = static_cast<Target&> (base);
                                hook (target);
                              };
                  }
              }
            
            template<typename HOOK, class FUN>
            Launch&&
            addHook (FUN Policy::*storedHook, HOOK&& hook)
              {
                return addLayer ([storedHook, hook = adaptedHook (storedHook, forward<HOOK> (hook))]
                                 (ThreadLifecycle& wrapper)
                                    {
                                      wrapper.*storedHook = move (hook);
                                    });
              }
            
            Launch&&
            addLayer (Act action)
              {
                launch = [action=move(action), chain=move(launch)]
                         (ThreadLifecycle& wrapper)
                            {
                              action(wrapper);
                              chain (wrapper);
                            };
                return move(*this);
              }
          };
        
        
        ThreadLifecycle (Launch launcher)
          : Policy{launcher.id}
          {
            launcher.launch (*this);
          }
        
        template<class FUN, typename...ARGS>
        ThreadLifecycle (string const& threadID, FUN&& threadFunction, ARGS&& ...args)
          : ThreadLifecycle{
              Launch{forward<FUN> (threadFunction), forward<ARGS> (args)...}
                    .threadID(threadID)}
          { }
        
        template<class SUB, typename...ARGS>
        ThreadLifecycle (RES (SUB::*memFun) (ARGS...), ARGS ...args)
          : ThreadLifecycle{
              Launch{std::move (memFun)
                    ,static_cast<SUB*> (this)
                    ,forward<ARGS> (args)...
                    }
                    .threadID(util::joinDash (typeSymbol<SUB>(), args...))}
          { }
        
        
        
        explicit
        operator bool()  const
          {
            return Policy::isLive();
          }
        
        using Policy::invokedWithinThread;
      };
    
  }//(End)base implementation.
  
  
  
  
  
  
  
  /************************************************************************/
  class Thread
    : public thread::ThreadLifecycle<thread::PolicyLaunchOnly>
    {
    public:
      using ThreadLifecycle::ThreadLifecycle;
    };
  
  
  
  
  /************************************************************************/
  template<typename RES =void>
  class ThreadJoinable
    : public thread::ThreadLifecycle<thread::PolicyResultJoin, RES>
    {
      using Impl = thread::ThreadLifecycle<thread::PolicyResultJoin, RES>;
    public:
      using Impl::Impl;
      
      lib::Result<RES>
      join ()
        {
          if (not Impl::threadImpl_.joinable())
            throw lumiera::error::Logic ("joining on an already terminated thread");
          
          Impl::threadImpl_.join();
          
          return Impl::result_;
        }
    };

  template<typename FUN, typename...ARGS>
  ThreadJoinable (string const&, FUN&&, ARGS&&...) -> ThreadJoinable<std::invoke_result_t<FUN,ARGS...>>;
  
  
  
  /************************************************************************/
  class ThreadHookable
    : public thread::ThreadLifecycle<thread::PolicyLifecycleHook>
    {
    public:
      using ThreadLifecycle::ThreadLifecycle;
    };
  
  
  
  template<class TAR = ThreadHookable>
  inline void
  launchDetached (ThreadHookable::Launch&& launchBuilder)
  {
    static_assert (lib::meta::is_Subclass<TAR, ThreadHookable>());
    
    new TAR{move(launchBuilder)
                 .atExit([](TAR& selfAllocation)
                           {
                             delete &selfAllocation;
                           })
                 .onOrphan([](thread::ThreadWrapper& wrapper)
                           {
                             wrapper.detach_thread_from_wrapper();
                           })};
     // Note: allocation tossed on the heap deliberately
  } //  The thread-function will pick up and manage *this
  
  template<class TAR = ThreadHookable, typename...INVO>
  inline void
  launchDetached (string const& threadID, INVO&& ...args)
  {
    using Launch = typename TAR::Launch;
    launchDetached<TAR> (Launch{forward<INVO> (args)...}
                          .threadID (threadID));
  }
  
  template<class TAR, typename...ARGS>
  inline void
  launchDetached (string const& threadID, void (TAR::*memFun) (ARGS...), ARGS ...args)
  {
    using Launch = typename TAR::Launch;
    launchDetached<TAR> (Launch{std::move (memFun)
                               ,lib::meta::InstancePlaceholder<TAR>{}
                               ,forward<ARGS> (args)...
                               }
                               .threadID (threadID));
  }
  
  template<class TAR, typename...ARGS>
  inline void
  launchDetached (void (TAR::*memFun) (ARGS...), ARGS ...args)
  {
    launchDetached (util::joinDash (lib::meta::typeSymbol<TAR>(), args...)
                   ,memFun
                   ,forward<ARGS> (args)...
                   );
  }
  
  
} // namespace lib
#endif /*LIB_THREAD_H*/