thread.hpp

Go to the documentation of this file.

#pragma once

#define FC_CONTEXT_STACK_SIZE (2048*1024)

#include <fc/thread/task.hpp>

namespace fc {
  class time_point;
  class microseconds;

   namespace detail
   {
      class worker_pool;
      void* get_thread_specific_data(unsigned slot);
      void set_thread_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
      unsigned get_next_unused_task_storage_slot();
      void* get_task_specific_data(unsigned slot);
      void set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
   }

   class thread_idle_notifier {
   public:
      virtual ~thread_idle_notifier() {}

      virtual task_base* idle() = 0;
      virtual void busy() = 0;
   };

  class thread {
    public:
      thread( const std::string& name = "", thread_idle_notifier* notifier = 0 );
      thread( thread&& m ) = delete;
      thread& operator=(thread&& t ) = delete;

      static thread& current();
      static void    cleanup();

     
      const string& name()const;

      void        set_name( const string& n );
       
      const char* current_task_desc() const;

      void    debug( const std::string& d );
     
     
      template<typename Functor>
      auto async( Functor&& f, const char* desc FC_TASK_NAME_DEFAULT_ARG, priority prio = priority()) -> fc::future<decltype(f())> {
         typedef decltype(f()) Result;
         typedef typename std::remove_const_t< std::remove_reference_t<Functor> > FunctorType;
         typename task<Result,sizeof(FunctorType)>::ptr tsk = 
              task<Result,sizeof(FunctorType)>::create( std::forward<Functor>(f), desc );
         tsk->retain(); // HERE BE DRAGONS
         fc::future<Result> r( std::dynamic_pointer_cast< promise<Result> >(tsk) );
         async_task(tsk.get(),prio);
         return r;
      }
      void poke();
     
     
      template<typename Functor>
      auto schedule( Functor&& f, const fc::time_point& when, 
                     const char* desc FC_TASK_NAME_DEFAULT_ARG, priority prio = priority()) -> fc::future<decltype(f())> {
         typedef decltype(f()) Result;
         typename task<Result,sizeof(Functor)>::ptr tsk = 
              task<Result,sizeof(Functor)>::create( std::forward<Functor>(f), desc );
         tsk->retain(); // HERE BE DRAGONS
         fc::future<Result> r( std::dynamic_pointer_cast< promise<Result> >(tsk) );
         async_task(tsk.get(),prio,when);
         return r;
      }
     
      void quit();

      void signal(int);
     
      bool is_running()const;
      bool is_current()const;
     
      priority current_priority()const;
      ~thread();

       template<typename T1, typename T2>
       int wait_any( const fc::future<T1>& f1, const fc::future<T2>& f2, const microseconds& timeout_us = microseconds::maximum()) {
          std::vector<fc::promise_base::ptr> proms(2);
          proms[0] = std::static_pointer_cast<fc::promise_base>(f1.m_prom);
          proms[1] = std::static_pointer_cast<fc::promise_base>(f2.m_prom);
          return wait_any_until(std::move(proms), fc::time_point::now()+timeout_us );
       }
    private:
      thread( class thread_d* ); // parameter is ignored, will create a new thread_d
      friend class promise_base;
      friend class task_base;
      friend class thread_d;
      friend class mutex;
      friend class detail::worker_pool;
      friend void* detail::get_thread_specific_data(unsigned slot);
      friend void detail::set_thread_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
      friend unsigned detail::get_next_unused_task_storage_slot();
      friend void* detail::get_task_specific_data(unsigned slot);
      friend void detail::set_task_specific_data(unsigned slot, void* new_value, void(*cleanup)(void*));
#ifndef NDEBUG
      friend class non_preemptable_scope_check;
#endif
      friend void yield();
      friend void usleep(const microseconds&);
      friend void sleep_until(const time_point&);
      friend void exec();
      friend int wait_any( std::vector<promise_base::ptr>&& v, const microseconds& );
      friend int wait_any_until( std::vector<promise_base::ptr>&& v, const time_point& tp );
      void wait_until( promise_base::ptr && v, const time_point& tp );
      void notify( const promise_base::ptr& v );

      void yield(bool reschedule=true);
      void sleep_until( const time_point& t );
      void  exec();
      int  wait_any_until( std::vector<promise_base::ptr>&& v, const time_point& );

      void async_task( task_base* t, const priority& p );
      void async_task( task_base* t, const priority& p, const time_point& tp );

      void notify_task_has_been_canceled();
      void unblock(fc::context* c);

      class thread_d* my;
  };

   void yield();

   void usleep( const microseconds& u );

   void sleep_until( const time_point& tp );

   void  exec();

   template<typename T1, typename T2>
   int wait_any( const fc::future<T1>& f1, const fc::future<T2>& f2, const microseconds timeout_us = microseconds::maximum()) {
      return fc::thread::current().wait_any(f1,f2,timeout_us);
   }
   int wait_any( std::vector<promise_base::ptr>&& v, const microseconds& timeout_us = microseconds::maximum() );
   int wait_any_until( std::vector<promise_base::ptr>&& v, const time_point& tp );

   template<typename Functor>
   auto async( Functor&& f, const char* desc FC_TASK_NAME_DEFAULT_ARG, priority prio = priority()) -> fc::future<decltype(f())> {
      return fc::thread::current().async( std::forward<Functor>(f), desc, prio );
   }
   template<typename Functor>
   auto schedule( Functor&& f, const fc::time_point& t, const char* desc FC_TASK_NAME_DEFAULT_ARG, priority prio = priority()) -> fc::future<decltype(f())> {
      return fc::thread::current().schedule( std::forward<Functor>(f), t, desc, prio );
   }

  template<typename Functor>
  auto sync_call( thread* t, Functor&& f, const char* desc FC_TASK_NAME_DEFAULT_ARG, priority prio = priority()) -> decltype(f())
  {
     if( t == nullptr )
         return f();

     typedef decltype(f()) Result;
     future<Result> r = t->async( f, desc, prio );
     return r.wait();
  }

} // end namespace fc

#ifdef _MSC_VER
struct _EXCEPTION_POINTERS;

namespace fc {
   /* There's something about the setup of the stacks created for fc::async tasks
    * that screws up the global structured exception filters installed by
    * SetUnhandledExceptionFilter().  The only way I've found to catch an 
    * unhaldned structured exception thrown in an async task is to put a 
    * __try/__except block inside the async task.
    * We do just that, and if a SEH escapes outside the function running 
    * in the async task, fc will call an exception filter privided by 
    * set_unhandled_structured_exception_filter(), passing as arguments
    * the result of GetExceptionCode() and GetExceptionInformation().
    *
    * Right now there is only one global exception filter, used for any 
    * async task.
    */
   typedef int (*unhandled_exception_filter_type)(unsigned, _EXCEPTION_POINTERS*);
   void set_unhandled_structured_exception_filter(unhandled_exception_filter_type new_filter);
   unhandled_exception_filter_type get_unhandled_structured_exception_filter();
}
#endif